English

意大利语

不靠谱的锁定指南¶

作者:: Rusty Russell

简介¶

欢迎阅读 Rusty 的“非常之不靠谱”的内核锁定问题指南。本文档描述了 Linux 内核 2.6 版本中的锁定系统。

随着超线程（HyperThreading）的广泛应用以及 Linux 内核中抢占机制的引入，每一位从事内核开发的黑客都需要了解对称多处理（SMP）下并发和锁定的基本原理。

并发带来的问题¶

（如果你已经知道什么是竞态条件 Race Condition，请跳过此段）。

在一个普通的程序中，你可以像这样增加一个计数器：

very_important_count++;

这是人们期望发生的情况：

预期结果¶
实例 1	实例 2
读取 very_important_count (5)
加 1 (6)
写入 very_important_count (6)
	读取 very_important_count (6)
	加 1 (7)
	写入 very_important_count (7)

而这才是可能发生的情况：

可能的结果¶
实例 1	实例 2
读取 very_important_count (5)
	读取 very_important_count (5)
加 1 (6)
	加 1 (6)
写入 very_important_count (6)
	写入 very_important_count (6)

竞态条件与临界区¶

这种重叠现象——即结果取决于多个任务执行的相对时间——被称为竞态条件（race condition）。包含并发问题的代码段被称为临界区（critical region）。特别是自从 Linux 开始在 SMP 机器上运行以来，这已成为内核设计和实现中的主要问题之一。

即使只有一个 CPU，抢占（Preemption）也可能产生同样的效果：通过在临界区内抢占一个任务，我们会遇到完全相同的竞态条件。在这种情况下，执行抢占的线程本身可能会运行该临界区。

解决方案是识别何时会发生这些同时访问，并使用锁来确保在任何时候只有一个实例可以进入临界区。Linux 内核中有许多友好的原语可以帮助你做到这一点。当然也有一些不那么友好的原语，但我会假装它们不存在。

Linux 内核中的锁定¶

如果我能给你一条关于锁定的建议，那就是：保持简单。

不要轻易引入新的锁。

两种主要的内核锁：自旋锁（Spinlock）和互斥锁（Mutex）¶

内核锁主要有两种类型。基本类型是自旋锁（include/asm/spinlock.h），这是一种非常简单的单一持有者锁：如果你无法获得自旋锁，你会一直尝试（自旋）直到获得为止。自旋锁非常小且快，可以在任何地方使用。

第二种类型是互斥锁（include/linux/mutex.h）：它类似于自旋锁，但你在持有互斥锁时可能会阻塞（休眠）。如果你无法锁定互斥锁，你的任务将挂起，并在互斥锁释放时被唤醒。这意味着 CPU 在你等待时可以做其他事情。在许多情况下你根本不能休眠（参见哪些函数可以安全地从中断中调用？），因此必须改用自旋锁。

这两种锁都不是递归的：参见死锁：简单与高级。

锁与单处理器内核¶

对于编译时没有开启 CONFIG_SMP 且没有开启 CONFIG_PREEMPT 的内核，自旋锁根本不存在。这是一个极好的设计决策：当没有其他人可以同时运行时，就没有理由使用锁。

如果内核编译时没有开启 CONFIG_SMP 但设置了 CONFIG_PREEMPT，那么自旋锁仅会禁用抢占，这足以防止任何竞态。在大多数情况下，我们可以认为抢占等同于 SMP，而无需单独担心它。

你应该始终在开启 CONFIG_SMP 和 CONFIG_PREEMPT 的情况下测试你的锁定代码，即使你没有 SMP 测试机，因为它仍然能捕获某些类型的锁定漏洞。

互斥锁仍然存在，因为它们是用户上下文之间同步所必需的，正如我们将在下面看到的。

仅在用户上下文中锁定¶

如果你有一个数据结构只从用户上下文访问，那么你可以使用简单的互斥锁（include/linux/mutex.h）来保护它。这是最简单的情况：初始化互斥锁，然后调用 mutex_lock_interruptible() 来获取互斥锁，调用 mutex_unlock() 来释放它。还有一个 mutex_lock() 应该避免使用，因为它在收到信号时不会返回。

示例：net/netfilter/nf_sockopt.c 允许通过 nf_register_sockopt() 注册新的 setsockopt() 和 getsockopt() 调用。注册和注销仅在模块加载和卸载（以及启动时，此时没有并发）时进行，并且仅在遇到未知的 setsockopt() 或 getsockopt() 系统调用时才咨询注册列表。nf_sockopt_mutex 非常适合保护这一点，特别是因为 setsockopt 和 getsockopt 调用很可能会休眠。

在用户上下文和软中断（Softirqs）之间锁定¶

如果软中断与用户上下文共享数据，你会面临两个问题。首先，当前用户上下文可能被软中断中断；其次，临界区可能会被另一个 CPU 进入。这就是使用 spin_lock_bh() (include/linux/spinlock.h) 的地方。它会禁用该 CPU 上的软中断，然后获取锁。spin_unlock_bh() 执行相反操作。（“_bh”后缀是对“下半部 Bottom Halves”的历史引用，这是软件中断的旧称。在一个完美的世界里，它其实应该被称为 spin_lock_softirq()）。

注意，你也可以在这里使用 spin_lock_irq() 或 spin_lock_irqsave()，它们也会停止硬件中断：参见硬中断上下文。

这对单处理器（UP）也同样有效：自旋锁会消失，这个宏仅变为 local_bh_disable() (include/linux/interrupt.h)，从而保护你免受软中断运行的影响。

在用户上下文和 Tasklets 之间锁定¶

这与上述情况完全相同，因为 tasklet 实际上是在软中断中运行的。

在用户上下文和定时器（Timers）之间锁定¶

这也与上述情况完全相同，因为定时器实际上是在软中断中运行的。从锁定的角度来看，tasklet 和定时器是相同的。

Tasklets/定时器之间锁定¶

有时一个 tasklet 或定时器可能想与另一个 tasklet 或定时器共享数据。

相同的 Tasklet/定时器¶

由于一个 tasklet 永远不会同时在两个 CPU 上运行，你不需要担心你的 tasklet 是重入的（即同时运行两次），即使在 SMP 上也是如此。

不同的 Tasklets/定时器¶

如果另一个 tasklet/定时器想与你的共享数据，你们都需要使用 spin_lock() 和 spin_unlock() 调用。这里不需要 spin_lock_bh()，因为你已经处于 tasklet 中，同一 CPU 上不会运行其他 tasklet。

软中断之间锁定¶

软中断通常可能想与自身或其他 tasklet/定时器共享数据。

相同的软中断¶

同一个软中断可以在其他 CPU 上运行：你可以使用每 CPU 数组（参见每 CPU 数据）来获得更好的性能。如果你已经用到了软中断，你可能已经足够关心可扩展性能，以证明额外增加的复杂性是值得的。

对于共享数据，你需要使用 spin_lock() 和 spin_unlock()。

不同的软中断¶

对于共享数据，无论是定时器、tasklet、不同的软中断，还是相同或另一个软中断，你都需要使用 spin_lock() 和 spin_unlock()：它们中的任何一个都可能在不同的 CPU 上运行。

硬中断（Hard IRQ）上下文¶

硬件中断通常与 tasklet 或软中断通信。这经常涉及将工作放入队列，然后由软中断取出。

在硬中断和软中断/Tasklets 之间锁定¶

如果硬件中断处理程序与软中断共享数据，你有两个顾虑。首先，软中断处理可能被硬件中断中断；其次，临界区可能会被另一个 CPU 上的硬件中断进入。这就是使用 spin_lock_irq() 的地方。它被定义为在该 CPU 上禁用中断，然后获取锁。spin_unlock_irq() 执行相反操作。

中断处理程序不需要使用 spin_lock_irq()，因为软中断在中断处理程序运行时无法运行：它可以使用稍快一点的 spin_lock()。唯一的例外是如果另一个不同的硬件中断处理程序使用了相同的锁：spin_lock_irq() 将阻止该处理程序中断我们。

这对单处理器（UP）也同样有效：自旋锁消失，宏仅变为 local_irq_disable() (include/asm/smp.h)，从而保护你免受软中断/tasklet/BH 运行的影响。

spin_lock_irqsave() (include/linux/spinlock.h) 是一个变体，它在一个 flags 变量中保存中断是开启还是关闭的，然后将其传递给 spin_unlock_irqrestore()。这意味着同样的代码既可以在硬中断处理程序内部使用（此时中断已经关闭），也可以在软中断中使用（此时需要禁用中断）。

注意，软中断（以及 tasklet 和定时器）在硬件中断返回时运行，因此 spin_lock_irq() 也会停止它们。从这个意义上说，spin_lock_irqsave() 是最通用且功能最强大的锁定函数。

在两个硬中断处理程序之间锁定¶

在两个中断处理程序之间共享数据的情况很少见，但如果你这样做，应该使用 spin_lock_irqsave()：在中断处理程序内部是否禁用所有中断是与架构相关的。

锁定速查表¶

Pete Zaitcev 给出如下总结：

如果你处于进程上下文（任何系统调用）并想锁定其他进程，请使用互斥锁（mutex）。你可以持有互斥锁并休眠（例如 copy_from_user() 或 kmalloc(x, GFP_KERNEL)）。
否则（== 数据可能在中断中被触及），使用 spin_lock_irqsave() 和 spin_unlock_irqrestore()。
避免持有自旋锁超过 5 行代码，且不要跨越任何函数调用（除了像 readb() 这样的访问器）。

最低要求表¶

下表列出了各种上下文之间的最低锁定要求。在某些情况下，相同的上下文一次只能在一个 CPU 上运行，因此该上下文不需要锁定（例如，特定的线程一次只能在一个 CPU 上运行，但如果它需要与其他线程共享数据，则需要锁定）。

记住上面的建议：你总是可以使用 spin_lock_irqsave()，它是所有其他自旋锁原语的超集。

.	中断处理程序 A	中断处理程序 B	软中断 A	软中断 B	Tasklet A	Tasklet B	定时器 A	定时器 B	用户上下文 A	用户上下文 B
中断处理程序 A	无
中断处理程序 B	SLIS	无
软中断 A	SLI	SLI	SL
软中断 B	SLI	SLI	SL	SL
Tasklet A	SLI	SLI	SL	SL	无
Tasklet B	SLI	SLI	SL	SL	SL	无
定时器 A	SLI	SLI	SL	SL	SL	SL	无
定时器 B	SLI	SLI	SL	SL	SL	SL	SL	无
用户上下文 A	SLI	SLI	SLBH	SLBH	SLBH	SLBH	SLBH	SLBH	无
用户上下文 B	SLI	SLI	SLBH	SLBH	SLBH	SLBH	SLBH	SLBH	MLI	无

表：锁定要求表

SLIS	spin_lock_irqsave
SLI	spin_lock_irq
SL	spin_lock
SLBH	spin_lock_bh
MLI	mutex_lock_interruptible

表：锁定要求表图例

trylock 系列函数¶

有些函数仅尝试获取一次锁，并立即返回一个值，告知获取锁是成功还是失败。如果你在其他线程持有锁时不需要访问被该锁保护的数据，就可以使用它们。如果你稍后需要访问被该锁保护的数据，则应在稍后获取该锁。

spin_trylock() 不会自旋，但如果在第一次尝试时获取了自旋锁，则返回非零值，否则返回 0。该函数可以像 spin_lock() 一样在所有上下文中使用：你必须已经禁用了可能中断你并获取自旋锁的上下文。

mutex_trylock() 不会挂起你的任务，但如果能在第一次尝试时锁定互斥锁，则返回非零值，否则返回 0。尽管该函数不会休眠，但仍不能在硬中断或软中断上下文中安全使用。

常见示例¶

让我们来看一个简单的例子：一个编号到名称映射的缓存。该缓存会记录每个对象的使用频率，当缓存满时，会剔除使用最少的对象。

全部处于用户上下文¶

在第一个例子中，我们假设所有操作都处于用户上下文（即来自系统调用），因此我们可以休眠。这意味着我们可以使用互斥锁来保护缓存及其中的所有对象。代码如下：

#include <linux/list.h>
#include <linux/slab.h>
#include <linux/string.h>
#include <linux/mutex.h>
#include <asm/errno.h>

struct object
{
        struct list_head list;
        int id;
        char name[32];
        int popularity;
};

/* Protects the cache, cache_num, and the objects within it */
static DEFINE_MUTEX(cache_lock);
static LIST_HEAD(cache);
static unsigned int cache_num = 0;
#define MAX_CACHE_SIZE 10

/* Must be holding cache_lock */
static struct object *__cache_find(int id)
{
        struct object *i;

        list_for_each_entry(i, &cache, list)
                if (i->id == id) {
                        i->popularity++;
                        return i;
                }
        return NULL;
}

/* Must be holding cache_lock */
static void __cache_delete(struct object *obj)
{
        BUG_ON(!obj);
        list_del(&obj->list);
        kfree(obj);
        cache_num--;
}

/* Must be holding cache_lock */
static void __cache_add(struct object *obj)
{
        list_add(&obj->list, &cache);
        if (++cache_num > MAX_CACHE_SIZE) {
                struct object *i, *outcast = NULL;
                list_for_each_entry(i, &cache, list) {
                        if (!outcast || i->popularity < outcast->popularity)
                                outcast = i;
                }
                __cache_delete(outcast);
        }
}

int cache_add(int id, const char *name)
{
        struct object *obj;

        if ((obj = kmalloc_obj(*obj)) == NULL)
                return -ENOMEM;

        strscpy(obj->name, name, sizeof(obj->name));
        obj->id = id;
        obj->popularity = 0;

        mutex_lock(&cache_lock);
        __cache_add(obj);
        mutex_unlock(&cache_lock);
        return 0;
}

void cache_delete(int id)
{
        mutex_lock(&cache_lock);
        __cache_delete(__cache_find(id));
        mutex_unlock(&cache_lock);
}

int cache_find(int id, char *name)
{
        struct object *obj;
        int ret = -ENOENT;

        mutex_lock(&cache_lock);
        obj = __cache_find(id);
        if (obj) {
                ret = 0;
                strcpy(name, obj->name);
        }
        mutex_unlock(&cache_lock);
        return ret;
}

注意，我们始终确保在添加、删除或查找缓存时持有 cache_lock：缓存基础设施本身和对象的内容都受到该锁的保护。在这种情况下这很容易，因为我们为用户拷贝了数据，从不让他们直接访问对象。

这里有一个细微（且常见）的优化：在 cache_add() 中，我们在获取锁之前先设置对象的字段。这是安全的，因为在我们将其放入缓存之前，没有其他人可以访问它。

从中断上下文访问¶

现在考虑 cache_find() 可以从中断上下文调用的情况：硬件中断或软中断。例如，一个从缓存中删除对象的定时器。

更改如下所示，采用标准补丁格式：- 是被删除的行，+ 是增加的行。

--- cache.c.usercontext 2003-12-09 13:58:54.000000000 +1100
+++ cache.c.interrupt   2003-12-09 14:07:49.000000000 +1100
@@ -12,7 +12,7 @@
         int popularity;
 };

-static DEFINE_MUTEX(cache_lock);
+static DEFINE_SPINLOCK(cache_lock);
 static LIST_HEAD(cache);
 static unsigned int cache_num = 0;
 #define MAX_CACHE_SIZE 10
@@ -55,6 +55,7 @@
 int cache_add(int id, const char *name)
 {
         struct object *obj;
+        unsigned long flags;

         if ((obj = kmalloc_obj(*obj)) == NULL)
                 return -ENOMEM;
@@ -63,30 +64,33 @@
         obj->id = id;
         obj->popularity = 0;

-        mutex_lock(&cache_lock);
+        spin_lock_irqsave(&cache_lock, flags);
         __cache_add(obj);
-        mutex_unlock(&cache_lock);
+        spin_unlock_irqrestore(&cache_lock, flags);
         return 0;
 }

 void cache_delete(int id)
 {
-        mutex_lock(&cache_lock);
+        unsigned long flags;
+
+        spin_lock_irqsave(&cache_lock, flags);
         __cache_delete(__cache_find(id));
-        mutex_unlock(&cache_lock);
+        spin_unlock_irqrestore(&cache_lock, flags);
 }

 int cache_find(int id, char *name)
 {
         struct object *obj;
         int ret = -ENOENT;
+        unsigned long flags;

-        mutex_lock(&cache_lock);
+        spin_lock_irqsave(&cache_lock, flags);
         obj = __cache_find(id);
         if (obj) {
                 ret = 0;
                 strcpy(name, obj->name);
         }
-        mutex_unlock(&cache_lock);
+        spin_unlock_irqrestore(&cache_lock, flags);
         return ret;
 }

注意，如果中断是开启的，spin_lock_irqsave() 会关闭它们，否则什么也不做（如果我们已经在中断处理程序中），因此这些函数可以安全地从任何上下文调用。

遗憾的是，cache_add() 调用了带有 GFP_KERNEL 标志的 kmalloc()，这仅在用户上下文中是合法的。我假设 cache_add() 仍然只在用户上下文中调用，否则这应该成为 cache_add() 的一个参数。

在文件外暴露对象¶

如果我们的对象包含更多信息，仅拷贝信息进出可能就不够了：例如，代码的其他部分可能希望保留指向这些对象的指针，而不是每次都查找 ID。这会产生两个问题。

第一个问题是我们使用 cache_lock 来保护对象：我们需要将其设为非静态，以便代码的其他部分可以使用它。这使锁定变得更棘手，因为锁定逻辑不再集中在一处。

第二个问题是生命周期问题：如果另一个结构体保留了指向对象的指针，它想必期望该指针保持有效。不幸的是，这只有在你持有锁时才能保证，否则有人可能会调用 cache_delete()，更糟的是，添加另一个对象并重用相同的地址。

由于只有一个锁，你不能永远持有它：否则其他人将无法完成任何工作。

解决这个问题的办法是使用引用计数（reference count）：每个持有对象指针的人在第一次获取对象时增加引用计数，并在使用完毕后减少引用计数。将计数减为零的人知道对象不再被使用，可以真正将其删除。

代码如下：

--- cache.c.interrupt   2003-12-09 14:25:43.000000000 +1100
+++ cache.c.refcnt  2003-12-09 14:33:05.000000000 +1100
@@ -7,6 +7,7 @@
 struct object
 {
         struct list_head list;
+        unsigned int refcnt;
         int id;
         char name[32];
         int popularity;
@@ -17,6 +18,35 @@
 static unsigned int cache_num = 0;
 #define MAX_CACHE_SIZE 10

+static void __object_put(struct object *obj)
+{
+        if (--obj->refcnt == 0)
+                kfree(obj);
+}
+
+static void __object_get(struct object *obj)
+{
+        obj->refcnt++;
+}
+
+void object_put(struct object *obj)
+{
+        unsigned long flags;
+
+        spin_lock_irqsave(&cache_lock, flags);
+        __object_put(obj);
+        spin_unlock_irqrestore(&cache_lock, flags);
+}
+
+void object_get(struct object *obj)
+{
+        unsigned long flags;
+
+        spin_lock_irqsave(&cache_lock, flags);
+        __object_get(obj);
+        spin_unlock_irqrestore(&cache_lock, flags);
+}
+
 /* Must be holding cache_lock */
 static struct object *__cache_find(int id)
 {
@@ -35,6 +65,7 @@
 {
         BUG_ON(!obj);
         list_del(&obj->list);
+        __object_put(obj);
         cache_num--;
 }

@@ -63,6 +94,7 @@
         strscpy(obj->name, name, sizeof(obj->name));
         obj->id = id;
         obj->popularity = 0;
+        obj->refcnt = 1; /* The cache holds a reference */

         spin_lock_irqsave(&cache_lock, flags);
         __cache_add(obj);
@@ -79,18 +111,15 @@
         spin_unlock_irqrestore(&cache_lock, flags);
 }

-int cache_find(int id, char *name)
+struct object *cache_find(int id)
 {
         struct object *obj;
-        int ret = -ENOENT;
         unsigned long flags;

         spin_lock_irqsave(&cache_lock, flags);
         obj = __cache_find(id);
-        if (obj) {
-                ret = 0;
-                strcpy(name, obj->name);
-        }
+        if (obj)
+                __object_get(obj);
         spin_unlock_irqrestore(&cache_lock, flags);
-        return ret;
+        return obj;
 }

我们将引用计数封装在标准的“get”和“put”函数中。现在我们可以从 cache_find() 返回对象本身，其优点是用户现在可以持有着对象并休眠（例如，通过 copy_to_user() 将名称拷贝到用户空间）。

另一点需要注意的是，我说过每一个指向对象的指针都应该持有引用：因此第一次插入缓存时，引用计数为 1。在某些版本中，框架本身不持有引用计数，但那会更复杂。

对引用计数使用原子操作¶

在实践中，通常会为 refcnt 使用 atomic_t。在 include/asm/atomic.h 中定义了许多原子操作：这些操作保证对系统中所有 CPU 都是原子可见的，因此不需要锁。在这种情况下，它比使用自旋锁更简单，尽管对于任何非琐碎的操作，使用自旋锁会更清晰。使用 atomic_inc() 和 atomic_dec_and_test() 代替标准的自增和自减运算符，并且不再使用锁来保护引用计数本身。

--- cache.c.refcnt  2003-12-09 15:00:35.000000000 +1100
+++ cache.c.refcnt-atomic   2003-12-11 15:49:42.000000000 +1100
@@ -7,7 +7,7 @@
 struct object
 {
         struct list_head list;
-        unsigned int refcnt;
+        atomic_t refcnt;
         int id;
         char name[32];
         int popularity;
@@ -18,33 +18,15 @@
 static unsigned int cache_num = 0;
 #define MAX_CACHE_SIZE 10

-static void __object_put(struct object *obj)
-{
-        if (--obj->refcnt == 0)
-                kfree(obj);
-}
-
-static void __object_get(struct object *obj)
-{
-        obj->refcnt++;
-}
-
 void object_put(struct object *obj)
 {
-        unsigned long flags;
-
-        spin_lock_irqsave(&cache_lock, flags);
-        __object_put(obj);
-        spin_unlock_irqrestore(&cache_lock, flags);
+        if (atomic_dec_and_test(&obj->refcnt))
+                kfree(obj);
 }

 void object_get(struct object *obj)
 {
-        unsigned long flags;
-
-        spin_lock_irqsave(&cache_lock, flags);
-        __object_get(obj);
-        spin_unlock_irqrestore(&cache_lock, flags);
+        atomic_inc(&obj->refcnt);
 }

 /* Must be holding cache_lock */
@@ -65,7 +47,7 @@
 {
         BUG_ON(!obj);
         list_del(&obj->list);
-        __object_put(obj);
+        object_put(obj);
         cache_num--;
 }

@@ -94,7 +76,7 @@
         strscpy(obj->name, name, sizeof(obj->name));
         obj->id = id;
         obj->popularity = 0;
-        obj->refcnt = 1; /* The cache holds a reference */
+        atomic_set(&obj->refcnt, 1); /* The cache holds a reference */

         spin_lock_irqsave(&cache_lock, flags);
         __cache_add(obj);
@@ -119,7 +101,7 @@
         spin_lock_irqsave(&cache_lock, flags);
         obj = __cache_find(id);
         if (obj)
-                __object_get(obj);
+                object_get(obj);
         spin_unlock_irqrestore(&cache_lock, flags);
         return obj;
 }

保护对象本身¶

在这些例子中，我们假设对象（除了引用计数）一旦创建就永远不会改变。如果我们想允许名称更改，有三种可能性：

你可以将 cache_lock 设为非静态，并要求人们在更改任何对象中的名称之前先获取该锁。
你可以提供一个 cache_obj_rename()，它为调用者获取此锁并更改名称，并告知所有人使用该函数。
你可以让 cache_lock 仅保护缓存本身，并使用另一个锁来保护名称。

从理论上讲，你可以将锁细化到为每个字段、每个对象设置一个锁。在实践中，最常见的变体是：

一个锁保护基础设施（在本例中为 cache 列表）和所有对象。这就是我们到目前为止所做的。
一个锁保护基础设施（包括对象内部的列表指针），而对象内部的另一个锁保护该对象的其余部分。
多个锁保护基础设施（例如，每个哈希链一个锁），可能还有一个单独的每个对象锁。

这是“每个对象一个锁”的实现：

--- cache.c.refcnt-atomic   2003-12-11 15:50:54.000000000 +1100
+++ cache.c.perobjectlock   2003-12-11 17:15:03.000000000 +1100
@@ -6,11 +6,17 @@

 struct object
 {
+        /* These two protected by cache_lock. */
         struct list_head list;
+        int popularity;
+
         atomic_t refcnt;
+
+        /* Doesn't change once created. */
         int id;
+
+        spinlock_t lock; /* Protects the name */
         char name[32];
-        int popularity;
 };

 static DEFINE_SPINLOCK(cache_lock);
@@ -77,6 +84,7 @@
         obj->id = id;
         obj->popularity = 0;
         atomic_set(&obj->refcnt, 1); /* The cache holds a reference */
+        spin_lock_init(&obj->lock);

         spin_lock_irqsave(&cache_lock, flags);
         __cache_add(obj);

注意，我决定使用频率计数（popularity count）应该由 cache_lock 保护，而不是由每个对象锁保护：这是因为它（像对象内部的 struct list_head 一样）在逻辑上是基础设施的一部分。这样，我在 __cache_add() 中寻找最不常用的对象时，就不需要获取每个对象的锁。

我还决定 id 成员是不可更改的，因此在 __cache_find() 中检查 id 时不需要获取每个对象的锁：对象锁仅由想要读取或写入 name 字段的调用者使用。

还要注意，我添加了注释来描述哪些数据由哪些锁保护。这极其重要，因为它描述了代码的运行时行为，而这仅通过阅读代码可能很难获得。正如 Alan Cox 所说，“锁定数据，而不是代码”。

常见问题¶

死锁：简单与高级¶

有一种编码漏洞是某段代码尝试两次获取同一个自旋锁：它将永远自旋，等待锁被释放（Linux 中的自旋锁、读写锁和互斥锁都不是递归的）。这很容易诊断：不是那种需要熬五个通宵对着代码发呆才能解决的问题。

对于稍微复杂一点的情况，想象你有一个由软中断和用户上下文共享的区域。如果你使用 spin_lock() 调用来保护它，那么用户上下文可能会在持有锁时被软中断中断，然后软中断会永远自旋以尝试获取同一个锁。

这两种情况都被称为死锁，如上所述，即使在单个 CPU 上也可能发生（尽管在开启 CONFIG_SMP=n 的内核编译中不会，因为自旋锁会消失。但在第二个例子中你仍会遇到数据损坏）。

这种完全死锁很容易诊断：在 SMP 机器上，看门狗定时器（watchdog timer）或者在编译时开启 DEBUG_SPINLOCK 设置（include/linux/spinlock.h）会在发生死锁时立即显示出来。

更复杂的问题是所谓的“致命拥抱”（deadly embrace），涉及两个或多个锁。假设你有一个哈希表：表中的每个条目都是一个自旋锁，以及一链表哈希对象。在软中断处理程序中，你有时想将一个对象从哈希表的一个位置移动到另一个位置：你获取旧哈希链的自旋锁和新哈希链的自旋锁，从旧链中删除对象，并将其插入新链中。

这里有两个问题。首先，如果你的代码尝试将对象移动到同一个链中，由于它尝试两次加锁，会发生自我死锁。其次，如果另一个 CPU 上的相同软中断正尝试反向移动另一个对象，则可能会发生以下情况：

CPU 1	CPU 2
获取锁 A -> 成功	获取锁 B -> 成功
获取锁 B -> 自旋	获取锁 A -> 自旋

表：后果

这两个 CPU 将永远自旋，等待对方释放锁。这看起来、闻起来、感觉起来都像是一次死机。

防止死锁¶

教科书会告诉你，如果你总是按相同的顺序锁定，就永远不会遇到这种死锁。实践会告诉你，这种方法无法扩展：当我创建一个新锁时，我对内核的理解还不足以弄清楚它在 5000 个锁的层级结构中处于什么位置。

最好的锁是封装好的：它们永远不会在头文件中暴露，也永远不会在调用同一文件之外的非琐碎函数时持有。通读这些代码，你可以看到它永远不会死锁，因为它在持有该锁时从不尝试获取另一个锁。使用你代码的人甚至不需要知道你使用了一个锁。

这里的一个经典问题是当你提供回调（callback）或钩子（hook）时：如果你在持有锁的情况下调用这些钩子，你就面临简单死锁或致命拥抱的风险（谁知道回调会做什么？）。

过度预防死锁¶

死锁确实有问题，但不如数据损坏严重。如果代码获取读锁，搜索列表，没找到想要的东西，释放读锁，然后获取写锁并插入对象，这就存在竞态条件。

竞态定时器：一种内核消遣¶

定时器会在竞态方面产生自己特有的问题。考虑一个对象集合（列表、哈希等），其中每个对象都有一个即将销毁它的定时器。

如果你想销毁整个集合（比如在卸载模块时），你可能会这样做：

/* THIS CODE BAD BAD BAD BAD: IF IT WAS ANY WORSE IT WOULD USE
   HUNGARIAN NOTATION */
spin_lock_bh(&list_lock);

while (list) {
        struct foo *next = list->next;
        timer_delete(&list->timer);
        kfree(list);
        list = next;
}

spin_unlock_bh(&list_lock);

迟早这会在 SMP 上崩溃，因为定时器可能恰好在 spin_lock_bh() 之前触发，它只会在我们执行 spin_unlock_bh() 之后才获得锁，然后尝试释放该元素（而该元素已经被释放了！）。

可以通过检查 timer_delete() 的结果来避免这种情况：如果返回 1，则定时器已被删除。如果返回 0，则意味着（在这种情况下）定时器当前正在运行，因此我们可以这样做：

retry:
        spin_lock_bh(&list_lock);

        while (list) {
                struct foo *next = list->next;
                if (!timer_delete(&list->timer)) {
                        /* Give timer a chance to delete this */
                        spin_unlock_bh(&list_lock);
                        goto retry;
                }
                kfree(list);
                list = next;
        }

        spin_unlock_bh(&list_lock);

另一个常见问题是删除会自动重启的定时器（通过在定时器函数末尾调用 add_timer()）。因为这是一种相当常见且容易产生竞态的情况，你应该使用 timer_delete_sync() (include/linux/timer.h) 来处理这种情况。

在释放定时器之前，应调用 timer_shutdown() 或 timer_shutdown_sync()，这会防止它被重新激活。随后任何尝试重新激活定时器的行为都会被核心代码静默忽略。

锁定速度¶

在考虑带有锁定的代码速度时，主要有三件事需要担心。首先是并发性：当某人持有锁时，有多少其他任务会处于等待状态。其次是实际获取和释放一个无竞争锁所花费的时间。第三是使用更少或更聪明的锁。我假设该锁被相当频繁地使用：否则，你就不会关心效率了。

并发性取决于锁通常持有的时间：你应该根据需要持有锁，但持有的时间越短越好。在缓存示例中，我们总是在不持有锁的情况下创建对象，只有当我们准备将其插入列表时才获取锁。

获取时间取决于锁定操作对流水线造成的损害（流水线停顿），以及此 CPU 是不是最后一个获取锁的 CPU（即锁对该 CPU 而言是否为缓存热 cache-hot）：在 CPU 较多的机器上，这种可能性会迅速下降。考虑一个 700MHz 的 Intel Pentium III：执行一条指令大约需要 0.7ns，原子自增大约需要 58ns，对于此 CPU 为缓存热的锁需要 160ns，而从另一个 CPU 传输缓存行（cacheline）还需要额外的 170 到 360ns。（这些数据来自 Paul McKenney 的 Linux Journal RCU 文章）。

这两个目标是冲突的：持有一个短时间的锁可以通过将锁拆分为多个部分来实现（例如在我们最后的“每个对象一个锁”的例子中），但这会增加锁获取的次数，结果往往比持有单个锁还要慢。这是提倡锁定简单性的另一个原因。

第三个顾虑将在下面解决：有一些方法可以减少需要进行的锁定操作量。

读写锁变体¶

自旋锁和互斥锁都有读写变体：rwlock_t 和 struct rw_semaphore。这些变体将用户分为两类：读者和写者。如果你只是读取数据，可以获取读锁；但要写入数据，你需要获取写锁。许多人可以同时持有读锁，但写者必须是唯一的持有者。

如果你的代码可以清晰地划分为读者/写者（如我们的缓存代码），且读锁被读者持有相当长的时间，使用这些锁可能会有所帮助。不过它们比普通锁稍微慢一点，所以实践中 rwlock_t 通常不值得使用。

避免锁：读取-拷贝-更新（RCU）¶

有一种特殊的读写锁定方法叫做读取-拷贝-更新（Read Copy Update，RCU）。使用 RCU，读者可以完全避免获取锁：由于我们期望缓存被读取的频率远高于更新的频率（否则缓存就是在浪费时间），它是进行这种优化的候选方案。

我们如何摆脱读锁？摆脱读锁意味着写者可能会在读者还在读取时更改列表。这其实很简单：如果写者非常小心地添加元素，我们可以在添加元素的同时读取单向链表。例如，将 new 添加到名为 list 的单向链表中：

new->next = list->next;
wmb();
list->next = new;

wmb() 是一个写内存屏障（write memory barrier）。它确保在第二个操作（将新元素放入列表）之前，第一个操作（设置新元素的 next 指针）已经完成并对所有 CPU 可见。这一点很重要，因为现代编译器和 CPU 除非被告知，否则都可以重新排序指令：我们希望读者要么根本看不到新元素，要么看到新元素的 next 指针正确地指向列表的其余部分。

幸运的是，有一个函数可以为标准的 struct list_head 列表执行此操作：list_add_rcu() (include/linux/list.h)。

从链表中删除一个元素甚至更简单：我们将指向旧元素的指针替换为其后继元素的指针，读者要么会看到它，要么会跳过它。

list->next = old->next;

有一个执行此操作的 list_del_rcu() (include/linux/list.h)（普通版本会“毒化”旧对象，这不是我们想要的）。

读者也必须小心：某些 CPU 可能会通过 next 指针提前开始读取下一个元素的内容，但当 next 指针在它们下方发生变化时，它们并没有意识到预取的内容是错误的。再一次，有一个 list_for_each_entry_rcu() (include/linux/list.h) 可以帮助你。当然，写者只需使用 list_for_each_entry()，因为不可能有两个写者同时存在。

我们最后的难题是：我们什么时候可以真正销毁被删除的元素？记住，可能有一个读者现在正遍历到列表中的这个元素：如果我们释放这个元素且 next 指针发生变化，读者将跳进垃圾堆并崩溃。我们需要等到我们确定在删除元素时正在遍历列表的所有读者都已经结束。我们使用 call_rcu() 来注册一个回调函数，一旦所有现有的读者都结束，该函数将真正销毁对象。或者，可以使用 synchronize_rcu() 来阻塞直到所有现有的读者都结束。

但是 RCU 是如何知道读者什么时候结束的呢？方法如下：首先，读者总是在 rcu_read_lock()/rcu_read_unlock() 对之间遍历列表：它们只是简单地禁用抢占，因此读者在读取列表时不会进入休眠。

RCU 然后等待，直到每个其他 CPU 都至少休眠过一次：由于读者无法休眠，我们知道在删除期间正在遍历列表的任何读者都已经结束，从而触发回调。真正的 RCU 代码比这更优化，但这是基本思想。

--- cache.c.perobjectlock   2003-12-11 17:15:03.000000000 +1100
+++ cache.c.rcupdate    2003-12-11 17:55:14.000000000 +1100
@@ -1,15 +1,18 @@
 #include <linux/list.h>
 #include <linux/slab.h>
 #include <linux/string.h>
+#include <linux/rcupdate.h>
 #include <linux/mutex.h>
 #include <asm/errno.h>

 struct object
 {
-        /* These two protected by cache_lock. */
+        /* This is protected by RCU */
         struct list_head list;
         int popularity;

+        struct rcu_head rcu;
+
         atomic_t refcnt;

         /* Doesn't change once created. */
@@ -40,7 +43,7 @@
 {
         struct object *i;

-        list_for_each_entry(i, &cache, list) {
+        list_for_each_entry_rcu(i, &cache, list) {
                 if (i->id == id) {
                         i->popularity++;
                         return i;
@@ -49,19 +52,25 @@
         return NULL;
 }

+/* Final discard done once we know no readers are looking. */
+static void cache_delete_rcu(void *arg)
+{
+        object_put(arg);
+}
+
 /* Must be holding cache_lock */
 static void __cache_delete(struct object *obj)
 {
         BUG_ON(!obj);
-        list_del(&obj->list);
-        object_put(obj);
+        list_del_rcu(&obj->list);
         cache_num--;
+        call_rcu(&obj->rcu, cache_delete_rcu);
 }

 /* Must be holding cache_lock */
 static void __cache_add(struct object *obj)
 {
-        list_add(&obj->list, &cache);
+        list_add_rcu(&obj->list, &cache);
         if (++cache_num > MAX_CACHE_SIZE) {
                 struct object *i, *outcast = NULL;
                 list_for_each_entry(i, &cache, list) {
@@ -104,12 +114,11 @@
 struct object *cache_find(int id)
 {
         struct object *obj;
-        unsigned long flags;

-        spin_lock_irqsave(&cache_lock, flags);
+        rcu_read_lock();
         obj = __cache_find(id);
         if (obj)
                 object_get(obj);
-        spin_unlock_irqrestore(&cache_lock, flags);
+        rcu_read_unlock();
         return obj;
 }

注意，读者将在 __cache_find() 中修改 popularity 成员，而现在它并不持有锁。一种解决方案是将其设为 atomic_t，但对于这种用途，我们并不真正关心竞态：一个近似的结果就足够了，所以我没有改动它。

结果是 cache_find() 不需要与任何其他函数同步，因此在 SMP 上的速度几乎与在 UP 上一样快。

这里还有进一步优化的可能：记住我们原始的缓存代码，那里没有引用计数，调用者在任何时候使用对象时只需持有锁。这仍然是可能的：如果你持有锁，就没有人可以删除该对象，因此你不需要获取和释放引用计数。

现在，由于 RCU 中的“读锁”仅仅是禁用抢占，因此在调用 cache_find() 和 object_put() 之间始终禁用抢占的调用者不需要真正获取和释放引用计数：我们可以通过将 __cache_find() 设为非静态来暴露它，这样的调用者只需直接调用该函数即可。

这里的好处是不会对引用计数执行写入操作：对象没有任何改动，由于缓存的作用，这在 SMP 机器上要快得多。

每 CPU 数据¶

另一种被广泛使用的避免锁定的技术是为每个 CPU 复制信息。例如，如果你想记录某种常见情况的计数，可以使用自旋锁和一个计数器。这很好也很简单。

如果那太慢了（通常不会，但如果你有一台非常大的机器进行测试并能证明它慢），你可以转而为每个 CPU 使用一个计数器，这样它们都不需要排他锁。参见 DEFINE_PER_CPU(), get_cpu_var() 和 put_cpu_var() (include/linux/percpu.h)。

对于简单的每 CPU 计数器特别有用的是 local_t 类型，以及 cpu_local_inc() 和相关函数，它们在某些架构上比简单代码更有效率 (include/asm/local.h)。

注意，如果不引入更多的锁，就没有简单、可靠的方法来获取此类计数器的准确值。对于某些用途来说，这并不是问题。

主要由中断处理程序使用的数据¶

如果数据总是从同一个中断处理程序中访问，你根本不需要锁：内核已经保证中断处理程序不会在多个 CPU 上同时运行。

Manfred Spraul 指出，即使数据偶尔在用户上下文或软中断/tasklet 中被访问，你仍然可以这样做。中断处理程序不使用锁，而所有其他访问都按如下方式进行：

mutex_lock(&lock);
disable_irq(irq);
...
enable_irq(irq);
mutex_unlock(&lock);

disable_irq() 会阻止中断处理程序运行（如果它当前正在其他 CPU 上运行，则等待它结束）。自旋锁防止任何其他访问同时发生。自然地，这比单纯调用 spin_lock_irq() 要慢，因此只有当此类访问极少发生时才有意义。

哪些函数可以安全地从中断中调用？¶

内核中的许多函数直接或间接地进入休眠（即调用 schedule()）：在持有自旋锁或禁用抢占的情况下，你绝不能调用它们。这也意味着你必须处于用户上下文中：从中断中调用它们是非法的。

一些会休眠的函数¶

下面列出了最常见的函数，但你通常必须阅读代码才能确定其他调用是否安全。如果调用它的其他所有人都能休眠，你可能也需要能够休眠。特别是，注册和注销函数通常期望从用户上下文调用，并且可以休眠。

访问用户空间：
- copy_from_user()
- copy_to_user()
- get_user()
- put_user()
kmalloc(GP_KERNEL) <kmalloc>`
mutex_lock_interruptible() 和 mutex_lock()

虽然有一个 mutex_trylock() 不会休眠，但仍不能在中断上下文中使用，因为其实现对此并不安全。mutex_unlock() 也从不休眠，但同样不能在中断上下文中使用，因为互斥锁必须由获取它的同一个任务释放。

一些不会休眠的函数¶

有些函数可以安全地从任何上下文调用，或在持有几乎任何锁的情况下调用。

Mutex API 参考¶

mutex_init¶

mutex_init (mutex)

初始化互斥锁

参数

mutex: 待初始化的互斥锁

描述

将互斥锁初始化为未锁定状态。

不允许初始化一个已经锁定的互斥锁。

mutex_init_with_key¶

mutex_init_with_key (mutex, key)

使用给定的 lockdep key 初始化互斥锁

参数

mutex: 待初始化的互斥锁
key: 要与互斥锁关联的 lockdep key

描述

将互斥锁初始化为未锁定状态。

不允许初始化一个已经锁定的互斥锁。

bool mutex_is_locked(struct mutex *lock)¶: 互斥锁是否已锁定

参数

struct mutex *lock: 要查询的互斥锁

描述

如果互斥锁已锁定则返回 true，未锁定则返回 false。

void mutex_lock(struct mutex *lock)¶: 获取互斥锁

参数

struct mutex *lock: 要获取的互斥锁

描述

为此任务排他性地锁定互斥锁。如果互斥锁当前不可用，它将休眠直到获取为止。

互斥锁随后必须由获取它的同一个任务释放。不允许递归锁定。任务在未解锁互斥锁之前不得退出。此外，互斥锁所在的内核内存不能在互斥锁仍处于锁定状态时被释放。互斥锁必须在锁定前先初始化（或静态定义）。不允许使用 memset() 将互斥锁清零。

（CONFIG_DEBUG_MUTEXES .config 选项会开启调试检查，以强制执行这些限制并进行死锁调试）

此函数类似于（但不等同于）down()。

void mutex_unlock(struct mutex *lock)¶: 释放互斥锁

参数

struct mutex *lock: 要释放的互斥锁

描述

解锁先前由该任务锁定的互斥锁。

此函数不得在中断上下文中使用。不允许解锁一个未锁定的互斥锁。

调用者必须确保互斥锁在该函数返回之前一直有效 —— mutex_unlock() 不能直接用于释放一个对象从而让另一个并发任务可以释放该对象。在这方面，互斥锁不同于自旋锁和引用计数。

此函数类似于（但不等同于）up()。

void ww_mutex_unlock(struct ww_mutex *lock)¶: 释放 w/w 互斥锁

参数

struct ww_mutex *lock: 要释放的互斥锁

描述

解锁先前由该任务通过任何 ww_mutex_lock* 函数（带有或不带有获取上下文）锁定的互斥锁。在释放获取上下文后禁止释放锁。

此函数不得在中断上下文中使用。不允许解锁一个已解锁的互斥锁。

int ww_mutex_trylock(struct ww_mutex *ww, struct ww_acquire_ctx *ww_ctx)¶: 尝试使用可选的获取上下文获取 w/w 互斥锁

参数

struct ww_mutex *ww: 要锁定的互斥锁
struct ww_acquire_ctx *ww_ctx: 可选的 w/w 获取上下文

描述

尝试使用可选的获取上下文对互斥锁进行 trylock；无法进行死锁检测。如果成功获取互斥锁则返回 1，否则返回 0。

与 ww_mutex_lock 不同，不执行死锁处理。但是，如果指定了 ctx，则在调用 ww_mutex_trylock 时可能会发生 -EALREADY 处理。

使用此函数获取的互斥锁必须使用 ww_mutex_unlock 释放。

int mutex_lock_interruptible(struct mutex *lock)¶: 获取互斥锁，可被信号中断。

参数

struct mutex *lock: 待获取的互斥锁。

描述

像 mutex_lock() 一样锁定互斥锁。如果进程在休眠时收到信号，此函数将返回且不获取互斥锁。

上下文

进程上下文。

返回

如果成功获取锁则返回 0，如果收到信号则返回 -EINTR。

int mutex_lock_killable(struct mutex *lock)¶: 获取互斥锁，可被致命信号中断。

参数

struct mutex *lock: 待获取的互斥锁。

描述

像 mutex_lock() 一样锁定互斥锁。如果进程在休眠时收到对当前进程致命的信号，此函数将返回且不获取互斥锁。

上下文

进程上下文。

返回

如果成功获取锁则返回 0，如果收到致命信号则返回 -EINTR。

void mutex_lock_io(struct mutex *lock)¶: 获取互斥锁并将进程标记为等待 I/O

参数

struct mutex *lock: 待获取的互斥锁。

描述

像 mutex_lock() 一样锁定互斥锁。当任务等待此互斥锁时，它将被调度器计入 IO 等待状态。

上下文

进程上下文。

int mutex_trylock(struct mutex *lock)¶: 尝试获取互斥锁，不等待

参数

struct mutex *lock: 要获取的互斥锁

描述

尝试原子性地获取互斥锁。如果成功获取则返回 1，若有竞争则返回 0。

注意

此函数遵循 spin_trylock() 的约定，因此它与 down_trylock() 的返回值相反！在将信号量用户转换为互斥锁时请务必小心。

此函数不得在中断上下文中使用。互斥锁必须由获取它的同一个任务释放。

int atomic_dec_and_mutex_lock(atomic_t *cnt, struct mutex *lock)¶: 如果减到 0 则返回并持有互斥锁

参数

atomic_t *cnt: 我们要递减的原子变量
struct mutex *lock: 如果减到 0 则持有该互斥锁返回

描述

如果减到 0 则返回 true 并持有锁，否则返回 false

Futex API 参考¶

struct futex_hash_bucket *__futex_hash(union futex_key *key, struct futex_private_hash *fph, struct futex_private_hash **fph_p)¶: 返回哈希桶

参数

union futex_key *key: 指向计算哈希值所针对的 futex key 的指针
struct futex_private_hash *fph: 指向私有哈希的指针（如果已知）
struct futex_private_hash **fph_p: 指向私有哈希指针的指针；输出设置时使用的私有哈希。

描述

我们对从 get_futex_key 返回的 key 进行哈希处理（见下文），并返回相应的哈希桶。如果 FUTEX 是 PROCESS_PRIVATE，则返回一个每进程哈希桶（来自私有哈希，如果存在）。否则返回全局哈希中的哈希桶。

struct hrtimer_sleeper *futex_setup_timer(ktime_t *time, struct hrtimer_sleeper *timeout, int flags, u64 range_ns)¶: 设置休眠的 hrtimer。

参数

ktime_t *time: 指向给定超时值的指针
struct hrtimer_sleeper *timeout: 要设置的 hrtimer_sleeper 结构体
int flags: futex 标志
u64 range_ns: 可选的纳秒范围

返回

初始化后的 hrtimer_sleeper 结构体，如果没有给出超时值则为 NULL

int get_futex_key(u32 __user *uaddr, unsigned int flags, union futex_key *key, enum futex_access rw)¶: 获取作为 futex key 的参数

参数

u32 __user *uaddr: futex 的虚拟地址
unsigned int flags: FLAGS_*
union futex_key *key: 存储结果的地址。
enum futex_access rw: 映射需要读/写权限（取值：FUTEX_READ, FUTEX_WRITE）

返回

负的错误代码或 0

描述

成功时，键值存储在 key 中。

对于共享映射（当 fshared 为真时），键为

( inode->i_sequence, 映射内的页面偏移, 页面内偏移 )

[ 另请参阅 get_inode_sequence_number() ]

对于私有映射（或当 !fshared 时），键为

( current->mm, 地址, 0 )

这允许（在适用时跨进程）标识 futex，而无需在 FUTEX_WAIT 期间保持页面锁定（pinned）。

lock_page() 可能会进入睡眠，调用者不应持有自旋锁。

int fault_in_user_writeable(u32 __user *uaddr)¶: 对用户地址触发缺页并验证读写（RW）访问权限

参数

u32 __user *uaddr: 指向触发缺页的用户空间地址的指针

描述

用于修复刚才在对 uaddr 进行原子写访问时触发的缺页的慢速路径。

我们没有对用户地址进行非破坏性写入的通用实现。既然我们已知是在禁用原子缺页处理的部分触发了缺页，我们可以直接调用 get_user_pages() 来避免 #PF（缺页异常）开销。

struct futex_q *futex_top_waiter(struct futex_hash_bucket *hb, union futex_key *key)¶: 返回 futex 上优先级最高的等待者

参数

struct futex_hash_bucket *hb: futex_q 所在的哈希桶
union futex_key *key: futex 键（用以将其与其他 futex 的 futex_q 区分开）

描述

必须在持有 hb 锁的情况下调用。

void wait_for_owner_exiting(int ret, struct task_struct *exiting)¶: 阻塞直到所有者退出

参数

int ret: 所有者当前的 futex 锁定状态
struct task_struct *exiting: 指向正在退出的任务的指针

描述

调用者必须持有对 exiting 的引用计数。

void __futex_unqueue(struct futex_q *q)¶: 从 futex_hash_bucket 中移除 futex_q

参数

struct futex_q *q: 要移出队列的 futex_q

描述

q->lock_ptr 必须不能为 NULL，且必须由调用者持有。

int futex_unqueue(struct futex_q *q)¶: 从 futex_hash_bucket 中移除 futex_q

参数

struct futex_q *q: 要移出队列的 futex_q

描述

调用者不得持有 q->lock_ptr。调用 futex_unqueue() 必须与之前的一次 futex_queue() 调用成对出现。

返回

1 - 如果 futex_q 仍在队列中（且我们成功将其移除）；
0 - 如果 futex_q 已经被唤醒线程移除

void futex_exit_recursive(struct task_struct *tsk)¶: 将任务的 futex 状态设置为 FUTEX_STATE_DEAD

参数

struct task_struct *tsk: 要设置状态的任务

描述

以无锁方式设置任务的 futex 退出状态。futex 等待者代码在任务退出时会观察该状态，并循环等待直到任务真正完成 futex 清理。最坏的情况是等待者在循环中运行，直到该状态变得可见。

此函数从 make_task_dead() 中的递归故障处理路径调用。

这是尽力而为的操作。要么 futex 退出代码已经运行，要么没有。如果 futex 上设置了 OWNER_DIED 位，则等待者可以接管它。如果没有设置，问题将推回给用户空间。如果 futex 退出代码尚未运行，则已入队的等待者可能会永久阻塞，但这无法处理。

struct futex_q¶: 哈希后的 futex 队列条目，每个等待任务一个

定义:

struct futex_q {
    struct plist_node list;
    struct task_struct *task;
    spinlock_t *lock_ptr;
    futex_wake_fn *wake;
    void *wake_data;
    union futex_key key;
    struct futex_pi_state *pi_state;
    struct rt_mutex_waiter *rt_waiter;
    union futex_key *requeue_pi_key;
    u32 bitset;
    atomic_t requeue_state;
    struct futex_private_hash *drop_fph;
#ifdef CONFIG_PREEMPT_RT;
    struct rcuwait requeue_wait;
#endif;
};

成员

list: 在该 futex 上等待的任务按优先级排序的列表
任务: 在该 futex 上等待的任务
lock_ptr: 哈希桶锁
唤醒: 此队列的唤醒处理程序
wake_data: 与唤醒处理程序关联的数据
key: futex 哈希所依据的键
pi_state: 可选的优先级继承状态
rt_waiter: 用于 requeue_pi 的 rt_waiter 存储空间
requeue_pi_key: requeue_pi 的目标 futex 键
位集: 用于可选的位掩码唤醒的位集
requeue_state: futex_requeue_pi() 的状态字段
drop_fph: 设置后，等待者应释放额外的私有哈希引用
requeue_wait: futex_requeue_pi() 的 RCU 等待（仅限 RT）

描述

我们使用这个哈希等待队列而不是普通的 wait_queue_entry_t，这样我们可以只唤醒相关的任务（哈希队列可能是共享的）。

futex_q 有一个已唤醒状态，就像任务有 TASK_RUNNING 一样。当 plist_node_empty(q->list) || q->lock_ptr == 0 时，它被视为已唤醒。唤醒顺序总是先使第一个条件为真，然后再使第二个条件为真。

PI futex 通常在通过 rt_mutex 代码从哈希列表中移除之前被唤醒。参见 futex_unqueue_pi()。

int futex_match(union futex_key *key1, union futex_key *key2)¶: 检查两个 futex 键是否相等

参数

union futex_key *key1: 指向 key1 的指针
union futex_key *key2: 指向 key2 的指针

描述

如果两个 futex_key 相等则返回 1，否则返回 0。

void futex_queue(struct futex_q *q, struct futex_hash_bucket *hb, struct task_struct *task)¶: 将 futex_q 排入 futex_hash_bucket 队列

参数

struct futex_q *q: 要排队的 futex_q
struct futex_hash_bucket *hb: 目标哈希桶
struct task_struct *task: 排队此 futex 的任务

描述

hb->lock 必须由调用者持有，并在此处释放。调用 futex_queue() 通常与恰好一次 futex_unqueue() 调用配对。例外情况涉及 PI 相关的操作，这些操作可能会使用 futex_unqueue_pi()，或者如果出队操作作为唤醒过程的一部分完成且出队状态隐含在唤醒任务的状态中，则不调用任何函数（示例参见 futex_wait_requeue_pi()）。

注意 task 可能为 NULL，用于 futex 的异步使用。

struct futex_vector¶: futex_waitv() 的辅助结构体

定义:

struct futex_vector {
    struct futex_waitv w;
    struct futex_q q;
};

成员

w: 用户空间提供的数据
q: 内核侧数据

描述

用于构建包含 futex_waitv() 所需所有数据数组的结构体

int futex_lock_pi_atomic(u32 __user *uaddr, struct futex_hash_bucket *hb, union futex_key *key, struct futex_pi_state **ps, struct task_struct *task, struct task_struct **exiting, int set_waiters)¶: 获取 pi 感知型 futex 所需的原子工作

参数

u32 __user *uaddr: pi futex 的用户地址
struct futex_hash_bucket *hb: pi futex 哈希桶
union futex_key *key: 与 uaddr 和 hb 关联的 futex 键
struct futex_pi_state **ps: 存储查找结果的 pi_state 指针
struct task_struct *task: 要执行原子锁定工作的任务。除了在 requeue pi 的情况下，这通常是 “current”。
struct task_struct **exiting: 用于存储正在退出中的所有者任务指针的指针
int set_waiters: 强制设置 FUTEX_WAITERS 位 (1) 与否 (0)

返回

0 - 准备好等待；
1 - 已获取锁；
<0 - 错误

描述

hb->lock 必须由调用者持有。

仅当返回值为 -EBUSY 时才会设置 exiting。如果是这种情况，返回时将持有该退出中任务的引用计数，调用者在等待退出完成后需要释放它。

int fixup_pi_owner(u32 __user *uaddr, struct futex_q *q, int locked)¶: 锁定后的 pi_state 管理和边界情况处理

参数

u32 __user *uaddr: futex 的用户地址
struct futex_q *q: futex_q（包含 pi_state 和对 rt_mutex 的访问）
int locked: 尝试获取 rt_mutex 是否成功（1）或失败（0）

描述

在尝试锁定 rt_mutex 后，调用此函数以清理 pi_state 所有者并处理可能允许我们获取锁的竞态条件。必须在持有 hb 锁的情况下调用。

返回

1 - 成功，已获取锁；
0 - 成功，未获取锁；
<0 - 出错（-EFAULT）

void requeue_futex(struct futex_q *q, struct futex_hash_bucket *hb1, struct futex_hash_bucket *hb2, union futex_key *key2)¶: 将 futex_q 从一个 hb 重排队到另一个 hb

参数

struct futex_q *q: 要重排队的 futex_q
struct futex_hash_bucket *hb1: 源哈希桶
struct futex_hash_bucket *hb2: 目标哈希桶
union futex_key *key2: 重排队后的 futex_q 的新键

void requeue_pi_wake_futex(struct futex_q *q, union futex_key *key, struct futex_hash_bucket *hb)¶: 唤醒在重排队期间获得锁的任务

参数

struct futex_q *q: futex_q
union futex_key *key: 重排队目标 futex 的键
struct futex_hash_bucket *hb: 重排队目标 futex 的哈希桶

描述

在 futex_requeue 且 requeue_pi=1 的过程中，如果目标 futex 无竞争或通过锁窃取（lock steal），则有可能获取它。

将 q::key 设置为重排队目标 futex 键，以便等待者能在正确的 futex 上检测到唤醒。
将 q 从哈希桶中出队。
将 q::rt_waiter 设置为 NULL，以便唤醒的任务能检测到原子锁获取。
将 q->lock_ptr 设置为重排队目标 hb->lock，以防等待者必须修复 pi 状态。
完成重排队状态，使等待者可以继续运行。在此之后，如果不需要修复 pi 状态，等待者任务可以立即从系统调用返回。
唤醒等待者任务。

必须在同时持有 q->lock_ptr 和 hb->lock 的情况下调用。

int futex_proxy_trylock_atomic(u32 __user *pifutex, struct futex_hash_bucket *hb1, struct futex_hash_bucket *hb2, union futex_key *key1, union futex_key *key2, struct futex_pi_state **ps, struct task_struct **exiting, int set_waiters)¶: 尝试为最高优先级等待者执行原子锁定

参数

u32 __user *pifutex: 目标 futex 的用户地址
struct futex_hash_bucket *hb1: 源 futex 哈希桶，必须由调用者锁定
struct futex_hash_bucket *hb2: 目标 futex 哈希桶，必须由调用者锁定
union futex_key *key1: 源 futex 键
union futex_key *key2: 目标 futex 键
struct futex_pi_state **ps: 用于存储 pi_state 指针的地址
struct task_struct **exiting: 用于存储正在退出中的所有者任务指针的指针
int set_waiters: 强制设置 FUTEX_WAITERS 位 (1) 与否 (0)

描述

如果可以原子地执行，尝试代表最高优先级等待者获取锁。如果成功，唤醒该等待者。如果调用者指定了 set_waiters，则指示 futex_lock_pi_atomic() 强制设置 FUTEX_WAITERS 位。hb1 和 hb2 必须由调用者持有。

仅当返回值为 -EBUSY 时才会设置 exiting。如果是这种情况，返回时将持有该退出中任务的引用计数，调用者在等待退出完成后需要释放它。

返回

0 - 无法原子地获取锁；
>0 - 已获取锁，返回值为 top_waiter 的 vpid
<0 - 错误

int futex_requeue(u32 __user *uaddr1, unsigned int flags1, u32 __user *uaddr2, unsigned int flags2, int nr_wake, int nr_requeue, u32 *cmpval, int requeue_pi)¶: 将等待者从 uaddr1 重排队到 uaddr2

参数

u32 __user *uaddr1: 源 futex 用户地址
unsigned int flags1: futex 标志（FLAGS_SHARED 等）
u32 __user *uaddr2: 目标 futex 用户地址
unsigned int flags2: futex 标志（FLAGS_SHARED 等）
int nr_wake: 要唤醒的等待者数量（对于 requeue_pi 必须为 1）
int nr_requeue: 要重排队的等待者数量 (0-INT_MAX)
u32 *cmpval: uaddr1 的预期值（或 NULL）
int requeue_pi: 如果我们正尝试从非 pi futex 重排队到 pi futex（不支持 pi 到 pi 的重排队）

描述

将 uaddr1 上的等待者重排队到 uaddr2。在 requeue_pi 情况下，尝试代表最高优先级等待者原子地获取 uaddr2。

返回

>=0 - 成功时，重排队或被唤醒的任务数量；
<0 - 出错

int handle_early_requeue_pi_wakeup(struct futex_hash_bucket *hb, struct futex_q *q, struct hrtimer_sleeper *timeout)¶: 处理在初始 futex 上的早期唤醒

参数

struct futex_hash_bucket *hb: futex_q 最初排入的哈希桶
struct futex_q *q: 在等待重排队期间被唤醒的 futex_q
struct hrtimer_sleeper *timeout: 与等待关联的超时时间（若无则为 NULL）

描述

确定早期唤醒的原因。

返回

-EWOULDBLOCK 或 -ETIMEDOUT 或 -ERESTARTNOINTR

int futex_wait_requeue_pi(u32 __user *uaddr, unsigned int flags, u32 val, ktime_t *abs_time, u32 bitset, u32 __user *uaddr2)¶: 在 uaddr 上等待并获取 uaddr2

参数

u32 __user *uaddr: 最初等待的 futex（非 pi）
unsigned int flags: futex 标志（FLAGS_SHARED, FLAGS_CLOCKRT 等），它们必须是相同类型，不能从私有重排队到共享等。
u32 val: uaddr 的预期值
ktime_t *abs_time: 绝对超时时间
u32 bitset: 由用户空间设置的 32 位唤醒位集，默认为全部
u32 __user *uaddr2: 在返回用户空间之前我们要获取的 pi futex

描述

调用者将在 uaddr 上等待，并将由 futex_requeue() 重排队到 uaddr2，uaddr2 必须感知 PI 且与 uaddr 不同。正常唤醒将在 uaddr2 上发生，并在返回用户空间之前完成 rt_mutex 的获取。这确保了 rt_mutex 在有等待者时始终维持一个所有者；如果没有所有者，PI 逻辑将不知道在需要时该提升/降低哪个任务的优先级。

我们在入队时在 futex_wait_queue() 中调用 schedule，并由于以下原因返回：1) 在 futex_requeue() 原子获取锁后在 uaddr2 上被唤醒 2) 重排队后在 uaddr2 上被唤醒 3) 信号 4) 超时

如果是 3，清理并返回 -ERESTARTNOINTR。

如果是 2，我们随后可能会在尝试获取 rt_mutex 时阻塞，并由于以下原因返回：5) 成功锁定 6) 信号 7) 超时 8) 其他获取锁失败

如果是 6，返回 -EWOULDBLOCK（重新启动系统调用也会得到同样结果）。

如果是 4 或 7，我们清理并返回 -ETIMEDOUT。

返回

0 - 成功；
<0 - 出错

void futex_do_wait(struct futex_q *q, struct hrtimer_sleeper *timeout)¶: 等待唤醒、超时或信号

参数

struct futex_q *q: 排队等待的 futex_q
struct hrtimer_sleeper *timeout: 准备好的 hrtimer_sleeper，若无超时则为 null

int futex_unqueue_multiple(struct futex_vector *v, int count)¶: 从哈希桶中移除多个 futex

参数

struct futex_vector *v: 要出队的 futex 列表
int count: 列表中的 futex 数量

描述

出队一系列 futex 的辅助函数。这不会失败。

返回

>=0 - 最后一个被唤醒的 futex 的索引；
-1
- 没有 futex 被唤醒

int futex_wait_multiple_setup(struct futex_vector *vs, int count, int *woken)¶: 准备等待并入队多个 futex

参数

struct futex_vector *vs: 要等待的 futex 列表
int count: 列表的大小
int *woken: 最后一个被唤醒的 futex 的索引（如果有）。用于通知调用者它可以将此索引返回给用户空间（返回参数）

描述

在一个步骤中准备多个 futex 并将其入队。如果 futex 列表无效或任何 futex 已被唤醒，此操作可能会失败。成功时，任务准备好进入可中断睡眠。

返回

1 - 其中一个 futex 被另一个线程唤醒
0 - 成功
<0 - -EFAULT, -EWOULDBLOCK 或 -EINVAL

void futex_sleep_multiple(struct futex_vector *vs, unsigned int count, struct hrtimer_sleeper *to)¶: 检查睡眠条件并进入睡眠

参数

struct futex_vector *vs: 等待的 futex 列表
unsigned int count: vs 的长度
struct hrtimer_sleeper *to: 超时时间

描述

当且仅当超时未过期且列表中的 futex 均未被唤醒时进入睡眠。

int futex_wait_multiple(struct futex_vector *vs, unsigned int count, struct hrtimer_sleeper *to)¶: 准备等待并入队多个 futex

参数

struct futex_vector *vs: 要等待的 futex 列表
unsigned int count: 对象数量
struct hrtimer_sleeper *to: 放弃等待并返回用户空间之前的超时时间

描述

FUTEX_WAIT_MULTIPLE futex 操作的入口点，此函数在一组 futex 上睡眠，并在第一个 futex 被唤醒或超时后返回。

返回

>=0 - 被唤醒的 futex 的提示（索引）
<0 - 出错

int futex_wait_setup(u32 __user *uaddr, u32 val, unsigned int flags, struct futex_q *q, union futex_key *key2, struct task_struct *task)¶: 准备在 futex 上等待

参数

u32 __user *uaddr: futex 用户空间地址
u32 val: 预期值
unsigned int flags: futex 标志（FLAGS_SHARED 等）
struct futex_q *q: 关联的 futex_q
union futex_key *key2: 如果用于 requeue PI，则为第二个 futex_key
struct task_struct *task: 排队此 futex 的任务

描述

设置 futex_q 并定位 hash_bucket。获取 futex 值并将其与预期值进行比较。在内部处理原子缺页异常。成功时持有 hb 锁返回，失败时则不锁定。

返回

0 - uaddr 包含 val 且 hb 已被锁定；
<0 - 出错且 hb 未锁定。可能的原因：uaddr 无法
读取、不包含预期值或未正确对齐。

延伸阅读¶

Documentation/locking/spinlocks.rst：内核源码中 Linus Torvalds 的自旋锁教程。
Unix Systems for Modern Architectures: Symmetric Multiprocessing and Caching for Kernel Programmers

Curt Schimmel 编写的非常出色的内核级锁定入门书（并非专门针对 Linux，但几乎所有内容都适用）。书很贵，但为了理解 SMP 锁定，每一分钱都值得。[ISBN: 0201633388]

致谢¶

感谢 Telsa Gwynne 编写 DocBooking、整理和添加样式。

感谢 Martin Pool, Philipp Rumpf, Stephen Rothwell, Paul Mackerras, Ruedi Aschwanden, Alan Cox, Manfred Spraul, Tim Waugh, Pete Zaitcev, James Morris, Robert Love, Paul McKenney, John Ashby 的校对、更正、激烈讨论、评论。

感谢密谋集团（cabal）未对本文档施加任何影响。

术语表¶

抢占（preemption）

在 2.5 版本之前，或者当未设置 CONFIG_PREEMPT 时，内核中处于用户上下文的进程不会互相抢占（即，除非你主动放弃或发生中断，否则你将一直持有 CPU）。随着 2.5.4 中加入 CONFIG_PREEMPT，情况发生了变化：在用户上下文中，更高优先级的任务可以“插队”：自旋锁被改为禁用抢占，即使在单处理器（UP）上也是如此。

bh

下半部（Bottom Half）：由于历史原因，现在名称中带有 ‘_bh’ 的函数通常指代任何软件中断（软中断），例如 spin_lock_bh() 会阻塞当前 CPU 上的任何软中断。下半部机制已被弃用，最终将被 tasklets 取代。在任何时候只能运行一个下半部。

硬件中断 / 硬件 IRQ

硬件中断请求。in_hardirq() 在硬件中断处理程序中返回 true。

中断上下文

非用户上下文：正在处理硬件中断或软件中断。通过 in_interrupt() 宏返回 true 来标识。

SMP

对称多处理器（Symmetric Multi-Processor）：为多 CPU 机器编译的内核（CONFIG_SMP=y）。

软件中断 / 软中断（softirq）

软件中断处理程序。in_hardirq() 返回 false；in_softirq() 返回 true。Tasklets 和 softirqs 都属于“软件中断”范畴。

严格来说，softirq 是最多 32 个枚举软件中断之一，可以同时在多个 CPU 上运行。有时也用来泛指 tasklet（即所有软件中断）。

tasklet

一种可以动态注册的软件中断，保证一次只能在一个 CPU 上运行。

timer

一种可以动态注册的软件中断，在给定时间（或接近该时间）运行。运行时，它就像一个 tasklet（事实上，它们是从 TIMER_SOFTIRQ 中调用的）。

UP

单处理器（Uni-Processor）：非 SMP 环境（CONFIG_SMP=n）。

用户上下文

内核代表特定进程（即系统调用或陷阱）或内核线程执行。你可以通过 current 宏获知是哪个进程。不要与用户空间混淆。可以被软件或硬件中断所中断。

用户空间

在内核之外执行自身代码的进程。