Linux sys_futex futex_wake与hashbucket锁定
Linux sys_futex futex_wake与hashbucket锁定
futex(2) 系统调用是 Linux 实现高效用户态同步的核心机制。与所有经典同步原语不同的是，futex 在无竞争时完全在用户态通过原子操作完成，仅在需要等待或唤醒时进入内核。sys_futex 的入口是 do_futex：
```c
long do_futex(u32 __user *uaddr, int op, u32 val, ktime_t *timeout,
u32 __user *uaddr2, u32 val2, u32 val3)
{
int ret = -ENOSYS;
switch (op) {
case FUTEX_WAIT:
ret = futex_wait(uaddr, flags, val, timeout, val3);
break;
case FUTEX_WAKE:
ret = futex_wake(uaddr, flags, val, val3);
break;
case FUTEX_REQUEUE:
ret = futex_requeue(uaddr, flags, uaddr2, val, val2, &val3, 0);
break;
case FUTEX_CMP_REQUEUE:
ret = futex_requeue(uaddr, flags, uaddr2, val, val2, &val3, 1);
break;
case FUTEX_WAIT_BITSET:
ret = futex_wait(uaddr, flags, val, timeout, val3);
break;
case FUTEX_WAKE_BITSET:
ret = futex_wake(uaddr, flags, val, val3);
break;
case FUTEX_LOCK_PI:
ret = futex_lock_pi(uaddr, flags, timeout, 0);
break;
case FUTEX_UNLOCK_PI:
ret = futex_unlock_pi(uaddr, flags);
break;
...
}
return ret;
}
```
do_futex 根据 op 分派到不同的处理函数。对于 FUTEX_WAKE，核心路径是 futex_wake：
```c
static int futex_wake(u32 __user *uaddr, unsigned int flags, int nr_wake, u32 bitset)
{
struct futex_hash_bucket *hb;
struct futex_q *this, *next;
union futex_key key = FUTEX_KEY_INIT;
int ret = 0;
if (!bitset)
return -EINVAL;
ret = get_futex_key(uaddr, flags, &key, FUTEX_READ);
if (unlikely(ret != 0))
goto out;
hb = hash_futex(&key);
spin_lock(&hb->lock);
plist_for_each_entry_safe(this, next, &hb->chain, list) {
if (match_futex(&this->key, &key)) {
if (this->pi_state || this->rt_waiter) {
ret = -EINVAL;
break;
}
if (!(this->bitset & bitset))
continue;
wake_futex(this);
if (++ret >= nr_wake)
break;
}
}
spin_unlock(&hb->lock);
out:
return ret;
}
```
get_futex_key 是第一个关键操作。它通过 get_user_pages_fast 锁定用户态的页，防止页面被换出导致物理地址变化，然后根据该页所在的位置（常规映射或匿名映射）构造一个独特的 futex_key：
```c
int get_futex_key(u32 __user *uaddr, unsigned int flags, union futex_key *key,
enum futex_access rw)
{
unsigned long address = (unsigned long)uaddr;
struct mm_struct *mm = current->mm;
struct page *page, *tail;
struct address_space *mapping;
int err, ro = 0;
if (unlikely((address % sizeof(u32)) != 0))
return -EINVAL;
address &= PAGE_MASK;
err = get_user_pages_fast(address, 1, rw == FUTEX_WRITE, &page);
if (err < 0)
return err;
...
}
```
key 的构成决定了 futex 的关联方式。对于基于物理页框的共享 futex（MAP_SHARED），key 使用 mapping + index；对于私有映射，key 使用 mm + address。这使得 fork 之后的父子进程通过 COW 页面触发不同的 key，避免交叉唤醒。
hash_futex 将 key 哈希到 futex_hash_bucket：
```c
static struct futex_hash_bucket *hash_futex(union futex_key *key)
{
u32 hash = jhash2((u32 *)key, offsetof(typeof(*key), both.offset) / 4,
key->both.offset);
return &futex_queues[hash & (futex_hashsize - 1)];
}
```
futex_queues 是一个 hash bucket 数组，每个桶包含一个 plist（优先级排序链表）和一个 spinlock。plist 按优先级排序，确保优先级继承机制的 futex 操作中高优先级等待者被优先唤醒。
wake_futex 执行实际的唤醒操作：
```c
static void wake_futex(struct futex_q *q)
{
struct task_struct *p = q->task;
get_task_struct(p);
plist_del(&q->list, &q->hb->chain);
WRITE_ONCE(q->lock_ptr, &q->hb->lock);
...
wake_up_state(p, TASK_NORMAL);
put_task_struct(p);
}
```
wake_futex 从 hash bucket 链表中删除该 futex_q，然后调用 wake_up_state 将等待者的状态从 TASK_INTERRUPTIBLE 或 TASK_UNINTERRUPTIBLE 切换为 TASK_RUNNING，并将其加入运行队列。
hb->lock 是保护同一个 hash bucket 内所有 futex_q 的 spinlock。在 futex_wait 路径中，等待者在调用 futex_wait_queue_me 时会将 futex_q 插入到 hb->chain，并使用 set_current_state 设置 TASK_INTERRUPTIBLE，随后检查用户态的 futex 值是否发生变化。这种 double-check 机制是 futex 的核心：wake 和 wait 基于同一个 hb->lock 保证原子性，避免唤醒信号丢失。全局的 futex hash 表大小在启动时根据物理内存调整，默认散列到 256 个桶。