经典同步问题

哲学家就餐问题

当初我在校招的时候，面试官也问过「哲学家就餐」这道题目，我当时听的一脸懵逼，无论面试官怎么讲述这个问题，我也始终没听懂，就莫名其妙的说这个问题会「死锁」。

当然，我这回答槽透了，所以当场 game over，残酷又悲惨故事，就不多说了，反正当时菜就是菜。

时至今日，看我来图解这道题。

先来看看哲学家就餐的问题描述：

• 5 个老大哥哲学家，闲着没事做，围绕着一张圆桌吃面；
• 巧就巧在，这个桌子只有 5 支叉子，每两个哲学家之间放一支叉子；
• 哲学家围在一起先思考，思考中途饿了就会想进餐；
• 奇葩的是，这些哲学家要两支叉子才愿意吃面，也就是需要拿到左右两边的叉子才进餐；
• 吃完后，会把两支叉子放回原处，继续思考；

那么问题来了，如何保证哲学家们的动作有序进行，而不会出现有人永远拿不到叉子呢？

方案一，拿叉子是互斥事件

我们用信号量的方式，也就是 PV 操作来尝试解决它，代码如下：

#define N 5					// 哲学家个数
semaphore fork[5];			// 信号量初值为1 (信号量数组)
							// 也就是叉子的个数

void smart_person(int i) {	// i 为哲学家编号 0-4
	while(TRUE) {
        think();			// 哲学家思考
        P(fork[i]);			// 去拿左边的又子
        P(fork[(i+1)%N]);	// 去拿右边的叉子
        eat();				// 进餐
        V(fork[i]);			// 放下左边的叉子
        V(fork[(i+1)%N]);	// 放下右边的叉子
	}
}

上面的程序，好似很自然。拿起叉子用 P 操作，代表有叉子就直接用，没有叉子时就等待其他哲学家放回叉子。

不过，这种解法存在一个极端的问题：假设五位哲学家同时拿起左边的叉子，桌面上就没有叉子了， 这样就没有人能够拿到他们右边的叉子，也就说每一位哲学家都会在 P(fork[(i + 1) % N ]) 这条语句阻塞了，很明显这发生了死锁的现象。

方案二，吃饭是互斥事件

既然「方案一」会发生同时竞争左边叉子导致死锁的现象，那么我们就在拿叉子前，加个互斥信号量，代码如下：

#define N 5					// 哲学家个数
semaphore fork[5];			// 每个叉子一个信号量，初值为1 (信号量数组)
semaphore mutex;			// 互斥信号量，初值为 1

void smart_person(int i) {	// i 为哲学家编号 0-4
	while(TRUE) {
        think();			// 哲学家思考
        P(mutex);			// 进入临界区
        P(fork[i]);			// 去拿左边的叉子
        P(fork[(i+1)%N]);	// 去拿右边的又子
        eat();				// 进餐
        V(fork[i]);			// 放下左边的又子
        V(fork[(i+1)%N]);	// 放下右边的又子
        V(mutex);			// 退出临界区
    }
}

上面程序中的互斥信号量的作用就在于，只要有一个哲学家进入了「临界区」，也就是准备要拿叉子时，其他哲学家都不能动，只有这位哲学家用完叉子了，才能轮到下一个哲学家进餐。

方案二虽然能让哲学家们按顺序吃饭，但是每次进餐只能有一位哲学家，而桌面上是有 5 把叉子，按道理是能可以有两个哲学家同时进餐的，所以从效率角度上，这不是最好的解决方案。

方案三，奇偶调整

那既然方案二使用互斥信号量，会导致只能允许一个哲学家就餐，那么我们就不用它。

另外，方案一的问题在于，会出现所有哲学家同时拿左边刀叉的可能性，那我们就避免哲学家可以同时拿左边的刀叉，采用分支结构，根据哲学家的编号的不同，而采取不同的动作。

即让偶数编号的哲学家「先拿左边的叉子后拿右边的叉子」，奇数编号的哲学家「先拿右边的叉子后拿左边的叉子」。

#define N 5					// 哲学家个数
semaphore fork[5];			// 每个叉子一个信号量，初值为1 (信号量数组)

void smart_person(int i) {	// i 为哲学家编号 0-4
	while(TRUE) {
        think();			// 哲学家思考
        
        if(i%2 == 0){
            P(fork[i]);			// 去拿左边的叉子
            P(fork[(i+1)%N]);	// 去拿右边的叉子
        } else {
            P(fork[(i+1)%N]);	// 去拿右边的叉子
            P(fork[i]);			// 去拿左边的叉子
        }
        
        eat();				// 进餐
        V(fork[i]);			// 放下左边的又子
        V(fork[(i+1)%N]);	// 放下右边的又子
    }
}

上面的程序，在 P 操作时，根据哲学家的编号不同，拿起左右两边叉子的顺序不同。另外，V 操作是不需要分支的，因为 V 操作是不会阻塞的。

方案三即不会出现死锁，也可以两人同时进餐。

方案四，状态码+信号量数组

在这里再提出另外一种可行的解决方案，我们用一个数组 state 来记录每一位哲学家的三个状态，分别是在进餐状态、思考状态、饥饿状态（正在试图拿叉子）。

那么，一个哲学家只有在两个邻居都没有进餐时，才可以进入进餐状态。

第 i 个哲学家的左邻右舍，则由宏 LEFT 和 RIGHT 定义：

• LEFT : ( i + 5 - 1 ) % 5
• RIGHT : ( i + 1 ) % 5

比如 i 为 2，则 LEFT 为 1，RIGHT 为 3。

具体代码实现如下：

#define N 5					// 哲学家个数
#define LEFT(i+N-1)%N		// i的左边邻居编号
#define RIGHT(i+1)% N		// i 的右边邻居编号

#define THINKING O			// 思考状态
#define HUNGRY 1			// 饥饿状态
#define EATING 2			// 进餐状态

int state[N];				// 数组记录每个哲学家的状态
semaphore s[5];				// 每个哲学家一个信号量，初值 0 (信号量数组)
semaphore mutex;			// 互斥信号量，初值为 1

void test(int i) {			// i为哲学家编号 0-4
	// 如果i号的左边右边哲学家都不是进餐状态，把i号哲学家标记为进餐状态
    if(state[i] == HUNGRY &&
        State[LEFT] != EATING &&
        State[RIGHT] != EATING)
    {
        state[i] = EATING	// 两把叉子到手，进餐状态
        V(s[i]);			// 通知第 i 哲学家可以进餐了
    }
}
    
// 功能:要么拿到两把叉子，要么被阻塞起来
void take_forks(int i) { 	// i为哲学家编号 0-4
	P(mutex);				// 进入临界区
	state[i] = HUNGRY;		// 标记哲学家处于饥饿状态
	test(i);				// 尝试获取 2 支叉子
	V(mutex);				// 离开临界区
	P(s[i]);				// 没有叉子则阻塞，有叉子则继续正常执行
}

// 功能:把两把叉子放回原处，并在需要的时候，去唤醒左邻右舍
void put forks(int i) {		// i为哲学家编号 0-4
	P(mutex);				// 进入临界区
    state[i]= THINKING;		// 吃完饭了，交出叉子，标记思考状态
    test(LEFT);				// 检查左边的左邻右舍是否在进餐，没则唤醒
    test(RIGHT);			// 检查右边的左邻右舍是否在进餐，没则唤醒
    V(mutex);				// 离开临界区
}

// 哲学家主代码
void smart_person(int i) {	// i为哲学家编号 0-4
	while(TRUE) {
        think();			// 思考
        take_forks(i);		// 准备拿去叉子吃饭
        eat();				// 就餐
        put_forks(i);		// 吃完放回叉子
    }
}

上面的程序使用了一个信号量数组，每个信号量对应一位哲学家，这样在所需的叉子被占用时，想进餐的哲学家就被阻塞。

注意，每个进程/线程将 smart_person 函数作为主代码运行，而其他 take_forks、put_forks 和 test 只是普通的函数，而非单独的进程/线程。

方案四同样不会出现死锁，也可以两人同时进餐

读者-写者问题

前面的「哲学家进餐问题」对于互斥访问有限的竞争问题（如 I/O 设备）一类的建模过程十分有用。

另外，还有个著名的问题是「读者-写者」，它为数据库访问建立了一个模型。

读者只会读取数据，不会修改数据，而写者即可以读也可以修改数据。

读者-写者的问题描述：

• 「读-读」允许：同一时刻，允许多个读者同时读
• 「读-写」互斥：没有写者时读者才能读，没有读者时写者才能写
• 「写-写」互斥：没有其他写者时，写者才能写

（话说这个有点像读写锁问题）

接下来，提出几个解决方案来分析分析。

方案一，读者优先

使用信号量的方式来尝试解决：

• 信号量 wMutex：控制写操作的互斥信号量，初始值为 1 ；
• 读者计数 rCount：正在进行读操作的读者个数，初始化为 0；
• 信号量 rCountMutex：控制对 rCount 读者计数器的互斥修改，初始值为 1；

接下来看看代码的实现：

semaphore wMutex;			// 控制写操作的互斥信号量，初始值为 1

int rCount = 0;				// 正在进行读操作的读者个数，初始化为0
semaphore rCountMutex;		// 控制对 Rcount 的互斥修改，初始值为 1

// 写者进程/线程执行的函数
void writer() {
	while(TRUE) {
		P(wMutex);			// 进入临界区
		write();
		V(wMutex);			// 进入临界区
    }
}

// 读者进程/线程执行的函数
void reader() {
    while(TRUE) {
        P(rCountMutex);		// 进入临界区
        if(rCount == 0) {
        	P(wMutex);		// 如果有写者，则阻塞写者
        }
        rCount++;			// 读者计数 +1
        V(rCountMutex);		// 离开临界区
        
        read( );			// 读数据
        
        P(rCountMutex);		// 进入临界区
        rCount--;			// 读完数据，准备离开
        if(rCount == 0) {
        	V(wMutex);		// 最后一个读者离开了，则唤醒写者
        }
        V(rCountMutex);		// 离开临界区
    }
}

上面的这种实现，是读者优先的策略，因为只要有读者正在读的状态，后来的读者都可以直接进入，如果读者持续不断进入，则写者会处于饥饿状态。

方案二，写者优先

那既然有读者优先策略，自然也有写者优先策略：

• 只要有写者准备要写入，写者应尽快执行写操作，后来的读者就必须阻塞；
• 如果有写者持续不断写入，则读者就处于饥饿；

在方案一的基础上新增如下变量：

• 信号量 rMutex：控制读者进入的互斥信号量，初始值为 1；
• 信号量 wDataMutex：控制写者写操作的互斥信号量，初始值为 1；
• 写者计数 wCount：记录写者数量，初始值为 0；
• 信号量 wCountMutex：控制 wCount 互斥修改，初始值为 1；

具体实现如下代码：

int rCount = 0;				// 正在进行读操作的读者个数，初始化为0
semaphore rCountMutex;		// 控制对 Rcount 的互斥修改，初始值为 1
semaphore rMutex;			// 控制读者进入的互斥信号量者，初始值为1

int wCount = 0;				// 正在进行读操作的写者个数，初始化为0
semaphore wCountMutex		// 控制 wCount 互斥修改，初始值为 1
semaphore wDataMutex		// 控制写者写操作的互斥信号量，初始值为 1

// 写者进程/线程执行的函数
void writer() {
	while(TRUE) {
        P(wCountMutex);		// 进入临界区
        if(wCount == 0) {
        	P(rMutex); 		// 当第一个写者进入，如果有读者则阻塞读者
        }
        wCount++;			// 写者计数 + 1
        V(wCountMutex);		// 离开临界区
        
        P(wDataMutex);		// 写者写操作之间互斥，进入临界区
        write();			// 写数据
        V(wDataMutex);		// 离开临界区
        
        P(wCountMutex);		// 进入临界区
        wCount--;			// 写完数据，准备离开
        if( wCount == 0) {
        	V(rMutex);		// 最后一个写者离开了，则唤醒读者
        }
        V(wCountMutex);		// 离开临界区
	}
}
   
// 读者进程/线程执行的函数
void reader() {
	while(TRUE) {
		P(rMutex);
		P(rCountMutex);		// 进入临界区
		if(rCount == 0) {
			P(wDataMutex);	// 当第一个读者进入，如果有写者则阻塞写者写操作
        }
		rCount++;
		V(rCountMutex);		// 离开临界区
		V(rMutex);
        
		read( );			// 读数据
		
        P(rCountMutex);		// 进入临界区
		rCount--;
		if(rCount == 0) {
			V(wDataMutex);	// 当没有读者了，则唤醒阻塞中写者的写操作
        }
        V(rCountMutex);		// 离开临界区
    }
}

注意，这里 rMutex 的作用，开始有多个读者读数据，它们全部进入读者队列，此时来了一个写者，执行了 P(rMutex) 之后，后续的读者由于阻塞在 rMutex 上，都不能再进入读者队列，而写者到来，则可以全部进入写者队列，因此保证了写者优先。

同时，第一个写者执行了 P(rMutex) 之后，也不能马上开始写，必须等到所有进入读者队列的读者都执行完读操作，通过 V(wDataMutex) 唤醒写者的写操作。

方案三，公平策略

既然读者优先策略和写者优先策略都会造成饥饿的现象，那么我们就来实现一下公平策略。

公平策略：

• 优先级相同；
• 写者、读者互斥访问；
• 只能一个写者访问临界区；
• 可以有多个读者同时访问临界资源；

具体代码实现：

int rCount = 0;				// 正在进行读操作的读者个数，初始化为 0
semaphore rCountMutex;		// 控制对 Rcount 的互斥修改，初始值为 1
semaphore wDataMutex;		// 控制写者写操作的互斥信号量，初始值为 1
semaphore flag;				// 用于实现公平竞争，初始值为 1

// 写者进程/线程执行的函数
void writer() {
	while(TRUE) {
		P(flag);
		P(wDataMutex);		// 写者写操作之间互斥，进入临界区
		write();			// 写数据
		V(wDataMutex);		// 离开临界区
        V(flag);
    }
}

// 读者进程/线程执行的函数
void reader() {
	while(TRUE) {
		P(flag);
		P(rCountMutex);		// 进入临界区
		if(rCount ==0) {
			P(wDataMutex); // 当第一个读者进入，如果有写者则阻塞写者写操作
        }
        rCount++:
		V(rCountMutex);		// 离开临界区
		V(flag);

		read(); 			// 读数据
		P(rCountMutex);		// 进入临界区
		rCount--:
		if(rCount == 0) {
			V(wDataMutex);	// 当没有读者了，则唤醒阻塞中写者的写操作
        }
		V(rCountMutex);		// 离开临界区
    }
}

看完代码不知你是否有这样的疑问，为什么加了一个信号量 flag，就实现了公平竞争？

对比方案一的读者优先策略，可以发现，读者优先中只要后续有读者到达，读者就可以进入读者队列， 而写者必须等待，直到没有读者到达。

没有读者到达会导致读者队列为空，即 rCount==0，此时写者才可以进入临界区执行写操作。

而这里 flag 的作用就是阻止读者的这种特殊权限（特殊权限是只要读者到达，就可以进入读者队列）。

比如：开始来了一些读者读数据，它们全部进入读者队列，此时来了一个写者，执行 P(falg) 操作，使得后续到来的读者都阻塞在 flag 上，不能进入读者队列，这会使得读者队列逐渐为空，即 rCount 减为 0。

这个写者也不能立马开始写（因为此时读者队列不为空），会阻塞在信号量 wDataMutex 上，读者队列中的读者全部读取结束后，最后一个读者进程执行 V(wDataMutex)，唤醒刚才的写者，写者则继续开始进行写操作。