多线程资源竞争的互斥与同步方法(转载)

多线程资源竞争的互斥与同步方法(转载)

对于共享资源,如果没有上锁,在多线程的环境里,那么就可能会发生并发访问的问题。

一、竞争与协作

在单核 CPU 系统里,为了实现多个程序同时运行的假象,操作系统通常以时间片调度的方式,让每个进程每次执行一个时间片,时间片用完了,就切换下一个进程运行,由于这个时间片的时间很短,于是就造成了「并发」的现象。

并发

另外,操作系统也为每个进程创建巨大、私有的虚拟内存的假象,这种地址空间的抽象让每个程序好像拥有自己的内存,而实际上操作系统在背后秘密地让多个地址空间「复用」物理内存或者磁盘。

虚拟内存管理-换入换出

如果一个程序只有一个执行流程,也代表它是单线程的。当然一个程序可以有多个执行流程,也就是所谓的多线程程序,线程是调度的基本单位,进程则是资源分配的基本单位。

多线程

那么问题就来了,多个线程竞争共享资源,如果不采取有效的措施,则会造成共享数据的混乱。

我们做个小实验,创建两个线程,它们分别对共享变量 i 自增 1 执行 10000 次,如下代码(虽然说是 C++ 代码,但是没学过 C++ 的同学也是看得懂的)

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
27
28
29
30
31
32
33
34
#include <iostream> // std::cout
#include <thread> // std::tread

int i = 0;

// 线程函数,对共享变量 i 自增 1 执行 10000 次
void test()
{
int num = 10000;

for(int n = 0; n < num; n++)
{
i = i + 1;
}
}

int main(void)
{
std::cout << "Start all threads." << std::endl;

// 创建线程
std::thread thread_test1(test);
std::thread thread_test2(test);

// 等待线程执行完成
thread_test1.join();
thread_test2.join();

std::cout << "All threads joined." << std::endl;

std::cout << "now i is" << i << std::endl;

return 0;
}

按理来说,i 变量最后的值应该是 20000,但很不幸,并不是如此。我们对上面的程序执行一下:

1
2
3
4
5
6
7
8
9
ubuntu@VM-0-9-ubuntu:~$ ./test_thread 
Start all threads.
All threads joined.
now i is 20000

ubuntu@VM-0-9-ubuntu:~$ ./test_thread
Start all threads.
All threads joined.
now i is 15173

运行了两次,发现出现了 i 值的结果是 15173,也会是 20000 的 i 值情况。

每次运行不但会产生错误,而且得到不同的结果。在计算机里是不能容忍的,虽然是小概率出现的错误,但是小概率事件它一定是会发生的,「墨菲定律」大家都懂吧。

为什么会发生这种情况?

为了理解为什么会发生这种情况,我们必须了解编译器为更新计数器 i 变量生成的代码序列,也就是要了解汇编指令的执行顺序。

在这个例子中,我们只是想给 i 加上数字 1,那么它对应的汇编指令执行过程是这样的:

可以发现,只是单纯给 i 加上数字 1,在 CPU 运行的时候,实际上要执行 3 条指令。

设想我们的线程 1 进入这个代码区域,它将 i 的值(假设此时是 50 )从内存加载到它的寄存器中,然后它向寄存器加 1,此时在寄存器中的 i 值是 51。

现在,一件不幸的事情发生了:时钟中断发生。因此,操作系统将当前正在运行的线程的状态保存到线程的线程控制块 TCP。

现在更糟的事情发生了,线程 2 被调度运行,并进入同一段代码。它也执行了第一条指令,从内存获取 i 值并将其放入到寄存器中,此时内存中 i 的值仍为 50,因此线程 2 寄存器中的 i 值也是 50。假设线程 2 执行接下来的两条指令,将寄存器中的 i 值 + 1,然后将寄存器中的 i 值保存到内存中,于是此时全局变量 i 值是 51。

最后,又发生一次上下文切换,线程 1 恢复执行。还记得它已经执行了两条汇编指令,现在准备执行最后一条指令。回忆一下, 线程 1 寄存器中的 i 值是51,因此,执行最后一条指令后,将值保存到内存,全局变量 i 的值再次被设置为 51。

简单来说,增加 i (值为 50 )的代码被运行两次,按理来说,最后的 i 值应该是 52,但是由于不可控的调度,导致最后 i 值却是 51。

蓝色表示线程 1 ,红色表示线程 2

1.1、互斥的概念

上面展示的情况称为竞争条件race condition),当多线程相互竞争操作共享变量时,由于运气不好,即在执行过程中发生了上下文切换,我们得到了错误的结果,事实上,每次运行都可能得到不同的结果,因此输出的结果存在不确定性indeterminate)。

由于多线程执行操作共享变量的这段代码可能会导致竞争状态,因此我们将此段代码称为临界区critical section),它是访问共享资源的代码片段,一定不能给多线程同时执行

我们希望这段代码是互斥(mutualexclusion)的,也就说保证一个线程在临界区执行时,其他线程应该被阻止进入临界区,说白了,就是这段代码执行过程中,最多只能出现一个线程。

互斥

另外,说一下互斥也并不是只针对多线程。在多进程竞争共享资源的时候,也同样是可以使用互斥的方式来避免资源竞争造成的资源混乱。

1.2、同步的概念

互斥解决了并发进程/线程对临界区的使用问题。这种基于临界区控制的交互作用是比较简单的,只要一个进程/线程进入了临界区,其他试图想进入临界区的进程/线程都会被阻塞着,直到第一个进程/线程离开了临界区。

我们都知道在多线程里,每个线程并一定是顺序执行的,它们基本是以各自独立的、不可预知的速度向前推进,但有时候我们又希望多个线程能密切合作,以实现一个共同的任务。

例子,线程 1 是负责读入数据的,而线程 2 是负责处理数据的,这两个线程是相互合作、相互依赖的。线程 2 在没有收到线程 1 的唤醒通知时,就会一直阻塞等待,当线程 1 读完数据需要把数据传给线程 2 时,线程 1 会唤醒线程 2,并把数据交给线程 2 处理。

所谓同步,就是并发进程/线程在一些关键点上可能需要互相等待与互通消息,这种相互制约的等待与互通信息称为进程/线程同步

举个生活的同步例子,你肚子饿了想要吃饭,你叫妈妈早点做菜,妈妈听到后就开始做菜,但是在妈妈没有做完饭之前,你必须阻塞等待,等妈妈做完饭后,自然会通知你,接着你吃饭的事情就可以进行了。

吃饭与做菜的同步关系

注意,同步与互斥是两种不同的概念:

  • 同步就好比:「操作 A 应在操作 B 之前执行」,「操作 C 必须在操作 A 和操作 B 都完成之后才能执行」等;
  • 互斥就好比:「操作 A 和操作 B 不能在同一时刻执行」;

二、互斥与同步的实现和使用

在进程/线程并发执行的过程中,进程/线程之间存在协作的关系,例如有互斥、同步的关系。

为了实现进程/线程间正确的协作,操作系统必须提供实现进程协作的措施和方法,主要的方法有两种:

  • :加锁、解锁操作;
  • 信号量:P、V 操作;

这两个都可以方便地实现进程/线程互斥,而信号量比锁的功能更强一些,它还可以方便地实现进程/线程同步。

2.1、锁

使用加锁操作和解锁操作可以解决并发线程/进程的互斥问题。

任何想进入临界区的线程,必须先执行加锁操作。若加锁操作顺利通过,则线程可进入临界区;在完成对临界资源的访问后再执行解锁操作,以释放该临界资源。

加锁-解锁

根据锁的实现不同,可以分为「忙等待锁」和「无忙等待锁」。

「忙等待锁」的实现

在说明「忙等待锁」的实现之前,先介绍现代 CPU 体系结构提供的特殊原子操作指令 —— 测试和置位(Test-and-Set)指令

如果用 C 代码表示 Test-and-Set 指令,形式如下:

1
2
3
4
5
6
int TestAndSet(int* old_ptr, int new)
{
int old = *old_ptr;
*old_ptr = new;
return old;
}

测试并设置指令做了下述事情:

  • old_ptr 更新为 new 的新值;
  • 返回 old_ptr 的旧值。

当然,关键是这些代码是原子执行。因为既可以测试旧值,又可以设置新值,所以我们把这条指令叫作「测试并设置」。

那什么是原子操作呢?原子操作就是要么全部执行,要么都不执行,不能出现执行到一半的中间状态

我们可以运用 Test-and-Set 指令来实现「忙等待锁」,代码如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
typedef struct lock_t
{
int flag;
}lock_t;

void init(lock_t* lock)
{
lock->flag = 0;
}

void lock(lock_t* lock)
{
while(TestAndSet(&lock->flag, 1) == 1); // do nothing
// 临界区
}

void unlock(lock_t* lock)
{
lock->flag = 0;
}

我们来确保理解为什么这个锁能工作:

  • 第一个场景是,首先假设一个线程在运行,调用 lock(),没有其他线程持有锁,所以 flag 是 0。当调用 TestAndSet(flag, 1) 方法,返回 0,线程会跳出 while 循环,获取锁。同时也会原子的设置 flag 为 1,标志锁已经被持有。当线程离开临界区,调用 unlock()flag 清理为 0。

  • 第二种场景是,当某一个线程已经持有锁(即 flag 为 1)。本线程调用 lock(),然后调用 TestAndSet(flag, 1),这一次返回 1。只要另一个线程一直持有锁,TestAndSet() 会重复返回 1,本线程会一直忙等。当 flag 终于被改为 0,本线程会调用 TestAndSet(),返回 0 并且原子地设置为 1,从而获得锁,进入临界区。

很明显,当获取不到锁时,线程就会一直 wile 循环,不做任何事情,所以就被称为「忙等待锁」,也被称为自旋锁spin lock)。

这是最简单的一种锁,一直自旋,利用 CPU 周期,直到锁可用。在单处理器上,需要抢占式的调度器(即不断通过时钟中断一个线程,运行其他线程)。否则,自旋锁在单 CPU 上无法使用,因为一个自旋的线程永远不会放弃 CPU。

「无等待锁」的实现

无等待锁顾明思议就是获取不到锁的时候,不用自旋。

既然不想自旋,那当获取不到锁的时候,就把当前线程放入到锁的等待队列,然后执行调度程序,把 CPU 让给其他线程执行。

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
27
28
29
30
31
32
33
34
35
typedef struct lock_t
{
int flag;
queue_t* q; // 等待队列
}lock_t;

void init(lock_t* lock)
{
lock->flag = 0;
queue_init(lock->q);
}

void lock(lock_t* lock)
{
while(TestAndSet(&lock->flag, 1) == 1)
{
// 保存现在运行线程 TCB;
// 将现在运行的线程 TCB 插入到等待队列;
// 设置该线程为等待状态;
// 调度程序;
}
// 临界区
}

void unlock(lock_t* lock)
{
if(lock->q != NULL)
{
// 移出等待队列的对头元素;
// 将该线程的 TCB 插入到就绪队列;
// 设置该线程为就绪状态;
}

lock->flag = 0;
}

本例只是提出了两种简单锁的实现方式。当然,在具体操作系统实现中,会更复杂,但也离不开本例两个基本元素。

如果你想要对锁的更进一步理解,推荐大家可以看《操作系统导论》第 28 章锁的内容,这本书在「微信读书」就可以免费看。

2.1、信号量

信号量是操作系统提供的一种协调共享资源访问的方法。

通常信号量表示资源的数量,对应的变量是一个整型(sem)变量。

另外,还有两个原子操作的系统调用函数来控制信号量的,分别是:

  • P 操作:将 sem 减 1,相减后,如果 sem < 0,则进程/线程进入阻塞等待,否则继续,表明 P 操作可能会阻塞;
  • V 操作:将 sem 加 1,相加后,如果 sem <= 0,唤醒一个等待中的进程/线程,表明 V 操作不会阻塞;

P 操作是用在进入临界区之前,V 操作是用在离开临界区之后,这两个操作是必须成对出现的。

类比如下,2 个资源的信号量,相当于 2 条火车轨道,PV 操作如下图过程:

信号量与火车轨道

操作系统是如何实现 PV 操作的呢?

信号量数据结构与 PV 操作的算法描述如下图:

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
27
28
29
30
31
32
33
34
35
36
37
38
// 信号量数据结构
typedef struct sem_t
{
int sem; // 资源个数
queue_t *q; // 等待队列
}sem_t;

// 初始化信号量
void init(sem_t* s, int sem)
{
s->sem = sem;
queue_init(s->q);
}

// P 操作
void P(sem_t* s)
{
s->sem--;
if(s->sem < 0)
{
1. 保留调用线程 CPU 现场;
2. 将该线程的 TCB 插入到 s 的等待队列;
3. 设置该线程为等待状态;
4. 执行调度程序;
}
}

// V 操作
void V(sem_t* s)
{
s->sem++;
if(s->sem <= 0)
{
1. 移出 s 等待队列首元素;
2. 将该线程的 TCB 插入就绪队列;
3. 设置该线程为就绪状态
}
}

PV 操作的函数是由操作系统管理和实现的,所以操作系统已经使得执行 PV 函数时是原子性的。

PV 操作如何使用的呢?

信号量不仅可以实现临界区的互斥访问控制,还可以实现线程间的事件同步。

我们先来说说如何使用信号量实现临界区的互斥访问

为每类共享资源设置一个信号量 s,其初值为 1,表示该临界资源未被占用。

只要把进入临界区的操作置于 P(s)V(s) 之间,即可实现进程/线程互斥:

此时,任何想进入临界区的线程,必先在互斥信号量上执行 P 操作,在完成对临界资源的访问后再执行 V 操作。由于互斥信号量的初始值为 1,故在第一个线程执行 P 操作后 s 值变为 0,表示临界资源为空闲,可分配给该线程,使之进入临界区。

若此时又有第二个线程想进入临界区,也应先执行 P 操作,结果使 s 变为负值,这就意味着临界资源已被占用,因此,第二个线程被阻塞。

并且,直到第一个线程执行 V 操作,释放临界资源,恢复 s 值为 0 后,才唤醒第二个线程,使之进入临界区,待它完成临界资源的访问后,又执行 V 操作,使 s 恢复到初始值 1。

对于两个并发线程,互斥信号量的值仅取 1、0 和 -1 三个值,分别表示:

  • 如果互斥信号量为 1,表示没有线程进入临界区;
  • 如果互斥信号量为 0,表示有一个线程进入临界区;
  • 如果互斥信号量为 -1,表示一个线程进入临界区,另一个线程等待进入。

通过互斥信号量的方式,就能保证临界区任何时刻只有一个线程在执行,就达到了互斥的效果。

再来,我们说说如何使用信号量实现事件同步

同步的方式是设置一个信号量,其初值为 0

我们把前面的「吃饭-做饭」同步的例子,用代码的方式实现一下:

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
semaphore s1 = 0;   // 表示需要吃饭
semaphore s2 = 0; // 表示饭还没做完

// 儿子线程函数
void son()
{
while(TRUE)
{
肚子饿;
V(s1); // 叫妈妈做饭
P(s2); // 等待妈妈做完饭
吃饭;
}
}

// 妈妈线程函数
void mom()
{
while(TRUE)
{
P(s1); // 询问需不需要做饭
做饭;
V(s2); // 做完饭,通知儿子吃饭
}
}

妈妈一开始询问儿子要不要做饭时,执行的是 P(s1) ,相当于询问儿子需不需要吃饭,由于 s1 初始值为 0,此时 s1 变成 -1,表明儿子不需要吃饭,所以妈妈线程就进入等待状态。

当儿子肚子饿时,执行了 V(s1),使得 s1 信号量从 -1 变成 0,表明此时儿子需要吃饭了,于是就唤醒了阻塞中的妈妈线程,妈妈线程就开始做饭。

接着,儿子线程执行了 P(s2),相当于询问妈妈饭做完了吗,由于 s2 初始值是 0,则此时 s2 变成 -1,说明妈妈还没做完饭,儿子线程就是等待状态。

最后,妈妈终于做完饭了,于是执行 V(s2)s2 信号量从 -1 变回了 0,于是就唤醒等待中的儿子线程,唤醒后,儿子线程就可以进行吃饭了。

2.2、生产者-消费者问题

生产者-消费者问题描述:

  • 生产者在生成数据后,放在一个缓冲区中;
  • 消费者从缓冲区取出数据处理;
  • 任何时刻,只能有一个生产者或消费者可以访问缓冲区;

我们对问题分析可以得出:

  • 任何时刻只能有一个线程操作缓冲区,说明操作缓冲区是临界代码,需要互斥
  • 缓冲区空时,消费者必须等待生产者生成数据;缓冲区满时,生产者必须等待消费者取出数据。说明生产者和消费者需要同步

那么我们需要三个信号量,分别是:

  • 互斥信号量 mutex:用于互斥访问缓冲区,初始化值为 1;
  • 资源信号量 fullBuffers:用于消费者询问缓冲区是否有数据,有数据则读取数据,初始化值为 0(表明缓冲区一开始为空);
  • 资源信号量 emptyBuffers:用于生产者询问缓冲区是否有空位,有空位则生成数据,初始化值为 n (缓冲区大小);
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
27
28
29
30
#define N 100
semaphore mutex = 1;
semaphore emptyBuffers = N;
semaphore fullBuffers = 0;

// 生产者线程函数
void producer()
{
while(TRUE)
{
P(emptyBuffers); // 讲空槽的个数 - 1
P(mutex); // 进入临界区
将生成的数据放到缓冲区中;
V(mutex); // 离开临界区
V(fullBuffers); // 将满槽的个数 + 1
}
}

// 消费者线程函数
void consumer()
{
while(TRUE)
{
P(fullBuffers); // 将满槽的个数 - 1
P(mutex); // 进入临界区
从缓冲区里读取数据;
V(mutex); // 离开临界区
V(emptyBuffers); // 将空槽的个数 + 1
}
}

如果消费者线程一开始执行 P(fullBuffers),由于信号量 fullBuffers 初始值为 0,则此时 fullBuffers 的值从 0 变为 -1,说明缓冲区里没有数据,消费者只能等待。

接着,轮到生产者执行 P(emptyBuffers),表示减少 1 个空槽,如果当前没有其他生产者线程在临界区执行代码,那么该生产者线程就可以把数据放到缓冲区,放完后,执行 V(fullBuffers) ,信号量 fullBuffers 从 -1 变成 0,表明有「消费者」线程正在阻塞等待数据,于是阻塞等待的消费者线程会被唤醒。

消费者线程被唤醒后,如果此时没有其他消费者线程在读数据,那么就可以直接进入临界区,从缓冲区读取数据。最后,离开临界区后,把空槽的个数 + 1。

三、经典同步问题

3.1、哲学家就餐问题

哲学家就餐的问题

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

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

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

方案一

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#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 ]) 这条语句阻塞了,很明显这发生了死锁的现象

方案二

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#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 把叉子,按道理是能可以有两个哲学家同时进餐的,所以从效率角度上,这不是最好的解决方案。

方案三

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

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

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

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 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 个哲学家的左邻右舍,则由宏 LEFTRIGHT 定义:

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

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

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
#define N 5
#define LEFT (i + N - 1) % N
#define RIGHT (i + 1) % N

#define THINKING 0
#define HUNGRY 1
#define EATING 2

int state[N];

semaphore s[5];
semaphore mutex;

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_forksput_forkstest 只是普通的函数,而非单独的进程/线程。

方案四也可解决问题

3.2、读者-写者问题

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

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

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

读者-写者的问题描述:

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

方案一

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

  • 信号量 wMutex:控制写操作的互斥信号量,初始值为 1 ;
  • 读者计数 rCount:正在进行读操作的读者个数,初始化为 0;
  • 信号量 rCountMutex:控制对 rCount 读者计数器的互斥修改,初始值为 1;
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
27
28
29
30
31
32
33
34
35
36
37
38
39
semaphore wMutex;           // 控制写操作的互斥信号量,初始值为 1
semaphore rCountMutex; // 控制对 rCount 的互斥修改,初始值为 1
int rCount; // 正在进行读操作的读者个数,初始化为 0

// 写者进程/线程执行的函数
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;
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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
semaphore rCountMutex;      // 控制对 rCount 的互斥修改,初始值为 1
semaphore rMutex; // 控制读者进入的互斥信号量,初始值为 1

semaphore wCountMutex; // 控制 wCount 互斥修改,初始值为 1
semaphore wDataMutex; // 控制写者写操作的互斥信号量,初始值为 1

int rCount = 0; // 正在进行读操作的读者个数,初始化为 0
int wCount = 0; // 正在进行写操作的写者个数,初始化为 0

// 写者进程/线程执行的函数
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++; // 读者计数 + 1
V(rCountMutex); // 离开临界区
V(rMutex);

read(); // 读数据

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

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

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

方案三

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

公平策略:

  • 优先级相同;
  • 写者、读者互斥访问;
  • 只能一个写者访问临界区;
  • 可以有多个读者同时访问临界资源;
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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
semaphore rCountMutex;      // 控制对 rCount 的互斥修改,初始值为 1
semaphore wDataMutex; // 控制写者写操作的互斥信号量,初始值为 1
semaphore flag; // 用于实现公平竞争,初始值为 1
int rCount = 0; // 正在进行读操作的读者个数,初始位为 0

// 写者进程/线程执行的函数
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); // 离开临界区
}
}

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

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

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

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

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

四、拓展

1、Java 多线程–三个线程分别打印 a、b、c,用多线程实现循环打印 15 次 abc

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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class PrintABC {
int count = 0; //打印次数
Lock lock = new ReentrantLock(); //可重写锁
Condition conditionA = this.lock.newCondition();
Condition conditionB = this.lock.newCondition();
Condition conditionC = this.lock.newCondition();

public class PrintA implements Runnable {
@Override
public void run() {
while (true)
if (count < 15) {
lock.lock();
System.out.print("A");
try {
conditionB.signal(); //线程b唤醒,因为a打印完应该打印b
conditionA.await(); //线程a进入等待队列
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}

}
}
}

public class PrintB implements Runnable {
@Override
public void run() {
while (true)
if (count < 15) {
lock.lock();
System.out.print("B");
try {
conditionC.signal();
conditionB.await();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}

}
}
}

public class PrintC implements Runnable {
@Override
public void run() {
while (true)
if (count < 15) {
lock.lock();
System.out.println("C" + count);
count++; //打印完c后,count++
try {
conditionA.signal();
conditionC.await();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}

}
}
}

public static void main(String[] args) {
PrintABC printABCD = new PrintABC();
new Thread(printABCD.new PrintA()).start();
new Thread(printABCD.new PrintB()).start();
new Thread(printABCD.new PrintC()).start();
}
}

原文链接:
多个线程为了同个资源打起架来了,该如何让他们安分?

评论

:D 一言句子获取中...

加载中,最新评论有1分钟缓存...