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

Question

我试图编写一个包含多个生产者和消费者的代码。我为生产者和消费者创建了多线程，并使用信号量进行同步。该代码适用于单个生产者和消费者。

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

我也想知道如何使多生产者消费者问题有效？从所有生产者和消费者分别获得平等的生产和消费机会的意义上说是有效的吗？ 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。

请注意，您创建竞争条件的方式正如有人在 cmets 中指出的那样：所有消费者和生产者线程取消引用指针，访问受互斥体保护的区域中的相同变量。还行吧。但是当他们读取变量时，主线程仍然可以很高兴地在任何锁之外更改共享变量。这绝对是种族风险。

解决方法

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

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

观察哪个线程真正在工作的一种快速解决方法是打印输出pthread_self() 的结果（请参阅here 怎么做）。或者将 id 存储在一个 int 数组中，并将地址传递给每个线程到该数组中的唯一元素。