实验、进程的同步与互斥——生产者消费者

1. 1.  实验目的

两个或两个以上的进程,不能同时进入关于同一组共享变量的临界区域,否则可能发生与时间有关的错误,这种现象被称作进程互斥。对CPU的速度和数目不做出任何假设的前提下，并发进程互斥访问临界资源，是一个较好的解决方案。另外，还需要解决异步环境下的进程同步问题。所谓异步环境是指：相互合作的一组并发进程，其中每一个进程都以各自独立的、不可预知的速度向前推进；但它们又需要密切合作，以实现一个共同的任务，即彼此“知道”相互的存在和作用。

实验目的：分析进程争用资源的现象，学习解决进程同步与互斥的方法。

本实验属于设计型实验，实验者可根据自身情况选用合适的开发环境和程序架构。

1. 2.  实验原理

信号量的PV操作与处理相关，P表示通过的意思，V表示释放的意思。

1962年，狄克斯特拉离开数学中心进入位于荷兰南部的艾恩德霍芬技术大学(Eindhoven Technical University)任数学教授。在这里，他参加了X8计算机的开发，设计与实现了具有多道程序运行能力的操作系统——THE Multiprogramming System。THE是艾恩德霍芬技术大学的荷兰文Tchnische Hoogeschool Eindhov –en的词头缩写。狄克斯特拉在THE这个系统中所提出的一系统方法和技术奠定了计算机现代操作系统的基础，尤其是关于多层体系结构，顺序进程之间的同步和互斥机制这样一些重要的思想和概念都是狄克斯特拉在THE中首先提出并为以后的操作系统如UNIX等所采用的。

为了在单处理机的情况下确定进程(process)能否占有处理机，狄克斯特拉将每个进程分为“就绪”(ready)、“运行”(running)和“阻塞”(blocking)三个工作状态。由于在任一时刻最多只有一个进程可以使用处理机，正占用着处理机的进程称为“运行”进程。当某进程已具备了使用处理机的条件，而当前又没有处理机供其使用，则使该进程处于“就绪”状态。当运行进程由于某种原因无法继续运行下去时，就停止其占用处理机，使之进入“阻塞”状态，待造成其退出运行的条件解除，再进入“就绪”状态。而对系统中所有同时运行的进程（在一个进程访问共享数据时，另一个进程不访问该数据），存在同步和互斥（mutually- exclusive，指两个进程不能同时在一个临界区中使用同一个可重复使用的资源，诸如读写缓冲区）两个关系，狄克斯特拉巧妙地利用火车运行控制系统中的“信号量”(semaphore，或叫”信号灯”)概念加以解决。

所谓信号量，实际上就是用来控制进程状态的一个代表某一资源的存储单元。例如，P1和P2是分别将数据送入缓冲B和从缓冲B读出数据的两个进程，为了防止这两个进程并发时产生错误，狄克斯特拉设计了一种同步机制叫“PV操作”，P操作和V操作是执行时不被打断的两个操作系统原语。执行P操作P（S）时信号量S的值减1，若结果不为负则P（S）执行完毕，否则执行P操作的进程暂停以等待释放。执行V操作V（S）时，S的值加1，若结果不大于0则释放一个因执行P（S）而等待的进程。对P1和P2可定义两个信号量S1和S2，初值分别为1和0。进程P1在向缓冲B送入数据前执行P操作P（S1），在送入数据后执行V操作V（S2）。进程P2在从缓冲B读取数据前先执行P操作P（S2），在读出数据后执行V操作V（S1）。当P1往缓冲B送入一数据后信号量S1之值变为0，在该数据读出后S1之值才又变为1，因此在前一数未读出前后一数不会送入，从而保证了P1和P2之间的同步。

这一同步机制叫PV操作，这是狄克斯特拉用荷兰文定义的，因为在荷兰文中，通过叫passeren，释放叫vrijgeven，PV操作因此得名。这是在计算机术语中不是用英语表达的极少数的例子之一。

1. 3.  实验内容

以下是进程同步与互斥的典型问题例析——生产者消费者，请模拟它的实现过程。

问题描述：一个仓库可以存放K件物品。生产者每生产一件产品，将产品放入仓库，仓库满了就停止生产。消费者每次从仓库中去一件物品，然后进行消费，仓库空时就停止消费。

数据结构：进程：Producer - 生产者进程，Consumer - 消费者进程

       buffer: array [0..k-1] of integer;

        in,out: 0..k-1;

       in记录第一个空缓冲区，out记录第一个不空的缓冲区

        s1,s2,mutex: semaphore;

       s1控制缓冲区不满,s2控制缓冲区不空,mutex保护临界区；

        初始化s1=k,s2=0,mutex=1

4.  实验步骤

         i.      选用合适的实验程序开发环境。

        ii.      设计程序结构，规划程序功能。

        iii.      完成程序的编码与测试。

        iv.      设计实验数据。

         v.      针对实验数据运行程序，并将结果记录在实验报告上。

5.  代码：

#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <pthread.h>
#include <unistd.h>
#include <signal.h>
#include <semaphore.h>
#define Maxbuf 10
#define TimesOfOp 10
#define true 1
#define false 0

struct Circlebuf
{
    int read;
    int write;
    int buf[Maxbuf];
} circlebuf;

sem_t mutex;//
sem_t empty;//
sem_t full;//

void writeCirclebuf(struct Circlebuf *circlebuf, int *value)
{
    circlebuf->buf[circlebuf->write] = (*value);
   circlebuf->write=(circlebuf->write+1)%Maxbuf;
}

int readCirclebuf(struct Circlebuf *circlebuf)
{
    int value = 0;
    value = circlebuf->buf[circlebuf->read];
    circlebuf->buf[circlebuf->read] = 0;
    circlebuf->read = (circlebuf->read + 1) % Maxbuf;
    return value;
}

void OutCirclebuf(struct Circlebuf *circlebuf)
{
    int i;
    printf("+++++缓冲区各单元的值.\n");
    for (i = 0; i <Maxbuf; i++)
        printf("%d  ", circlebuf->buf[i]);
    printf("\n");
}

void sigend(int sig)
{
    exit(0);
}

void *productThread(void *i)
{
    int *n = (int *)i;
    int t = TimesOfOp;
    while (t--)
    {
        sem_wait(&empty);//
        sem_wait(&mutex);//
        writeCirclebuf(&circlebuf, n);
        printf("----------------生产者%d 写入缓冲区的值%d \n", *n, *n);

        OutCirclebuf(&circlebuf);
       	sem_post(&mutex);
	sem_post(&full);
    }
}

void *consumerThread(void *i)
{
    int *n = (int *)i;
    int value = 0;
    int t = TimesOfOp;
    while (t--)
    {
        sem_wait(&full);
		sem_wait(&mutex);
        value = readCirclebuf(&circlebuf);
        printf("----------------消费者%d 取走产品的值%d \n",*n,value);
        OutCirclebuf(&circlebuf);
       sem_post(&mutex);
		sem_post(&empty);
    }
}

 main()
{
    int i;
    int ConsNum = 0, ProdNum = 0, ret;
    pthread_t cpid, ppid;//
    sem_init(&mutex, 0, 1);
    sem_init(&empty, 0, Maxbuf);

    sem_init(&full, 0, 0);
    signal(SIGINT, sigend);
    signal(SIGTERM, sigend);

    circlebuf.read = circlebuf.write = 0;
    for ( i = 0; i < Maxbuf; i++)
        circlebuf.buf[i] = 0;
    printf("请输入生产者的数目 :");
    scanf("%d",&ProdNum);
    int *pro=(int*)malloc(ProdNum*sizeof(int));
     printf("请输入消费者的数目 :");
    scanf("%d", &ConsNum);
    int *con = (int *)malloc(ConsNum * sizeof(int));
    for ( i = 1; i <= ConsNum; i++)
    {

        cpid = i;
        con[i - 1] = i;
        ret = pthread_create(&cpid, NULL, consumerThread,(void*)&con[i-1] ); 
        if (ret != 0)
        {
            printf("Create thread error");
            exit(1);
        }
    }
    for (i = 1; i <= ProdNum; i++)
    {
        ppid = i + 100;
        pro[i - 1] = i;
        ret = pthread_create(&ppid, NULL, productThread, (void *)&pro[i - 1]);
        if (ret != 0)
        {
            printf("Create thread error");
            exit(1);
        }
    }
    sleep(1);
    sem_destroy(&mutex);//
    sem_destroy(&empty);//
    sem_destroy(&full);//
    pthread_exit(NULL);//
}