1 定義
互斥鎖（mutex）是一種用于多線程編程的機(jī)制，用于防止多條線程同時(shí)對(duì)同一公共資源進(jìn)行讀寫(xiě)操作。
為了達(dá)到這個(gè)目的，互斥鎖將代碼劃分為臨界區(qū)域（critical section），這部分代碼涉及對(duì)公共資源的讀寫(xiě)操作。一個(gè)程序、進(jìn)程或線程可以擁有多個(gè)臨界區(qū)域，但并不一定都需要應(yīng)用互斥鎖。
舉例來(lái)說(shuō)，如果一條線程正在修改數(shù)據(jù)，而另一條線程被喚醒并嘗試讀取這些數(shù)據(jù)，那么就會(huì)導(dǎo)致數(shù)據(jù)的狀態(tài)不確定，甚至可能導(dǎo)致數(shù)據(jù)損壞。為了保護(hù)多個(gè)線程共享的數(shù)據(jù)，必須確保同一時(shí)間只有一個(gè)臨界區(qū)域處于運(yùn)行狀態(tài)，其他的臨界區(qū)域必須被掛起并無(wú)法獲得運(yùn)行機(jī)會(huì)。
互斥鎖實(shí)現(xiàn)多線程同步的核心思想是，當(dāng)一個(gè)線程訪問(wèn)公共資源時(shí)，這個(gè)線程會(huì)執(zhí)行“加鎖”操作，阻止其他線程訪問(wèn)該資源。訪問(wèn)完成后，線程會(huì)執(zhí)行“解鎖”操作，讓其他線程可以訪問(wèn)資源。當(dāng)多個(gè)線程競(jìng)爭(zhēng)訪問(wèn)資源時(shí)，只有最先執(zhí)行“加鎖”操作的線程可以訪問(wèn)資源。
在多個(gè)線程競(jìng)爭(zhēng)訪問(wèn)已被鎖定的公共資源時(shí)，其他線程只能等待資源解鎖，形成一個(gè)等待隊(duì)列。一旦資源被解鎖，操作系統(tǒng)會(huì)喚醒等待隊(duì)列中的線程，第一個(gè)獲得資源鎖的線程將開(kāi)始訪問(wèn)資源，而其他線程則繼續(xù)等待。
mutex有什么缺點(diǎn)？
不同于mutex最初的設(shè)計(jì)與目的，現(xiàn)在的struct mutex是內(nèi)核中最大的鎖之一，比如在x86-64上，它差不多有32bytes的大小，而struct samaphore是24bytes，rw_semaphore為40bytes，更大的數(shù)據(jù)結(jié)構(gòu)意味著占用更多的CPU緩存和更多的內(nèi)存占用。
什么時(shí)候應(yīng)該使用mutex？
除非mutex的嚴(yán)格語(yǔ)義要求不合適或者臨界區(qū)域阻止鎖的共享，否則相較于其他鎖原語(yǔ)來(lái)說(shuō)更傾向于使用mutex
mutex與spinlock的區(qū)別？
spinlock是讓一個(gè)嘗試獲取它的線程在一個(gè)循環(huán)中等待的鎖，線程在等待時(shí)會(huì)一直查看鎖的狀態(tài)。而mutex是一個(gè)可以讓多個(gè)進(jìn)程輪流分享相同資源的機(jī)制
spinlock通常短時(shí)間持有，mutex可以長(zhǎng)時(shí)間持有
spinlock任務(wù)在等待鎖釋放時(shí)不可以睡眠，mutex可以
看到一個(gè)非常有意思的解釋：
spinlock就像是坐在車后座的熊孩子，一直問(wèn)”到了嗎？到了嗎？到了嗎？…“
mutex就像一個(gè)司機(jī)返回的信號(hào)，說(shuō)”我們到了！“
2 實(shí)現(xiàn)
看一下Linux kernel-5.8是如何實(shí)現(xiàn)mutex的2 實(shí)現(xiàn)
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_head??wait_list; #ifdef?CONFIG_DEBUG_MUTEXES ??void??????*magic; #endif #ifdef?CONFIG_DEBUG_LOCK_ALLOC ??struct?lockdep_map??dep_map; #endif };
可以看到，mutex使用了原子變量owner來(lái)追蹤鎖的狀態(tài)，owner實(shí)際上是指向當(dāng)前mutex鎖擁有者的struct task_struct *指針，所以當(dāng)鎖沒(méi)有被持有時(shí)，owner為NULL。
/* ?*?This?is?the?control?structure?for?tasks?blocked?on?mutex, ?*?which?resides?on?the?blocked?task's?kernel?stack: ?*?表示等待隊(duì)列wait_list中進(jìn)程的結(jié)構(gòu)體 ?*/ struct?mutex_waiter?{ ??struct?list_head??list; ??struct?task_struct??*task; ??struct?ww_acquire_ctx??*ww_ctx; #ifdef?CONFIG_DEBUG_MUTEXES ??void??????*magic; #endif };
上鎖
當(dāng)要獲取mutex時(shí)，通常有三種路徑方式
fastpath：通過(guò) cmpxchg() 當(dāng)前任務(wù)與所有者來(lái)嘗試原子性的獲取鎖。這僅適用于無(wú)競(jìng)爭(zhēng)的情況（cmpxchg() 檢查 0UL，因此上面的所有 3 個(gè)狀態(tài)位都必須為 0）。如果鎖被爭(zhēng)用，它會(huì)轉(zhuǎn)到下一個(gè)可能的路徑。
midpath：又名樂(lè)觀旋轉(zhuǎn)（optimistic spinning）—在鎖的持有者正在運(yùn)行并且沒(méi)有其他具有更高優(yōu)先級(jí)（need_resched）的任務(wù)準(zhǔn)備運(yùn)行時(shí)，通過(guò)旋轉(zhuǎn)來(lái)獲取鎖。理由是如果鎖的所有者正在運(yùn)行，它很可能很快就會(huì)釋放鎖。mutex spinner使用 MCS 鎖排隊(duì)，因此只有一個(gè)spinner可以競(jìng)爭(zhēng)mutex。
MCS 鎖（由 Mellor-Crummey 和 Scott 提出）是一個(gè)簡(jiǎn)單的自旋鎖，具有公平的理想屬性，每個(gè) cpu 都試圖獲取在本地變量上旋轉(zhuǎn)的鎖，排隊(duì)采用的是鏈表實(shí)現(xiàn)的FIFO。它避免了常見(jiàn)的test-and-set自旋鎖實(shí)現(xiàn)引起的昂貴的cacheline bouncing。類似MCS的鎖是專門為睡眠鎖的樂(lè)觀旋轉(zhuǎn)而量身定制的（畢竟如果只是短暫的自旋比休眠效率要高）。自定義 MCS 鎖的一個(gè)重要特性是它具有額外的屬性，即當(dāng)spinner需要重新調(diào)度時(shí)，它們能夠直接退出 MCS 自旋鎖隊(duì)列。這有助于避免需要重新調(diào)度的 MCS spinner持續(xù)在mutex持有者上自旋，而僅需直接進(jìn)入慢速路徑獲取MCS鎖。
slowpath：最后的手段，如果仍然無(wú)法獲得鎖，則將任務(wù)添加到等待隊(duì)列并休眠，直到被解鎖路徑喚醒。在正常情況下它阻塞為 TASK_UNINTERRUPTIBLE。
雖然正式的內(nèi)核互斥鎖是可休眠的鎖，但midpath路徑 (ii) 使它們更實(shí)際地成為混合類型。通過(guò)簡(jiǎn)單地不中斷任務(wù)并忙于等待幾個(gè)周期而不是立即休眠，此鎖的性能已被視為顯著改善了許多工作負(fù)載。請(qǐng)注意，此技術(shù)也用于 rw 信號(hào)量。
具體代碼調(diào)用鏈很長(zhǎng)…
/*不可中斷的獲取鎖*/ void?__sched?mutex_lock(struct?mutex?*lock) { ????might_sleep(); ????/*fastpath*/ ????if?(!__mutex_trylock_fast(lock)) ????????/*midpath?and?slowpath*/ ????????__mutex_lock_slowpath(lock); } __mutex_trylock_fast(lock)?->?atomic_long_try_cmpxchg_acquire(&lock->owner,?&zero,?curr)?->?atomic64_try_cmpxchg_acquire(v,?(s64?*)old,?new); __mutex_lock_slowpath(lock)->__mutex_lock(lock,?TASK_UNINTERRUPTIBLE,?0,?NULL,?_RET_IP_)?->??__mutex_lock_common(lock,?state,?subclass,?nest_lock,?ip,?NULL,?false) ?? ?? /*可中斷的獲取鎖*/ int?mutex_lock_interruptible(struct?mutex?*lock);
嘗試上鎖
int?__sched?mutex_trylock(struct?mutex?*lock) { ??bool?locked; #ifdef?CONFIG_DEBUG_MUTEXES ??DEBUG_LOCKS_WARN_ON(lock->magic?!=?lock); #endif ??locked?=?__mutex_trylock(lock); ??if?(locked) ????mutex_acquire(&lock->dep_map,?0,?1,?_RET_IP_); ??return?locked; } static?inline?bool?__mutex_trylock(struct?mutex?*lock) { ??return?!__mutex_trylock_or_owner(lock); }
釋放鎖
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_); }
跟加鎖對(duì)稱，也有fastpath, midpath, slowpath三條路徑。
判斷鎖狀態(tài)
bool?mutex_is_locked(struct?mutex?*lock) { ??return?__mutex_owner(lock)?!=?NULL; }
很顯而易見(jiàn)，mutex持有者不為NULL即表示鎖定狀態(tài)。
3 實(shí)際案例
實(shí)驗(yàn)：
#include? #include? #define?LOOP?1000000 int?cnt?=?0; int?cs1?=?0,?cs2?=?0; void*?task(void*?args)?{ ????while(1) ????{ ????????if(cnt?>=?LOOP) ????????{ ????????????break; ????????} ????????cnt++; ????????if((int)args?==?1)?cs1?++;?else?cs2++; ????} ????return?NULL; } int?main()?{ ????pthread_t?tid1; ????pthread_t?tid2; ????/*?create?the?thread?*/ ????pthread_create(&tid1,?NULL,?task,?(void*)1); ????pthread_create(&tid2,?NULL,?task,?(void*)2); ????/*?wait?for?thread?to?exit?*/ ????pthread_join(tid1,?NULL); ????pthread_join(tid2,?NULL); ????printf("cnt?=?%d?cs1=%d?cs2=%d?total=%d ",?cnt,cs1,cs2,cs1+cs2); ????return?0; }
輸出：
cnt?=?1000000?cs1=958560?cs2=1520226?total=2478786
正確結(jié)果不應(yīng)該是1000000嗎？為什么會(huì)出錯(cuò)呢，我們可以從匯編角度來(lái)分析一下。
$>?g++?-E?test.c?-o?test.i $>?g++?-S?test.i?-o?test.s $>?vim?test.s ?? .file?"test.c" ??.globl??_cnt ??.bss ??.align?4 _cnt: ??.space?4 ??.text ??.globl??__Z5task1Pv ??.def??__Z5task1Pv;??.scl??2;??.type?32;?.endef __Z5task1Pv: ??...
我們可以看到一個(gè)簡(jiǎn)單的cnt++，對(duì)應(yīng)
movl??_cnt,?%eax addl??$1,?%eax movl??%eax,?_cnt
CPU先將cnt的值讀到寄存器eax中，然后將[eax] + 1，最后將eax的值返回到cnt中，這些操作不是**原子性質(zhì)(atomic)**的，這就導(dǎo)致cnt被多個(gè)線程操作時(shí)，+1過(guò)程會(huì)被打斷。
加入mutex保護(hù)臨界資源
#include? #include? #define?LOOP?1000000 pthread_mutex_t?mutex; int?cnt?=?0; int?cs1?=?0,?cs2?=?0; void*?task(void*?args)?{ ????while(1) ????{ ????????pthread_mutex_lock(&mutex); ????????if(cnt?>=?LOOP) ????????{ ????????????pthread_mutex_unlock(&mutex); ????????????break; ????????} ????????cnt++; ????????pthread_mutex_unlock(&mutex); ????????if((int)args?==?1)?cs1?++;?else?cs2++; ????} ????return?NULL; } int?main()?{ ????pthread_mutex_init(&mutex?,?NULL); ????pthread_t?tid1; ????pthread_t?tid2; ????/*?create?the?thread?*/ ????pthread_create(&tid1,?NULL,?task,?(void*)1); ????pthread_create(&tid2,?NULL,?task,?(void*)2); ????/*?wait?for?thread?to?exit?*/ ????pthread_join(tid1,?NULL); ????pthread_join(tid2,?NULL); ????printf("cnt?=?%d?cs1=%d?cs2=%d?total=%d ",?cnt,cs1,cs2,cs1+cs2); ????return?0; } 輸出： cnt?=?1000000?cs1=517007?cs2=482993?total=1000000
.png)
推廣.jpg)