Charon/blockchain
1
0
Fork 0
forked from lthn/blockchain
Code Pull requests Activity

added option for manual specifying number of threads

Browse source
This commit is contained in:
cryptozoidberg 2022-02-08 14:55:35 +01:00
parent 761b75ad72
commit 5d808ad39e
No known key found for this signature in database
GPG key ID: 22DEB97A54C6FDEC
1 changed files with 39 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
src/common/threads_pool.h
View file

@ -13,7 +13,7 @@ namespace utils
struct call_executor_base
{
virtual void execute()=0;
virtual void execute() = 0;
};
template<typename t_executor_func>
@ -28,7 +28,7 @@ namespace utils
}
};
template<typename t_executor_func>
template<typename t_executor_func>
std::shared_ptr<call_executor_base> build_call_executor(t_executor_func func)
{
std::shared_ptr<call_executor_base> res(static_cast<call_executor_base*>(new call_executor_t<t_executor_func>(func)));
@ -40,26 +40,30 @@ namespace utils
public:
typedef std::list<std::shared_ptr<call_executor_base>> jobs_container;
template<typename t_executor_func>
static void add_job_to_container(jobs_container& cntr, t_executor_func func)
template<typename t_executor_func>
static void add_job_to_container(jobs_container& cntr, t_executor_func func)
{
cntr.push_back(std::shared_ptr<call_executor_base>(static_cast<call_executor_base*>(new call_executor_t<t_executor_func>(func))));
}
void init()
{
m_is_stop = false;
int num_threads = std::thread::hardware_concurrency();
this->init(num_threads);
}
void init(unsigned int num_threads)
{
m_is_stop = false;
for (int i = 0; i < num_threads; i++)
{
m_threads.push_back(std::thread([this](){this->worker_func(); }));
m_threads.push_back(std::thread([this]() {this->worker_func(); }));
}
}
threads_pool(): m_is_stop(false), m_threads_counter(0)
threads_pool() : m_is_stop(false), m_threads_counter(0)
{}
template<typename t_executor_func>
bool add_job(t_executor_func func)
{
@ -109,41 +113,41 @@ namespace utils
}
}
private:
void worker_func()
private:
void worker_func()
{
LOG_PRINT_L0("Worker thread is started");
while (true)
{
LOG_PRINT_L0("Worker thread is started");
while (true)
std::shared_ptr<call_executor_base> job;
{
std::shared_ptr<call_executor_base> job;
{
std::unique_lock<std::mutex> lock(m_queue_mutex);
std::unique_lock<std::mutex> lock(m_queue_mutex);
m_condition.wait(lock, [this]()
{
return !m_jobs_que.empty() || m_is_stop;
});
if (m_is_stop)
{
LOG_PRINT_L0("Worker thread is finished");
return;
}
job = m_jobs_que.front();
m_jobs_que.pop_front();
m_condition.wait(lock, [this]()
{
return !m_jobs_que.empty() || m_is_stop;
});
if (m_is_stop)
{
LOG_PRINT_L0("Worker thread is finished");
return;
}
job->execute();
job = m_jobs_que.front();
m_jobs_que.pop_front();
}
job->execute();
}
}
jobs_container m_jobs_que;
std::condition_variable m_condition;
std::mutex m_queue_mutex;
std::vector<std::thread> m_threads;
std::atomic<bool> m_is_stop;
std::atomic<int64_t> m_threads_counter;
jobs_container m_jobs_que;
std::condition_variable m_condition;
std::mutex m_queue_mutex;
std::vector<std::thread> m_threads;
std::atomic<bool> m_is_stop;
std::atomic<int64_t> m_threads_counter;
};
}