信号量解决生产者-消费者问题

#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<semaphore.h>
#include<pthread.h>

#define NBUF 10
struct
{
int buf[NBUF];
sem_t mutex,nempty,nstored;
}shared;

void* produce(void*);
void* consume(void*);

int nitems;

int main(int argc,char** argv)
{
pthread_t tid_produce,tid_consume;
if(argc!=2)
{
fprintf(stderr,"usage:prodcons1<#items>\n");
exit(1);
}

nitems=atoi(argv[1]);

sem_init(&shared.mutex,0,1);
sem_init(&shared.nempty,0,NBUF);
sem_init(&shared.nstored,0,0);

pthread_create(&tid_produce,NULL,produce,NULL);
pthread_create(&tid_consume,NULL,consume,NULL);

pthread_join(tid_produce,NULL);
pthread_join(tid_consume,NULL);

sem_destroy(&shared.mutex);
sem_destroy(&shared.nempty);
sem_destroy(&shared.nstored);

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

#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<semaphore.h>
#include<pthread.h>

#define NBUF 10
#define MAXTHREADS 10

struct
{
int buf[NBUF];
int count;       //生产者当前生产的下标
sem_t mutex,nempty,nstored;
}shared;

void* produce(void*);
void* consume(void*);

int nitems;
int nproducers;

int main(int argc,char** argv)
{
pthread_t tid_produce[MAXTHREADS];
pthread_t tid_consume;
int i;

if(argc!=3)
{
fprintf(stderr,"usage:prodcons2<#items>\n");
exit(1);
}

nitems=atoi(argv[1]);
nproducers=atoi(argv[2])>MAXTHREADS?MAXTHREADS:atoi(argv[2]);

sem_init(&shared.mutex,0,1);
sem_init(&shared.nempty,0,NBUF);
sem_init(&shared.nstored,0,0);
for(i=0;i<nproducers;i++)
{
pthread_create(&tid_produce[i],NULL,produce,NULL);
}
pthread_create(&tid_consume,NULL,consume,NULL);
for(i=0;i<nproducers;i++)
{
pthread_join(tid_produce[i],NULL);
}

pthread_join(tid_consume,NULL);

sem_destroy(&shared.mutex);
sem_destroy(&shared.nempty);
sem_destroy(&shared.nstored);

exit(0);
}
void* produce(void* arg)
{
int i;
for(;;)
{
sem_wait(&shared.nempty);
sem_wait(&shared.mutex);
if(shared.count>=nitems)   //当生产者完成生产时，互斥量+1，退出
{
sem_post(&shared.mutex);
sem_post(&shared.nempty);
return NULL;
}
shared.buf[shared.count%NBUF]=shared.count;
shared.count++;
sem_post(&shared.mutex);
sem_post(&shared.nstored);
}
return NULL;
}
void* consume(void* arg)
{
int i;
for(i=0;i<nitems;i++)
{
sem_wait(&shared.nstored);
sem_wait(&shared.mutex);
if(shared.buf[i%NBUF]!=i)
{
printf("buf[%d]=%d\n",i,shared.buf[i%NBUF]);
}
sem_post(&shared.mutex);
sem_post(&shared.nempty);
}
return NULL;
}

#include<stdio.h>
#include<stdlib.h>
#include<unistd.h>
#include<semaphore.h>
#include<pthread.h>

#define NBUF 10
#define MAXTHREADS 10

struct
{
int buf[NBUF];
int count;      //生产者生产下标
int concount;   //消费者消费下标
sem_t mutex,nempty,nstored;
}shared;

void* produce(void*);
void* consume(void*);

int nitems;
int nproducers;
int nconsumers;

int main(int argc,char** argv)
{
pthread_t tid_produce[MAXTHREADS];
pthread_t tid_consume[MAXTHREADS];
int i;

if(argc!=4)
{
fprintf(stderr,"usage:prodcons2<#items>\n");
exit(1);
}

nitems=atoi(argv[1]);
nproducers=atoi(argv[2])>MAXTHREADS?MAXTHREADS:atoi(argv[2]);
nconsumers=atoi(argv[3])>MAXTHREADS?MAXTHREADS:atoi(argv[3]);

sem_init(&shared.mutex,0,1);
sem_init(&shared.nempty,0,NBUF);
sem_init(&shared.nstored,0,0);

for(i=0;i<nproducers;i++)
{
pthread_create(&tid_produce[i],NULL,produce,NULL);
}
for(i=0;i<nconsumers;i++)
{
pthread_create(&tid_consume[i],NULL,consume,NULL);
}
for(i=0;i<nproducers;i++)
{
pthread_join(tid_produce[i],NULL);
}
for(i=0;i<nconsumers;i++)
{
pthread_join(tid_consume[i],NULL);
}
sem_destroy(&shared.mutex);
sem_destroy(&shared.nempty);
sem_destroy(&shared.nstored);
exit(0);
}
void* produce(void* arg)
{
for(;;)
{
sem_wait(&shared.nempty);
sem_wait(&shared.mutex);
if(shared.count>=nitems)
{
sem_post(&shared.nstored);     /*这里多了一个sem_post(shared.nstored);
我们让每个生产者给nstored信号量加1以给每个消费者线程解除阻塞，
让每个消费者线程能进入if()语句，从而退出线程。*/
sem_post(&shared.mutex);
sem_post(&shared.nempty);
return NULL;
}
shared.buf[shared.count%NBUF]=shared.count;
shared.count++;
sem_post(&shared.mutex);
sem_post(&shared.nstored);
}
return NULL;
}
void* consume(void* arg)
{
for(;;)
{
sem_wait(&shared.nstored);
sem_wait(&shared.mutex);
if(shared.concount>=nitems)
{
sem_post(&shared.mutex);
sem_post(&shared.nstored);
return NULL;
}
if(shared.buf[shared.concount%NBUF]!=shared.concount)
{
printf("buf[%d]=%d\n",shared.concount,shared.buf[shared.concount%NBUF]);
}
shared.concount++;
sem_post(&shared.mutex);
sem_post(&shared.nempty);
}
return NULL;
}