乐于分享
好东西不私藏

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

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 的状态变化有以下三种情况:

  1. 无等待者:owner = task | 0 → 0

  2. 有等待者且设置了 HANDOFF:owner = task | HANDOFF | WAITERS → first_task | PICKUP | WAITERS

  3. 有等待者但无 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 的状态变化有以下几种情况:

  1. 无竞争,快速取锁:0 → curr_task

  2. 有等待者,锁刚好被释放:WAITERS → new_task | WAITERS

  3. 队首进程请求交接:owner = old_task | WAITERS → old_task | HANDOFF | WAITERS

    当队首进程在慢速路径中被唤醒但仍无法获取锁时,会设置 HANDOFF 标志,请求解锁者将锁定向交接给自己。

  4. 被 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 内核性能卓越的基石。

相关学习资料