本地自旋鎖與信號量/多服務檯自旋隊列-spin wait風格的信號量

週日傍晚，我去家附近的超市(...)買蘇打水，準備自制青檸蘇打，我感受我作的比買的那個巴黎水要更爽口。因爲天氣太熱，不少人都去超市避暑去了，超市 也不攆人，這彷彿是他們的策略，人過來避暑了，走的時候不免要買些東西的，就跟不少美女在公交地鐵上看淘寶消磨時光，而後就下單了...這是多麼容易一件 事，反之開車的美女網購就少不少。對於超市的避暑者，要比公交車上下單更麻煩些，由於有一個成本問題，這就是排隊成本。
       其實這是一個典型的多服務檯排隊問題，可是超市處理的並很差，存在隊頭擁塞問題，我就好幾回遇到過。好幾回，我排的那個隊，前面結帳出現了糾紛，咱們後面 的就必須等待，眼睜睜看着旁邊的結帳隊伍向前推動，可是這種排隊方案足夠簡單，把調度任務交給了排隊者本人，結帳的人想排到哪一個隊列就排到哪一個隊列，判斷 一個隊列是否會擁塞也有不少辦法，好比看購物的多少，是否有衣物(鎖卡拔出糾紛)，是否有稱重的東西(會忘記稱重)，是否有打折物，是否有老年人，收銀員 的手法是否嫺熟等，全靠本身的判斷，無異於一場***。  我改造的Open×××多線程實現就是這種。
       銀行服務以及飯店的排隊服務就要好不少，顧客排隊時，自取一個號碼，排入單一的隊列，由空閒服務檯叫號，這就是一個調度系統。這種單隊列多服務檯是不會出 現隊頭擁塞的，等候的顧客持ticket排隊，自己沒必要排在隊伍裏，而ticket號邏輯上組成一個虛擬的隊列，沒叫到號的能夠暫時乾點別的，自身沒必要排 隊。

       暫時幹別的？並不意味着你能夠離開，特別是業務處理流程很快的狀況下。你離開大廳，剛走出去，準備去旁邊的小店逛逛，結果聽到叫到你的號了，趕忙返回，其 實還不如不出去呢。可是對於等待比較久的叫號系統，那卻是能夠暫時出去。出去再返回的過程意味着體力開銷，可是若是出去的時間久，能夠完成另外一件重要的 事，意味着爲這另外這件事的收益付出的體力開銷是值得的。

       知道我想到什麼了嗎？我想到了信號量。信號量就是一個單隊列多服務檯排隊系統，信號量的初始值就是服務檯的數量。一個執行流被服務意味着少了一個可服務的 服務檯，這就是down操做，而up操做則是一個服務檯從新變成空閒的信號，這意味着有一個新的排隊者能夠獲得服務了，我能夠把」服務「理解成進入臨界 區。

       我在想一個問題，爲何信號量必定要設計成sleep-wait的模式，爲何就沒有spin-wait的模式啊。而我目前面臨的問題，若是使用 sleep-wait，切換開銷太大，perf顯示的頭幾名大頭都在schedule，wake up，之類的，也就是說，你切換出去了，沒多久就又把你叫回來了，好在Linux調度系統基於CFS徹底公平機制，抖動不會太厲害，不過這麼切換一次形成 的開銷也不算小，起碼等到再次切換回來的時候，cache變涼了。

回顧Linux版的ticket自旋鎖，我以爲全部的排隊者以及持鎖者 touch同一個變量，該變量會cache到全部的當事者cpu的cache中，被持鎖者以及爭鎖者read/write時，會涉及到多個處理器之間的 cache一致性問題，這也是一筆很大的底層開銷。因而我設計了一個本地接力自旋鎖改變了這個局面，保持每個爭鎖者都只touch一個別的爭鎖者不會 touch的變量，且cache line要着色以保證不會cache到同一line，此外，持鎖者在釋放鎖的時候，只會write下一個爭鎖者的本地變量。這樣就確保了cache一致性 被最少的觸發。
       本着這個新的自旋鎖設計，結合我在超市的經歷，我想把我這個自旋鎖發展成一個能夠有多個CPU持有鎖的自旋隊列。後來我忽然發現，這不就是信號量嘛... 惋惜信號量並無如期被我所用，由於Linux實現的信號量是sleep-wait機制的，我須要的是spin-wait，由於我知道一個數據包的發送是 很快的，之因此引入隊列，構建VOQ，是由於我想避開N加速比問題，然而個人算法是軟實現，根本不存在N加速比問題，因此後來我想取消VOQ，又怕引起隊 頭擁塞，因此採用了多服務檯單隊列機制，爲了實現這個，我本能夠採用信號量的，可是又不想sleep，因此採用極其複雜的多個spin lock的機制，超市排隊引起的遐想致使我想到用spin-wait來實現信號量，事實上，簡單測試以後，發現效果還真不錯。

先看一下Linux原生的信號量實現，代碼比較簡單。順便說一句，這篇文章並不意味着我又開始源代碼分析了，而是也許它意味着某種終結，先後的呼應。

/*
 * 爲了突出重點問題,不至於迷失在代碼細節.我作了如下的假設:
 * 1.我省去了操做信號量自己的自旋鎖,我假設P/V操做過程的任意序列都是原子的.
 * 2.我取消了超時參數以及state,我假設除非獲得信號量,不然必定等下去,我還假設睡眠不會被打斷,除非有人喚醒.
 * 3.我取消了inline，由於我想突出圍繞本地棧變量本地自旋，這樣不會cache pingpong.
 */
struct semaphore {
    raw_spinlock_t        lock;
    unsigned int        count;
    struct list_head    wait_list;
};

struct semaphore_waiter {
    struct list_head list;
    struct task_struct *task;
    // 本地局部檢測變量
    bool up;
};


static int down(struct semaphore *sem)
{
    if (likely(sem->count > 0)) {
        sem->count--;
    }
    else {
        struct task_struct *task = current;
        struct semaphore_waiter waiter;
        // 棧上的排隊體,至關於ticket,得到信號量(函數返回)後就沒有用了
        list_add_tail(&waiter.list, &sem->wait_list);
        waiter.task = task;
        waiter.up = false;

        for (;;) {
            __set_task_state(task, TASK_UNINTERRUPTIBLE);
            schedule();

            // 本地棧變量的檢測，減小了多處理器之間的cache同步，不會cache乒乓
            // ********************************************************************
            // 可是要想到一種狀況，若是多個進程試圖寫這個變量，仍是要有鎖操做的。
            // 雖然個人假設是全部操做以及操做序列都是原子的，可是在up操做中，持有信
            // 號量的進程只是簡單的wake up了隊列，而這並不能確保被喚醒的task就必定可
            // 以獲得執行，中間還有一個schedule層呢。鑑於這種複雜的局面，我想到了不
            // sleep，而是本地自旋版本的信號量，無論怎樣，它確實解決了個人問題。
            // [事實上，因爲sem自己擁有一把自旋鎖，這就禁止了多個「服務檯」同時召喚
            //  同一個等待者的局面，而我在個人描述中，忽略了這把自旋鎖，這是爲何呢？
            //  由於，我想爲個人自旋信號量版本貼金，否則人家都把問題解決了，我還扯啥
            //  玩意兒啊！]
            // ********************************************************************
            // 這種狀況在spin lock下不會存在，由於同時只有一個進程會持有lock，
            // 不可能多個進程同時操做。

            if (waiter.up) {
                return 0;
            }
        }
    }
}

void up(struct semaphore *sem)
{
    unsigned long flags;
    if (likely(list_empty(&sem->wait_list))) {
        sem->count++;
    } else {
        struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
                            struct semaphore_waiter, list);
        // 標準的Linux kernel中，該操做被spin lock保護，這意味着不可能多個服務檯同時將
        // 服務給與同一個等待者。
        list_del(&waiter->list);
        waiter->up = true;
        // 簡單wake up進程，它什麼時候投入運行，看調度器什麼時候調度它了。
        wake_up_process(waiter->task);
    }
}

因爲我忽略了信號量自己的保護自旋鎖，當你詳細分析上述實現的時候，會發現不少競爭條件，好比同時多個服務檯召喚一個等待者，可是 不要緊，該說的我都寫到冗長的註釋裏面了。我之因此忽略信號量的自旋鎖，是由於我想把信號量該形成一個通用的自旋等待隊列，自旋鎖只是其中一個特殊狀況， 該狀況對應只有一個服務檯的情形。
       若是看懂了原生的實現，那麼改造後的實現應該是如下的樣子：

/*
 * 我引入了BEGIN_ATOMIC和END_ATOMIC兩個宏，由於我不想貼彙編碼，因此這兩個宏的意思就是它們之間的代碼都是由
 * lock前綴修飾的，鎖總線。
 * 此外，什麼事情都沒有作，只是改了名稱。若是想初始化一個標準的排隊自旋鎖，將初始化宏的val設置成1便可。
 */
struct spin_semaphore {
    unsigned int        count;
    struct list_head    wait_list;
};

struct spin_semaphore_waiter {
    struct list_head list;
    struct task_struct *task;
    // 本地局部檢測變量
    bool up;
};


static int spin_down(struct spin_semaphore *sem)
{
    if (likely(sem->count > 0)) {
        sem->count--;
    }
    else {
        struct task_struct *task = current;
        struct spin_semaphore_waiter waiter;
BEGIN_ATOMIC
        list_add_tail(&waiter.list, &sem->wait_list);
        waiter.task = task;
        waiter.up = false;
END_ATOMIC

        for (;;) {
            cpu_relax();  // PAUSE
            if (waiter.up) {
                return 0;
            }
        }
    }
}

void up(struct spin_semaphore *sem)
{
    unsigned long flags;
BEGIN_ATOMIC
    if (likely(list_empty(&sem->wait_list))) {
        sem->count++;
END_ATOMIC
    }
    else {
        struct spin_semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
                            struct spin_semaphore_waiter, list);
        list_del(&waiter->list);
        waiter->up = true;
END_ATOMIC
    }
}

全部名稱加上了spin_前綴修飾。不錯，這個應該是和Windows NT內核的排隊自旋鎖的實現很接近了。在此不談優化，然而實際使用時，應該是先用匯編編碼，而後彙編碼優化它了。