how to terminates all the preallocated threads in a threadpool?

user_7189286 0
Viewed 145

I have used below structure to create a threadpool, now the question is how do I let all the preallocate threads end properly?

std::vector<pthread_t> preallocatedThreadsPool; // threadpool
std::queue<int> tcpQueue;  // a queue to hold my task

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t condition_var = PTHREAD_COND_INITIALIZER;

void* threadFunctionUsedByThreadsPool(void *arg);

main () {
    preallocatedThreadsPool.resize(preallocatThreadsNumber);
    for(pthread_t i : preallocatedThreadsPool) {
        pthread_create(&i, NULL, threadFunctionUsedByThreadsPool, NULL);
    }


    pthread_mutex_lock(&mutex); // one thread mess with the queue at one time
 
    tcpQueue.push(task);

    pthread_cond_signal(&condition_var);
    pthread_mutex_unlock(&mutex);

}


void* threadFunctionUsedByThreadsPool(void *arg) {
    while (true) {
        pthread_mutex_lock(&mutex);
        if (tcpQueue.empty()) {  // can't get work from the queue then just wait
        pthread_cond_wait(&condition_var, &mutex); // wait for the signal from other thread to deal with client otherwise sleep
        task = tcpQueue.front();
        tcpQueue.pop(); 
        }
        pthread_mutex_unlock(&mutex);
    
        if (task) {
            // do task
        }
    }
    return NULL;
}

I have been searching for days for this problem still can not find a decent solution, the closest one I have tried is , when the program wants to quit, push a special item into the queue, then inside threadFunctionUsedByThreadsPool, when detecting such item, I will call pthread_join, however, when I using gdb tool to debug it , those pre-allocated threads are still there, anyone could help, better with some code, for example, how do I modify the threadFunctionUsedByThreadsPool, so that I can quit all the pre-allocated threads properly? Thanks so much!!!

server threadpool pthreads multithreading c++
2 Answers

TLDR: You just need a thread-safe variable that all threads can check for an exit condition in between work items. Use pthread_join to wait for a thread to exit.

First, let's get the while loop in your thread function correct with respect to condition variables.

Instead of this:

    pthread_cond_wait(&condition_var, &mutex); // wait for the signal from other thread to deal with client otherwise sleep
    task = tcpQueue.front();
    tcpQueue.pop();

Check the state of the queue before before and after waking up on the condition variable. Spurious wake up is a real thing and there's no guarantee another thread didn't wake up and grab the last work item. You definitely don't want to be popping from an empty queue.

Better:

    while (tcpQueue.empty()) {  
        pthread_cond_wait(&condition_var, &mutex); // wait for the signal from other thread to deal with client otherwise sleep
    }
    task = tcpQueue.front();
    tcpQueue.pop();

With that addressed, we can introduce a new global bool that represents the stop condition:

 bool stopCondition = false;

Whenever we want to tell all the threads in the pool to stop, we can set stopCondition to true and signal the condition var to alert all threads of a state change. Reading or writing stopCondition should be done under a lock. (I suppose you could also use std::atomic<bool>)

Putting it all together, your thread function becomes this:

void* threadFunctionUsedByThreadsPool(void* arg) {

    pthread_mutex_lock(&mutex);

    while (!stopCondition) {

        // wait for a task to be queued
        while (tcpQueue.empty() && !stopCondition) {  
            pthread_cond_wait(&condition_var, &mutex); // wait for the signal from other thread to deal with client otherwise sleep
        }

        if (stopCondition == false) {
            task = tcpQueue.front();
            tcpQueue.pop();

            // exit lock while operating on a task
            pthread_mutex_unlock(&mutex);

            if (task) {
                // do task
            }

            // re-acquire the lock
            pthread_mutex_lock(&mutex);

        }
 
    }

    // release the lock before exiting the function
    pthread_mutex_unlock(&mutex);
    return NULL;
}

And then a helper function to signal all the threads to exit and also waits for each thread to stop. notice that we're using pthread_cond_broadcast to notify all threads to wake up from their condition variable wait instead of pthread_cond_signal which only wakes up one thread.

void stopThreadPool()
{

    // signal all threads to exit after they finish their current work item
    pthread_mutex_lock(&mutex);
        stopCondition = true;
        pthread_cond_broadcast(&condition_var); // notify all threads
    pthread_mutex_unlock(&mutex);

    // wait for all threads to exit
    for (auto& t : preAllocatedThreadsPool) {
        pthread_join(t, nullptr);
    }
    preAllocatedThreadsPool.clear();
}

One last bug that I just caught - your main isn't property initializing your preAllocatedThreadsPool vector like you think it is. You're making a copy of the pthread_t, instead of using the handle actually in the vector.

Instead of this:

for(pthread_t i : preallocatedThreadsPool) {

Your loop needs to enumerate by reference:

Better:

for(pthread_t &i : preallocatedThreadsPool) {
user_1044580

Send a task that instructs the pool thread to requeue the task and then terminate. The poison-task will then run through all the threads in the pool, killing them all off. I have used a null as the poison, (ie, an illegal task) - it does not need to be destructed when it has killed the last thread. You may wish to purge the task queue before sending the null/whatever. If you use a null, you only need a null check in the threads, just after dequeing the task.

You need very little extra code, you don't need to know how many threads in the pool and it will work:)

user_7581330
Related