unix 多进程下的生产者和消费者实现

#include<iostream.h>
#include<pthread.h>
#include<stdio.h>
#include<stdlib.h>
#include <sys/stat.h>
#include<time.h>
#include<unistd.h>
#include <semaphore.h>
#include<sys/mman.h>

typedef struct
{
int buf[129];
int occupied;
int nextin;
int nextout;
pthread_mutex_t mutex;
pthread_cond_t more;
pthread_cond_t less;
} buffer_t;

void producer(buffer_t *b,char c)
{
pthread_mutex_lock(&(b->mutex));

if(b->occupied > 128)
{
pthread_cond_wait(&(b->less),&(b->mutex));
}

b->buf[b->nextin++]=c;
b->nextin=b->nextin%128;
b->occupied++;

pthread_cond_signal(&(b->more));
pthread_mutex_unlock(&(b->mutex));
}

void producer_driver(buffer_t *b)
{
char item;
while (1)
{
item = getchar();
if (item == EOF)
{
producer(b,'/0');
break;
}
else
{
producer(b, (char)item);
sleep(0.5);
}
}
}

char consumer(buffer_t *b)
{
char c;
pthread_mutex_lock(&(b->mutex));
if(b->occupied <= 0)
{
pthread_cond_wait(&(b->more),&(b->mutex));
}

c=b->buf[b->nextout++];
b->nextout=b->nextout%128;
b->occupied--;

pthread_cond_signal(&(b->less));
pthread_mutex_unlock(&(b->mutex));

return c;
}

void consumer_driver(buffer_t *b)
{
char item;
while (1)
{
if ((item = consumer(b))=='/0')
{
break;
}

putchar(item);
sleep(1);
//item = consumer(b);
//cout<<item<<endl;
//cout.flush();
}
}

int main()
{
int zfd;
buffer_t *buffer;
pthread_mutexattr_t mattr;
pthread_condattr_t cvattr_less, cvattr_more;

zfd = open("/dev/zero", O_RDWR);
buffer = (buffer_t *)mmap(NULL, sizeof(buffer_t),
PROT_READ|PROT_WRITE, MAP_SHARED, zfd, 0);

buffer->occupied = buffer->nextin = buffer->nextout = 0;

pthread_mutexattr_init(&mattr);
pthread_mutexattr_setpshared(&mattr,PTHREAD_PROCESS_SHARED);
pthread_mutex_init(&buffer->mutex, &mattr);

pthread_condattr_init(&cvattr_less);
pthread_condattr_setpshared(&cvattr_less, PTHREAD_PROCESS_SHARED);
pthread_cond_init(&buffer->less, &cvattr_less);

pthread_condattr_init(&cvattr_more);
pthread_condattr_setpshared(&cvattr_more,PTHREAD_PROCESS_SHARED);
pthread_cond_init(&buffer->more, &cvattr_more);

if (fork() == 0)
{
consumer_driver(buffer);
}
else
{
producer_driver(buffer);
}

sleep(120);
exit(0);
}