Linux同步:mutex 源码分析(下)

Linux同步：mutex 源码分析（下）

前篇Linux同步：Linux mutex 设计思想（上）介绍了mutex的设计思想，这篇继续分析mutex的源码。高性能意味着无竞争时要极快，公平性意味着等待者不能被饿死。Linux mutex 如何同时兼顾这两者？答案就藏在三条路径和几个标志位里。

一、mutex结构体

struct mutex结构体主要成员如下：

// include/linux/mutex_types.hstruct mutex {        atomic_long_t           owner;        raw_spinlock_t          wait_lock;#ifdef CONFIG_MUTEX_SPIN_ON_OWNER        struct optimistic_spin_queue osq; /* Spinner MCS lock */#endif        struct list_head        wait_list;};

其中 owner 是一个 atomic_long_t 类型的原子变量。它利用了内存对齐的特性，将 64 位（或 32 位）空间划分为两部分：

高位（bits 3-63/31）：存储指向锁持有者的 struct task_struct * 指针。

低位（bits 0-2）：存储三个状态标志。

由于 task_struct 指针按 L1_CACHE_BYTES 对齐，其低 3 位恒为 0，所以可以安全地将这些“闲置位”用作标志位。

三个状态标志（MUTEX_FLAGS）这三个标志位于 owner 的最低 3 位：

MUTEX_FLAG_WAITERS (0x01)：Bit 0。标记当前锁的等待队列中至少有一个进程在排队。解锁时，如果存在此标志位，说明不能简单地直接释放锁，必须进入慢速路径去唤醒队列中的队首进程。

MUTEX_FLAG_HANDOFF (0x02)：Bit 1。当等待者发现自己处于等待队列队首时，会设置此标志，表示自己已准备好请求接手锁。此标志位为了防止饥饿。

MUTEX_FLAG_PICKUP (0x04)：Bit 2。解锁者在释放锁时，如果检测到 HANDOFF 标志，会清除 HANDOFF，然后设置 PICKUP，并将锁的 owner 字段指向队首等待者。这相当于向外界宣告：“我已经指定了下一个锁的拥有者，其他人不能抢了”。

MUTEX_FLAGS (0x07)：用于一次性提取或清除全部三个标志位的掩码。

内核提供了两个辅助函数，用于从 owner 字段中提取出 task_struct * 指针和标志位：

static inline struct task_struct *__mutex_owner(struct mutex *lock){    return (struct task_struct *)(atomic_long_read(&lock->owner) & ~MUTEX_FLAGS);}static inline unsigned long __owner_flags(unsigned long owner){        return owner & MUTEX_FLAGS;}

二、mutex解锁流程

查看mutex的代码时，可以先看解锁流程。解锁流程比较简单，可以帮助理解加锁流程怎么设置这些标志位。

2.1 快速路径

lock->owner == curr 满足表明没有入队等待者（WAITERS 标志为 0），可以直接释放锁。此时即使有进程在乐观自旋，它们也会在锁释放后的原子操作中竞争获取。

static __always_inline bool __mutex_unlock_fast(struct mutex *lock){        unsigned long curr = (unsigned long)current;        return atomic_long_try_cmpxchg_release(&lock->owner, &curr, 0UL);}

因为自旋等待的时候也没设置标志位。所以unlock直接释放后，当前正在乐观自旋的进程可以直接获取锁。

2.2 handoff 表明有waiter在申请该锁，唤醒第一个waiter

关键代码如下，lock->owner = task | MUTEX_FLAG_PICKUP;表明唤醒指定的进程。task变量为等待链表的第一个waiter。

__mutex_unlock_slowpath里调用__mutex_handoff实现了该功能：

static void __mutex_handoff(struct mutex *lock, struct task_struct *task){        unsigned long owner = atomic_long_read(&lock->owner);        for (;;) {                unsigned long new;                new = (owner & MUTEX_FLAG_WAITERS);                new |= (unsigned long)task;   //唤醒指定的进程。防止饥饿。                if (task)                        new |= MUTEX_FLAG_PICKUP;                if (atomic_long_try_cmpxchg_release(&lock->owner, &owner, new))                        break;        }}

2.3 如果没有进程在请求锁，但有waiter，则也是唤醒第一个waiter。

__mutex_unlock_slowpath忽略了部分跟该流程无关的代码。

//__mutex_unlock_slowpath()    for (;;) {	    // 忽略MUTEX_FLAG_HANDOFF和调试相关            if (atomic_long_try_cmpxchg_release(&lock->owner, &owner, __owner_flags(owner))) {                    if (owner & MUTEX_FLAG_WAITERS)  //有waiter                            break;                    return;            }    }    raw_spin_lock_irqsave(&lock->wait_lock, flags);    if (!list_empty(&lock->wait_list)) {            /* get the first entry from the wait-list: */            struct mutex_waiter *waiter =                    list_first_entry(&lock->wait_list,                                     struct mutex_waiter, list);            next = waiter->task;            __clear_task_blocked_on(next, lock);            wake_q_add(&wake_q, next);//添加第一个要唤醒的进程    }    raw_spin_unlock_irqrestore_wake(&lock->wait_lock, flags, &wake_q);

2.4 解锁路径的状态流转汇总

解锁流程中，owner 的状态变化有以下三种情况：

无等待者：owner = task | 0 → 0
有等待者且设置了 HANDOFF：owner = task | HANDOFF | WAITERS → first_task | PICKUP | WAITERS
有等待者但无 HANDOFF：owner = task | WAITERS → WAITERS（保留标志，等待唤醒）

三、mutex加锁流程

下面分别介绍mutex加锁流程，包含三个路径。

3.1 快速路径：如果没有进程获取锁，则owner为空，直接获取到锁。

lock->owner为空，表明无人持有锁，此时通过原子指令获取锁，速度很快。

static __always_inline bool __mutex_trylock_fast(struct mutex *lock){        unsigned long curr = (unsigned long)current;        unsigned long zero = 0UL;        MUTEX_WARN_ON(lock->magic != lock);        if (atomic_long_try_cmpxchg_acquire(&lock->owner, &zero, curr))                return true;        return false;}

3.2 中速路径：自旋等待

当发现持有锁的进程处于running，则认为这个锁很快就会被释放，调用mutex_optimistic_spin自旋等待。

伪代码如下：

mutex_optimistic_spin() osq_lock // 保证只能有一个waiter自旋在mutex lock锁    for (;;) {   	    if (__mutex_trylock_or_owner() 获取锁) {		        break        	}	    mutex_spin_on_owner 自旋等待锁持有者释放mutex lock或者退出循环	} osq_unlock

代码实现上调用osq_lock获取osq锁后，并循环等待直到锁被释放、持有锁进入睡眠、被抢占的场景出现。 osq_lock、osq_unlock 实现的是mcs队列锁，防止出现缓存颠簸问题。

注：本文重点聚焦 mutex 的加解锁流程与状态机。osq 所实现的 MCS 队列锁，其设计原理与源码实现，将在后续文章中单独展开分析。

mutex_spin_on_owner自旋等待mutex lock锁释放或者主动放弃。

//mutex_spin_on_owner()  while (__mutex_owner(lock) == owner) {                barrier();                /*                 * Use vcpu_is_preempted to detect lock holder preemption issue.                 */                if (!owner_on_cpu(owner) || need_resched()) {                        ret = false;                        break;                }                // 忽略ww相关代码                cpu_relax();        }

!owner_on_cpu(owner) || need_resched()这个是主动放弃的条件，如果持有者不再cpu上，或者自身需要被抢占，就主动放弃获取锁。

3.3 慢速路径

走到慢速流程，则加到等待队列里，设置waiter标志位，并睡眠。如果睡眠过程中被唤醒，则调用__mutex_trylock_or_handoff(lock, true)申请锁，这样下次唤醒会优先唤醒。下面的代码做过一些删减，为了能够更容易看清流程。

  static __always_inline int __sched__mutex_lock_common(struct mutex *lock, unsigned int state, unsigned int subclass,                    struct lockdep_map *nest_lock, unsigned long ip,                    struct ww_acquire_ctx *ww_ctx, const bool use_ww_ctx){        DEFINE_WAKE_Q(wake_q);        struct mutex_waiter waiter;        unsigned long flags;        int ret;        raw_spin_lock_irqsave(&lock->wait_lock, flags);        waiter.task = current;        __mutex_add_waiter(lock, &waiter, &lock->wait_list); //添加列表的时候，会设置waiter标志位。        set_current_state(state);        for (;;) {                bool first;                if (__mutex_trylock(lock))                        goto acquired;                raw_spin_unlock_irqrestore_wake(&lock->wait_lock, flags, &wake_q);                schedule_preempt_disabled();   //第一次睡眠，此时未设置handoff标志。                first = __mutex_waiter_is_first(lock, &waiter);                set_current_state(state);                if (__mutex_trylock_or_handoff(lock, first))  //如果第一次睡眠被唤醒后没有获取到锁，并且自己排在队头，则会设置handoff标志，请求优先获取锁                        break;if (first) {                        clear_task_blocked_on(current, lock);                        if (mutex_optimistic_spin(lock, ww_ctx, &waiter))    //先乐观自旋获取锁                                break;                        set_task_blocked_on(current, lock);                }                raw_spin_lock_irqsave(&lock->wait_lock, flags);        }        raw_spin_lock_irqsave(&lock->wait_lock, flags);acquired:        __set_current_state(TASK_RUNNING);        __mutex_remove_waiter(lock, &waiter);        raw_spin_unlock_irqrestore_wake(&lock->wait_lock, flags, &wake_q);        preempt_enable();        return 0;}

mutex_optimistic_spin(lock, ww_ctx, &waiter)这行代码里，传递了waiter，不需要获取mcs锁，这样保证优先获取mutex lock。

小结：队首进程第一次被唤醒时，如果乐观自旋仍无法获取锁，则会设置 HANDOFF 标志，向解锁者请求"定向交接"。这避免了队首进程被新来的乐观自旋者反复"插队"而导致饥饿。

3.4 锁定路径的状态流转汇总

加锁流程中，owner 的状态变化有以下几种情况：

无竞争，快速取锁：0 → curr_task
有等待者，锁刚好被释放：WAITERS → new_task | WAITERS
队首进程请求交接：owner = old_task | WAITERS → old_task | HANDOFF | WAITERS
当队首进程在慢速路径中被唤醒但仍无法获取锁时，会设置 HANDOFF 标志，请求解锁者将锁定向交接给自己。
被 PICKUP 唤醒并成功取锁：owner = first_task | PICKUP | WAITERS → first_task | WAITERS
解锁者已通过 __mutex_handoff 将 owner 指向队首进程并设置了 PICKUP。该进程被唤醒后，在 __mutex_trylock_or_handoff() 中检测到该标志，会原子性地清除 PICKUP 位，成功持有锁。

mutex实现的时候，为了高性能，引入了乐观自旋，这个会引起不公平（__mutex_lock_common优先调用mutex_optimistic_spin获取锁）。同时又通过handoff、pickup机制来缓解这个不公平性。

四、总结：性能与公平的优雅权衡

纵观 mutex 的三条路径，我们可以看到内核工程师在性能与公平之间的极致权衡：

路径	触发条件	策略	目标
快速路径	`owner == 0` （无竞争）	单次原子操作	极致低延迟
中速路径	持有者正在运行	乐观自旋（CPU 忙等）	避免上下文切换开销
慢速路径	持有者睡眠或竞争激烈	入队睡眠 + 交接机制	保证公平，防止饥饿

而 HANDOFF 与 PICKUP 这对标志位，如同接力赛中的交接棒，确保即使在超高并发下，等待队列中的进程也不会被新来的竞争者"插队"饿死。

设计哲学：利用内存对齐的闲置位存储状态，用轻量级的原子操作替代重量级的锁保护，再通过状态机流转串联起三条路径。这种在微观层面"锱铢必较"的优化精神，正是 Linux 内核性能卓越的基石。