使用CAS實現無鎖的SkipList
感謝同事【付哲】釋出此文。
無鎖
併發環境下最常用的同步手段是互斥鎖和讀寫鎖，例如pthread_mutex和pthread_readwrite_lock，常用的正規化為：
void ConcurrencyOperation() {
mutex.lock();
// do something
mutex.unlock();
}
這種方法的優點是：
- 程式設計模型簡單，如果小心控制上鎖順序，一般來說不會有死鎖的問題；
- 可以通過調節鎖的粒度來調節效能。
缺點是：
- 所有基於鎖的演算法都有死鎖的可能；
- 上鎖和解鎖時程序要從使用者態切換到核心態，並可能伴隨有執行緒的排程、上下文切換等，開銷比較重；
- 對共享資料的讀與寫之間會有互斥。
無鎖程式設計（嚴格來講是非阻塞程式設計）可以分為lock free和wait-free兩種，下面是對它們的簡單描述：
- lock free：鎖無關，一個鎖無關的程式能夠確保它所有執行緒中至少有一個能夠繼續往下執行。這意味著有些執行緒可能會被任意的延遲，然而在每一個步驟中至少有一個執行緒能夠執行下去。因此這個系統作為一個整體總是在前進的，儘管有些執行緒的進度可能沒有其它執行緒走的快。
- wait free：等待無關，一個等待無關的程式可以在有限步之內結束，而不管其它執行緒的相對執行速度如何。
- lock based：基於鎖，基於鎖的程式無法提供上面的任何保證，任一執行緒持有了某互斥體並處於等待狀態，那麼其它想要獲取同意互斥體的執行緒只有等待，所有基於鎖的演算法無法擺脫死鎖的陰影。
本文提到的無鎖單指lock free。
lock free與CAS
常見的lock free程式設計一般是基於CAS(Compare And Swap)操作：
CAS(void *ptr, Any oldValue, Any newValue);
即檢視記憶體地址ptr處的值，如果為oldValue則將其改為newValue，並返回true，否則返回false。X86平臺上的CAS操作一般是通過CPU的CMPXCHG指令來完成的。CPU在執行此指令時會首先鎖住CPU匯流排，禁止其它核心對記憶體的訪問，然後再檢視或修改*ptr的值。簡單的說CAS利用了CPU的硬體鎖來實現對共享資源的序列使用。它的優點是：
- 開銷較小：不需要進入核心，不需要切換執行緒；
- 沒有死鎖：匯流排鎖最長持續為一次read+write的時間；
- 只有寫操作需要使用CAS，讀操作與序列程式碼完全相同，可實現讀寫不互斥。
缺點是：
- 程式設計非常複雜，兩行程式碼之間可能發生任何事，很多常識性的假設都不成立。
- CAS模型覆蓋的情況非常少，無法用CAS實現原子的複數操作。
而在效能層面上，CAS與mutex/readwrite lock各有千秋，簡述如下：
- 單執行緒下CAS的開銷大約為10次加法操作，mutex的上鎖+解鎖大約為20次加法操作，而readwrite lock的開銷則更大一些。
- CAS的效能為固定值，而mutex則可以通過改變臨界區的大小來調節效能；
- 如果臨界區中真正的修改操作只佔一小部分，那麼用CAS可以獲得更大的併發度。
- 多核CPU中執行緒排程成本較高，此時更適合用CAS。
使用CAS實現無鎖單向連結串列
單向連結串列實現的核心就是insert函式，這裡我們用兩個版本的insert函式來進行簡單的演示，使用的CAS操作為GCC提供的__sync_compare_and_swap函式。
首先是無序的insert操作，即將新結點插入到指定結點的後面。
void insert(Node *prev, Node *node) {
while (true) {
node->next = prev->next;
if (__sync_compare_and_swap(&prev->next, node->next, node)) {
return;
}
}
}
程式碼分析：
- 首先修改node->next，此時node還沒有完成插入，只能被本執行緒看到，因此這個修改可以直接進行。
- 在if中嘗試修改prev->next，如果失敗，則表明prev->next剛剛被其它執行緒修改了，則重複這一過程。
然後是有序的insert操作，即保證prev<= node <= next。
void insert(Node *prev, Node *node) {
while (true) {
Node *next = prev->next;
while (next != NULL && next->item < node->item) {
prev = next;
next = prev->next;
}
node->next = next;
if (__sync_compare_and_swap(&prev->next, next, node)) {
return;
}
}
}
這段程式碼相比上一版本多了一個next變數。如果去掉next變數，那麼程式碼就是下面的樣子。
void insert(Node *prev, Node *node) {
while (true) {
while (prev->next != NULL && prev->next->item < node->item) {
prev = prev->next;
}
node->next = prev->next;
if (__sync_compare_and_swap(&prev->next, node->next, node)) {
return;
}
}
}
上面的程式碼有著很嚴重的安全隱患：prev是共享資源，因此每個prev->next的值不一定是相等的！解決辦法就是用一個區域性變數來儲存某個時刻prev的值，從而保證我們在不同地方進行比較的結點是一致的。
Key-Value資料結構
目前常用的key-value資料結構有三種：Hash表、紅黑樹、SkipList，它們各自有著不同的優缺點（不考慮刪除操作）：
- Hash表：插入、查詢最快，為O(1)；如使用連結串列實現則可實現無鎖；資料有序化需要顯式的排序操作。
- 紅黑樹：插入、查詢為O(logn)，但常數項較小；無鎖實現的複雜性很高，一般需要加鎖；資料天然有序。
- SkipList：插入、查詢為O(logn)，但常數項比紅黑樹要大；底層結構為連結串列，可無鎖實現；資料天然有序。
如果要實現一個key-value結構，需求的功能有插入、查詢、迭代、修改，那麼首先Hash表就不是很適合了，因為迭代的時間複雜度比較高；而紅黑樹的插入很可能會涉及多個結點的旋轉、變色操作，因此需要在外層加鎖，這無形中降低了它可能的併發度。而SkipList底層是用連結串列實現的，可以實現為lock free，同時它還有著不錯的效能（單執行緒下只比紅黑樹略慢），非常適合用來實現我們需求的那種key-value結構。LevelDB、Reddis的底層儲存結構就是用的SkipList。
SkipList
那麼，SkipList是什麼呢？它由多層有序連結串列組成，每層連結串列的結點數量都是上一層的X倍，而它的插入和查詢操作都從頂層開始進行。
（圖片取自wiki）
從上圖可以很容易看出查詢的方式：
- 從頂層的頭結點出發；
- 若下一結點為目標值，則返回結果；
- 若下一結點小於目標值，則前進；
- 若下一結點大於目標值或為NULL，則：
- 若當前處於最底層，則返回NULL；
- 下降一層，重複2-4步。
在SkipList中，結點層數非常關鍵，如果各個結點的層數均勻分佈，那麼插入與查詢的效率就會比較高。為了實現這一目的，SkipList中每個結點的層數是在插入前隨機算出來的，其基本原理就是令結點在i層的概率是i+1層的X倍，程式碼如下：
int RandLevel(int X, int maxLevel) {
int r = rand();
int level = 1;
for (int j = X; r < RAND_MAX / j && level < maxLevel; ++level, j *= X)
continue;
return level;
}
插入新結點的過程與查詢很類似，這裡我們假設連結串列中的各結點不允許重複：
- 計算出新結點的層數lv；
- 從lv層的頭結點出發，開始查詢過程；
- 如果找到目標值，返回NULL；
- 如果當前處於最底層，則建立新結點，並依次將新結點插入到1-lv層；
可以看出，插入操作的1-3步是單純的讀操作，只有第4步才是對共享資源的寫操作。而第4步的插入實質上就是有序連結串列的插入操作，我們在前面已經簡述瞭如何用CAS實現它。因此，只要保證插入順序是從底層向上依次插入，那麼就可以將SkipList實現為lock free。插入順序從底向上進行的原因如下。
N個插入操作肯定需要至少N次CAS，而任意一個CAS成功後就意味著新結點已經成為了SkipList的一部分，變成了共享資源，則新結點就需要遵循其它結點的原則：每個結點都同時存在於1-lv層。容易看出，只有從底層向上插入才能滿足這一條件。
多個CAS操作本身沒有原子性，即在N次插入沒有完成前，新結點會表現出一定的不一致性，具體來說就是多個執行緒先後訪問新結點時，看到的它的層數並不相同。這種不一致性會比較輕微的影響SkipList的效能，而不會影響它的正確性。
SkipList的插入程式碼如下：
void Insert(Node *node) {
node->level = RandLevel(2, MAX_LEVEL);
InsertInternal(head, node->level, node);
}
Node *InsertInternal(Node *prev, int lv, Node *node) {
Node *next = prev->next[lv];
while (next != NULL && next->item < node->item) {
prev = next;
next = prev->next[lv];
}
if (next == NULL || next->item > node->item) {
if (lv != 0) {
if (InsertInternal(prev, lv - 1, node) != NULL) {
ListInsert(prev, node, lv);
}
}
} else if (next->item == node->item) {
return NULL;
}
return node;
}
其中ListInsert就是對前面有序連結串列插入的一個簡單改寫。整個插入過程遞迴實現，從而滿足了插入順序要從底向上的要求。
更多思考
在設計無鎖SkipList時，不光需要我們將顯式的鎖用CAS替換掉，還需要儘量避免一些隱式的鎖，以及一些非執行緒安全的函式。
- RandLevel中的rand()是非執行緒安全的函式，需要替換為執行緒安全的版本（如非標準庫的rand_r()），或是由各執行緒自己來儲存rand使用的seed。
- 在建立SkipList的時候需要指定一個MAX_LEVEL，即頭結點的層數，這個值在此SkipList生命期中固定不變。一般來說12-20層都是可以接受的。
- 全域性new內部會加鎖，如果這裡有瓶頸的話需要換用自定義的記憶體池。
- 如果使用了記憶體池，那麼必須確保記憶體池本身是無鎖且支援併發寫的。否則就只能將SkipList改寫為單寫多讀版本。
- 在計算新結點的層數時，需要傳入一個maxLevel，這裡有兩種常見做法：可以傳入SkipList的最大層數MAX_LEVEL，也可以傳入當前最大層數topLevel + 1。兩種做法的優缺點為：
- 傳入MAX_LEVEL可能在SkipList中結點數量較少時就達到很高的層數，降低了此時插入與查詢的效能；但如果有序插入多個新結點，能保證各結點的層數均勻分佈。
- 傳入topLevel + 1可以保證在結點數較少時不太可能出現很高的層數，但在有序插入多個新結點時，可能導致前面插入結點的層數整體要低於後面插入的結點。
- SkipList的修改操作也需要是lock free的，因此需要將Node中的item改為指標，在修改某結點值的時候用CAS來替換掉舊指標，並在完成後刪除。
- SkipList也可以在最底層加入反向指標prev，這樣就能直接O(1)的反向迭代。帶來的問題是更大的不一致性——在插入未完成時兩個執行緒分別正向和反向迭代，看到的SkipList是不一致的。但可以保證SkipList在插入完成後的最終狀態是一致的。
本文只是對無鎖SkipList設計的一個簡單回顧，不包括詳細的實現程式碼。因為還不確定自己設計的有沒有紕漏，還需要認真學習一下LevelDB和Reddis中的SkipList程式碼。
參考：
