futex 操作的字面语义 (内核不理解"锁"):

 futex_wait(&uaddr, expected):
   "若 *uaddr == expected 则睡到有人 futex_wake(uaddr)"
 futex_wake(&uaddr, n):
   "唤醒正在等 *uaddr 的至多 n 个线程"

 所有锁语义 (谁持锁/递归/错误检查) 全在用户态 glibc:
   pthread_mutex_lock:
     state CAS 0→1            [无竞争: 完成]
     失败 → state=2(有等待者) → futex_wait(state, 2)
   pthread_mutex_unlock:
     state 1→0 或 2→0
     若原值是 2 → futex_wake(&state, 1)   [定点唤醒一个]

 为什么值校验必须是协议的一部分:
   唤醒与状态变更的顺序无法原子 — "先改值后 wake" 与
   "先查值后 sleep" 之间存在窗口, wait 的 expected 参数
   就是关闭这个窗口的凭据 (31.2.2 节竞态纪律)

// include/linux/futex.h:32-53
union futex_key {
    struct {            /* 共享映射 (FLAGS_SHARED) */
        u64 i_seq;      /* 后备 inode 的序号 */
        unsigned long pgoff;    /* 文件内页偏移 */
        unsigned int offset;    /* 页内偏移 */
    } shared;
    struct {            /* 私有映射 (默认) */
        union {
            struct mm_struct *mm;   /* 进程地址空间 */
            u64 __tmp;
        };
        unsigned long address;  /* 进程内地址 */
        unsigned int offset;
    } private;
    struct {            /* 内核内部用 */
        u64 ptr;
        unsigned long word;
        unsigned int offset;
        unsigned int node;  /* NOT hashed! */
    } both;
};

同一物理锁在不同进程眼中的地址不同 (各自 mmap),
"身份"必须锚定到物理层:

 私有锁 (FLAGS_PRIVATE, 默认):
   key = (mm, address, offset)
   → 只在进程内匹配, 哈希也只需进程内唯一
   → 7.0 的 futex_private_hash: 进程可申请
     本地哈希表 (futex_hash_prctl), 多线程
     大户不再挤全局桶 — 锁竞争随核数线性化的
     长年顽疾的解法

 共享锁 (FLAGS_SHARED, shm/mmap 共享内存上的锁):
   key = (inode->i_seq, pgoff, offset)
   → 锚定后备文件页 — 不同进程地址不同,
     但页缓存页唯一 (30.1 节"共享=同文件"的回声)

 get_futex_key() (core.c:548) 的职责:
   逐页表/页缓存解析地址 → 物理身份; 共享模式
   需持页引用防遍历中途映射被拆

// kernel/futex/futex.h:134-139
struct futex_hash_bucket {
    atomic_t waiters;       /* 快查: 桶上有没有人 */
    spinlock_t lock;
    struct plist_head chain;    /* 等待者链 (plist: 优先级排序) */
    struct futex_private_hash *priv;
} ____cacheline_aligned_in_smp;

futex_wait(uaddr) 的容器视角:

 get_futex_key → key ──hash──> 全局哈希表 (futex_hash)
                              大小随内存规模自适配
        ┌──────────────────────────────────────┐
        │ bucket[0] [chain: (key1,q1)(key2,q2)]│
        │ bucket[1] [chain: (key3,q3)]         │  多把锁的等待者
        │ ...                                  │  共享一个桶,
        │ bucket[k] [chain: (keyN,qN)]         │  哈希碰撞即共桶
        └──────────────────────────────────────┘
 futex_wake 遍历桶内链表, 以 futex_match(key) 定点
 (同桶不同锁的等待者被扫描但不被唤醒 — 31.2.1 节)

 waiters 原子计数的意义: futex_wake 先查
 futex_hb_waiters_pending(hb) (waitwake.c:173),
 为零则连桶锁都不拿 — "无等待者时的 wake syscall
 只剩一次查表" 的快速否决

// kernel/futex/waitwake.c:155-215（节选）
int futex_wake(u32 __user *uaddr, unsigned int flags, int nr_wake, u32 bitset)
{
    struct futex_q *this, *next;
    union futex_key key = FUTEX_KEY_INIT;
    DEFINE_WAKE_Q(wake_q);
    int ret;

    if (!bitset)
        return -EINVAL;

    ret = get_futex_key(uaddr, flags, &key, FUTEX_READ);    /* 地址→键 */
    if (unlikely(ret != 0))
        return ret;
    ...
    CLASS(hb, hb)(&key);            /* 定位哈希桶 */

    /* Make sure we really have tasks to wakeup */
    if (!futex_hb_waiters_pending(hb))  /* :173 快速否决 */
        return ret;

    spin_lock(&hb->lock);

    plist_for_each_entry_safe(this, next, &hb->chain, list) {
        if (futex_match(&this->key, &key)) {        /* 键精确匹配 */
            if (this->pi_state || this->rt_waiter) {
                ret = -EINVAL;      /* PI 等待者不许裸 wake */
                break;
            }

            /* Check if one of the bits is set in both bitsets */
            if (!(this->bitset & bitset))
                continue;       /* bitset 过滤 */

            this->wake(&wake_q, this);  /* 摘队+挂唤醒批 */
            if (++ret >= nr_wake)
                break;
        }
    }
    ...
}

// kernel/futex/waitwake.c:591-612（注释原文即协议）
int futex_wait_setup(u32 __user *uaddr, u32 val, unsigned int flags,
             struct futex_q *q, union futex_key *key2,
             struct task_struct *task)
{
    ...
    /*
     * Access the page AFTER the hash-bucket is locked.
     * Order is important:
     *
     *   Userspace waiter: val = var; if (cond(val)) futex_wait(&var, val);
     *   Userspace waker:  if (cond(var)) { var = new; futex_wake(&var); }
     *
     * The basic logical guarantee of a futex is that it blocks ONLY
     * if cond(var) is known to be true at the time of blocking, for
     * any cond.  If we locked the hash-bucket after testing *uaddr,
     * that would open a race condition where we could block
     * indefinitely with cond(var) false, which would violate the
     * guarantee.
     *
     * On the other hand, we insert q and release the hash-bucket only
     * after testing *uaddr.  This guarantees that futex_wait() will
     * NOT absorb a wakeup if *uaddr does not match the desired values
     * while the syscall executes.
     */
retry:
    ret = get_futex_key(uaddr, flags, &q->key, FUTEX_READ);
    ...
    if (1) {
        CLASS(hb, hb)(&q->key);

        futex_q_lock(q, hb);            /* [1] 先锁桶 */

        ret = futex_get_value_locked(&uval, uaddr); /* [2] 后读值 */
        if (ret) {
            futex_q_unlock(hb);
            ret = get_user(uval, uaddr);    /* 缺页: 供货后重试 */
            ...
        }

        if (uval != val) {          /* [3] 值已变: 不睡! */
            futex_q_unlock(hb);
            return -EWOULDBLOCK;        /* 即用户态的 EAGAIN */
        }
        queue_me(q, hb);            /* [4] 挂队+放锁 */
    }
    return ret;
}

两个线程的交错矩阵 (W=唤醒方, S=睡眠方):

 S: 锁桶 → 读值=val → 挂队 → 放锁 → 睡
 W: 改值 → futex_wake → 锁桶 → 摘队 → 唤醒

 无论 W 的"改值+wake"落在 S 时序的哪个点:
 - W 先完成全部: S 读值时已 val'≠val → -EWOULDBLOCK,
   不睡, 回用户态重验条件 (不丢唤醒)
 - S 挂队后 W 才 wake: W 在桶上找到 q, 定点唤醒
   (不丢唤醒)
 - W 在 S 读值与挂队之间改值: 因 S 已持桶锁, W 的
   锁桶被推迟到 S 放锁后 → 顺序化, 无窗口

 "唤醒不丢"的唯一前提: 等待与唤醒共享同一把桶锁
 作为交汇点 — 这就是 futex 的全部同步魔法

// kernel/futex/waitwake.c:706（入口）
int futex_wait(u32 __user *uaddr, unsigned int flags, u32 val, ktime_t *abs_time, u32 bitset)

// kernel/futex/requeue.c（futex_requeue 主体的协议面）

requeue 的使用场景 (glibc 条件变量):

 pthread_cond_broadcast:
   等待 cond 的 N 个线程若全部唤醒 → N 个线程同时抢
   mutex → 惊群 + 锁风暴
   FUTEX_REQUEUE(cond, mutex, 1, INT_MAX):
     唤醒 1 个去抢锁, 其余改挂到 mutex 的等待队列
     → 锁逐个移交, 零风暴
 CMP 变体: 先校验 *uaddr1 == val3 防陈旧引用 (29 章同款)

 FUTEX_REQUEUE_PI (pi.c): PI 版转移, 目标锁的所有权
 同步移交 — 31.3.3 节

glibc 的 state 字编码 (futex 字 = mutex->data.lock):

 值 0: 无锁            — lock 即 CAS 0→1, 完成
 值 1: 有主, 无等待者   — unlock 直接 1→0, 不需要 syscall!
 值 2: 有主, 有等待者   — unlock 必须 futex_wake
                            wait 以 futex_wait(&state, 2) 睡

 pthread_mutex_lock 的完整决策树:

   CAS(0→1) 成功? ──是──> 返回                       [层1 快路径]
        │否
   自旋若干轮重试 (adaptive 模式)? ──成功──> 返回     [层2 自旋]
        │否
   state CAS → 2 (登记"有等待者")
   futex_wait(&state, 2) ──EWOULDBLOCK──> 回层1      [层3 内核]
        │睡眠... 被唤醒
   回到层1 重试 CAS (醒来后重新抢锁)

 pthread_mutex_unlock:
   原 1 → CAS 1→0, 无等待者: 完事 (零系统调用!)
   原 2 → 置 0 + futex_wake(&state, 1)

// include/linux/futex.h:80-100（futex_init_task, 每线程初始化）
static inline void futex_init_task(struct task_struct *tsk)
{
    tsk->robust_list = NULL;        /* 用户态注册的链表头 */
#ifdef CONFIG_COMPAT
    tsk->compat_robust_list = NULL;
#endif
    INIT_LIST_HEAD(&tsk->pi_state_list);    /* PI 状态链 (31.3.3) */
    tsk->pi_state_cache = NULL;
    tsk->futex_state = FUTEX_STATE_OK;
    mutex_init(&tsk->futex_exit_mutex);
}

robust 协议 (用户态与内核的分工):

 用户态 (set_robust_list 注册):
   每把 robust 锁是链表节点; 加锁时先把节点挂链
   ("预设死亡善后"), 再 CAS 拿锁; 解锁后摘链
   → "链上但已加锁" = 死亡时正在持有的锁

 内核 (exit_robust_list, 进程退出钩子):
   遍历死者的 robust 链:
     对每把"持有中"的锁: 置 FUTEX_OWNER_DIED 位
     + futex_wake_op 定点唤醒等待者
   等待者醒来看到 OWNER_DIED → 得知前任暴毙
   → glibc 返回 EOWNERDEAD, 应用决定恢复或标记
   ENOTRECOVERABLE

优先级反转问题 (13.4 节 RT 抢占的延续):

 低优先级线程 L 持锁 → 高优先级 H 睡在锁上 →
 中优先级 M 抢占 L 无限跑 → H 的优先级被 M 践踏

 PI 的解法: H 睡在锁上期间, L "继承" H 的优先级,
 M 无法抢占 L → L 尽快出临界区

 普通 futex 无解: 内核不知道"睡在哪个锁上" (只有键),
 更不知道持锁者是谁 (值在用户态)!

 FUTEX_LOCK_PI 的做法: 把"持锁者 PID"写进 futex 字
 (值 = tid, 锁值即持锁者身份!), 内核接管:
   futex_lock_pi (pi.c:918)
     futex_lock_pi_atomic (:515) 原子地:
       校验/登记 owner = 持锁 tid
     建 futex_pi_state (futex.h:144-160):
       pi_mutex = rt_mutex (17 章 RT 互斥的内核形态!)
       owner   = 持锁 task
     H 以 rt_mutex_waiter 挂入 → 优先级沿
       owner 链传播 (boost), L 跑在 H 的优先级上
   futex_unlock_pi (pi.c:1133):
     解除 boost, 把锁原子移交最高优先等待者