# 进程的定义、组成、组织方式、特征

整体框架

# 进程的定义

$\color{red}程序$ ：就是一个指令序列

早期的计算机（只支持 $\color{red}单道$ 程序）

引入 $\color{red}多道$ 程序之后：

为了方便操作操作系统管理，完成各个程序并发执行、

引入了 $\color{red}进程、进程实体$ 的概念

PCB、程序段、数据段三部分构成了

$\color{red}进程实体$ （ $\color{red}进程映像$ ）

$\color{red}程序段、数据段、\texttt{PCB}$ 三部分组成了 $\color{red}进程实体$ （ $\color{red}进程映像$ ）。

一般情况下，我们把进程实体就简称为进程，

例如，所谓创建进程，实质上是创建进程实体中的 PCB ;

而撤销进程，实质上是撤销进程实体中的 PCB。

注意： $\color{red}\texttt{PCB}是进程存在的唯一标志!$

从不同的角度，进程可以有不同的定义，比较传统典型的定义有:

进程是程序的一次 $\color{blue}执行过程$ 。
进程是一个程序及其数据在处理机上顺序执行时所 $\color{blue}发生的活动$ 。
进程是具有独立功能的程序在数据集合上运行的过程，

它是系统进行资源分配和调度的一个独立单位

上述都是强调进程的 “动态性”

引入进程实体的概念后，可把进程定义为:

$\color{red}进程$ 是进程实体的 $\color{blue}运行过程$ ，

是系统进行 $\color{red}资源分配$ 和 $\color{red}调度$ 的一个独立单位。

注：严格来说，进程实体和进程并不一样，

进程实体是 $\color{red}静态$ 的，

进程则是 $\color{red}动态$ 的。
不过，除非题目专门考察二者区别，否则可以认为进程实体就是进程。

因此我们也可以说 “进程由程序段、数据段、PCB 三部分组成”

# 进程的组成

$\color{red}进程实体$ （ $\color{red}进程映像$ ）由 $\color{red}程序段、数据段、\texttt{PCB}$ 三部分组成。

不同的书籍，分类方式可能不同

# 进程的组织

在一个系统中，通常有数十、数百乃至数千个 PCB 。为了能对他们加以有效的管理，应该用适当的方式把这些 PCB 组织起来。

注：进程的 $\color{red}组成$ 讨论的是一个 $\color{red}进程内部$ 由哪些部分构成的问题，

而进程的 $\color{red}组织$ 讨论的是 $\color{red}多个进程之间$ 的组织方式问题

# 链接方式

# 索引方式

# 进程的特征

进程和程序是两个截然不同的概念，相比于程序，进程拥有以下特征:

动态性是进程最基本的特征

进程是资源分配、接受调度的基本单位

异步性会导致并发程序执行结果的不确定性。

具体在 "进程同步"

# 整体框架

# 进程的状态与转换

整体框架

# 三种基本状态

进程是程序的一次执行。在这个执行过程中，有时进程正在被 CPU 处理，有时又需要等待 CPU 服务，

可见进程的状态是会有各种变化。

为了方便对各个进程的管理，操作系统需要将进程合理地划分为几种状态

① 运行态：

注意：单核处理机的环境下，每一个时刻最多只有一个进程处于运行态

（双核环境下可以同时有两个进程处于运行态）

② 就绪态：

进程已经拥有除处理机之外所有需要的组员，一旦获得处理机。即可立即进入运行态开始运行
即：万事具备，只欠 CPU

③ 阻塞态：

如：等待操作系统分配打印机、等待读磁盘操作的结果。

CPU 是计算机中最昂贵的部件，为了提高 CPU 的利用率，需要先将其他进程需要的资源分配到位，才能得到 CPU 的服务

例如：运行态中的某个进程向请求 I/O 设备，就需要进入阻塞态，等待这些资源的分配完成再进入就绪态，等待 CPU 的服务

# 另外两种状态

操作系统需要完成创建进程。

操作系统为该进程分配所需的内存空间等系统资源，

并为其创建、初始化 PCB（如：为进程分配 PID）

进程运行结束（或者由于 bug 导致进程无法继续执行下去，比如数组越界错误），

需要撤销进程。

操作系统需要完成撤销进程相关的工作。

完成将分配给进程的资源回收，撤销进程 PCB 等工作

# 进程状态的转换

# 整体框架

# 进程控制

整体框架

# 什么是进程控制？

进程控制的主要功能是对系统中的所有进程实施有效的管理，

它具有创建新进程、撤销已有进程、实现进程状态转换等功能。

简单来说：进程控制就是要实现进程状态转换

# 如何实现进程控制？

# 进程控制

① 创建进程：需要初始化 PCB、分配系统资源

创建态 -> 就绪态，需要修改 PCB 内容和相应队列

② 就绪态 -> 运行态需要恢复进程运行环境、修改 PCB 内容和相应队列

恢复进程运行环境因为进程可能在运行中就因为某个原因让出了 CPU，这一次又抢到了 CPU 时间片，需要从之前中断的地方开始执行

③ 运行态 -> 阻塞态需要保存进程运行环境、修改 PCB 内容和相应队列

保存进程运行环境因为进程因为某个原因让出了 CPU，在某段时间后可能还会再次抢到 CPU，需要从之前中断的地方开始执行

④ 阻塞态 -> 就绪态需要修改 PCB 内容和相应队列。

如果等待的是资源，则还需为进程分配系统资源

⑤ 运行态 -> 就绪态（进程切换）需要保存进程运行环境、修改 PCB 内容和相应队列

⑥ 运行态 -> 终止态需回收进程拥有的资源，撤销 CPU

# 原语

原语是一种特殊的程序，处于操作系统最顶层，是最接近硬件的部分

用 $\color{red}原语$ 实现进程控制。

原语的特点是执行期间 $\color{red}不允许中断$ ，

只能一气呵成。

这种不可被中断的操作即 $\color{red}原子操作$ 。

原语采用 “ $\color{red}关中断$ 指令” 和 “ $\color{red}开中断$ 指令” 实现

显然， $\color{red}关/开中断指令$ 的权限非常大，

必然是只允许在 $\color{red}核心态$ 下执行的 $\color{red}特权指令$

原语属于操作系统内核的一部分

# 进程控制相关的原语

学习技巧：进程控制会导致进程状态的转换。无论哪个原语，要做的无非三类事情:

更新 PCB 中的信息（如修改进程状态标志、将运行环境保存到 PCB 、从 PCB 恢复运行环境)

a. 所有的进程控制原语一定都会修改进程状态标志

b. 剥夺当前运行进程的 CPU 使用权必然需要保存其运行环境

c. 某进程开始运行前必然要恢复期运行环境
将 PCB 插入合适的队列
分配 / 回收资源

# 进程的创建原语

无 -> 创建态 -> 就绪态

# 进程的终止原语

就绪态 / 阻塞态 / 运行态 -> 终止态 -> 无

# 进程的阻塞和唤醒原语

阻塞原语与唤醒原语必须成对使用

阻塞原语：运行态 -> 阻塞态

唤醒原语：阻塞态 -> 运行态

因何事阻塞，就应何事唤醒

即：解铃还须系铃人

# 进程的切换原语

运行态 -> 阻塞态 / 就绪态

就绪态 -> 运行态

# 整体框架

各原语可以实现怎样的状态转换

各原语大概做了哪些事情

# 进程通信

整体框架

# 什么是进程通信？

顾名思义，进程通信就是指进程之间的信息交换。

进程是分配系统资源的单位（包括内存地址空间)，

因此 $\color{red}各进程$ 拥有的 $\color{red}内存地址空间相互独立$ 。

为了保证安全， $\color{red}一个进程不能直接访问另一个进程的地址空间$

但是进程之间的信息交换又是必须实现的。

为了保证进程间的安全通信，操作系统提供了一些方法。

# 共享存储

两个进程对共享空间的访问必须是互斥的（互斥访问通过操作系统提供的工具实现）。

操作系统只负责提供共享空间和同步互斥工具（如 P、V 操作）

$\color{red}基于数据结构$ 的共享：

比如共享空间里只能放一个长度为 10 的数组。
这种共享方式速度慢、限制多，
是一种 $\color{red}低级通信$ 方式

$\color{red}基于存储区$ 的共享：

在内存中画出一块共享存储区，数据的形式、存放位置都由进程控制，而不是操作系统。
相比之下，这种共享方式速度更快，
是一种 $\color{red}高级通信$ 方式。

# 管道通信

“管道” 是指用于连接读写进程的一个共享文件，

又名 pipe 文件。
其实就是在内存中开辟一个大小固定的缓冲区

管道只能采用 $\color{red}半双工通信$ ，某一时间段内只能实现单向的传输。

如果要实现双向同时通信，则需要设置两个管道。

首先进程 $1$ 会往管道中写数据，当管道中的数据写满了之后进程 $2$ 才可以开始往外读数据

只有这个数据全部被读出后，进程 $1$ 才可以继续往里面写数据

数据以字符流的形式写入管道，当管道写满时，写进程的 write() 系统调用将被阻塞，等待读进程将数据取走。

当读进程将数据全部取走后 **，管道变空 **，此时读进程的 read() 系统调用将被阻塞。

如果 $\color{red}没写满,就不允许读$ 。

如果 $\color{red}没读空,就不允许写$ 。

数据一旦被读出，就从管道中被抛弃，

这就意味着 $\color{red}读进程最多只能有一个$ ，

否则可能会有读错数据的情况。

# 消息传递

进程间的数据交换以 $\color{red}格式化的消息$ （ Message ）为单位。

进程通过操作系统提供的 “发送消息 / 接收消息” 两个 $\color{red}原语$ 进行数据交换。

例如： TCP 报文段的首部就是消息头，数据载荷就是消息体

# 直接通信方式

消息直接挂到接收进程的消息缓存队列上

# 间接通信方式

消息要先发送到中间实体（信箱）中，因此也称 “信箱通信方式” 。

Eg：计网中的电子邮件系统

# 整体框架

# 线程概念和多线程模型

整体框架

# 什么是线程，为什么要引入线程？

还没有引入进程之前，各个程序只能串行执行

进程是程序的一次执行过程，但这些功能显然不可能是由一个程序顺序处理就能实现的

有的进程可能需要 “同时” 做很多事，而传统的进程只能串行地执行一系列程序。

为此，引入了 “线程”，来增加并发度。

引入线程后，线程成为了程序执行流的最小单位

可以把线程理解为 “轻量级进程” 。

$\color{red}线程$ 是一个 $\color{red}基本的\texttt{CPU}执行单元$ ，

也是 $\color{red}程序执行流的最小单位$ 。

引入线程之后，不仅是进程之间可以并发，

进程内的 $\color{red}各线程之间$ 也可以 $\color{red}并发$ ，

从而进一步 $\color{red}提升了系统的并发度$ ，

使得一个进程内也可以并发处理各种任务（如 QQ 视频、文字聊天、传文件)

引入线程后， $\color{red}进程$ 只作为 $\color{red}除\texttt{CPU}之外的系统资源的分配单元$

（如打印机、内存地址空间等都是分配给进程的）。

即：进程只作为分配资源的基本单位，而将线程作为调度的基本单位

# 引入线程带来的变化

# 线程的属性

例如：从 QQ 聊天窗口跳到 b 站视频就是不同进程之间线程的切换

# 线程的实现方式

# 用户级线程 (User-Level Thread,ULT)

用户级线程由应用程序通过线程库实现。

所有的 $\color{red}线程管理工作$ 都由 $\color{red}应用程序负责$ （包括线程切换)

用户级线程中，

$\color{red}线程切换$ 可以在 $\color{red}用户态下即可完成$ ，

无需操作系统干预。

在用户看来，是有多个线程。但是在操作系统内核看来，并意识不到线程的存在。

（用户级线程对用户不透明，对操作系统透明)

可以这样理解，“ $\color{red}用户级线程$ ” 就是 “ $\color{red}从用户视角看能看到的线程$ ”

# 内核级线程 (Kernel-Level Thread, KLT)

又称 "内核支持的线程"

内核级 $\color{red}线程的管理工作$ 由 $\color{red}操作系统内核$ 完成。

线程调度、切换等工作都由内核负责，

因此 $\color{red}内核级线程的切换$ 必然需要在 $\color{red}核心态$ 下才能完成。

可以这样理解，“ $\color{red}内核级线程$ ” 就是 “ $\color{red}从操作系统内核视角看能看到的线程$ ”

# 二者组合

在同时支持用户级线程和内核级线程的系统中，可采用二者组合的方式:

将 n 个用户级线程映射到 m 个内核级线程上（ n >= m ）

$\color{red}重点重点重点$ ：

操作系统只 “看得见” 内核级线程，

因此只有 $\color{red}内核级线程才是处理机分配的单位$ 。

例如：

上述这个模型中，该进程由两个内核级线程，三个用户级线程，在用户看来，这个进程中有三个线程。

但即使该进程在一个 4 核处理机的计算机上运行，也最多只能被分配到两个核，最多只能有两个用户线程并行执行。

# 多线程模型

在同时支持用户级线程和内核级线程的系统中，由几个用户级线程映射到几个内核级线程的问题引出了 “多线程模型” 问题。

# 多对一模型

$\color{red}多对一$ 模型：

多个用户及线程映射到一个内核级线程。每个用户进程只对应一个内核级线程。

优点：

用户级线程的切换在用户空间即可完成，不需要切换到核心态，
线程管理的系统开销小，效率高

缺点：

当一个用户级线程被阻塞后，整个进程都会被阻塞，并发度不高。
多个线程不可在多核处理机上并行运行

因为线程是处理机调度的基本单位，而不是资源分配的基本单位。多个用户级映射在一个内核级线程，内核认为仅有一个线程，当一个用户级线程被阻塞，内核认为所映射的内核级线程也被阻塞，故所有映射到该内核级的用户级线程都被阻塞

# 一对一模型

$\color{red}一对一$ 模型：

一个用户及线程映射到一个内核级线程。
每个用户进程有与用户级线程同数量的内核级线程。

优点：

当一个线程被阻塞后，别的线程还可以继续执行，并发能力强
多线程可在多核处理机上并行执行。

缺点：

一个用户进程会占用多个内核级线程，线程切换由操作系统内核完成，需要切换到核心态，
因此线程管理的成本高，开销大。

# 多对多模型

多对多模型： n 用户及线程映射到 m 个内核级线程（ n >= m ）。

每个用户进程对应 m 个内核级线程。

克服了多对一模型并发度不高的缺点，

又克服了一对一模型中一个用户进程占用太多内核级线程，开销太大的缺点。

# 整体框架

# 处理机调度的概念、层次

# 调度的基本概念

当有一堆任务要处理，但由于资源有限，这些事情没法同时处理。、

这就需要确定某种规则来决定处理这些任务的顺序，这就是 “调度” 研究的问题。

在多道程序系统中，进程的数量往往是多于处理机的个数的，这样不可能同时并行地处理各个进程。

$\color{red}处理机调度$ ，

就是从就绪队列中 $\color{red}按照一定的算法选择一个进程$ 并将 $\color{red}处理机分配给它$ 运行，

以实现进程的并发执行。

# 调度的三个层次

# 高级调度

由于内存空间有限，有时无法将用户提交的作业全部放入内存，因此就需要确定某种规则来决定将作业调入内存的顺序。

$\color{red}高级调度（作业调度)$ 。按一定的原则从外存上处于后备队列的作业中挑选一个 (或多个）作业，给他们分配内存等必要资源，

并 $\color{red}建立相应的进程(建立\texttt{PCB})$ ，
以使它（们） $\color{red}获得竞争处理机的权利$ 。

高级调度是辅存（外存）与内存之间的调度。每个作业只调入一次，调出一次。

$\color{red}作业调入时会建立相应的\texttt{PCB}，作业调出时才撤销\texttt{PCB}$ 。

高级调度主要是指 调入 的问题，因为只有调入的时机需要操作系统来确定，

但调出的时机必然是作业运行结束才调出。

# 中级调度

引入了虚拟存储技术之后，可将暂时不能运行的进程调至外存等待。

等它重新具备了运行条件且内存又稍有空闲时，再重新调入内存。

这么做的目的：

是为了 $\color{red}提高内存利用率$ 和 $\color{red}系统吞吐量$ 。

暂时调到外存等待的进程状态为 $\color{red}挂起状态$ 。

值得注意的是， PCB 并不会一起调到外存，而是 $\color{red}会常驻内存$ 。

PCB 中会记录进程数据在外存中的存放位置，进程状态等信息，操作系统通过内存中的 PCB 来保持对各个进程的监控、管理。
被挂起的进程 PCB 会被放到的 $\color{red}挂起队列$ 中。

$\color{red}中级调度（内存调度）$ ，

就是要决定将哪个处于挂起状态的进程重新调入内存。

一个进程可能会被多次调出、调入内存，

因此 $\color{red}中级调度$ 发生的 $\color{red}频率$ 要比高级调度 $\color{red}更高$ 。

# 进程的挂起态与七状态模型

暂时调到外存等待的进程状态为 $\color{red}挂起状态$ （挂起态，suspend）

挂起态又可以进一步细分为 $\color{red}就绪挂起$ 、 $\color{red}阻塞挂起$ 两种状态

五状态模型→七状态模型

就是内存不够用了，将进程挂到就绪挂起或者挂到阻塞挂起

等到内存空间空闲，这个进程需要继续执行，这个进程就会被激活，相应的数据就会被移到内存中

一个处于创建态的进程，当它创建 PCB 之后，有可能出现内存空间不够的情况，就会先进入到就绪挂起状态

当阻塞挂起中的进程等待某一事件发生时候，这个进程就会进入到就绪挂起状态

注意 “挂起” 和 “阻塞” 的区别，两种状态都是暂时不能获得 CPU 的服务，

但挂起态是将进程映像（程序段、数据段）调到外存去了，
而阻塞态下进程映像还在内存中。

有的操作系统会把就绪挂起、阻塞挂起分为两个挂起队列，甚至会根据阻塞原因不同再把阻塞挂起进程进一步细分为多个队列。

# 低级调度

$\color{red}低级调度（进程调度）$ ，

其主要任务是按照某种方法和策略从就绪队列中选取一个进程，将处理机分配给它。

进程调度是操作系统中 $\color{red}最基本的一种调度$ ，

在一般的操作系统中都必须配置进程调度。

进程调度的 $\color{red}频率很高$ ，一般几十毫秒一次。

# 三层调度的联系、对比

	要做什么	调度发生在...	发生频率	队进程状态的影响
高级调度（作业调度）	按照某种规则，从后备队列中选择合适的作业（类似于程序段和数据段）将其调入内存，并为其创建进程	外存 -> 内存（面向作业）	最低	无 -> 创建态 -> 就绪态
中级调度（内存调度）	按照某种规则，从挂起队列中选择合适的进程将其数据调回内存	外存 -> 内存（面向进程）	中等	就绪挂起 -> 就绪态（阻塞挂起 -> 阻塞态）
低级调度（进程调度）	按照某种规则，从就绪队列中选择一个进程为其分配处理机	内存 -> CPU	最高	就绪态 -> 运行态

# 整体框架

# 进程调度的时机、切换与过程、方式

# 进程调度的时机

$\color{red}进程调度$ （低级调度），就是按照某种算法从就绪队列中选择一个进程为其分配处理机

进程在 $\color{red}操作系统内核程序临界区$ 中 $\color{red}不能$ 进行调度与切换 √

（2012 年联考真题）进程处于 $\color{red}临界区$ 时 $\color{red}不能$ 进行处理机调度 ❌

临界资源 ：一个时间段内只允许一个进程使用的资源。

各进程需要 $\color{red}互斥地$ 访问临界资源。

临界区 ：访问临界资源的那段代码。

$\color{red}内核程序临界区$ 一般是用来访问 $\color{red}某种内核数据结构$ 的，

比如进程的就绪队列（由各就绪进程的 PCB 组成)

若还没有退出内核程序临界区（还没解锁）就进行进程调度（访问就绪队列），

但是进程调度的相关程序也需要访问就绪队列，

但此时就绪队列被锁住了，因此又无法顺利进行进程调度

若此时进程访问的是普通的临界资源，例如：打印机

在打印机打印完成之前，进程一直处于临界区内，临界资源不会解锁。

但打印机又是慢速设备，此时如果一直不允许进程调度的话就会导致 CPU 一直空闲

普通 I 临界区访问的临界资源不会直接影响操作系统内核的管理工作。

因此在访问普通临界区时可以进行调度与切换。

# 进程调度的方式

有的系统中，只允许进程主动放弃处理机

有的系统中，进程可以主动放弃处理机，

当有更紧急的任务需要处理时，也会强行剥夺处理机（被动放弃)

# 非剥夺方式（非抢占方式）

$\color{red}非剥夺调度方式$ ，又称 $\color{red}非抢占方式$ 。

即，只允许进程主动放弃处理机。

在运行过程中即便有更紧迫的任务到达，当前进程依然会继续使用处理机，

直到该进程终止或主动要求进入阻塞态。

实现简单，系统开销小但是无法及时处理紧急任务，适合于早期的批处理系统

# 剥夺调度方式（抢占方式）

$\color{red}剥夺调度方式$ ，又称 $\color{red}抢占方式$ 。

当一个进程正在处理机上执行时，如果有一个更重要或更紧迫的进程需要使用处理机，则立即暂停正在执行的进程，将处理机分配给更重要紧迫的那个进程。

可以优先处理更紧急的进程，也可实现让各进程按时间片轮流执行的功能（通过时钟中断）。

适合于分时操作系统、实时操作系统

# 进程的切换与过程

# "狭义的进程调度" 与 "进程切换" 的区别

“狭义的进程调度” 与 “进程切换” 的区别:

$\color{red}狭义的进程调度$ 指的是从就绪队列中 $\color{red}选中一个要运行的进程$ 。

这个进程可以是刚刚被暂停执行的进程，也可能是另一个进程，后一种情况就需要进程切换

$\color{red}进程切换$ 是指一个进程让出处理机，由另一个进程占用处理机的过程。

$\color{red}广义的进程调度$ 包含了选择一个进程和进程切换两个步骤。

进程切换的过程主要完成了:

对原来运行进程各种数据的保存（信息一般保存在 PCB 中）
对新的进程各种数据的恢复

（如：程序计数器、程序状态字、各种数据寄存器等处理机现场信息，这些信息一般保存在进程控制块）

注意： $\color{red}进程切换是有代价的$

因此如果过于频繁的进行进程调度、切换，必然会使整个系统的效率降低，使系统大部分时间都花在了进程切换上，而真正用于执行进程的时间减少。

# 整体框架

# 调度算法的评价指标

# CPU 利用率

由于早期的 CPU 造价极其昂贵，

因此人们会 $\color{red}希望让 \texttt{CPU} 尽可能多地工作$

$\color{red}\texttt{CPU}利用率$ ：指 CPU “忙碌” 的时间占总时间的比例。

$\color{red}{利用率}$ $=\large\frac{忙碌的时间}{总时间} \,$

有的题目还会要求计算某种设备的利用率

Eg ：某计算机只支持单道程序，某个作业刚开始需要在 CPU 上运行 $5$ 秒，再用打印机打印输出 $5$ 秒，之后再执行 $5$ 秒，才能结束。在此过程中， CPU 利用率、打印机利用率分别是多少？

CPU 利用率 = $\frac{5+5}{5+5+5}=66.66\%$
打印机利用率 = $\frac{5}{5+5+5}=33.33\%$

通常会考察多道程序并发执行的情况，可以用 “甘特图” 来辅助计算

# 系统吞吐量

对于计算机来说，希望能用尽可能少的时间处理完尽可能多的作业

$\color{red}系统吞吐量$ ：

单位时间内完成作业的数量

$\textcolor{red}{系统吞吐量}=\large\frac{总共完成了多少道作业}{总共花了多少时间}\,$

Eg ：某计算机系统处理完 $10$ 道作业，共花费 $100$ 秒，则系统吞吐量为？

$\frac{10}{100}= 0.1道/秒$

# 周转时间 (4 个指标)

对于计算机的用户来说，他很关心自己的作业从提交到完成花了多少时间。

$\color{red}周转时间$ ，

是指从作业被提交给系统开始，到作业完成为止的这段时间间隔。

它包括四个部分:

作业在外存后备队列上等待作业调度（高级调度）的时间、
进程在就绪队列上等待进程调度（低级调度）的时间（就绪态）、
进程在 CPU 上执行的时间（运行态）、
进程等待 I/0 操作完成的时间（阻塞态）。

后三项在一个作业的整个处理过程中，可能发生多次。

（作业） $\color{red}周转时间$ $=作业完成时间-作业提交时间$

上述对于用户来说，更关心自己的单个作业的周转时间

$\color{red}平均周转时间$ $=\large\frac {各作业周转时间之和}{作业数}\,$

对于操作系统来说，更关心系统的整体表现，因此更关心所有作业周转时间的平均值

思考：有的作业运行时间短，有的作业运行时间长，因此在周转时间相同的情况下，运行时间不同的作业，给用户的感觉肯定是不一样的

例如：排队上厕所，另外个人只需等待一分钟，上十分钟，而你却等待十分钟，上一分钟

$\color{red}带权周转时间$ $=\large\frac {作业周转时间}{作业实际运行的时间}\, = \frac {作业完成时间-作业提交时间}{作业实际运行的时间}\,$

对于周转时间相同的两个作业，实际运行时间长的作业在相同时间内被服务的时间更多，带权周转时间更小，用户满意度更高。
对于实际运行时间相同的两个作业，周转时间短的带权周转时间更小，用户满意度更高。

带权周转时间必然 $≥1$

带权周转时间与周转时间都是越小越好

$\color{red}平均带权周转时间$ $=\large\frac {各作业带权周转时间之和}{作业数}\,$

# 等待时间

计算机的用户希望自己的作业尽可能少的等待处理机

$\color{red}等待时间$ ，

指进程 / 作业处于等待处理机状态时间之和，等待时间越长，用户满意度越低。

对于 $\color{red}进程$ 来说，

等待时间就是指进程建立后 $\color{red}等待被服务的时间之和$
在等待 I/O 完成的期间其实进程也是在被服务的，所以不计入等待时间。

对于 $\color{red}作业$ 来说，

不仅要考虑 $\color{red}建立进程后的等待时间$
$\color{red}还要加上作业在外存后备队列中等待的时间$ 。

一个作业总共需要被 CPU 服务多久，被 l/O 设备服务多久一般是确定不变的，

因此调度算法其实只会影响作业 / 进程的等待时间。

当然，与前面指标类似，也有 “ $\color{red}平均等待时间$ ” 来评价整体性能。

# 响应时间

对于计算机用户来说，会希望自己的提交的请求（比如通过键盘输入了一个调试命令）尽早地开始被系统服务、回应。

$\color{red}响应时间$ ：

指从用户 $\color{red}提交请求$ 到 $\color{red}首次产生响应$ 所用的时间。

# 整体框架

# FCFS、SJF、HRRN 调度算法

Tips ：各种调度算法的学习思路

算法思想
算法规则
这种调度算法是用于作业调度还是进程调度？
抢占式？非抢占式？
优点和缺点
是否会导致 $\color{red}饥饿$

某进程 / 作业长期得不到服务

算法	可抢占？	优点	缺点	考虑到等待时间 & 运行时间	会导致饥饿？
FCFS	非抢占式	公平；实现简单；利于长作业	不利于短作业	等待时间 ✔；<br /> 运行时间 ❌	不会
SJF/SPF	默认为非抢占式；也有 SJF 抢占式版本最短剩余时间优先算法（SRTN）	"最短的" 平均等待 / 周转时间；利于短作业	不利于长作业，可能会导致饥饿；难以做到正真的短作业优先	等待时间 ❌；<br /> 运行时间 ✔	会（导致长作业饥饿）
HRRN	非抢占式	上述两种算法的均衡考虑，综合考虑的等待时间和运行时间		等待时间 ✔；<br /> 运行时间 ✔	不会

# FCFS

例题：各进程到达就绪队列的时间、需要的运行时间如下表所示。使用先来先服务调度算法，计算各进程的等待时间、平均等待时间、周转时间、平均周转时间、带权周转时间、平均带权周转时间。

# SJF

# 非抢占式的 SJF

例题：各进程到达就绪队列的时间、需要的运行时间如下表所示。使用非抢占式的短作业优先（严格来说，用于进程调度应该称为短进程优先调度 SPF ）调度算法，计算各进程的等待时间、平均等待时间、周转时间、平均周转时间、带权周转时间、平均带权周转时间。

# 抢占式的 SJF（最短剩余时间优先算法 SRTN）

# 注意

1 . 如果题目中未特别说明，所提到的 “短作业 / 进程优先算法” 默认是非抢占式的

2 . 很多书上都会说 “ SJF 调度算法的平均等待时间、平均周转时间最少”

严格来说，这个表述是错误的，不严谨的。之前的例子表明，最短剩余时间优先算法得到的平均等代时间、平均周转时间还要更少

应该加上一个条件 “在 所有进程同时可运行 时，采用 SJF 调度算法的平均等待时间、平均周转时间最少”;

或者说 “在 所有进程都几乎同时到达时 ，采用 SJF 调度算法的平均等待时间、平均周转时间最少” ;

如果不加上述前提条件，则应该说 “ 抢占式的 短作业 / 进程优先调度算法（ 最短剩余时间优先, SRTN 算法）的平均等待时间、平均周转时间最少”

3．虽然严格来说， SJF 的平均等待时间、平均周转时间并不一定最少，

但相比于其他算法（如 FCFS ) ， SJF 依然可以获得较少的平均等待时间、平均周转时间

4 ．如果选择题中遇到 “ SIF 算法的平均等待时间、平均周转时间最少” 的选项，那最好判断其他选项是不是有很明显的错误，如果没有更合适的选项，那也应该选择该选项

# 二者思考

FCFS 算法是在每次调度的时候选择一个等待时间最长的作业（进程）为其服务。

但是没有考虑到作业的运行时间，因此导致了对短作业不友好的问题

SJF 算法是选择一个执行时间最短的作业为其服务。

但是又完全不考虑各个作业的等待时间，因此导致了对长作业不友好的问题，甚至还会造成饥饿问题

能不能设计一个算法，即考虑到各个作业的等待时间，也能兼顾运行时间呢？

高响应比优先算法

# HRRN

$\color{red}响应比$ $=\large\frac{等待时间+要求服务时间}{要求服务时间}\,$

例题：各进程到达就绪队列的时间、需要的运行时间如下表所示。使用高响应比优先调度算法，计算各进程的等待时间、平均等待时间、周转时间、平均周转时间、带权周转时间、平均带权周转时间。

# 总结

算法	可抢占？	优点	缺点	考虑到等待时间 & 运行时间	会导致饥饿？
FCFS	非抢占式	公平；实现简单；利于长作业	不利于短作业	等待时间 ✔；<br /> 运行时间 ❌	不会
SJF/SPF	默认为非抢占式；也有 SJF 抢占式版本最短剩余时间优先算法（SRTN）	"最短的" 平均等待 / 周转时间；利于短作业	不利于长作业，可能会导致饥饿；难以做到正真的短作业优先	等待时间 ❌；<br /> 运行时间 ✔	会（导致长作业饥饿）
HRRN	非抢占式	上述两种算法的均衡考虑，综合考虑的等待时间和运行时间		等待时间 ✔；<br /> 运行时间 ✔	不会

注：这几种算法主要关心对用户的公平性、平均周转时间、平均等待时间等评价系统整体性能的指标，但是不关心 “响应时间”，也并不区分任务的紧急程度，因此对于用户来说，交互性很糟糕。

因此这三种算法一般适合用于 $\color{red}早期的批处理系统$ ，

当然， FCFS 算法也常结合其他的算法使用，在现在也扮演着很重要的角色。

而适合用于交互式系统的调度算法将在下个小节介绍...

# 时间片轮转、优先级调度算法、多级反馈队列调度算法

Tips ：各种调度算法的学习思路

算法思想
算法规则
这种调度算法是用于作业调度还是进程调度？
抢占式？非抢占式？
优点和缺点
是否会导致 $\color{red}饥饿$

某进程 / 作业长期得不到服务

# 时间片轮转（RR, Round-Robin）

# 例子

常用于分时操作系统，更注重 "响应时间"，因而此处不计算周转时间

例题：各进程到达就绪队列的时间、需要的运行时间如下表所示。使用时间片轮转调度算法，计算各进程的等待时间、平均等待时间、周转时间、平均周转时间、带权周转时间、平均带权周转时间。

时间片为 5 的情况

若按照先来先服务调度算法

# 时间片太大 / 小的影响

如果时间片太大，使得每个进程都可以在一个时间片内就完成，

则时间片轮转调度算法 退化 为先来先服务调度算法，

并且 $\color{red}会增大进程响应时间$ 。

比如：系统中有 10 个进程在并发执行，如果时间片为 1 秒，则一个进程被响应可能需要等 9 秒...
- 也就是说，如果用户在自己进程的时间片外通过键盘发出调试命令，
  
  可能需要等待 9 秒才能被系统响应
- 其实就是时间片太长了，导致后面的进程等待时间也会太长
因此时间片不能太大。

另一方面，进程调度、切换是有时间代价的（保存、恢复运行环境)，

因此如果时间片太小，会导致 进程切换过于频繁 （不断地频繁切换用户态与内核态），系统会花大量的时间来处理进程切换，

从而导致实际用于进程执行的时间比例减少。
可见时间片也不能太小。

一般来说，设计时间片时要让切换进程的开销占比不超过 $1\%$

# 优先级调度算法

例题：各进程到达就绪队列的时间、需要的运行时间、进程优先数如下表所示。使用非抢占式的优先级调度算法，分析进程运行情况。（注：优先数越大，优先级越高)

# 非抢占式的

# 抢占式的

# 静态 / 动态优先级

就绪队列未必只有一个，可以按照不同优先级来组织。

另外，也可以把优先级高的进程排在更靠近队头的位置

根据优先级是否可以动态改变，可将优先级分为静态优先级和动态优先级两种。

$\color{red}静态优先级$ ：

创建进程时确定，之后一直不变。

$\color{red}动态优先级$ ：

创建进程时有一个初始值，之后会根据情况动态地调整优先级。

如何合理地设置各类进程地优先级？

通常：

系统进程优先级高于用户进程
前台进程优先级高于后台进程
操作系统更偏好 I/O 型进程（或称 I/O 繁忙型进程）

注：与 I/O 型进程相对的是计算型进程（或称 CPU 繁忙型进程）

I/O 设备和 CPU 可以并行工作。如果优先让 I/O 繁忙型进程优先运行的话，

则越有可能让 I/O 设备尽早地投入工作，则资源利用率、系统吞吐量都会得到提升
I/O 花费的时间一般比较长，尽早处理完 I/O

若采用 CPU 繁忙型进程的话，

CPU 优先于 I/O ，则 CPU 运行完之后，此时后备队列中没有队列了，需要等待 I/O 设备输入

如果采用的是动态优先级，什么时候应该调整？

可以从追求公平、提升资源利用率等角度考虑

如果某进程在就绪队列中等待了很长时间，则可以适当提升其优先级
如果某进程占用处理机运行了很长时间，则可适当降低其优先级
如果发现一个进程频繁地进行 I/O 操作，则可适当提升其优先级

HRRN 高响应比优先调度算法，可以认为是动态地优先级调度算法

# 多级反馈队列调度算法

FCFS 算法的优点是公平

SJF 算法的优点是能尽快处理完短作业，平均等待 / 周转时间等参数很优秀

时间片轮转调度算法可以让各个进程得到及时的响应

优先级调度算法可以灵活地调整各种进程被服务的机会

能否对其他算法做个折中权衡？得到一个综合表现优秀平衡的算法呢？

多级反馈队列调度算法

由于进程源源不断地到来会使得高优先级的队列始终非空，而低优先级的队列需要等待高优先级的队列为空时才能使用，导致低优先级队列中的进程饥饿

# 总结

算法	可抢占？	优点	缺点	会导致饥饿？	补充
时间片轮转	抢占式	公平；适用于分时系统	频繁切换有开销；不区分优先级	不会	时间片太大或太小导致的影响
优先级调度	有的抢占式的，有的非抢占式的	区分优先级；适用于实时系统	可能导致饥饿	会	动态 / 静态优先级。各类型型进程如何设置优先级？如何调整优先级
多级反馈队列	抢占式	平衡优秀；6（9 翻了）	一般不说它优缺点；不过可能导致饥饿	会

注：比起早期的批处理操作系统来说，由于计算机造价大幅降低，因此之后出现的交互式操作系统（包括分时操作系统、实时操作系统等）更注重系统的响应时间、公平性、平衡性等指标。

而这几种算法恰好也能较好地满足交互式系统的需求。

因此这三种算法适合用于 $\color{red}交互式系统$

比如 UNIX 使用的就是多级反馈队列调度算法

# 进程同步与互斥

# 什么是进程同步

# 回顾异步性

$\color{red}异步$ 是指，

在多道程序环境下，允许多个程序并发执行，但由于资源有限，进程的执行不是一贯到底的，而是走走停停，以不可预知的速度向前推进，这就是进程的异步性。

由于系统的有限资源导致的

例如：老渣要和两个女孩并发约会

一号的指令 1 ：老渣陪我吃饭

一号的指令 2 ：老渣把心给我

二号的指令 1 ：老渣把心给我

二号的指令 2 ：老渣陪我吃饭

与一号、二号的约会相当于对两个进程的处理，每个进程都有各自需要执行的指令。

老渣的心相当于有限的系统资源。

若女一号只想做老渣的初恋

女二号只想交一个有恋爱经验的渣男

那么，老渣在并发执行这两个约会进程的时候，就必须保证 “一号的指令 2” 一定要在二号的指令 1” 之前执行。

操作系统需要提供 "进程同步机制" 来实现上述需求

# 回顾进程通信 -- 管道通信

读进程和写进程并发地运行，由于并发必然导致异步性，

因此 “写数据” 和 “读数据” 两个操作执行的先后顺序是不确定的。

而实际应用中，又必须按照 “写数据→读数据” 的顺序来执行的。

如何解决这种 $\color{red}异步$ 问题，

就是 “进程 $\color{red}同步$ ” 所讨论的内容。

# 同步（直接制约关系）

$\color{red}同步$ 亦称 $\color{red}直接制约关系$ ，

它是指为完成某种任务而建立的两个或多个进程，这些进程因为需要在某些位置上协调它们的工作次序而产生的制约关系。
进程间的直接制约关系就是源于它们之间的相互合作。

# 进程互斥（间接制约关系）

进程的 “并发” 需要 “共享” 的支持。各个并发执行的进程不可避免的需要共享一些系统资源

比如内存，又比如打印机、摄像头这样的 I/O 设备

$\color{red}临界资源$

$\textcolor{red}{一个时间段内只允许一个进程使用}$ 的资源

许多物理设备（比如摄像头、打印机）都属于临界资源。

此外还有许多变量、数据、内存缓冲区等都属于临界资源。

对临界资源的访问，必须互斥地进行。

$\color{red}互斥$ ，亦称 $\color{red}间接制约关系$ 。

进程互斥指当一个进程访问某临界资源时，另一个想要访问该临界资源的进程必须等待。当前访问临界资源的进程访问结束，释放该资源之后另一个进程才能去访问临界资源。

# 对临界资源的互斥访问（四个部分）

注意：

$\textcolor{red}{临界区}$ 是进程中 $\textcolor{red}{访问临界资源}$ 的代码段。
$\textcolor{red}{进入区}$ 和 $\textcolor{red}{退出区}$ 是 $\textcolor{red}{负责实现互斥}$ 的代码段。
临界区也可以称为 "临界段"

如果一个进程暂时不能进入临界区，

那么该进程是否应该一直占着处理机？
该进程有没有可能一直进不了临界区？

# 遵循的原则

空闲让进。临界区空闲时，可以允许一个请求进入临界区的进程立即进入临界区；
忙则等待。当已有进程进入临界区时，其他试图进入临界区的进程必须等待；
有限等待。对请求访问的进程，应保证能在有限时间内进入临界区（保证不会饥饿）;
让权等待。当进程不能进入临界区时，应立即释放处理机，防止进程忙等待。

# 整体框架

# 进程互斥的软件实现方法

学习提示:

理解各个算法的思想、原理
结合上小节学习的 “实现互斥的四个逻辑部分”，重点理解各算法在进入区、退出区都做了什么
分析各算法存在的缺陷（结合 “实现互斥要遵循的四个原则” 进行分析）

# 单标志法

算法思想：两个进程在访问完临界区后会把使用临界区的权限转交给另一个进程。

也就是说 $\color{red}每个进程进入临界区的权限只能被另一个进程赋予$

turn 的初值为 0，即刚开始只允许 0 号进程进入临界区。

若 P1 先上处理机运行，则会一直卡在 ⑤。直到 P1 的时间片用完，发生调度，切换 P0 上处理机运行。

代码 ① 不会卡住 P0 ， P0 可以正常访问临界区，在 P0 访问临界区期间即时切换回 P1 ， P1 依然会卡在⑤。

只有 P0 在退出区将 turn 改为 1 后， P1 才能进入临界区。

因此，该算法可以实现 “同一时刻最多只允许一个进程访问临界区”

turn 表示当前允许进入临界区的进程号，而只有当前允许进入临界区的进程在访问了临界区之后，才会修改 turn 的值。

也就是说，对于临界区的访问，一定是按 P0→P1→P0→P1→...... 这样轮流访问。

这种必须 “轮流访问” 带来的问题是，

如果此时允许进入临界区的进程是 P0 ，而 P0 一直不访问临界区，那么虽然此时临界区空闲，但是并不允许 P1 访问。

因此，单标志法存在的主要问题是：

$\color{red}违背空闲让进原则$ 。

# 双标志先检查法

算法思想：设置一个布尔型数组 flag [] ，数组中各个元素用来标记各进程想进入临界区的意愿，

比如 “flag [0] = true” 意味着 0 号进程 P0 现在想要进入临界区。

每个进程在进入临界区之前先检查当前有没有别的进程想进入临界区，

如果没有，则把自身对应的标志 flag [i] 设为 true，之后开始访问临界区。

若按照 ①⑤②⑥③⑦.... 顺序执行， P0 和 P1 将会同时访问临界区。

因此，双标志先检查法存在的主要问题是：

$\color{red}违背忙则等待原则$ 。

原因在于，进入区的 “检查” 和 “上锁” 两个处理不是一气呵成的。“检查” 后，“上锁” 前可能发生讲程切换

# 双标志后检查法

算法思想：双标志先检查法的改版。前一个算法的问题是先 “检查” 后 “上锁” ，但是这两个操作又无法一气呵成，因此导致了两个进程同时进入临界区的问题。

因此，人们又想到先 “上锁” 后 “检查” 的方法，来避免上述问题。

若按照 ①⑤②⑥.... 的顺序执行， P0 和 P1 将都无法进入临界区

相当于死锁，两方都一直等待对方

因此，双标志后检查法虽然解决了 “忙则等待” 的问题，

但是 $\color{red}又违背了空闲让进和有限等待原则$

会因各进程都长期无法访问临界资源而产生 “ 饥饿 ” 现象。

两个进程都争着想进入临界区，但是谁也不让谁，最后谁都无法进入临界区。

# Peterson 算法

算法思想：双标志后检查法中，两个进程都争着想进入临界区，但是谁也不让谁，最后谁都无法进入临界区。

Gary L.Peterson 想到了一种方法，如果双方都争着想进入临界区，那可以让进程尝试 “孔融让梨” ，主动让对方先使用临界区。

两种双标志法的问题都是由于进入区的几个操作不能一气呵成导致的。

我们可以推理验证在 Peterson 算法中，两个进程进入区中的各个操作按不同的顺序穿插执行会发生什么情况:

①②③⑥⑦⑧...
①⑥②③...

因为 turn 只有一个，必然有一个判断是 false，所以不会产生死锁

抛开 turn 不谈，就回到了双标志后检查法，所以不会出现单标志法的问题

单标志法的问题：如果此时允许进入临界区的进程是 P0 ，而 P0 一直不访问临界区，那么虽然此时临界区空闲，但是并不允许 P1 访问。

进入区:

主动争取；
主动谦让；
检查对方是否也想使用，且最后一次是不是自己说了 “客气话”

例如：

香香想用马桶，如果需要的话，可以先让臭臭用，

检查发现臭臭不需要，于是香香访问临界区 —―马桶。
臭臭想用马桶，可以让香香先用

香香想用马桶，可以让臭臭先用

经检查发现香香也要使用，但最后是臭臭说了 "客气话"，因此臭臭循环等待

Peterson 算法用软件方法解决了进程互斥问题，

$\color{red}遵循了空闲让进、忙则等待、有限等待三个原则$
但是依然 $\color{red}未遵循让权等待$ 的原则。

Peterson 算法相较于之前三种软件解决方案来说，是最好的，但依然不够好。

# 整体框架

上述并不是概览，不同的算法应该相应的分析

# 进程互斥的硬件实现方法

# 中断屏蔽方法

利用 “开 / 关中断指令” 实现（与原语的实现思想相同，即在某进程开始访问临界区到结束访问为止都不允许被中断，也就不能发生进程切换，因此也不可能发生两个同时访问临界区的情况）

优点：简单、高效

缺点：

不适用于多处理机（多个处理机访问同一个临界区，导致其中剩余的处理机会一值等待）；
只适用于操作系统内核进程；
不适用于用户进程（因为开 / 关中断指令只能运行在内核态，这组指令如果能让用户随意使用会很危险）

# TestAndSet

简称 TS 指令，也有地方称为 TestAndSetLock 指令，或 TSL 指令

TSL 指令是用硬件实现的，执行的过程不允许被中断，只能一气呵成。以下是用 C 语言描述的逻辑

若刚开始 lock 是 false，则 TSL 返回的 old 值为 false， while 循环条件不满足，直接跳过循环，进入临界区。若刚开始 lock 是 true ，则执行 TLS 后 old 返回的值为 true ， while 循环条件满足，会一直循环，直到当前访问临界区的进程在退出区进行 “解锁”。

相比软件实现方法， TSL 指令把 “上锁” 和 “检查” 操作用硬件的方式变成了一气呵成的原子操作。

优点：

实现简单，无需像软件实现方法那样严格检查是否会有逻辑漏洞；
适用于多处理机环境

其他进程会卡在 ts 循环里面

缺点：

不满足 “让权等待” 原则，暂时无法进入临界区的进程会占用 CPU 并循环执行 TSL 指令，从而导致 “忙等” 。

# Swap 指令

有的地方也叫 Exchange 指令，或简称 XCHG 指令。
Swap 指令是用硬件实现的，执行的过程不允许被中断，只能一气呵成。以下是用 C 语言描述的逻辑

逻辑上来看 Swap 和 TSL 并无太大区别，都是先记录下此时临界区是否已经被上锁（记录在 old 变量上)，再将上锁标记 lock 设置为 true，最后检查 old ，如果 old 为 false 则说明之前没有别的进程对临界区上锁，则可跳出循环，进入临界区。

优点：

实现简单，无需像软件实现方法那样严格检查是否会有逻辑漏洞；
适用于多处理机环境

其他进程会卡在 ts 循环里面

缺点：

不满足 “让权等待” 原则，暂时无法进入临界区的进程会占用 CPU 并循环执行 Swap 指令，从而导致 “忙等” 。

# 整体框架

# 信号量机制

复习回顾 + 思考：之前学习的这些进程互斥的解决方案分别存在哪些问题？

进程互斥的四种软件实现方式（单标志法、双标志先检查、双标志后检查、 Peterson 算法)

进程互斥的三种硬件实现方式（中断屏蔽方法、 TS/TSL 指令、 Swap/XCHG 指令)

在双标志先检查法中，进入区的 “检查”、“上锁” 操作无法一气呵成，从而导致了两个进程有可能同时进入临界区的问题；
所有的解决方案都无法实现 “让权等待”

1965 年，荷兰学者 Dijkstra 提出了一种卓有成效的实现进程互斥、同步的方法――信号量机制

用户进程可以通过使用操作系统提供的一对原语来对信号量进行操作，从而很方便的实现了进程互斥、进程同步。

$\color{red}信号量$ ：

其实就是一个变量（ $\color{blue}可以是一个整数，也可以是更复杂的记录型变量$ )，
可以用一个信号量来 $\color{red}表示系统中某种资源的数量$

比如：系统中只有一台打印机，就可以设置一个初值为 $1$ 的信号量。

$\color{red}原语$

是一种特殊的程序段，其 $\color{red}执行只能一气呵成，不可被中断$ 。
原语是由 $\color{red}关中断/开中断指令$ 实现的。
软件解决方案的主要问题是由 “进入区的各种操作无法一气呵成” ，因此如果能把进入区、退出区的操作都用 “原语” 实现，使这些操作能 “一气呵成” 就能避免问题。

$\color{red}一对原语$ :

$\color{red}\texttt{wait(S)}$ 原语和 $\color{red}\texttt{signal(S)}$ 原语
可以把原语理解为我们自己写的函数，函数名分别为 wait 和 signal ，
括号里的 $\color{red}信号量 \texttt{S}$ 其实就是函数调用时传入的一个参数。

wait 、 signal 原语常 $\color{red}简称为 \texttt{P、V}操作$ （来自荷兰语 proberen ：尝试和 verhogen ：增加）

因此，做题的时候常把 wait(S) 、 signal(S) 两个操作分别写为 P(S) 、 V(S)

# 整形信号量

用一个 $\color{red}整数型的变量$ 作为信号量，用来 $\color{red}表示系统中某种资源的数量$ 。

与普通整数变量的区别：对信号量的操作只有三种，

即初始化、 P 操作、 V 操作

P 操作中：“检查” 和 “上锁” 一气呵成，避免了并发、异步导致的问题

因为这是原语

存在的问题：不满足 “让权等待” 原则，会发生 “忙等”

卡在 while 循环中，因为这是原语，不可中断，会一直卡在 while 循环中，除非有资源

例如：

# 记录型信号量

整型信号量的缺陷是存在 “忙等” 问题，因此人们又提出了 “记录型信号量”，即用记录型数据结构表示的信号量。

Java 中典型的 AQS

若剩余资源 > 0 的话，说明没有进程在等待这种资源，就不需要唤醒了

在考研题目中 wait(S) 、 signal(S) 也可以记为 P(S) 、 V(S) ,

这对原语可用于实现系统资源的 “申请” 和 “释放”。

$\color{red}\texttt{S.value} 的初值$ 表示系统中 $\color{red}某种资源的数目$

对信号量 S 的 $\textcolor{red}{一次 \texttt{Р}操作}意味着进程\textcolor{red}{请求一个单位的该类资源}$

因此需要执行 S.value-- ，表示资源数减 $1$ ，
当 S.value < 0 时表示该类资源已分配完毕，因此进程应调用 block 原语进行自我阻塞（当前运行的进程从运行态→阻塞态)，主动放弃处理机，并插入该类资源的等待队列 S.L 中。
可见，该机制遵循了 “让权等待” 原则，不会出现 “忙等” 现象。

对信号量 S 的 $\textcolor{red}{一次 \texttt{V}操作}意味着进程\textcolor{red}{释放一个单位的该类资源}$

因此需要执行 S.value++ ，表示资源数加 $1$ ，
若加 $1$ 后仍是 S.value <=0 ，表示依然有进程在等待该类资源，因此应调用 wakeup 原语唤醒等待队列中的第一个进程（被唤醒进程从阻塞态→就绪态）。

# 整体框架

注：若考试中出现 P(S) 、 V(S) 的操作，除非特别说明，否则默认 S 为记录型信号量。

# 用信号量机制实现进程互斥、同步、前驱关系

# 信号量机制实现进程互斥

分析并发进程的关键活动，划定临界区（如：对临界资源打印机的访问就应放在临界区)
设置 $\color{red}互斥信号量$ mutex ， $\color{red}初值为 1$
在临界区之前执行 P(mutex)
在临界区之后执行 V(mutex)

要会自己定义记录型信号量，但如果题目中没特别说明，可以把信号量的声明简写成这种形式

semaphore mytex = 1

注意：对不同的临界资源需要设置不同的互斥信号量

$\color{red}\texttt{P、V} 操作必须成对出现$

缺少 P(mutex) 就不能保证临界资源的互斥访问
缺少 V(mutex) 会导致资源用不被释放，等待进程永不被唤醒

# 用信号量机制实现进程同步

比如， P1 、 P2 并发执行，由于存在异步性，因此二者交替推进的次序是不确定的。

若 P2 的 “代码 4” 要基于 P1 的 “代码 1” 和 “代码 2 ” 的运行结果才能执行，

那么我们就必须保证 “代码 4” 一定是在 “代码 2 ” 之后才会执行。

这就是进程同步问题，让本来异步并发的进程互相配合，有序推进。

用信号量实现进程同步

分析什么地方需要实现 “同步关系” ，即必须保证 “一前一后” 执行的两个操作（或两句代码）
设置 $\color{red}同步信号量$ S， $\color{red}初始为0$
在 “前操作” $\color{red}之后$ 执行 V(S)
在 “后操作” $\color{red}之前$ 执行 P(S)

若先执行到 V(S) 操作，则 S++ 后 S=1 。

之后当执行到 P(S) 操作时，由于 S=1 ，表示有可用资源，会执行 S-- ， S 的值变回 0 ， P2 进程不会执行 block 原语，而是继续往下执行代码 $4$ 。
说明代码 $2$ 肯定是先于代码 $4$ 执行的

若先执行到 P(S) 操作，由于 S=0 ， S-- 后 S=-1 ，表示此时没有可用资源，

因此 P 操作中会执行 block 原语，主动请求阻塞。之后当执行完代码 $2$ ，继而执行 V(S) 操作， S++ ，使 S 变回 $0$ ，由于此时有进程在该信号量对应的阻塞队列中，因此会在 V 操作中执行 wakeup 原语，唤醒 P2 进程。这样 P2 就可以继续执行代码 $4$ 了

可以理解为 Java 中的 LockSupport.park() 与 LockSupport.unpark(...)

LockSupport.park() 对应 P(S)
LockSupport.unpark(...) 对应 V(S)

LockSupport 与线程中断

# 用信号量机制实现前驱关系！！

# 整体框架

除了互斥、同步问题外，还会考察有多个资源的问题，有多少资源就把信号量初值设为多少。

申请资源时进行 P 操作，释放资源时进行 V 操作即可

# 生产者 - 消费者问题

系统中有一组生产者进程和一组消费者进程，生产者进程每次生产一个产品放入缓冲区，消费者进程每次从缓冲区中取出一个产品并使用。（注：这里的 “产品” 理解为某种数据）

生产者、消费者共享一个初始为空、大小为 n 的缓冲区。

只有缓冲区没满时，生产者才能把产品放入缓冲区，否则必须等待。

同步关系。缓冲区满时，生产者要等待消费者取走产品

只有缓冲区不空时，消费者才能从中取出产品，否则必须等待。

同步关系。缓冲区为空时，消费者要等待生产者放入产品

缓冲区是临界资源，各进程必须互斥地访问

若其他进程并发地向缓冲区放入数据，会导致之前的数据会被覆盖

如何用信号量机制（ P 、 V 操作）实现生产者、消费者进程的这些功能呢？

信号量机制可实现互斥、同步、对一类系统资源的申请和释放。

设置初值为 $1$ 的互斥信号量
设置初值为 $0$ 的同步信号量（实现 “一前一后”）
设置一个信号量，初始值即为资源的数量（本质上也属于 “同步问题”，若无空闲资源，则申请资源的进程需要等待别的进程释放资源后才能继续往下执行)

# PV 操作题目分析步骤

关系分析。找出题目中描述的各个进程，分析它们之间的同步
整理思路。根据各进程的操作流程确定 P 、 V 操作的大致顺序。
设置信号量。设置需要的信号量，并根据题目条件确定信号量初值。

(互斥信号量初值一般为 1，同步信号量的初始值要看对应资源的初始值是多少）

生产者每次要消耗（P）一个空闲缓冲区，并生产（V）一个产品

消费者每次要消耗（P）一个产品，并且释放一个空闲缓冲区（V）

往缓冲区中放入 / 取走产品需要互斥

# 能否改变相邻 P、V 操作的顺序？

若此时缓冲区内已经放满产品，则 empty=0 ， full=n 。

则生产者进程执行 ① 使 mutex 变为 0 ，再执行 ② ，由于已没有空闲缓冲区，因此生产者被阻塞。由于生产者阻塞，因此切换回消费者进程。消费者进程执行 ③ ，由于 mutex 为 0，即生产者还没释放对临界资源的 “锁”，因此消费者也被阻塞。

这就造成了生产者等待消费者释放空闲缓冲区，而消费者又等待生产者释放临界区的情况，生产者和消费者循环等待被对方唤醒，出现 “死锁”。

同样的，若缓冲区中没有产品，即 full=0 ， empty=n 。按③④①的顺序执行就会发生死锁。

因此， $\color{red}实现互斥的\texttt{P}操作一定要在实现同步的\texttt{P}操作之后$ 。

V 操作不会导致进程阻塞，因此 $\color{red}两个\text{V}操作顺序可以交换$

若放到 PV 操作之间会导致并发度降低，消费者花费更多的时间去使用产品，此时还没有释放临界资源，会导致生产者不断地等待消费者。

例如：业务代码放入核心代码里

# Java 案例

	import java.util.Deque;
	import java.util.LinkedList;
	import java.util.Random;
	import java.util.concurrent.*;
	import java.util.concurrent.locks.Condition;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;

	class ProducerConsumerByLock {
	// 互斥资源
	private static Deque<Integer> queue = new LinkedList<>();
	// 最大容量
	private static int maxSize = 5;
	private static Lock lock = new ReentrantLock(false);
	private static Condition full = lock.newCondition();
	private static Condition empty = lock.newCondition();

	public static void main(String[] args) throws ExecutionException, InterruptedException {
	for (int i = 0; i < 3; i++) {
	CompletableFuture.runAsync(new Producer());
	}
	for (int i = 0; i < 3; i++) {
	CompletableFuture.runAsync(new Consumer());
	}
	//CompletableFuture.runAsync (new Consumer ()); 产生的进程默认是守护进程
	try {
	TimeUnit.SECONDS.sleep(10);
	} catch (InterruptedException e) {
	e.printStackTrace();
	}

	}

	// 生产者
	public static class Producer implements Runnable {
	public void produce() {
	while (true) {
	lock.lock();
	try {
	while (queue.size() == maxSize) {
	System.out.println("产品已满, 等待消费者进行消费, 当前生产者: " + Thread.currentThread().getName());
	// 阻塞生产者，并释放当前线程的锁
	try {
	full.await();
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}

	// 生产产品
	queue.offerLast(1);
	// 日志打印，业务逻辑尽量放在锁外面，这里只是为了顺序打印到控制台
	System.out.println("当前生产者生产产品 " + 1 + ", 当前生产者: " + Thread.currentThread().getName());
	// 唤醒其他所有的消费者与生产者
	empty.signalAll();
	full.signalAll();// 并发，让其他生产者也参与竞争
	} finally {
	lock.unlock();
	}

	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	produce();
	}
	}

	// 消费者
	public static class Consumer implements Runnable {
	// 消费产品
	public void consume() {
	while (true) {
	int x;
	lock.lock();
	try {
	while (0 == queue.size()) {
	System.out.println("产品为空, 等待生产者进行生产, 当前消费者: " + Thread.currentThread().getName());
	// 阻塞消费者者，并释放当前线程的锁
	try {
	empty.await();
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}

	// 从缓冲区获取资源
	x = queue.poll();
	// 日志打印，业务逻辑尽量放在锁外面，这里只是为了顺序打印到控制台
	System.out.println("消费者消费产品 " + x + ", 当前消费者: " + Thread.currentThread().getName());
	// 唤醒其他所有的消费者与生产者
	empty.signalAll();// 并发，让其他消费者也参与竞争
	full.signalAll();
	} finally {
	lock.unlock();
	}

	// 防止一瞬间消费完成
	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	consume();
	}
	}

	}

stdout：

当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-9
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-11
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-2
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-4
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-6
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-13
产品为空, 等待生产者进行生产, 当前消费者: ForkJoinPool.commonPool-worker-6
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-2
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-6
产品为空, 等待生产者进行生产, 当前消费者: ForkJoinPool.commonPool-worker-4
产品为空, 等待生产者进行生产, 当前消费者: ForkJoinPool.commonPool-worker-13
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-2
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-4
产品为空, 等待生产者进行生产, 当前消费者: ForkJoinPool.commonPool-worker-13
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-9
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-13
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-11
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-6
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-9
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-13
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-9
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-4
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-2
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-4
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-11
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-2
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-6
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-9
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-4
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-9
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-13
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-13
产品为空, 等待生产者进行生产, 当前消费者: ForkJoinPool.commonPool-worker-4
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-2
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-4
产品为空, 等待生产者进行生产, 当前消费者: ForkJoinPool.commonPool-worker-13
当前生产者生产产品 1, 当前生产者: ForkJoinPool.commonPool-worker-11
消费者消费产品 1, 当前消费者: ForkJoinPool.commonPool-worker-13

# 总结

PV 操作题目的解题思路

关系分析。找出题目中描述的各个进程，分析它们之间的同步
整理思路。根据各进程的操作流程确定 P 、 V 操作的大致顺序。
设置信号量。设置需要的信号量，并根据题目条件确定信号量初值。

(互斥信号量初值一般为 1，同步信号量的初始值要看对应资源的初始值是多少）

生产者消费者问题是一个互斥、同步的综合问题。

对于初学者来说最难的是发现题目中隐含的两对同步关系。

有时候是消费者需要等待生产者生产，有时候是生产者要等待消费者消费，这是两个不同的 “一前一后问题”，因此也需要设置两个同步信号量。

易错点：实现互斥和实现同步的两个 P 操作的先后顺序

实现互斥的操作要在实现同步的操作之后，否则会产生 "死锁"

# 多生产者 - 多消费者进程

桌子上有一只盘子，每次只能向其中放入一个水果。爸爸专向盘子中放苹果，妈妈专向盘子中放橘子，儿子专等着吃盘子中的橘子，女儿专等着吃盘子中的苹果。只有盘子空时，爸爸或妈妈才可向盘子中放一个水果。仅当盘子中有自己需要的水果时，儿子或女儿可以从盘子中取出水果。

生产者生产不同的产品，消费者消费相应的产品

# 问题分析

① 关系分析。找出题目中描述的各个进程，分析它们之间的同步、互斥关系。

互斥关系:

对缓冲区（盘子）的访问要互斥地进行同步关系（一前一后）:

同步关系（一前一后）

父亲将苹果放入盘子后，女儿才能取苹果
母亲将橘子放入盘子后，儿子才能取橘子
只有盘子为空时，父亲或母亲才能放入水果
- “盘子为空” 这个事件可以由儿子或女儿触发，事件发生后才允许父亲或母亲放水果

② 整理思路。根据各进程的操作流程确定 P 、 V 操作的大致顺序。

互斥：在临界资源前后分别 PV

同步：前 V 后 P

在前操作后面进行 V 操作
在后操作前面进行 P 操作

③ 设置信号量。设置需要的信号量，并根据题目条件确定信号量初值。

(互斥信号量初值一般为 1，同步信号量的初始值要看对应资源的初始值是多少)

# 具体实现

问题：可不可以不用互斥信号量？

分析：刚开始，儿子、女儿进程即使上处理机运行也会被阻塞。如果刚开始是父亲进程先上处理机运行，则：
父亲 P(plate) ，可以访问盘子 → 母亲 P(plate) ，阻塞等待盘子 → 父亲放入苹果 V(apple) ，女儿进程被唤醒，其他进程即使运行也都会阻塞，暂时不可能访问临界资源（盘子) → 女儿 P(apple) ，访问盘子， V(plate) ，等待盘子的母亲进程被唤醒→母亲进程访问盘子（其他进程暂时都无法进入临界区）>...

结论：即使不设置专门的互斥变量 mutex ，也不会出现多个进程同时访问盘子的现象

原因在于：

本题中的缓冲区大小为 1，在任何时刻， apple 、 orange 、 plate 三个同步信号量中最多只有一个是 1。
因此在任何时刻，最多只有一个进程的 P 操作不会被阻塞，并顺利地进入临界区...

若盘子容量设置为 2

父亲 P(plate) ，可以访问盘子 → 母亲 P(plate) ，可以访问盘子 → 父亲在往盘子里放苹果，同时母亲也可以往盘子里放橘子。

于是就出现了两个进程同时访问缓冲区的情况，有可能导致两个进程写入缓冲区的数据相互覆盖的情况。

因此，如果缓冲区大小大于 $1$ ，就必须专门设置一个互斥信号量 mutex 来保证互斥访问缓冲区。

# Java 案例

	import java.util.Deque;
	import java.util.LinkedList;
	import java.util.Random;
	import java.util.concurrent.*;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.locks.Condition;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;

	class ProducerConsumerByLock {
	// 互斥资源
	private static Deque<String> queueApple = new LinkedList<>();
	private static Deque<String> queueOrange = new LinkedList<>();
	private static int maxSize = 1;
	private static Lock lock = new ReentrantLock(false);
	private static Condition apple = lock.newCondition();
	private static Condition orange = lock.newCondition();
	private static Condition plate = lock.newCondition();

	public static void main(String[] args) throws ExecutionException, InterruptedException {
	CompletableFuture.runAsync(new Dad());
	CompletableFuture.runAsync(new Mom());
	CompletableFuture.runAsync(new Daughter());
	CompletableFuture.runAsync(new Son());
	//CompletableFuture.runAsync (new Consumer ()); 产生的进程默认是守护进程
	try {
	TimeUnit.SECONDS.sleep(10);
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}

	public static class Dad implements Runnable {

	// 放入苹果
	public void put() {
	while (true) {
	lock.lock();
	try {
	while (queueApple.size() == maxSize \|\| queueOrange.size() == maxSize) {
	System.out.println("盘子不为空, 父亲阻塞");
	try {
	plate.await();
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	// 放入苹果
	queueApple.offerLast("apple");
	System.out.println("父亲放入了 1 个苹果");
	// 通知苹果的所有消费者
	apple.signalAll();
	} finally {
	lock.unlock();
	}
	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	put();
	}
	}

	public static class Mom implements Runnable {

	// 放入橘子
	public void put() {
	while (true) {
	lock.lock();
	try {
	while (queueOrange.size() == maxSize \|\| queueApple.size() == maxSize) {
	System.out.println("盘子不为空, 母亲阻塞");
	try {
	plate.await();
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	// 放入橘子
	queueOrange.offerLast("orange");
	System.out.println("母亲放入了 1 个橘子");
	// 通知橘子的所有消费者
	orange.signalAll();
	} finally {
	lock.unlock();
	}
	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	put();
	}
	}


	public static class Son implements Runnable {

	// 获取橘子
	public void get() {
	while (true) {
	String v = "";
	lock.lock();
	try {
	while (queueOrange.isEmpty()) {
	System.out.println("盘子为空或者没有橘子，儿子阻塞");
	try {
	orange.await();
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	// 获取橘子
	v = queueOrange.poll();
	System.out.println("儿子吃了一个橘子: " + v);
	// 通知所有生产者，父亲和母亲
	plate.signalAll();
	} finally {
	lock.unlock();
	}
	// 防止一瞬间消费完成
	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	get();
	}
	}

	public static class Daughter implements Runnable {

	// 获取苹果
	public void get() {
	String v = "";
	while (true) {
	lock.lock();
	try {
	while (queueApple.isEmpty()) {
	System.out.println("盘子为空或者没有苹果，女儿阻塞");
	try {
	apple.await();
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	// 获取橘子
	v = queueApple.poll();
	System.out.println("女儿吃了一个苹果: " + v);
	// 通知所有生产者，父亲和母亲
	plate.signalAll();
	} finally {
	lock.unlock();
	}
	// 防止一瞬间消费完成
	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	get();
	}
	}
	}

stdout：

父亲放入了 1 个苹果
盘子不为空, 母亲阻塞
女儿吃了一个苹果: apple
母亲放入了 1 个橘子
儿子吃了一个橘子: orange
父亲放入了 1 个苹果
盘子为空或者没有橘子，儿子阻塞
女儿吃了一个苹果: apple
母亲放入了 1 个橘子
儿子吃了一个橘子: orange
盘子为空或者没有橘子，儿子阻塞
盘子为空或者没有苹果，女儿阻塞
父亲放入了 1 个苹果
女儿吃了一个苹果: apple
母亲放入了 1 个橘子
儿子吃了一个橘子: orange
母亲放入了 1 个橘子
盘子不为空, 父亲阻塞
儿子吃了一个橘子: orange
父亲放入了 1 个苹果
女儿吃了一个苹果: apple
盘子为空或者没有橘子，儿子阻塞
母亲放入了 1 个橘子
儿子吃了一个橘子: orange
盘子为空或者没有苹果，女儿阻塞

# 总结

总结：在生产者 - 消费者问题中，如果缓冲区大小为 1 ，那么有可能不需要设置互斥信号量就可以实现互斥访问缓冲区的功能。

当然，这不是绝对的，要具体问题具体分析。

建议：在考试中如果来不及仔细分析，可以加上互斥信号量，保证各进程一定会互斥地访问缓冲区。

但需要注意的是，实现互斥的 P 操作一定要在实现同步的 P 操作之后，否则可能引起 “死锁” 。

解决 “多生产者 - 多消费者问题” 的关键在于理清复杂的同步关系。

在分析同步问题（一前一后问题）的时候不能从单个进程行为的角度来分析，要把 “一前一后” 发生的事看做是两种 “事件” 的前后关系。

比如，如果从单个进程行为的角度来考虑的话，我们会有以下结论:

$\color{red}如果盘子里装有苹果，那么一定要女儿取走苹果后父亲或母亲才能再放入水果$
$\color{red}如果盘子里装有橘子，那么一定要儿子取走橘子后父亲或母亲才能再放入水果$
这么看是否就意味着要设置四个同步信号量分别实现这四个 “一前一后” 的关系了？

正确的分析方法应该从 “ $\textcolor{red}{事件}$ ” 的角度来考虑

我们可以把上述四对 “进程行为的前后关系” 抽象为一对 “事件的前后关系”
盘子变空事件 → 放入水果事件。“盘子变空事件” 既可由儿子引发，也可由女儿引发；“放水果事件”" 既可能是父亲执行，也可能是母亲执行。这样的话，就可以用一个同步信号量解决问题了

# 吸烟者问题

假设一个系统有 $\color{red}三个抽烟者进程和一个供应者进程$ 。

每个抽烟者不停地卷烟并抽掉它，但是要卷起并抽掉一支烟，抽烟者需要有三种材料：烟草、纸和胶水。三个抽烟者中，第一个拥有烟草、第二个拥有纸、第三个拥有胶水。供应者进程无限地提供三种材料，供应者每次将两种材料放桌子上，拥有剩下那种材料的抽烟者卷一根烟并抽掉它，并给供应者进程一个信号告诉完成了，供应者就会放另外两种材料再桌上，这个过程一直重复（让三个抽烟者轮流地抽烟)

smoking

# 问题分析

本质上这题也属于 “生产者 - 消费者” 问题，更详细的说应该是 “可生产多种产品的单生产者 - 多消费者”。

① 关系分析。找出题目中描述的各个进程，分析它们之间的同步、互斥关系。

桌子可以抽象为容量为 1 的缓冲区，要互斥访问

组合一：纸 + 胶水
组合二：烟草 + 胶水
组合三：烟草 + 纸

同步关系（从事件的角度来分析）:

桌上有组合一：第一个抽烟者取走东西
桌上有组合二：第二个抽烟者取走东西
桌上有组合三：第三个抽烟者取走东西
发出完成信号：供应者将下一个组合放到桌上

② 整理思路。根据各进程的操作流程确定 P 、 V 操作的大致顺序

同步关系：前 V 后 P

③ 设置信号量。设置需要的信号量，并根据题目条件确定信号量初值。

(互斥信号量初值一般为 1，同步信号量的初始值要看对应资源的初始值是多少)

# 具体实现

由于互斥资源只有一个，所以上述案例可以不需要互斥量

上述所示，若 P(finish) 放在 if 前面，则需要设置 finish 初值为 $1$ ，否则会产生死锁

放在 if 前面就跟之前案例中的盘子互斥资源一样了

缓冲区大小为 1，同一时刻，四个同步信号量中至多有一个的值为 1

# Java 案例

	import java.util.Random;
	import java.util.concurrent.*;
	import java.util.concurrent.locks.Condition;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;

	class ProducerConsumerByLock {
	// 互斥资源，由于直接在修改 i 的时候加锁了，所以不需要设置为原子类
	private static int i = 1;
	// 最大容量
	private static int capacity = 1;
	// 产品个数
	private static int count = 0;
	//flag
	private static Lock lock = new ReentrantLock(false);
	private static Condition offer1 = lock.newCondition();
	private static Condition offer2 = lock.newCondition();
	private static Condition offer3 = lock.newCondition();
	private static Condition finish = lock.newCondition();

	public static void main(String[] args) throws ExecutionException, InterruptedException {

	// 三个吸烟者
	for (int i = 1; i <= 3; i++) {
	CompletableFuture.runAsync(new Smokers(i));
	}
	Thread.sleep(1);
	// 一个生产者
	CompletableFuture.runAsync(new Producer());
	//CompletableFuture.runAsync (new Consumer ()); 产生的进程默认是守护进程
	try {
	TimeUnit.SECONDS.sleep(10);
	} catch (InterruptedException e) {
	e.printStackTrace();
	}

	}

	// 生产者
	public static class Producer implements Runnable {
	public void produce() {
	while (true) {
	lock.lock();
	try {
	while (capacity == count) {
	System.out.println("产品 " + i + " 还没有被消费,阻塞");
	finish.await();
	}
	count++;
	// 根据 i 的值不同唤醒不同的进程
	switch (i) {
	case 1:
	offer1.signalAll();
	break;
	case 2:
	offer2.signalAll();
	break;
	case 3:
	offer3.signalAll();
	break;
	default:
	}
	i = (i + 1) % 3 == 0 ? 3 : (i + 1) % 3;
	System.out.println("生产当前一个产品: " + i);

	} catch (InterruptedException e) {
	e.printStackTrace();
	} finally {
	lock.unlock();
	}
	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	produce();
	}
	}

	// 吸烟者
	@AllArgsConstructor
	public static class Smokers implements Runnable {

	private int number;

	public void smoke() {
	while (true) {
	lock.lock();
	try {
	// 如果当前吸烟者对应的材料不同，阻塞
	while (count == 0 \|\| i != number) {
	if (count == 0) {
	System.out.println("吸烟者" + number + "阻塞, 材料为空");
	} else {
	System.out.println("吸烟者" + number + "阻塞, 当前材料: " + i);
	}
	switch (i) {
	case 1:
	offer1.await();
	break;
	case 2:
	offer2.await();
	break;
	case 3:
	offer3.await();
	break;
	default:
	// 刚开始默认全部阻塞
	switch (number) {
	case 1:
	offer1.await();
	break;
	case 2:
	offer2.await();
	break;
	case 3:
	offer3.await();
	break;
	default:
	}
	}
	}
	count--;
	// 进行吸烟
	System.out.println("吸烟者" + number + "吸烟 " + i + ", 吸烟完成");
	// 唤醒生产者进行生产下一个吸烟者
	finish.signalAll();
	} catch (InterruptedException e) {
	e.printStackTrace();
	} finally {
	lock.unlock();
	}

	// 防止一瞬间消费完成
	try {
	Thread.sleep(new Random().nextInt(1000));
	} catch (InterruptedException e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void run() {
	smoke();
	}
	}

	}

stdout:

吸烟者1阻塞, 材料为空
吸烟者2阻塞, 材料为空
吸烟者3阻塞, 材料为空
生产当前一个产品: 2
吸烟者1阻塞, 当前材料: 2
吸烟者2吸烟 2, 吸烟完成
吸烟者3阻塞, 材料为空
生产当前一个产品: 3
吸烟者1阻塞, 当前材料: 3
吸烟者3吸烟 3, 吸烟完成
吸烟者2阻塞, 材料为空
吸烟者3阻塞, 材料为空
生产当前一个产品: 1
吸烟者1吸烟 1, 吸烟完成
吸烟者2阻塞, 材料为空
吸烟者3阻塞, 材料为空
生产当前一个产品: 2
吸烟者2吸烟 2, 吸烟完成
吸烟者3阻塞, 材料为空
吸烟者2阻塞, 材料为空
生产当前一个产品: 3
吸烟者3吸烟 3, 吸烟完成
吸烟者2阻塞, 材料为空
吸烟者1阻塞, 材料为空
生产当前一个产品: 1
吸烟者2阻塞, 当前材料: 1
吸烟者1吸烟 1, 吸烟完成
吸烟者3阻塞, 材料为空
生产当前一个产品: 2
吸烟者2吸烟 2, 吸烟完成
吸烟者3阻塞, 材料为空
生产当前一个产品: 3
吸烟者3吸烟 3, 吸烟完成
生产当前一个产品: 1
吸烟者1吸烟 1, 吸烟完成
吸烟者3阻塞, 材料为空
吸烟者2阻塞, 材料为空
吸烟者1阻塞, 材料为空
生产当前一个产品: 2
吸烟者3阻塞, 当前材料: 2
吸烟者2吸烟 2, 吸烟完成
吸烟者1阻塞, 材料为空
生产当前一个产品: 3
吸烟者3吸烟 3, 吸烟完成
吸烟者1阻塞, 材料为空
吸烟者2阻塞, 材料为空
生产当前一个产品: 1
吸烟者1吸烟 1, 吸烟完成
吸烟者2阻塞, 材料为空
吸烟者3阻塞, 材料为空
生产当前一个产品: 2
吸烟者2吸烟 2, 吸烟完成
吸烟者3阻塞, 材料为空
吸烟者2阻塞, 材料为空
生产当前一个产品: 3
吸烟者3吸烟 3, 吸烟完成
吸烟者2阻塞, 材料为空
吸烟者1阻塞, 材料为空
吸烟者3阻塞, 材料为空
生产当前一个产品: 1
吸烟者2阻塞, 当前材料: 1
吸烟者1吸烟 1, 吸烟完成
吸烟者3阻塞, 材料为空
吸烟者1阻塞, 材料为空
生产当前一个产品: 2
吸烟者2吸烟 2, 吸烟完成
吸烟者3阻塞, 材料为空
吸烟者1阻塞, 材料为空
生产当前一个产品: 3
吸烟者3吸烟 3, 吸烟完成
吸烟者1阻塞, 材料为空
吸烟者2阻塞, 材料为空
生产当前一个产品: 1
吸烟者1吸烟 1, 吸烟完成
吸烟者2阻塞, 材料为空
吸烟者3阻塞, 材料为空
吸烟者1阻塞, 材料为空
生产当前一个产品: 2
吸烟者2吸烟 2, 吸烟完成
吸烟者3阻塞, 材料为空
吸烟者1阻塞, 材料为空

# 总结

吸烟者问题可以为我们解决 “可以生产多个产品的单生产者” 问题提供一个思路。

值得吸取的精华是：“轮流让各个吸烟者吸烟” 必然需要 “轮流的在桌上放上组合一、二、三”，注意体会我们是如何用一个整型变量 i 实现这个 “轮流” 过程的。

如果题目改为 “每次随机地让一个吸烟者吸烟”，我们有应该如何用代码写出这个逻辑呢？

random 变量即可

若一个生产者要生产多种产品（或者说会引发多种前驱事件），那么各个 V 操作应该放在各自对应的 “前驱事件” 发生之后的位置。

# 读者 - 写者问题

有读者和写者两组并发进程，共享一个文件，当两个或两个以上的读进程同时访问共享数据时不会产生副作用，

但若某个写进程和其他进程（读进程或写进程）同时访问共享数据时则可能导致数据不一致的错误。

因此要求:

①允许多个读者可以同时对文件执行读操作；
②只允许一个写者往文件中写信息；
③任一写者在完成写操作之前不允许其他读者或写者工作；
④写者执行写操作前，应让已有的读者和写者全部退出。

# 问题分析

两类进程：写进程、读进程

互斥关系：写进程一写进程、写进程一读进程。读进程与读进程不存在互斥问题。

写者进程和任何进程都互斥，设置一个互斥信号量 rw ，在写者访问共享文件前后分别执行 P 、 V 操作。

读者进程和写者进程也要互斥，因此读者访问共享文件前后也要对 rw 执行 P 、 V 操作。

如果所有读者进程在访问共享文件之前都执行 P(rw) 操作，那么会导致各个读进程之间也无法同时访问文件。

Key：读者写者问题的核心思想――怎么处理该问题呢？

P(rw) 和 V(rw) 其实就是对共享文件的 “加锁” 和 “解锁”。既然各个读进程需要同时访问，而读进程与写进程又必须互斥访问，那么我们可以让第一个访问文件的读进程 “加锁”，让最后一个访问完文件的读进程 “解锁”。

可以设置一个整数变量 count 来记录当前有几个读进程在访问文件。

# 具体实现

潜在的问题：只要有读进程还在读，写进程就要一直阻塞等待，可能 “饿死”。

因此，这种算法中，读进程是优先的

# 总结

读者 - 写者问题为我们解决复杂的互斥问题提供了一个参考思路。

其核心思想在于设置了一个计数器 count 用来记录当前正在访问共享文件的读进程数。我们可以用 count 的值来判断当前进入的进程是否是第一个 / 最后一个读进程，从而做出不同的处理。

另外，对 count 变量的检查和赋值不能一气呵成导致了一些错误，

$\color{red}如果需要实现一气呵成，自然应该想到用互斥信号量$

最后，还要认真体会我们是如何解决 “写进程饥饿” 问题的。

绝大多数的考研 PV 操作大题都可以用之前介绍的几种生产者 - 消费者问题的思想来解决，如果遇到更复杂的问题，可以想想能否用读者写者问题的这几个思想来解决。

# 哲学家进餐问题

一张圆桌上坐着 5 名哲学家，每两个哲学家之间的桌上摆一根筷子，桌子的中间是一碗米饭。

哲学家们倾注毕生的精力用于思考和进餐，哲学家在思考时，并不影响他人。

只有当哲学家饥饿时，才试图拿起左、右两根筷子（一根一根地拿起)。如果筷子已在他人手上，则需等待。

饥饿的哲学家只有同时拿起两根筷子才可以开始进餐，当进餐完毕后，放下筷子继续思考。

# 问题分析

关系分析。系统中有 5 个哲学家进程，5 位哲学家与左右邻居对其中间筷子的访问是互斥关系。
整理思路。这个问题中只有互斥关系，但与之前遇到的问题不同的事，每个哲学家进程需要同时持有两个临界资源才能开始吃饭。如何避免临界资源分配不当造成的死锁现象，是哲学家问题的精髓。
信号量设置。定义互斥信号量数组 chopstick[5]={1,1,1,1,1} 用于实现对 5 个筷子的互斥访问。

并对哲学家按 0~4 编号，哲学家 i 左边的筷子编号为 i ，右边的筷子编号为 (i+1)%5 。

错误

循环的等待

# 如何避免死锁的发生呢？（附 Java 实现）

① 可以对哲学家进程施加一些限制条件，比如最多允许四个哲学家同时进餐。（破坏循环等待条件，本来 5 个哲学家对应 5 个筷子，现在 4 个哲学家对应 5 个筷子）

这样可以保证至少有一个哲学家是可以拿到左右两只筷子的

	semaphore chopstick [5] = {1,1,1,1,1};
	semaphore mutex = 4; // 只允许 4 个哲学家进行争抢
	pi () {
	while (1) {
	P(mutex);
	P(chopstick[i]); // 拿左
	P(chopstick[(i + 1) % 5]); // 拿右
	V(mutex); //V 顺序无所谓
	吃饭...
	V(chopstick[i]); // 放左
	V(chopstick[(i + 1) % 5]); // 放右
	思考...
	}
	}

Java 实现

	/**
	* 最多允许四个哲学家同时进餐
	*/
	class DiningPhilosophers {

	public DiningPhilosophers() {

	}

	Lock[] chopstick = new ReentrantLock[]{
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock()
	};
	Semaphore mutex = new Semaphore(4);

	// call the run() method of any runnable to execute its code
	public void wantsToEat(int philosopher,
	Runnable pickLeftFork,
	Runnable pickRightFork,
	Runnable eat,
	Runnable putLeftFork,
	Runnable putRightFork) throws InterruptedException {
	// 信号量 -1
	mutex.acquire();
	// 拿起左右筷子
	chopstick[philosopher].lock();
	chopstick[(philosopher + 1) % 5].lock();
	pickLeftFork.run();
	pickRightFork.run();
	//eat
	eat.run();
	// 放下左右筷子
	putLeftFork.run();
	putRightFork.run();
	chopstick[philosopher].unlock();
	chopstick[(philosopher + 1) % 5].unlock();
	// 信号量 + 1
	mutex.release();
	}
	}

② 要求奇数号哲学家先拿左边的筷子，然后再拿右边的筷子，而偶数号哲学家刚好相反。（破坏循环等待，分为奇偶数，给上编号）

用这种方法可以保证如果相邻的两个奇偶号哲学家都想吃饭，那么只会有其中一个可以拿起第一只筷子，另一个会直接阻塞。这就避免了占有一支后再等待另一只的情况。

	semaphore chopstick [5] = {1,1,1,1,1};
	semaphore mutex = 1; // 互斥地取筷子
	pi () {
	while (1) {
	P(mutex);
	// 偶数
	if (i & 1 == 0) {
	P(chopstick[(i + 1) % 5]); // 拿右
	P(chopstick[i]); // 拿左
	}else {
	P(chopstick[i]); // 拿左
	P(chopstick[(i + 1) % 5]); // 拿右
	}
	V(mutex); //V 顺序无所谓
	吃饭...
	V(chopstick[i]); // 放左
	V(chopstick[(i + 1) % 5]); // 放右
	思考...
	}
	}

Java 实现

	/**
	* 奇数号哲学家先左后右，偶数号哲学家先右后左
	*/
	class DiningPhilosophers {

	public DiningPhilosophers() {

	}

	Lock[] chopstick = new ReentrantLock[]{
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock()
	};

	// call the run() method of any runnable to execute its code
	public void wantsToEat(int philosopher,
	Runnable pickLeftFork,
	Runnable pickRightFork,
	Runnable eat,
	Runnable putLeftFork,
	Runnable putRightFork) throws InterruptedException {
	// 尝试拿起左右筷子
	if ((philosopher & 1) != 0) {
	chopstick[philosopher].lock();
	chopstick[(philosopher + 1) % 5].lock();
	}else {
	chopstick[(philosopher + 1) % 5].lock();
	chopstick[philosopher].lock();
	}

	pickLeftFork.run();
	pickRightFork.run();
	//eat
	eat.run();
	putLeftFork.run();
	putRightFork.run();
	// 放下左右筷子
	chopstick[philosopher].unlock();
	chopstick[(philosopher + 1) % 5].unlock();
	}
	}

③ 仅当一个哲学家左右两支筷子都可用时才允许他抓起筷子。（破坏请求和保持条件，预先分配 2 个筷子）

如下所示：

若 0 号哲学家正在 P(chopstick[i]) 中，此时 2 号哲学家也想拿筷子，将会阻塞在 P(mutex) ，直到等待 0 号哲学家 V(mutex)
若 0 号哲学家正在吃饭，0, 1 号筷子被使用；此时 1 号哲学家将会阻塞 P(chopstick[i]) ;（0 号筷子）， mutex = 0 ; 若 2 号哲学家也想拿筷子，则会被阻塞到 P(mutex)
- 即使 2 号左右两边的筷子都在，也暂时无法取得
若 0 号哲学家正在吃饭，0, 1 号筷子被使用；此时 4 号哲学家将拿起左边筷子，但是阻塞在 P(chopstick[(i + 1) % 5]) ，直到等待 0 号哲学家吃完饭 V(chopstick[i]);
- 此时 4 号右边的筷子不可用，但 4 号仍然会拿起左边的筷子

	semaphore chopstick [5] = {1,1,1,1,1};
	semaphore mutex =1; // 互斥地取筷子
	Pi () { //i 号哲学家的进程
	while (1) {
	P(mutex);
	P(chopstick[i]); // 拿左
	P(chopstick[(i + 1) % 5]); // 拿右
	V(mutex);
	吃饭...
	V(chopstick[i]); // 放左
	V(chopstick[(i + 1) % 5]); // 放右
	思考...
	}
	}

因此上述方法并不能保证只有两边的筷子都可用时，才允许哲学家拿起筷子

例如上述的情况 1 和 2 都会导致 2 号哲学家两边的筷子都可以用，但是却被阻塞

更准确的说法应该是：

各哲学家拿筷子这件事必须互斥的执行。

这就保证了即使一个哲学家在拿筷子拿到一半时被阻塞，也不会有别的哲学家会继续尝试拿筷子。

这样的话，当前正在吃饭的哲学家放下筷子后，被阻塞的哲学家就可以获得等待的筷子了。

Java 实现

	class DiningPhilosophers {

	public DiningPhilosophers() {

	}

	Lock[] chopstick = new ReentrantLock[]{
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock(),
	new ReentrantLock()
	};
	Lock mutex = new ReentrantLock();

	// call the run() method of any runnable to execute its code
	public void wantsToEat(int philosopher,
	Runnable pickLeftFork,
	Runnable pickRightFork,
	Runnable eat,
	Runnable putLeftFork,
	Runnable putRightFork) throws InterruptedException {

	// 由于要同时持有，所以要互斥
	mutex.lock();
	chopstick[philosopher].lock();
	chopstick[(philosopher + 1) % 5].lock();
	mutex.unlock();

	pickLeftFork.run();
	pickRightFork.run();
	//eat
	eat.run();
	putLeftFork.run();
	putRightFork.run();
	// 放下左右筷子
	chopstick[philosopher].unlock();
	chopstick[(philosopher + 1) % 5].unlock();
	}
	}

# 位运算 + CAS 优化

方式① 最多允许四个哲学家同时进餐

	/**
	* 最多允许四个哲学家同时进餐
	*/
	class DiningPhilosophers {

	public DiningPhilosophers() {

	}

	// 若 chopstick 等于 0, 表示筷子未被使用
	AtomicInteger chopstick = new AtomicInteger();
	// 00001, 00010, 00100, 01000, 10000
	int[] chopstickMask = new int[]{1, 2, 4, 8, 16};

	Semaphore mutex = new Semaphore(4);

	// call the run() method of any runnable to execute its code
	public void wantsToEat(int philosopher,
	Runnable pickLeftFork,
	Runnable pickRightFork,
	Runnable eat,
	Runnable putLeftFork,
	Runnable putRightFork) throws InterruptedException {
	// 左右筷子
	int leftC = chopstickMask[philosopher];
	int rightC = chopstickMask[(philosopher + 1) % 5];
	// 信号量 -1
	mutex.acquire();
	// 尝试拿起左右筷子
	//CAS
	while (!get(leftC)) {
	Thread.sleep(1);
	}
	while (!get(rightC)) {
	Thread.sleep(1);
	}
	pickLeftFork.run();
	pickRightFork.run();
	//eat
	eat.run();
	putLeftFork.run();
	putRightFork.run();
	// 放下左右筷子
	while (!put(leftC)) {
	Thread.sleep(1);
	}
	while (!put(rightC)) {
	Thread.sleep(1);
	}

	// 信号量 + 1
	mutex.release();
	}

	private boolean put(int mask) {
	int expect = chopstick.get();
	// 例如：1 号放下筷子 00001 和 00010
	// 左边的筷子 expect : 00011 ^ 00001 -> 00010
	// 右边的筷子 expect : 00010 ^ 00010 -> 00000
	return chopstick.compareAndSet(expect, expect ^ mask);
	}

	// 尝试拿起左右筷子
	private boolean get(int mask) {
	int expect = chopstick.get();
	// 只有当前筷子没有被拿起才返回 true, 并且更新筷子状态
	// 例如：1 号拿起筷子 00001 和 00010 , 更新后 : 00011 (说明 1, 2 号筷子被拿了)
	// 此时 2 号想拿起左边的筷子 : 00010 , & 运算发现被拿了，返回 false
	return (expect & mask) <= 0 && chopstick.compareAndSet(expect, expect ^ mask);
	}
	}

方式② 奇数号哲学家先左后右，偶数号哲学家先右后左

	import java.util.concurrent.Semaphore;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.locks.Condition;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;

	/**
	* 奇数号哲学家先左后右，偶数号哲学家先右后左
	*/

	class DiningPhilosophers {

	public DiningPhilosophers() {

	}

	// 若 chopstick 等于 0, 表示筷子未被使用
	AtomicInteger chopstick = new AtomicInteger();
	// 00001, 00010, 00100, 01000, 10000
	int[] chopstickMask = new int[]{1, 2, 4, 8, 16};


	// call the run() method of any runnable to execute its code
	public void wantsToEat(int philosopher,
	Runnable pickLeftFork,
	Runnable pickRightFork,
	Runnable eat,
	Runnable putLeftFork,
	Runnable putRightFork) throws InterruptedException {
	// 左右筷子
	int leftC = chopstickMask[philosopher];
	int rightC = chopstickMask[(philosopher + 1) % 5];
	// 信号量 -1
	if ((philosopher & 1) != 0) {
	// 尝试拿起左右筷子
	while (!get(leftC)) {
	Thread.sleep(1);
	}
	while (!get(rightC)) {
	Thread.sleep(1);
	}
	} else {
	// 尝试拿起右左筷子
	while (!get(rightC)) {
	Thread.sleep(1);
	}
	while (!get(leftC)) {
	Thread.sleep(1);
	}

	}

	// 退出临界区

	pickLeftFork.run();
	pickRightFork.run();
	//eat
	eat.run();
	putLeftFork.run();
	putRightFork.run();
	// 放下左右筷子
	while (!put(leftC)) {
	Thread.sleep(1);
	}
	while (!put(rightC)) {
	Thread.sleep(1);
	}
	// 信号量 + 1
	}

	private boolean put(int mask) {
	int expect = chopstick.get();
	// 例如：1 号放下筷子 00001 和 00010
	// 左边的筷子 expect : 00011 ^ 00001 -> 00010
	// 右边的筷子 expect : 00010 ^ 00010 -> 00000
	return chopstick.compareAndSet(expect, expect ^ mask);
	}

	// 尝试拿起左右筷子
	private boolean get(int mask) {
	int expect = chopstick.get();
	// 只有当前筷子没有被拿起才返回 true, 并且更新筷子状态
	// 例如：1 号拿起筷子 00001 和 00010 , 更新后 : 00011 (说明 1, 2 号筷子被拿了)
	// 此时 2 号想拿起左边的筷子 : 00010 , & 运算发现被拿了，返回 false
	return (expect & mask) <= 0 && chopstick.compareAndSet(expect, expect ^ mask);
	}
	}

方式③ 仅当一个哲学家左右两支筷子都可用时才允许他抓起筷子。

	import java.util.concurrent.Semaphore;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.locks.Condition;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;

	/**
	* 仅当一个哲学家左右两支筷子都可用时才允许他抓起筷子。
	*/

	class DiningPhilosophers {

	public DiningPhilosophers() {

	}

	// 若 chopstick 等于 0, 表示筷子未被使用
	AtomicInteger chopstick = new AtomicInteger();
	// 00001, 00010, 00100, 01000, 10000
	int[] chopstickMask = new int[]{1, 2, 4, 8, 16};

	// 进入临界区
	AtomicInteger mutex = new AtomicInteger();


	// call the run() method of any runnable to execute its code
	public void wantsToEat(int philosopher,
	Runnable pickLeftFork,
	Runnable pickRightFork,
	Runnable eat,
	Runnable putLeftFork,
	Runnable putRightFork) throws InterruptedException {
	// 左右筷子
	int leftC = chopstickMask[philosopher];
	int rightC = chopstickMask[(philosopher + 1) % 5];
	// 信号量 -1
	// 进入临界区
	while (!mutex.compareAndSet(0, 1)) {
	Thread.sleep(1);
	}
	// 尝试拿起左右筷子
	while (!get(leftC)) {
	Thread.sleep(1);
	}
	while (!get(rightC)) {
	Thread.sleep(1);
	}
	// 退出临界区
	while (!mutex.compareAndSet(1, 0)) {
	Thread.sleep(1);
	}

	pickLeftFork.run();
	pickRightFork.run();
	//eat
	eat.run();
	putLeftFork.run();
	putRightFork.run();
	// 放下左右筷子
	while (!put(leftC)) {
	Thread.sleep(1);
	}
	while (!put(rightC)) {
	Thread.sleep(1);
	}
	// 信号量 + 1
	}

	private boolean put(int mask) {
	int expect = chopstick.get();
	// 例如：1 号放下筷子 00001 和 00010
	// 左边的筷子 expect : 00011 ^ 00001 -> 00010
	// 右边的筷子 expect : 00010 ^ 00010 -> 00000
	return chopstick.compareAndSet(expect, expect ^ mask);
	}

	// 尝试拿起左右筷子
	private boolean get(int mask) {
	int expect = chopstick.get();
	// 只有当前筷子没有被拿起才返回 true, 并且更新筷子状态
	// 例如：1 号拿起筷子 00001 和 00010 , 更新后 : 00011 (说明 1, 2 号筷子被拿了)
	// 此时 2 号想拿起左边的筷子 : 00010 , & 运算发现被拿了，返回 false
	return (expect & mask) <= 0 && chopstick.compareAndSet(expect, expect ^ mask);
	}
	}

# 总结

哲学家进餐问题的关键在于解决进程死锁。

这些进程之间只存在互斥关系，但是与之前接触到的互斥关系不同的是，每个进程都需要同时持有两个临界资源，因此就有 “死锁” 问题的隐患。

如果在考试中遇到了一个进程需要同时持有多个临界资源的情况，应该参考哲学家问题的思想，分析题中给出的进程之间是否会发生循环等待，是否会发生死锁。

可以参考哲学家就餐问题解决死锁的三种思路。

# 管程

管程就是一个软件模块，里面封装了实现同步，互斥的函数

# 为什么要引入管程

信号量机制存在的问题：编写程序困难、易出错

能不能设计一种机制，让程序员写程序时不需要再关注复杂的 PV 操作，让写代码更轻松呢？

1973 年， Brinch Hansen 首次在程序设计语言（ Pascal ）中引入了 “管程” 成分 —―一种高级同步机制

# 管程的定义和基本特征

管程是一种特殊的软件模块，有这些部分组成：

局部于管程的 $\color{red}共享数据结构$ 说明；
对该数据结构进行操作的 $\color{red}一组过程$
对局部于管程的共享数据设置初始值的语句；
管程有一个名字。

Tips ："过程" 其实就是 "函数"

管程的基本特征：

局部于管程的数据只能被局部于管程的过程所访问；
一个进程只有通过调用管程内的过程才能进入管程访问共享数据；
- 类似于 Java 中 private，要访问私有变量的数据，需要提供 public 方法才可以访问
$\color{red}每次仅允许一个进程在管程内执行某个内部过程$

例：若想访问这些数据结构的话，需要调用管程中的函数间接的访问这些数据

# 拓展 1∶用管程解决生产者消费者问题

如下所示的伪代码：

除了由编译器负责实现各进程互斥地进管程中的过程之外

可以在管程中设置条件变量和等待 / 唤醒操作，已解决同步问题
例 2 ：两个消费者进程先执行，生产者进程后执行...

引入管程的目的无非就是要更方便地实现进程互斥和同步。

需要在管程中定义共享数据（如生产者消费者问题的缓冲区)
需要在管程中定义用于访问这些共享数据的 “入口” ――其实就是一些函数（如生产者消费者问题中，可以定义一个函数用于将产品放入缓冲区，再定义一个函数用于从缓冲区取出产品)
只有通过这些特定的 “入口” 才能访问共享数据
管程中有很多 “入口”，但是每次只能开放其中一个 “入口”，并且只能让一个进程或线程进入

（如生产者消费者问题中，各进程需要互斥地访问共享缓冲区。管程的这种特性即可保证一个时间段内最多只会有一个进程在访问缓冲区。)

$\color{red}注意:这种互斥特性是由编译器负责实现的，程序员不用关心$
可在管程中设置条件变量及等待 / 唤醒操作以解决同步问题。可以让一个进程或线程在条件变量上等待

（此时，该进程应先释放管程的使用权，也就是让出 “入口”)；可以通过唤醒操作将等待在条件变量上的进程或线程唤醒。

程序员可以用某种特殊的语法定义一个管程（比如: monitor ProducerConsumer .....end monitor 😉

之后其他程序员就可以使用这个管程提供的特定 “入口” 很方便地使用实现进程同步 / 互斥了。
"封装" 思想

# 拓展 2：Java 中类似于管程的机制

Java 中，如果用关键字 synchronized 来描述一个函数，那么这个函数同一时间段内只能被一个线程调用

每次只能有一个线程进入 insert 函数，如果多个线程同时调用 insert 函数，则后来者需要排队等待

# 整体框架

# 死锁

# 死锁的概念

# 什么是死锁

我等待你，你等待他，他等待她，她等待我..... 这世界每个人都爱别人.....

我们从资源占有的角度来分析，这段关系为什么看起来那么纠结...

在并发环境下，各进程因竞争资源而造成的一种互相等待对方手里的资源，导致各进程都阻塞，都无法向前推进的现象，就是 "死锁"。

发生死锁后若无外力干涉，这些进程都将无法向前推进。

# 死锁、饥饿、死循环的区别

死锁 ：各进程互相等待对方手里的资源，导致各进程都阻塞，无法向前推进的现象。

饥饿 ：由于长期得不到想要的资源，某进程无法向前推进的现象。

比如：在短进程优先（ SPF ）算法中，若有源源不断的短进程到来，则长进程将一直得不到处理机，从而发生长进程 “饥饿”。

死循环 ：某进程执行过程中一直跳不出某个循环的现象。有时是因为程序逻辑 bug 导致的，有时是程序员故意设计的。

	相同点	不同点
死锁	都是进程无法顺利向前推进的现象 (故意设计的死循环除外)	死锁一定是 “循环等待对方手里的资源” 导致的，因此如果有死锁现象，那至少有两个或两个以上的进程同时发生死锁。另外，发生死锁的进程一定处于阻塞态。
饥饿		可能只有一个进程发生饥饿。发生饥饿的进程既可能是阻塞态(如长期得不到需要的 I/O 设备)，也可能是就绪态(长期得不到处理机)
死循环		可能只有一个进程发生死循环。死循环的进程可以上处理机运行（可以是运行态），只不过无法像期待的那样顺利推进。死锁和饥饿问题是由于操作系统分配资源的策略不合理导致的，而死循环是由代码逻辑的错误导致的。死锁和饥饿是管理者（操作系统）的问题,死循环是被管理者的问题。

# 死锁产生的必要条件

产生死锁必须同时满足一下四个条件，只要其中任一条件不成立，死锁就不会发生。

$\color{red}互斥条件$ ：只有对必须互斥使用的资源的争抢才会导致死锁（如哲学家的筷子、打印机设备）。

像内存、扬声器这样可以同时让多个进程使用的资源是不会导致死锁的（因为进程不用阻塞等待这种资源）。

$\color{red}不剥夺条件$ ：进程所获得的资源在未使用完之前， $\color{red}不能由其他进程强行夺走$ ，只能主动释放。

例如：哲学家进程问题中，若每一个哲学家都持有一只筷子，则需要等待某一个哲学家释放该筷子，而不是强行夺走筷子

$\color{red}请求和保持条件$ ：进程已经 $\color{red}保持了至少一个资源$ ，但又提出了新的资源 $\color{red}请求$ ，而该资源又被其他进程占有，此时请求进程被阻塞，但又对自己已有的资源 $\color{red}保持$ 不放。

$\color{red}循环等待条件$ ：存在一种进程 $\color{red}资源的循环等待链$ ，链中的每一个进程已获得的资源同时被下一个进程所请求。

$\color{red}注意！发生死锁时一定有循环等待，但是发生循环等待时未必死锁$ （循环等待是死锁的必要不充分条件)

如果同类资源数大于 1 ，则即使有循环等待，也未必发生死锁。

但如果系统中每类资源都只有一个，那循环等待就是死锁的充分必要条件了。

# 什么时候会发生死锁

① 对系统资源的竞争。各进程对不可剥夺的资源（如打印机）的竞争可能引起死锁，对可剥夺的资源（CPU）的竞争是不会引起死锁的。

② 进程推进顺序非法。请求和释放资源的顺序不当，也同样会导致死锁。

例如，并发执行的进程 P1 、 P2 分别申请并占有了资源 R1 、 R2 ，之后进程 P1 又紧接着申请资源 R2 ，而进程 P2 又申请资源 R1 ，两者会因为申请的资源被对方占有而阻塞，从而发生死锁。

③ 信号量的使用不当也会造成死锁。

如生产者 - 消费者问题中，如果实现互斥的 P 操作在实现同步的 P 操作之前，就有可能导致死锁。(可以把互斥信号量、同步信号量也看做是一种抽象的系统资源）

总之 ，对不可剥夺资源的不合理分配，可能导致死锁。

# 死锁的处理策略

预防死锁。

破坏死锁产生的四个必要条件中的一个或几个。

避免死锁。

用某种方法防止系统进入不安全状态，从而避免死锁（银行家算法)

死锁的检测和解除。

允许死锁的发生，不过操作系统会负责检测出死锁的发生，然后采取某种措施解除死锁。

# 整体框架

# 死锁的处理策略

# 预防死锁

# 破坏互斥条件

$\color{red}互斥条件$ ：只有对必须互斥使用的资源的争抢才会导致死锁。

如果把只能互斥使用的资源改造为允许共享使用，则系统不会进入死锁状态。

比如： $\color{red}SPOOLing$ 技术。操作系统可以采用 SPOOLing 技术把独占设备在逻辑上改造成共享设备。
比如，用 SPOOLing 技术将打印机改造为共享设备...

如上右所示：进程 1 和进程 2 将作业交给输出进程即可，不需要阻塞，具体的打印操作由输出进程操作即可

可以理解为菜鸟驿站

该策略的 $\color{red}缺点$ ：并不是所有的资源都可以改造成可共享使用的资源。

并且为了系统安全，很多地方还必须保护这种互斥性。
因此， $\color{red}很多时候都无法破坏互斥条件$ 。

# 破坏不剥夺条件

$\color{red}不剥夺条件$ ：进程所获得的资源在未使用完之前，不能由其他进程强行夺走，只能主动释放。

破坏不剥夺条件:

① 方案一：当某个进程请求新的资源得不到满足时，它必须立即释放保持的所有资源，待以后需要时再重新申请。也就是说，即使某些资源尚未使用完，也需要主动释放，从而破坏了不可剥夺条件。

② 方案二：当某个进程需要的资源被其他进程所占有的时候，可以由操作系统协助，将想要的资源强行剥夺。这种方式一般需要考虑各进程的优先级（比如：剥夺调度方式，就是将处理机资源强行剥夺给优先级更高的进程使用)

该策略的 $\color{red}缺点$

实现起来比较复杂
释放已获得的资源可能造成前一阶段工作的失效。
因此这种方法一般只适用于易保存和恢复状态的资源，如 CPU 。
反复地申请和释放资源会增加系统开销，降低系统吞吐量。
若采用方案一，意味着只要暂时得不到某个资源、之前获得的那些资源就都需要放弃，以后再重新申请。如果一直发生这样的情况，就会导致进程饥饿。

# 破坏请求和保持条件

$\color{red}请求和保持条件$ ：进程 $\color{red}已经保持了至少一个资源$ ，但又提出了新的资源 $\color{red}请求$ ，而该资源又被其他进程占有，此时请求进程被阻塞，但又对自己已有的资源 $\color{red}保持$ 不放。

可以 $\color{red}采用静态分配方法$ ，即进程在运行前一次申请完它所需要的全部资源，在它的资源未满足前，不让它投入运行。一旦投入运行后，这些资源就一直归它所有，该进程就不会再请求别的任何资源了。

该策略实现起来简单，但也有明显的 $\color{red}缺点$ :

有些资源可能只需要用很短的时间，因此如果进程的整个运行期间都一直保持着所有资源，就会造成严重的资源浪费， $\color{red}资源利用率极低$ 。
另外，该策略也有 $\color{red}可能导致某些进程饥饿$ 。

如下所示：若有源源不断地 A 类进程进来，资源 1 一直被分配给 A 类进程，会导致 C 类进程饥饿

# 破坏循环等待条件

$\color{red}循环等待条件$ ：存在一种进程资源的 $\color{red}循环等待链$ ，链中的每一个进程已获得的资源同时被下一个进程所请求。

可采用 $\color{red}顺序资源分配法$ 。首先给系统中的资源编号，规定每个进程必 $\color{red}须按编号递增的顺序请求资源$ ，同类资源（即编号相同的资源）一次申请完。

原理分析：一个进程只有已占有小编号的资源时，才有资格申请更大编号的资源。按此规则，已持有大编号资源的进程不可能逆向地回来申请小编号的资源，从而就不会产生循环等待的现象。

该策略的 $\color{red}缺点$ :

不方便增加新的设备，因为可能需要重新分配所有的编号；
进程实际使用资源的顺序可能和编号递增顺序不一致，会导致资源浪费；

例如：5 号资源为打印机，7 号资源为扫描仪。 P3 进程应该先使用扫描仪，在使用打印机，但是实际上缺失先使用打印机，再使用扫描仪，导致打印机空闲了很长一段时间，等到扫描仪使用完了在使用打印机资源。因此导致系统资源的浪费

必须按规定次序申请资源，用户编程麻烦。

# 避免死锁（银行家算法）

# 什么是安全序列

你是一位成功的银行家，手里掌握着 $100$ 个亿的资金...
有三个企业想找你贷款，分别是企业 B 、企业 A 、企业 T ，为描述方便，简称 BAT 。

B 表示:“大哥，我最多会跟你借 $70$ 亿...”

A 表示:“大哥，我最多会跟你借 $40$ 亿..”

T 表示:“大哥，我最多会跟你借 $50$ 亿..”

然而... 江湖中有个不成文的规矩：如果你借给企业的钱总数达不到企业提出的最大要求，那么不管你之前给企业借了多少钱，那些钱都拿不回来了....

刚开始， BAT 三个企业分别从你这儿借了 $20$ 、 $10$ 、 $30$ 亿....

① 先给 B 借 $30$ 亿

所以给 B 借 $30$ 亿是不安全的...

② 先给 A 借 $20$ 亿，再借给 B $50$ 亿

③ 先给 A 借 $20$ 亿，再借给 T $20$ 亿

# 安全序列、不安全状态、死锁的联系

所谓 $\color{red}安全序列$ ，就是指如果系统按照这种序列分配资源，则每个进程都能顺利完成。只要能找出一个安全序列，系统就是 $\color{red}安全状态$ 。当然， $\color{red}安全序列可能有多个$ 。

如果分配了资源之后，系统中找不出任何一个安全序列，系统就进入了 $\color{red}不安全状态$ 。这就意味着之后 $\color{red}可能$ 所有进程都无法顺利的执行下去。当然，如果有进程提前归还了一些资源，那 $\color{red}系统也有可能重新回到安全状态$ ，不过我们在分配资源之前总是要考虑到最坏的情况。

如果系统处于 $\color{red}安全状态$ ，就 $\color{red}一定不会$ 发生 $\color{red}死锁$ 。如果系统进入 $\color{red}不安全状态$ ，就 $\color{red}可能$ 发生 $\color{red}死锁$ （处于不安全状态未必就是发生了死锁，但发生死锁时一定是在不安全状态)

因此可以 $\color{red}在资源分配之前预先判断这次分配是否会导致系统进入不安全状态$ ，以此决定是否答应资源分配请求。这也是 “ $\color{red}银行家算法$ ” 的核心思想。

# 银行家算法

银行家算法是荷兰学者 Dijkstra 为银行系统设计的，以确保银行在发放现金贷款时，不会发生不能满足所有客户需要的情况。后来该算法被用在操作系统中，用于 $\color{red}避免死锁$ 。

$\color{red}核心思想$ ：在进程提出资源申请时， $\color{red}先预判此次分配是否会导致系统进入不安全状态$

如果会进入不安全状态，就暂时不答应这次请求，让该进程先阻塞等待。

此时系统是否处于安全状态？

思路：尝试找出一个安全序列... $\color{red}{P1}$

依次检查剩余可用资源 (3,3,2) 是否能满足各进程的需求

① 可满足 P1 需求，将 P1 加入安全序列，并更新剩余可用资源值为 (5,3,2)

依次检查剩余可用资源 (5,3,2) 是否能满足剩余进程（ $\color{red}不包括已加入安全序列的进程$ ）的需求

说明如果优先把资源分配给 P3 , 那 P3 一定是可以顺利执行结束的。等 P3 结束了就会归还资源。

于是，资源数就可以增加到 (2,1,1)+(5,3,2)=(7,4,3)

② 可满足 P3 需求，将 P3 加入安全序列，并更新剩余可用资源值为 (7,4,3)

依次检查剩余可用资源 (7,4,3) 是否能满足剩余进程（不包括已加入安全序列的进程）的需求.....

……….

以此类推，共五次循环检查即可将 $5$ 个进程都加入安全序列中，最终可得一个安全序列。

该算法称为 $\color{red}安全性算法$ 。可以很方便地用代码实现以上流程，每一轮检查都从编号较小的进程开始检查实际做题时可以更快速的得到安全序列。

对于手算

① 可以找到安全序列的例子

实际做题（手算）时可用更快速的方法找到一个安全序列:
经对比发现， (3,3,2) 可满足 P1 、 P3 ，说明无论如何，这两个进程的资源需求一定是可以依次被满足的，因此 P1 、 P3 一定可以顺利的执行完，并归还资源。可把 P1 、 P3 先加入安全序列。
(2,0,0)+(2,1,1)+(3,3,2)= (7,4,3)

剩下的 P0 、 P2 、 P4 都可被满足。同理，这些进程都可以加入安全序列。

于是， $5$ 个进程全部加入安全序列，说明此时系统 $\color{red}处于安全状态$ ，暂 $\color{red}不可能发生死锁$

② 找不到安全序列的例子

经对比发现， (3,3,2) 可满足 P1 、 P3 ，说明无论如何，这两个进程的资源需求一定是可以依次被满足的，因此 P1 、 P3 一定可以顺利的执行完，并归还资源。可把 P1 、 P3 先加入安全序列。

(2,0,0)+(2,1,1)+(3,3,2)= (7,4,3)

剩下的 P0 需要 (8,4,3) ， P2 需要 (6,5,0) ， P4 需要 (4,3,4) 任何一个进程都不能被完全满足

于是，无法找到任何一个安全序列，说明此时系统处于 $\color{red}不安全状态$ ， $\color{red}有可能发生死锁$ 。

# Java 实现银行家算法

BankerAlgorithm.class

	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.List;

	public class BankerAlgorithm {

	/**
	* @param maximumDemands 最大需求，第 i 个进程的最大需求为数组 j
	* @param assigneds 已分配，第 i 个进程的已分配的数量为数组 j
	* @param totalResource 可用资源数
	* @return 是否是安全序列
	*/
	public boolean isSafeSequence(int[][] maximumDemands, int[][] assigneds, int[] totalResource) {
	// 预处理，获取当前进程最多还需要的需求
	int[][] mostNeededs = new int[maximumDemands.length][maximumDemands[0].length];
	int[] remainResource = totalResource;
	for (int i = 0; i < maximumDemands.length; i++) {
	// 最多还需要
	int[] curNeed = new int[maximumDemands[i].length];
	// 最大需求
	int[] curMaxDemand = maximumDemands[i];
	// 已分配
	int[] curAssigned = assigneds[i];
	for (int j = 0; j < curMaxDemand.length; j++) {
	curNeed[j] = (curMaxDemand[j] - curAssigned[j]);
	int total = remainResource[j];
	remainResource[j] = total - curAssigned[j];
	}
	mostNeededs[i] = curNeed;
	}

	return dfs(assigneds, mostNeededs, remainResource, new boolean[maximumDemands.length], 0);
	}

	public boolean dfs(int[][] assigneds, int[][] mostNeededs, int[] remainResource, boolean[] isVisited, int cnt) {
	// 被分配完了
	if (cnt == mostNeededs.length) {
	return true;
	}
	for (int i = 0; i < mostNeededs.length; i++) {
	// 被访问过了
	if (isVisited[i]) {
	continue;
	}
	// 满足需求
	int[] check = check(assigneds[i], mostNeededs[i], remainResource);
	if (check != null) {
	int[] temp = remainResource;
	remainResource = check;
	// 移除当前进程
	isVisited[i] = true;
	if (dfs(assigneds, mostNeededs, remainResource, isVisited, cnt + 1)) {
	return true;
	}
	isVisited[i] = false;
	remainResource = temp;
	}
	}
	return false;
	}

	// 检查是否满足需求，并且重新计算剩余资源数
	public int[] check(int[] assigned, int[] mostNeeded, int[] remainResource) {
	int[] ans = new int[mostNeeded.length];
	for (int i = 0; i < mostNeeded.length; i++) {
	int diff = remainResource[i] - mostNeeded[i];
	if (diff < 0) {
	return null;
	}
	ans[i] = assigned[i] + remainResource[i];
	}
	return ans;
	}
	}

Test.class

	public class Test{
	public static void main(String[] args) {
	BankerAlgorithm bankerAlgorithm = new BankerAlgorithm();
	//int[][] maximumDemands = new int[][]{ {7, 5, 3}, {3, 2, 2}, {9, 0, 2}, {2, 2, 2}, {4, 3, 3}};
	//int[][] assigneds = new int[][]{ {0, 1, 0}, {2, 0, 0}, {3, 0, 2}, {2, 1, 1}, {0, 0, 2}};
	//true

	int[][] maximumDemands = new int[][]{ {8, 5, 3}, {3, 2, 2}, {9, 5, 2}, {2, 2, 2}, {4, 3, 6}};
	int[][] assigneds = new int[][]{ {0, 1, 0}, {2, 0, 0}, {3, 0, 2}, {2, 1, 1}, {0, 0, 2}};
	//false

	int[] totalResource = new int[]{10, 5, 7};
	System.out.println(bankerAlgorithm.isSafeSequence(maximumDemands, assigneds, totalResource));
	}
	}

# 总结

银行家算法步骤:

① 检查此次申请是否超过了之前声明的最大需求数

② 检查此时系统剩余的可用资源是否还能满足这次请求

③ 试探着分配，更改各数据结构

④ 用安全性算法检查此次分配是否会导致系统进入不安全状态

安全性算法步骤:

检查当前的剩余可用资源是否能满足某个进程的最大需求，如果可以，就把该进程加入安全序列，并把该进程持有的资源全部回收。

系统处于不安全状态未必死锁，但死锁时一定处于不安全状态。系统处于安全状态一定不会死锁。

# 检测和解除

如果系统中既不采取预防死锁的措施，也不采取避免死锁的措施，系统就很 $\color{red}可能发生死锁$ 。在这种情况下，系统应当提供两个算法:

① 死锁检测算法：用于检测系统状态，以确定系统中是否发生了死锁。
② 死锁解除算法：当认定系统中已经发生了死锁，利用该算法可将系统从死锁状态中解脱出来。

# 死锁的检测

为了能对系统是否已发生了死锁进行检测，必须:

① 用 $\color{red}某种数据结构$ 来保存资源的请求和分配信息；

② 提供 $\color{red}一种算法$ ，利用上述信息来检测系统是否已进入死锁状态。

# Java 定义该数据结构（资源分配图）

① map 构造邻接表

	public class ResourceAllocationGraph {
	/**
	* 对于 Map<Node, List<Pair>> map
	* 若 Node 为进程则 List<Pair> 为申请某些资源数量，
	* 例如：P1 (false, 1) : [Pair (R1, 1)]
	* 表示 p1 请求资源的总数量为 1, 申请资源 R1 的数量为 1
	* 若 Node 为资源则 List<Pair> 为分配给进程的资源数量
	* 例如：R1 (true, 0) : [Pair (P1, 2), Pair (P2, 1)]
	* 表示 R1 剩余资源为 0, 分配给 P1, p2 的资源数分别为 2, 1
	*/
	List<Node> processes;
	List<Node> resources;
	}

	class Node {
	// 是否是资源的标志， true : 资源， false : 进程
	boolean isResource;

	// 若为资源：当前剩余资源的总数量
	// 若为进程：当前进程请求资源的总数量
	int val;

	// 进程: pi 被分配的资源的节点 + 相应的分配数
	// 资源: ri 被请求的进程节点 + 相应的请求数
	Map<Node, Integer> inDegreeNodes;

	// 进程：pi 申请的资源节点 + 相应的申请数
	// 资源: ri 分配出去的进程节点 + 相应的分配数
	Map<Node, Integer> outDegreeNodes;
	public Node(boolean isResource, int val) {
	this.isResource = isResource;
	this.val = val;
	}
	}

② 双向链表

	public class ResourceAllocationGraph {
	Node prevProcess;
	Node nextProcess;
	Node prevResource;
	Node nextResource;
	int pSize();
	int rSize();
	}
	class Node {
	// 是否是资源的标志， true : 资源， false : 进程
	boolean isResource;

	// 若为资源：当前剩余资源的总数量
	// 若为进程：当前进程请求资源的总数量
	int val;

	// 进程: pi 被分配的资源的节点 + 相应的分配数
	// 资源: ri 被请求的进程节点 + 相应的请求数
	Map<Node, Integer> inDegreeNodes;

	// 进程：pi 申请的资源节点 + 相应的申请数
	// 资源: ri 分配出去的进程节点 + 相应的分配数
	Map<Node, Integer> outDegreeNodes;
	public Node(boolean isResource, int val) {
	this.isResource = isResource;
	this.val = val;
	}
	}

# 死锁检测的流程

① 如果系统中剩余的可用资源数足够满足进程的需求，那么这个进程暂时是不会阻塞的，可以顺利地执行下去。

② 如果这个进程执行结束了把资源归还系统，就可能使某些正在等待资源的进程被激活，并顺利地执行下去。

③ 相应的，这些被激活的进程执行完了之后又会归还一些资源，这样可能又会激活另外一些阻塞的进程..

如果按上述过程分析，最终 $\color{red}能消除所有边$ ，就称这个图是 $\color{red}<i>可完全简化的</i>$ 。此时一定 $\color{red}没有发生死锁$ （相当于能找到一个安全序列)

如果最终 $\color{red}不能消除所有边$ ，那么此时就是 $\color{red}发生了死锁$

$\color{red}最终还连着边的那些进程就是处于死锁状态的进程$ 。

安全序列的情况

上述所示：

对于 P2 来说，若 P2 想申请 R1 资源，但是 R1 已经分配出去 $3$ 个，因此 P2 的申请请求不能被满足
对于 P1 来说，若 P1 想申请 R2 资源， R2 已经分给 P2 一个，还剩余 $1$ 个，所以 P1 可以得到满足。然后等待 P1 运行完毕之后， P1 会释放自己所拥有的资源（即： R1 ：0 -> 2）。之后 P2 也可以得到满足...

具体流程如下所示：

产生死锁的情况

$\color{red}最终还连着边的那些进程就是处于死锁状态的进程$

如下所示

# 死锁检测的算法

1）在资源分配图中，找出 既不阻塞又不是孤点的进程 Pi （即找出一条有向边与它相连，且该有向边对应资源的申请数量小于等于系统中已有空闲资源数量。

如下图中， R1 没有空闲资源， R2 有一个空闲资源。若所有的连接该进程的边均满足上述条件，则这个进程能继续运行直至完成，然后释放它所占有的所有资源）。消去它所有的请求边和分配变，使之称为孤立的结点。在下图中， P1 是满足这一条件的进程结点，于是将 P1 的所有边消去。

2）进程 Pi 所释放的资源，可以唤醒某些因等待这些资源而阻塞的进程，原来的阻塞进程可能变为非阻塞进程。

在下图中， P2 就满足这样的条件。根据 1) 中的方法进行一系列简化后，若能消去途中所有的边，则称该图是 $\color{red}可完全简化的$ 。

P1 的所有边消去后的资源分配图

P2 的所有边消去后的可完全简化的资源分配图

# Java 实现死锁检测的算法

资源分配图

	// 资源分配图
	class ResourceAllocationGraph {
	/**
	* 对于 Map<Node, List<Pair>> map
	* 若 Node 为进程则 List<Pair> 为申请某些资源数量，
	* 例如：P1 (false, 1) : [Pair (R1, 1)]
	* 表示 p1 请求资源的总数量为 1, 申请资源 R1 的数量为 1
	* 若 Node 为资源则 List<Pair> 为分配给进程的资源数量
	* 例如：R1 (true, 0) : [Pair (P1, 2), Pair (P2, 1)]
	* 表示 R1 剩余资源为 0, 分配给 P1, p2 的资源数分别为 2, 1
	*/
	List<Node> processes;
	List<Node> resources;
	}

	class Node {
	// 是否是资源的标志， true : 资源， false : 进程
	boolean isResource;

	// 若为资源：当前剩余资源的总数量
	// 若为进程：当前进程请求资源的总数量
	int val;

	// 进程: pi 被分配的资源的节点 + 相应的分配数
	// 资源: ri 被请求的进程节点 + 相应的请求数
	Map<Node, Integer> inDegreeNodes;

	// 进程：pi 申请的资源节点 + 相应的申请数
	// 资源: ri 分配出去的进程节点 + 相应的分配数
	Map<Node, Integer> outDegreeNodes;
	public Node(boolean isResource, int val) {
	this.isResource = isResource;
	this.val = val;
	}
	}

step 1 : 构建资源分配图

	public ResourceAllocationGraph builderResourceGraph() {
	// 资源
	Node r1 = new Node(true, 0);
	Node r2 = new Node(true, 1);
	// 进程
	// Node p1 = new Node(false, 1);
	Node p1 = new Node(false, 2);
	Node p2 = new Node(false, 1);

	//r1 分配出去的资源和请求 r1 资源的进程
	r1.outDegreeNodes = new HashMap<>();
	r1.outDegreeNodes.put(p1, 2);
	r1.inDegreeNodes = new HashMap<>();
	r1.inDegreeNodes.put(p2, 1);
	//r2 分配出去的资源和请求 r1 资源的进程
	r2.outDegreeNodes = new HashMap<>();
	r2.outDegreeNodes.put(p2, 1);
	r2.inDegreeNodes = new HashMap<>();
	// r2.inDegreeNodes.put(p1, 1);
	r2.inDegreeNodes.put(p1, 2);
	//p1 申请的资源和被分配的资源
	p1.outDegreeNodes = new HashMap<>();
	// p1.outDegreeNodes.put(r2, 1);
	p1.outDegreeNodes.put(r2, 2);
	p1.inDegreeNodes = new HashMap<>();
	p1.inDegreeNodes.put(r1, 2);
	//p2 申请的资源和被分配的资源
	p2.outDegreeNodes = new HashMap<>();
	p2.outDegreeNodes.put(r1, 1);
	p2.inDegreeNodes = new HashMap<>();
	p2.inDegreeNodes.put(r2, 1);
	// 建立资源分配图
	ResourceAllocationGraph graph = new ResourceAllocationGraph();
	graph.processes = new ArrayList<>();
	graph.processes.add(p1);
	graph.processes.add(p2);
	graph.resources = new ArrayList<>();
	graph.resources.add(r1);
	graph.resources.add(r2);
	return graph;
	}

step 2 : 死锁检测，true : 死锁，false : 不产生死锁

	public boolean hasDeadlock(ResourceAllocationGraph graph, boolean[] isolated, int cnt) {
	//1. 找出即不阻塞又不是孤立的进程 pi
	// 即找出一条有向边与它相连，且该有向边对应资源的申请数量小于等于系统中已有空闲资源数量
	List<Node> processes = graph.processes;
	// 所有的 pi 都孤立了
	if (cnt == 0) {
	return false;
	}
	// 可能有多个
	List<Node> pis = new ArrayList<>();
	//pi 申请的资源必须全部满足条件
	for (int i = 0; i < processes.size(); i++) {
	// 说明 pi 已经被孤立了
	if (isolated[i]) {
	continue;
	}
	Node pi = processes.get(i);
	//pi 申请的资源节点剩余资源 >= pi 申请的资源
	boolean flag = true;
	for (Map.Entry<Node, Integer> entry : pi.outDegreeNodes.entrySet()) {
	Node node = entry.getKey();
	if (node.val < entry.getValue()) {
	flag = false;
	break;
	}
	}
	if (flag) {
	pis.add(pi);
	}
	}
	// 没有 pi , 产生了死锁
	if (pis.size() == 0) {
	return true;
	}
	//2. pi 释放资源
	for (int i = 0; i < pis.size(); i++) {
	Node pi = pis.get(i);
	//pi 对应申请的相应资源节点，释放 pi
	pi.outDegreeNodes.forEach((node, integer) -> {
	node.inDegreeNodes.remove(pi);
	});
	// 分配给 pi 的相应资源节点增加相应的资源数，并释放 pi
	pi.inDegreeNodes.forEach((node, integer) -> {
	node.val += integer;
	node.outDegreeNodes.remove(pi);
	});
	// 让 pi 孤立
	isolated[i] = true;
	cnt--;
	}
	// 递归地检测，没有死锁 false, 有死锁 true
	return hasDeadlock(graph, isolated, cnt);
	}

整体代码

	public class DeadlockDetection {
	public static void main(String[] args) {
	//1. 构建资源分配图
	DeadlockDetection deadlockDetection = new DeadlockDetection();
	ResourceAllocationGraph graph = deadlockDetection.builderResourceGraph();
	//2. 死锁检测，true : 死锁，false : 不产生死锁
	System.out.println(deadlockDetection.hasDeadlock(graph, new boolean[graph.processes.size()], graph.processes.size()));
	}

	public boolean hasDeadlock(ResourceAllocationGraph graph, boolean[] isolated, int cnt) {
	//1. 找出即不阻塞又不是孤立的进程 pi
	// 即找出一条有向边与它相连，且该有向边对应资源的申请数量小于等于系统中已有空闲资源数量
	List<Node> processes = graph.processes;
	// 所有的 pi 都孤立了
	if (cnt == 0) {
	return false;
	}
	// 可能有多个
	List<Node> pis = new ArrayList<>();
	//pi 申请的资源必须全部满足条件
	for (int i = 0; i < processes.size(); i++) {
	// 说明 pi 已经被孤立了
	if (isolated[i]) {
	continue;
	}
	Node pi = processes.get(i);
	//pi 申请的资源节点剩余资源 >= pi 申请的资源
	boolean flag = true;
	for (Map.Entry<Node, Integer> entry : pi.outDegreeNodes.entrySet()) {
	Node node = entry.getKey();
	if (node.val < entry.getValue()) {
	flag = false;
	break;
	}
	}
	if (flag) {
	pis.add(pi);
	}
	}
	// 没有 pi , 产生了死锁
	if (pis.size() == 0) {
	return true;
	}
	//2. pi 释放资源
	for (int i = 0; i < pis.size(); i++) {
	Node pi = pis.get(i);
	//pi 对应申请的相应资源节点，释放 pi
	pi.outDegreeNodes.forEach((node, integer) -> {
	node.inDegreeNodes.remove(pi);
	});
	// 分配给 pi 的相应资源节点增加相应的资源数，并释放 pi
	pi.inDegreeNodes.forEach((node, integer) -> {
	node.val += integer;
	node.outDegreeNodes.remove(pi);
	});
	// 让 pi 孤立
	isolated[i] = true;
	cnt--;
	}
	// 递归地检测，没有死锁 false, 有死锁 true
	return hasDeadlock(graph, isolated, cnt);
	}

	public ResourceAllocationGraph builderResourceGraph() {
	// 资源
	Node r1 = new Node(true, 0);
	Node r2 = new Node(true, 1);
	// 进程
	// Node p1 = new Node(false, 1);
	Node p1 = new Node(false, 2);
	Node p2 = new Node(false, 1);

	//r1 分配出去的资源和请求 r1 资源的进程
	r1.outDegreeNodes = new HashMap<>();
	r1.outDegreeNodes.put(p1, 2);
	r1.inDegreeNodes = new HashMap<>();
	r1.inDegreeNodes.put(p2, 1);
	//r2 分配出去的资源和请求 r1 资源的进程
	r2.outDegreeNodes = new HashMap<>();
	r2.outDegreeNodes.put(p2, 1);
	r2.inDegreeNodes = new HashMap<>();
	// r2.inDegreeNodes.put(p1, 1);
	r2.inDegreeNodes.put(p1, 2);
	//p1 申请的资源和被分配的资源
	p1.outDegreeNodes = new HashMap<>();
	// p1.outDegreeNodes.put(r2, 1);
	p1.outDegreeNodes.put(r2, 2);
	p1.inDegreeNodes = new HashMap<>();
	p1.inDegreeNodes.put(r1, 2);
	//p2 申请的资源和被分配的资源
	p2.outDegreeNodes = new HashMap<>();
	p2.outDegreeNodes.put(r1, 1);
	p2.inDegreeNodes = new HashMap<>();
	p2.inDegreeNodes.put(r2, 1);
	// 建立资源分配图
	ResourceAllocationGraph graph = new ResourceAllocationGraph();
	graph.processes = new ArrayList<>();
	graph.processes.add(p1);
	graph.processes.add(p2);
	graph.resources = new ArrayList<>();
	graph.resources.add(r1);
	graph.resources.add(r2);
	return graph;
	}
	}

	// 资源分配图
	class ResourceAllocationGraph {
	/**
	* 对于 Map<Node, List<Pair>> map
	* 若 Node 为进程则 List<Pair> 为申请某些资源数量，
	* 例如：P1 (false, 1) : [Pair (R1, 1)]
	* 表示 p1 请求资源的总数量为 1, 申请资源 R1 的数量为 1
	* 若 Node 为资源则 List<Pair> 为分配给进程的资源数量
	* 例如：R1 (true, 0) : [Pair (P1, 2), Pair (P2, 1)]
	* 表示 R1 剩余资源为 0, 分配给 P1, p2 的资源数分别为 2, 1
	*/
	List<Node> processes;
	List<Node> resources;
	}

	class Node {
	// 是否是资源的标志， true : 资源， false : 进程
	boolean isResource;

	// 若为资源：当前剩余资源的总数量
	// 若为进程：当前进程请求资源的总数量
	int val;

	// 进程: pi 被分配的资源的节点 + 相应的分配数
	// 资源: ri 被请求的进程节点 + 相应的请求数
	Map<Node, Integer> inDegreeNodes;

	// 进程：pi 申请的资源节点 + 相应的申请数
	// 资源: ri 分配出去的进程节点 + 相应的分配数
	Map<Node, Integer> outDegreeNodes;
	public Node(boolean isResource, int val) {
	this.isResource = isResource;
	this.val = val;
	}
	}

# 死锁定理

$\color{red}死锁定理$ ：如果某时刻系统的资源分配图是 $\color{red}不可完全简化$ 的，那么此时系统 $\color{red}死锁$

# 死锁的解除

一旦检测出死锁的发生，就应该立即解除死锁。

补充：并不是系统中所有的进程都是死锁状态，用死锁检测算法 $\color{red}化简资源分配图后，还连着边的那些进程就是死锁进程$

# 资源剥夺法

$\color{red}资源剥夺法$ ：

挂起（暂时放到外存上）某些死锁进程，并抢占它的资源，将这些资源分配给其他的死锁进程。
但是应防止被挂起的进程长时间得不到资源而 饥饿 。

# 撤销进程法 (终止进程法)

$\color{red}撤销进程法$ （或称 $\color{red}终止进程法$ ）：

强制撤销部分、甚至全部死锁进程，并剥夺这些进程的资源。

这种方式的优点是实现简单，但所付出的代价可能会很大。

因为有些进程可能已经运行了很长时间，已经接近结束了，一旦被终止可谓 功亏一篑 ，以后还得从头再来。

# 进程回退法

$\color{red}进程回退法$ ：

让一个或多个死锁进程回退到足以避免死锁的地步。
这就要求系统要记录进程的历史信息，设置 还原点 。

# 如何决定 "对谁动手"

① 进程优先级

例如：进程优先级低的对其下手

② 已执行多长时间

例如：执行时间更少的进程，让其牺牲

③ 还要多久能完成

例如：牺牲掉还需要更长的时间完成，优先让即将完成的进程获得资源

④ 进程已经使用了多少资源

例如：优先牺牲掉已经使用更多资源的进程

⑤ 进程是交互式的还是批处理式的

若牺牲掉交互式的（即：正在与用户交互的进程），则用户不满意
批处理式的无非就是在做一些计算，对用户的及时反馈并不那么在意
所以优先牺牲掉批处理式的