在編寫並發同步程序的時候，如果臨界區非常小，比如說只有幾條或幾十條指令，那么我們可以選擇自旋鎖（spinlock）。使用普通的互斥鎖會涉及到操作系統的調度，因此小臨界區一般首選自旋鎖。自旋鎖的工作方式就是讓競爭的線程不斷地讀取一個變量的狀態，判斷是否滿足可以進入臨界區的條件。
最簡單的自旋鎖應該如何實現？假設我們用一個布爾變量表示臨界區是否被占用，true表示被占用，false表示沒有被占用，那么我們可以考慮這樣的（偽）代碼：
1 void lock(bool *lck) 2 { 3 while (*lck == true); 4 *lck = true; 5 } 6 7 void unlock(bool *lck) 8 { 9 *lck = false; 10 }
這段代碼顯然是有問題的。lock函數分為兩步執行，第一步是讀lck變量的值，第二步是在發現lck變為false之后將其設置為true表示自己進入臨界區。問題就在於這兩步之間可能會交錯執行其他線程的lock，導致多個線程同時進入臨界區，因此這個算法是不正確的。
好在現代的硬件都提供了各種強大的read-modify-write（RMW）原子操作，可以原子地執行讀改寫操作，所謂原子操作就是可以在執行這幾個操作的時候不會被其他指令打斷，可以實現排他的內存訪問，實際是在內存總線層次的臨界區操作。現代硬件一般都支持強大的原子操作，例如Intel平台支持在指令前面加上lock前綴鎖定總線，實現原子的比較-交換操作（CAS，通過cmpxchg指令，原子地執行判斷條件並交換值的操作），原子遞增和原子遞減（inc和dec指令）操作。還支持一個交換操作，xchg指令，不需要lock前綴，可以原子地交換兩個寄存器或一個寄存器和一個內存位置的值。在Intel平台上通過這些基本的原子操作支持幾乎所有的原子操作。
在上面這個自旋鎖的例子中，我們可以使用一種稱為測試-設置（TAS）的原子操作。TAS操作是實際上是一種xchg操作：原子地寫入true，並返回之前的值。使用TAS實現自旋鎖的思路就是：先嘗試將true寫進共享的鎖變量lck，看返回的值是什么。如果返回false，說明之前lck的值為false，原子操作可以保證只有我一個線程成功地將lck從false變為true了，其他線程如果和我並發爭搶這個lck，那么他們得到的都是我改過的值，即true，所以讓他們繼續自旋等待去吧，等我unlock的時候把lck寫入false之后，那些失敗的線程中必然有一個能成功地寫入true。
下面是使用TAS操作的自旋鎖算法的lock()函數的代碼：
1 void lock(bool *lck) 2 { 3 while (TAS(lck) == true); 4 }
那么這個自旋算法是否正確了呢？答案是不一定，因為現代的硬件不一定滿足順序一致性，為了提高單核性能，處理器可能會將寫操作歸並，導致讀寫亂序，盡管保證單個線程自己看自己的執行結果符合程序順序，但是在多核多線程環境中，線程看到其他線程的執行順序不一定符合線程的程序順序。如果符合的話，這種正確性條件稱為順序一致性，但是現代的硬件一般都會采用有所放松（relax）的條件。好在這種硬件平台都提供了內存屏障（fence或barrier）指令來強制該順序一致性的地方實現順序一致性。因此為了編寫跨平台且可靠的代碼，我們還要加入屏障指令。為了方便，我們用一些跨平台的原子庫就好了，比如說libatomic_ops。完整代碼如下所示：
1 #include <atomic_ops.h> 2 3 #define SPIN_BODY __asm__ __volatile__("rep;nop" : : : "memory") 4 5 typedef struct { 6 AO_TS_t lock; 7 } spinlock_t; 8 9 AO_INLINE void spinlock_init(spinlock_t *lock) 10 { 11 lock->lock = AO_TS_INITIALIZER; 12 } 13 14 AO_INLINE void spinlock_lock(spinlock_t *lock) 15 { 16 while (AO_test_and_set_acquire(&lock->lock) == AO_TS_SET) { 17 SPIN_BODY; 18 } 19 } 20 21 AO_INLINE void spinlock_unlock(spinlock_t *lock) 22 { 23 AO_CLEAR(&lock->lock); 24 }
其中AO_test_and_set_acquire可以保證這個調用之后的所有操作都一定會出現在這個原子操作之后，即進入臨界區的操作不會被亂序到這個原子操作之前。
AO_CLEAR里面也包含了一個屏障，可以保證這條語句之前的操作（即臨界區操作）不會被亂序到這條原子操作之后。
用庫的好處就是可以方便地編寫跨平台的並發代碼。庫提供的是例如CAS、SWAP、INC、DEC之類的高層語義以及各種屏障的組合，並且將這些接口映射到底層具體的硬件平台。庫一般還會提供功能探測的宏用於判斷指定的原子操作能否在指定的硬件平台上通過硬件指令高效地實現，如果不能實現，庫則會提供相對低效的fallback實現。
順便提一下關於原子庫的選擇。
對比一些原子庫，我覺得Erlang運行時中自用的原子庫是最強大最完整的，提供了所有常用原子操作和所有屏障的完整組合，盡管有很多組合沒有實際意義，不過也不費電啊。libatomic_ops同樣是一個系出名門的庫，但是竟然沒有提供SWAP的操作。Erlang運行時中的原子庫也可以選擇libatomic_ops作為后端，但是由於libatomic_ops沒有swap操作，所以ERTS中的ethr_atomic[32]_xchg_<barrier>系列接口都是用CAS模擬的。此外美國阿貢國家實驗室有一個openpa，略顯簡陋。gcc-4.1之后的版本也提供了atomic擴展，不過從這頁文檔看是針對Intel平台設計的。所以，綜上，如果不需要SWAP，那就用libatomic_ops，如果追求全面完整，可以考慮把ERTS中的原子庫剝離出來用（甚至可以把ERTS中整個lib都弄出來，這個庫里面好東西不少，應該可以滿足大部分需求，而且免費跨平台哦）。
好了言歸正傳，還沒有開始介紹本文的主角MCS自旋鎖。
MCS自旋鎖是用發明人的名字命名的，也就是John Mellor-Crummey和Michael Scott，所以MCS三個字母不涉及這個鎖本身具體的意義。我們先說一下MCS鎖的工作原理，然后再說為什么一個自旋鎖要弄得這么復雜，前面那個不挺好的么，又簡潔又高效。關於工作原理，簡單地說，MCS鎖是一個鏈表形式的鎖，每一個線程是鏈表中的一個節點，MCS鎖用一個tail指針維護鏈表中的最后一個節點，每一個節點用一個布爾值locked表示自己是否被鎖定，以及一個next指針表示在鏈表中的下一個節點。MCS鎖的工作原理是這樣的：
- lock：用SWAP操作將tail更新為指向自己；如果tail之前的值為NULL，則表示沒有線程在等待，我可以直接獲得鎖進入臨界區。如果tail的值不為NULL，說明前面有線程在等待，那么我也插入鏈表尾端，將自己的locked設置為true，然后自旋自己的lock等待別人把我的locked的值設置為false退出循環結束等待進入臨界區；
- unlock：判斷自己的next是否為NULL，如果為NULL，則說明在我后面沒有線程在等待了，那么我就是鏈表中最后一個線程（至少目前是，而且tail還指向我），然后用CAS操作賭一把，將tail設置由指向我自己設置為NULL並返回，如果CAS成功了，那么表示成功釋放鎖直接返回；如果CAS不成功，剛才在CAS之前有一個比我快的線程先我一步插到我后面，改了tail，那我只能等一會。等什么呢？插到我后面的線程肯定要把我的next設置為他，那我就等next變為不是NULL就好了。如果我next不為NULL了，說明我后面插入了在自旋等待進入臨界區的線程了，根據前面lock的操作流程，他們肯定在自旋等待自己的locked。那么既然我要退出臨界區了，那我就把臨界區進入權轉交給插在我后面的那個線程，所以把他的locked字段設置為false，我出來，他進去。
有圖有真相，下面的圖示展示了MCS鎖的工作原理：
有3個線程，A、B和C。A的locked為false，A在臨界區，B和C先后試圖lock，由於A在臨界區，所以B和C都依次插在后面，並且locked都為true，都在等待。tail指向最后一個插入的節點C。下圖是線程A釋放鎖：
線程A將下一個節點，即線程B的locked設置為false，線程B可以退出等待，進入臨界區。線程A將自己從鏈表中脫離開。
下面展示的是MCS鎖的代碼。數據結構定義如下：
1 typedef struct _mcs_qnode { 2 int locked; 3 struct _mcs_qnode *next; 4 } mcs_qnode_t; 5 6 typedef union { 7 mcs_qnode_t node; 8 char pad__[CACHE_LINE_ALIGN_SIZE(sizeof(mcs_qnode_t))]; 9 } aligned_mcs_qnode_t; 10 11 typedef struct { 12 AO_t tail; /* 指向aligned_mcs_qnode_t的指針 */ 13 pthread_key_t qnode_key; 14 } mcslock_t;
mcs_qnode_t表示的是線程的節點，由於locked字段是要自旋的，為了避免偽共享的問題，aligned_mcs_qnode_t結構體將mcs_qnode_t對齊到cache線。
下面是初始化、鎖和解鎖的實現。這段代碼用的是libatomic_ops庫，但是由於這個庫沒有SWAP操作，而我是在Intel平台上做這個實驗，所以我借用了ERTS中的xchg代碼，就是xchg_mb()的實現，后面_mb尾巴的意思是在Intel平台上，xchg本身具有完整的讀寫屏障語義。
1 static __inline__ size_t 2 xchg_mb (AO_t *var, AO_t val) 3 { 4 AO_t tmp = val; 5 __asm__ __volatile__( 6 "xchgq" " %0, %1" 7 : "=r"(tmp) 8 : "m"(*var), "0"(tmp) 9 : "memory"); 10 /* now tmp is the atomic's previous value */ 11 return tmp; 12 } 13 14 #define SPIN_BODY __asm__ __volatile__("rep;nop" : : : "memory") 15 #define COMPILER_BARRIER __asm__ __volatile__("" : : : "memory") /* 防止編譯器亂序 */ 16 17 __inline__ int 18 mcslock_init(mcslock_t *lock) 19 { 20 int res; 21 AO_store(&lock->tail, (AO_t)NULL); 22 res = pthread_key_create(&lock->qnode_key, NULL); 23 return res; 24 } 25 26 __inline__ void 27 mcslock_lock(mcslock_t *lock) 28 { 29 aligned_mcs_qnode_t *qnode, *pred; 30 qnode = (aligned_mcs_qnode_t *)pthread_getspecific(lock->qnode_key); 31 /* 獲得當前線程的MCS節點，如果沒有就創建一個 */ 32 if (qnode == NULL){ 33 /* 分配節點的時候cache對齊 */ 34 qnode = malloc(sizeof(aligned_mcs_qnode_t) + CACHE_LINE_SIZE - 1); 35 if (qnode == NULL) { 36 abort(); 37 } 38 if (((AO_t) qnode) & CACHE_LINE_MASK) { 39 qnode = (aligned_mcs_qnode_t *) 40 ((((AO_t) qnode) & ~CACHE_LINE_MASK) 41 + CACHE_LINE_SIZE); 42 } 43 qnode->node.locked = 0; 44 qnode->node.next = NULL; 45 pthread_setspecific(lock->qnode_key, (void *)qnode); 46 } 47 pred = (aligned_mcs_qnode_t *)xchg_mb(&lock->tail, (AO_t)qnode); 48 if (pred != NULL) { 49 qnode->node.locked = 1; 50 pred->node.next = &qnode->node; 51 AO_nop_write(); 52 COMPILER_BARRIER; 53 while (qnode->node.locked) { 54 SPIN_BODY; 55 } 56 } 57 } 58 59 __inline__ void 60 mcslock_unlock(mcslock_t *lock) 61 { 62 aligned_mcs_qnode_t *qnode = (aligned_mcs_qnode_t *)pthread_getspecific(lock->qnode_key); 63 if (qnode->node.next == NULL) { 64 /* 我應該是最后一個，賭一把將tail從我設置為NULL */ 65 if (AO_compare_and_swap(&lock->tail, 66 (AO_t)qnode, 67 (AO_t)NULL)) { 68 return; 69 } 70 /* 我的CAS沒成功，之前有人插到我后面了，那我只能等 */ 71 AO_nop_write(); 72 COMPILER_BARRIER; 73 while (qnode->node.next == NULL) { 74 SPIN_BODY; 75 } 76 } 77 /* 解鎖我后面的那個線程 */ 78 qnode->node.next->locked = 0; 79 qnode->node.next = NULL; 80 AO_nop_write(); 81 COMPILER_BARRIER; 82 }
那么John Mellor-Crummey和Michael Scott為什么要發明這么復雜的自旋鎖呢，而且還用到了SWAP和CAS這些高端大氣上檔次的原子操作？一句話，前面基於TAS的自旋鎖在多核系統上不具有很好的可伸縮性。每一次TAS操作都會寫入cache，那么所有線程的這條cache線都會失效，當線程數多的時候會導致大量的cache一致性流量。MCS鎖的優點在於每一個線程只自旋自己的變量，因此自旋完全可以在自己所在的核心的L1中完成，完全沒有任何cache一致性的問題，也不會產生內存訪問，也不會產生NUMA跨節點的流量。只有在轉交鎖的時候需要寫入一次后一個線程的lock變量。此外，MCS鎖由於使用了鏈表，所以還能保證公平，線程自然地按照先后次序排好隊。
那么由於cache造成的可伸縮性問題到底有多嚴重？畢竟MCS鎖本身也夠復雜。我在Intel Xeon Phi 5110P協處理器上做了實驗比較上文中兩種鎖的並發性能。實驗創建多個線程同時增加一個共享的計數器，通過自旋鎖保護臨界區。共享計數器的最終值設置為12000000，每一個線程均分自己的任務，統計不同線程數的情況下的執行時間。由於多個線程的並發鎖開銷，明顯只有1個線程的時候是耗時最短的。Intel Xeon Phi 5110P協處理器有60個通過環狀網絡連接的核心，每一個核心支持4個硬件線程，也就是說總共支持240個硬件線程。運行頻率1.1GHz，而且是不帶亂序執行的，所以單核性能還是挺低的。
這款協處理器的具體硬件規范參見 http://ark.intel.com/products/71992/Intel-Xeon-Phi-Coprocessor-5110P-8GB-1_053-GHz-60-core。
將線程數從1逐漸遞增到240，看看TAS自旋鎖和MCS自旋鎖的表現：
圖中藍色的圓點表示TAS鎖的數據，黑色的三角表示MCS鎖的數據。可以看出TAS鎖的運行時間在線程數較少的時候耗時並不長，但是隨着線程數增多的時候，運行時間在不斷增長。而MCS鎖的運行時間除了在線程數很少的時候非常短，然后基本上保持在一定的水平，說明MCS鎖在多核硬件上的伸縮性非常好。
再看下面的profile數據。
下圖是TAS鎖的profile數據，這個圖對應60個線程的時候，這60個線程用滿了60個核心，也就是說每一個核心上占用一個硬件線程：
可以看出CPI（clock per instruction）值非常高，表示指令的延遲非常高，這就是cache不斷invalidate造成的結果。
下面是MCS鎖在同樣條件下的profile數據
這個CPI小多了，只有2.幾，屬於非常正常的范圍。雖然MCS鎖的程序retire的指令是TAS鎖的80多倍，但是CPI卻只有后者的幾百分之一。此外由於MCS鎖的公平性，MCS的CPU利用率也比TAS的更為平衡。
所以，小小的自旋鎖里，也有不少學問哦。