操作系统实验报告生产者消费者问题

1、操作系统实验报告2016年 1月 8日指导教师对实验报告的评语成绩： 指导教师签字： 年 月 日一、设计目标l 完成N个生产者和M个消费者线程之间的并发控制，N、M不低于30，数据发送和接收缓冲区尺寸不小于20个（每个产品占据一个）。l 其中生产者线程1、3、5、7、9生产的产品供所有奇数编号的消费者线程消费，只有所有奇数编号的消费者线程都消费后，该产品才能从缓冲区中撤销。l 其中生产者线程2、4、6、8、10生产的产品所有偶数编号的消费者线程都可消费，任一偶数编号消费者线程消费该消息后，该产品都可从缓冲区中撤销。l 其中11-20号生产者线程生产的产品仅供对应编号的消费者线程消费。l 其他编

2、号生产者线程生产的产品可由任意的消费者线程消费。l 每个生产线程生产30个消息后结束运行。如果一个消费者线程没有对应的生产者线程在运行后，也结束运行。所有生产者都停止生产后，如果消费者线程已经没有可供消费的产品，则也退出运行。二、背景知识说明原理利用进程间共享的信号量、互斥锁等控制线程同步。相关函数说明pthread_create()：创建一个线程pthread_join()：阻塞当前的线程，直到另外一个线程运行结束pthread_mutex_init()：初始化互斥锁pthread_mutex_lock()：占有互斥锁（阻塞操作）pthread_mutex_unlock()：释放互斥锁sem

3、_wait()：获取信号量sem_post()：释放信号量三、设计设计环境Linux操作系统（ubuntu12.04）gcc 4.6c语言Posix线程库编译命令：gcc consumer.c -lpthread -o consumer概要设计及详细设计图注释：(1)图：主函数流程(2)图：生产者流程(3)图：消费者流程 （1）图(2) 图（3）图重要代码注释：#include<stdio.h>#include<malloc.h>#include<pthread.h>#include<semaphore.h>#define BUFFER_SIZE

4、 30#define OVER (-1)struct Product int tid; int data;struct producers /定义生产者条件变量结构 struct Product bufferBUFFER_SIZE;/缓冲区 sem_t sem_read; / 读信号量 sem_t sem_write; / 写信号量 pthread_mutex_t wlock; / 缓冲区写锁 pthread_mutex_t rlock; / 缓冲区读锁 pthread_mutex_t lock; / thread_count的读写锁 int readpos , writepos;/读写位置;

5、struct producers buffer;int thread_count = 30; /存活生产者计数/用于在线程内部标识线程IDint ids30 = 1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30;int count = 0; / 计数消费产品数量int pcount = 0; / 计数生产产品数量void init() /初始化相关锁和变量以及信号量 buffer.readpos = 0; buffer.writepos = 0; /写信号量比缓冲区小1，防止缓冲区满

6、和缓冲区空分不清 sem_init(&buffer.sem_write, 0, BUFFER_SIZE-1); sem_init(&buffer.sem_read, 0, 0); pthread_mutex_init(&buffer.wlock, NULL); pthread_mutex_init(&buffer.rlock, NULL); pthread_mutex_init(&buffer.lock, NULL);void put(int tid, int data) /缓冲区中放入一个数据 sem_wait(&buffer.sem_writ

7、e); /生产前先加锁，已防止其他线程同时生产 pthread_mutex_lock(&buffer.wlock); buffer.bufferbuffer.writepos.tid = tid; buffer.bufferbuffer.writepos.data = data; buffer.writepos +; + pcount; if( buffer.writepos >= BUFFER_SIZE ) buffer.writepos = 0; pthread_mutex_unlock(&buffer.wlock); sem_post(&buffer.sem

8、_read);/读数据并移走struct Product * get(int tid) struct Product * produce = NULL; /消费前先上锁，以防止其他线程同时消费 pthread_mutex_lock(&buffer.rlock); / 如果生产者线程没有全部退出，或者缓冲区内仍有产品，则说明可以尝试去获取产品 if(thread_count > 0 | buffer.readpos != buffer.writepos) /从信号量的值减去一个“1”，但它永远会先等待该信号量为一个非零值才开始做减法 sem_wait( &buffer.se

9、m_read ); int pos = buffer.readpos; / 在已有产品中迭代，查找适合的产品 while(pos != buffer.writepos) int id = buffer.bufferpos.tid; if(id >10 && id <=20 && tid = id) /如果产品是的生产者线程号 10<id<=20 则可以供和它的线程号相同的消费者线程消费 produce = (struct Product *)malloc(sizeof(struct Product); produce->tid =

10、id; produce->data = buffer.bufferpos.data; break; else if(id <= 10 && (id%2 = tid%2) /如果产品是的生产者线程号 <=10 则可以供和它的线程号奇偶性相同的消费者消费 produce = (struct Product *)malloc(sizeof(struct Product); produce->tid = id; produce->data = buffer.bufferpos.data; break; else if(id > 20) /如果产品是的

11、生产者线程号>20则可以供任何消费者消费 produce = (struct Product *)malloc(sizeof(struct Product); produce->tid = id; produce->data = buffer.bufferpos.data; break; pos = (pos+1)%BUFFER_SIZE; if( produce ) /如果取得了产品，消费计数+1，将在队头的元素覆盖到被取走的位置 + count; buffer.bufferpos.tid = buffer.bufferbuffer.readpos.tid; buffer.

12、bufferpos.data = buffer.bufferbuffer.readpos.data; + buffer.readpos; if( buffer.readpos >= BUFFER_SIZE ) buffer.readpos = 0; /如果取得了产品则释放一个缓冲区可写的信号量，否则释放一个可读的信号量 if( produce ) sem_post(&buffer.sem_write); else sem_post(&buffer.sem_read); pthread_mutex_unlock(&buffer.rlock); else pthrea

13、d_mutex_unlock(&buffer.rlock); return produce;void *producer(void *data) /每个线程循环生产30个产品 int tid = *(int *)data); int n; for(n = 1; n <= 30; n+) printf("producer %d product %d n", tid, n); put(tid, n); / 每退出一个生产者线程后，thread_count 减1 pthread_mutex_lock(&buffer.lock); - thread_count

14、; pthread_mutex_unlock(&buffer.lock); return NULL;void *consumer(void * data) int tid = *(int *)data); struct Product *d = NULL; while(1) d = get(tid); if( d ) printf("consumer %d consum %d from Producer %d n", tid, d->data,d->tid); free(d); pthread_mutex_lock(&buffer.lock);

15、/ 当所有生产者线程已退出，且缓冲区内已没有该线程可消费的产品时，退出该线程 if(d = NULL && thread_count = 0 ) pthread_mutex_unlock(&buffer.lock); break; else pthread_mutex_unlock(&buffer.lock); return NULL;int main(void) pthread_t th_a30, th_b30; void *retval; init(&buffer); int i; for(i = 0; i<30; +i) / 创建生产者和消费

16、者 pthread_create(&th_ai, NULL, producer, &idsi); pthread_create(&th_bi, NULL, consumer, &idsi); for(i = 0; i<30; +i) / 将线程加入到主线程组 pthread_join(th_ai, &retval); pthread_join(th_bi, &retval); printf("product %d productsn", pcount); printf("consume %d productsn

17、", count); return 0;四、测试测试数据文件格式测试数据由程序内部生成运行结果分析运行 ./consumer cat > output.txt 将输出结果保存在output.txt中待分析。cat output.txt | grep producer 2 由上图可知，生产者2 顺利完成了生产工作。运行 cat output.txt | grep from Producer 5结果如下图：可知生产者5 生产的产品全部被奇数消费者线程消费运行 cat output.txt | grep from Producer 6可知生产者6 生产的产品全部被偶数消费者线程消费。运行： cat output.txt | grep from Producer 13由图可知，所有13号生产这线程生产的产品都被相应的线程消费。运行：cat output.txt | grep from Producer 23由图可知，所有23号生产者生产的产品可以被任意消费者消费。由上图可知，总计生产了900个产品，共消费900个产品，五、总结生产者消费者问题，也称有限缓冲问题，是一个多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程即所谓的“生产者”和“消费者”在实际运行时会发生的问题。生产者的主要作用是生