c++ - pthread_cond_wait 有时会收不到信号

Question

pthread_cond_wait 和 pthread_cond_signal 有一个奇怪的问题。我已经安排了一系列线程。启动时都处于休眠状态。唤醒函数将向这些线程发出信号、做一些工作并等待结果。

在下面的设置中，td 是线程数据，包含互斥量和条件，th 是包含线程指针的数组:

for (size_t i = 0; i < NUM_THREADS; i++) {
    pthread_cond_init(&td[i].cond, NULL);
    pthread_mutex_init(&td[i].cond_mutex, NULL);
    pthread_mutex_init(&td[i].work_mutex, NULL);
    pthread_mutex_lock(&td[i].cond_mutex);
    pthread_mutex_lock(&td[i].work_mutex);
    pthread_create(&th[i], NULL, thread_worker, (void *)&td[i]);
}

Thread worker是这样的:

void*
thread_worker(void* data)
{
    THREAD_DATA *td = (THREAD_DATA *)data;
    while (1) {
        pthread_cond_wait(&td->cond, &td->cond_mutex);  // marker

        // do work ...

        pthread_mutex_unlock(&td->work_mutex);
    }
    pthread_exit(NULL);
}

这个 job 函数应该唤醒所有线程，完成工作，并等待它们完成:

void
job()
{
    for (size_t i = 0; i < NUM_THREADS; i++) {
        pthread_cond_signal(&td[i].cond);
    }
    for (size_t i = 0; i < NUM_THREADS; i++) {
        pthread_mutex_lock(&td[i].work_mutex);  // block until the work is done
    }
}

在极少数情况下(可能是 1000 次运行中的 1 次)，上述设置会遇到卡住。发生这种情况时，thread_worker 中的“标记”行将不会被 pthread_cond_signal 发出信号，它只是继续等待。这种情况非常罕见，但时有发生。我生成了大量日志消息，并且我验证了 pthread_cond_wait 总是在 pthread_cond_signal 之前被调用。我在这里做错了什么？

Answer 1

没有任何东西可以强制 pthread_cond_wait() 在 pthread_cond_signal() 之前被调用。不管你怎么说日志记录，记录的行完全有可能与实际发生的事情顺序不一致。

您没有正确使用互斥锁和条件变量:互斥锁只能由锁定它们的同一个线程解锁，并且条件变量应该与对某些共享状态(称为谓词).共享状态应该受到传递给 pthread_cond_wait() 的互斥量的保护。

例如，您的示例可以重新设计以正确使用互斥锁和条件变量。首先在THREAD_DATA结构中添加一个int work_status，其中0表示线程正在等待工作，1 表示工作可用，2 表示工作已完成。

您似乎不需要在每个 THREAD_DATA 中使用两个互斥量，并且您不想在设置时将互斥量锁定在主线程中:

for (size_t i = 0; i < NUM_THREADS; i++) {
    pthread_cond_init(&td[i].cond, NULL);
    pthread_mutex_init(&td[i].cond_mutex, NULL);
    td[i].work_status = 0;
    pthread_create(&th[i], NULL, thread_worker, (void *)&td[i]);
}

使用条件变量让线程等待 work_status:

void*
thread_worker(void* data)
{
    THREAD_DATA *td = (THREAD_DATA *)data;

    while (1) {
        /* Wait for work to be available */
        pthread_mutex_lock(&td->cond_mutex);
        while (td->work_status != 1)
            pthread_cond_wait(&td->cond, &td->cond_mutex);
        pthread_mutex_unlock(&td->cond_mutex);

        // do work ...

        /* Tell main thread that the work has finished */
        pthread_mutex_lock(&td->cond_mutex);
        td->work_status = 2;
        pthread_cond_signal(&td->cond);
        pthread_mutex_unlock(&td->cond_mutex);
    }
    pthread_exit(NULL);
}

...并在 job() 中适本地设置并等待 work_status:

void
job()
{
    /* Tell threads that work is available */
    for (size_t i = 0; i < NUM_THREADS; i++) {
        pthread_mutex_lock(&td[i].cond_mutex);
        td[i].work_status = 1;
        pthread_cond_signal(&td[i].cond);
        pthread_mutex_unlock(&td[i].cond_mutex);
    }

    /* Wait for threads to signal work complete */
    for (size_t i = 0; i < NUM_THREADS; i++) {
        pthread_mutex_lock(&td[i].cond_mutex);
        while (td[i].work_status != 2)
            pthread_cond_wait(&td[i].cond, &td[i].cond_mutex);
        pthread_mutex_unlock(&td[i].cond_mutex);
    }
}