RT-mutex 实现设计

Copyright (c) 2006 Steven Rostedt

依照 GNU 自由文档许可证（GNU Free Documentation License）1.2 版发布

本文档旨在描述 rtmutex.c 的实现设计。它没有描述 rtmutex.c 存在的原因。有关这方面的内容，请参阅 支持 PI 的 RT-mutex 子系统。尽管本文档确实解释了如果没有此代码会发生什么问题，但这主要是为了便于理解该代码实际所做的工作。

本文档的目标是帮助他人理解所使用的优先级继承 (PI) 算法，以及决定以这种特定方式实现 PI 的原因。

无界优先级反转

优先级反转是指低优先级进程在更高优先级进程想要运行时执行。发生这种情况有几个原因，而且大多数时候是不可避免的。每当高优先级进程想要使用低优先级进程拥有的资源（例如互斥锁）时，高优先级进程必须等待，直到低优先级进程使用完该资源。这就是优先级反转。我们想要防止的是被称为无界优先级反转的情况。即高优先级进程被低优先级进程阻止运行不确定的时间。

无界优先级反转的经典例子是：假设有三个进程，我们称之为进程 A、B 和 C，其中 A 是最高优先级进程，C 是最低的，B 处于中间。A 尝试获取 C 拥有的锁，因而必须等待并让 C 运行以释放该锁。但在同时，B 开始执行，由于 B 的优先级高于 C，它会抢占 C，但通过这样做，它实际上抢占了比其优先级更高的 A。现在，我们无法预知 A 会等待 C 释放锁休眠多久，因为据我们所知，B 可能是一个 CPU 吞噬者，永远不会给 C 释放锁的机会。这就是所谓的无界优先级反转。

这里用一个小型 ASCII 图来展示该问题

   grab lock L1 (owned by C)
     |
A ---+
        C preempted by B
          |
C    +----+

B         +-------->
                B now keeps A from running.

优先级继承 (PI)

有几种方法可以解决此问题，但其他方法超出了本文档的范围。这里我们仅讨论 PI。

PI 是指如果另一个进程阻塞在当前进程拥有的锁上，该进程就会继承另一个进程的优先级。为了更容易理解，让我们再次使用前面的例子，即进程 A、B 和 C。

这一次，当 A 阻塞在 C 拥有的锁上时，C 将继承 A 的优先级。因此，如果此时 B 变为可运行状态，它将不会抢占 C，因为 C 现在拥有 A 的高优先级。一旦 C 释放了锁，它就会失去继承的优先级，然后 A 就可以继续使用 C 曾拥有的资源。

术语

在这里，我解释一些本文档中使用的术语，以帮助描述用于实现 PI 的设计。

PI 链

  • PI 链是一系列有序的锁和进程，它们导致进程从阻塞在其某个锁上的前一个进程继承优先级。本文档后面将对此进行更详细的描述。

mutex

  • 在本文档中，为了区分实现 PI 的锁和 PI 代码中使用的自旋锁，从现在开始，将 PI 锁称为互斥锁（mutex）。

lock

  • 从本文档的现在开始，当我提及用于保护 PI 算法某些部分的自旋锁时，我将使用术语“锁（lock）”。对于 UP（启用 CONFIG_PREEMPT 时），这些锁会禁用抢占；在 SMP 上，它们可防止多个 CPU 同时进入临界区。

自旋锁 (spin lock)

  • 同上文的锁。

等待者 (waiter)

  • 等待者是一个存储在被阻塞进程栈上的结构体。由于 waiter 的作用域在进程阻塞于互斥锁的代码范围内，因此在进程的栈上（作为局部变量）分配 waiter 是完全可以的。该结构体包含一个指向任务的指针，以及任务所阻塞的互斥锁。它还包含红黑树（rbtree）节点结构，用于将任务放置在互斥锁的 waiters 红黑树中，以及互斥锁所有者任务的 pi_waiters 红黑树中（如下所述）。

    waiter 有时也用来指代等待互斥锁的任务。这与 waiter->task 相同。

等待者列表 (waiters)

  • 阻塞在某个互斥锁上的进程列表。

顶层等待者 (top waiter)

  • 等待特定互斥锁的最高优先级进程。

顶层 PI 等待者 (top pi waiter)

  • 等待特定进程拥有的某个互斥锁的最高优先级进程。

注意

在本文档中，task（任务）和 process（进程）可以互换使用，主要是为了区分同时描述的两个进程。

PI 链

PI 链是一系列可能导致优先级继承发生的进程和互斥锁列表。多个链可能会汇聚，但链绝不会分叉，因为一个进程在同一时间不能阻塞在多个互斥锁上。

示例

Process:  A, B, C, D, E
Mutexes:  L1, L2, L3, L4

A owns: L1
        B blocked on L1
        B owns L2
               C blocked on L2
               C owns L3
                      D blocked on L3
                      D owns L4
                             E blocked on L4

该链将是

E->L4->D->L3->C->L2->B->L1->A

为了展示两条链在哪里合并，我们可以添加另一个进程 F 和另一个互斥锁 L5，其中 B 拥有 L5，而 F 阻塞在互斥锁 L5 上。

针对 F 的链将是

F->L5->B->L1->A

由于一个进程可以拥有多个互斥锁，但绝不会同时阻塞在多个互斥锁上，因此这些链会合并。

这里我们展示两条链

E->L4->D->L3->C->L2-+
                    |
                    +->B->L1->A
                    |
              F->L5-+

为了让 PI 发挥作用，这些链右端的进程（或者我们也可以称之为链的顶部）的优先级必须大于或等于链中左侧或下方的进程。

此外，由于一个互斥锁上可能有多个进程阻塞，我们可能会在互斥锁处有多个链合并。如果我们添加另一个阻塞在互斥锁 L2 上的进程 G

G->L2->B->L1->A

再次为了展示这可以如何增长，我将再次展示合并的链

E->L4->D->L3->C-+
                +->L2-+
                |     |
              G-+     +->B->L1->A
                      |
                F->L5-+

如果进程 G 在链中拥有最高优先级，那么链中之上的所有任务（本例中的 A 和 B）的优先级必须提升至与 G 相同。

互斥锁等待者树

每个互斥锁都会跟踪所有阻塞在它上面的等待者。互斥锁有一个 rbtree，按优先级存储这些等待者。该树由位于互斥锁的 struct of 中的自旋锁保护。这个锁被称为 wait_lock。

任务 PI 树

为了跟踪 PI 链，每个进程都有自己的 PI rbtree。这是一棵由该进程拥有的所有互斥锁的顶层等待者组成的树。请注意，这棵树只保存顶层等待者，而不保存阻塞在该进程拥有的互斥锁上的所有等待者。

任务的 PI 树的顶部始终是等待该任务所拥有的互斥锁的最高优先级任务。因此，如果该任务继承了优先级，它将始终是位于该树顶部的任务的优先级。

该树作为名为 pi_waiters 的 rbtree 存储在进程的任务结构体中。它由同样位于任务结构体中的名为 pi_lock 的自旋锁保护。该锁也可能在中断上下文中被获取，因此在对 pi_lock 加锁时，必须禁用中断。

PI 链的深度

PI 链的最大深度不是动态的，实际上是可以被定义的。但要弄清楚这一点非常复杂，因为它取决于互斥锁的所有嵌套情况。让我们看一个例子：我们有 3 个互斥锁 L1、L2 和 L3，以及四个独立的函数 func1、func2、func3 和 func4。下面展示了 L1->L2->L3 的加锁顺序，但实际上可能并非以这种方式直接嵌套

void func1(void)
{
      mutex_lock(L1);

      /* do anything */

      mutex_unlock(L1);
}

void func2(void)
{
      mutex_lock(L1);
      mutex_lock(L2);

      /* do something */

      mutex_unlock(L2);
      mutex_unlock(L1);
}

void func3(void)
{
      mutex_lock(L2);
      mutex_lock(L3);

      /* do something else */

      mutex_unlock(L3);
      mutex_unlock(L2);
}

void func4(void)
{
      mutex_lock(L3);

      /* do something again */

      mutex_unlock(L3);
}

现在我们添加 4 个分别运行这些函数的进程。进程 A、B、C 和 D 分别运行函数 func1、func2、func3 和 func4，其中 D 最先运行，A 最后运行。当 D 在 func4 的“再次做某事”区域被抢占时，我们得到了如下的加锁情况

D owns L3
       C blocked on L3
       C owns L2
              B blocked on L2
              B owns L1
                     A blocked on L1

And thus we have the chain A->L1->B->L2->C->L3->D.

这给我们的 PI 深度为 4（四个进程），但如果单独看任何一个函数，似乎它们的加锁深度最多只有 2。因此，尽管加锁深度在编译时是确定的，但要找出这种深度的可能性仍然非常困难。

既然互斥锁可以由用户空间应用程序定义，我们就不希望出现那种嵌套大量互斥锁以创建巨大 PI 链的拒绝服务（DOS）类型的应用程序，并且让代码在查看大量数据时持有自旋锁。因此为了防止这种情况，该实现不仅实现了最大加锁深度，而且在遍历 PI 链时，一次最多只持有两个不同的锁。下面会对此进行更多介绍。

互斥锁所有者和标志

互斥锁结构体包含一个指向互斥锁所有者的指针。如果该互斥锁未被拥有，则此所有者设置为 NULL。由于所有架构的任务结构体至少按两字节对齐（如果不是这样，rtmutex.c 代码将会崩溃！），这允许将最低有效位用作标志。第 0 位用作“有等待者（Has Waiters）”标志。每当互斥锁上有等待者时，它就被置位。

有关更多详细信息，请参阅 支持 PI 的 RT-mutex 子系统。

cmpxchg 技巧

某些架构实现了原子的 cmpxchg（比较并交换，Compare and Exchange）。这（在适用的情况下）用于保持获取和释放互斥锁的快速路径简短。

cmpxchg 基本上是以原子方式执行的以下函数

unsigned long _cmpxchg(unsigned long *A, unsigned long *B, unsigned long *C)
{
      unsigned long T = *A;
      if (*A == *B) {
              *A = *C;
      }
      return T;
}
#define cmpxchg(a,b,c) _cmpxchg(&a,&b,&c)

有这个功能真的很好，因为它允许你仅在变量符合预期时才更新变量。如果返回值（A 的旧值）等于 B，你就知道它成功了。

宏 rt_mutex_cmpxchg 用于尝试加锁和解锁互斥锁。如果架构不支持 CMPXCHG，则该宏被简单地设置为每次都失败。但如果支持 CMPXCHG，这将极大地有助于保持快速路径简短。

将 rt_mutex_cmpxchg 与所有者字段中的标志结合使用，有助于为支持它的架构优化系统。本文档后面也会对此进行解释。

优先级调整

rtmutex.c 中的 PI 代码实现有几个地方需要进程调整其优先级。借助于进程的 pi_waiters，了解需要调整的内容变得相当容易。

实现任务调整的函数是 rt_mutex_adjust_prio 和 rt_mutex_setprio。rt_mutex_setprio 仅在 rt_mutex_adjust_prio 中使用。

rt_mutex_adjust_prio 检查任务的优先级，以及等待该任务所拥有的任何互斥锁的最高优先级进程。由于任务的 pi_waiters 按优先级顺序保存了该任务拥有的所有互斥锁的所有顶层等待者，我们只需将顶层 pi 等待者与它自己的普通/截止期限优先级进行比较，并取较高者。然后调用 rt_mutex_setprio 将任务的优先级调整为新优先级。请注意，rt_mutex_setprio 定义在 kernel/sched/core.c 中，用于实现优先级的实际更改。

注意

对于 task_struct 中的 “prio” 字段，数字越小，优先级越高。“prio” 为 5 的优先级高于 “prio” 为 10 的优先级。

有趣的是，rt_mutex_adjust_prio 既可以提高也可以降低任务的优先级。在更高优先级进程刚刚阻塞在由该任务拥有的互斥锁上的情况下，rt_mutex_adjust_prio 会提高/提升任务的优先级。但如果一个更高优先级的任务由于某种原因离开互斥锁（超时或收到信号），这个相同的函数会降低/取消提升任务的优先级。这是因为 pi_waiters 始终包含等待该任务拥有的互斥锁的最高优先级任务，所以我们只需要将该顶层 pi 等待者的优先级与给定任务的普通优先级进行比较即可。

PI 链遍历的高级概述

PI 链遍历由函数 rt_mutex_adjust_prio_chain 实现。

该实现经过了多次迭代，最终形成了我们认为最好的方案。它通过每次最多只获取两个锁来遍历 PI 链，非常高效。

rt_mutex_adjust_prio_chain 可用于提升或降低进程优先级。

调用 rt_mutex_adjust_prio_chain 时传入的参数有：要检查 PI 提升/取消提升的任务（进程正在阻塞其上的互斥锁的所有者）、用于检查死锁的标志、该任务拥有的互斥锁、指向阻塞在该互斥锁上的进程 waiter 结构体的指针（尽管对于取消提升，此参数可以为 NULL）、指向任务阻塞于其上的互斥锁的指针，以及作为互斥锁顶层等待者的 top_task。

在此解释中，我将不提及死锁检测。本解释将尽量保持在较高抽象级别。

调用此函数时，没有持有任何锁。这也意味着进入此函数时，所有者和锁的状态可能会发生变化。

在调用此函数之前，已经对该任务执行了 rt_mutex_adjust_prio。这意味着该任务已被设置为它应有的优先级，但任务 waiter 的 rbtree 节点尚未更新为新优先级，并且该任务可能不在其阻塞所处的 pi_waiters 和 waiters 树的正确位置上。此函数解决了所有这些问题。

Thomas Gleixner 在 rtmutex.c 中总结了该函数的主要操作。有关更多详细信息，请参阅“链遍历基础和保护范围（Chain walk basics and protection scope）”注释。

获取互斥锁（遍历过程）

好的，现在让我们详细看一下获取互斥锁时发生的过程。

首先尝试的是快速获取互斥锁。这在启用了 CMPXCHG 时进行（否则快速获取会自动失败）。只有当互斥锁的 owner 字段为 NULL 时，才能通过 CMPXCHG 获取锁，并且无需再做其他事情。

如果锁存在争用，我们将走慢速路径（rt_mutex_slowlock）。

慢速路径函数是在栈上创建任务的 waiter 结构体的地方。这是因为 waiter 结构体仅在此函数的作用域内需要。waiter 结构体包含用于将任务存储在互斥锁的 waiters 树中的节点，如果需要，还包含所有者的 pi_waiters 树中的节点。

获取互斥锁的 wait_lock，因为解锁互斥锁的慢速路径也需要获取该锁。

然后我们调用 try_to_take_rt_mutex。这就是未实现 CMPXCHG 的架构总是获取锁的地方（如果没有争用的话）。

每当任务在慢速路径中尝试获取互斥锁时，都会使用 try_to_take_rt_mutex。这里做的第一件事是以原子方式设置互斥锁 owner 字段的“有等待者（Has Waiters）”标志。通过现在设置此标志，正在被争用的互斥锁的当前所有者在不进入慢速解锁路径的情况下就无法释放互斥锁，而这又需要获取 wait_lock，该锁正由当前代码持有。因此，设置“Has Waiters”标志迫使当前所有者与此代码进行同步。

如果满足以下条件，则获取锁

  1. 锁没有所有者

  2. 相对于该锁的所有其他等待者，当前任务具有最高优先级

如果任务成功获取了锁，则该任务被设置为锁的所有者，并且如果该锁仍然有等待者，则将 top_waiter（等待该锁的最高优先级任务）添加到该任务的 pi_waiters 树中。

如果锁未被 try_to_take_rt_mutex() 获取，则调用 task_blocks_on_rt_mutex() 函数。这会将任务添加到锁的等待者树中，并传播锁的 pi 链以及锁的所有者的 pi_waiters 树。下一节将对此进行描述。

任务阻塞在互斥锁上

互斥锁和进程的统计是通过进程的 waiter 结构体完成的。“task” 字段设置为该进程，“lock” 字段设置为该互斥锁。waiter 的 rbtree 节点初始化为进程当前的优先级。

由于在进入慢速加锁时获取了 wait_lock，我们可以安全地将 waiter 添加到任务等待者树中。如果当前进程是当前等待此互斥锁的最高优先级进程，那么我们从所有者的 pi_waiters 中移除先前的顶层等待者进程（如果存在），并将当前进程添加到该树中。由于所有者的 pi_waiter 已经改变，我们在所有者身上调用 rt_mutex_adjust_prio，以查看所有者是否应相应地调整其优先级。

如果所有者也阻塞在某个锁上，并且其 pi_waiters 发生了变化（或者开启了死锁检查），我们就会释放互斥锁的 wait_lock，并继续在所有者上运行 rt_mutex_adjust_prio_chain，如前所述。

现在所有锁都已释放，如果当前进程仍然阻塞在互斥锁上（waiter “task” 字段不为 NULL），那么我们就进入睡眠状态（调用 schedule）。

在循环中被唤醒

然后，任务可能由于以下几种原因而被唤醒

  1. 先前的锁所有者释放了锁，并且该任务现在是 top_waiter

  2. 我们收到了信号或超时

在这两种情况下，任务都会再次尝试获取锁。如果成功获取，它将把自己从 waiters 树中移除，并将自己重新设置为 TASK_RUNNING 状态。

在第一种情况下，如果在此任务获取锁之前，锁已被另一个任务获取，那么它将回到睡眠状态并等待再次被唤醒。

第二种情况仅适用于以下任务：它们正在获取一个由于信号或超时而能够在获取锁之前被唤醒的互斥锁（即 rt_mutex_timed_futex_lock()）。被唤醒时，它将再次尝试获取锁；如果成功，则任务返回时持有该锁；否则，如果任务是被信号唤醒的，则返回 -EINTR，如果超时，则返回 -ETIMEDOUT。

释放互斥锁

对于具有 CMPXCHG 的那些架构，互斥锁的解锁也有一条快速路径。由于在存在争用时获取互斥锁总是会设置互斥锁所有者的“Has Waiters”标志，因此我们利用这一点来判断在解锁互斥锁时是否需要走慢速路径。如果互斥锁没有任何等待者，则互斥锁的 owner 字段将等于当前进程，并且只需将 owner 字段替换为 NULL 即可解锁互斥锁。

如果 owner 字段设置了“Has Waiters”位（或者 CMPXCHG 不可用），则走慢速解锁路径。

慢速解锁路径中所做的第一件事是获取互斥锁的 wait_lock。这同步了互斥锁的加锁和解锁操作。

进行检查以确定互斥锁是否有等待者。在没有 CMPXCHG 的架构上，这是互斥锁所有者确定是否需要唤醒等待者的位置。在确实有 CMPXCHG 的架构上，该检查是在快速路径中完成的，但慢速路径中仍然需要它。如果互斥锁的一个等待者在所有者未通过快速路径 CMPXCHG 检查与获取 wait_lock 之间的时间内由于信号或超时而醒来，则该互斥锁可能没有任何等待者，因此所有者仍然需要进行此检查。如果没有等待者，则将互斥锁 owner 字段设置为 NULL，释放 wait_lock，无需再做其他事情。

如果有等待者，那么我们需要唤醒一个。

在唤醒代码中，获取当前所有者的 pi_lock。找到锁的顶层等待者，并将其从互斥锁的 waiters 树以及当前所有者的 pi_waiters 树中移除。“Has Waiters”位被标记，以防止低优先级任务窃取锁。

最后，我们释放待处理所有者的 pi_lock 并唤醒它。

联系方式

有关本文档的更新，请发邮件给 Steven Rostedt <rostedt@goodmis.org>

致谢

作者：Steven Rostedt <rostedt@goodmis.org>

更新：Alex Shi <alex.shi@linaro.org> - 7/6/2017

原审阅者

Ingo Molnar, Thomas Gleixner, Thomas Duetsch, and Randy Dunlap

更新 (7/6/2017) 审阅者：Steven Rostedt and Sebastian Siewior

更新日志

本文档最初是为 2.6.17-rc3-mm1 编写的，在 4.12 版本进行了更新