Linux 内核 futex 机制与同步原语实战：从系统调用到高性能锁的工程真相

在多线程编程中，"锁"是最基础的同步原语之一。你可能每天都在使用 pthread_mutex_lock()，但你是否想过——为什么它如此之快？当没有竞争时，一次互斥锁的加锁和解锁根本不需要陷入内核。这个"魔法"的核心正是本文的主角：futex（Fast Userspace muTEX）。

本文将深入 futex 的内核实现机制、优先级继承协议、健壮锁（Robust futexes），以及如何基于 futex 构建高性能的自定义同步原语。

---

一、为什么需要 futex？

在 futex 出现之前，Linux 的线程同步主要依赖以下几种方式：

1. System V 信号量（semop）：每次操作都需要陷入内核，系统调用开销巨大。
2. 管道/事件通知：使用 read()/write() 管道来实现通知，怪异且低效。
3. 信号（SIGUSR1）：信号处理程序的执行上下文受限，无法安全地加锁。

核心矛盾在于：大多数锁操作没有竞争。如果每次加锁都进入内核，99% 的系统调用都是白白浪费。futex 的设计哲学就是：在用户态完成无竞争的加锁/解锁，只在真正需要等待或唤醒时才进入内核。

这个设计原则后来被广泛应用于各种同步机制：

• pthread_mutex_t：futex 作为底层等待机制
• pthread_cond_t：条件变量通过 futex 实现阻塞和唤醒
• pthread_rwlock_t：读写锁
• sem_t：信号量

---

二、futex 系统调用解析

Linux 提供了 futex(2) 系统调用，原型如下：

#include <linux/futex.h>

long futex(uint32_t *uaddr, int futex_op, uint32_t val,
           const struct timespec *timeout, uint32_t *uaddr2, uint32_t val3);

关键参数：

• uaddr：指向用户态 32 位整数的指针（即 futex word），也称为"futex 变量"。
• futex_op：操作类型（如 FUTEX_WAIT、FUTEX_WAKE 等）。
• val：操作相关的值。
• timeout：超时时间（用于 FUTEX_WAIT）。

futex 核心在于：uaddr 指向的内存是用户态与内核共享的 32 位变量。线程在用户态通过原子指令修改变量值，只在必要时才通过系统调用进入内核等待或唤醒其他线程。

2.1 核心操作

操作	说明
FUTEX_WAIT	如果 *uaddr == val，让调用线程阻塞等待
FUTEX_WAKE	唤醒最多 val 个在 uaddr 上等待的线程
FUTEX_WAIT_BITSET	带位掩码的等待，用于条件变量的精确唤醒
FUTEX_WAKE_BITSET	带位掩码的唤醒
FUTEX_REQUEUE	将等待者从一个 futex 重排队到另一个
FUTEX_CMP_REQUEUE	带比较的 requeue，避免竞态
FUTEX_WAIT_REQUEUE_PI	用于优先级继承的 requeue 等待
FUTEX_LOCK_PI	优先级继承锁的加锁操作
FUTEX_UNLOCK_PI	优先级继承锁的解锁操作
FUTEX_TRYLOCK_PI	尝试获取优先级继承锁

2.2 内联汇编实现无竞争加锁

在没有竞争的情况下，一次用户态的 lock cmpxchg 就完成了加锁：

#include <stdatomic.h>
#include <linux/futex.h>
#include <sys/syscall.h>
#include <unistd.h>
#include <stdint.h>
#include <time.h>
#include <stdio.h>

// 直接使用 futex 构建一个简单的互斥锁
typedef struct {
    uint32_t futex_word; // 0=未锁定, 1=锁定无等待者, 2=锁定有等待者
} raw_mutex_t;

static long sys_futex(uint32_t *uaddr, int op, uint32_t val,
                      const struct timespec *timeout) {
    return syscall(SYS_futex, uaddr, op, val, timeout, NULL, 0);
}

void raw_mutex_init(raw_mutex_t *m) {
    atomic_store(&m->futex_word, 0);
}

void raw_mutex_lock(raw_mutex_t *m) {
    // 第一阶段：无竞争路径（用户态原子操作）
    uint32_t expected = 0;
    if (atomic_compare_exchange_strong_explicit(
            &m->futex_word, &expected, 1,
            memory_order_acquire, memory_order_relaxed)) {
        // 成功获取锁，无竞争，直接返回
        return;
    }

    // 第二阶段：竞争路径
    // 将 futex_word 设为 2（表示有等待者）
    while (atomic_exchange_explicit(&m->futex_word, 2,
                                    memory_order_acquire) != 0) {
        // 内核等待：只有当 futex_word 为 2 时才阻塞
        sys_futex(&m->futex_word, FUTEX_WAIT, 2, NULL);
        // 被唤醒后重新尝试 CAS
        expected = 0;
        if (!atomic_compare_exchange_strong_explicit(
                &m->futex_word, &expected, 2,
                memory_order_acquire, memory_order_relaxed)) {
            // 被唤醒后锁已被其他线程抢走
            // expected 现在为 1 或 2，继续循环
        } else {
            break; // 成功获取锁
        }
    }
}

void raw_mutex_unlock(raw_mutex_t *m) {
    uint32_t prev = atomic_fetch_sub_explicit(&m->futex_word, 1,
                                              memory_order_release);
    if (prev != 1) {
        // 之前是 2，说明有等待者，需要唤醒
        atomic_store(&m->futex_word, 0);
        sys_futex(&m->futex_word, FUTEX_WAKE, 1, NULL);
    }
}

上面的实现虽然简化，但揭示了 futex 的核心思路：

1. 无竞争时：用户态 CAS 操作完成加锁（~10ns 量级）
2. 竞争时：futex_word 保持为 2，线程进入 FUTEX_WAIT，进入内核等待队列
3. 解锁时：检查 futex_word，如果有等待者（值为 2），则先设为 0 再唤醒

与传统的 sem_wait() 相比（每次操作都陷入内核，~100ns-1μs），futex 在无竞争时有 10-100 倍的性能优势。

---

三、futex 内核实现：哈希桶与等待队列

futex 的内核实现位于 kernel/futex.c（5.x 内核约 4000 行代码）。核心数据结构是一个全局的哈希表。

3.1 核心数据结构

struct futex_q {
    struct plist_node list;           // 挂在哈希桶的链表上
    struct task_struct *task;         // 等待的任务
    union futex_key key;              // 用户态地址对应的 key
    struct futex_pi_state *pi_state;  // PI 状态（优先级继承）
    struct rt_mutex_waiter *rt_wait;  // rt_mutex 等待对象
    union futex_key *requeue_pi_key;  // requeue 目标 key
    uint32_t bitset;                  // 位掩码
};

struct futex_hash_bucket {
    spinlock_t lock;
    struct plist_head chain;
} ____cacheline_aligned_in_smp;

static struct {
    struct futex_hash_bucket *queues;
    unsigned long            queues_count;
} futex_queues ____cacheline_aligned_in_smp;

关键设计：

• 全局 256 个（可配置到更多）哈希桶，按用户态地址哈希。
• 每个桶有自旋锁保护。
• 挂在桶上的等待队列是一个优先队列（priority-sorted linked list）。
• 等待者按任务优先级排序，唤醒时优先唤醒高优先级任务。

3.2 FUTEX_WAIT 的执行路径

用户态调用 futex(uaddr, FUTEX_WAIT, val)
  -> sys_futex() [kernel/futex.c]
  -> do_futex()
       |
       +-- futex_wait_setup()
       |     +-- futex_key 计算（根据 uaddr 计算哈希）
       |     +-- futex_hashbucket_lock()（获取桶自旋锁）
       |     +-- 检查 *uaddr != val 则返回 -EAGAIN
       |
       +-- futex_wait()
             +-- 将 futex_q 插入等待队列
             +-- 设置任务状态为 TASK_INTERRUPTIBLE
             +-- schedule() 让出 CPU
             +-- 被唤醒后检查是否超时/被信号中断

3.3 FUTEX_WAKE 的执行路径

用户态调用 futex(uaddr, FUTEX_WAKE, nr)
  -> do_futex() -> futex_wake()
       +-- futex_hashbucket_lock()
       +-- 遍历链表，按优先级选择前 nr 个等待者
       +-- wake_up_state() 唤醒选中的任务
       +-- futex_hashbucket_unlock()

---

四、优先级继承（Priority Inheritance）：PI Futexes

在实时系统中，经典问题是优先级反转（Priority Inversion）：

高优先级任务 H（优先级 90）
    -> 等待锁
中优先级任务 M（优先级 50）-> 被 H 阻塞
低优先级任务 L（优先级 10）-> 持锁运行
    -> 问题：如果 M 开始运行，L 被 M 抢占 -> H 被 M 间接阻塞

优先级继承协议（PIP）要求：当高优先级任务被阻塞在锁上时，锁的持有者应临时继承高优先级任务的优先级，避免被中优先级任务抢占。

4.1 内核实现

PI futex 的核心数据结构是 struct futex_pi_state，它将用户态的 futex 与内核的 rt_mutex（实时互斥锁）绑定：

struct futex_pi_state {
    struct rt_mutex pi_mutex;       // 内核 rt_mutex
    struct task_struct *owner;      // 当前所有者
    atomic_t refcount;
    union futex_key key;
};

执行流程：

// 简化版 FUTEX_LOCK_PI 路径
futex_lock_pi()
  +-- 尝试获取锁（CAS user space from 0 to caller_tid）
  +-- 失败时 -> futex_lock_pi_atomic()
  |     +-- 识别锁持有者（通过 futex word 中存储的 tid）
  |     +-- 设置 FUTEX_WAITERS 标志
  |     +-- 构造 futex_pi_state，关联 rt_mutex
  +-- rt_mutex_start_proxy_lock()
        +-- 将锁持有者的调度优先级提升到与等待者相同
        +-- 内核的 boost 机制通过 rt_mutex 实现

4.2 用户态使用 PI 互斥锁

#include <pthread.h>

int main() {
    pthread_mutex_t mutex;
    pthread_mutexattr_t attr;

    pthread_mutexattr_init(&attr);
    // 设置协议为 PTHREAD_PRIO_INHERIT
    pthread_mutexattr_setprotocol(&attr, PTHREAD_PRIO_INHERIT);
    // 设置类型为普通互斥
    pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_NORMAL);
    pthread_mutexattr_setrobust(&attr, PTHREAD_MUTEX_ROBUST);

    pthread_mutex_init(&mutex, &attr);
    // 现在 pthread_mutex_lock() 会在内部使用 FUTEX_LOCK_PI
    pthread_mutex_lock(&mutex);
    // 临界区
    pthread_mutex_unlock(&mutex);

    pthread_mutex_destroy(&mutex);
    return 0;
}

在 Linux 上验证：

# 查看线程的调度优先级
cat /proc/<TID>/sched | grep -E "prio|policy"

---

五、Robust Futexes：处理异常终止

在多进程共享互斥锁的场景下（PTHREAD_PROCESS_SHARED），一个进程异常终止会导致锁永远持有。Robust futexes 正是为解决这个问题设计的。

5.1 约定

• futex word 的低 30 位是锁状态（包含 owner tid）。
• 第 30 位 FUTEX_WAITERS：有等待者存在。
• 第 31 位 FUTEX_OWNER_DIED：上次持有者已死。

5.2 解锁时的检查

void robust_mutex_unlock(pthread_mutex_t *m) {
    int ret = pthread_mutex_unlock(m);
    if (ret == EOWNERDEAD) {
        // 上次所有者已死，需要调用 consistent 恢复共享状态
        pthread_mutex_consistent(m);
        // 然后正常 unlock
        pthread_mutex_unlock(m);
    }
}

5.3 内核检测机制

当进程退出时，内核遍历该进程注册的所有 robust futex 列表：

do_exit()
  -> exit_robust_list()
       +-- 遍历 robust_list
       +-- 将 futex word 设为 (tid | FUTEX_OWNER_DIED)
       +-- futex_wake() 唤醒等待者

---

六、Requeue 艺术：条件变量的实现

futex 的 FUTEX_CMP_REQUEUE 操作是条件变量（pthread_cond）实现的关键——它能原子地将等待者从一个 futex 搬运到另一个，避免唤醒丢失。

6.1 pthread_cond_wait 的实现逻辑

// 简化版：pthread_cond_wait 的底层实现
int pthread_cond_wait_impl(cond_t *cond, mutex_t *mutex) {
    // 1. 把 mutex 的锁释放
    atomic_store(&mutex->futex_word, 0);
    if (有等待者) {
        sys_futex(&mutex->futex_word, FUTEX_WAKE, INT_MAX, NULL);
    }

    // 2. 原子地将自己加入 cond 的等待队列
    //    同时验证 mutex 仍然未被其他线程修改
    sys_futex(&cond->seq, FUTEX_CMP_REQUEUE,
               /*nr_wake=*/1,          // 唤醒 1 个 cond 上的线程
               /*nr_requeue=*/INT_MAX, // 其余全部搬到 mutex
               /*target=*/&mutex->futex_word,
               /*val=*/cond->序列值);
    // 这一步确保：从加入 cond 等待队列到重新获取 mutex
    // 这个过程中如果有 signal/broadcast，不会丢失

    // 3. 重新获取 mutex
    raw_mutex_lock(mutex);
}

6.2 FUTEX_REQUEUE vs FUTEX_CMP_REQUEUE 的必要性

为什么需要比较？考虑这个时间线：

线程 A：pthread_mutex_unlock() -> mutex word = 0 -> FUTEX_WAKE
线程 B：pthread_cond_wait() -> 还未进入 FUTEX_WAIT 内核
线程 A：再次 lock -> mutex word = 1（lock 成功）
线程 B：进入 FUTEX_WAIT -> 永久阻塞！

FUTEX_CMP_REQUEUE 的 val 参数要求内核在 requeue 之前再次检查 futex word 的值。如果不一致（说明有人抢占了），可以返回错误，让调用者重试，避免丢失唤醒。

---

七、现代演进：Priority-Weighted Queues 与 RT Mutex 嵌套

自 Linux 5.14 起，futex 内部使用一种增强的优先级排序结构，不仅考虑任务的静态优先级，还考虑：

1. 等待时间：长时间等待的任务防止饥饿
2. CPU 亲和性：优先唤醒与上次运行在同一 CPU 上的任务（缓存友好）
3. 嵌套 rt_mutex：在复杂的锁层次中正确传播优先级提升

7.1 The Futex2 扩展提案

近年来社区提出了多种 futex 扩展：

• rwmutex futexes：在一套操作中实现读写锁（减少系统调用次数）
• 等待多个 futex：类似 epoll 的批量等待接口
• NUMA-aware futex：在 NUMA 系统上根据内存位置优化唤醒策略

// 概念性示例：假设的 futex_wait_multiple
struct futex_wait_entry {
    uint32_t *uaddr;
    uint32_t   val;
    uint32_t   flags;
};

int futex_wait_multiple(struct futex_wait_entry *entries, int nr,
                        const struct timespec *timeout);

这对于需要同时等待多个条件的高性能场景（如组合锁、多源事件队列）有重要意义。

---

八、实战：构建一个高性能读写锁

基于 futex 系统调用，我们可以直接实现一个读写锁：

#include <stdatomic.h>
#include <linux/futex.h>
#include <sys/syscall.h>
#include <unistd.h>
#include <stdint.h>
#include <stdio.h>

typedef struct {
    _Atomic uint32_t state;
    // state 位定义：
    // [31:1] = 读者计数器（右移 1 位）
    // [0] = 写者标记（1=有写者等待或持有）
} futex_rwlock_t;

#define FUTEX_RWLOCK_WRLOCK   0x00000001u
#define FUTEX_RWLOCK_RDCOUNT  0xFFFFFFF2u

static long futex_call(uint32_t *uaddr, int op, uint32_t val,
                       const struct timespec *timeout) {
    return syscall(SYS_futex, uaddr, op, val, timeout, NULL, 0);
}

void futex_rwlock_init(futex_rwlock_t *rw) {
    atomic_store(&rw->state, 0);
}

void futex_rwlock_rdlock(futex_rwlock_t *rw) {
    for (;;) {
        uint32_t expected = atomic_load_explicit(&rw->state,
                                                 memory_order_relaxed);
        if ((expected & FUTEX_RWLOCK_WRLOCK) == 0) {
            // 尝试增加读者计数
            if (atomic_compare_exchange_weak_explicit(
                    &rw->state, &expected, expected + 2,
                    memory_order_acquire, memory_order_relaxed)) {
                return;
            }
        } else {
            // 有写者，阻塞等待
            futex_call(&rw->state, FUTEX_WAIT, expected, NULL);
        }
    }
}

void futex_rwlock_wrlock(futex_rwlock_t *rw) {
    for (;;) {
        uint32_t expected = atomic_load_explicit(&rw->state,
                                                 memory_order_relaxed);
        if (expected == 0) {
            // 无人持有，尝试直接获取写锁
            if (atomic_compare_exchange_weak_explicit(
                    &rw->state, &expected, FUTEX_RWLOCK_WRLOCK,
                    memory_order_acquire, memory_order_relaxed)) {
                return;
            }
        } else if ((expected & FUTEX_RWLOCK_WRLOCK) == 0) {
            // 设置写者标记，表示写者等待
            if (atomic_compare_exchange_weak_explicit(
                    &rw->state, &expected, expected | FUTEX_RWLOCK_WRLOCK,
                    memory_order_relaxed, memory_order_relaxed)) {
                // 等待所有读者退出
                uint32_t target = expected | FUTEX_RWLOCK_WRLOCK;
                while (atomic_load_explicit(&rw->state,
                                            memory_order_relaxed) != FUTEX_RWLOCK_WRLOCK) {
                    futex_call(&rw->state, FUTEX_WAIT, target, NULL);
                }
                return;
            }
        } else {
            // 另一个写者，阻塞等待
            futex_call(&rw->state, FUTEX_WAIT, expected, NULL);
        }
    }
}

void futex_rwlock_unlock(futex_rwlock_t *rw) {
    uint32_t prev = atomic_fetch_add_explicit(&rw->state, -2,
                                              memory_order_release);
    if (prev == FUTEX_RWLOCK_WRLOCK) {
        // 之前是写锁
        atomic_store(&rw->state, 0);
        futex_call(&rw->state, FUTEX_WAKE, INT_MAX, NULL);
    } else if ((prev & FUTEX_RWLOCK_RDCOUNT) == 2 && (prev & FUTEX_RWLOCK_WRLOCK)) {
        // 最后一个读者且有等待的写者
        atomic_store(&rw->state, 0);
        futex_call(&rw->state, FUTEX_WAKE, INT_MAX, NULL);
    }
}

注意这个实现使用了"写者偏好"策略——设置写者标记后会阻止新读者获取锁，防止写者饥饿。这在高写入频率场景下是正确的，但对于读者远多于写者的场景，可能需要不同的策略。

---

九、调试与观测

9.1 使用 strace 观察 futex 调用

# 追踪所有 futex 系统调用
strace -e trace=futex -p <PID>

# 示例输出：
# futex(0x7f8a4c000e68, FUTEX_WAKE_PRIVATE, 1) = 0
# futex(0x7f8a4c000e68, FUTEX_WAIT_PRIVATE, 2, NULL) = 0

9.2 查看进程的 robust list

cat /proc/<PID>/robust_list

9.3 futex 延迟分析

# 使用 bpftrace 跟踪 futex 延迟
bpftrace -e '
tracepoint:syscalls:sys_enter_futex {
    @start[tid] = nsecs;
}
tracepoint:syscalls:sys_exit_futex /@start[tid]/ {
    @us = hist((nsecs - @start[tid]) / 1000);
    delete(@start[tid]);
}
'

9.4 性能基准

典型结果（无竞争）：futex-based mutex 约 20-30ns，仅比纯原子操作慢 2-3 倍。相比之下，传统的 System V 信号量每次操作约 200-500ns。

---

十、总结：futex 设计的启示

futex 的设计哲学是操作系统与用户态库协作的典范：

1. 用户态快速路径：无竞争时不陷入内核，原子指令完成同步。
2. 内核慢速路径：竞争时进入内核，利用内核的调度器和等待队列做高效的线程管理。
3. 优先级集成：通过 rt_mutex 与内核调度器深度集成，支持实时调度策略。
4. 健壮性设计：robust futexes 为进程崩溃场景提供了恢复机制。

从 futex 的设计中，我们可以获得以下工程启示：

• 分层设计：快速路径与慢速路径分离是高性能系统的不二法则。
• 只在必要时进入内核：系统调用是重要的资源，应当被珍惜使用。
• 与调度器协同：锁不仅是互斥工具，也应与优先级调度、CPU 亲和性等系统特性协同工作。

现代运行时（Go、Rust、Zig）在各自的同步原语实现中，都借鉴了类似 futex 的用户态-内核协作模式。理解 futex 不仅是操作系统知识的积累，更是理解所有现代同步原语底层逻辑的钥匙。