如何進行l(wèi)inux futex淺析

如何進行l(wèi)inux futex淺析，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

Futex，F(xiàn)ast Userspace muTEXes，作為linux下的一種快速同步（互斥）機制，已經(jīng)存在了很長一段時間了（since linux 2.5.7）。它有什么優(yōu)勢？又提供了怎樣一些功能，

futex誕生之前

在futex誕生之前，linux下的同步機制可以歸為兩類：用戶態(tài)的同步機制 和 內(nèi)核同步機制。 用戶態(tài)的同步機制基本上就是利用原子指令實現(xiàn)的spinlock。最簡單的實現(xiàn)就是使用一個整型數(shù)，0表示未上鎖，1表示已上鎖。trylock操作就利用原子指令嘗試將0改為1：

bool trylock(int lockval) {int old;
    atomic { old = lockval; lockval = 1; }  // 如：x86下的xchg指令return old == 0;
}

無論spinlock事先有沒有被上鎖，經(jīng)歷trylock之后，它肯定是已經(jīng)上鎖了。所以lock變量一定被置1。而trylock是否成功，取決于spinlock是事先就被上了鎖的（old==1），還是這次trylock上鎖的(old==0)。而使用原子指令則可以避免多個進程同時看到old==0，并且都認為是自己改它改為1的。

spinlock的lock操作則是一個死循環(huán)，不斷嘗試trylock，直到成功。
對于一些很小的臨界區(qū)，使用spinlock是很高效的。因為trylock失敗時，可以預期持有鎖的線程（進程）會很快退出臨界區(qū)（釋放鎖）。所以死循環(huán)的忙等待很可能要比進程掛起等待更高效。
但是spinlock的應用場景有限，對于大的臨界區(qū)，忙等待則是件很恐怖的事情，特別是當同步機制運用于等待某一事件時（比如服務器工作線程等待客戶端發(fā)起請求）。所以很多情況下進程掛起等待是很有必要的。

內(nèi)核提供的同步機制，諸如semaphore、等，其實骨子里也是利用原子指令實現(xiàn)的spinlock，內(nèi)核在此基礎上實現(xiàn)了進程的睡眠與喚醒。
使用這樣的鎖，能很好的支持進程掛起等待。但是最大的缺點是每次lock與unlock都是一次系統(tǒng)調(diào)用，即使沒有鎖沖突，也必須要通過系統(tǒng)調(diào)用進入內(nèi)核之后才能識別。（關于系統(tǒng)調(diào)用開銷大的問題，可以參閱：《從"read"看系統(tǒng)調(diào)用的耗時》。）

理想的同步機制應該是在沒有鎖沖突的情況下在用戶態(tài)利用原子指令就解決問題，而需要掛起等待時再使用內(nèi)核提供的系統(tǒng)調(diào)用進行睡眠與喚醒。換句話說，用戶態(tài)的spinlock在trylock失敗時，能不能讓進程掛起，并且由持有鎖的線程在unlock時將其喚醒？
如果你沒有較深入地考慮過這個問題，很可能想當然的認為類似于這樣就行了：

void lock(int lockval) {while (!trylock(lockval)) {
        wait();  // 如：raise(SIGSTOP)}
}

但是如果這樣做的話，檢測鎖的trylock操作和掛起進程的wait操作之間會存在一個窗口，如果其間lock發(fā)生變化（比如鎖的持有者釋放了鎖），調(diào)用者將進入不必要的wait，甚至于wait之后再沒有人能將它喚醒。（詳見《linux線程同步淺析》的討論。）

在futex誕生之前，要實現(xiàn)我們理想中的鎖會非常別扭。比如可以考慮用sigsuspend系統(tǒng)調(diào)用來實現(xiàn)進程掛起：

class mutex {private:int lockval;
    spinlocked_set<pid_t> waiters;    // 使用spinlock做保護的setpublic:void lock() {pid_t mypid = getpid();
        waiters.insert(mypid);        // 先將自己加入mutex的等待隊列while (!trylock(lockval)) {   // 再嘗試加鎖// 進程初始化時需要將SIGUSER1 mask掉，并在此時開啟sigsuspend(MASK_WITHOUT_SIGUSER1);
        }
        waiters.remove(mypid)         // 上鎖成功之后將自己從等待隊列移除}void unlock() {
        lockval = 0;                  // 先釋放鎖pid_t waiter = waiters.first();  // 再檢查等待隊列if (waiter != 0) {            // 如果有人等待，發(fā)送SIGUSER1信號將其喚醒kill(waiter, SIGUSER1);
        }
    }
}

注意，這里的sigsuspend不同于簡單的raise(SIGSTOP)之類wait操作。如果unlock時用于喚醒的kill操作先于sigsuspend發(fā)生，sigsuspend也一樣能被喚醒。（詳見《linux線程同步淺析》的討論。）
這樣的實現(xiàn)有點類似于老版本的phread_cond，應該還是能work的。有些不太爽的地方，比如sigsuspend系統(tǒng)調(diào)用是全局的，并不單單考慮某一把鎖。也就是說，lockA的unlock可以將等待lockB的進程喚醒。盡管進程被喚醒之后會繼續(xù)trylock，并不影響正確性；盡管多數(shù)情況下lockA.unlock也并不會試圖去喚醒等待lockB的進程（除了一些競爭情況下），因為后者很可能并不在lockA的等待隊列中。
另一方面，用戶態(tài)實現(xiàn)的等待隊列也不太爽。它對進程的生命周期是無法感知的，很可能進程掛了，pid卻還留在隊列中（甚至于一段時間之后又有另一個不相干的進程重用了這個pid，以至于它可能會收到莫名其妙的信號）。所以，unlock的時候如果僅僅給隊列中的一個進程發(fā)信號，很可能喚醒不了任何等待者。保險的做法只能是全部喚醒，從而引發(fā)“驚群“現(xiàn)象。不過，如果僅僅用在多線程（同一進程內(nèi)部）倒也沒關系，畢竟多線程不存在某個線程掛掉的情況（如果線程掛掉，整個進程都會掛掉），而對于線程響應信號而主動退出的情況也是可以在主動退出前注意處理一下等待隊列清理的問題。

futex來了

現(xiàn)在看來，要實現(xiàn)我們想要的鎖，對內(nèi)核就有兩點需求：1、支持一種鎖粒度的睡眠與喚醒操作；2、管理進程掛起時的等待隊列。
于是futex就誕生了。futex主要有futex_wait和futex_wake兩個操作：

// 在uaddr指向的這個鎖變量上掛起等待（僅當*uaddr==val時）int futex_wait(int *uaddr, int val);// 喚醒n個在uaddr指向的鎖變量上掛起等待的進程int futex_wake(int *uaddr, int n);

內(nèi)核會動態(tài)維護一個跟uaddr指向的鎖變量相關的等待隊列。
注意futex_wait的第二個參數(shù)，由于用戶態(tài)trylock與調(diào)用futex_wait之間存在一個窗口，其間lockval可能發(fā)生變化（比如正好有人unlock了）。所以用戶態(tài)應該將自己看到的*uaddr的值作為第二個參數(shù)傳遞進去，futex_wait真正將進程掛起之前一定得檢查lockval是否發(fā)生了變化，并且檢查過程跟進程掛起的過程得放在同一個臨界區(qū)中。（參見《linux線程同步淺析》的討論。）如果futex_wait發(fā)現(xiàn)lockval發(fā)生了變化，則會立即返回，由用戶態(tài)繼續(xù)trylock。

futex實現(xiàn)了鎖粒度的等待隊列，而這個鎖卻并不需要事先向內(nèi)核申明。任何時候，用戶態(tài)調(diào)用futex_wait傳入一個uaddr，內(nèi)核就會維護起與之配對的等待隊列。
這件事情聽上去好像很復雜，實際上卻很簡單。其實它并不需要為每一個uaddr單獨維護一個隊列，futex只維護一個總的隊列就行了，所有掛起的進程都放在里面。當然，隊列中的節(jié)點需要能標識出相應進程在等待的是哪一個uaddr。這樣，當用戶態(tài)調(diào)用futex_wake時，只需要遍歷這個等待隊列，把帶有相同uaddr的節(jié)點所對應的進程喚醒就行了。
作為優(yōu)化，futex維護的這個等待隊列由若干個帶spinlock的鏈表構成。調(diào)用futex_wait掛起的進程，通過其uaddr hash到某一個具體的鏈表上去。這樣一方面能分散對等待隊列的競爭、另一方面減小單個隊列的長度，便于futex_wake時的查找。每個鏈表各自持有一把spinlock，將"*uaddr和val的比較操作"與"把進程加入隊列的操作"保護在一個臨界區(qū)中。

另一個問題是關于uaddr參數(shù)的比較。futex支持多進程，需要考慮同一個物理內(nèi)存單元在不同進程中的虛擬地址不同的問題。那么不同進程傳遞進來的uaddr如何判斷它們是否相等，就不是簡單數(shù)值比較的事情。相同的uaddr不一定代表同一個內(nèi)存，反之亦然。
兩個進程（線程）要想共享同存，無外乎兩種方式：通過文件映射（映射真實的文件或內(nèi)存文件、ipc shmem，以及有親緣關系的進程通過帶MAP_SHARED標記的匿名映射共享內(nèi)存）、通過匿名內(nèi)存映射（比如多線程），這也是進程使用內(nèi)存的唯二方式。
那么futex就應該支持這兩種方式下的uaddr比較。匿名映射下，需要比較uaddr所在的地址空間（mm）和uaddr的值本身；文件映射下，需要比較uaddr所在的文件inode和uaddr在該inode中的偏移。注意，上面提到的內(nèi)存共享方式中，有一種比較特殊：有親緣關系的進程通過帶MAP_SHARED標記的匿名映射共享內(nèi)存。這種情況下表面上看使用的是匿名映射，但是內(nèi)核在暗中卻會轉(zhuǎn)成到/dev/zero這個特殊文件的文件映射。若非如此，各個進程的地址空間不同，匿名映射下的uaddr永遠不可能被futex認為相等。

futex和它的兄弟姐妹們

futex_wait和futex_wake就是futex的基本。之后，為了對其他同步方式做各種優(yōu)化，futex又增加了若干變種。
futex等待系列的調(diào)用一般都可以傳遞timeout參數(shù)，支持超時喚醒。這一塊邏輯相對較獨立，本文中不再展開。

Bitset系列

int futex_wait_bitset(int *uaddr, int val, int bitset);int futex_wake_bitset(int *uaddr, int n, int bitset);

額外傳遞了一個bitset參數(shù)，使用特定bitset進行wait的進程，只能被使用它的bitset超集的wake調(diào)用所喚醒。
這個東西給讀寫鎖很好用，進程掛起的時候通過bitset標記自己是在等待讀還是等待寫。unlock時決定應該喚醒一個寫等待的進程、還是喚醒全部讀等待的進程。
沒有bitset這個功能的話，要么只能unlock的時候不區(qū)分讀等待和寫等待，全部喚醒；要么只能搞兩個uaddr，讀寫分別futex_wait其中一個，然后再用spinlock保護一下兩個uaddr的同步。
（參閱：http://locklessinc.com/articles/sleeping_rwlocks/）

Requeue系列

int futex_requeue(int *uaddr, int n, int *uaddr2, int n2);int futex_cmp_requeue(int *uaddr, int n, int *uaddr2, int n2, int val);

功能跟futex_wake有點相似，但不僅僅是喚醒n個等待uaddr的進程，而更進一步，將n2個等待uaddr的進程移到uaddr2的等待隊列中（相當于也futex_wake它們，然后強制讓它們futex_wait在uaddr2上面）。
在futex_requeue的基礎上，futex_cmp_requeue多了一個判斷，僅當*uaddr與val相等時才執(zhí)行操作，否則直接返回，讓用戶態(tài)去重試。
這個東西是為pthread_cond_broadcast準備的。還是先來回顧一下pthread_cond的邏輯（列一下，后面會多次用到）：

pthread_cond_wait(mutex, cond):value = cond->value; /* 1 */pthread_mutex_unlock(mutex); /* 2 */retry:pthread_mutex_lock(cond->mutex); /* 10 */if (value == cond->value) { /* 11 */me->next_cond = cond->waiter;cond->waiter = me;
        pthread_mutex_unlock(cond->mutex);
        unable_to_run(me);
        goto retry;
    } else
        pthread_mutex_unlock(cond->mutex); /* 12 */pthread_mutex_lock(mutex); /* 13 */pthread_cond_signal(cond):pthread_mutex_lock(cond->mutex); /* 3 */cond->value++; /* 4 */if (cond->waiter) { /* 5 */sleeper = cond->waiter; /* 6 */cond->waiter = sleeper->next_cond; /* 7 */able_to_run(sleeper); /* 8 */}
    pthread_mutex_unlock(cond->mutex); /* 9 */

pthread_cond_broadcast跟pthread_cond_signal類似，不過它會喚醒所有（而不是一個）等待者。注意，pthread_cond_wait在被喚醒之后，第一件事情就是lock(mutex)（第13步）。如果pthread_cond_broadcast一下子喚醒了N個等待者，它們醒來之后勢必會爭搶mutex，造成千軍萬馬過獨木橋的"驚群"現(xiàn)象。
作為一種優(yōu)化，pthread_cond_broadcast不應該用futex_wake去喚醒所有等待者，而應該用futex_requeue喚醒一個等待者，然后將其他進程都轉(zhuǎn)移到mutex的等待隊列上去（隨后再由mutex的unlock來逐個喚醒）。

為什么要有futex_cmp_requeue呢？因為futex_requeue其實是有問題的，它相當于直接把一批進程拖到uaddr2的等待隊列里面去了，而沒有在臨界區(qū)里面做狀態(tài)檢查（回想一下futex_wait里面檢查*uaddr==val的重要性）。那么，在進入futex_requeue和真正將進程移到uaddr2之間就存在一個窗口，這個間隙內(nèi)可能有其他線程futex_wake(uaddr2)，這將無法喚醒這些正要移動卻尚未移動的進程，可能造成這些進程今后再也無法被喚醒了。
不過盡管futex_requeue并不嚴謹，pthread_cond_broadcast這個case卻是OK的，因為在pthread_cond_broadcast喚醒等待者的時候，不可能有人futex_wake(uaddr2)，因為這個鎖正在被pthread_cond_broadcast持有，它將在喚醒操作結(jié)束后（第9步）才會釋放。這也就是為什么futex_requeue有問題，卻堂而皇之的被release了。

Wake & Operator

int futex_wake_op(int *uaddr1, int *uaddr2, int n1, int n2, int op);

這個系統(tǒng)調(diào)用有點像CISC的思路，一個調(diào)用中搞了很多動作。它嘗試在uaddr1的等待隊列中喚醒n1個進程，然后修改uaddr2的值，并且在uaddr2的值滿足條件的情況下，喚醒uaddr2隊列中的n2個進程。uaddr2的值如何修改？又需要滿足什么樣的條件才喚醒uaddr2？這些邏輯都pack在op參數(shù)中。
int類型的op參數(shù)，其實是一個struct：

struct op {// 修改*uaddr2的方法：SET (*uaddr2=OPARG)、ADD(*uaddr2+=OPARG)、// OR(*uaddr2|=OPARG)、ANDN(*uaddr2&=~OPARG)、XOR(*uaddr2^=OPARG)int OP : 4;// 判斷*uaddr2是否滿足條件的方法：EQ(==)、NE(!=)、LT(<)、LE(<=)、GT(>)、GE(>=)int CMP : 4;int OPARG : 12;// 修改*uaddr2的參數(shù)int CMPARG : 12;// 判斷*uaddr2是否滿足條件的參數(shù)}

futex_wake_op搞這么一套復雜的邏輯，無非是希望一次系統(tǒng)調(diào)用里面處理兩把鎖，相當于用戶態(tài)調(diào)用兩次futex_wake。
假設用戶態(tài)需要釋放uaddr1和uaddr2兩把鎖（值為0代表未上鎖、1代表上鎖、2代表上鎖且有進程掛起等待），不使用futex_wake_op的話需要這么寫：

int old1, old2;atomic { old1 = *uaddr1; *uaddr1 = 0; }if (old1 == 2) {
    futex_wake(uaddr1, n1);}
atomic { old2 = *uaddr2; *uaddr2 = 0; }if (old2 == 2) {
    futex_wake(uaddr2, n2);}

而使用futex_wake_op的話，只需要這樣：

int old1;
atomic { old1 = *uaddr1; *uaddr1 = 0; }if (old1 == 2) {
    futex_wake_op(uaddr1, n1, uaddr2, n2, {
        // op參數(shù)的意思：設置*uaddr2=0，并且如果old2==2，則執(zhí)行喚醒
        OP=SET, OPARG=0, CMP=EQ, CMPARG=2
    } );
}else {
    ... // 單獨處理uaddr2
}

搞這么復雜，其實并不僅僅是省一次系統(tǒng)調(diào)用的問題。因為有可能在unlock(uaddr1)之后，被喚醒的進程立馬會去lock(uaddr2)。而這時如果這邊還沒來得及unlock(uaddr2)的話，被喚醒的進程立刻又將被掛起，然后隨著這邊unlock(uaddr2)又會再度被喚醒。這不折騰么？
這個場景就可能發(fā)生在pthread_cond_wait和pthread_cond_signal之間。當pthread_cond_signal在喚醒等待者之后，會釋放內(nèi)部的鎖（第9步）。而pthread_cond_wait在被喚醒之后立馬又會嘗試獲取內(nèi)部的鎖，以重新檢查狀態(tài)（第10步）。若不是futex_wake_op將喚醒和釋放鎖兩個動作一筆帶過，這中間必定會有強烈的競爭。
當然，使用前面提到的futex_cmp_requeue也能避免過分競爭，pthread_cond_signal不要直接喚醒等待者，而是將其requeue到內(nèi)部鎖的等待隊列，等這邊釋放鎖之后才真正將其喚醒。不過既然pthread_cond_signal立馬就會釋放內(nèi)部鎖，先requeue再wake多少還是啰嗦了些。

Priority Inheritance系列

int futex_lock_pi(int *uaddr);int futex_trylock_pi(int *uaddr);int futex_unlock_pi(int *uaddr);int futex_wait_requeue_pi(int *uaddr1, int val1, int *uaddr2);int futex_cmp_requeue_pi(int *uaddr, int n1, int *uaddr2, int n2, int val);

Priority Inheritance，優(yōu)先級繼承，是解決優(yōu)先級反轉(zhuǎn)的一種辦法。
futex_lock_pi/futex_trylock_pi/futex_unlock_pi，是帶優(yōu)先級繼承的futex鎖操作。
futex_cmp_requeue_pi是帶優(yōu)先級繼承版本的futex_cmp_requeue，futex_wait_requeue_pi是與之配套使用的，用于替代普通的futex_wait。

看完上述內(nèi)容，你們掌握如何進行l(wèi)inux futex淺析的方法了嗎？如果還想學到更多技能或想了解更多相關內(nèi)容，歡迎關注億速云行業(yè)資訊頻道，感謝各位的閱讀！