背景 By 魯迅 By 高爾基說明： 1. Kernel版本：4.14 2. ARM64處理器，Contex A53，雙核 3. 使用工具：Source Insight 3.5， Visio 1. 概述信號量，是操作系統中一種常用的同步與互斥的機制；信號量允許多個進程（計數值 1）同時進入臨 ...

背景

Read the fucking source code! --By 魯迅
A picture is worth a thousand words. --By 高爾基

說明：

Kernel版本：4.14
ARM64處理器，Contex-A53，雙核
使用工具：Source Insight 3.5， Visio

1. 概述

信號量semaphore，是操作系統中一種常用的同步與互斥的機制；
信號量允許多個進程（計數值>1）同時進入臨界區；
如果信號量的計數值為1，一次只允許一個進程進入臨界區，這種信號量叫二值信號量；
信號量可能會引起進程睡眠，開銷較大，適用於保護較長的臨界區；
與讀寫自旋鎖類似，linux內核也提供了讀寫信號量的機制；

本文將分析信號量與讀寫信號量的機制，開始吧。

2. 信號量

2.1 流程分析

可以將信號量比喻成一個盒子，初始化時在盒子里放入N把鑰匙，鑰匙先到先得，當N把鑰匙都被拿走完後，再來拿鑰匙的人就需要等待了，只有等到有人將鑰匙歸還了，等待的人才能拿到鑰匙；

信號量的實現很簡單，先看一下數據結構：

struct semaphore {
	raw_spinlock_t		lock;       //自旋鎖，用於count值的互斥訪問
	unsigned int		count;      //計數值，能同時允許訪問的數量，也就是上文中的N把鎖
	struct list_head	wait_list;      //不能立即獲取到信號量的訪問者，都會加入到等待列表中
};

struct semaphore_waiter {
	struct list_head list;      //用於添加到信號量的等待列表中
	struct task_struct *task;   //用於指向等待的進程，在實際實現中，指向current
	bool up;                    //用於標識是否已經釋放
};

流程如下：

down介面用於獲取信號量，up用於釋放信號量；
調用down時，如果sem->count > 0時，也就是盒子裡邊還有多餘的鎖，直接自減並返回了，當sem->count == 0時，表明盒子裡邊的鎖被用完了，當前任務會加入信號量的等待列表中，設置進程的狀態，並調用schedule_timeout來睡眠指定時間，實際上這個時間設置的無限等待，也就是只能等著被喚醒，當前任務才能繼續運行；
調用up時，如果等待列表為空，表明沒有多餘的任務在等待信號量，直接將sem->count自加即可。如果等待列表非空，表明有任務正在等待信號量，那就需要對等待列表中的第一個任務（等待時間最長）進行喚醒操作，並從等待列表中將需要被喚醒的任務進行刪除操作；

2.2 信號量缺點

對比下《Linux Mutex機制分析》說過的Mutex，Semaphore與Mutex在實現上有一個重大的區別：ownership。Mutex被持有後有一個明確的owner，而Semaphore並沒有owner，當一個進程阻塞在某個信號量上時，它沒法知道自己阻塞在哪個進程（線程）之上；
沒有ownership會帶來以下幾個問題：
1. 在保護臨界區的時候，無法進行優先順序反轉的處理；
2. 系統無法對其進行跟蹤斷言處理，比如死鎖檢測等；
3. 信號量的調試變得更加麻煩；

因此，在Mutex能滿足要求的情況下，優先使用Mutex。

2.3 其他介面

信號量提供了多種不同的信號量獲取的介面，介紹如下：

/* 未獲取信號量時，進程輕度睡眠： TASK_INTERRUPTIBLE */
int down_interruptible(struct semaphore *sem)
/* 未獲取到信號量時，進程中度睡眠： TASK_KILLABLE */
int down_killable(struct semaphore *sem)
/* 非等待的方式去獲取信號量 */
int down_trylock(struct semaphore *sem)
/* 獲取信號量，並指定等待時間 */
int down_timeout(struct semaphore *sem, long timeout)

3. 讀寫信號量

【原創】linux spinlock/rwlock/seqlock原理剖析（基於ARM64）文章中，我們分析過讀寫自旋鎖，讀寫信號量的功能類似，它能有效提高併發性，我們先明確下它的特點：

允許多個讀者同時進入臨界區；
讀者與寫者不能同時進入臨界區（讀者與寫者互斥）；
寫者與寫者不能同時進入臨界區（寫者與寫者互斥）；

3.1 數據結構

讀寫信號量的數據結構與信號量的結構比較相似：

struct rw_semaphore {
	atomic_long_t count;        //用於表示讀寫信號量的計數
	struct list_head wait_list;     //等待列表，用於管理在該信號量上睡眠的任務
	raw_spinlock_t wait_lock;   //鎖，用於保護count值的操作
#ifdef CONFIG_RWSEM_SPIN_ON_OWNER
	struct optimistic_spin_queue osq; /* spinner MCS lock */    //MCS鎖，參考上一篇文章Mutex中的介紹
	/*
	 * Write owner. Used as a speculative check to see
	 * if the owner is running on the cpu.
	 */
	struct task_struct *owner;      //當寫者成功獲取鎖時，owner會指向鎖的持有者
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
	struct lockdep_map	dep_map;
#endif
};

最關鍵的需要看一下count欄位，掌握了這個欄位的處理，才能比較好理解讀寫信號量的機制；
【原創】linux spinlock/rwlock/seqlock原理剖析（基於ARM64）文章中提到過讀寫自旋鎖，讀寫自旋鎖中的lock欄位，bit[31]用於寫鎖的標記，bit[30:0]用於讀鎖的統計，而讀寫信號量的count欄位也大體類似；

以32位的count值為例，高16bit代表的是waiting part，低16bit代表的是active part；
RWSEM_UNLOCKED_VALUE：值為0，表示鎖未被持有，沒有讀者也沒有寫者；
RWSEM_ACTIVE_BIAS：值為1，，該值用於定義RWSEM_ACTIVE_READ_BIAS和RWSEM_ACTIVE_WRITE_BIAS；
RWSEM_WAITING_BIAS：值為-65536，當有任務需要加入到等待列表中時，count值需要加RWSEM_WAITING_BIAS，有任務需要從等待列表中移除時，count值需要減去RWSEM_WAITING_BIAS；
RWSEM_ACTIVE_READ_BIAS：值為1，當有讀者去獲取鎖的時候，count值將加RWSEM_ACTIVE_READ_BIAS，釋放鎖的時候，count值將減去RWSEM_ACTIVE_READ_BIAS；
RWSEM_ACTIVE_WRITE_BIAS，值為-65535，當有寫者去獲取鎖的時候，count值將加RWSEM_ACTIVE_WRITE_BIAS，釋放鎖的時候，count值需要減去RWSEM_ACTIVE_WRITE_BIAS；

在獲取釋放讀鎖和寫鎖的全過程中，count值伴隨著上述這幾個巨集定義的加減操作，用於標識不同的狀態，可以羅列如下：

0x0000000X：活躍的讀者和正在申請讀鎖的讀者總共為X個，沒有寫者來干擾；
0x00000000：沒有讀者和寫者來操作，初始化狀態；
0xFFFF000X：分為以下幾種情況：
1. 0xFFFF000X = RWSEM_WAITING_BIAS + X * RWSEM_ACTIVE_READ_BIAS，表示活躍的讀者和正在申請讀鎖的讀者總共有X個，並且還有一個寫者在睡眠等待；
2. 0xFFFF000X = RWSEM_ACTIVE_WRITE_BIAS + (X - 1)* RWSEM_ACTIVE_READ_BIAS，表示有一個寫者在嘗試獲取鎖，活躍的讀者和正在申請讀鎖的讀者總共有X-1個；
0xFFFF0001：分為以下幾種情況：
1. 0xFFFF0001 = RWSEM_ACTIVE_WRITE_BIAS，有一個活躍的寫者，或者寫者正在嘗試獲取鎖，沒有讀者干擾；
2. 0xFFFF0001 = RWSEM_ACTIVE_READ_BIAS + RWSEM_WAITING_BIAS，有個寫者正在睡眠等待，還有一個活躍或嘗試獲取鎖的讀者；

3.1 讀信號量

3.1.1 讀者獲取鎖

特點：讀者與讀者可以併發執行，讀者與寫者互斥執行，因此當有寫者持有鎖的時候，讀者將進入睡眠狀態；
當sem->count加1後還是小於0，代表鎖已經被寫者持有了，讀者獲取鎖失敗，進入rwsem_down_read_failed函數；
如果sem->wait_list是空時，代表沒有任務在等待列表中，首次加入時，sem->count值需要加上RWSEM_WAITING_BIAS，表示有任務在等待列表中；
如果此時sem->count == RWSEM_WAITING_BIAS或者count > RWSEM_WAITING_BIAS && adjustment != RWSEM_ACTIVE_READ_BIAS，表示此時寫者將鎖釋放了，因此需要去喚醒在等待列表中的任務；
如果寫者沒有釋放鎖，那就進入迴圈，並調用schedule讓出CPU，直到鎖被釋放了，那麼從代碼流程中看，只有!waiter.task時才會跳出迴圈，也就是waiter.task == NULL時，才是獲取成功，這個操作是在__rwsem_mark_wake中通過smp_store_release(&waiter->task, NULL)實現的；
在等待獲取鎖的迴圈中，需要對信號進行處理，如果對應的等待任務沒被喚醒，那麼直接跳轉到out_nolock處，接下來的處理就是一些逆操作了，包括從等待列表中刪除，如果是等待列表中的首個任務，還需要減去RWSEM_WAITING_BIAS等；

總結一下：
讀者獲取鎖的時候，如果沒有寫者持有，那就可以支持多個讀者直接獲取；而如果此時寫者持有了鎖，讀者獲取失敗，它將把自己添加到等待列表中，（這個等待列表中可能已經存放了其他來獲取鎖的讀者或者寫者），在將讀者真正睡眠等待前，還會再一次判斷此時是否有寫者釋放了該鎖，釋放了的話，那就需要對睡眠等待在該鎖的任務進行喚醒操作了

3.1.2 讀者釋放鎖

釋放鎖的時候sem->count值進行減1操作；
減1操作之後得到的count值小於-1，並且active part是全零，代表等待列表中有寫任務在睡眠等待，因此需要進行喚醒操作；
喚醒操作中，如果有自旋等待的任務，那就可以直接返回了，畢竟人家在自旋呢，又沒有睡眠；
沒有自旋等待任務，那就去喚醒等待列表中的任務了；

3.2 寫信號量

3.2.1 寫者獲取鎖

寫者的特點：看誰都不順眼，跟誰都互斥，有我沒你。只要有一個寫者在持有鎖，其他的讀者與寫者都無法獲取；
在寫者獲取鎖的時候，將sem->count值加上RWSEM_ACTIVE_WRITE_BIAS，如果這個值不等於RWSEM_ACTIVE_WRITE_BIAS，表示有其他的讀者或寫者持有鎖，因此獲取鎖失敗，調用rwsem_down_write_failed來處理；
調用rwsem_optimistic_spin進行樂觀自旋去嘗試獲取鎖，獲取了的話，則直接返回，optimistic spin可以參考《Linux Mutex機制分析》文章中的分析，它的作用也是性能的優化，認為鎖的持有者會很快釋放，因此當前進程選擇自旋而不是讓出CPU，減少上下文切換帶來的開銷；
如果等待列表中有讀者任務在睡眠等待，此時假如寫者釋放了鎖，那麼需要先將讀者任務都給喚醒了；如果等待列表中沒有任務，也就意味著當前的寫者是第一個任務，因此將sem->count值加上RWSEM_WAITING_BIAS；
迴圈等待獲取鎖，這個過程與down_read是類似的；

總結
寫者獲取鎖時，只要鎖被其他讀者或者寫者持有了，則獲取鎖失敗，然後進行失敗情況處理。在失敗情況下，它本身會嘗試進行optimistic spin去嘗試獲取鎖，如果獲取成功了，那就是皆大歡喜了，否則還是需要進入慢速路徑。慢速路徑中去判斷等待列表中是否有任務在睡眠等待，並且會再次嘗試去查看是否已經有寫者釋放了鎖，寫者釋放了鎖，並且只有讀者在睡眠等待，那麼此時應該優先讓這些先等待的任務喚醒

3.2.2 寫者釋放鎖

寫者釋放鎖的時候，有一個關鍵的操作，將sem->owner進行清零操作，在寫者獲取鎖的時候會將該值設置成持有鎖的進程；
釋放鎖的時候，需要減去RWSEM_ACTIVE_WRITE_BIAS，然後再去判斷值，如果此時還有任務在睡眠等待，那就進行喚醒操作；

3.3 總結

理解讀寫信號量有幾個關鍵點：

讀寫信號量的特性可以與讀寫自旋鎖進行類比（讀者與讀者併發、讀者與寫者互斥、寫者與寫者互斥），區別在於讀寫信號量可能會發生睡眠，進而帶來進程切換的開銷；
為了優化讀寫信號量的性能，引入了MCS鎖機制，進一步減少切換開銷。第一個寫者獲取了鎖後，第二個寫者去獲取時自旋等待，而讀者去獲取時則會進入睡眠；
讀寫信號量的count值很關鍵，代表著讀寫信號量不同狀態的切換，因此也決定了執行流程；
讀者或寫者釋放鎖的時候，去喚醒等待列表中的任務，需要分情況處理。等待列表中可能存放的是讀者與寫者的組合，如果第一個任務是寫者，則直接喚醒該寫者，否則將喚醒排在前邊的連續幾個讀者；

參考

Real-world Concurrency

歡迎關註公眾號，不定期分享內核機制文章

【原創】Linux信號量機制分析

背景