Linux►Kernel

kernel/futex 快速用户态互斥（Futex）机制与原理详解

kernel/futex 快速用户态互斥（Futex）机制与原理详解

源码路径：rk3588/kernel-6.1/kernel/futex/
内核版本：Linux 6.1（RK3588 平台）
平台：RK3588（4×Cortex-A76 + 4×Cortex-A55 big.LITTLE，ARM64）

该目录实现 Linux 内核 Futex（Fast Userspace Mutex，快速用户态互斥） 机制，是 pthread 互斥锁、条件变量、信号量等用户态同步原语的底层支撑。核心思想：无竞争时在用户态通过原子操作完成加锁/解锁，仅在需要阻塞/唤醒时才进入内核。

目录

一、源码目录结构

1.1 编译依赖（Makefile）

1
obj-y += core.o syscalls.o pi.o requeue.o waitwake.o
配置项 说明
CONFIG_FUTEX 启用 futex 子系统（默认 y）
CONFIG_FUTEX_PI 优先级继承 futex（默认 y，依赖 RT_MUTEXES）
CONFIG_FAIL_FUTEX 故障注入测试（debug）
CONFIG_PREEMPT_RT RT 内核下 hash bucket 锁变为 sleeping spinlock

1.2 源文件

文件 行数 功能
core.c ~1159 哈希表初始化、futex key 计算、用户态原子操作封装、robust 清理
waitwake.c ~708 futex_wait / futex_wake / futex_wake_op / futex_waitv 多路等待
requeue.c ~897 futex_requeue / requeue_pi / wait_requeue_pi
pi.c ~1233 优先级继承锁：lock_pi / unlock_pi / pi_state 管理
syscalls.c ~379 系统调用入口：futex / futex_waitv / robust_list
futex.h ~294 内部头文件：数据结构、函数声明

二、整体架构

2.1 设计哲学

1
2
3
4
5
6
7
8
9
10
┌─────────────────────────────────────────────────────────┐
│  用户态（glibc/pthread/bionic）                          │
│  原子 CAS 操作 futex 变量（userspace fast path）         │
│  仅在需要阻塞/唤醒时调用 futex(2) 进入内核               │
└───────────────────────────┬─────────────────────────────┘
                            │ syscall
┌───────────────────────────▼─────────────────────────────┐
│  kernel/futex/                                           │
│  哈希桶管理等待队列 → 阻塞/唤醒任务 → PI/robust 扩展     │
└─────────────────────────────────────────────────────────┘

Fast path（用户态）：lock() 尝试 CAS(0→1)，成功则直接返回，零 syscall 开销。

Slow path（内核态）：CAS 失败（已有持有者或 WAITERS 位），调用 FUTEX_WAIT 阻塞；解锁方调用 FUTEX_WAKE 唤醒。

2.2 模块职责划分

模块 职责
core.c 基础设施：hash 表、key 计算、cmpxchg、robust 退出清理
waitwake.c 核心 wait/wake 语义、bitset 过滤、wake_op 原子操作
requeue.c 将一个 futex 上的 waiter 批量转移到另一个 futex（条件变量基础）
pi.c 优先级继承，防止优先级反转（PI mutex）
syscalls.c 系统调用分发、robust_list、多路 waitv

2.3 与用户态库的映射

用户态 API 内核 futex 操作
pthread_mutex_lock() 用户态 CAS + FUTEX_WAIT / FUTEX_LOCK_PI
pthread_mutex_unlock() 用户态 store + FUTEX_WAKE / FUTEX_UNLOCK_PI
pthread_cond_wait() FUTEX_WAIT + FUTEX_REQUEUE
pthread_cond_signal() FUTEX_WAKE
pthread_cond_broadcast() FUTEX_WAKE (nr=INT_MAX) + FUTEX_REQUEUE
sem_wait() / sem_post() FUTEX_WAIT / FUTEX_WAKE

三、核心数据结构

3.1 union futex_key — 哈希键

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
union futex_key {
    struct {                    // PROCESS_SHARED（文件映射）
        u64 i_seq;              // inode 序列号
        unsigned long pgoff;    // 页索引
        unsigned int offset;    // 页内偏移
    } shared;
    struct {                    // PROCESS_PRIVATE
        struct mm_struct *mm;   // 地址空间
        unsigned long address;  // 虚拟地址（页对齐）
        unsigned int offset;    // 页内偏移
    } private;
    struct {                    // 哈希比较用
        u64 ptr;
        unsigned long word;
        unsigned int offset;
    } both;
};
  • Private futex：key = (mm, address, offset)，无需 pin 页面，速度快
  • Shared futex：key = (inode->i_sequence, page->index, offset)，跨进程共享

3.2 struct futex_hash_bucket — 哈希桶

1
2
3
4
5
struct futex_hash_bucket {
    atomic_t waiters;           // 等待者计数（SMP 优化用）
    spinlock_t lock;
    struct plist_head chain;    // 优先级排序的等待队列
} ____cacheline_aligned_in_smp;

全局哈希表 futex_queues[]，大小为 256 × num_possible_cpus() 的 2 的幂。

3.3 struct futex_q — 等待队列条目

1
2
3
4
5
6
7
8
9
10
struct futex_q {
    struct plist_node list;         // 按优先级排序
    struct task_struct *task;
    spinlock_t *lock_ptr;           // 所属 hash bucket 的锁
    union futex_key key;
    struct futex_pi_state *pi_state; // PI futex 专用
    struct rt_mutex_waiter *rt_waiter;
    u32 bitset;                     // bitset 唤醒过滤
    atomic_t requeue_state;         // requeue_pi 状态机
};

3.4 struct futex_pi_state — 优先级继承状态

1
2
3
4
5
6
7
struct futex_pi_state {
    struct list_head list;          // 挂在 owner->pi_state_list
    struct rt_mutex_base pi_mutex;  // 底层 RT mutex
    struct task_struct *owner;
    refcount_t refcount;
    union futex_key key;
};

四、Futex Key 与哈希

4.1 get_futex_key()

1
2
3
4
5
6
7
8
9
10
11
get_futex_key(uaddr, fshared, key, rw)
  │
  ├── Private (!fshared):
  │     key = (current->mm, page_align(uaddr), offset)
  │     无需 pin 页面，直接返回
  │
  └── Shared (fshared):
        get_user_pages_fast(uaddr)
        → 获取 page → compound_head → mapping
        → key = (inode->i_sequence, page->index, offset)
        → 处理 COW / swap / truncate 等边界情况

4.2 futex_hash()

1
2
3
4
5
struct futex_hash_bucket *futex_hash(union futex_key *key)
{
    u32 hash = jhash2((u32 *)key, ..., key->both.offset);
    return &futex_queues[hash & (futex_hashsize - 1)];
}

4.3 哈希表初始化

1
2
3
4
5
6
7
8
9
10
11
static int __init futex_init(void)
{
    futex_hashsize = roundup_pow_of_two(256 * num_possible_cpus());
    futex_queues = alloc_large_system_hash(...);
    for (i = 0; i < futex_hashsize; i++) {
        atomic_set(&futex_queues[i].waiters, 0);
        plist_head_init(&futex_queues[i].chain);
        spin_lock_init(&futex_queues[i].lock);
    }
}
core_initcall(futex_init);

五、基本 Wait/Wake 流程

5.1 FUTEX_WAIT

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
用户态：
  val = *futex;
  if (cond(val))
      futex(FUTEX_WAIT, futex, val);

内核 futex_wait():
  futex_wait_setup(uaddr, val, flags, &q, &hb)
    → get_futex_key()           计算 key
    → futex_q_lock(q)           获取 hash bucket 锁
    → futex_get_value_locked()  再次读取 *uaddr
    → if (*uaddr != val) return -EWOULDBLOCK  // 值已变，无需等待

  futex_queue(q, hb)            入队，释放 bucket 锁
  futex_wait_queue(hb, q, to)   schedule() 阻塞

  被唤醒后：
    futex_unqueue(q)            检查是否成功出队
    → 0: 正常唤醒
    → -ETIMEDOUT: 超时
    → -ERESTARTSYS: 信号中断

5.2 FUTEX_WAKE

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
用户态：
  *futex = newval;
  futex(FUTEX_WAKE, futex, nr_wake);

内核 futex_wake():
  get_futex_key(uaddr) → futex_hash(&key)

  // 优化：无 waiter 时跳过加锁
  if (!futex_hb_waiters_pending(hb))
      return 0;

  spin_lock(&hb->lock)
  plist_for_each_entry_safe(this, next, &hb->chain, list) {
      if (futex_match(&this->key, &key)) {
          if (this->bitset & bitset)    // bitset 过滤
              futex_wake_mark(&wake_q, this);
          if (++ret >= nr_wake) break;
      }
  }
  spin_unlock(&hb->lock)
  wake_up_q(&wake_q)            批量唤醒

5.3 FUTEX_WAKE_OP

在一个 syscall 中同时对两个 futex 执行 wake + 原子操作 + 条件 wake：

1
2
3
4
futex_wake_op(uaddr1, uaddr2, nr_wake, nr_wake2, op)
  → 对 uaddr1 执行 futex_wake(nr_wake)
  → 对 uaddr2 执行 arch_futex_atomic_op_inuser(op)  // 原子加减/位操作
  → 若 atomic_op 条件满足，对 uaddr2 执行 futex_wake(nr_wake2)

用途：用户态读写锁、复杂同步原语的高效实现。

六、内存序与竞态防护

6.1 核心问题

Waiter 和 Waker 并发时，必须保证：waiter 要么看到 futex 值变化，要么被 waker 唤醒，不能两者都错过。

6.2 SMP 解决方案

1
2
3
4
5
6
7
8
9
10
11
12
13
14
CPU 0 (waiter)                    CPU 1 (waker)
val = *futex;
futex_wait(futex, val);
  waiters++            (a)
  smp_mb()             (A)  ←─── 配对 ───→  smp_mb()  (B)
  lock(hb);                                  *futex = newval;
  uval = *futex;                             futex_wake(futex);
  if (uval == val)
    queue(); schedule();                     if (waiters)
                                               lock(hb);
                                               wake();
                                               unlock(hb);
  else
    waiters--          (b)
  • (A) ↔ (B)：futex_hb_waiters_inc() 的 smp_mb__after_atomic() 与 futex_hb_waiters_pending() 的 smp_mb() 配对
  • Waker 看到 waiters > 0 才加锁，避免无 waiter 时的锁开销
  • Waiter 在加锁前递增 waiters，保证 waker 不会错过

6.3 futex_q 唤醒判定

1
2
// futex_q 被认为已唤醒当且仅当：
plist_node_empty(&q->list) || q->lock_ptr == NULL

唤醒顺序：先清空 list 节点，再 store_release lock_ptr = NULL。

七、Requeue 机制

7.1 概述

FUTEX_REQUEUE 将一个 futex（uaddr1）上的 waiter 批量转移到另一个 futex（uaddr2），是 pthread 条件变量 的核心实现。

7.2 流程

1
2
3
4
5
6
futex_requeue(uaddr1, uaddr2, nr_wake, nr_requeue, cmpval, requeue_pi)
  → double_lock_hb(hb1, hb2)     按地址顺序加两把 bucket 锁
  → futex_wake(uaddr1, nr_wake)  先唤醒 uaddr1 上部分 waiter
  → 对 uaddr1 上剩余 waiter：
      requeue_futex(q, hb1, hb2, key2)  转移到 hb2
  → double_unlock_hb(hb1, hb2)

7.3 requeue_futex()

1
2
3
4
5
6
7
8
9
10
11
12
13
static inline void requeue_futex(struct futex_q *q,
    struct futex_hash_bucket *hb1, struct futex_hash_bucket *hb2,
    union futex_key *key2)
{
    if (&hb1->chain != &hb2->chain) {
        plist_del(&q->list, &hb1->chain);
        futex_hb_waiters_dec(hb1);
        futex_hb_waiters_inc(hb2);
        plist_add(&q->list, &hb2->chain);
        q->lock_ptr = &hb2->lock;
    }
    q->key = *key2;
}

7.4 FUTEX_CMP_REQUEUE

在 requeue 前检查 *uaddr1 == cmpval，仅当条件满足时才执行 requeue。用于条件变量的 “while loop” 语义。

7.5 Requeue-PI

FUTEX_CMP_REQUEUE_PI / FUTEX_WAIT_REQUEUE_PI：requeue 到 PI futex 时，需建立优先级继承链。涉及复杂的状态机（Q_REQUEUE_PI_* 枚举），处理 RT 内核下的并发 wakeup 与 requeue 交错。

八、Priority Inheritance（PI）

8.1 问题背景

优先级反转：高优先级任务等待低优先级任务持有的锁，而中优先级任务抢占低优先级任务，导致高优先级任务被间接阻塞。

8.2 PI Futex 原理

1
2
3
4
5
6
7
8
9
10
11
12
futex 值编码：
  bit 31-1: owner TID
  bit 0:    FUTEX_WAITERS 标志

FUTEX_LOCK_PI:
  尝试 CAS(0 → current_tid)
  失败 → 解析 owner TID → 建立 PI 链 → 阻塞在 rt_mutex 上
  owner 优先级被临时提升到 waiters 中最高优先级

FUTEX_UNLOCK_PI:
  查找 top waiter → 将 futex 值设为 waiter 的 TID
  → 唤醒 waiter 并通过 PI 恢复 owner 优先级

8.3 关键函数（pi.c）

函数 功能
futex_lock_pi() PI 加锁：atomic 尝试 → rt_mutex 阻塞
futex_unlock_pi() PI 解锁：传递锁给 top waiter
futex_lock_pi_atomic() 原子路径：解析 owner、attach pi_state
attach_to_pi_owner() 建立 PI 链，处理 owner 退出竞态
fixup_pi_owner() 修正 pi_state owner 并获取锁
wake_futex_pi() PI 唤醒 top waiter

8.4 pi_state 生命周期

1
2
3
4
alloc_pi_state()          ← 从 current->pi_state_cache 分配
  → attach_to_pi_state()  ← 关联到 futex key
  → rt_mutex 阻塞/唤醒
  → put_pi_state()        ← refcount 归零，回收到 cache 或 kfree

Owner 退出时：exit_pi_state_list() 遍历 task->pi_state_list，清理 PI 链并设置 FUTEX_OWNER_DIED。

九、Robust Futex

9.1 问题背景

持有 robust mutex 的线程异常退出（crash/exit）时，锁不会被释放，导致其他 waiter 永久阻塞。

9.2 机制

用户态维护 per-thread robust list（通过 set_robust_list 注册）：

1
2
3
4
5
struct robust_list_head {
    struct robust_list list;
    long futex_offset;
    struct robust_list *list_op_pending;
};

线程退出时内核遍历 robust list：

1
2
3
4
5
6
7
futex_exit_release(task)
  → futex_cleanup_begin()     设置 FUTEX_STATE_EXITING
  → exit_robust_list()        遍历 robust_list
      → handle_futex_death()  设置 FUTEX_OWNER_DIED 位
      → futex_wake()          唤醒 waiter
  → exit_pi_state_list()      清理 PI futex
  → futex_cleanup_end()       设置 FUTEX_STATE_DEAD

9.3 FUTEX_OWNER_DIED

Waiter 被唤醒后检查 FUTEX_OWNER_DIED 位：

  • 若 owner 已死，waiter 可尝试 CAS(OWNER_DIED → current_tid) 接管锁
  • 否则正常竞争

十、Futex2 扩展（futex_waitv）

10.1 概述

Linux 6.x 引入的 futex2 接口，支持在单个 syscall 中等待多个 futex。

10.2 futex_waitv 系统调用

1
2
3
4
5
6
SYSCALL_DEFINE5(futex_waitv,
    struct futex_waitv __user *, waiters,   // 等待数组
    unsigned int, nr_futexes,              // 数组长度（≤128）
    unsigned int, flags,
    struct __kernel_timespec __user *, timeout,
    clockid_t, clockid)
1
2
3
4
5
6
struct futex_waitv {
    __u64 val;          // 期望值
    __u64 uaddr;        // futex 地址
    __u32 flags;        // FUTEX_32 | FUTEX_PRIVATE_FLAG
    __u32 __reserved;
};

10.3 实现（waitwake.c）

1
2
3
4
5
futex_wait_multiple(vs, count, to)
  → futex_wait_multiple_setup()  逐个 futex 检查值并入队
  → futex_sleep_multiple()       统一 schedule
  → 任一 futex 被 wake 或值变化 → 返回对应 index
  → unqueue_multiple()           清理其余 waiter

用途：eventfd + futex 等多路复用等待，减少 syscall 次数。

十一、系统调用接口

11.1 futex(2) 操作码

操作 值 说明
FUTEX_WAIT 0 等待 futex 值等于 val
FUTEX_WAKE 1 唤醒最多 val 个 waiter
FUTEX_REQUEUE 3 转移 waiter 到 uaddr2
FUTEX_CMP_REQUEUE 4 条件 requeue
FUTEX_WAKE_OP 5 wake + 原子操作 + 条件 wake
FUTEX_LOCK_PI 6 PI 加锁
FUTEX_UNLOCK_PI 7 PI 解锁
FUTEX_TRYLOCK_PI 8 PI 尝试加锁
FUTEX_WAIT_BITSET 9 带 bitset 过滤的 wait
FUTEX_WAKE_BITSET 10 带 bitset 过滤的 wake
FUTEX_WAIT_REQUEUE_PI 11 wait + requeue 到 PI futex
FUTEX_CMP_REQUEUE_PI 12 条件 requeue 到 PI futex
FUTEX_LOCK_PI2 13 PI 加锁（CLOCK_REALTIME 超时）

标志位：

标志 值 说明
FUTEX_PRIVATE_FLAG 128 进程私有 futex（默认）
FUTEX_CLOCK_REALTIME 256 使用 CLOCK_REALTIME 超时

11.2 其他系统调用

调用 功能
set_robust_list(2) 注册当前线程的 robust futex 列表
get_robust_list(2) 获取指定线程的 robust 列表
futex_waitv(2) 多 futex 等待（futex2）

11.3 do_futex 分发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
long do_futex(uaddr, op, val, timeout, uaddr2, val2, val3)
{
    cmd = op & FUTEX_CMD_MASK;
    if (!(op & FUTEX_PRIVATE_FLAG)) flags |= FLAGS_SHARED;

    switch (cmd) {
    case FUTEX_WAIT:       return futex_wait(...);
    case FUTEX_WAKE:       return futex_wake(...);
    case FUTEX_REQUEUE:    return futex_requeue(...);
    case FUTEX_LOCK_PI:    return futex_lock_pi(...);
    case FUTEX_UNLOCK_PI:  return futex_unlock_pi(...);
    // ...
    }
}

十二、RK3588/ARM64 平台说明

12.1 架构支持

ARM64 完整支持 futex 全部特性：

1
2
3
4
# arch/arm64/Kconfig
select HAVE_FUTEX
select ARCH_HAS_FUTEX_ATOMIC
select FUTEX_PI

12.2 用户态原子操作（arch/arm64/include/asm/futex.h）

ARM64 使用 LL/SC（Load-Linked / Store-Conditional） 指令实现 futex 原子操作：

1
2
3
4
5
6
7
8
9
10
11
// futex_atomic_cmpxchg_inatomic — CAS 实现
asm volatile(
    "1: ldxr   %w1, %2\n"       // Load-Exclusive
    "   sub    %w3, %w1, %w5\n"  // compare with oldval
    "   cbnz   %w3, 4f\n"        // mismatch → exit
    "2: stlxr  %w3, %w6, %2\n"   // Store-Exclusive
    "   cbz    %w3, 3f\n"        // success
    "   ... retry loop ...\n"
    "3: dmb    ish\n"            // Data Memory Barrier
    "4:\n"
    ...);
操作 ARM64 实现
FUTEX_OP_SET mov + STLRX
FUTEX_OP_ADD add + STLRX
FUTEX_OP_OR orr + STLRX
FUTEX_OP_ANDN and + STLRX
FUTEX_OP_XOR eor + STLRX
cmpxchg ldxr + stlxr + dmb ish
  • 最大重试次数：FUTEX_MAX_LOOPS = 128
  • 使用 uaccess_enable_privileged() 允许在内核态访问用户页
  • dmb ish（Inner Shareable）保证 ARM64 弱内存序下的正确性

12.3 big.LITTLE 注意点

  • Futex key 基于虚拟地址/mm，与运行在哪个核心无关
  • PI futex 的优先级继承跨 A76/A55 核心正常工作（基于 RT mutex）
  • futex_hashsize = 256 × 8 = 2048（RK3588 8 核）

12.4 Android/Bionic 相关

RK3588 常用于 Android 平台，Bionic libc 大量使用 futex：

  • pthread_mutex → FUTEX_WAIT/WAKE 或 FUTEX_LOCK_PI/UNLOCK_PI
  • pthread_cond → FUTEX_WAIT + FUTEX_REQUEUE
  • sem → FUTEX_WAIT/WAKE
  • Java synchronized / Object.wait/notify → 底层 futex

十三、完整 pthread mutex 时序

以 两个线程竞争 pthread_mutex 为例（非 PI、非 robust）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
[Thread A — 加锁成功]
  lock(&mutex):
    CAS(0 → 1) 成功 → 返回（纯用户态，零 syscall）

[Thread B — 加锁阻塞]
  lock(&mutex):
    CAS(0 → 1) 失败（值为 1）
    val = 1; 设置 WAITERS 位: CAS(1 → 0x80000001)  // bit31=TID|WAITERS
    futex(FUTEX_WAIT, &mutex, 2)    // val=2 含 WAITERS 位
      → 内核: futex_wait_setup() → *mutex==2 ✓
      → futex_queue() → schedule() 阻塞

[Thread A — 解锁]
  unlock(&mutex):
    读取 waiter 存在 → store(0)
    futex(FUTEX_WAKE, &mutex, 1)
      → 内核: futex_wake() → 找到 Thread B
      → futex_wake_mark() → wake_up_q()
      
[Thread B — 被唤醒]
  futex_wait 返回 → CAS(0 → 1) 成功 → 持有锁

条件变量 broadcast 时序：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
[Thread B — cond_wait]
  lock(&cond_mutex)
  unlock(&user_mutex)
  futex(FUTEX_WAIT, &cond, 0)         // 等待条件
    → 内核阻塞在 cond futex 上

[Thread A — cond_broadcast]
  lock(&cond_mutex)
  修改条件
  unlock(&user_mutex)
  futex(FUTEX_CMP_REQUEUE, &cond, &mutex, 1, INT_MAX, &cond_val)
    → wake cond 上 1 个 waiter
    → 将其余 waiter requeue 到 mutex futex
  unlock(&cond_mutex)

[Thread B — 被 requeue]
  在 mutex futex 上被唤醒 → 竞争 mutex → 重新 lock(&cond_mutex)

十四、总结

kernel/futex 是 Linux 用户态同步的基石：

  1. Fast path 在用户态 — 无竞争时零 syscall，性能接近纯用户态锁
  2. 哈希桶等待队列 — 按 futex key 哈希，plist 优先级排序，支持高效 wake
  3. 内存序保证 — SMP 下 waiters 计数 + memory barrier 防止 lost wakeup
  4. Requeue — 条件变量的核心原语，批量转移 waiter
  5. Priority Inheritance — 通过 RT mutex 解决优先级反转
  6. Robust Futex — 线程异常退出时自动清理持有的锁
  7. Futex2 — futex_waitv 支持多 futex 单 syscall 等待

RK3588（ARM64）通过 LL/SC 指令（ldxr/stlxr/dmb ish）实现高效的用户态原子操作，完整支持 PI futex 和 robust futex，是 Android/Linux 桌面系统中 pthread 同步的底层支撑。

附录：源文件清单

文件 行数 分类
core.c ~1159 基础设施、key、robust 清理
waitwake.c ~708 wait/wake/wake_op/waitv
requeue.c ~897 requeue / requeue_pi
pi.c ~1233 优先级继承
syscalls.c ~379 系统调用入口
futex.h ~294 内部头文件

相关头文件：

文件 功能
include/linux/futex.h 内核公共 API
include/uapi/linux/futex.h 用户态 API、操作码定义
arch/arm64/include/asm/futex.h ARM64 原子操作实现

文章互动

阅读 --

留言

0 条留言

正在加载留言…