Linux內核同步機制mutex

mutex鎖概述

在linux內核中，互斥量mutex是一種保證CPU串行運行的睡眠鎖機制。和spinlock類似，都是同一個時刻只有一個線程進入臨界資源，不同的是，當無法獲取鎖的時候，spinlock原地自旋，而mutex則是選擇掛起當前線程，進入阻塞狀態。所以，mutex無法在中斷上下文中使用。

mutex鎖使用注意事項

mutex一次只能有一個進程或線程持有該鎖mutex只有它的擁有者可以釋放該鎖不能多次釋放同一把鎖不可以重復獲取同一把鎖，否則會造成死鎖必須使用mutex提供的專用初始化函數初始化該鎖不能重復初始化同一把鎖不能使用memset或memcpy等內存處理函數初始化mutex鎖線程退出時要釋放自己持有的所有mutex鎖不能用于設備中斷或軟中斷上下文中

mutex鎖結構體定義

owner：記錄mutex的持有者wait_lock：spinlock自旋鎖soq：MCS鎖隊列，用于支持mutex樂觀自旋機制wait_list：當無法獲取鎖的時候掛起在此magic：用于debug調試dep_map：用于debug調試

struct mutex { atomic_long_t  owner; spinlock_t  wait_lock;#ifdef CONFIG_MUTEX_SPIN_ON_OWNER struct optimistic_spin_queue osq; /* Spinner MCS lock */#endif struct list_headwait_list;#ifdef CONFIG_DEBUG_MUTEXES void   *magic;#endif#ifdef CONFIG_DEBUG_LOCK_ALLOC struct lockdep_map dep_map;#endif};

mutex鎖主要接口函數

mutex_init	初始化mutex對象
__mutex_init	mutex_init會調用此函數
DEFINE_MUTEX	靜態定義并初始化一個mutex對象
__MUTEX_INITIALIZER	DEFINE_MUTEX會調用此函數
mutex_lock	獲取mutex鎖，失敗進程進入D狀態
mutex_lock_interruptible	獲取mutex鎖，失敗進入S狀態
mutex_trylock	嘗試獲取mutex鎖，失敗直接返回
mutex_unlock	釋放mutex鎖
mutex_is_locked	判斷當前mutex鎖的狀態

獲取鎖流程分析

mutex_lock()函數調用might_sleep()函數判斷鎖的狀態，調用__mutex_trylock_fast()函數嘗試快速獲取mutex鎖，如果失敗，則調用__mutex_lock_slowpath()函數獲取mutex鎖

void __sched mutex_lock(struct mutex *lock){ might_sleep(); if (!__mutex_trylock_fast(lock))  __mutex_lock_slowpath(lock);}

如果沒有定義CONFIG_DEBUG_ATOMIC_SLEEP宏，might_sleep函數退化為 might_resched()函數。

(資料圖片)

# define might_sleep() \\ do { __might_sleep(__FILE__, __LINE__, 0); might_resched(); } while (0)# define sched_annotate_sleep() (current- >task_state_change = 0)#else  static inline void ___might_sleep(const char *file, int line,       int preempt_offset) { }  static inline void __might_sleep(const char *file, int line,       int preempt_offset) { }# define might_sleep() do { might_resched(); } while (0)# define sched_annotate_sleep() do { } while (0)

在配置了搶占式內核或者非搶占式內核的情況下，might_resched()函數最終都是空函數。如果配置了主動搶占式內核CONFIG_PREEMPT_VOLUNTARY，則might_resched()函數會調用 _cond_resched()函數來主動觸發一次搶占。

#ifdef CONFIG_PREEMPT_VOLUNTARYextern int _cond_resched(void);# define might_resched() _cond_resched()#else# define might_resched() do { } while (0)#endif#ifndef CONFIG_PREEMPTextern int _cond_resched(void);#elsestatic inline int _cond_resched(void) { return 0; }#endif

——cond_resched()函數調用should_resched()函數判斷搶占計數器是否為0，如果搶占計數器為0并且設置了重新調度標記則調用preempt_schedule_common()函數進行搶占式調度

#ifndef CONFIG_PREEMPTint __sched _cond_resched(void){ if (should_resched(0)) {  preempt_schedule_common();  return 1; } return 0;}EXPORT_SYMBOL(_cond_resched);#endif

__mutex_trylock_fast()函數調用atomic_long_cmpxchg_acquire()函數判斷lock->owner的值是否等于0，如果等于0，則直接將當前線程的task struct的指針賦值給lock->owner，表示該mutex鎖已經被當前線程持有。如果lock->owner的值不等于0，則表示該mutex鎖已經被其他線程持有或者鎖正在傳遞給top waiter線程，當前線程需要阻塞等待。上面描述的操作（比較和賦值）都是原子操作，不會有任何指令插入。

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

慢速獲取mutex鎖的路徑就是__mutex_lock_common()函數，所謂慢速其實就是阻塞當前線程，將current task掛入mutex的等待隊列的尾部。讓所有等待mutex的任務按照時間的先后順序排列起來，當mutex被釋放的時候，會首先喚醒隊首的任務，即最先等待的任務最先被喚醒。此外，在向空隊列插入第一個任務的時候，會給mutex flag設置上MUTEX_FLAG_WAITERS標記，表示已經有任務在等待這個mutex鎖了。

static noinline void __sched__mutex_lock_slowpath(struct mutex *lock){ __mutex_lock(lock, TASK_UNINTERRUPTIBLE, 0, NULL, _RET_IP_);}static int __sched__mutex_lock(struct mutex *lock, long state, unsigned int subclass,      struct lockdep_map *nest_lock, unsigned long ip){ return __mutex_lock_common(lock, state, subclass, nest_lock, ip, NULL, false);}static __always_inline int __sched__mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,      struct lockdep_map *nest_lock, unsigned long ip,      struct ww_acquire_ctx *ww_ctx, const bool use_ww_ctx){ struct mutex_waiter waiter; bool first = false; struct ww_mutex *ww; int ret; might_sleep(); ww = container_of(lock, struct ww_mutex, base); if (use_ww_ctx && ww_ctx) {  if (unlikely(ww_ctx == READ_ONCE(ww- >ctx)))   return -EALREADY; } preempt_disable(); mutex_acquire_nest(&lock- >dep_map, subclass, 0, nest_lock, ip); if (__mutex_trylock(lock) ||     mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx, NULL)) {  /* got the lock, yay! */  lock_acquired(&lock- >dep_map, ip);  if (use_ww_ctx && ww_ctx)   ww_mutex_set_context_fastpath(ww, ww_ctx);  preempt_enable();  return 0; } spin_lock(&lock- >wait_lock); /*  * After waiting to acquire the wait_lock, try again.  */ if (__mutex_trylock(lock)) {  if (use_ww_ctx && ww_ctx)   __ww_mutex_wakeup_for_backoff(lock, ww_ctx);  goto skip_wait; } debug_mutex_lock_common(lock, &waiter); debug_mutex_add_waiter(lock, &waiter, current); lock_contended(&lock- >dep_map, ip); if (!use_ww_ctx) {  /* add waiting tasks to the end of the waitqueue (FIFO): */  list_add_tail(&waiter.list, &lock- >wait_list);#ifdef CONFIG_DEBUG_MUTEXES  waiter.ww_ctx = MUTEX_POISON_WW_CTX;#endif } else {  /* Add in stamp order, waking up waiters that must back off. */  ret = __ww_mutex_add_waiter(&waiter, lock, ww_ctx);  if (ret)   goto err_early_backoff;  waiter.ww_ctx = ww_ctx; } waiter.task = current; if (__mutex_waiter_is_first(lock, &waiter))  __mutex_set_flag(lock, MUTEX_FLAG_WAITERS); set_current_state(state); for (;;) {  /*   * Once we hold wait_lock, we"re serialized against   * mutex_unlock() handing the lock off to us, do a trylock   * before testing the error conditions to make sure we pick up   * the handoff.   */  if (__mutex_trylock(lock))   goto acquired;  /*   * Check for signals and wound conditions while holding   * wait_lock. This ensures the lock cancellation is ordered   * against mutex_unlock() and wake-ups do not go missing.   */  if (unlikely(signal_pending_state(state, current))) {   ret = -EINTR;   goto err;  }  if (use_ww_ctx && ww_ctx && ww_ctx- >acquired > 0) {   ret = __ww_mutex_lock_check_stamp(lock, &waiter, ww_ctx);   if (ret)    goto err;  }  spin_unlock(&lock- >wait_lock);  schedule_preempt_disabled();  /*   * ww_mutex needs to always recheck its position since its waiter   * list is not FIFO ordered.   */  if ((use_ww_ctx && ww_ctx) || !first) {   first = __mutex_waiter_is_first(lock, &waiter);   if (first)    __mutex_set_flag(lock, MUTEX_FLAG_HANDOFF);  }  set_current_state(state);  /*   * Here we order against unlock; we must either see it change   * state back to RUNNING and fall through the next schedule(),   * or we must see its unlock and acquire.   */  if (__mutex_trylock(lock) ||      (first && mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx, &waiter)))   break;  spin_lock(&lock- >wait_lock); } spin_lock(&lock- >wait_lock);acquired: __set_current_state(TASK_RUNNING); mutex_remove_waiter(lock, &waiter, current); if (likely(list_empty(&lock- >wait_list)))  __mutex_clear_flag(lock, MUTEX_FLAGS); debug_mutex_free_waiter(&waiter);skip_wait: /* got the lock - cleanup and rejoice! */ lock_acquired(&lock- >dep_map, ip); if (use_ww_ctx && ww_ctx)  ww_mutex_set_context_slowpath(ww, ww_ctx); spin_unlock(&lock- >wait_lock); preempt_enable(); return 0;err: __set_current_state(TASK_RUNNING); mutex_remove_waiter(lock, &waiter, current);err_early_backoff: spin_unlock(&lock- >wait_lock); debug_mutex_free_waiter(&waiter); mutex_release(&lock- >dep_map, 1, ip); preempt_enable(); return ret;}

釋放鎖流程分析

釋放鎖的流程也分快速釋放和慢速釋放兩種路徑

void __sched mutex_unlock(struct mutex *lock){#ifndef CONFIG_DEBUG_LOCK_ALLOC if (__mutex_unlock_fast(lock))  return;#endif __mutex_unlock_slowpath(lock, _RET_IP_);}

如果一個線程獲取了mutex鎖之后，沒有其他的線程試圖獲取，此時的mutex的owner成員就是該線程的task struct地址，并且所有的mutex flag都沒有被設置。這時候只需要將mutex的owner成員清零即可，不需要其它操作，這就是快速釋放鎖的路徑。

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

如果有其他線程在競爭該mutex鎖，這時候就會進入慢速釋放鎖路徑，慢速釋放鎖路徑的邏輯分成兩段：一段是釋放mutex鎖，另外一段是喚醒top waiter線程。

static noinline void __sched __mutex_unlock_slowpath(struct mutex *lock, unsigned long ip){ struct task_struct *next = NULL; DEFINE_WAKE_Q(wake_q); unsigned long owner; mutex_release(&lock- >dep_map, 1, ip); /*  * Release the lock before (potentially) taking the spinlock such that  * other contenders can get on with things ASAP.  *  * Except when HANDOFF, in that case we must not clear the owner field,  * but instead set it to the top waiter.  */ owner = atomic_long_read(&lock- >owner); for (;;) {  unsigned long old;#ifdef CONFIG_DEBUG_MUTEXES  DEBUG_LOCKS_WARN_ON(__owner_task(owner) != current);  DEBUG_LOCKS_WARN_ON(owner & MUTEX_FLAG_PICKUP);#endif  if (owner & MUTEX_FLAG_HANDOFF)   break;  old = atomic_long_cmpxchg_release(&lock- >owner, owner,        __owner_flags(owner));  if (old == owner) {   if (owner & MUTEX_FLAG_WAITERS)    break;   return;  }  owner = old; } spin_lock(&lock- >wait_lock); debug_mutex_unlock(lock); 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;  debug_mutex_wake_waiter(lock, waiter);  wake_q_add(&wake_q, next); } if (owner & MUTEX_FLAG_HANDOFF)  __mutex_handoff(lock, next); spin_unlock(&lock- >wait_lock); wake_up_q(&wake_q);}

總結

本篇主要介紹了mutex互斥鎖的使用注意事項，介紹了mutex的主要函數接口以及獲取鎖的流程分析和釋放鎖的流程分析。通過本文的學習，我們基本可以了解了mutex的實現機制和使用方法。