I would like to make a portable thread pool using boost::thread
. Some one could tell me to look at boost::thread_group
and boost::asio
. I know, but I am not really familiar with these things so I don't know if they are suitable for my purpose.
Please give me comments on what I supposed to do here.
JobItem and JobQueue:
class JobItem
{
private:
uint32_t job_type_; //!< Job type
char* job_ptr_; //!< Pointer to job content
// ...
};
class JobQueue
{
private:
queue<JobItem*> job_queue_;
public:
// Definitely I need to use mutex and condition variable here
JobItem* pick();
void push(JobItem* job_item);
};
JobWorker (worker thread):
class JobWorker
{
private:
JobQueue* shared_job_queue_;
boost::thread the_thread_;
public:
JobWorker(JobQueue* shared_queue) : shared_job_queue_(shared_queue);
// exec_job will be overridden by concrete subclasses
virtual void exec_job(JobItem* job_item) = 0;
void run()
{
while(true)
{
JobItem* job = shared_job_queue_->pick();
exec_job(job);
delete job;
}
}
void start()
{
the_thread_ = boost::thread(boost::bind(&JobWorker::run, this));
}
void stop()
{
the_thread_.join();
}
};
Concrete JobWorker subclasses:
class Task_1_Worker : public JobWorker
{
// override exec_job
void exec_job(JobItem* job)
{
// Process job item here
}
}
class Task_2_Worker : public JobWorker
{
// override exec_job
void exec_job(JobItem* job)
{
// Process job item here
}
}
Now, WorkerPool with template:
template <class JobWorkerT>
class WorkerPool
{
private:
JobQueue shared_job_queue_;
list<JobWorkerT> job_workers_;
public:
WorkerPool(const int num_threads);
~WorkerPool(void) {};
void push(JobItem* job)
{
shared_job_queue_.push(job);
}
void start();
void stop();
};
template <class JobWorkerT>
WorkerPool<JobWorkerT>::WorkerPool(const int num_threads)
{
for(int i=0; i<num_threads; i++)
{
JobWorkerT job_worker(&shared_job_queue_);
job_workers_.push_back(job_worker);
}
}
template <class JobWorkerT>
void WorkerPool<JobWorkerT>::start()
{
for(std::list<JobWorkerT>::iterator iter = job_workers_.begin(); iter != job_workers_.end(); iter++)
{
((JobWorkerT)(*iter)).start();
}
}
template <class JobWorkerT>
void WorkerPool<JobWorkerT>::stop()
{
for(std::list<JobWorkerT>::iterator iter = job_workers_.begin(); iter != job_workers_.end(); iter++)
{
((JobWorkerT)(*iter)).stop();
}
}
main()
int main()
{
WorkerPool<Task_1_Worker>* task1_grp = new WorkerPool<Task_1_Worker>();
WorkerPool<Task_2_Worker>* task2_grp = new WorkerPool<Task_2_Worker>();
task1_grp->start();
task2_grp->start();
// Producers should call task1_grp.push(JobItem*) or task2_grp.push(JobItem*) to dispatch new jobs to thread pools
}