I have read the book "C++ concurrency in action" and understood the thread_pool implementation. I have changed a few things according to my project requirements.
I have used std::variant to support heterogeneous work-queue to store different task arriving on a epoll-event loop. Currently In my project I have only two different types of task arrive on the epoll loop. Those are TaskECB and TaskRPCB. I have created classes for both of them and overloaded the operator()
#define THREAD_POOL_SIZE 100
std::map<std::string,std::string> tidToTname;
template<typename T>
class threadSafeQueue {
private:
mutable std::mutex mut;
std::queue<std::shared_ptr<T>> taskQueue; /* task as pushed here and
task are processed in FIFO
style */
std::condition_variable dataCond; /* used to protect the queue */
public:
threadSafeQueue(){}
void waitAndPop(T& value); /* wait untill task is not available in
the queue */
std::shared_ptr<T> waitAndPop();/* same but returns a shared pointer */
bool tryPop(T& value); /* does not block */
std::shared_ptr<T> tryPop(); /* does not block and returns a pointer*/
void Push(T newData);
bool Empty() const; /* check if queue is empty or not */
void notifyAllThreads(); /* notify all the waiting threads
used in Thpool decallocation */
};
template<typename T>
void threadSafeQueue<T>::notifyAllThreads() {
dataCond.notify_all();
}
template<typename T>
void threadSafeQueue<T>::waitAndPop(T& value) {
std::unique_lock<std::mutex> lk(mut);
dataCond.wait(lk,[this](){return !taskQueue.empty();});
value = std::move(*taskQueue.front());
taskQueue.pop();
}
template<typename T>
std::shared_ptr<T> threadSafeQueue<T>::waitAndPop() {
std::unique_lock<std::mutex> lk(mut);
dataCond.wait(lk,[this](){return !taskQueue.empty();});
std::shared_ptr<T> res = taskQueue.front();
taskQueue.pop();
return res;
}
template<typename T>
bool threadSafeQueue<T>::tryPop(T& value) {
std::lock_guard<std::mutex> lk(mut);
if(taskQueue.empty())
return false;
value = std::move(*taskQueue.front());
taskQueue.pop();
return true;
}
template<typename T>
std::shared_ptr<T> threadSafeQueue<T>::tryPop() {
std::lock_guard<std::mutex> lk(mut);
if(taskQueue.empty())
return std::shared_ptr<T>(); /* return nullptr */
std::shared_ptr<T> res = taskQueue.front();
taskQueue.pop();
return res;
}
template<typename T>
void threadSafeQueue<T>::Push(T newData) { /* TODO: size check before pushing */
std::shared_ptr<T> data(std::make_shared<T>(std::move(newData)));
/* construct the object before lock*/
std::lock_guard<std::mutex> lk(mut);
taskQueue.push(data);
dataCond.notify_one();
}
template<typename T>
bool threadSafeQueue<T>::Empty() const {
std::lock_guard<std::mutex> lk(mut);
return taskQueue.empty();
}
class TaskRPCB { /* Task RecvAndProcessCallbacks */
private:
int len;
uint fdId;
int streamId;
void* msgBlob;
std::function<void(void*,int,uint,int)> func;
public:
TaskRPCB(std::function<void(void*,int,uint,int)>&f , void* msgBlob,int len,
uint fdId, int streamId) {
this->func = f;
this->msgBlob = msgBlob;
this->len = len;
this->fdId = fdId;
this->streamId = streamId;
}
void operator()() {
higLog("%s","TaskRPCB function is executing...");
func(msgBlob,len,fdId,streamId);
}
};
class TaskECB { /* Task eventCallBack */
private:
std::function<void(void)> func;
public:
TaskECB(std::function<void(void)>&f) : func(f) {}
void operator()() {
higLog("%s","TaskECB function is executing...");
func();
}
};
typedef variant<TaskECB,TaskRPCB> taskTypes;
class Thpool {
std::atomic_bool done;
threadSafeQueue<taskTypes> workQ;
std::vector<std::thread> threads;
void workerThread() {
auto tid = std::this_thread::get_id();
std::stringstream ss;
ss << tid;
std::string s = ss.str();
while(!done) {
auto task = workQ.waitAndPop();
if(task != nullptr and !done) {
printf("%s is executing now : ",tidToTname[s].c_str());
if((*task).index() == 0) { // TODO: change 0 and 1 to enums
auto func = get<TaskECB>(*task);
func();
}else if((*task).index() == 1) {
auto func = get<TaskRPCB>(*task);
func();
}
}
}
}
public:
Thpool(): done(false) {
// unsigned const maxThreadCount = std::thread::hardware_concurrency();
unsigned const maxThreadCount = THREAD_POOL_SIZE;
printf("ThreadPool Size = %d",maxThreadCount);
/* save thread names for logging purpose */
std::vector<std::string> tnames;
for(unsigned int i = 0;i<maxThreadCount;i++) {
tnames.push_back("Thread_" + std::to_string(i+1));
}
try { /* exception might arise due to thread creation */
for(unsigned int i = 0;i<maxThreadCount;i++) {
threads.push_back(std::thread(&Thpool::workerThread,this));
/*map this ith threadID to a name Thread_i*/
auto tid = threads[i].get_id();
std::stringstream ss;
ss << tid;
tidToTname[ss.str()] = tnames[i];
}
}catch(...) {
done = true;
throw;
}
}
~Thpool() {
// done = true;
}
template<typename taskType>
void submit(taskType task) {
workQ.Push(task);
}
void deAllocatePool() {
done = true;
workQ.notifyAllThreads();
// unsigned const maxThreadCount = std::thread::hardware_concurrency();
unsigned const maxThreadCount = THREAD_POOL_SIZE;
for(unsigned int i = 0;i<maxThreadCount;) {
if(threads[i].joinable()) {
threads[i].join();
i++; /* go for the next thread in the pool */
}else {
workQ.notifyAllThreads();
}
}
}
};
/*============== Thread Pool code ends ============== */
How I use this thread_pool ?
Thpool pool;
Following code will be inserted in required functions. I am showing for the TaskRPCB type task only.
std::function<void(void*,int,uint,int)> func = NFVInstance->CallBackTable[channel];
/* create a task object : members of it :
- the callback function
- msgBlob
- len of the msgBlob
- some other ID
- stream ID on which this message was received
*/
TaskRPCB task(func,msg,rc,fdd.id,streamId);
/* submit this task to the pool.
one of the waiting threads will pick this task
*/
pool.submit(task);
std::variantrequires that). I hope you get good reviews! \$\endgroup\$THREAD_POOL_SIZE 100I hope you are running on some large hardware. The point os a thread pool is that you create a thread for each available processor (or very close to that) then each processor gets new work from the pool without having to switch threads (as that is a relatively expensive operation). \$\endgroup\$