基于Hoare管程实现环形缓冲器

1.代码展示

代码文件circle_buffer.cpp。

Hoare管程的实现参照《操作系统教程:第5版》（费翔林著），使用了POSIX中的sem_t信号量，即通过结构体InterfaceModule实现霍尔管程，并定义了Enter、Leave、Wait、Signal四种方法，原理和书上说的一样，就不再赘述了。环形缓冲容器的实现用了C++的类和模板，Get和Put方法就是通过管程来实现的。其实这里就和上课讲的一样，Get/Put主操作完了之后就可以直接返回，没有后面的对共享变量的操作了，Hoare管程的实现是非必须的。

测试样例用两个生产者和两个消费者进行。

具体代码如下：


#include 
#include 
#include 
#include 
#include 
using namespace std;

//通过宏定义P、V操作
#define P(sem) 
    sem_wait(sem);
#define V(sem) 
    sem_post(sem);
typedef sem_t Semaphore;

typedef struct InterfaceModule
{
    Semaphore mutex;  //进程调用管程之前使用的互斥信号量
    Semaphore next;   //发出Signal()操作的进程挂起自己的信号量
    int next_count;   //在next上等待的进程数
    InterfaceModule() //默认构造函数
    {
        sem_init(&mutex, 0, 1);
        sem_init(&next, 0, 1);
        next_count = 0;
    }
    ~InterfaceModule() //析构函数，销毁信号量
    {
        sem_destroy(&mutex);
        sem_destroy(&next);
    }
} InterfaceModule;

void Enter(InterfaceModule &IM)
{
    P(&(IM.mutex)); //互斥进入管程
}
void Leave(InterfaceModule &IM)
{
    if (IM.next_count > 0) //判断有否发出Signal()操作的进程
    {
        IM.next_count--; //等待在next上的进程数减1
        V(&(IM.next));   //释放一个挂起在next上的进程
    }
    else
    {
        V(&(IM.mutex)); //否则开放管程
    }
}
void Wait(Semaphore &x_sem, int &x_count, InterfaceModule &IM)
{
    x_count++;             //等待资源的进程数加1，x_count初始化为0
    if (IM.next_count > 0) //判断有否发出Signal()操作的进程
    {
        IM.next_count--; //等待在next上的进程数减1
        V(&(IM.next));   //释放一个挂起在next上的进程
    }
    else
    {
        V(&(IM.mutex)); //否则开放进程
    }
    P(&(x_sem)); //等待资源的进程阻塞自己，x_sem初始化为0
}
void Signal(Semaphore &x_sem, int &x_count, InterfaceModule &IM)
{
    if (x_count > 0) //判断是否有等待资源的进程
    {
        x_count--;       //等待资源的进程数减1
        IM.next_count++; //发出Signal()操作的进程数加1
        V(&(x_sem));     //释放一个等待资源的进程
        P(&(IM.next))    //发出Signal()操作的进程阻塞自己
    }
}

template 
class circleBuffer
{
public:
    circleBuffer(int K) //构造函数
    {
        sz = K;                   //缓冲区总大小为K
        cnt = 0;                  //初始数量为0
        buffer = new T[sz];       //初始化缓冲区
        head = tail = 0;          //初始化头尾指针
        sem_init(¬full, 0, 0); //分别给“生产者”和“消费者”初始化两个信号量
        sem_init(¬empty, 0, 0);

        notfull_count = notempty_count = 0; //用于记录等待资源的进程数量
    }
    ~circleBuffer() //析构函数
    {
        delete[] buffer;       //删除缓冲区
        sem_destroy(¬full); //销毁信号量
        sem_destroy(¬empty);
    }

    void Get(T &t) // Get方法
    {
        Enter(im);    //进入管程
        if (cnt == 0) //如果缓冲区为空，则阻塞当前进程
            Wait(notempty, notempty_count, im);
        t = buffer[head];
        head = (head + 1) % sz;
        cnt--;                              //拿去一个产品
        Signal(notfull, notfull_count, im); //唤醒等待的生产者
        Leave(im);                          //离开管程
    }

    void Put(T &t) // Put方法
    {
        Enter(im);                            //进入管程
        if (cnt == sz)                        //如果缓冲区满
            Wait(notfull, notfull_count, im); //阻塞当前进程
        buffer[tail] = t;
        tail = (tail + 1) % sz;
        cnt++;                                //添加一个产品
        Signal(notempty, notempty_count, im); //唤醒一个消费者进程
        Leave(im);                            //离开管程
    }

private:
    T *buffer;                         //缓冲区
    int sz;                            //缓冲区大小
    int cnt;                           //当前待处理元素数量
    int head;                          //头指针
    int tail;                          //尾指针
    Semaphore notfull, notempty;       //生产者、消费者的缓冲区信号量
    int notfull_count, notempty_count; //记录等待资源进程的个数
    InterfaceModule im;                // IM，管程结构体
};

//测试样例
#define BUFFER_SIZE 5

//规定生产者、消费者等待时间，如果要测试缓冲区满，生产者等待的情况，则上1下5
//如果要测试缓冲区空，消费者等待的情况，则上5下1
#define PRODUCER_SLEEP_TIME 5
#define CONSUMER_SLEEP_TIME 1
circleBuffer c(BUFFER_SIZE); //大小为BUFFER_SIZE的环形缓冲队列
void *producer(void *arg)         //生产者进程
{
    char id = *(char *)arg;
    int num;
    sleep(PRODUCER_SLEEP_TIME);
    while (1)
    {
        num = rand() % 10;
        c.Put(num);
        printf("Producer %c puts num %d.n", id, num);
        sleep(PRODUCER_SLEEP_TIME);
    }
}
void *consumer(void *arg) //消费者进程
{
    char id = *(char *)arg;
    int num;
    sleep(CONSUMER_SLEEP_TIME);
    while (1)
    {
        c.Get(num);
        printf("Consumer %c gets num %d.n", id, num);
        sleep(CONSUMER_SLEEP_TIME);
    }
}
int main()
{
    srand((unsigned int)time(0)); //修改种子
    pthread_t p1, p2, c1, c2;     //两个生产者，两个消费者
    char p1_name = 'A', p2_name = 'B', c1_name = 'C', c2_name = 'D';
    pthread_create(&p1, NULL, producer, (void *)&p1_name);
    pthread_create(&p1, NULL, producer, (void *)&p2_name);
    pthread_create(&c1, NULL, consumer, (void *)&c1_name);
    pthread_create(&c1, NULL, consumer, (void *)&c2_name);

    pthread_join(p1, NULL);
    pthread_join(p2, NULL);
    pthread_join(c1, NULL);
    pthread_join(c2, NULL);
}

2.测试结果

缓冲区的大小设置为5。

先测试Put速度快，Get速度慢，从而缓冲区满，生产者等待的情况：

修改生产者、消费者等待时间：
```
#define PRODUCER_SLEEP_TIME 5
#define CONSUMER_SLEEP_TIME 1
```
执行命令：
```
g++ circle_buffer.cpp -o circle_buffer -lpthread
./circle_buffer
```
得到结果：

可以看到由于生产者速度过快，生产者会先将缓冲区填满，然后生产者进入等待状态。之后消费者每次取出一个，生产者（即Put进程）都会被立刻唤醒，生产者立即把容器填满。

再测试Get速度快，Put速度慢，从而缓冲区空，消费者等待的情况：

修改生产者、消费者等待时间：
```
#define PRODUCER_SLEEP_TIME 1
#define CONSUMER_SLEEP_TIME 5
```
执行命令：
```
g++ circle_buffer.cpp -o circle_buffer -lpthread
./circle_buffer
```
得到结果：

可以看到由于消费者速度大于生产者生产速度，每次生产者一生产产品，消费者就会立马消费，并且进入到阻塞态，并唤醒一个生产者进程，然后生产者再生产，消费者再立即消费。。。

基于Hoare管程机制的环形缓冲容器同步Get和Put实现成功。

基于Hoare管程实现环形缓冲器

Java相关栏目本月热门文章