并发与同步、信号量与管程、生产者消费者问题

正文

 　　计算机硬件发展到今天，不管是专业服务器还是PC，甚至于最普遍的移动设备基本上都是多核CPU，程序的并发执行可以更加充分利用这些计算资源。除此之后，为了协调CPU与外设（如磁盘）的速度差异，我们也需要并发。本文是笔者学习清华大学和UCSD（加州大学圣迭戈分校）的操作系统课程的笔记和总结，以及自己的思考和实践。

并发与同步：

回到顶部

    并发不是多核时代的产物，在早期的多核CPU已经通过时分复用来实现程序之间的并发。并发带来的好处很多，比如资源共享、提高程序执行效率、模块化等等。但并发对编程带来了很多挑战，比如互斥、死锁。

    所谓同步互斥，就是在并发的情况下，保证一些操作的原子性。原子操作（Atomic Operation）即在一次执行过程中要么全部成功，要么全部失败的操作，不存在部分执行的情况。原子性在各种应用场景都非常重要，比如数据库的ACID，其中A就是原子性。生活中，也要很多原子性的例子，比如银行卡之间转账，分成两个操作：A扣钱，B加钱，但这两个操作必须满足原子性，不然就出大问题了。

    我们常见意义上的并发一般是指多进程之间或者多线程之间。当然，近些年越来越流行的协程也算一种并发。对于非操作系统内核开发人员，需要同步互斥的场景大多数都是线程之间的并发。更好的资源共享是线程的特性之一，下图（来自UCSD：Principles of Operating Systems ）所示是同一个进程内三个线程的内存使用情况：

    

    对于每一个进程内的多个线程，static data segment（包括全局变量、static对象）、Heap（堆，malloc和new分配的空间）是共享的。每个线程有自己独立的Stack（栈），存储局部变量。

    后文主要以多线程为例，但同样适用于多进程。

临界区

    提到并发编程，首先就想到临界区（critical section）这个概念，临界区是线程中访问临界资源的一段需要互斥执行的代码。临界资源是指线程之间共享的资源，但不同的执行序列结果不确定的，这也叫做竞态条件（race condition）。举个例子，我们都知道linux的系统调用fork会创建一个子进程，该调用在父进程会返回子进程的PID，操作系统中每个进程的PID必须是独一无二的。假设pid的生成是这样的：

new_pid = next_pid++

    上面的代码中next_pid就是临界资源，多个进程可能同时访问这个资源， 但上述代码不是原子的（在汇编下是几条指令），但是需要互斥执行的，因此需要临界区。当然，临界区只是一个概念，具体怎么实现依赖于系统与编程语言。在编码中使用临界区的伪代码如下：

entry section        

     critical section 

exit section         

    分为三部分：

    entry section： 判断能否进入，如果能则设置标志，不能则等待

    critical section：需要互斥访问的代码段

    exit section： 清除设置的标志，使得其他进程（线程）可以进入临界区

    临界区的实现有几种方式：

• 禁用硬件中断：

　　我们知道，系统调用以及执行流程的切换都是依靠软中断。禁用中断之后，进程（线程）就不会被切换出去，从而保证代码段能执行结束。但坏处也很明显，由于中断被禁用，如果临界区代码一直执行，其他进程就没机会执行了。而且，只能禁止单个CPU的中断。

• 基于软件同步：

　　即基于代码实现同步互斥，比较有名的是peterson算法，用来解决两个进程对临界区的互斥访问问题。

• 基于原子操作原语的方法：

　　上述两种方式都比较复杂，我们需要更加高级的武器。支持并发的语言都提供了锁（lock）这个概念，在现实生活中也很好理解，如果只能一个人在屋子里，那么进去之后就锁上，出来的时候再打开锁；没有锁的人只能在外面等着。在编程语言中，大概是这样样子的：

acquire(lock)

　　critical section 

release(lock)

　　acquire，release实现的也就是entry section和 exit section的功能，上面的代码是面向过程的写法，面向对象一般写成lock.acquire和lock.release，或者使用RALL（Resource Acquisition Is Initialization）来避免release函数未被调用到的问题。

    lock的实现需要基于硬件提供的“原子操作指令”，这些操作虽然理解起来是几步操作，但硬件保证其原子性。比较常见的原子操作指令包括 test and set、compare and swap，接下来通过test and set来看看lock的实现。

Spinlock实现

　　test-and-set是一个原子操作，其作用对某个变量赋值为1（set），并返回变量之前的值，下面用C语言描述这个过程

1     #define LOCKED 1
2     int TestAndSet(int* lockPtr) {
3         int oldValue;
4          
5         oldValue = *lockPtr;
6         *lockPtr = LOCKED;
7 
8         return oldValue;
9     }

    对应的acqurie和release的伪码如下：

void acquire(int *lock){
    while(TestAndSet(*lock));
}

void release(int *lock){
    *lock = 0;
}

　　在acquire函数中，如果TestAndSet返回1，那么while循环就一直执行（也就是在这里等待），直到另一个线程调用release。当然，这个实现看起来不太好，主要是等待的线程会不停的检查，浪费CPU，这个问题称之为忙等待（busy-wait or spin-wait），所以这个lock的实现也叫自旋锁spinlock。解决办法是如果需要等待，那么该线程主动交出执行权，让其他线程有机会执行，这种方式称之为让权等待（yield-wait or sleep-wait），应用开发人员使用的互斥锁一都是指这种情况。

信号量与管程

回到顶部

　　在上一部分介绍了并发、同步互斥、临界区等基本概念，也介绍了利用原子操作指令实现锁（lock）的机制与实现。但这个的lock（spinlock）是最基础的同步原语，实际操作系统需要封装更高级的同步原语来实现更复杂、更实用的功能。信号量（semaphore）和管程（monitor）就是操作系统提供的两种更高级的同步方式，在操作系统（linux）和编程语言都有对应的实现和封装，本章节对二者进行介绍和对比。

信号量

　　semaphore是大牛Dijkstra（没错，就是出现在数据结构或者图论中的Dijkstra）在20世纪60年代发明的概念，用来协调并发程序对共享资源的访问。注意，这里是协调，而不是互斥，协调的概念更广泛一些，也包括并发程序之间的协作。semaphore有一个整型变量（S）和两个原子操作组成。S了资源的数量，而两个原子操作一般被成为P操作和V操作（有时也被成为wait、signal）。

　　P操作表示申请一个资源，P操作的定义：S=S-1，若S>=0，则执行P操作的线程继续执行；若S<0，则置该线程为阻塞状态，并将其插入阻塞队列。伪码如下：

Procedure P（Var S：Semaphore）；
　　Begin
　　　　S：=S-1；
　　　　If S<0 then w(S) {执行P操作的线程插入等待队列}
　　End；

　　V操作表示释放一个资源，V操作定义：S=S+1，若S>0则执行V操作的线程继续执行；若S<0，则从阻塞状态唤醒一个线程，并将其插入就绪队列。伪代码如下：

Procedure V（Var S：Semaphore）；
　　Begin 
　　　　S：=S+1
　　　　If S<=0 then R(s) {从阻塞队列中唤醒一个线程}
　　End；

　　信号量在现实生活中很容易找到对比的例子，比如银行的窗口数量就是S，在窗口办理业务就是P操作，业务办理结束就是V操作。

　　根据S初始值的不同，semaphore就有不同的作用。如果S初始值为1，那么这个semaphore就是一个mutex semaphore，效果就是临界区的互斥访问。如果S初始值为0，那么就是用来做条件同步，效果就是必须等待某些条件发生。如果S初始值为N（N一般大于1），那么就是用来限制并发数目，也被称之为counting semaphone。

　　后文会利用具体的例子（生产者消费者问题）来阐述semaphore上面的三种用法。

管程

 　　管程是编程语言提供的一种抽象数据结构，用于多线程互斥访问共享资源。首先，是互斥访问，即任一时刻只有一个线程在执行管程代码；第二，正在管程内的线程可以放弃对管程的控制权，等待某些条件发生再继续执行。第二点就厉害了，不管是之前提到的互自旋锁还是信号量，进入了临界区域除非代码执行完，否则是不会出现线程切换的，而管程可以主动放弃执行权，这反映到编码上也会有一些差异。

　　条件变量（condition variable）是管程内部的实现机制，每个条件变量都代表一种等待的原因，也对应一个等待队列。条件变量有两个操作：wait和signal，或者在加上一个signal_all操作。条件变量都是配合互斥锁一起使用，互斥锁保证了对临界资源的互斥访问。简单来说，管程就是互斥锁（称之为monitor's lock）与条件变量的配合使用。下面是基本编程范式：

　　首先是线程共享的变量，分别是互斥锁、条件变量、是否需要等待的原因

mutex global_mutex 
condition global_cv
bool p

　　可能需要等待的线程（code snippet1）： 

　　

acquire(global_mutex) 
while(!p): # 如果条件不满足，则需要等待
　　global_cv.wait(global_mutex) # wait操作将当前线程阻塞在等到队列中，释放对管程的互斥访问

// do a lot of thing
release(global_mutex)

　　另外一个线程

acquire(global_mutex)
// do sth make "p" is true
global_cv.signal() # 唤醒在管程中等待的线程

release(global_mutex)

　　看了上面的代码，可能会产生两个疑问：第一个是code snippet1中，global-wait在互斥锁内，那怎么实现将当前线程阻塞的呢，必须要释放（release）global_mutex才行啊？可以看到，wait操作其中一个参数是global_mutex, 可以把wait操作理解成下面的伪码：

contidion::wait(mutex& mut){
　　release(mut)
　　yield current thread # 当前线程释放控制权
　　acquire(mut)
}

　　另外一个疑问，对条件p的判断用了while循环，为什么if不行呢？这个疑问，大多数人（包括写这篇文章之前的我:-D）都是知其然而不知其所以然。while形式称之为hansen模式，if形式称之为hoare模式。这两种模式对应着不同的管程实现，具体来说是调用signal操作时，是继续在当前线程执行，还是切换到被唤醒的线程执行？下图来自清华大学的网络课程《操作系统》：

　　可以看到，在T2线程调用x.signal之后，hansen模式会继续执行，所以当重新回到wait线程的时候，可能情况已经发生了变化，所以需要重新判断；而Hoare模式会立刻从T2线程切换到T1线程。Hansen看起来变得复杂，引入了不确定性，但是相比hoare模式，少了一次线程的切换，在真实的操作系统中就是这么实现的，所以我们编码的时候都需要用while循环判断条件是否成立。

信号量 VS 管程

　　上面的介绍可以看到，信号量与管程都能够解决我们再编程中遇到的同步互斥问题，所以，难免需要对二者进行对比。首先，二者本质上时互通的，hoare在论文中也证明了可以用信号量实现管程、也可以用管程实现信号量。下面简单归纳一下二者的区别：

• 信号量本质是可共享的资源的数量； 而管程是一种抽象数据结构用来限制同一时刻只有一个线程进入临界区

• 信号量是可以并发的，并发量取决于S初始值；而管程内部同一时刻最多只能有一个线程执行

• 信号量与管理的资源紧耦合（即信号量S的初始值等同于资源的数目，且通过P V操作修改剩余可用的资源数量）；而在管程中需自行判断是否还有可共享的资源。这一点可以参见下面生产者消费者的实现代码

• 信号量的P操作可能阻塞，也可能不阻塞；而管程的wait操作一定会阻塞

• 信号量的V操作如果唤醒了其他线程，当前线程与被唤醒线程并发执行；对于管程的signal操作，要么当前线程继续执行（Hansen），要么被唤醒线程继续执行（Hoare），二者不能并发。

　　简而言之，管程相比信号量更好实用，也更容易出错。大多数支持多线程的编程语言会同时提供管程与信号量，如Java、Python、Linux C；而在在C++11中，提供了mutex和contidion_variable，但是没有提供semaphore。管程（互斥锁 + 条件变量）是并发编程中最为常见的同步方式，在陈硕大牛的《多线程服务器的常用编程模型》一文中提到，如果需要使用底层同步原语，也是推荐互斥锁+条件变量。

生产者消费者问题

回到顶部

 　　生产者消费者是非常经典的同步模型，在编程中也有非常多的应用。生产者消费者问题描述如下：一个或者多个生产者往消息队列里面插入数据；单个消费者从消息队列里面取出数据处理；并且对消息队列的操作必须是互斥的。分析问题可知需要解决三个同步问题：

• 对消息队列的操作必须是互斥的，需要加锁（如果是lockfree的数据结构，就不用加锁，如boost::lockfree::queue）

• 消息队列中没有数据时，消费者需要等待生产者产生数据，这就是条件同步

• 消息队列满时，生产者需要等到消费者消费数据，这也是条件同步

信号量实现

 　　从上面分析生产者消费者需要解决的同步问题（互斥与条件同步），都能用信号量来解决。对于互斥，信号量S为1就可以；对于消费者等待的情况，信号量初始值为0即可；对于生产者等待的情况，信号量初始值为消息队列长度即可。linux下提供了信号量的实现，头文件在/usr/include/semaphore.h，代码实现如下(producer_consumer_semaphore.cpp)：

 1 #include <stdio.h>
 2 #include <pthread.h>
 3 #include <time.h>
 4 #include <stdlib.h>
 5 #include <unistd.h>
 6 #include <semaphore.h>
 7 
 8 #define BUFF_SIZE  3
 9 #define PRODUCE_THREAD_SIZE 5
10 int g_buff[BUFF_SIZE];
11 int g_write_index = 0;
12 int g_read_index = 0;
13 
14 sem_t lock;
15 sem_t consume_sem, produce_sem;
16 
17 
18 void* produce(void *ptr){
19     int idx = *(int*)ptr;
20     printf("in produce %d %d %d
",idx, g_write_index, g_read_index);
21     while(1){
22         sem_wait(&produce_sem); # 限制了生产者并发的数目
23 
24         sem_wait(&lock);　　　　　# 对临界区的访问要加锁
25         g_buff[g_write_index] = idx;
26         g_write_index = (g_write_index + 1) % BUFF_SIZE;
27         sem_post(&lock);
28 
29         sem_post(&consume_sem);
30     }
31     return NULL;
32 }
33 
34 void* consume(void *ptr){
35     while(1){
36         sem_wait(&consume_sem);
37         sem_wait(&lock);
38         int data = g_buff[g_read_index];
39         g_buff[g_read_index] = -1;
40         g_read_index = (g_read_index + 1) % BUFF_SIZE;
41         printf("consume %d %d %d
", data, g_read_index, g_write_index);
42         sem_post(&lock);
43         sem_post(&produce_sem);
44     }
45     return NULL;
46 }
47 
48 int main(int argc, char * argv[]){
49     pthread_t con;
50     pthread_t pros[PRODUCE_THREAD_SIZE];
51     sem_init(&lock, 0, 1);
52     sem_init(&consume_sem,0, 0);
53     sem_init(&produce_sem,0, BUFF_SIZE);
54 
55     pthread_create(&con, NULL, consume, NULL);
56     int thread_args[PRODUCE_THREAD_SIZE];
57     for(int i = 0; i < PRODUCE_THREAD_SIZE; i++){
58         thread_args[i] = i + 1;
59         pthread_create(&pros[i], NULL, produce, (thread_args + i));
60     }
61 
62     pthread_join(con,0);
63     for(int i = 0; i < PRODUCE_THREAD_SIZE; i++)
64         pthread_join(pros[i],0);
65 
66     sem_destroy(&lock);
67     sem_destroy(&consume_sem);
68     sem_destroy(&produce_sem);
69 
70     return 0;
71 }

 　　代码中，消息队列的大小为3，produce_sem的初始值一定与消息队列的大小相同。总共有5个生产者线程，多余可并发的数量（produce_sem），因此很大概率会有生产者线程阻塞在produce_sem对应的等待队列。 另外两点需要注意：第一点在produce和consume线程中都是需要加锁（互斥锁lock），因为信号量是可以并发的，需要对临界资源（g_buff,g_read_index,g_write_index)互斥访问。另外，在produce线程，需要先判断能否并发，然后再对临界区加锁，二者的顺序不能反过来，否则会产生死锁。

　　上面的代码用：g++ -lpthread producer_consumer_semaphore.cpp producer_consumer_semaphore， 然后运行 ./producer_consumer_semaphore 即可

　　

管程实现

 　　生产者消费者问题用管程（互斥锁+条件变量）来实现也很简单，而且个人感觉更直观，linux下提供了条件变量和互斥锁的实现，头文件在/usr/include/pthread.h，代码实现如下（producer_consumer_monitor.cpp）：

　　

 1 #include <stdio.h>
 2 #include <pthread.h>
 3 #include <time.h>
 4 #include <stdlib.h>
 5 #include <unistd.h>
 6 
 7 
 8 #define BUFF_SIZE  3
 9 #define PRODUCE_THREAD_SIZE 5
10 int g_buff[BUFF_SIZE];
11 int g_write_index = 0;
12 int g_read_index = 0;
13 
14 pthread_mutex_t lock;
15 pthread_cond_t consume_cond, produce_cond;
16 
17 
18 void* produce(void *ptr){
19     int idx = *(int*)ptr;
20     printf("in produce %d %d %d
",idx, g_write_index, g_read_index);
21     while(1){
22         pthread_mutex_lock(&lock);
23         while((g_write_index + 1) % BUFF_SIZE == g_read_index)
24             pthread_cond_wait(&produce_cond, &lock);
25 
26         g_buff[g_write_index] = idx;
27         g_write_index = (g_write_index + 1) % BUFF_SIZE;
28 
29         pthread_cond_signal(&consume_cond);
30         pthread_mutex_unlock(&lock);
31 
32     }
33     return NULL;
34 }
35 
36 void* consume(void *ptr){
37     while(1){
38         pthread_mutex_lock(&lock);
39         while(g_read_index == g_write_index)
40              pthread_cond_wait(&consume_cond, &lock);
41 
42         int data = g_buff[g_read_index];
43         g_buff[g_read_index] = -1;
44         g_read_index = (g_read_index + 1) % BUFF_SIZE;
45         printf("consume %d
", data);
46 
47         pthread_cond_signal(&produce_cond);
48         pthread_mutex_unlock(&lock);
49     }
50     return NULL;
51 }
52 
53 int main(int argc, char * argv[]){
54     pthread_t con;
55     pthread_t pros[PRODUCE_THREAD_SIZE];
56 
57     srand((unsigned)time(NULL));
58     pthread_mutex_init(&lock, 0);
59     pthread_cond_init(&consume_cond,0);
60     pthread_cond_init(&produce_cond,0);
61 
62     pthread_create(&con, NULL, consume, NULL);
63     int thread_args[PRODUCE_THREAD_SIZE];
64     for(int i = 0; i < PRODUCE_THREAD_SIZE; i++){
65         thread_args[i] = i + 1;
66         pthread_create(&pros[i], NULL, produce, (thread_args + i));
67     }
68 
69     pthread_join(con,0);
70     for(int i = 0; i < PRODUCE_THREAD_SIZE; i++)
71         pthread_join(pros[i],0);
72 
73     pthread_mutex_destroy(&lock);
74     pthread_cond_destroy(&consume_cond);
75     pthread_cond_destroy(&produce_cond);
76 
77     return 0;
78 }

 　　上面的代码用：g++ -lpthread producer_consumer_monitor.cpp producer_consumer_monitor， 然后运行 ./producer_consumer_monitor 即可

总结：

回到顶部

　　并发可能带来互斥、饥饿、死锁等问题，操作系统为了解决这些问题提供了很多的支持，从最底层的禁止硬件终端和原子操作指令（test-and-set），到更高级的同步原语：锁（互斥锁、自旋锁）、信号量、管程。管程是一种抽象数据结构，编程中使用互斥锁配合信号量使用。管程有两种不同的模式：hansen vs hoare，区别在于signal操作之后是否立即切换到被唤醒线程，实际的操作系统中，多使用hansen模式。