生产者-消费者问题：posix 互斥锁在使用条件变量时被锁定了两次？

Question

以下代码只是展示如何使用条件变量来同步线程（一个生产者和多个消费者）作为练习。 请参阅代码“usleep(100);”的行。当我评论这一行时，两个消费者线程似乎在生产者线程退出后同时锁定了互斥锁，然后等待“cond”，这引发了竞争条件。 但如果我取消注释它。一切顺利（似乎）。

我的问题是两个消费者线程如何同时锁定一个互斥锁？即使我不调用 usleep()，这个演示也应该可以工作？提前感谢您的时间。

这是在生产者的循环中移除usleep后的输出。请注意输出中的最后两个'lock'。

$ ./pthread_cond 
Producer 3062414192 beginning...
Producer locked mutex self:3062414192
Producer is creating work:1
Producer finished creating work:1
Producer unlock self:3062414192
Producer locked mutex self:3062414192
Producer is creating work:2
Producer finished creating work:2
Producer unlock self:3062414192
Producer locked mutex self:3062414192
Producer is creating work:3
Producer finished creating work:3
Producer unlock self:3062414192
Producer locked mutex self:3062414192
Producer is creating work:4
Producer finished creating work:4
Producer unlock self:3062414192
Producer 3062414192 exit after creating 4 works...
produce joined,but 4 work remained
Consumer 3070806896 beginning...
Consumer locked mutex self:3070806896
to wait on cond self:3070806896
Consumer 3079199600 beginning...
Consumer locked mutex self:3079199600
to wait on cond self:3079199600

已实现：

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

#define MAX_COUSUMER 2

#define TOTAL_WORK 4

int g_work_counter=0;

pthread_mutex_t mut = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;

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

    printf("Producer %lu beginning...\n",pthread_self());
    for(i=0;i<TOTAL_WORK;i++)
    {
        assert(pthread_mutex_lock(&mut)==0);
        printf("Producer locked mutex self:%lu\n",pthread_self());
        printf("Producer is creating work:%d\n",g_work_counter+1);
        g_work_counter++;
        printf("Producer finished creating work:%d\n",g_work_counter);
        pthread_cond_broadcast(&cond);
        assert(pthread_mutex_unlock(&mut)==0);
        printf("Producer unlock self:%lu\n",pthread_self());

        //usleep(100);
    }

    printf("Producer self:%lu exit after creating %d works...\n",pthread_self(),i);//counter starts from 0
    pthread_exit(NULL);
}

void *consumer_thread(void *arg)
{
    printf("Consumer %lu beginning...\n",pthread_self());
    //use pthread_cancel in main
    pthread_detach(pthread_self());

    while(1)
    {
        assert(pthread_mutex_lock(&mut)==0);
        printf("Consumer locked mutex self:%lu\n",pthread_self());
        printf("to wait on cond self:%lu\n",pthread_self());
        assert(pthread_cond_wait(&cond,&mut)==0);
        if(g_work_counter)
        {
            printf("Consumer %lu is performing work:%d\n",pthread_self(),g_work_counter);
            g_work_counter--;
            printf("Consumer %lu finished performing work:%d\n",pthread_self(),g_work_counter+1);
        }
        assert(pthread_mutex_unlock(&mut)==0);
        printf("Consumer unlock self:%lu\n",pthread_self());
    }

    //no output (pthread_cancel is called)
    printf("Consumer %lu exit...\n",pthread_self());
    pthread_exit(NULL);
}

int main(int argc,char* argv[])
{
    pthread_t producer;
    pthread_t consumers[MAX_COUSUMER];
    int i;

    for(i=0;i<MAX_COUSUMER;i++)
    {
        if(pthread_create(&consumers[i],NULL,consumer_thread,NULL)!=0)
        {
            printf("pthread_create failed for consumer_thread %d\n",i);
        }
    }

    pthread_create(&producer,NULL,producer_thread,NULL);

    if(pthread_join(producer,NULL)!=0)
    {
        printf("pthread_join failed for producer_thread %lu\n",consumers[i]);
    }
    printf("producer joined,but %d work remained\n",g_work_counter);

    //wait for the consumers
    while(g_work_counter>0)
        ;

    //cancel the consumer,for they are detached
    for(i=0;i<MAX_COUSUMER;i++)
    {
        if(pthread_cancel(consumers[i])!=0)
        {
            printf("pthread_cancel failed for consumer_thread %d\n",i);
        }
    }

    pthread_mutex_destroy(&mut);
    pthread_cond_destroy(&cond);
    return 0;
}

Answer 1

当线程等待某个条件时，它释放锁。当它被唤醒时，它会重新获得它。在这段代码中，消费者只应在缓冲区为空时等待。

另一个问题在于main，实际上，这行：while(g_work_counter>0)。那时你没有锁，所以检查g_work_counter 是不安全的。我也不太确定pthread_detach(pthread_self());。 main 不应该在它自己的孩子上调用它吗？

一般来说，如果你想检查死锁或者调试你的 pthreads 代码，你应该使用pthread_mutexattr_foo 函数来设置一个错误检查互斥锁，并且检查返回值不仅仅是一个@ 987654327@.

Answer 2

正如Borealid 所说，当线程等待条件变量时，锁被释放。您的消费者函数应如下所示：

    /* Wait for some work to be available */
    pthread_mutex_lock(&mut);
    while (g_work_counter == 0)
    {
        pthread_cond_wait(&cond, &mut);
    }

    /* Have lock, g_work_counter is > 0, so do work */
    while (g_work_counter > 0)
    {
        g_work_counter--;
    }

    pthread_mutex_unlock(&mut);

（pthread_cond_wait() 应该始终在这样的旋转 while 循环中使用）。

Answer 3

我建议您不要使用 pthread_cancel() API。在不引入资源泄漏、死锁和不一致的情况下使用 pthread_cancel() 非常困难 在你的程序中。

作为重击规则： 将分离的线程用于“一次性/我不关心结果/完成某事”之类的事情。 对于其他事情，使用以合作方式关闭的“普通线程”（通过检查 一个标志）。

因此，在您的示例中，您希望在退出主程序之前完成所有工作。 不要将消费者创建为分离的并替换取消所有 带有循环的消费者为所有消费者执行 pthread_join()。