本地自旋鎖與信號量/多服務臺自旋隊列-spin wait風格的信號量
其實這是一個典型的多服務臺排隊問題，可是超市處理的並不好。存在隊頭擁塞問題。我就好幾次遇到過。好幾次，我排的那個隊。前面結賬出現了糾紛。我們後面的就必須等待，眼睜睜看著旁邊的結賬隊伍向前推進。可是這樣的排隊方案足夠簡單。把調度任務交給了排隊者本人，結賬的人想排到哪個隊列就排到哪個隊列，推斷一個隊列是否會擁塞也有非常多辦法，比方看購物的多少。是否有衣物(鎖卡拔出糾紛)，是否有稱重的東西(會忘記稱重)。是否有打折物。是否有老年人。收銀員的手法是否嫻熟等。全靠自己的推斷。無異於一場賭博。 我改造的OpenVPN多線程實現就是這樣的。
銀行服務以及飯店的排隊服務就要好非常多，顧客排隊時。自取一個號碼，排入單一的隊列，由空暇服務臺叫號。這就是一個調度系統。這樣的單隊列多服務臺是不會出現隊頭擁塞的，等候的顧客持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內核的排隊自旋鎖的實現非常接近了。在此不談優化。然而實際使用時，應該是先用匯編編碼。然後匯編碼優化它了。
本地自旋鎖與信號量/多服務臺自旋隊列-spin wait風格的信號量