c++ - 如何终止线程池中的所有预分配线程？

Question

我已经使用以下结构创建了一个线程池，现在的问题是如何让所有预分配线程正确结束？

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;
}

我一直在为这个问题寻找几天仍然找不到像样的解决方案，我尝试过的最接近的是，当程序要退出时，将一个特殊项目插入队列，然后在threadFunctionUsedByThreadsPool中，当检测到此类项目时，我将调用 pthread_join，但是，当我使用 gdb 工具调试它时，那些预先分配的线程仍然存在，任何人都可以提供帮助，更好地使用一些代码，例如，如何修改 threadFunctionUsedByThreadsPool，以便我可以退出所有正确预分配线程？
非常感谢!!!

Answer 1

TLDR:您只需要一个线程安全变量，所有线程都可以在工作项之间检查退出条件。使用 pthread_join等待线程退出。
首先，让我们在你的线程函数中的 while 循环就条件变量而言是正确的。
取而代之的是:

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

在条件变量上检查唤醒前后队列的状态。虚假唤醒是真实存在的，并且不能保证另一个线程不会唤醒并获取最后一个工作项。您绝对不想从空队列中弹出。
更好的:

    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();

解决这个问题后，我们可以引入一个新的全局 bool 值来表示停止条件:

 bool stopCondition = false;

每当我们想告诉池中的所有线程停止时，我们可以设置 stopCondition至 true并通知条件变量以提醒所有线程状态更改。读或写 stopCondition 应该在锁下完成。 (我想你也可以使用 std::atomic<bool> )
把它们放在一起，你的线程函数变成了这样:

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;
}

然后是一个辅助函数来通知所有线程退出并等待每个线程停止。请注意，我们正在使用 pthread_cond_broadcast通知所有线程从它们的条件变量等待中唤醒，而不是 pthread_cond_signal只唤醒一个线程。

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();
}

我刚刚发现的最后一个错误 - 你的 main不是属性初始化您的 preAllocatedThreadsPool像你想的那样的 vector 。您正在制作 pthread_t 的拷贝，而不是在 vector 中实际使用句柄。
取而代之的是:

for(pthread_t i : preallocatedThreadsPool) {

您的循环需要通过引用枚举:
更好的:

for(pthread_t &i : preallocatedThreadsPool) {