经典进程同步问题--生产者消费者问题

朱雀 2022-04-16 20:14 237阅读 0赞

**问题描述：**

一群生产者进程在生产产品，并将这些产品提供给消费者去消费。为了使生产者进程与消费者进程能够并发进行，在两者之间设置一个具有n个缓冲区的缓冲池，生产者进程将产品放入一个缓冲区中；消费者可以从一个缓冲区取走产品去消费。尽管所有的生产者进程和消费者进程是以异方式运行，但它们必须保持同步：当一个缓冲区为空时不允许消费者去取走产品，当一个缓冲区满时也不允许生产者去存入产品。

**解决方案：**

我们这里利用一个一个数组buffer来表示这个n个缓冲区的缓冲池

用输入指针和输出指针+1来表示在缓冲池中存入或取出一个产品

由于这里的缓冲池是循环缓冲的，故应把in和out表示成：in = ( in +1 ) % n (或把out表示为 out = ( out +1 ) % n )

当( in +1) % n= out的时候说明缓冲池满，in = out 则说明缓冲池空。

在这里还要引入一个整型的变量counter（初始值0），每当在缓冲区存入或取走一个产品时，counter +1或-1。

那么问题的关键就是，把这个counter作为临界资源处理，即令生产者进程和消费者进程互斥的访问它。

首先解释一下信号量（Semaphores）机制。这是荷兰学者Dijkstra在1965年提出的一种有效的进程同步工具。Dijkstra定义的整型信号量是一个用于表示资源数目的整型量，不同于一般的整型量，它除了初始化以外只能通过两个标准的原子操作（Atomic Operation）waits(S)和signal(S)来访问。很长时间以来这两个操作分别被称为P、V操作。wait和signal操作描述如下：

wait(S){
    
            while (S <= 0); //无可用资源
            S-- ;
    }
    
    signal(S){
    
            S-- ;
    }

waits(S)和signal(S)是原子操作，执行时不可中断。即：当一个进程在修改某个信号量时，没有其他进程可以对其修改。

**解决办法**：

**1.利用记录型信号量解决问题**

上述整型信号量机制中，只要S<=0就会不断的测试，没有遵循“让权等待”准则，可能导致“忙等”。记录型信号量机制采取“让权等待”，但会出现多个进程等待访问同一个临界资源的情况，因此在此机制中，除了一个表示资源数量的整型量value以外，还需要增加一个进程链表指针list，用于链接所有等待进程。

记录型信号量的数据项描述如下：

1 typedef struct{
    2 
    3         int value ;
    4         struct processControlBlock *list ;
    5 
    6 }semaphore;

相应的wait(S)和signa(S)为：

wait(semaphore *S){
    
            S->value -- ;
            if( S->value < 0 )
                    block( S->list ); //资源分配完毕，自我阻塞
            
    }
    
    signal(semaphore *S){
    
            S->value ++ ;
            if( S->value <= 0 )
                    weakup( S->list ); //仍有等待该资源的进程被阻塞，将list中的第一个进程唤醒
            
    }

现在是对生产者-消费者问题解决方案的描述：

假定在生产者和消费者之间的公共缓冲池中具有n个缓冲区，利用互斥信号量mutex来实现诸进程对缓冲池的互斥使用；利用信号量empty和full分别表示缓冲池中的空槽数量和满槽数量。

int in = 0 , out = 0 ;
    item buffer[n];
    semaphore mutex = 1 ; //信号量，控制对临界资源的互斥访问
    semaphore empty = n ; //信号量，表示缓冲池当前的空槽数目
    semaphore full = 0 ; //信号量，表示缓冲池当前的满槽数目
    
    //producer 
    void producer {
        
        do {
    
            item = produce_item();
            wait(empty); //空槽数目减一
            wait(mutex); //进入临界区
            buffer[in] = insert_item(item); //存入产品
            in = ( in +1 ) % n ;
            signal(mutex); //退出临界区
            signal(full); //满槽数目加一
    
               } while（1）
    }
    
    //consumer 
    void consumer{
        
        do {
    
            wait(full); //满槽数目减一
            wait(mutex); //进入临界区 ，
            //注意：两个P操作的顺序不能颠倒
            remove_item(item) = buffer[out] ; //取出产品
            out = ( in +1 ) % n ;
            signal(mutex); //退出临界区
            signal(empty); //空槽数目加一
            //V的顺序都可以
            consumer_item(item);
               } while（1）
    }
    
    void main() {
    
        cobegin //并发执行
        producer();consumer();
        coend
    
    }

注意：

1.用于实现互斥的wait(mutex) 和signal(mutex)必须成对出现，而对资源的full和empty的wait和signal也要成对出现，这里是指分别出现在不同的程序中。

2.应该先进行对资源的wait操作再进行对互斥信号量的wait操作。

**2.利用AND信号量解决问题**

首先还是解释一下AND型信号量。上述是针对多个并发程序仅共享一个临界资源的情况，AND型信号量则是解决多个并发程序共享多个资源的问题。基本思想是：对若干个临界资源的分配采取原子操作的方法，要么把它请求的所有资源都分配到进程，要么就一个也不分配。这里用Swait()表示要么全部分配，要么全不分配，用Ssignal()表示全部释放。

Swait() 和Ssignal的描述如下：

Swait(S1,S2,S3,...,Sn){
        while (1) {
            if( S1>=1 && S2>=1 && ... && Sn>=1){
    
            for( i = 1 ; i <= n ; i++) Si --;
            break ;
                }else{
    
                将这个进程放进等待队列，第一个<1的Si作为等待队列的开始
                }
        }    
    }
    
    Ssignal(S1,S2,...,Sn) {
    
        while(1) {
            for( i = 1 ; i <= n ; i++) {
            Si ++;
            清空等待队列并将其存入就绪队列
            }
        }
    }

具体实现：

consumer_item(item);
    
    int in = 0 , out = 0 ;
    item buffer[n];
    semaphore mutex = 1 ; //信号量，控制对临界资源的互斥访问
    semaphore empty = n ; //信号量，表示缓冲池当前的空槽数目
    semaphore full = 0 ; //信号量，表示缓冲池当前的满槽数目
    
    //producer 
    void producer {
        
        do {
    
            item = produce_item();
            Swait(empty,mutex); 
            buffer[in] = insert_item(item); //存入产品
            in = ( in +1 ) % n ;
            signal(mutex，full); 
    
    
               } while（1）
    }
    
    //consumer 
    void consumer {
        
        do {
    
            wait(full，mutex); 
            remove_item(item) = buffer[out] ; //取出产品
            out = ( in +1 ) % n ;
            signal(mutex，empty); 
                consumer_item(item);
               } while（1）
    }
    
    void main() {
    
        cobegin //并发执行
        producer();consumer();
        coend
    
    }

**3.利用管程解决问题**

首先介绍一下管程。代表共享资源的数据结构以及对该数据结构实施操作的一组过程所组成的资源管理程序共同构成了一个操作系统的资源管理模块，称为管程。

这里先定义一个管程PC，描述如下：

Monitor PC {
    
        item buffer[N];
        int in , out ;
        condition notfull , notempty; //条件变量
        int count ; //缓冲池中已有的产品数目
    
        public:
        //没满就往里存，满了就阻塞
        void put( item x ){ 
    
        if( count >= N)
            cwait( notfull ); //缓冲池满，阻塞，挂在条件condition的队列上
            buffer[in] = x;
                in = ( in +1 ) % n ;
            count ++ ;
            csignal(notempty); // 唤醒一个阻塞在condition队列中的进程，队列空则无操作
        }
    
        void get( item x ){
    
        if( count <= 0)
            cwait( notempty ); //缓冲池空，阻塞，挂在条件condition的队列上
            x = buffer[out];
                out = ( out +1 ) % n ;
            count -- ;
            csignal(notfull); // 唤醒一个阻塞在condition队列中的进程，队列空则无操作
        }
    
        { in = 0, out = 0 , count = 0 ;}
    
    }PC;

具体方案：

//producer
    void producer(){
    
        item x;
        while(1){
    
        x = produce_item();
        PC.put(x);
        }
    }
    
    //consumer
    
    void consumer(){
    
        item x;
        while(1){
    
        PC.get(x);
        consumer_item() = x ;
        
        }
    }
    
    void main() {
    
        cobegin //并发执行
        producer();consumer();
        coend
    
    }