多生产者-消费者执行的效率

Question

我试着和多个生产者和消费者一起做一个代码。我为生产者和使用者创建了多线程，并使用信号量进行同步。该代码在单个生产者和消费者中运行良好。

我面临的问题是，经过一段时间的程序执行之后，只有consumer1和producer1参与了这个过程。我无法理解其他生产者和消费者发生了什么。

我也想知道如何使多生产者-消费者问题的效率？在所有生产者和消费者分别获得平等的生产和消费机会的意义上的效率？C++代码(它包含大量的C)：

#include <iostream>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>
#include <queue>
using namespace std;
sem_t empty;
sem_t full;
int cnt = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
queue<int> q;
void *producer(void *a)
{   
    int *num = (int *)a;
    while(1) {
        sem_wait(&empty);
        pthread_mutex_lock(&mutex);
        cnt = cnt+1;
        q.push(cnt);
        cout<<cnt<<" item produced by producer "<<(*num+1)<<endl;
        pthread_mutex_unlock(&mutex);
        sem_post(&full);
        sleep(1);
    }
}
void *consumer(void *a)
{   
    int *num = (int *)a;
    while(1) {
        sem_wait(&full);
        pthread_mutex_lock(&mutex);
        cout<<q.front()<<" item consumed by consumer "<<(*num+1)<<endl;
        q.pop();
        pthread_mutex_unlock(&mutex);
        sem_post(&empty);
        sleep(1);
    }
}
int main()
{   
    pthread_t p[5];
    pthread_t c[5];
    sem_init(&empty,0,5);
    sem_init(&full,0,0);
    int i;
    for(i = 0; i < 5; i++) {
        pthread_create(&p[i],NULL,producer,(void *)(&i));
    }
    for(i = 0; i < 5; i++) {
        pthread_create(&c[i],NULL,consumer,(void *)(&i));
    }
    for(i = 0; i < 5; i++) {
        pthread_join(p[i],NULL);
        pthread_join(c[i],NULL);
    }
}

更新代码：

#include <iostream>
#include <pthread.h>
#include <semaphore.h>
#include <unistd.h>
#include <queue>
#include <map>
using namespace std;
sem_t empty;
sem_t full;
int cnt = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
map<pthread_t,int> mc,mp;
queue<int> q;
void *producer(void *a)
{   
    while(1) {
        sem_wait(&empty);
        pthread_mutex_lock(&mutex);
        cnt = cnt+1;
        q.push(cnt);
        cout<<cnt<<" item produced by producer "<<mp[pthread_self()]<<endl;
        pthread_mutex_unlock(&mutex);
        sem_post(&full);
        sleep(1);
    }
}
void *consumer(void *a)
{   
    while(1) {
        sem_wait(&full);
        pthread_mutex_lock(&mutex);
        cout<<q.front()<<" item consumed by consumer "<<mc[pthread_self()]<<endl;
        q.pop();
        pthread_mutex_unlock(&mutex);
        sem_post(&empty);
        sleep(1);
    }
}
int main()
{   
    pthread_t p[5];
    pthread_t c[5];
    sem_init(&empty,0,5);
    sem_init(&full,0,0);
    int i;
    pthread_mutex_lock(&mutex);
    for(i = 0; i < 5; i++) {
        pthread_create(&p[i],NULL,producer,NULL);
        pthread_create(&c[i],NULL,consumer,NULL);
        mc[c[i]] = i+1;
        mp[p[i]] = i+1; 
    }
    pthread_mutex_unlock(&mutex);
    for(i = 0; i < 5; i++) {
        pthread_join(p[i],NULL);
        pthread_join(c[i],NULL);
    }
}

Answer 1

短答案

线程实际上是以平等的机会执行的，但它们只是打印出一个不属于它们的标识符。

详细解释

在每个线程中都保留一个指向线程号的指针num。它是指向保存的值的指针，而不是值本身。所以所有的线程都指向同一个计数器，想在那里找到自己的标识符。

每次访问*num时，您都不能访问i在启动线程时拥有的值，而是访问它的当前值。

不幸的是，在main()的每个循环中，您都重用变量i。最后一个循环，您将i设置为0，并等待第一个线程加入。但是所有这些线程都会永远循环，所以这个循环很难超过这个初始的0值。所以每个线程都认为这个数字*num+1是1。

注意，正如有人在注释中指出的那样，您创建了一个争用条件:所有使用者和生产者线程取消对指针的引用，访问互斥保护区域中的相同变量。这样没问题。但是，在读取变量时，主线程仍然可以愉快地更改任何锁之外的共享变量。这无疑是一种种族风险。

解决方案

std::thread将允许您通过walue传递i，这样每个线程都有自己的未更改的id副本。

对于线程，您必须传递一个指向值的指针。不幸的是，即使您在线程开始时执行指向值的本地副本，您仍然处于争用状态。

观察哪个线程真正在执行这项工作的一个快速解决方法是打印出pthread_self()的结果(参见here如何完成)。或者将ids存储在int数组中，并将地址传递给每个线程，将地址传递给该数组中的唯一元素。