【读书笔记】linux系统用semaphore来解决经典的生产者-消费者问题

在Linux系统下处理多进程或多线程的并发编程时，进程/线程同步是经常要遇到的问题。而在众多同步问题的场景中，生产者-消费者问题（Producer-Consumer Problem）是一个几乎每部涉及到同步问题的经典教材都会提到的经典模型。在linux系统中，实现同步的典型思路是借助内核提供的3种变量，分别是：1)
互斥量（mutex）; 2) 信号量（semaphore）; 3) 条件变量（condition variable）。

本文下面的内容为《Computer Systems: A Programmer's Perspective》一书第12.5小节—用信号量同步线程的读书笔记，旨在说明如何利用semaphore解决生产者-消费者的同步问题。

1. 信号量semaphore的语义及用途

semaphore的概念是著名的图灵奖得主—荷兰CS科学家Edsger Dijkstra（在提出semaphore之前，此君因提出Dijkstra's algorithm 解决了图搜索中的最短路径问题而闻名于世）于1965年提出的。简单来说，semaphore是一个特殊类型的变量，它具有非负整数值，支持两种特殊的操作，这两种操作称为P和V：

P(s): 若s是非零的，则P将s减1并立即返回。若s为零，那就挂起调用P(s)的这个线程，直到s变为非零，而一个V操作会重启这个线程。在重启之后，P操作将s减1，并将控制返回给调用者。

V(s): V操作将s加1。若有任何线程阻塞在P操作等待s变为非零，那么V操作会重启这些线程中的一个，然后该线程将s减1，完成它的P操作。

注：P/V来源于荷兰语Proberen（测试）和Verhogen（增加）。

需要注意的几点：

1) P中的测试和减1操作是不可分割的，即一旦预测信号量s变为非零，就会将s减1，不能有中断。V中的加1操作也是不可分割的，即加载、加1和存储信号量的过程中不能有中断。也即，P/V均为原子操作。

2) V的定义中没有定义等待线程被重新启动的顺序，唯一的要求是V必须只能重启一个正在等待的线程。因此，当有多个线程在等待同一个信号量时，你不能预测V操作要重启哪一个线程。

3) 由2)可知，当有多个线程在P操作处阻塞时，另一个线程的V操作只能"唤醒"其中的一个，这需要内核的实现来保证，以避免惊群效应（Thundering Herd Problem）。

linux内核中提供了P/V原语对应的具体函数，具体可见<semaphore.h>。

semaphore提供了一种方便的方法来确保对共享变量的互斥访问，基本思想是将每个共享变量（或一组相关的共享变量）与一个信号量s（初始为1）联系起来，然后用P(s)和V(s)操作将相应临界区保护起来。以这种方式来保护共享变量的信号量叫做二元信号量（binary semaphore），因为它的值总是0或1。除此之外，信号量还可以用来对一组可用资源进行计数，这样的信号量称为计数信号量（counting semaphore）。
2. 生产者-消费者问题

典型的生产者-消费者问题如下图所示。生产者和消费者线程共享一个由n个槽的有限缓冲区，生产者线程反复生成新的item并将其插入缓冲区尾部，消费者线程不断从缓冲区头部取出这些item并消费他们。



由于插入和取出item都涉及更新共享变量，所以必须保证对缓冲区的访问是互斥的。但仅保证互斥访问是不够的，我们还需调度对缓冲区的访问：若缓冲区是满的（无空的slot），则生产者必须等待直到有一个slot变为可用；类似地，若缓冲区是空的（无可用item），则消费者必须等待直到有一个item变为可用。

3. 用Semaphore解决Producer-Consumer问题

我们用下面这个自定义变量类型来抽象生产者-消费者模型。

typedef struct
{
int * buf;     // buffer array
int   n;       // maximum number of slots
int   front;   // buf[(front+1) % n] is first item
int   rear;    // buf[rear % n] is last item
sem_t mutex;   // protects accesses to buf
sem_t slots;   // counts available slots
sem_t items;   // counts available items
} sbuf_t;

由上面sbuf_t类型定义可知，我们维护了一个最大可以放置n个items的有限缓冲区buf，front和rear分别指向buf中的第一个item和最后一个item，三个信号量同步对缓冲区的访问，其中，mutex信号量提供互斥的缓冲区访问，slots和items分别对空槽位和可用items进行计数。

下面我们分别定义sbuf_init()、sbuf_deinit()、sbuf_insert()和sbuf_remove()来模拟生产者-消费者问题。注意：限于篇幅，下面给出的示例代码中均未对异常情况做处理，实际编码实现中，异常处理是不可忽略的。

// create an empty, bounded, shared FIFO buffer with n slots
void sbuf_init(sbuf_t * sp, int n)
{
sp->buf = calloc(n, sizeof(int));
sp->n   = n;                        // buffer holds max of n items
sp->front = sp->rear = 0;           // empty buffer if front == rear
sem_init(&sp->mutex, 0, 1);         // binary semaphore for mutex-locking
sem_init(&sp->slots, 0, n);         // initially, buf has n empty slots
sem_init(&sp->items, 0, 0);         // initially, buf has zero data items
}

上面的函数中，我们为有限缓冲区在heap上分配空间，设置front和rear表示这是一个空的缓冲区，并为三个信号量赋初始值。调用完该函数后，我们创建了一个带保护的、初始为空的有限缓冲区。

// clean up buffer sp
void sbuf_deinit(sbuf_t * sp)
{
free(sp->buf);
}

// insert item onto the rear of shared buffer sp, it's an abstract of producer
void sbuf_insert(sbuf_t * sp, int item)
{
P(&sp->slots);                            // wait for available slot
P(&sp->mutex);                            // lock the buffer
sp->buf[(++sp->rear) % (sp->n)] = item;   // insert the item
V(&sp->mutex);                            // unlock the buffer
V(&sp->items);                            // announce available ite
}

// remove and return the first item from buffer sp,  it's an abstract of consumer
void sbuf_remove(sbuf_t * sp)
{
int item;
P(&sp->items);                             // wait for available item
P(&sp->mutex);                             // lock the buffer
item = sp->buf[(++sp->front) % (sp->n)];   // remove the item
V(&sp->mutex);                             // unlock the buffer
V(&sp->slots);                             // announce available slot
return item;
}

关于上面的代码需要说明的是：

有限缓冲区buf是被当做circular buffer来用的（从访问buf的index计算式可以看出这一点），因此，从哪里开始写第一个item是不重要的（因此，代码中的第一个item被写到buf[1]的位置）。而且从代码可知，这里的circular buffer无论是空还是满，其条件均为front == rear。采用这种判空/判满条件的好处是不浪费slot，缺点是空/满条件都是front
== rear，是比较容易引人迷惑从而引入bug的。关于circular buffer用其它方法来判空/判满的思路，不在本笔记的讨论范围内，感兴趣的同学，可以参考这里。

思考题：

设p表示生产者数量，c表示消费者数量，n表示缓冲区中最多的items数量。对于下面的场景，指出sbuf_insert和sbuf_remove中互斥锁信号量是否是必须的。

A. p = 1, c = 1, n > 1

B. p = 1, c = 1, n = 1

C. p > 1, c > 1, n = 1

Answers:

A场景下，mutex信号量是必须的。因为producer和consumer会并发访问缓冲区。

B/C场景下，mutex非必须。因为buffer只有1个slot，非空即满，producer生产了一个item后，buffer满导致其阻塞在P(&sp->slots)；而consumer消费完仅有的这个item并通过V(&sp->slots)试图唤醒producer后，在buffer中有可用item之前，会由于buffer空而阻塞在P(&sp->items)。可见，当n = 1时，生产者和消费者可以借助sp->slots和sp->items实现互斥访问缓冲区，因此，此时可以省去显式的互斥锁。
【参考资料】

1. <Computer Systems: A Programmer's Perspective>. chapter 12.5

2. wikipedia: Producer-Consumer Problem

3. wikipedia: semaphore

4. wikipedia: Thundering Herd Problem

5. wikipedia: Circular buffer
================= EOF ================