15.md

十五、使用 Pthread 的多线程——第二部分：同步

多线程功能强大并对性能有很大影响的关键原因之一是，它支持并行性或并发性的概念；从我们在前面的第 14 章，使用 Pthreadsi-essentials学到的知识中，我们了解到一个进程的多个线程可以(并且确实确实)并行执行。 在大型多核系统上(多核几乎是现在的标准，甚至在嵌入式系统中也是如此)，这种影响被放大了。

然而，正如经验告诉我们的那样，总是有取舍的。 平行随之而来的是两场比赛的丑陋潜力，以及随后的缺陷。 不仅如此，像这样的情况通常会变得非常难以调试，因此也很难修复。

在本章中，我们将尝试：

• 让读者知道这些并发(竞争)缺陷的确切位置和具体内容
• 如何在多线程应用中通过良好的设计和编码实践来避免它们

同样，本章分为两大领域：

• 在第一部分，我们清楚地解释了问题，比如原子性是如何重要的，以及僵局问题。
• 在本章的后半部分，我们将介绍 pthreadsAPI 集为应用开发人员提供的锁定(和其他)机制，以帮助解决并完全避免这些问题。

赛车问题

首先，也是最重要的，让我们试图了解我们试图解决的问题到底是什么，在哪里。 在上一章中，我们了解到一个进程的所有线程可以共享除堆栈之外的所有内容；每个线程都有自己的私有堆栈内存空间。

再仔细看看第 14 章，使用 PthreadsPart I-Essentials 进行多线程：图 2，2(省略内核内容)；虚拟地址空间-文本和数据段，而不是堆栈段-在进程的所有线程之间共享。 当然，数据段是全局变量和静态变量所在的位置。

冒着夸大这些事实的风险，这意味着给定进程的所有线程(如果不是 POSS，则使 COW 也不是写入时复制(COW)的正常字体)共享以下内容：

• 文本片段
• 数据段-初始化数据、未初始化数据(以前称为 BSS)和堆段
• 几乎所有内核级别的对象和数据都是由操作系统为进程维护的(同样，请参阅第 14 章，使用 Pthread 多线程，第 I 部分-Essentials和*：图 2*)

需要理解的真正重要的一点是，分享新的文本片段并不是什么问题。 为什么？ 文本就是代码；机器代码--构成我们所说的机器语言的操作码和操作数--驻留在这些内存页面中。 回想一下第 2 章，虚拟内存，所有的文本(代码)页都有相同的权限：读取-执行(r-x)。这一点很重要，因为多个线程并行执行文本(代码)不仅很好-我们鼓励这样做！ 毕竟，这就是并行的全部意义所在。 想想看，如果我们只读取和执行代码，我们就不会以任何方式修改它；因此，即使在并行执行时，它也是完全安全的。

另一方面，数据页具有读写(RW)的权限。这意味着线程 A 与另一个线程 B 并行处理数据页本身就是危险的。 为什么呢？这是相当直观的：它们最终可能会重创页面中的内存值。 (例如，可以想象两个线程同时写入全局链表。)。 关键的一点是，共享的可写内存必须受到保护，以防并发访问，以便始终保持数据完整性。

要真正理解为什么我们如此关注这些问题，请继续阅读。

并发性和原子性

并发执行意味着多个线程可以真正在多个 CPU 核上并行运行。 当这种情况发生在文本(代码)上时，这很好；我们可以获得更高的吞吐量。 然而，当我们在处理共享可写数据的同时并行运行时，我们就会遇到数据完整性问题，这是因为文本是只读的(并且是可执行的)，而数据是读写的。

当然，我们真正想要的是贪婪和两全其美：通过多个线程并发执行代码，但在我们必须处理共享数据时，停止并发(并行性)，只让一个线程按顺序运行数据段，直到完成，然后恢复并行执行。

教育性的银行账户示例

一个经典的(教学)例子是有问题的银行账户和软件应用。 想象一下，自由职业雕塑家 Kaloor(不用说，这里使用了虚构的名字和人物)在他的银行有一个账户；他目前的余额是 12,000.00 美元。 同时发放了两笔交易，押金 3000 美元和 8000 美元，这是他成功完成工作的报酬。如果不需要天才就能看到(假设没有其他交易)，他的账户余额很快就会反映出 23,000.00 美元。

出于本例的目的，让我们直观地看到银行软件应用是一个多线程进程；为简单起见，我们考虑派生一个线程来处理事务。 运行该软件的服务器系统是一台功能强大的多核机器--比如说，它有 12 个 CPU 核心。 当然，这意味着线程可以同时在不同的内核上并行运行。

因此，让我们想象一下，对于 Kaloor 的每个事务，我们都有一个线程运行来执行它-线程 A 和线程 B。线程 A(运行在比方说 CPU#0 上)处理第一笔 3,000 美元的存款，线程 B(运行在比方说 CPU#1 上)处理(几乎立即)第二笔 8,000 美元的存款。

这里我们考虑两个案例：

• 这种情况是偶然的，可能是交易成功完成的。 下图清楚地显示了这种情况：

Figure 1: The bank account; correct, by chance

• 再一次偶然发现，这些交易并没有成功完成。 下图显示了这种情况：

Figure 2: The bank account; incorrect, by chance

问题区域在前面的表格中突出显示：很明显，线程 B 对余额执行了无效读取-它读取了 12,000 美元(截至时间T4的值)的陈旧价值，而不是获取 15,000 美元的实际现值-导致可怜的 Kaloor 实际上损失了 3,000 美元。

怎么会出这事？ 简而言之，种族状况不佳导致了这个问题。 要了解这场比赛，请仔细查看前面的表格，并将活动形象化：

• 代表账户当前余额的变量；余额为全局变量：
  • 它驻留在数据段中
  • 它由进程的所有线程共享
• 在时间 t3，CPU#0 上的线程 A：交了$3,000；balance仍然是$12,000(尚未更新)
• 在时间 t4，CPU#1 上的线程 B**：预存 8,000 美元；余额仍为 12,000 美元(尚未更新)** *** 在时间 t5：
  • CPU#0 上的线程 A：更新余额
  • 同时，但在另一个核心上：
    • CPU#1 上的线程 B：更新余额
    • 偶然情况下，如果线程 B 在 CPU#1 上运行几微秒后，CPU#0 上的线程 A 可以更新 BALANCE 变量会怎么样？
    • 然后，线程 B 将余额读取为$12,000($3,000！)，这被称为脏读和，这是问题的核心。 这种情况被称为比赛；比赛是一种结果不确定和不可预测的情况。在大多数情况下，这将是一个问题(就像这里一样)；在少数无关紧要的情况下，它被称为良性比赛。**

**需要强调的事实是，存放资金和更新余额(或者反过来，提取资金和更新余额)的操作必须保证是原子的。 他们不能比赛，因为那会是一个缺陷(Bug)。

在软件编程上下文中，短语原子操作(或原子性)意味着操作一旦开始，将不间断地运行到完成。

临界截面

我们该如何安排前一场比赛呢？ 这很简单，真的：我们必须确保，如前所述，银行业务-存款、取款等-保证做两件事：

• 成为在该时间点运行代码的唯一线程
• 原子运行至完成，不间断

一旦实现了这一点，共享数据就不会被破坏。 必须以上述方式运行的代码段称为关键代码段。

在我们虚构的银行应用中，运行代码以执行银行操作(存款或取款)的线程必须在关键部分执行此操作，如下所示：

Figure 3: The critical section

因此，现在，让我们假设银行应用已被更正，以考虑到这些事实；线程 A 和线程 B 的垂直时间线执行路径现在如下所示：

Fig 4: Correct banking application—critical section

在这里，线程 A 和线程 B 一旦开始它们的(存放)操作，就会单独运行并完成(没有中断)；因此，它们是顺序的和原子的。

总结一下：

• 一个非常关键的部分是必须执行以下操作的代码：
  • 在不受进程中其他线程干扰的情况下运行(因为它使用某些共享资源，如全局数据)
  • 自动运行(完成，无中断)
• 如果临界区的代码可以与其他线程并行运行，这就是一个缺陷(Bug)，称为资源竞争
• 为了防止竞争，我们必须保证临界区的代码单独和原子地运行
• 要做到这一点，我们必须首先同步关键部分

现在，问题是：我们如何才能同步一个关键的部分？ 继续读下去。

锁定概念

软件中有几种形式的同步；其中一种常见的形式，实际上也是我们将会用到的一种形式，称为锁定。 在编程术语中，正如应用开发人员所见，锁最终是实例化为变量的数据结构。

当需要临界区时，只需将临界区的代码封装在解锁命令和相应的解锁命令操作之间即可。 (目前，不要担心代码级 API 的细节；我们将在稍后讨论这一点。 在这里，我们只关注正确的概念。)

让我们使用图(图 3 中前面的*的超集)来表示关键部分，以及同步机制-一个锁(的超集)：*

Fig 5: Critical section with locking

锁仓的基本前提是：

• 在任何给定的时间点，只有一个线程可以持有或拥有锁；该线程是锁的所有者。
• 在解锁时，当多个线程试图获取或获取锁时，内核将保证只有一个线程将获得锁。
• 获得锁的线程称为赢家(或锁所有者)；试图获得锁但未获得锁的线程称为失败者。

因此，想象一下：假设我们有三个线程，A、B 和 C，它们在不同的 CPU 核心上并行运行，所有线程都试图获取一个锁。 锁的保证是只有一个线程获得它-比方说线程 C 赢了，获得了锁(因此线程 C 是锁的赢家或所有者)；线程 A 和 B 是输家。 在那之后会发生什么？

• Winner 线程将锁定操作视为非阻塞调用；它继续进入关键部分(可能正在处理某些共享的可写资源，如全局数据)。
• 失败的线程将锁定操作视为阻塞调用；它们现在阻塞(等待)，但具体是基于什么呢？ (回想一下，阻塞调用是我们等待事件发生并在事件发生后解除阻塞的调用。)。 嗯，当然是解锁和操作！
• 胜利者线程在(原子地)完成临界区后，执行解锁操作。
• 线程 A 或线程 B 现在将获得锁，并且整个序列重复。

用一种更一般的方式，我们现在可以理解为：如果 N 个线程在竞争一个锁，那么锁操作(由操作系统)的保证是正好有一个线程-赢家-将获得锁。因此，我们将有一个赢家和 N-1 个输家。 胜利者线程进入临界区的代码；在此期间，所有 N-1 个失败者线程等待(阻塞)解锁操作。 在未来的某个时候(希望很快)，赢家会执行解锁；这将再次触发整个序列：前 N-1 个输家再次争夺锁；我们将有一个赢家，N-2 个输家；赢家的线程进入第二个关键部分的代码。 在此期间，所有 N-2 个失败者线程都会在解锁操作后等待(阻塞)，直到所有失败者线程都成为赢家并因此运行了临界区的代码。

它是原子的吗？

前面关于原子执行临界区的必要性的讨论可能会让您(程序员)感到担忧：也许您在想，如何识别临界区？ 这很简单：如果您具有并行性(多个线程可以并行运行代码路径)的潜力，并且代码路径正在处理某些共享资源(通常是全局或静态数据)，那么您就有了一个临界区，这意味着您将通过锁定来保护它。

A quick thumb rule: in the majority of cases, multiple threads will be running through code paths. Thus, in a general sense, the mere presence of some writable shared resource of any sort—a global, a static, an IPC shared-memory region, (even) a data item representing a hardware register in a device driver— makes the code path into a critical section. The rule is this: just protect it.

我们在上一节中看到的虚构的银行账户示例充分表明，我们有一个需要保护的关键部分(通过锁定)。 然而，我们确实遇到过这样的情况：我们是否真的需要锁定可能不那么明显。 举个例子：我们在一个多线程 C 应用中有一个全局整数g；但是在某个时刻，我们会递增它的值，比如：g ++;

这看起来很简单，但是等等！ 它是一种可写的共享资源--全局数据；多线程可能会并行运行这段代码，从而使其成为需要保护(通过锁)的临界区。 或者不是？

从表面上看，一个简单的递增(或递减)操作可能看起来是原子的(回想一下，原子在没有中断的情况下运行到完成)本身，因此不需要通过锁或任何其他形式的同步进行特殊保护。 但事实真的是这样吗？

在我们进一步讨论之前，还有另一个关键事实需要了解，那就是，在现代微处理器上，唯一可以保证在单一机器语言指令中是原子的东西。 在每条机器指令完成后，CPU 上的控制单元检查它是否必须服务于其他任何事情，通常是硬件中断或(软件)异常条件；如果是，它将程序计数器(IP 或 PC)设置为该地址并分支；如果不是，执行继续，PC 寄存器适当递增。

所以，仔细想想这一点：增量操作g++是否原子，确实取决于两个因素：

• 正在使用的微处理器的指令集体系结构(ISA)(简而言之，取决于 CPU 本身)
• 该处理器的 C 编译器如何生成代码

如果编译器为g++和 C 代码生成一条单一的机器语言指令，那么执行将确实是原子的。但它会是原子的吗？让我们来看看！！(经验性的重要性-实验，尝试-是一个关键特性；我们的第 19 章，故障排除和最佳实践涵盖了这些要点的更多内容)。

有一个非常有趣的网站，名为https://godbolt.org(后面会有截图)，和让人可以看到各种编译器如何编译给定的高级语言代码(在撰写本书时，它支持包括 C 和 C++在内的 14 种语言，当然还有各种编译器，包括 GCC(1)和 clang(1))。 有趣的是，随着 Language 下拉菜单设置为 C++，用户也可以通过 GCC for ARM 进行编译！)

让我们首先访问此网站，然后执行以下操作：

1. 通过下拉菜单选择 C 作为语言
2. 在右窗格中，将编译器选择为 x86_64 GCC 8.2
3. 在左窗格中，键入以下程序：

int g=41;
int main(void)
{
    g ++;
}

以下是输出：

Figure 6: g++ increment via gcc 8.2 on x86_64, no optimization

看看右边的窗格--可以看到编译器生成的汇编语言(当然，这些汇编语言随后会变成与处理器 ISA 相对应的机器码)。 所以?。 请注意，g++C 高级语言语句在其左窗格中以淡黄色突出显示；在右窗口中使用相同的颜色突出显示相应的程序集。 人们注意到了什么？C 代码的一行g++；变成了四条汇编语言指令。因此，根据我们前面的学习，这段代码本身不能被认为是原子的(但我们当然可以通过使用锁来强制它成为原子代码)。

下一个实验：保持一切不变，但请注意，在右窗格中有一个文本小部件，您可以在其中键入要传递给编译器的选项开关；我们键入-O2，这意味着我们希望编译器使用优化级别 2(相当高的优化级别)。 现在，对于输出：

Figure 7: g++ increment via gcc 8.2 on x86_64, optimization level 2

g++C 代码现在归结为只有一条汇编指令，因此确实变成了原子指令。

有了 ARM 编译器，而且没有优化，g++就可以转换成几行装配线-清楚地、非原子的：

Fig 8: g++ increment via gcc 7.2.1 on ARM, no optimization

我们的结论是什么？ 对于应用来说，我们编写的代码保持跨(CPU)体系结构的可移植性通常很重要。 在前面的示例中，我们清楚地发现，编译器为简单的g++和操作生成的代码有时是原子的，有时不是。 (这将取决于几个因素：CPU 的 ISA、编译器和编译时的优化级别-On，等等。)。 因此，人们可以得出的唯一安全的结论是：要安全，无论哪里有危险的部分，都要保护它到(用锁或其他手段)。

脏读数

许多刚接触这些主题的程序员都会做出致命的假设，他们会这样想：好的，我知道在修改共享资源(如全局数据结构)时，我需要将代码视为临界区，并使用锁定来保护它，但是，我的代码只在全局链表上迭代；它只对其进行读操作，而从不对其进行写入，因此，这不是临界区，不需要保护(我甚至会获得高性能的布朗尼分数)。

请把泡泡吹破！ 这是一个关键的部分。为什么？ 想象一下：当您的代码在全局链表上迭代(只读取它)时，恰恰是因为您没有以其他方式获取锁或同步，所以当您读取数据结构时，另一个写线程很可能正在向数据结构写入数据。我想一想：这是导致灾难的秘诀；您的代码完全有可能最终读取陈旧的或写了一半的不一致数据。 这称为脏读，当您不保护临界区时可能会发生这种情况。 事实上，这正是我们虚构的银行应用示例中的缺陷。

我们再次(再次)强调这些事实：

• 如果代码正在访问任何类型的可写共享资源，并且存在并行性的可能性，那么它就是一个关键部分。 保护好它。
• 这样做的一些副作用包括：
  • 如果您的代码确实具有并行性，但只在局部变量上工作，那么就不会有问题，也不是关键部分。 (请记住：每个线程都有自己的私有堆栈，因此可以在没有显式保护的情况下使用局部变量。)
  • 如果全局变量被标记为 1const，那么它当然是好的-在任何情况下它都是只读的。

(不过请注意，C 中的 const 关键字实际上并不保证该值确实是常量(正如人们通常所理解的那样)！ 这只是意味着变量是只读的，但如果另一个指针可以使用宏从下面访问它，那么它引用的数据仍然可以更改)。

正确使用锁有一个学习曲线，可能比其他编程构造要陡峭一些；这是因为，人们必须首先学习识别临界区，因此需要使用锁(在上一节中介绍)，然后学习和使用好的设计锁准则，第三，了解并避免讨厌的死锁！

锁定指南

在本节中，我们将为开发人员提供一组很小但很重要的启发式或指导原则，供开发人员在设计和实现使用锁的多线程代码时牢记在心。 这些可能在特定情况下适用，也可能不适用；有了经验，一个人会学会在适当的时间应用正确的指导方针。

不用再多说了，下面就是它们：

• 保持足够精细的锁定粒度：锁定数据，而不是代码。
• 简单是关键：涉及多个锁和线程的复杂锁定场景不仅会导致性能问题(极端情况是死锁)，还会导致其他缺陷。 保持设计尽可能简单总是很好的做法。
• 防止饥饿：长时间持有锁会导致失败者线程饥饿；必须进行设计-甚至是测试-以确保作为经验法则，每个关键部分(解锁操作和解锁操作之间的代码)尽快完成。 良好的设计可以确保代码的关键部分不会花费太长时间；将超时与锁结合使用是缓解此问题的一种方法(稍后将对此进行更多介绍)。
• 了解锁定会造成瓶颈，这一点非常重要。下面是锁定的物理类比：
  • 漏斗：把漏斗的茎想象成关键部分--它的宽度一次只能让一根线通过(最后的赢家)；输家的线仍然堵在漏斗的嘴里。
  • 在一条多车道繁忙的高速公路上，一个单独的收费站

因此，避免过长的临界区是关键：

• 在设计中构建同步，并避免这样的诱惑：好吧，我先写代码，然后再回来看锁定。它通常不太顺利；锁定本身就是一项复杂的业务；试图推迟其正确的设计和实现只会加剧问题。

让我们更详细地研究其中的第一点。

锁定粒度

在使用应用时，假设有几个地方需要通过锁定来保护数据-换句话说，有几个关键部分：

Fig 9: Timeline with several critical sections

我们已经在时间线上用实心红色矩形显示了关键部分(据我们所知，需要同步锁定的地方)。开发人员可能会意识到，为什么不简化一下呢？ 只需在时间t1获取单个锁，并在时间t6解锁：

Figure 10: Coarse granularity locking

这将在保护所有关键部分方面发挥作用。 但这是以牺牲性能为代价的，想想看，线程每次运行前面的代码路径，都必须拿到锁，执行工作，然后解锁。 这很好，但是并行性呢？ 它实际上失败了；从t1到t6的代码现在被序列化了。 这种过度放大的 locking-of-all-critical-sections-with-one-big-fat-lock 被称为粗粒度锁定。

回想我们之前的讨论：代码(文本)从来不是问题-这里根本不需要锁定；只需锁定正在访问任何类型的可写共享数据的位置。 这些是关键部分！ 这就产生了细粒度锁定-我们仅在关键部分开始的时间点进行锁定，并在结束的时间点解锁；下图反映了这一点：

Figure 11: Fine granularity locking

正如我们前面所说的，要记住的一个好的经验法则是锁定数据，而不是锁定代码。

Is super-fine granularity locking always best? Perhaps not; locking is a complex business. Practical work has shown that, sometimes, holding a lock while even working on code (pure text—the code between the critical sections), is okay. It is a balancing act; the developer must ideally use experience and trial-and-error to judge locking granularity and efficiency, constantly testing and re-evaluating the code paths for robustness and performance as one goes along.

Straying too far in either direction might be a mistake; too coarse a locking granularity yields poor performance, but too fine a granularity can too.

僵局及其避免

所谓死锁是指相关线程不可能取得进一步进展的不受欢迎的情况。 死锁的典型症状是应用(或设备驱动程序或任何软件)似乎挂起。

常见死锁类型

思考几个典型的死锁场景会帮助读者更好地理解，回想一下，锁和的基本前提是，只有一个赢家(获得锁的线程)和 N-1 个输家。 另一个关键点是，只有优胜者和线程可以执行解锁操作-没有其他线程可以这样做。

自死锁(重锁)

知道前面提到的信息，想象一下这样的场景：有一个锁(我们只称它为 L1)，有三个线程在竞争它(让我们只称它们为线程 A、B、B 和 C 线程)；假设线程 B 是最后的赢家。这很好，但如果线程 B 在其临界区内，再次尝试获取相同的锁 L1 会发生什么？好的，想想看：L1 锁当前处于锁定状态，从而迫使线程 B 阻止该锁：L1 锁当前处于锁定状态，因此强制线程 B 阻止。但如果线程 B 在其临界区内，再次尝试获取相同的锁 L1 会发生什么情况？嗯，想想看：L1 锁当前处于锁定状态，从而迫使线程 B 阻止。 然而，除了线程 B 本身之外，没有任何线程可以执行解锁操作，所以线程 B 将以永远等待告终！ 这就是问题所在：僵局。 这种类型的死锁被称为自我死锁，或重新锁定错误。

有人可能会争辩说，确实存在这种情况，难道不能递归地获取锁吗？是的，正如我们稍后将看到的那样，这可以在 pthread 和 API 中完成。 然而，好的设计经常反对使用递归锁；实际上，Linux 内核不允许使用递归锁。

ABBA 僵局

在涉及嵌套锁定的场景中，可能会出现一种更复杂的死锁形式：两个或更多竞争线程以及两个或更多锁。 这里，让我们来看一个最简单的例子：有两个线程(A 和 B)使用两个锁(L1 和 L2)的场景。

让我们假设这是在垂直时间线上展开的，如下表所示：

| 时间 | 线程 A | 线程 B | | T1 期 | 尝试获取 L1 锁 | 尝试获取 L2 锁 | | T2 | 获取锁定 L1 | 获取锁定 L2 | | T3 | | | | T4 | 尝试获取 L2 锁 | 尝试获取 L1 锁 | | T5 | 正在解锁 L2 上的数据块 | 正在解锁 L1 上的数据块 | | | | |

很明显，每个线程都在等待另一个线程解锁它想要的锁；因此，每个线程都会永远等待，这就保证了死锁。 这种僵局通常被称为致命拥抱或 ABBA 僵局。

避免僵局

避免僵局显然是我们想要确保的事情。 除了第锁定准则部分和中涉及的要点之外，还有一个关键点，即多个锁的获取顺序很重要；始终保持一致的锁顺序将提供防止死锁的保护。

要理解其中的原因，让我们重新看看刚才介绍的 ABBA 死锁场景(请参阅上表)。 再看看表：注意线程 A 获取锁 L1，然后尝试获取锁 L2，而线程 B 执行相反的操作。 现在我们将代表这个场景，但有一个关键的警告：锁定排序！这次，我们将有一个锁定排序规则；它可能很简单：首先，取 L1 锁，然后取 L2 锁：

锁定 L1-->锁定 L2

考虑到这一锁定订单，我们发现情况可能如下所示：

| 时间 | 线程 A | 线程 B | | T1 期 | 尝试获取 L1 锁 | 尝试获取 L1 锁 | | T2 | | 获取锁定 L1 | | T3 | | | | T4 | | 锁定 L1 L1 | | T5 | 获取锁定 L1 | | | T6 | | 尝试获取 L2 锁 | | T7 | 锁定 L1 L1 | 获取锁 L2 | | T8 | 尝试获取 L2 锁 | | | T9 | | ---> | | T10 | | 锁定 L2 | | T11 | 获取锁定 L2 | | | T12 | | ..。 | | T13 | 锁定 L2 | ..。 |

这里的关键点是，两个线程都试图以给定的顺序获取锁；首先是 L1，然后是 L2。 在上表中，我们可以想象这样一种情况：线程 B 首先获得锁，强制线程 A 等待。 这完全没有问题，这是意料之中的；关键是不会出现死锁。

精确的顺序本身并不重要；重要的是设计者和开发人员能够记录要遵循的锁顺序并遵守它。

The lock ordering semantics, and indeed developer comments regarding this key point, can be often found within the source tree of the Linux kernel (ver 4.19, as of this writing). Here's one example: virt/kvm/kvm_main.c``... /*  * Ordering of locks:  *  * kvm->lock --> kvm->slots_lock --> kvm->irq_lock  */ ...

那么，回过头来看我们的第一个表，我们现在可以清楚地看到，死锁的发生是因为违反了锁的顺序规则：线程 B 先占用锁 L2，然后才占用锁 L1！

使用 pthreadAPI 进行同步

既然我们已经介绍了所需的理论背景信息，让我们继续进行实际操作：在本章的其余部分，我们将重点介绍如何使用 pthreadsAPI 执行同步，从而避免竞争。

我们了解到，要保护临界区中任何类型的可写共享数据，我们需要锁定。 PthreadsAPI 为这个用例提供了互斥锁；我们只打算在临界区的持续时间内保持锁一小段时间。

不过，在某些情况下，我们需要不同类型的同步-我们需要基于某个数据元素的值进行同步；pthreadsAPI 为此用例提供了条件变量(CV)。

让我们依次介绍这些内容。

互斥锁

单词“mutex实际上是互斥的缩写；对于所有其他(输家)线程的互斥，一个线程-赢家-持有(或拥有)互斥锁。只有当它被解锁时，另一个线程才能获得锁。

An FAQ: What really is the difference between the semaphore and the mutex lock? Firstly, the semaphore can be used in two ways—one, as a counter (with the counting semaphore object), and two (relevant to us here), essentially as a mutex lock—the binary semaphore.  Between the binary semaphore and the mutex lock, there exists two primary differences: one, the semaphore is meant to be used to synchronize between processes and not the threads internal to a single process (it is indeed a well-known IPC facility); the mutex lock is meant to synchronize between the threads of a given (single) process. (Having said that, it is possible to create a process-shared mutex, but it's never the default). Two, the SysV IPC implementation of the semaphore provides the possibility of having the kernel unlock the semaphore (via the semop(2) SEM_UNDO flag) if the owner process is abruptly killed (always possible via signal #9); no such possibility even exists for the mutex—the winner must unlock it (we shall cover how the developer can ensure this later).

让我们从一个初始化、使用和销毁互斥锁的简单示例开始。 在本程序中，我们将创建三个线程，并且仅递增三个全局整数，每个全局整数在线程的 Worker 例程中各递增一次。

For readability, only key parts of the source code are displayed; to view the complete source code, build, and run it. The entire tree is available for cloning from GitHub here: https://github.com/PacktPublishing/Hands-on-System-Programming-with-Linux. Code: ch15/mutex1.c:

static long g1=10, g2=12, g3=14;    /* our globals */
pthread_mutex_t mylock;   /* lock to protect our globals */ 

为了使用互斥锁，必须首先将其初始化为解锁状态；这可以按如下方式完成：

 if ((ret = pthread_mutex_init(&mylock, NULL)))
     FATAL("pthread_mutex_init() failed! [%d]\n", ret);

或者，我们可以将初始化作为声明执行，例如：

pthread_mutex_t mylock = PTHREAD_MUTEX_INITIALIZER;

事实上，可以为互斥锁指定几个互斥锁属性(通过 APIpthread_mutexattr_init(3))；我们将在本章后面讨论这一点。 目前，这些属性将是系统默认值。

此外，一旦完成，我们必须立即销毁互斥锁：

 if ((ret = pthread_mutex_destroy(&mylock)))
     FATAL("pthread_mutex_destroy() failed! [%d]\n", ret);

像往常一样，我们在一个循环中创建(三个)工作线程(我们不在这里显示此代码，因为它是重复的)。 下面是线程的辅助例程：

void * worker(void *data)
{
     long datum = (long)data + 1;
     if (locking)
         pthread_mutex_lock(&mylock);

     /*--- Critical Section begins */
 g1 ++; g2 ++; g3 ++;
     printf("[Thread #%ld] %2ld %2ld %2ld\n", datum, g1, g2, g3);
     /*--- Critical Section ends */

     if (locking)
         pthread_mutex_unlock(&mylock);

     /* Terminate with success: status value 0.
      * The join will pick this up. */
     pthread_exit((void *)0);
}

因为我们正在处理的每个线程的数据都是完全可写的共享(它在数据段中！)。 资源，我们认识到这是一个关键的部分！

因此，我们必须保护它-在这里，我们使用互斥锁来保护它。 因此，在进入临界区之前，我们首先获取互斥锁，然后处理全局数据，然后解锁，从而使操作不受竞争的影响。 (请注意，在前面的代码中，我们只在名为locking的变量为真的情况下执行锁定和解锁；这是一种有意测试代码的方式。 当然，在生产中，请取消 If 条件，只执行锁定！)。 细心的读者还会注意到，我们将关键部分保持得相当简短-它只封装了全局更新和随后的printf(3)，没有其他内容。 (这对于良好的性能很重要；回想一下我们在前面关于锁定粒度的内容。)

正如前面提到的，我们故意向用户提供一个选项，以避免完全使用锁定-这当然会，或者更确切地说，这可能会导致错误行为。 让我们试试看：

$ ./mutex1 
Usage: ./mutex1 lock-or-not
 0 : do Not lock (buggy!)
 1 : do lock (correct)
$ ./mutex1 1
At start:   g1 g2 g3
            10 12 14
[Thread #1] 11 13 15
[Thread #2] 12 14 16
[Thread #3] 13 15 17
$ 

它确实能像预期的那样工作。 即使我们将参数传递为零(从而关闭锁定)，程序似乎(通常)工作正常：

$ ./mutex1 0
At start:   g1 g2 g3
            10 12 14
[Thread #1] 11 13 15
[Thread #2] 12 14 16
[Thread #3] 13 15 17
$ 

为什么？ 啊，理解这一点很重要：回想一下我们在前面部分学到的东西：它是原子的吗？ 通过将简单的整数增量和编译器优化设置为较高级别(实际上，此处为-O2)，很可能整数增量是原子级的，因此实际上不需要锁定。 然而，情况可能并不总是这样，特别是当我们做一些比仅仅对整数变量递增或递减更复杂的事情时。 (考虑读/写大型全局链表，等等)！ 底线：我们必须始终认识到关键部分，并确保我们保护它们。

观看比赛

为了准确演示这个问题(实际看到的是数据竞赛)，我们将编写另一个演示程序。 在这个例子中，我们将计算给定数字的阶乘系数(快速提示：3！=3x2x1=6；回想一下您的学生时代-符号 N！的意思是 N 的阶乘)。 相关代码如下：

For readability, only key parts of the source code are displayed; to view the complete source code, build, and run it. The entire tree is available for cloning from GitHub here: https://github.com/PacktPublishing/Hands-on-System-Programming-with-Linux. Code: ch15/facto.c:

在main()中，我们初始化互斥锁(并创建两个工作线程；我们没有显示创建、销毁线程以及互斥锁的代码)：

printf( "Locking mode : %s\n" 
        "Verbose mode : %s\n",
          (gLocking == 1?"ON":"OFF"),
          (gVerbose == 1?"ON":"OFF"));

if (gLocking) {
     if ((ret = pthread_mutex_init(&mylock, NULL)))
         FATAL("pthread_mutex_init() failed! [%d]\n", ret);
 }
...

这条线的工作人员例程如下：

void * worker(void *data)
{
    long datum = (long)data + 1;
    int N=0;
...
    if (gLocking)
        pthread_mutex_lock(&mylock);

    /*--- Critical Section begins! */
 factorize(N);
    printf("[Thread #%ld] (factorial) %d ! = %20lld\n",
      datum, N, gFactorial);
    /*--- Critical Section ends */

    if (gLocking)
        pthread_mutex_unlock(&mylock);
...

识别出临界区后，我们获取(并随后解锁)互斥锁。 *factorize函数的代码如下：

/*
 * This is the function that calculates the factorial of the given   parameter. 
Stress it, making it susceptible to the data race, by turning verbose mode On; then, it will take more time to execute, and likely end up "racing" on the value of the global gFactorial. */
static void factorize(int num)
{
     int i;
     gFactorial = 1;
     if (num <= 0)
         return;
    for (i=1; i<=num; i++) {
         gFactorial *= i;
         VPRINT(" i=%2d fact=%20lld\n", i, gFactorial);
    }
}

仔细阅读前面的评论；它是本演示的关键。 让我们试试看：

$ ./facto 
Usage: ./facto lock-or-not [verbose=[0]|1]
Locking mode:
 0 : do Not lock (buggy!)
 1 : do lock (correct)
(TIP: turn locking OFF and verbose mode ON to see the issue!)
$ ./facto 1
Locking mode : ON
Verbose mode : OFF
[Thread #2] (factorial) 12 ! =     479001600
[Thread #1] (factorial) 10 ! =       3628800
$ 

结果是正确的(自己验证这一点)。 现在，我们在锁定和详细模式打开的情况下重新运行它：

$ ./facto 0 1
Locking mode : OFF
Verbose mode : ON
facto.c:factorize:50: i= 1 fact=                 1
facto.c:factorize:50: i= 2 fact=                 2
facto.c:factorize:50: i= 3 fact=                 6
facto.c:factorize:50: i= 4 fact=                24
facto.c:factorize:50: i= 5 fact=               120
facto.c:factorize:50: i= 6 fact=               720
facto.c:factorize:50: i= 7 fact=              5040
facto.c:factorize:50: i= 8 fact=             40320
facto.c:factorize:50: i= 9 fact=            362880
facto.c:factorize:50: i=10 fact=            3628800
[Thread #1] (factorial) 10 ! =           3628800
facto.c:factorize:50: i= 1 fact=                        1
facto.c:factorize:50: i= 2 fact=         7257600  *<-- Dirty Read!*
facto.c:factorize:50: i= 3 fact=                 21772800
facto.c:factorize:50: i= 4 fact=                 87091200
facto.c:factorize:50: i= 5 fact=                435456000
facto.c:factorize:50: i= 6 fact=               2612736000
facto.c:factorize:50: i= 7 fact=              18289152000
facto.c:factorize:50: i= 8 fact=             146313216000
facto.c:factorize:50: i= 9 fact=            1316818944000
facto.c:factorize:50: i=10 fact=           13168189440000
facto.c:factorize:50: i=11 fact=          144850083840000
facto.c:factorize:50: i=12 fact=         1738201006080000
[Thread #2] (factorial) 12 ! =        1738201006080000
$ 

啊哈！ 在这种情况下，10!可以工作，但12!是错误的！ 我们可以从前面的输出中确切地看到，已经发生了脏读取(在 12！的计算的 i==2 次迭代中)，从而导致了缺陷。 嗯，当然：我们没有保护这里的临界区(锁定被关闭)；真的难怪它出了问题。

我们要再次强调的是，这些竞赛是微妙的计时巧合；在有缺陷的实现中，您的测试用例仍然可能成功，但当然这不能保证任何事情(正如墨菲定律告诉我们的那样，它很可能在该领域失败！)。(一个不幸的事实是，测试可以发现错误的存在，但不能揭示它们的存在。最重要的是，第 19 章，故障排除和最佳实践涵盖了这些要点)。

The reader will realize that, as these data races are delicate timing coincidences, they may or may not occur exactly as shown here on your test systems. Retrying the app a few times may help reproduce these scenarios.

我们让读者在锁定模式打开和详细模式打开的情况下试用用例；当然，它应该可以工作。

互斥锁属性

互斥锁可以有几个与之关联的属性。 此外，我们列举了其中的几个。

互斥类型

互斥锁可以是四种类型中的一种，默认情况下通常(但并非总是)是普通互斥锁(取决于实现)。 使用的互斥类型会影响锁定和解锁的行为。 这些类型包括：PTHREAD_MUTEX_NORMAL、PTHREAD_MUTEX_ERRORCHECK、PTHREAD_MUTEX_RECSIVE 和 PTHREAD_MUTEX_DEFAULT。

pthread_mutex_lock(3)上的系统手册页使用一个表描述了取决于互斥类型的行为；为了方便读者，我们在这里复制了相同的内容。

如果线程试图重新锁定它已经锁定的互斥体，pthread_mutex_lock(3)的行为应如下表的重新锁定列中所述。 如果线程尝试解锁未锁定的互斥体或解锁的互斥体，则pthread_mutex_unlock(3)应按照下表的Unlock When Not Owner列中的说明进行操作：

如果互斥体类型为 PTHREAD_MUTEX_DEFAULT，则pthread_mutex_lock(3)的行为可能对应于其他三种标准互斥体类型之一，如上表所述。 如果它不符合这三种情况中的一种，则对于标记为†的情况，行为也是未定义的。

重新锁定一栏直接对应于我们在本章前面描述的自死锁场景，例如，尝试重新锁定已经锁定的锁会有什么效果(也许是诗意的措辞？)。 会有的。 显然，除了递归测试和错误检查互斥锁的情况外，最终结果要么是未定义的，要么是没有定义的(这意味着任何事情都可能发生！)。 或者真的陷入僵局。

同样，尝试由除所有者以外的任何线程解锁互斥锁要么会导致未定义的行为，要么会导致错误。

有人可能会问：为什么锁定 API 的行为会因互斥类型的不同而不同--在错误返回或失败方面？ 为什么不为所有类型设置一个标准行为，从而简化情况呢？ 这是简单性和性能之间通常的权衡：例如，它的实现方式允许编写良好、经过编程验证的正确实时嵌入式应用放弃额外的错误检查，从而提高速度(这在关键代码路径上尤其重要)。 另一方面，在开发或调试环境中，开发人员可能会选择允许额外检查以在发布之前捕获缺陷。 (pthread_mutex_destroy(3)上的手册页有一个标题为在错误检查和支持的性能之间的权衡的章节，更详细地描述了这一方面。)

互斥锁的 type 属性(上表中的第一列)的get和set对 API 非常简单：

include <pthread.h>
int pthread_mutexattr_gettype(const pthread_mutexattr_t *restrict attr,     int *restrict type);
int pthread_mutexattr_settype(pthread_mutexattr_t *attr, int type);

健壮的互斥属性

看一眼前面的表格，你会发现健壮性这一列；它是什么意思？ 回想一下，只有互斥锁的所有者线程才有可能解锁互斥锁；现在，我们问，如果某个偶然的机会，所有者线程死了怎么办？ (首先，好的设计将确保这种情况永远不会发生；其次，即使发生这种情况，也有防止线程取消的方法，这是我们将在下一章讨论的主题。)。 从表面上看，没有任何帮助；等待锁的任何其他线程现在都会死锁(实际上，它们只会挂起)。 此行为实际上是默认行为；它也是由名为“PTHREAD_MUTEX_STALTED”的健壮属性设置的行为。 为了在这种情况下进行(可能的)救援，不存在另一个健壮的互斥体属性的值：*PTHREAD_MUTEX_ROBLE。 用户始终可以通过以下两对 API 在互斥体上查询和设置这些属性：

#include <pthread.h>
int pthread_mutexattr_getrobust(const pthread_mutexattr_t *attr,
    int *robustness);
int pthread_mutexattr_setrobust(const pthread_mutexattr_t *attr,
    int robustness);

如果在互斥锁上设置了此属性(值为 PTHREAD_MUTEX_ROBLY)，则如果拥有者线程在持有互斥锁时死亡，则在锁上执行的后续操作pthread_mutex_lock(3)将成功，并返回值为EOWNERDEAD。 不过，坚持住！ 尽管调用返回(所谓的)成功返回，但重要的是要理解，问题锁现在被认为处于不一致状态，必须通过pthread_mutex_consistent(3)命令 API 重置为一致状态：

int pthread_mutex_consistent(pthread_mutex_t *mutex);

这里返回值为零表示成功；互斥体现在回到非常一致(稳定)的状态，可以正常使用(使用它，在某个时候，当然必须解锁它)。

要总结这一点，要使用更健壮的属性互斥锁，请使用以下命令：

• 初始化互斥锁： pthread_mutexattr_t attr； pthread_mutexattr_init(&attr)；
  • 设置健壮属性：pthread_mutexattr_setrobust(&attr, PTHREAD_MUTEX_ROBUST)；
• 所有者线程
  • 锁定：pthread_mutex_lock(&mylock)
  • 现在，假设线程所有者突然死亡(同时持有互斥锁)
• 另一个线程(可能是 Main)可以承担所有权：
  • 首先，侦破案件：
    • ret = pthread_mutex_lock(&mylock); if (ret == EOWNERDEAD) {
  • 然后，使其保持一致： 和pthread_mutex_consistent(&mylock)
  • 使用它(或者干脆解锁)
  • 解锁：pthread_mutex_unlock(&mylock)

我们没有重复轮子，而是将读者引向一个简单的、可读的示例，该示例使用了前面描述的强大的互斥属性特性。 请在pthread_mutexattr_setrobust(3)的手册页中找到它。

Under the hood, the Linux pthreads mutex lock is implemented via the futex(2) system call (and thus by the OS). The futex (fast user mutex) provides a fast, robust, atomic-only instructions locking implementation. Links with more details can be found in the *Further reading *section on the GitHub repository.

IPC、线程和进程共享的互斥锁

想象一个由几个独立的多线程进程组成的大型应用。 现在，如果这些进程想要彼此通信(它们通常会想要这样做)，那么如何才能实现这一点呢？ 当然，答案是进程间通信(IPC)-为此目的而存在的机制。 一般而言，典型的 Unix/Linux 平台上有几种 IPC 机制；这些机制包括共享内存(以及mmap(2))、消息队列、信号量(通常用于同步)、命名管道(FIFO)和未命名管道、套接字(Unix 和 Internet 域)，以及某种程度上的信号。

Unfortunately, due to space constraints, we do not cover process IPC mechanisms in this book; we urge the interested reader to look into the links (and books) provided on IPC in the *Further reading *section on the GitHub repository.

这里要强调的是，所有这些 IPC 机制都是为独立于 VM 的进程之间的通信而设计的。因此，我们这里讨论的重点是多线程，给定进程内的所有线程是如何相互通信的？ 这真的很简单：就像人们可以设置和使用共享内存区域来有效和高效地在进程之间通信(写入和读取该区域，通过信号量同步访问)一样，线程可以简单而有效地使用全局内存缓冲区(或任何适当的数据结构)作为彼此通信的媒介，当然，还可以通过互斥锁同步对全局内存区域的访问。

有趣的是，可以使用互斥锁作为属于不同进程的线程之间的同步原语。这是通过设置称为 pShared 或进程共享的互斥锁属性来实现的。用于获取和设置进程共享互斥锁属性的 API 对如下所示：

int pthread_mutexattr_getpshared(const pthread_mutexattr_t *attr,
    int *pshared);
int pthread_mutexattr_setpshared(pthread_mutexattr_t *attr,
    int pshared);

第二个参数pshared可以设置为以下值之一：

• PTHREAD_PROCESS_PRIVATE：默认设置；在这里，互斥锁只对创建互斥锁的进程内的线程可见。
• PTHREAD_PROCESS_SHARED：在这里，互斥锁对任何可以访问创建互斥锁的内存区域的线程都是可见的，包括不同进程的线程。

但是，如何真正确保互斥锁所在的内存区域在进程之间共享(如果没有该内存区域，相关进程就不可能使用互斥锁)呢？ 好了，这真的回到了基础：我们必须利用我们提到的 IPC 机制之一-共享内存，事实证明它是正确的使用机制。因此，我们让应用设置了一个共享内存区域(通过传统的 SysV IPCshmget(2))或较新的 POSIX IPCshm_open(2)系统调用)，并在这个共享内存中实例化了我们的进程共享互斥锁。

因此，让我们用一个简单的应用将所有这些联系在一起：我们将编写一个创建两个共享内存区域的应用：

• 其一，一个较小的共享内存区，用作进程共享互斥体锁和只有一次的初始化控件的共享空间(马上详细介绍)
• 第二，共享内存区作为存储 IPC 消息的简单缓冲区

我们将使用进程共享属性初始化一个互斥锁，这样它就不能在不同进程的线程之间使用来同步访问；在这里，我们派生并让原始父进程和新生的子进程的一个线程竞争互斥锁。 一旦它们(按顺序)获得它，它们就会将一条消息写入第二个共享内存段。 在应用结束时，我们销毁资源并显示共享内存和缓冲区(作为简单的概念验证)。

让我们试用一下我们的应用(ch15/pshared_mutex_demo.c)：

We have added some blank lines in the following code for readability.

$ ./pshared_mutex_demo 
./pshared_mutex_demo:15317: shmem segment successfully created / accessed. ID=38928405
./pshared_mutex_demo:15317: Attached successfully to shmem segment at 0x7f45e9d50000
./pshared_mutex_demo:15317: shmem segment successfully created / accessed. ID=38961174
./pshared_mutex_demo:15317: Attached successfully to shmem segment at 0x7f45e9d4f000

[pthread_once(): calls init_mutex(): from PID 15317]

Worker thread #0 [15317] running ...
 [thrd 0]: attempting to take the shared mutex lock...
 [thrd 0]: got the (shared) lock!
#0: work done, exiting now

 Child[15319]: attempting to taking the shared mutex lock...
 Child[15319]: got the (shared) lock!

main: joining (waiting) upon thread #0 ...
Thread #0 successfully joined; it terminated with status=0

Shared Memory 'comm' buffer:
00000000 63 63 63 63 63 00 63 68 69 6c 64 20 31 35 33 31 ccccc.child 1531
00000016 39 20 68 65 72 65 21 0a 00 74 74 74 74 74 00 74 9 here!..ttttt.t
00000032 68 72 65 61 64 20 31 35 33 31 37 20 68 65 72 65 hread 15317 here
00000048 21 0a 00 00 00 00 00 00 00 00 00 00 00 00 00 00 !...............
00000064 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................
00000080 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................
00000096 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................
00000112 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................

在现实世界中，事情并没有这么简单；确实存在一个额外的同步问题需要考虑：如何确保互斥锁被正确地、原子地初始化(只由一个进程或线程初始化)，并且只初始化一次，其他线程是否应该尝试使用它？ 在我们的演示程序中，我们使用pthread_once(3)API 实现了互斥对象的保证只初始化一次(但没有忽略 have-threads-wait-and-only-use-it-once-initialized 问题)。 问题)。 (关于堆栈溢出的一个有趣的问题&A 强调了这一点；请看一下：https://stackoverflow.com/questions/42628949/using-pthread-mutex-shared-between-processes-correctly#*。)*然而，实际情况是，pthread_once(3)API 是在进程的线程之间使用的。 另外，POSIX 要求once_control函数的初始化是静态完成的；在这里，我们只能在运行时执行，所以它并不完美。

在其主要功能中，我们设置并初始化(IPC)共享内存段；我们敦促读者仔细阅读源代码(阅读所有评论)，并亲自试用：

For readability, only key parts of the source code are displayed; to view the complete source code, build, and run it. The entire tree is available for cloning from GitHub here: https://github.com/PacktPublishing/Hands-on-System-Programming-with-Linux.

...

  /* Setup a shared memory region for the process-shared mutex lock.
   * A bit of complexity due to the fact that we use the space within for:
   * a) memory for 1 process-shared mutex
   * b) 32 bytes of padding (not strictly required)
   * c) memory for 1 pthread_once_t variable.
   * We need the last one for performing guaranteed once-only
   * initialization of the mutex object.
   */
  shmaddr = shmem_setup(&gshm_id, argv[0], 0, 
              (NUM_PSMUTEX*sizeof(pthread_mutex_t) + 32 +  
                sizeof(pthread_once_t)));
  if (!shmaddr)
      FATAL("shmem setup 1 failed\n");

  /* Associate the shared memory segment with the mutex and 
   * the pthread_once_t variable. */
  shmtx = (pthread_mutex_t *)shmaddr;
  mutex_init_once = (pthread_once_t *)shmaddr +     
                      (NUM_PSMUTEX*sizeof(pthread_mutex_t)) + 32;
  *mutex_init_once = PTHREAD_ONCE_INIT; /* see below comment on pthread_once */

  /* Setup a second shared memory region to be used as a comm buffer */
  gshmbuf = shmem_setup(&gshmbuf_id, argv[0], 0, GBUFSIZE);
  if (!gshmbuf)
      FATAL("shmem setup 2 failed\n");
  memset(gshmbuf, 0, GBUFSIZE);

  /* Initialize the mutex; here, we come across a relevant issue: this
   * mutex object is already instantiated in a shared memory region that
   * other processes might well have access to. So who will initialize
   * the mutex? (it must be done only once).
   * Enter the pthread_once(3) API: it guarantees that, given a
   * 'once_control' variable (1st param), the 2nd param - a function
   * pointer, that function will be called exactly once.
   * However: the reality is that the pthread_once is meant to be used
   * between the threads of a process. Also, POSIX requires that the
   * initialization of the 'once_control' is done statically; here, we
   * have performed it at runtime...
   */
  pthread_once(mutex_init_once, init_mutex);
...

使用进程共享属性初始化互斥锁的*init_mutex和函数如下所示：

static void init_mutex(void)
{
  int ret=0;

  printf("[pthread_once(): calls %s(): from PID %d]\n",
      __func__, getpid());
  ret = pthread_mutexattr_init(&mtx_attr);
  if (ret)
      FATAL("pthread_mutexattr_init failed [%d]\n", ret);

  ret = pthread_mutexattr_setpshared(&mtx_attr, PTHREAD_PROCESS_SHARED);
  if (ret)
      FATAL("pthread_mutexattr_setpshared failed [%d]\n", ret);

  ret = pthread_mutex_init(shmtx, &mtx_attr);
  if (ret)
      FATAL("pthread_mutex_init failed [%d]\n", ret);
}

工作线程的代码-工作线程例程-如以下代码所示。 在这里，我们需要对第二个共享内存段进行操作，这当然意味着这是一个关键部分。 因此，我们使用进程共享锁，执行工作，然后解锁互斥锁：

void * worker(void *data)
{
  long datum = (long)data;
  printf("Worker thread #%ld [%d] running ...\n", datum, getpid());
  sleep(1);
  printf(" [thrd %ld]: attempting to take the shared mutex lock...\n", datum);

  LOCK_MTX(shmtx);
  /*--- critical section begins */
  printf(" [thrd %ld]: got the (shared) lock!\n", datum);
  /* Lets write into the shmem buffer; first, a 5-byte 'signature',
     followed by a message. */
  memset(&gshmbuf[0]+25, 't', 5);
  snprintf(&gshmbuf[0]+31, 32, "thread %d here!\n", getpid());
  /*--- critical section ends */
  UNLOCK_MTX(shmtx);

  printf("#%ld: work done, exiting now\n", datum);
  pthread_exit(NULL);
}

请注意，锁定和解锁操作是由宏执行的；它们如下所示：

#define LOCK_MTX(mtx) do {                           \
  int ret=0;                                         \
  if ((ret = pthread_mutex_lock(mtx)))               \
    FATAL("pthread_mutex_lock failed! [%d]\n", ret); \
} while(0)

#define UNLOCK_MTX(mtx) do {                           \
  int ret=0;                                           \
  if ((ret = pthread_mutex_unlock(mtx)))               \
    FATAL("pthread_mutex_unlock failed! [%d]\n", ret); \
} while(0)

我们把它留给读者来查看我们可以分叉的代码，并让新生的子进程基本上做与前面的工作线程相同的事情-在(相同的)第二个共享内存段上操作；作为一个临界区，它也试图获取进程共享的线程锁，一旦获得它，就执行工作，随后解锁互斥锁。

Unless there is some compelling reason not to do so, when setting up IPC between processes, we suggest that you use one (or some) of the numerous IPC mechanisms that have been explicitly designed for this very purpose. Using the process-shared mutex as a synchronization mechanism between the threads of two or more processes is possible, but ask yourself if it is really required.

Having said that, there are some advantages to using a mutex over the traditional (binary) semaphore object; these include the fact that the mutex is always associated with an owner thread, and only the owner can operate upon it (preventing some illegal or defective scenarios), and that mutexes can be set up to use nested (recursive) locking, and deal with the priority inversion problem effectively (via the inheritance protocol and/or priority ceiling attributes).

优先级反转、看门狗和火星

实时操作系统(RTOS)上通常运行着时间关键型多线程应用。 非常简单但仍然正确的是，RTOS 调度程序决定下一个运行哪个线程的主要规则是，优先级最高的可运行线程必须是正在运行的线程。(顺便说一句，我们将在章，Linux 上的 CPU 调度中讨论与 Linux OS 相关的 CPU 调度；现在不必担心细节。)

优先级反转

让我们来设想一个包含三个线程的应用；其中一个是高优先级线程(让我们称其为优先级为 90 的线程 A)，另一个是低优先级线程(让我们称其为优先级为 10 的线程 B)，最后是一个中等优先级线程 C.(SCHED_FIFO 调度策略的优先级范围是 1 到 99，99 是可能的最高优先级；在后面的章节中将详细介绍这一点)。 因此，我们可以想象在一个进程中有这三个优先级不同的线程：

• 线程 A：高优先级，90
• 线程 B：低优先级，10
• 线程 C：中等优先级，45

此外，让我们考虑一下我们有一些共享资源 X，这是线程 A 和 B 梦寐以求的；当然，这构成了一个关键的数据部分，因此，我们需要同步对它的访问以确保正确性。 我们将使用一个新的互斥锁工具来做到这一点。

正常情况下可能是这样工作的(让我们暂时忽略线程 C)：线程 B 在 CPU 上运行一些代码；线程 A 在另一个 CPU 内核上运行其他代码。 这两个线程都不在关键线程段；因此，互斥锁处于解锁状态。

现在，线程 B(在时间T1)命中临界区的代码并获得互斥锁，从而成为线程的所有者。它现在运行临界区内的代码(在 X 上工作)。 同时，如果线程 A 在时间t2也碰巧命中关键线程部分，从而试图获取互斥锁，该怎么办？ 嗯，我们知道它已经被锁定了，因此线程 A 将不得不等待(阻塞)线程 B 将要(希望很快)执行的解锁操作。一旦线程 B 解锁互斥锁(在时间T3)，线程 A 就会获取它(在时间T4；我们认为T4-T3的延迟非常小)，生活(非常愉快地)继续。 这看起来很好：

Fig 12: Mutex locking: the normal good case

然而，一个潜在的坏情况也存在！ 继续读下去。

看门狗定时器简介

看门狗是一种机制，用于定期检测系统是否处于健康状态，如果认为不是，则重新启动系统。 这是通过设置(内核)计时器(也就是说，60 秒超时)来实现的。 如果一切正常，看门狗守护进程(守护进程只是一个系统后台进程)将始终如一地取消计时器(当然，在计时器过期之前)，然后重新启用它；这称为抚摸狗。 如果守护进程没有(由于出现严重错误)，看门狗会感到恼火，并重新启动系统！ 纯软件看门狗实现将不受内核错误和故障的保护；硬件看门狗(锁存到板重置电路中)将始终能够在需要时重新启动系统。

通常，嵌入式应用的高优先级线程被设计为具有非常真实的最后期限，它们必须在此期限内完成一些工作；否则，系统将被视为失败。 有人想知道，如果在运行时操作系统本身-由于一个不幸的错误-简单地崩溃或挂起(死机)怎么办？ 这样，应用线程就无法继续运行；我们需要一种方法来检测并摆脱这种混乱。 嵌入式设计者通常利用看门狗定时器(WDT)和硬件电路(以及相关的设备驱动程序)来精确地实现这一点。 如果系统或关键线程未达到其截止日期(未能抚摸狗)，则系统将重新启动。

那么，回到我们的场景。 假设我们的高优先级线程 A 的截止时间是 100ms；在头脑中重复前面的锁定场景，但要有这个不同之处(也请参阅图 13：)：

• 线程 B(低优先级线程)，在时间t1获得互斥锁。
• 线程 A还在时间t2请求互斥锁(但必须等待线程 B 解锁)。
• 在线程 B 可以完成临界区之前，另一个中等优先级的线程 C#(运行在同一个 CPU 内核上，优先级为 45)被唤醒！ 它将立即抢占线程 B，因为它的优先级更高(回想一下，优先级最高的可运行线程必须是正在运行的线程)。
• 现在，在线程 C 离开 CPU 之前，线程 B 无法完成临界区，因此无法执行解锁。
• 这反过来又会显著延迟线程 A，线程 A 在线程 B 即将进行的解锁时被阻塞：
  • 但是，线程 B 已被线程 C 抢占，因此无法执行解锁。
• 如果解锁时间超过线程 A 的截止日期 A(时间T4)怎么办？
  • 则看门狗计时器将到期，强制系统重新启动：

Fig 13: Priority inversion

有趣而不幸的是，您是否注意到最高优先级线程(A)实际上被迫等待系统上最低优先级线程(B)？ 这种现象实际上是一种有据可查的软件风险评估，正式名称为优先级反转。

不仅如此，考虑如果几个中等优先级的线程在线程 B 处于其临界区(从而持有锁)时被唤醒，会发生什么情况？ 线程 A 的潜在等待时间现在可能会变得非常长；这种情况称为无界优先级反转。

火星探路者任务简介

非常有趣的是，这种精确的场景优先级倒置在一个完全不同的世界中上演了相当戏剧性的一幕：在火星表面！ 美国国家航空航天局于 1997 年 7 月 4 日成功将一艘机器人航天器(探路者着陆器)降落在火星表面；然后开始卸货并在火星表面部署一个较小的机器人--美国旅居者漫游者号。 然而，控制器发现着陆器遇到了问题-它经常会重新启动。 对实时遥测馈送的详细分析最终揭示了根本问题--是软件遇到了优先级反转问题！ 值得称赞的是，NASA 的喷气推进实验室(JPL)团队与 Wind River 的工程师(该公司向 NASA 提供了定制的 VxWorks RTOS)从地球上诊断并调试了情况，确定了根本原因缺陷是一个优先反转问题，修复了它，并将新固件上传到月球车上，一切都奏效了：

Figure 14: Photo from the Mars Pathfinder Lander

当微软工程师迈克·琼斯(Mike Jones)在 IEEE 实时研讨会上写了一封有趣的电子邮件，讲述了 NASA 的探路者任务发生了什么时，这一消息(以病毒式的方式)传播开来；NASA 喷气推进实验室(JPL)的团队负责人格伦·里夫斯(Glenn Reeves)最终详细地回复了这一消息，并发表了一篇现在相当著名的文章，标题为：火星上到底发生了什么？。在这篇文章和随后写的关于这个主题的文章中，捕捉到了许多有趣的见解。 在我看来，所有的软件工程师都会通过阅读这些内容来帮自己的忙！ (请务必查找在进一步阅读一节中提供的链接，该部分位于火星探路器和优先级反转下的 GitHub 存储库。)

格伦·里夫斯强调了几个重要的经验教训，以及他们能够重现和解决这个问题的原因，其中一个是这样的：我们坚信测试就是你飞行的东西，飞行你测试的哲学。实际上，由于设计决定将相关的详细诊断和调试信息保存在跟踪/日志环形缓冲区中，这些缓冲区可以随意转储(并发送到地球)，他们能够调试手头的根本问题。

优先级继承-避免优先级反转

好的，很好；但是如何解决优先级反转这样的问题呢？ 有趣的是，这是一个已知的风险，互斥体的设计包括一个内置的解决方案。关于帮助解决优先级反转问题，存在两个互斥体属性-优先级继承(PI)和优先级上限。

PI 是一个有趣的解决方案。 想想看，关键问题是操作系统调度线程的方式。 在操作系统(尤其是 RTOS)中，实时线程的调度-决定谁运行-本质上与竞争线程的最高优先级成正比：优先级越高，运行的机会就越大。 因此，让我们快速回顾一下前面的场景示例。 回想一下，我们有这三个优先级不同的线程：

• 线程 A：高优先级，90
• 线程 B：低优先级，10
• 线程 C：中等优先级，45

当线程 B 长时间持有互斥锁时，就会发生优先级反转，从而迫使线程 A 在解锁过程中阻塞太长时间(超过最后期限)。 因此，想想看：如果线程 B 抓住互斥锁的那一刻，我们可以将它的优先级提高到系统上也在等待同一互斥锁的最高优先级线程的优先级。然后，当然，线程 B 将获得优先级 90，因此它不能被(线程 C 或任何其他线程)抢占！这确保了它快速完成其临界区并解锁互斥锁；一旦解锁，它就回到原来的优先级。 这就解决了问题；这种方法被称为 PI。

PthreadsAPI 集合提供了一对 API 来查询和设置互斥锁的协议属性，在此基础上可以使用 PI：

int pthread_mutexattr_getprotocol(const pthread_mutexattr_t
     *restrict attr, int *restrict protocol);
int pthread_mutexattr_setprotocol(pthread_mutexattr_t *attr,
     int protocol);

协议参数可以采用以下值之一：PTHREAD_PRIO_INSTORITIT、 PTHREAD_PRIO_NONE 或*PTHREAD_PRIO_PRIO_PROTECT(默认值为 PTHREAD_PRIO_NONE)。 当互斥锁具有继承或保护协议之一时，其所有者线程在调度优先级方面会受到影响。

持有使用 PTHREAD_PRIO_INSTORITE 协议初始化的任何互斥体上的锁(拥有它)的线程将继承也使用此协议的任何这些互斥体(健壮或非健壮)上被阻塞(等待)的任何线程的最高优先级(因此以该优先级执行)。

持有使用 TPTHREAD_PRIO_PRIO_PROTECT 协议初始化的任何互斥体上的锁(拥有它)的线程将继承也使用此协议的任何线程的最高优先级上限(并因此以该优先级执行)，无论它们当前是否阻塞(等待)这些互斥体中的任何一个(健壮或非健壮)。

如果线程使用使用不同协议初始化的互斥锁，则它将以其中定义的最高优先级执行。

在探路者任务中，使用的实时操作系统是风河公司著名的 VxWorks。 互斥锁(或信号量)当然有 PI 属性；只是 JPL 软件团队错过了打开互斥锁的 PI 属性，导致了优先级反转问题！ (实际上，软件团队很清楚这一点，并在几个地方使用了它，但不是一个地方-墨菲定律在起作用！)

此外，开发人员可以利用优先级上限-这是所有者线程执行临界区代码的最低优先级。 因此，能够指定这一点，就可以确保它处于足够高的值，以保证所有者线程在临界区中不会被抢占。 Pthreadspthread_mutexattr_getprioceiling(3)和 pthreadpthread_mutexattr_setprioceiling(3)接口可用于查询和设置互斥锁的优先级上限属性。 (它必须在有效的 SCHED_FIFO 优先级范围内，在 Linux 平台上通常为 1 到 99)。

Again, in practice, there are some challenges in using priority inheritance and ceiling attributes, which are, mostly, performance overheads:

• 可能会导致更繁重的任务/上下文切换
• 优先级传播可能会增加开销
• 如果线程和锁都很多，则会产生性能开销，而且死锁可能会攀升

互斥锁属性用法摘要

实际上，如果您希望彻底测试和调试您的应用，并且并不真正关心性能(至少现在如此)，那么可以按如下方式设置您的互斥体：

• 在其上设置最健壮的属性(允许一个人在不解锁的情况下抓住车主死亡)：pthread_mutexattr_setrobust(&attr, PTHREAD_MUTEX_ROBUST)
• 将类型设置为错误检查(允许捕获错误自死锁/重新锁定的情况)： pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK)

另一方面，需要挤出性能的设计良好且经过验证的应用将使用普通(默认)互斥锁类型和属性。 前面的情况不会被捕获(反而会导致未定义的行为)，但是它们永远不会发生！

如果需要递归锁，(显然)应该将互斥锁类型设置为 PTHREAD_MUTEX_RECURSIVE。对于递归互斥锁，重要的是要认识到，如果互斥锁被执行了n次，那么它也必须被解锁n次，这样才能被认为是真正处于解锁状态(因此可以再次锁定)。

在多进程、多线程的应用中，如果需要在不同进程的线程之间使用互斥锁，可以通过进程互斥对象的进程共享属性来实现。 请注意，在这种情况下，包含互斥锁的内存本身必须在进程之间共享(我们通常使用共享内存段)。

PPI 和优先级上限属性允许开发人员保护应用免受众所周知的软件风险：优先级反转。

互斥锁-其他变体

本节帮助您理解互斥锁的附加语义(稍有不同)。 我们将讨论超时互斥变量、“忙碌等待”用例和读取器-写入器锁定。

互斥锁尝试超时

在前面的小节锁定指南中，标签为防止饥饿，我们了解到长时间持有互斥锁会导致性能问题；值得注意的是，失败的线程可能会饿死。这是避免这个问题的一种方法(当然，修复任何饥饿的根本原因是要做的最重要的事情！)(锁指南标签下的锁定准则，我们了解长时间持有互斥锁会导致性能问题；值得注意的是，失败的线程将会挨饿。)。 就是让失败的线程只等待互斥锁一段时间；如果需要更长时间才能解锁，那就算了。 这正是pthread_mutex_timedlock(3)API 提供的功能：

#include <pthread.h>
#include <time.h>
int pthread_mutex_timedlock(pthread_mutex_t *restrict mutex,
        const struct timespec *restrict abstime);

很明显：所有锁定语义都与通常的pthread_mutex_lock(3)相同，只是如果锁上的阻塞(等待)时间超过第二个参数-指定为绝对值的时间，其中 API 返回失败-返回的值将为ETIMEDOUT。 (我们已经在第 13 章、定时器中对超时进行了详细编程。)

不过请注意，其他错误返回值也是可能的(例如，EOWNERDEAD表示前一个所有者终止的健壮互斥锁，EDEADLK表示在错误检查互斥锁上检测到死锁，依此类推)。 有关详细信息，请参阅pthread_mutex_timedlock(3)上的手册页。

BUSY-正在等待锁定(非阻塞变量)

我们了解互斥锁的正常工作方式：如果锁已经被锁定，那么尝试获取锁将导致该线程阻止(等待)解锁的发生。 如果你想要这样的设计怎么办：如果锁被锁了，不要让我等待；我会做一些其他的工作，然后重试。这种语义通常被称为忙碌等待或非阻塞，由 trylock 变体提供。 顾名思义，我们可以尝试锁，如果我们得到它，那就太好了；如果没有，那也没关系--我们不会强迫线程等待。 锁可能被进程内的任何线程(如果它是进程共享的互斥体，甚至是外部线程)获取，包括同一线程-如果它被标记为递归的。 但请稍等；如果互斥锁确实是递归锁，那么将立即成功获取它，并且调用将立即返回。

*具体接口如下：

​int pthread_mutex_trylock(pthread_mutex_t *mutex);。

虽然这种忙碌等待语义有时很有用-具体地说，它用于检测和防止某些类型的死锁-但在使用它时要小心。 想想看：对于一个不太满意的锁(一个不经常使用的锁，尝试获取锁的线程很可能会立即获得它)，使用这种忙碌-等待的语义可能会很有用。 但是，对于一个高度满足的锁(热码路径上的锁，经常被拿走和释放)，这实际上会伤害一个人获得锁的机会！ 为什么？ 因为你不愿意等待。 (有趣的是，软件有时会模仿生活，对吧？)

读取器-写入器互斥锁

设想一个具有大约 10 个工作线程的多线程应用；假设在大多数情况下(比如 90%的时间)，8 个工作线程忙于扫描全局链表(或类似的数据结构)。 当然，现在由于它是全局的，我们可以知道它是一个关键部分；如果不能用互斥保护它，很容易导致脏读错误。 但是，这是以较大的性能为代价的：因为每个工作线程都想要搜索列表，所以它被迫等待来自所有者的解锁事件。

计算机科学家已经针对这种情况提出了一种相当创新的替代方案(也称为虚拟读写器问题)，在这种情况下，数据访问是这样的：在大多数时间里，(共享的)数据只被读取而不被写入。我们使用互斥锁的一种特殊变体，称为虚拟读取器-写入器锁：

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);

请注意，这是一种全新的锁类型：Thepthread_wrlock_t。

如果一个线程为自己获得了一个读锁，关键在于：实现现在相信这个线程只读不写；因此，不会进行实际的锁定，API 将只返回 SUCCESS！ 这样，阅读器实际上是并行运行的，从而保持高性能；没有安全问题或竞争，因为他们保证只读不到。

然而，当一个线程想要写数据时，它必须获得一个写锁定：当这种情况发生时，应用正常的锁定语义。写线程现在必须等待所有读线程执行解锁，然后写线程获得写锁定并继续。 当它在临界区内时，没有线程-读取器和写入器-都无法干预；他们将不得不像往常一样阻止(等待)写入器的解锁。 因此，这两个场景现在都进行了优化。

存在用于设置读取器-写入器互斥锁和属性的常见可疑对象-API(按字母顺序)：

• pthread_rwlockattr_destroy(3P)
• pthread_rwlockattr_getpshared(3P)
• pthread_rwlockattr_setkind_np(3P)
• pthread_rwlockattr_getkind_np(3P)
• pthread_rwlockattr_init(3P)
• pthread_rwlockattr_setpshared(3P)

Note that the APIs suffixed with _np imply they are non-portable, and Linux-only.

类似地，读取器-写入器锁定 API 遵循通常的模式-也存在超时和尝试变量：

• pthread_rwlock_destroy(3P)
• pthread_rwlock_init(3P)
• pthread_rwlock_timedrdlock(3P)
• pthread_rwlock_tryrdlock(3P)
• pthread_rwlock_unlock(3P)
• pthread_rwlock_rdlock(3P)
• pthread_rwlock_timedwrlock(3P)
• pthread_rwlock_trywrlock(3P)
• pthread_rwlock_wrlock(3P)

我们期望程序员以正常的方式进行设置-初始化 rrwlock 属性对象，初始化 rrwlock 属性本身(使用pthread_rwlock_init(3P))，一旦完成就销毁属性结构，然后根据需要执行实际的锁定。(=

不过，请注意，在使用读取器-写入器锁时，应该仔细测试应用的性能；已经注意到，它的实现速度比通常的互斥锁慢。 此外，还有一个额外的担忧，即在负载下，读写器锁定语义可能会导致写入器饥饿。我认为：如果读者不断涌现，写入器线程可能需要等待很长时间才能获得锁。

显然，有了读写器锁，相反的动态也可能发生：读取器可能会饿死。有趣的是，Linux 提供了一个非常不可移植的 API，允许程序员指定要防止哪种类型的饥饿-读取器还是写入器-默认情况下是写入器挨饿。 要调用以设置此设置的 API 为pthread_rwlockattr_setkind_np(3)。 这允许根据您的特定工作负载进行一定程度的调优。 (然而，该实现显然仍然存在一个缺陷，实际上，编写器匮乏仍然是现实。 我们不打算进一步讨论这一点；如果需要进一步的帮助，读者可以参考手册页。)

尽管如此，读取器-写入器锁变体通常很有用；想想那些需要经常扫描某些键值映射数据结构并执行某种类型的表查找的应用。 (例如，操作系统通常具有经常查找路由表但很少更新路由表的网络代码路径。)。 不变的是，所讨论的全局共享数据通常是从数据读取的，而很少是写入的。

自旋锁变种

这里重复一下：我们已经了解了互斥锁是如何正常工作的；如果锁已经被锁定，那么尝试获取锁将导致该线程阻塞(等待)解锁的发生。让我们更深入地挖掘一下；失败的线程究竟是如何阻塞-等待-互斥锁的解锁的呢？ 答案是，对于互斥锁，它们是通过休眠进程(由操作系统调度离开 CPU)来完成的。事实上，这是进程互斥锁的定义属性之一。

另一方面，还有一种完全不同的锁-自旋锁(在 Linux 内核中非常常用)，它的行为与之完全相反：它的工作方式是让失败的线程通过旋转(轮询)来等待解锁操作-嗯，现实情况是，实际的自旋锁实现比在这里听起来要精致和高效得多；不过，这个讨论远远超出了本书的范围。 乍一看，轮询似乎不是让失败的线程等待解锁的一种糟糕方式；它与自旋锁配合良好的原因是，在临界区内花费的时间肯定非常短(从技术上讲，少于执行两次上下文切换所需的时间)，因此当临界区很小时，自旋锁的使用效率比互斥体高得多。

尽管 pthreads 的实现确实提供了自旋锁，但您应该清楚地了解以下几点：

• 自旋锁仅供采用实时操作系统调度策略(SCHED_FIFO，可能还有 SCHED_RR；我们将在第 17 章，Linux 上的 CPU 调度)的极高性能实时线程使用。
• Linux 平台上的默认调度策略从来不是实时的；它是非实时的 SCHED_OTHER 策略，非常适合不确定的应用；使用互斥锁是可行的。
• 在用户空间中使用自旋锁不被认为是正确的设计方法；此外，代码更容易受到死锁和(无界的)优先级反转场景的影响。

基于上述原因，我们不再深入研究以下 pthread 和 Spinlock API：

• pthread_spin_init(3)
• pthread_spin_lock(3)
• pthread_spin_trylock(3)
• pthread_spin_unlock(3)
• pthread_spin_destroy(3)

如果需要，一定要在它们各自的手册页中查找它们(但如果使用它们，也要加倍小心！)

更多的互斥体使用指南

除了前面提供的提示和指南(请参阅锁定指南章节)外，还请考虑以下几点：

• 一个人应该使用多少把锁？
• 对于许多锁实例，如何知道使用哪个锁变量以及何时使用呢？
• 测试互斥锁是否锁定。

让我们把这些要点逐一提出来。

在小型应用中(如此处所示)，也许只使用一个锁来保护临界区就足够了；它的优点是使事情变得简单(这是一件大事)。 然而，在大型项目中，仅使用一个锁对可能遇到的每个临界区执行锁定都有可能成为主要的性能破坏者！ 想想看为什么会发生这种情况：一旦代码中的任何地方命中一个互斥锁，所有并行都会停止，并且代码以序列化方式运行；如果这种情况发生得足够频繁，性能将会迅速下降。

Interestingly, the Linux kernel, for years, had a major performance headache precisely because of one lock that was being used throughout large cross sections of the codebase—so much so, that it was nicknamed the big kernel lock (BKL) (a giant lock). It was finally gotten rid of only in the 2.6.39 version of the Linux kernel (see the Further reading section on the GitHub repository for a link to more on the BKL).

因此，虽然没有规则来确定应该使用多少锁，但启发式的做法是考虑简单性与性能之间的权衡。 作为提示，在大型生产质量项目(如 Linux 内核)中，我们通常使用单个锁来保护单个数据-即数据对象；通常，这只是一种数据结构。 这将确保全局数据在访问时受到保护，但只受实际访问它的代码路径的保护，而不是每个代码路径，从而确保数据安全和并行性(性能)。

好的，太好了。 现在，如果我们真的遵循这个指导方针，那么如果我们最终只有几百个锁呢！？ (是的，在具有数百个全局数据结构的大型项目中，这是完全可能的。)。 现在，我们有了另一个实际问题：开发人员必须确保他们使用正确的锁来保护给定的数据结构(在访问数据结构 Y 的同时，使用锁 X 对数据结构 X 意味着什么？ 这将是一个严重的缺陷)。 因此，一个实际问题是，我如何确定哪个数据结构受哪个锁保护，或者用另一种方式来说明：如何(我如何确定哪个锁变量保护哪个数据结构？)。 天真的解决办法就是给每把锁起一个合适的名字，也许是像lock_<DataStructureName>这样的名字。嗯，不像看起来那么简单！

Informal polls have revealed that, often, one of the hardest things a programmer does is variable naming! (See the Further reading section on the GitHub repository for a link to this.)

因此，这里有一个提示：将保护给定数据结构的锁嵌入到数据结构本身中；换句话说，使其成为其保护的数据结构的成员！ (同样，Linux 内核经常使用这种方法。)

互斥体锁定了吗？

在某些情况下，开发人员可能会问：给定一个互斥体，我可以知道它是处于锁定状态还是解锁状态？ 或许理由是：如果锁上了，我们就解锁吧。

有一种方法可以测试这一点：使用pthread_mutex_trylock(3)API。 如果它返回EBUSY，则意味着互斥当前被锁定(否则，它应该返回0，意味着它是解锁的)。 但是等等！ 这里有一个固有的竞争条件；只要想一想就知道了：

if (pthread_mutex_trylock(&mylock) != EBUSY)) {    <-- time t1
    // it's unlocked                               <-- time t2
}
// it's locked

当我们到达时间 t2 时，不能保证到目前为止，另一个线程没有锁定有问题的互斥体！ 所以，这种做法是不正确的。 (实现这种同步的唯一现实方法是放弃通过互斥锁进行同步，而使用条件变量；这将在下一节中介绍。)

关于互斥锁的(相当长的)报道到此结束。 在我们完成之前，我们想指出另一个有趣的地方：我们在前面已经说过，成为原子级意味着能够不间断地运行关键代码部分直到完成。 但现实是，我们的现代系统确实不会以(令人震惊的)规律性打断我们-硬件中断和异常是常态！因此，人们应该意识到：

• 在用户空间，由于不可能屏蔽硬件中断，进程和线程不会在任何时间点因硬件中断而中断。 因此，用户空间代码基本上不可能是真正的原子代码。(但是，如果我们被硬件中断/故障/异常打断，那又怎么样呢？ 他们将完成他们的工作，并将控制权交还给我们，这一切都非常迅速。 我们不太可能竞争，与这些代码实体共享全局可写数据)。
• 然而，在内核空间中，我们以 OS 特权运行，实际上甚至可以屏蔽硬件中断，从而允许我们以真正的原子方式运行(您认为著名的 Linux 内核 Spinlock 是如何工作的？)。

现在我们已经介绍了用于锁定的典型 API，我们鼓励读者，第一，动手试用示例；第二，重温前面的章节，即锁定准则和死锁。

条件变量

CCV 是一种线程间事件通知机制。当我们使用互斥锁来同步(序列化)对临界区的访问，从而保护它时，我们使用条件变量来促进进程线程之间的高效通信-基于数据项的值进行同步。 下面的讨论将使这一点更加清楚。

在多线程应用的设计和实现中，经常会遇到这样的情况：一个线程 B 正在执行某些工作，而另一个线程 A 正在等待该工作的完成。 只有当线程 B 完成工作时，线程 A 才应该继续；我们如何在代码中有效地实现这一点？

没有简历--天真的做法

您可能还记得，线程的退出状态(通过pthread_exit(3))被传递回调用pthread_join(3)的线程；我们可以利用这个特性吗？ 嗯，不是的：首先，一旦指定的工作完成，线程 B 不一定会终止(这可能只是一个里程碑，而不是它必须执行的所有工作)，其次，即使它确实终止了，除了调用pthread_join(3)的线程之外，可能还有其他线程可能需要知道。

好的；为什么不让线程 A 在完成时通过简单的技术进行轮询，即在工作完成时让线程 B 将全局整数(称为gWorkDone)设置为 1(当然，还有让线程 A 对其进行轮询)，在伪代码中可能类似于以下内容：

| 时间 | 线程 B | 线程 A | | T0 | 初始化：gWorkDone = 0 | < common > | | T1 期 | 完成这项工作..。 | while (!gWorkDone) ; | | T2 | ..。 | ..。 | | T3 | 已完成的工作；gWorkDone = 1 | ..。 | | T4 | | 检测到；中断循环并继续 |

可能行得通，但不行。有何不可？

• 首先，在一个变量上进行无限时间的轮询在 CPU 方面是非常昂贵的(这只是一个糟糕的设计)。
• 第二，请注意，我们正在对一个完全共享的可写全局变量进行操作，而没有对其进行保护；*这正是将数据争用引入应用的方式，因此也就是将错误引入应用的方式。

因此，上表所示的方法被认为是幼稚的、低效的，甚至可能是错误的(Racy)。

使用条件变量

正确的方法是使用新的 CV。条件变量是线程可以高效地根据数据的值进行同步的一种方式。 它实现了与幼稚的轮询方法相同的最终结果，但以一种更有效、更重要的方式实现了这一点。

请查看下表：

| 时间 | 线程 B | 线程 A | | T0 | 初始化：gWorkDone=0；初始化{cv，mutex}对 | < common > | | T1 期 | | 等待来自线程 B 的信号：锁定关联的互斥体；pthread_cond_wait() | | T2 | 完成这项工作..。 | < ... blocking ... > | | T3 | 工作完成； 锁定关联的互斥锁；信号线程 A：pthread_cond_signal()；解锁关联的互斥锁 | ...... | | T4 | | 解锁；检查工作是否真的完成，如果是，解锁与其关联的互斥，然后继续... |

虽然上表向我们展示了步骤的顺序，但仍需要一些解释。 在幼稚的方法中，我们看到其中一个(严重的)缺点是全局共享数据变量在没有保护的情况下被操纵！ 条件变量通过要求条件变量始终与互斥锁相关联来解决这个问题；我们可以将其视为**{CV，互斥锁}对**。

这个想法很简单：每次我们打算使用全局谓词来告诉我们工作是否已经完成时(gWorkDone，在我们的示例中)，我们锁定互斥体，读/写全局互斥体，解锁互斥体，因此--重要的是！--保护它。

CV 的美妙之处在于我们根本不需要轮询：等待工作完成的线程在该事件发生时使用pthread_cond_wait(3)阻塞(等待)，并且已经完成工作的线程通过pthread_cond_signal(3)API“通知”对应的线程：

int pthread_cond_wait(pthread_cond_t *restrict cond,
                      pthread_mutex_t *restrict mutex);
int pthread_cond_signal(pthread_cond_t *cond);

Though we use the word signal here, this has nothing to do with Unix/Linux signals and signaling that we covered in earlier Chapters 11, Signaling - Part I, and Chapter 12, Signaling - II.

(请注意{cv，mutex}对是如何组合在一起的)。 当然，与线程一样，我们必须首先初始化 CV 及其关联的互斥锁；CV 可以通过以下方式静态初始化：

pthread_cond_t cond = PTHREAD_COND_INITIALIZER; 

或通过以下 API 动态(在运行时)：

int pthread_cond_init(pthread_cond_t *restrict cond,
                          const pthread_condattr_t *restrict attr);

如果要设置 CV 的特定非默认属性，可以通过pthread_condattr_set*(3P)API 进行设置，或者只需通过首先调用pthread_condattr_init(3P)API 并将已初始化的 CV 属性对象作为第二个参数传递给pthread_cond_init(3P)来将 CV 设置为默认值即可：

int pthread_condattr_init(pthread_condattr_t *attr);

相反，完成后，请使用以下 API 来销毁 CV 属性对象和 CV 本身：

int pthread_condattr_destroy(pthread_condattr_t *attr);
int pthread_cond_destroy(pthread_cond_t *cond);

一个简单的 CV 使用演示应用

初始化/破坏太多了吗？ 请看下面的简单代码(ch15/cv_simple.c)，它将阐明它们的用法；我们将编写一个更小的程序来演示条件变量及其关联的互斥锁的用法。 在这里，我们创建了两个线程 A 和 B。然后，我们让线程 B 执行一些工作，线程 A 使用{cv，mutex}对在这些工作完成后进行同步：

For readability, only key parts of the source code are displayed; to view the complete source code, build and run it. The entire tree is available for cloning from GitHub here: https://github.com/PacktPublishing/Hands-on-System-Programming-with-Linux.

...
#define LOCK_MTX(mtx) do { \
  int ret=0; \
  if ((ret = pthread_mutex_lock(mtx))) \
    FATAL("pthread_mutex_lock failed! [%d]\n", ret); \
} while(0)

#define UNLOCK_MTX(mtx) do { \
  int ret=0; \
  if ((ret = pthread_mutex_unlock(mtx))) \
    FATAL("pthread_mutex_unlock failed! [%d]\n", ret); \
} while(0)

static int gWorkDone=0;
/* The {cv,mutex} pair */
static pthread_cond_t mycv;
static pthread_mutex_t mycv_mutex = PTHREAD_MUTEX_INITIALIZER;

在前面的代码中，我们再次展示了实现互斥锁和解锁的宏、全局谓词(布尔)变量gWorkDone，当然还有{cv，mutex}对变量。

在下面的代码中，在 Main 中，我们初始化 CV 属性对象和 CV 本身：

// Init a condition variable attribute object
  if ((ret = pthread_condattr_init(&cvattr)))
      FATAL("pthread_condattr_init failed [%d].\n", ret);
  // Init a {cv,mutex} pair: condition variable & it's associated mutex
  if ((ret = pthread_cond_init(&mycv, &cvattr)))
      FATAL("pthread_cond_init failed [%d].\n", ret);
  // the mutex lock has been statically initialized above.

创建工作线程 A 和 B 并开始它们的工作(我们在这里不重复显示线程创建的代码)。 在这里，您会发现线程 A 的工作例程-它必须等待线程 B 完成工作。 我们使用{cv，mutex}对轻松高效地实现这一点。

然而，该库确实要求应用保证在调用pthread_cond_wait(3P)API 之前，关联的互斥体锁定已被获取(锁定)；否则，这将导致未定义的行为(或者当互斥体类型为非PTHREAD_MUTEX_ERRORCHECK或健壮的互斥体时，实际失败)。 一旦线程在 CV 上阻塞，互斥锁就会自动释放。

此外，如果在线程处于等待状态时发送信号，则应处理该信号并恢复等待；这还可能导致虚假唤醒的返回值为零(请立即详细介绍)：

static void * workerA(void *msg)
{
  int ret=0;

  LOCK_MTX(&mycv_mutex);
  while (1) {
      printf(" [thread A] : now waiting on the CV for thread B to finish...\n");
      ret = pthread_cond_wait(&mycv, &mycv_mutex);
      // Blocking: associated mutex auto-released ...
      if (ret)
          FATAL("pthread_cond_wait() in thread A failed! [%d]\n", ret);
      // Unblocked: associated mutex auto-acquired upon release from the condition wait...

      printf(" [thread A] : recheck the predicate (is the work really "
 "done or is it a spurious wakeup?)\n");
 if (gWorkDone)
 break;
      printf(" [thread A] : SPURIOUS WAKEUP detected !!! "
             "(going back to CV waiting)\n");
  }
 UNLOCK_MTX(&mycv_mutex);
  printf(" [thread A] : (cv wait done) thread B has completed it's work...\n");
  pthread_exit((void *)0);
}

理解这一点非常重要：仅仅从线程pthread_cond_wait(3P)返回并不一定意味着我们等待(阻塞)的条件-在本例中，线程 B 完成了工作-实际上发生了！ 在软件中，可能会出现完全虚假的唤醒状态(由于其他事件-可能是信号而导致的虚假唤醒)；健壮的软件会在循环中重新检查条件，以确定我们被唤醒的原因是正确的-在我们的情况下，工作确实已经完成。 这就是为什么我们在无限循环中运行，一旦解除对pthread_cond_wait(3P)的阻塞，就会检查全局整数gWorkDone是否真的具有我们期望的值(在本例中，1 表示工作完成)。

好的，但也要考虑到这一点：即使读取共享全局变量也会成为临界区(否则可能导致脏读取)；因此，我们需要在执行此操作之前获取互斥锁。 啊，这就是{cv，mutex}配对的想法有一个内置的自动机制真正帮助我们的地方-当我们调用pthread_cond_wait(3P)的那一刻，相关的互斥锁被自动和原子地释放(解锁)，然后我们阻塞条件变量信号。 当另一个线程(这里是 B)向我们发出信号时(显然是在同一个 CV 上)，我们就会从pthread_cond_wait(3P)中解锁，并且与之关联的互斥锁将自动和原子锁定，从而允许我们重新检查全局(或其他)。 所以，我们做我们的工作，然后打开它。

以下是线程 B 的 Worker 例程的代码，该例程执行一些样本工作，然后向线程 A 发出信号：

static void * workerB(void *msg)
{
  int ret=0;

  printf(" [thread B] : perform the 'work' now (first sleep(1) :-)) ...\n");
  sleep(1);
  DELAY_LOOP('b', 72);
  gWorkDone = 1;

  printf("\n [thread B] : work done, signal thread A to continue ...\n");
  /* It's not strictly required to lock/unlock the associated mutex
   * while signalling; we do it here to be pedantically correct (and
   * to shut helgrind up).
   */
  LOCK_MTX(&mycv_mutex);
  ret = pthread_cond_signal(&mycv);
  if (ret)
      FATAL("pthread_cond_signal() in thread B failed! [%d]\n", ret);
  UNLOCK_MTX(&mycv_mutex);
  pthread_exit((void *)0);
}

请注意详细说明了为什么我们在信号之前再次使用互斥锁的注释。 好的，让我们试一试(我们建议您构建并运行调试版本，因为这样延迟循环就会正确显示)：

$ ./cv_simple_dbg 
 [thread A] : now waiting on the CV for thread B to finish...
 [thread B] : perform the 'work' now (first sleep(1) :-)) ...
bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
 [thread B] : work done, signal thread A to continue ...
 [thread A] : recheck the predicate (is the work really done or is it a spurious wakeup?)
 [thread A] : (cv wait done) thread B has completed it's work...
$ 

API 还提供阻塞调用的超时变量：

int pthread_cond_timedwait(pthread_cond_t *restrict cond,
    pthread_mutex_t *restrict mutex, const struct timespec *restrict abstime);

语义与pthread_cond_wait相同，不同之处在于，如果第三个参数中指定的时间已经(已经)过去，则 API 返回(带故障值为 1ETIMEDOUT)。(如果已过了第三个参数中指定的时间，则返回pthread_cond_wait，不同之处在于 API 返回的故障值为 0ETIMEDOUT)。 用于测量经过的时间的时钟是 CV 的一个属性，可以通过 npthread_condattr_setclock(3P)API 设置。

(pthread_cond_wait和pthread_cond_timedwait都是取消点；此主题将在下一章中讨论。)

CV 广播唤醒

正如我们之前看到的，pthread_cond_signal(3P)API 用于解锁在特定 CV 上被阻塞的线程。 该接口的变体如下：

int pthread_cond_broadcast(pthread_cond_t *cond);

此 API 允许您解除对同一 CV 上阻塞的多个线程的阻塞。例如，如果我们在同一 CV 上有三个线程阻塞；当应用调用线程pthread_cond_broadcast(3P)时，哪个线程将首先运行？ 嗯，这就像在问，当创建线程时，哪个线程将首先运行(回想一下上一章中的讨论)。 当然，答案是，在没有特定的调度策略的情况下，它是不确定的。 当应用于 CV 解锁并在 CPU 上运行时，同样的答案也适用于这个问题。

要继续，一旦等待的线程被解锁，请回想一下，关联的互斥体将被获取，但当然只有一个被解锁的线程会首先获得它。 同样，这取决于调度策略和优先级。 在所有默认情况下，它仍然不确定哪个线程首先获取它。 在任何情况下，在没有实时特征的情况下，这对应用应该无关紧要(如果应用是实时的，那么请阅读我们的文章第 17 章，Linux 上的 CPU 调度，以及首先在每个应用线程上设置实时调度策略和优先级)。

此外，这些 API 的手册页面清楚地说明，尽管调用前面的 API 的线程(pthread_cond_signal和pthread_cond_broadcast)在执行此操作时并不要求您持有关联的互斥锁(回想一下，我们总是有一对{CV，mutex})，但严格纠正语义要求它们确实持有互斥，执行信号或广播，然后解锁互斥锁(我们的示例应用，ch15/cv_simple.c，它确实遵循这个指导原则)。

为了让这篇关于简历的讨论更加完整，这里有几个小贴士：

• 不要在信号处理程序中使用条件变量方法；代码不被认为是异步信号安全的(回想一下前面的第 11 章、信号-第 I 部分和第 12 章、信号-第 II 部分)。
• 使用著名的 Valgrind 工具套件(回想一下，我们在第章，中介绍了 Valgrind 的 Memcheck 工具以解决内存问题)，特别是名为 dhelgrind 的工具，对于检测 p 线程和多线程应用中的同步错误(数据竞争)非常有用(有时)。 用法很简单： $ valgrind --tool=helgrind [-v] <app_name> [app-params ...]：
  • 不过，与许多此类工具一样，helgrind 经常会引发许多误报。 例如，我们发现消除我们之前编写的cv_simple应用中的printf(3)会从 Helgrind 中移除大量(误报)错误和警告！
  • 在调用pthread_cond_signal函数和/或pthread_cond_broadcast函数 API 之前，如果没有首先获取与之关联的互斥锁函数(这不是必需的)，则 Helgrind 会抱怨。

一定要尝试 helgrind Out(同样，GitHub 存储库上的进一步阅读部分有一个指向其(非常好的)文档的链接)。

简略的 / 概括的 / 简易判罪的 / 简易的

本章首先介绍了并发性、原子性等关键概念，以及识别临界区并对其进行保护的必要性。锁定是实现这一点的典型方法；pthreadAPI 提供了强大的互斥锁来实现这一点。 然而，使用锁，特别是在大型项目中，充满了隐藏的问题和危险--我们讨论了有用的锁定准则、死锁和及其避免。

然后，本章继续指导读者使用 pthread 和互斥锁。 这里涵盖了大量内容，包括各种互斥锁属性、认识和避免优先级反转问题的重要性以及互斥锁的变化。 最后，我们介绍了环境条件变量(CV)的需求和用法，以及如何使用它来高效地促进线程间事件通知。

下一章是这个关于多线程的三部曲中的最后一章；在其中，我们将集中讨论线程安全(和线程安全 API)、线程取消和清理、与 MT 混合信号、一些常见问题和提示等重要问题，并看看多进程模型与传统多线程模型的优缺点。**