在线程中，如何可靠地将信号传递给另一个线程？

Question

我试图用p线程编写一个简单的线程池程序。然而，pthread_cond_signal似乎没有阻塞，这就产生了一个问题。例如，假设我有一个“生产者-消费者”计划：

pthread_cond_t my_cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t my_cond_m = PTHREAD_MUTEX_INITIALIZER;

void * liberator(void * arg)
{
    // XXX make sure he is ready to be freed
    sleep(1);

    pthread_mutex_lock(&my_cond_m);
    pthread_cond_signal(&my_cond);
    pthread_mutex_unlock(&my_cond_m);

    return NULL;
}

int main()
{
    pthread_t t1;
    pthread_create(&t1, NULL, liberator, NULL);

    // XXX Don't take too long to get ready. Otherwise I'll miss 
    // the wake up call forever
    //sleep(3);

    pthread_mutex_lock(&my_cond_m);
    pthread_cond_wait(&my_cond, &my_cond_m);
    pthread_mutex_unlock(&my_cond_m);

    pthread_join(t1, NULL);

    return 0;
}

正如在两个XXX标记中所描述的，如果我去掉sleep调用，那么main()可能会暂停，因为它错过了liberator()的唤醒呼叫。当然，sleep也不是一个非常健壮的方法来确保这一点。

在现实生活中，这将是一个工作线程，它告诉经理线程它已经准备好工作，或者经理线程宣布新工作是可用的。

您如何在p线程中可靠地做到这一点？

精化

@Borealid的回答有点见效，但他对这个问题的解释可能会更好。我建议任何看这个问题的人阅读评论中的讨论内容，了解正在发生的事情。

特别是，我自己会修改他的答案和代码示例如下，以使这一点更清楚。(自从Borealid最初的答案，在编译和工作时，我非常困惑)

// In main
pthread_mutex_lock(&my_cond_m);

// If the flag is not set, it means liberator has not 
// been run yet. I'll wait for him through pthread's signaling 
// mechanism

// If it _is_ set, it means liberator has been run. I'll simply 
// skip waiting since I've already synchronized. I don't need to 
// use pthread's signaling mechanism
if(!flag) pthread_cond_wait(&my_cond, &my_cond_m);

pthread_mutex_unlock(&my_cond_m);

// In liberator thread
pthread_mutex_lock(&my_cond_m);

// Signal anyone who's sleeping. If no one is sleeping yet, 
// they should check this flag which indicates I have already 
// sent the signal. This is needed because pthread's signals 
// is not like a message queue -- a sent signal is lost if 
// nobody's waiting for a condition when it's sent.
// You can think of this flag as a "persistent" signal
flag = 1;
pthread_cond_signal(&my_cond);
pthread_mutex_unlock(&my_cond_m);

Answer 1

使用同步变量。

在main中

pthread_mutex_lock(&my_cond_m);
while (!flag) {
    pthread_cond_wait(&my_cond, &my_cond_m);
}
pthread_mutex_unlock(&my_cond_m);

在线程中：

pthread_mutex_lock(&my_cond_m);
flag = 1;
pthread_cond_broadcast(&my_cond);
pthread_mutex_unlock(&my_cond_m);

对于生产者-消费者问题，这将是消费者睡觉时，缓冲区是空的，生产者睡觉时，它是满的。记住在访问全局变量之前获得锁。

Answer 2

我找到了解决方案这里。对我来说，理解这个问题的棘手之处在于：

1. 生产者和消费者必须能够双向沟通。无论哪种方式都是不够的。
2. 这种双向通信可以打包成一个线程条件.

为了说明，上面提到的博客文章证明了这实际上是有意义和可取的行为：

pthread_mutex_lock(&cond_mutex);
pthread_cond_broadcast(&cond):
pthread_cond_wait(&cond, &cond_mutex);
pthread_mutex_unlock(&cond_mutex);

这样的想法是，如果生产者和消费者都采用这种逻辑，他们中的任何一方都可以先睡觉，因为每个人都能唤醒另一个角色。换句话说，在一个典型的生产者--消费者的角度--如果一个消费者需要睡觉，那是因为一个生产者需要醒来，反之亦然。将此逻辑封装在单个p线程条件下是有意义的。

当然，上面的代码有一种意想不到的行为，即当工作线程实际上只想唤醒生产者时，它也会唤醒另一个休眠的工作线程。这可以通过@Borealid建议的简单变量检查来解决：

while(!work_available) pthread_cond_wait(&cond, &cond_mutex);

在worker广播时，所有工作线程都将被唤醒，但是一个接一个地被唤醒(因为pthread_cond_wait中隐含的互斥锁)。由于其中一个辅助线程将消耗该工作(将work_available设置为false)，当其他工作线程苏醒并实际开始工作时，该工作将不可用，因此工作人员将再次睡眠。

下面是一些我测试过的注释代码，供感兴趣的人使用：

// gcc -Wall -pthread threads.c -lpthread

#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <assert.h>

pthread_cond_t my_cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t my_cond_m = PTHREAD_MUTEX_INITIALIZER;

int * next_work = NULL;
int all_work_done = 0;

void * worker(void * arg)
{
    int * my_work = NULL;

    while(!all_work_done)
    {
        pthread_mutex_lock(&my_cond_m);

        if(next_work == NULL)
        {
            // Signal producer to give work
            pthread_cond_broadcast(&my_cond);

            // Wait for work to arrive
            // It is wrapped in a while loop because the condition 
            // might be triggered by another worker thread intended 
            // to wake up the producer
            while(!next_work && !all_work_done)
                pthread_cond_wait(&my_cond, &my_cond_m);
        }

        // Work has arrived, cache it locally so producer can 
        // put in next work ASAP
        my_work = next_work;
        next_work = NULL;
        pthread_mutex_unlock(&my_cond_m);

        if(my_work)
        {
            printf("Worker %d consuming work: %d\n", (int)(pthread_self() % 100), *my_work);
            free(my_work);
        }
    }

    return NULL;
}

int * create_work()
{
    int * ret = (int *)malloc(sizeof(int));
    assert(ret);
    *ret = rand() % 100;
    return ret;
}

void * producer(void * arg)
{
    int i;

    for(i = 0; i < 10; i++)
    {
        pthread_mutex_lock(&my_cond_m);
        while(next_work != NULL)
        {
            // There's still work, signal a worker to pick it up
            pthread_cond_broadcast(&my_cond);

            // Wait for work to be picked up
            pthread_cond_wait(&my_cond, &my_cond_m);
        }

        // No work is available now, let's put work on the queue
        next_work = create_work();
        printf("Producer: Created work %d\n", *next_work);

        pthread_mutex_unlock(&my_cond_m);
    }

    // Some workers might still be waiting, release them
    pthread_cond_broadcast(&my_cond);

    all_work_done = 1;
    return NULL;
}

int main()
{
    pthread_t t1, t2, t3, t4;

    pthread_create(&t1, NULL, worker, NULL);
    pthread_create(&t2, NULL, worker, NULL);
    pthread_create(&t3, NULL, worker, NULL);
    pthread_create(&t4, NULL, worker, NULL);

    producer(NULL);

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);
    return 0;
}