一、Linux為何會引入讀寫鎖？

除了mutex，在linux內(nèi)核中，還有一個經(jīng)常用到的睡眠鎖就是rw semaphore（后文簡稱為rwsem），它到底和mutex有什么不同呢？為何會有rw semaphore？無他，僅僅是為了增加內(nèi)核的并發(fā)，從而增加性能而已。Mutex嚴格的限制只有一個thread可以進入臨界區(qū)，但是實際應用中，有些場景對共享資源的訪問可以嚴格區(qū)分讀和寫的，并且是讀多寫少，這時候，其實多個讀的thread同時進入臨界區(qū)是OK的，使用mutex則限制一個線程進入臨界區(qū)，從而導致性能的下降。

本文會描述linux5.15.81中讀寫鎖的數(shù)據(jù)結(jié)構(gòu)和邏輯過程。

二、如何抽象讀寫鎖的數(shù)據(jù)結(jié)構(gòu)？

下圖可以抽象rwsem相關(guān)的數(shù)據(jù)結(jié)構(gòu)：

一個rwsem對象需要記錄兩種數(shù)據(jù)：

1、讀寫鎖的狀態(tài)信息

2、和該讀寫鎖相關(guān)的任務信息

我們先看看讀寫鎖的狀態(tài)。讀寫鎖狀態(tài)字需要分別記錄讀鎖和寫鎖的狀態(tài)：由于多個reader可以同時處于臨界區(qū)，所以對于reader-owned的場景，讀鎖狀態(tài)變成了一個counter，來記錄臨界區(qū)內(nèi)reader的數(shù)量，counter等于0表示讀鎖為空鎖狀態(tài)。對于writer，其行為和互斥鎖一致，因此其寫鎖狀態(tài)和mutex一樣，仍然使用一個bit表示。

和讀寫相關(guān)的任務有兩類，一類是已經(jīng)持鎖的線程（即在臨界區(qū)的線程），另外一類是無法持鎖而需要等待的任務。對于writer持鎖情況，由于排他性，我們很清楚的知道是哪個task持鎖，那么一個task struct指針就足夠了記錄owner了。然而對于讀側(cè)可以多個reader進入臨界區(qū)，那么owner們需要組成一個隊列才可以記錄每一個臨界區(qū)的reader。

不過在實際的rwsem實現(xiàn)中，由于跟蹤owner們開銷比較大，因此也是用一個task struct指針指向其一。具體linux代碼是這樣處理的：reader進入的時候會設(shè)置owner task，但是離開讀臨界區(qū)并不會清除task指針。這樣，實際上對于讀，owner task應該表示該任務曾經(jīng)擁有該鎖，并不表示是目前持鎖的owner task，也有可能已經(jīng)離開臨界區(qū)，甚至該任務已經(jīng)銷毀。

如果持鎖失敗，無法進入臨界區(qū)，我們有兩種選擇：

1、樂觀自旋

2、掛入等待隊列

兩種選擇各有優(yōu)點和缺點，總結(jié)如下：

在5.15的內(nèi)核中，只有在write持鎖路徑上有樂觀自旋的操作，reader路徑?jīng)]有，只有偷鎖的操作。當樂觀自旋失敗后就會掛入等待隊列，阻塞當前線程。（樂觀自旋功能有一個很有意思的發(fā)展過程，從開始支持writer的樂觀自旋，到支持全場景的樂觀自旋，然后又回到最初，有興趣可以查閱內(nèi)核的patch了解詳情）

在了解了rwsem的基本概念之后，我們一起來看看struct rw_semaphore數(shù)據(jù)結(jié)構(gòu)，其成員描述如下：

由于是sleep lock，我們需要把等待的任務掛入隊列。在內(nèi)核中，struct rwsem_waiter用來抽象等待rwsem的任務，其成員描述如下：

三、Rwsem外部接口API為何？

Rwsem模塊的外部接口API如下：

四、嘗試獲取讀鎖

和down_read不一樣，down_read_trylock只是嘗試獲取讀鎖，如果成功，那么自然是好的，直接返回1，如果失敗，也不會阻塞，只是返回0就可以了。代碼主邏輯在__down_read_trylock函數(shù)中，如下：

A、tmp的初始值設(shè)定為RWSEM_UNLOCKED_VALUE（0值），因此第一次循環(huán)是為當前是空鎖而做的優(yōu)化：如果當前的sem->count等于0，那么給sem->count賦值RWSEM_READER_BIAS，標記持鎖成功，然后設(shè)定owner返回1即可。

B、如果快速獲取空鎖不成功，這時候tmp已經(jīng)賦值（等于sem->count），不再是0值了。通過對當前sem->count的值可以判斷是否是可以進入臨界區(qū)。持讀鎖失敗的情況包括：

如果判斷可以進入讀臨界區(qū)（臨界區(qū)僅有reader并且沒有writer等待的場景），那么重新進入循環(huán)，如果sem->count保持不變，那么可以持鎖成功，給進入臨界區(qū)的reader數(shù)目加一，并設(shè)置owner task和reader持鎖標記（non-spinnable比特保持不變）。如果這期間有其他線程插入修改了count值，那么需要再次判斷是否能持讀鎖，重復上面的循環(huán)。如果判斷不可以進入臨界區(qū)，退出循環(huán)，持鎖失敗。

五、獲取讀鎖

Reader獲取讀鎖的代碼主要在__down_read_common函數(shù)中，如下：

1、快速路徑

rwsem_read_trylock是快速路徑，代碼如下：

A、reader直接會給sem->count加RWSEM_READER_BIAS來增加讀臨界區(qū)的線程個數(shù)，當然這有可能失敗，那么就進入慢速路徑（需要回退錯誤增加讀臨界區(qū)線程數(shù)量）。如果恰好能夠進入臨界區(qū)，那么就直接設(shè)定owner返回即可。注意：這里*cntp保存了atomic add之后的新值。rwsem_down_read_slowpath會使用這個新值作為參數(shù)。

B、當reader的數(shù)量過多（以至于都溢出了）的時候，需要禁止樂觀自旋。

C、這里是持鎖成功的路徑。RWSEM_READ_FAILED_MASK上一節(jié)已經(jīng)解釋，這里不再贅述。這里需要注意的是rwsem_set_reader_owned函數(shù)中flag的設(shè)定，由于reader進入臨界區(qū)，因此RWSEM_READER_OWNED也需要設(shè)定。RWSEM_RD_NONSPINNABLE標記保持不變。

在快速路徑中，有兩種常見的情況會持鎖成功：一種是空鎖，另外一種是沒有任何waiter等待的純reader并發(fā)。

2、慢速路徑

如果快速路徑持鎖失敗，那么進入慢速路徑。慢速路徑代碼比較長，我們分段解析。首先是防止等待隊列中waiter任務餓死的代碼：

如果當前的鎖被reader持有（至少有一個reader在臨界區(qū)），那么不再樂觀偷鎖而是直接進行掛等待隊列的操作。為何怎么做呢？因為需要在餓死waiter和reader吞吐量上進行平衡。一方面，連續(xù)的reader持續(xù)偷鎖的話會餓死等待隊列上的任務。另外，在喚醒路徑上，被喚醒的top reader會順便將隊列中的若干（不大于256個）reader也同時喚醒，以便增加rwsem的吞吐量。所以這里的reader直接掛入隊列，累計多個reader以便可以批量喚醒。

Reader偷鎖的場景主要發(fā)生在喚醒top waiter的過程中，這時候臨界區(qū)沒有線程，被喚醒的reader或者writer也沒有持鎖（writer需要被調(diào)度到CPU上執(zhí)行之后才會試圖持鎖，高負載的場景下，鎖被偷的概率比較大，reader是喚醒后立刻持鎖，被偷的幾率小一點）。具體樂觀偷鎖（optimistic lock stealing）的代碼如下：

A、所謂偷鎖就是不樂觀自旋（要有排隊），不管先來后到，直接獲取鎖。允許偷鎖的場景是這樣的：臨界區(qū)沒有writer持鎖，也沒有設(shè)置handoff，正在喚醒top waiter的過程中，并且有任務在等待隊列的情況。這時候進入慢速路徑的reader可以先于top waiter喚醒之前把鎖偷走。需要特別說明的是：這時候reader counter已經(jīng)加一，還是盡量讓reader偷鎖成功，否則還需要回退。

B、當前線程獲得了讀鎖，需要設(shè)置owner，畢竟它是臨界區(qū)的新客

C、如果偷鎖成功并且它是臨界區(qū)第一個reader，那么它還會把等待隊列中的reader都喚醒（前提是top waiter不是writer），帶領(lǐng)大家一起往前沖（這里會打破FIFO的順序，懲罰了隊列中的writer）。具體是通過rwsem_mark_wake來標記喚醒的reader，然后通過wake_up_q將reader喚醒并進入讀臨界區(qū)。為了減低對等待中的writer線程的影響，這時候?qū)eader的并發(fā)是受限的，最多可以喚醒MAX_READERS_WAKEUP個reader。

如果偷鎖不成功，當前的reader還是需要進入阻塞狀態(tài)：

A、準備好掛入等待隊列的rwsem waiter數(shù)據(jù)，需要特別說明的是這里的timeout時間：目前手機平臺的HZ設(shè)置的是250，也就是說在觸發(fā)handoff機制之前waiter需要至少在隊列中等待一個tick（4ms）的時間。這里的timeout是指handoff timeout，為了防止偷鎖或者自旋導致等待隊列中的top waiter有一個長時間的持鎖延遲。在timeout時間內(nèi)，樂觀偷鎖或者自旋可以順利進行，但是一旦超時就會設(shè)定handoff標記，樂觀偷鎖或者自旋被禁止，鎖的所有權(quán)需要遞交給等待隊列中的top waiter。

B、如果目前等待隊列為空，那么要做一些額外的處理。例如入隊之前肯定給安排上RWSEM_FLAG_WAITERS這個標記。

C、當然，在入隊之前還要垂死掙扎一下（等待隊列為空的時候邏輯簡單一些，不需要喚醒隊列上的wait），看看是不是當前有機可乘，如果是這樣，那么就順勢而為，直接持鎖成功，而且counter都已經(jīng)準備好了，前面已經(jīng)加一了。

D、等待隊列非空的時候，邏輯稍微負載一點。調(diào)用rwsem_add_waiter函數(shù)即可以把當前任務掛入等待隊列尾部。這時候也需要把之前武斷增加的counter給修正回來了（adjustment初始化為-RWSEM_READER_BIAS）。如果是第一個waiter，也順便設(shè)置了RWSEM_FLAG_WAITERS標記。

在當前線程進入阻塞之前，我們需要進行試圖持鎖的動作（上面是空隊列場景檢查，這里的邏輯稍微復雜一點，由于已經(jīng)入隊，這里需要調(diào)用rwsem_mark_wake函數(shù)來完成阻塞后喚醒的動作），畢竟這時候可能恰好owner離開臨界區(qū)，變成空鎖。

A、如果這時候發(fā)現(xiàn)鎖的owner恰好都離開了臨界區(qū)，那么我們是需要執(zhí)行喚醒top waiter操作的，喚醒之前需要清除禁止樂觀自旋的標記，畢竟目前臨界區(qū)沒有任何線程。

B、除了上面說的場景需要喚醒，在reader持鎖并且我們是隊列中的第一個waiter的時候，也需要喚醒的動作（喚醒自己）。

阻塞部分的代碼邏輯如下：

A、在rwsem_mark_wake函數(shù)中我們會喚醒reader并將其等待對象的task成員（waiter.task）設(shè)置為NULL。因此，這里如果發(fā)現(xiàn)waiter.task等于NULL，那么說明是該線程被正常喚醒，那么從阻塞狀態(tài)返回，持鎖成功。

B、如果在該線程阻塞的時候，有其他任務發(fā)送信號給該線程，那么就持鎖失敗退出。如果已經(jīng)被喚醒，同時又收到信號，這時候需要首先完成喚醒，持鎖成功，然后在其他的合適點再處理該信號。當然，大部分的rwsem都是D狀態(tài)，也就不需要處理信號了。

C、進入阻塞狀態(tài)，讓調(diào)度器選擇next task

六、釋放讀鎖

釋放讀鎖的代碼邏輯主要在__up_read函數(shù)中，如下：

需要強調(diào)的是：這里僅僅是減去了讀臨界區(qū)的counter計數(shù)，并沒有清除owner中的task pointer。此外，當?shù)却犃杏衱aiter并且沒有writer或者reader在臨界區(qū)的時候，我們會調(diào)用rwsem_wake來喚醒等待隊列的線程。因為臨界區(qū)已經(jīng)沒有線程，所以需要清除nonspinable標記。喚醒的動作主要是通過rwsem_mark_wake和wake_up_q來完成的，wake_up_q比較簡單，我們就不贅述了，主要看看rwsem_mark_wake的邏輯。

我們首先給出wake type的解釋：

在RWSEM_WAKE_READERS場景中，多個reader被喚醒，并且當前很可能是空鎖狀態(tài)，為了防止writer搶鎖，因此會先讓top waiter持有讀鎖，然后慢慢處理后續(xù)。RWSEM_WAKE_READ_OWNED則沒有這個顧慮，因為喚醒者已經(jīng)持有讀鎖。

在釋放讀鎖的場景中，rwsem_mark_wake使用的是RWSEM_WAKE_ANY參數(shù)，具體的代碼如下：

這段代碼是處理top waiter是writer的邏輯。這時候，如果wake type是RWSEM_WAKE_ANY，即不關(guān)心喚醒的是reader還是writer，只要喚醒等待隊列頭部的waiter就好。如果top waiter是writer，我們只需要將這個writer喚醒即可，不需要修改鎖的狀態(tài)，出隊等操作，這些都是在喚醒之后完成。如果wake type是其他兩種類型（都是喚醒reader的），那么就直接返回。也就是說在rwsem_mark_wake想要喚醒reader的場景中，如果top waiter是writer，那么將不會喚醒任何reader線程。如果top waiter是reader的話，那么基本上是需要喚醒一組reader了。

A、執(zhí)行到這里，我們需要喚醒等待隊列頭部的若干reader線程去持鎖。由于writer有可能會在這個階段偷鎖，因此，這里我們會先讓top waiter（reader）持鎖，然后再慢慢去計算到底需要喚醒多少個reader并將其喚醒。如果當前線程已經(jīng)持有了讀鎖（wake type的類型是RWSEM_WAKE_READ_OWNED），則不需要提前持鎖，直接越過這部分的邏輯即可。

B、如果的確發(fā)生了writer通過樂觀自旋偷鎖，那么我們需要檢查設(shè)置handoff的條件。如果reader被writer阻塞太久，那么我們設(shè)定handoff標記，要求rwsem的writer停止通過樂觀自旋偷鎖，將鎖的所有權(quán)轉(zhuǎn)交給top waiter（reader）

C、上面已經(jīng)向rwsem的count增加reader計數(shù)，這里把owner也設(shè)定上（flag也同步安排，這里non-spinnable bit保持不變）。隨后top waiter的reader會喚醒若干隊列中的non top reader，但是它們都不配擁有名字。

讀鎖已經(jīng)安排的妥妥的了，下面就是慢慢喚醒等待隊列的reader了。我們通過兩步來完成喚醒：

1、將等待隊列中的reader摘下放入到一個單獨的列表中（wlist），同時對reader進行計數(shù)。后續(xù)這個計數(shù)會寫入rwsem 的reader counte域。

2、對于wlist中的每一個waiter對象（reader任務），清除waiter->task并將它們放入wake_q以便稍后被喚醒。

我們先看第一輪計算喚醒reader個數(shù)的計數(shù)：

A、對于rwsem，其公平性是區(qū)分讀寫的。對于讀，如果top waiter是reader，那么所有的reader都可以進入臨界區(qū)，不管reader在隊列中的順序。對于writer，我們要確保其公平性，我們要按照writer在隊列中的順序依次持鎖。根據(jù)上面的原則，我們會略過隊列中的writer，將盡量多的reader喚醒并進入臨界區(qū)

B、喚醒數(shù)量不能大于256，否則會餓死writer

C、根據(jù)喚醒的reader數(shù)量計算count調(diào)整值