第 6 章 进程同步

背景

临界资源：一次仅允许一个进程使用的资源

竞争条件（race condition）：多个进程并发访问和操作同一数据并且执行结果与特定访问顺序有关

临界区问题

每个进程有一段代码，称为临界区（critical section），进程在执行该区时可能修改公共变量、更新一个表、写一个文件等。

临界区问题：设计一个协议以便协作进程

在进入临界区前，每个进程应请求许可，实现这一请求的代码区称为进入区（entry section），临界区之后可以有退出区（exit section），其它代码为剩余区（remainder section）。

• 临界区问题的解决方案应满足三条要求：
  • 互斥（mutual exclusion）：如果进程 $P_i$ 在临界区内执行，那么其它进程不能在其临界区内执行
  • 进步（progress）：如果没有进程在其临界区内执行，并且有进程需要进入临界区，那么只有那些不在剩余区内执行的进程可以参加选择，以便确定谁能下次进入临界区，而且这种选择不能无限延迟
  • 有限等待（bounded waiting）：从一个进程做出进入临界区的请求直到这个请求允许为止，其它进程允许进入临界区的次数具有上限

Peterson Solution

Peterson Solution 适用于两个进程（$P_i$ 和 $P_j$，$j==1-i$）交错执行临界区与剩余区。

Peterson Solution 要求两个进程共享两个数据项：

int turn;
boolean flag[2];

变量turn表示哪个进程可以进入临界区，数组flag表示哪个进程准备进入临界区。

硬件同步

对于单处理器环境，解决临界区问题可以通过关闭中断即禁止中断出现完成

多处理器的中断禁止很耗时，因为消息要传递到所有处理器，系统效率降低

许多现代操作系统提供特殊硬件指令，用于检测和修改字的内容，或者用于原子地交换两个字（作为不可中断的指令）。

bool test_and_set(bool *target) {
    bool rv = *target;
    *target = true;

    return rv;
}

如果一台机器支持指令test_and_set()，可以通过声明一个布尔变量lock，初始化为false，来实现互斥。

while(true) {
    while (test_and_set(&lock)); /*do nothing*/
    
    /*critical section*/

    lock = false;

    /*remainder section*/
}

int compare_and_swap(int *value, int expected, int new_value) {
    int temp = *value;

    if (*value == expected) *value = new_value;

    return temp;
}

声明一个全局布尔变量lock，并初始化为 0，来实现互斥。

while (true) {
    while (compare_and_swap(&lock, 0, 1) != 0); /*do nothing*/

    /*critical section*/

    lock = 0;

    /*remainder section*/
}

互斥锁（Mutex Lock）

采用互斥锁保护临界区，从而防止竞争条件

一个进程在进入临界区时应得到锁，在退出临界区时释放锁

自旋锁（spinlock）：进程不停地旋转，以等待锁变得可用

信号量（Semaphore）

一个信号量是个整型变量，除了初始化外只能通过两个标准原子操作wait()和signal()来访问。

wait(S) {
    while (S <= 0); //busy wait
    S--;
}

signal(S) {
    S++;
}

在wait()和signal()操作中，信号量整数值的修改应不可分割地执行，即当一个进程修改信号量值时，没有其它进程能够同时修改同一信号量的值。

信号量的使用

计数信号量（counting semaphore）：值不受限制

二元信号量（binary semaphore）：值只能为 0 或 1（1 表示该信号量正在被使用）

二元信号量可以提供互斥锁：

Semaphore S;  //initailized to 1
wait(S);
//critical section
signal(S);

二元信号量实现计数信号量：

struct S {
    integer val;  //initially K
    BinarySemaphore sem;  //initially 0
    integer waitCount;  //initially 0, number of processes waiting on S
    BinarySemaphore mutex;  //initially 1, protects val and waitCount
};

wait(S) {
    wait(S.mutex);
    if (S.val == 0) {
        S.waitCount += 1;
        signal(S.mutex);
        wait(S.sem);
    } else {
        S.val -= 1;
        signal(S.mutex);
    }
}

signal(S) {
    wait(S.mutex);
    if (S.waitCount > 0) {
        S.waitCount -= 1;
        signal(S.sem);
    }else {
        S.val += 1;
    }
    signal(S.mutex);
}

信号量的实现

忙等待浪费 CPU 周期，这原本可以有效用于其他进程

为了克服忙等待，可以修改wait()和signal()操作的定义，将忙等待换成阻塞自己

整型变量value用于表示资源数目

阻塞操作将一个进程放到与信号量相关的等待队列中，并且将该进程状态切换为等待状态，然后 CPU 调度程序选择执行另一个进程。

操作wakeup()可以让等待信号量而阻塞的进程在其他进程执行操作signal()之后重新执行，将进程从等待状态改为就绪状态。

wait(semaphore *S) { // 相当于申请资源    S->value--;    if (S->value < 0) { // 资源已被分配完毕        add this process to waiting queue;        block();    }}signal(semaphore *S) { // 相当于释放资源    S->value++;    if (S->value <= 0) {        remove a process P from waiting queue;        wakeup(P);    }}

死锁与饥饿

死锁（deadlock）：两个或多个进程无限等待一个事件，而该事件只能由这些等待进程之一来产生

饥饿（starvation）：进程无限等待信号量

经典同步问题

有界缓冲问题（Bounded-Buffer Problem）

int n;semaphore mutex = 1;semaphore empty = n;semaphore full = 0;

n个缓冲区，每个缓冲区可存一个数据项

信号量mutex提供缓冲池访问的互斥要求，并初始化为 1

信号量empty用于表示空的缓冲区数量，并初始化为 n

信号量full用于表示满的缓冲区数量，并初始化为 0

while (true) {    //produce an item    wait(empty);    wait(mutex);    //add the item to the buffer    signal(mutex);    signal(full);}

while (true) {    wait(full);    wait(mutex);    //remove an item from buffer    signal(mutex);    signal(empty);    //consume the removed item}

读者-写者问题（Readers and Writers Problem）

一个数据库为多个并发进程所共享，读者进程只需要读数据库，写者进程可以读和写数据库

要求写者在写入数据库时具有共享数据库独占的访问权

semaphore rw_mutex = 1;semaphore mutex = 1;int read_count = 0;

信号量rw_mutex供写者作为互斥信号量，也为第一个进入临界区和最后一个离开临界区的读者所使用，初始化为 1

信号量mutex用于确保在更新变量read_count时互斥，初始化为 1

变量read_count用于跟踪多少进程正在读对象，初始化为 0

while (true) {    wait(rw_mutex)    //writing is performed    signal(rw_mutex);}

while (true) {    wait(mutex);    read_count++;    if (read_count == 1) wait(rw_mutex);    signal(mutex);    //reading is performed    wait(mutex);    read_count--;    if (read_count == 0) signal(rw_mutex);    signal(mutex);}

哲学家就餐问题（Dining-Philosophers Problem）

semaphore chopstick[5];

chopstick的所有元素都初始化为 1

while (true) {    wait(chopstick[i]);    wait(chopstick[(i+1)%5];    //eat    signal(chopstick[i]);    signal(chopstick[(i+1)%5];    //think}

如果五个哲学家同时拿起左边的筷子将导致死锁，有多种可能的补救措施

没有死锁的解决方案不一定能消除饥饿的可能性

管程（Monitor）

虽然信号量提供了一种方便且有效的进程同步机制，但它们的错误使用可能导致难以检测的时序错误，因为这些错误只有在特定执行顺序时才会出现，而这些顺序并不总是出现。

编译器无法检查语义逻辑层面的错误

使用方法

管程类型（monitor type）属于抽象数据类型，提供了一组由程序员定义的、在管程内互斥的操作，把对共享资源的操作封装起来

monitor monitor_name {    /*shared variable declarations*/    function P1(...) {        ...    }    function P2(...) {        ...    }    ...    function Pn(...) {        ...    }    initialization_code(...) {        ...    }}

一个进程只有调用管程内的函数才能进入管程访问共享资源

管程结构确保每次只有一个进程在管程内处于活动状态

管程的示意图

当需要编写定制的同步方案时，可定义一个或多个类型为condition的变量，对应一个或多个进程被阻塞的原因

每个条件变量保存一个等待队列，用于记录因该条件变量而阻塞的所有进程

条件变量有操作wait()和signal()可调用

x.wait()：当x对应的条件不满足时，正在调用管程的进程调用x.wait()将自己插入x条件的等待队列并释放管程。此时其它进程可以使用该管程

x.signal()：x对应的条件发生了变化，则调用x.signal()，唤醒一个因x条件而阻塞的进程

• 假设操作x.signal()被一个进程 $P$ 调用，x 上有一个挂起的进程 $Q$，存在两种可能性：
  • 唤醒并等待：进程 $P$ 等待直到 $Q$ 离开管程，或者等待另一个条件
  • 唤醒并继续：进程 $Q$ 等待直到 $P$ 离开管程，或者等待另一个条件

条件变量与信号量的相似点：wait/signal 操作类似，可以实现进程的阻塞/释放

条件变量与信号量的不同点：条件变量是没有值的，仅实现了排队等待的功能；而信号量是有值的，反映了剩余资源数。在管程中，剩余资源数用共享数据结构记录

哲学家就餐问题的管程解决方案

强加限制：只有当一个哲学家的两根筷子都可用时才能拿起筷子

monitor DiningPhilosophers   {    enum { THINKING, HUNGRY, EATING) state [5] ;    condition self [5];  //philosopher i can delay herself when unable to get chopsticks    void pickup (int i) {        state[i] = HUNGRY;        test(i);        if (state[i] != EATING) self [i].wait;    }    void putdown (int i) {        state[i] = THINKING;        // test left and right neighbors        test((i + 4) % 5);        test((i + 1) % 5);    }    void test (int i) {        if ( (state[(i + 4) % 5] != EATING) &&        (state[i] == HUNGRY) &&        (state[(i + 1) % 5] != EATING) ) {            state[i] = EATING;	          self[i].signal();	        }	    }    initialization_code() {        for (int i = 0; i < 5; i++)            state[i] = THINKING;    }}

哲学家 i应按如下顺序来调用操作pickup()和putdown()：

DiningPhilosophers.pickup(i);...EAT...DiningPhilosophers.putdown(i);

无死锁，但可能饥饿

采用信号量的管程实现

对于每个管程，都有一个信号量mutex，初始化为 1

进程在进入管程前应执行wait(mutex)，在离开管程之后应执行signal(mutex)

由于唤醒进程必须等待，直到重新启动的进程离开或等待，所以引入了一个额外的信号量next，初始化为 0。唤醒进程可使用next来挂起自己。

整型变量next_count用于对在next上挂起的进程进行计数

每个外部函数F替换为：

wait(mutex);.../*body of F*/...if (next_count > 0)    signal(next);else    signal(mutex);

确保了管程内的互斥

每个条件变量x有一个信号量x_sem和一个整型变量x_count，均初始化为 0

x.wait()实现如下：

x_count++;if (next_count > 0)    signal(next);else    signal(mutex);wait(x_sem);x_count--;

x.signal()实现如下：

if (x_count > 0) {    next_count++;    signal(x_sem);    wait(next);    next.count--;}

管程内的进程重启