posix 信号量---生产者消费者（二）

多生产者单消费者

#include "../unipc.h"

#define NBUFF 10
#define MAX_PRODUCE 100
#define FILE_MODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH)
int nitems;
int nproducers;
struct {
    int index;
    int buff[NBUFF];
    sem_t mutex,nempty,nstored;
} shared;

void *produce(void *arg)
{
    for(;;) {
        sem_wait(&shared.nempty);
        sem_wait(&shared.mutex);
        if(shared.index >= nitems){
            sem_post(&shared.nempty);
            sem_post(&shared.mutex);
            return NULL;
        }
        shared.buff[shared.index %NBUFF] = shared.index;
        shared.index++;
        sem_post(&shared.mutex);
        sem_post(&shared.nstored);
        (*(int*)arg)++;
    }
}

void *consume(void *arg)
{
    int i;
    for(i = 0;i < nitems; i++) {
        sem_wait(&shared.nstored);
        sem_wait(&shared.mutex);
        if(i != shared.buff[i%NBUFF]) {
            printf("i = %d,buff[%d]=%d\n",i,i,shared.buff[i]);
        }
        sem_post(&shared.mutex);
        sem_post(&shared.nempty);
    }
    return NULL;
}

int main(int argc ,char *argv[])
{
    int i;
    int count[MAX_PRODUCE];
    pthread_t ptid_produce[MAX_PRODUCE],ptid_consume;

    if(argc != 3) {
        printf("usage produce3 <#produces> <#items>\n");
        return -1;
    }
    nproducers = atoi(argv[1]);
    nitems = atoi(argv[2]);
    if(nproducers > MAX_PRODUCE)
        nproducers = MAX_PRODUCE;
    printf("nproducers = %d,nitems=%d\n",nproducers,nitems);
    sem_init(&shared.mutex,0,1);
    sem_init(&shared.nempty,0,NBUFF);
    sem_init(&shared.nstored,0,0);
    shared.index = 0;

    for(i = 0;i < nproducers; i++) {
        count[i] = 0;
        pthread_create(&ptid_produce[i],NULL,produce,&count[i]);
    }
    pthread_create(&ptid_consume,NULL,consume,NULL);

    for(i = 0;i < nproducers; i++) {
        pthread_join(ptid_produce[i],NULL);
        printf("count[%d] = %d\n",i,count[i]);
    }
    pthread_join(ptid_consume,NULL);

    sem_destroy(&shared.mutex);
    sem_destroy(&shared.nempty);
    sem_destroy(&shared.nstored);
    return 0;
}

生产者线程的终止

当最后一个条目生产出来后，每个生产者线程都执行循环顶端的语句：

sem_wait(&shared.nempty);

它将nempty信号量减1。
1.）然而每个生产者线程在终止前必须给该信号量加1（sem_post(&shared.nempty)），因为它在最后一次走过循环时并没有往缓存区存入一个条目。要是线程终止时我们没有给nempty信号量加1，而且生产者线程数大于缓冲区槽位数（譬如说14个生产者和10个缓冲区槽位），那么多余的线程(4个)将永远阻塞，等待nempty信号量，从而永远终止不了。

if(shared.index >= nitems){
            //sem_post(&shared.nempty);
            sem_post(&shared.mutex);
            return NULL;
}

注释掉sem_post(&shared.nempty); 运行produce3 14 10000
发现程序阻塞起来了
[e-mail]:~/test/mutex/pxsem>gcc -g produces3.c -o produces3 -Wall -pthread
[e-mail]:~/test/mutex/pxsem>./produces3 14 100000                         
nproducers = 14,nitems=100000
count[0] = 6994
count[1] = 7056
count[2] = 7079
count[3] = 7460
-----阻塞
通过查看线程的命令查看，发现一共有5个线程存在着，应该是一个主线程main，另外4个是生产者线程。

[e-mail]:~/codes/ipv6_12_11>ps -eLf | grep pts/3
gwwu      8529 25419  8529  0    5 10:39 pts/3    00:00:00 ./produces3 14 100000
gwwu      8529 25419  8534  0    5 10:39 pts/3    00:00:00 ./produces3 14 100000
gwwu      8529 25419  8535  0    5 10:39 pts/3    00:00:00 ./produces3 14 100000
gwwu      8529 25419  8539  0    5 10:39 pts/3    00:00:00 ./produces3 14 100000
gwwu      8529 25419  8542  0    5 10:39 pts/3    00:00:00 ./produces3 14 100000
gwwu      8561 25756  8561  0    1 10:41 pts/7    00:00:00 grep pts/3
gwwu     25395 25393 25395  0    1 09:49 ?        00:00:00 sshd: gwwu@pts/3 
gwwu     25419 25395 25419  0    1 09:49 pts/3    00:00:00 -bash
[e-mail]:~/codes/ipv6_12_11>

用pstree 查看更加明显：
[e-mail]:~/codes/ipv6_12_11>ps -t pts/3             
  PID TTY          TIME CMD
 8529 pts/3    00:00:00 produces3
25419 pts/3    00:00:00 bash
[e-mail]:~/codes/ipv6_12_11>pstree -p 8529
produces3(8529)-+-{produces3}(8534)
                |-{produces3}(8535)
                |-{produces3}(8539)
                `-{produces3}(8542)

2.）假设只有一个生产者，当它用完所有NBUFF后，将阻塞在sem_wait(&shared.nempty),等待消费者消费完处理sem_post(&shared.nempty).当生产者完成所有nitem个数据的生成后，最终由于消费者没有来得急消费，使得该生产者阻塞于sem_wait(&shared.nempty)，等待最终消费者sem_post,然后进入if(shared.index >= nitems)，然后退出。