进程间通信

进程间需要通信，我们需要设计一了良好的结构，不使用中断的方式实现。在本节中将讨论进程间通信（Inter Process Comminucation, IPC）的问题。

竞争条件

什么是竞争条件？

竞争条件 (race condition)：两个或多个进程共同读写某些共享资源，而最后的执行解决取决于进行运行时间的精确时序时，这种情况称为竞争条件。

怎样避免竞争条件？

互斥 (mutual exclusion)：以某种手段确保当一个进程在使用一个资源时，其他进程就不能对资源的做同样的操作；
我们把共享的内存进行访问的程序片段称作临界区域 (critical region)。如果我们通过合适的安排使得两个进程不可能同时处于临界区，就能够避免竞争条件。

忙等待的互斥

下面列举的这些实现互斥的方案，绝对性地禁止了两个进程共享一个资源：

屏蔽中断：顾名思义，一个进程或线程进入临界区域之后立即屏蔽所有中断，离开之前再打开中断；
锁变量：一种软件层面的解决方案。即设置一个共享的锁变量，其初始值为 0：
1. 一个线程想要进入临界区之前需要先测试锁变量的值，为 1 则继续等待；
2. 如果为 0，则线程进入这个临界区，并且将锁变量的值设置为 1，离开时重置；
显然，但其实这种方法仍然会有概率导致多个线程同时进入临界区。
严格轮换法：设置一个变量用于记录当前可以进入临界区的线程，各个线程连续测试这个变量是否出现给定的值。这种方式称为忙等待 (busy waiting)，忙于等待的锁称为自旋锁 (spin lock)。
由于这种方式是浪费 CPU 时间的，通常会避免。只有在有理由认为等待时间非常短的情况下，才使用。

Peterson 解法：荷兰数学家 T.Dekker 提出的一种不需要严格轮换的软件互斥算法。后来 Peterson 发现了一种更简单的互斥算法。

这个算法的核心是 enter_region 与 leave_rigion 这样两个函数，比如我们有 2 个线程：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
#define N 2				// 表示一共有两个线程，下面两个函数的传参都是指线程号，共有 01 两种取值
int turn;					// 当前实际可以占用资源的进程号
int interested[N];

void enter_region(int process) {
  int other = 1 - process;			// 另一个进程号
  interested[other] = TRUE;
  turn = process;
  while(turn==process && interested[other] == TRUE);
}

void leave_region(int process) {
  interested[process] = FALSE;
}

如果一个线程调用了 enter_region 函数，但是并没有获得 turn 的赋值，那么说明这个线程获取资源失败，可以再次调用函数，则函数会进入等待状态。

TSL 指令：指 Test and Set Lock，测试并加锁。它的汇编格式如下：
1
TSL RX,LOCK
它将一个内存字 LOCK 读取到 RX 中，并且在 LOCK 上置放一个非零值，这两步读写操作是不可分割的。

上面列举的这些方法都是正确的，但是它们都有忙等的缺点，也就是说：

当一个进程想要进入临界区时，先检查是否会允许进入，若不允许，该进程将原地等待直到允许为止；

解决这个缺点最简单的方法是 sleep 和 wakeup：

sleep：一个将引起进程阻塞的系统调用，即被挂起，知道收到被唤醒的信号；
wakeup：也就是唤醒一个指定进程的操作。

消费者生产者问题

问题

考虑这样一个问题：两个进程共享一个缓冲区。其中一个生产者，将信息放入缓冲区；另一个是消费者，将信息从缓冲区中取出。一个比较正常的处理办法是以下的方式：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
#define N 100					// 缓冲区的数目
int count = 0;				// 当前存在信息的缓冲区数目

void producer() {
 int item;
 while(TRUE) {
   item = produce_item();
   if (count == N) sleep();
   insert_item(item);
   count += 1;
   if (count == 1) wakeup(consumer);
 }
}

void consumer() {
 int item;
 while(TRUE) {
   if (count == 0) sleep();
   item = remove_item();
   count -= 1;
   if (count == N-1) wakeup(producer);
   consume_item(item);
 }
}

这种处理办法看起来很好，但是它是存在问题的：

如果消费者检测到 0，sleep 函数还没开始执行，生产者发送的 wakeup 信号就已经到达，那么传递给消费者的 wakeup 信号将丢失，两个线程有可能永久沉睡下去。
这个问题就叫做消费者生产者问题。

信号量

Dijkstra 引入了一个新的变量类型信号量 (semaphore) 用来记录线程被唤醒的次数。他建议设定两种操作，将其命名为 down 和 up，分别对应一个线程的 sleep 与 wakeup 函数，伪代码大致如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
int semaphore = 0;		// 信号量

void down() {
  while (TRUE) {
    if (semaphore > 0) {
      semaphore -= 1;
      return;
    }
    sleep();
  }
}

void up() {
  semaphore += 1;
  wakeup();
}

引入信号量的概念可以解决消费者生产者问题：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#define N 100
typedef int semaphore;

semaphore full = 0;
semaphore empty = N;
semaphore mutex = 1;

void producer() {
 int item;
 while (TURE) {
   item = produce_item();
   down(&empty);
   down(&mutex); insert_item(item); up(&mutex);
   up(&full);
 }
}

void consumer() {
 int item;
 while (TRUE) {
   down(&full);
   down(&mutex); item = remove_item(); up(&mutex);
   up(&empty);
   consume_item(item);
 }
}

解决方案中一共提供了三个信号量：
1. 一个称为 full，用来记录被占用缓冲区的数目，初始值为 0；
2. 一个称为 empty，用来记录空闲缓冲区的数目，初始值为 N（缓冲区的总数）；
3. 一个称为 mutex，锁变量，用来确保消费者生产者不会同时访问缓冲区。

信号量的另一种用途是实现同步 (synchronization)。

互斥量

如果不需要信号量的计数能力，有时可以使用信号量的简化版本互斥量 (mutex)。它仅仅适用于管理或共享一小段代码，因为互斥量实现时既容易又有效，这在实现用户线程包的时候非常有用。