生产者-消费者

1、生产者-消费者(producer-consume r)问题，也称作有界缓冲区(bounded-buffe r)问题，两 个进程共享一个公共的固定大小的缓冲区。其中一个是生产者，用于将消息放入缓冲区；另 外一个是消费者，用于从缓冲区中取出消息。问题出现在当缓冲区已经满了，而此时生产者 还想向其中放入一个新的数据项的情形，其解决方法是让生产者此时进行休眠，等待消费者 从缓冲区中取走了一个或者多个数据后再去唤醒它。同样地，当缓冲区已经空了，而消费者 还想去取消息，此时也可以让消费者进行休眠，等待生产者放入一个或者多个数据时再唤醒 它。听起来好像蛮对的，无懈可击似的，但其实在实现时会有一个竞争条件存在

2、的。为了跟踪缓 冲区中的消息数目，需要一个变量count。如果缓冲区最多存放N个消息，则生产者的代 码会首先检查count是否达到N，如果是，则生产者休眠；否则，生产者向缓冲区中放入 一个消息，并增加count的值。消费者的代码也与此类似，首先检测count是否为0，如果是，则休眠；否则，从缓冲区 中取出消息并递减count的值。同时，每个进程也需要检查是否需要唤醒另一个进程。代 码可能如下：/缓冲区大小#define N 100int count = 0;/跟踪缓冲区的记录数/ 缓冲区中的数据项/*生产者进程*/ void procedure(void) int item;while(tru

3、e) /无限循环/ 产生下一个数据项/ 如果缓冲区满了，/ 将新数据项放入缓冲区/计数器加1/表明插入之前为空，/ 消费/ 唤醒消费者item = produce_item();if (count = N)进行休眠sleep();insert_item(item);count = count + 1;if (count = 1)者等待wakeup(consumer);while(true)/ 无限循环if (count = 0)入休眠sleep();item = remove_item();count = count - 1;if (count = N -1)生产者wakeup(produce

4、r);consume_item(item);如果缓冲区为空，进/从缓冲区中取出一个数据项/计数器减1/缓冲区有空槽/唤醒/打印出数据项/*消费者进程*/ void consumer(void) /缓冲区中的数据项int item;看上去很美，哪里出了问题，这里对count的访问是有可能出现竞争条件的：缓冲区为空， 消费者刚刚读取count的值为0，而此时调度程序决定暂停消费者并启动执行生产者。生 产者向缓冲区中加入一个数据项，count加1。现在count的值变成了 1.它推断刚才count 为0，所以此时消费者一定在休眠，于是生产者开始调用wakeup（consumer）来唤醒消费者。 但是

5、，此时消费者在逻辑上并没有休眠，所以wakeup信号就丢失了。当消费者下次运行 时，它将测试先前读到的count值，发现为0（注意，其实这个时刻count已经为1 了）， 于是开始休眠（逻辑上）。而生产者下次运行的时候，count会继续递增，并且不会唤醒 consumer 了，所以迟早会填满缓冲区的，然后生产者也休眠，这样两个进程就都永远的休 眠下去了。1，使用信号量解决生产者-消费者问题首先了解一下信号量吧，信号量是E.W.Dijkstra在1965年提出的一种方法，它是使用一 个整型变量来累计唤醒的次数，供以后使用。在他的建议中，引入了一个新的变量类型，称 为信号量（semaphore），

6、一个信号量的取值可以为0 （表示没有保存下来的唤醒操作）或者为 正值（表示有一个或多个唤醒操作）。并且设立了两种操作：down和up（分别为一般化后的sleep和wakeup，其实也是一般教 科书上说的P/V向量）。对一个信号量执行down操作，表示检查其值是否大于0，如果 该值大于0，则将其值减1（即用掉一个保存的唤醒信号）并继续；如果为0，则进程休眠， 而且此时down操作并未结束。另外，就是检查数值，修改变量值以及可能发生的休眠操作都作为单一的，不可分割的原子操作来完成。下面开始考虑用信号量来解决生产者-消费者问题了，不过在此之前，再次分析一下这个问 题的本质会更清晰点：问题的实质在于发

7、给一个（尚）未休眠进程（如上的消费者进程在只 判断了 count = 0后即被调度出来，还未休眠）的wakeup信号丢失（如上的生产者进程 在判断了 count = 1后以为消费者进程休眠，而唤醒它）了。如果它没有丢失，则一切都 会很好。#define N 100typedef int semaphore;semaphore mutex = 1;semaphore empty = N;semaphore full= 0;/缓冲区中的槽数目/信号量一般被定义为特殊的整型数据/控制对临界区的访问/计数缓冲区中的空槽数目/计数缓冲区中的满槽数目/*生产者进程*/void proceducer(voi

8、d)int item;while(1)item = procedure_item();down(&empty);down(&mutex);insert_item(item);up(&mutex);up(&full);加1/*消费者进程*/void consumer(voi)int item;while(1)down(&full);down(&mutex);item = remove_item();up(&mutex);up(&empty);/生成数据/将空槽数目减1/进入临界区/将新数据放入缓冲区/离开临界区/ 将满槽的数目/将满槽数目减1/ 进入临界区/从缓冲区中取出数据项/ 离开临界区/

9、将空槽数目加1consumer_item(item);/处理数据项该解决方案使用了三个信号量：一个为full,用来记录充满的缓冲槽的数目，一个为empty, 记录空的缓冲槽总数，一个为mutex，用来确保生产者和消费者不会同时访问缓冲区。mutex 的初始值为1，供两个或者多个进程使用的信号量，保证同一个时刻只有一个进程可以进入 临界区，称为二元信号量(binary semaphore)0如果每一个进程在进入临界区前都执行一个 down(.)，在刚刚退出临界区时执行一个up(.)，就能够实现互斥。另外，通常是将down和up操作作为系统调用来实现，而且OS只需要在执行以下操作 时暂时禁止全部中

10、断：测试信号量，更新信号量以及在需要时使某个进程休眠。这里使用了三个信号量，但是它们的目的却不相同，其中full和empty用来同步(synchronization)，而 mutex 用来实现互斥。2,使用消息传递解决生产者-消费者问题这种IPC方式使用两条原语send和receive，也是系统调用。如：send(dest, &msg) /将消息msg发送到目标(进程)dest中receive(src, &msg) /接收由src过来的msg,如果没有消息可用，则可能阻塞接收者消息传递系统会面临位于网络中不同机器上的通信进程的情形，所以会更加的复杂。如：消 息可能被网络丢失，一般使用确认(AC

11、K)消息。如果发送方在一定的时间段内没有收到 确认消息，则重发消息。如果消息本身被正确接收，但是返回的ACK消息丢失，发送方则重发消息，这样接收方 就会收到两份同样的消息。一般使用在每条原始消息的头部嵌入一个连续的序号来解决这个 问题。另外，消息传递系统还需要解决进程命名的问题，在send和receive系统调用中指定的进 程必须没有二义性的。还有其他的一些问题，如性能问题，身份认证等等，不过那个就会扯 多了，还是看看如果解决这个生产者-消费者的问题吧：#define N 100/缓冲区中的槽数目/*生产者进程*/void proceducer(void)int item;message ms

12、g;/ 消息缓冲区while(1)/生成数据/等待消费者发送空的缓冲区/创建待发送消息/发送数据项给消费者item = procedure_item(); receive(consumer, &msg); build_msg(&msg, item);send(consumer, &msg); /*消费者进程*/void consumer(voi)int item, i;message msg;for(i=0; iN; i+)send(producer, &msg);while(1)receive(producer, &msg);item = extract_item(&msg);send(pr

13、oceduer, &msg);产者consumer_item(item);/发送给生产者N个空缓冲区/接收包含数据项的消息/解析消息，并组装成数据项/然后又将空缓冲区发送回生/处理数据项在这个解决方案中，共使用了 N条消息，有点类似于上一个的共享内存缓冲区的N个槽， 消费者进程这边首先通过一个for循环将N条空消息发送给生产者。当生产者向消费者传 递一个数据项时，是通过取走每一条接收到的空消息，然后送回填充了内容的消息给消费者 的。通过这种方式，整个消息传递系统中的总的消息数（包括空的消息+存了数据项的消 息=N）是不变的。如果运行过程中，生产者进程的速度比消费者快，则所有的消息最终都会塞满，然后生产者 进程就会等待消费者（即使调用procedure也是阻塞在receive处），直到消费者返回一条 空的消息；反之亦然。下面再来看一下消息传递方式的两种变体。一种是：为每一个进程分配一个唯一的地址，让 消息按照这个进程的地址进行编址。也就是send和receive调用的第一个