1、用鎖來防止數據競跑

在進行并行編程時，我們常常需要使用鎖來保護共享變量，以防止多個線程同時對該變量進行更新時產生數據競跑（Data Race）。所謂數據競跑，是指當兩個（或多個）線程同時對某個共享變量進行操作，且這些操作中至少有一個是寫操作時所造成的程序錯誤。例1中的兩個線程可能同時執行“counter++”從而產生數據競跑，造成counter最終值為1（而不是正確值2）。

例1：

#include <pthread.h>
int counter = 0;
void *func(void *params)
{
    counter++; //數據競跑
}
void main()
{
    pthread_t thread1, thread2;
    pthread_create(&thread1, 0, func, 0);
    pthread_create(&thread2, 0, func, 0);
    pthread_join(thread1, 0 );
    pthread_join(thread2, 0 );
}

這是因為counter++本身是由三條匯編指令構成的（從主存中將counter的值讀到寄存器中；對寄存器進行加1操作；將寄存器中的新值寫回主存），所以例1中的兩個線程可能按如下交錯順序執行，導致counter的最終值為1：

例2：

load [%counter], rax; // 線程1從counter讀取0到寄存器rax
add rax, 1; // 線程1對寄存器rax進行加1
load [%counter], rbx; // 線程2從counter讀取0到寄存器rbx
store rax [%counter]; // 線程1把1寫入counter的主存地址
add rbx, 1; // 線程2對寄存器rbx進行加1
store rbx, [%counter]; // 線程2把1寫入counter的主存地址

為了防止例1中的數據競跑現象，我們可以使用鎖來保證每個線程對counter++操作的獨占訪問（即保證該操作是原子的）。在例3的程序中，我們使用mutex鎖將counter++操作放入臨界區中，這樣同一時刻只有獲取鎖的線程能訪問該臨界區，保證了counter++的原子性：即只有在線程1執行完counter++的三條指令之后線程2才能執行counter++操作，保證了counter的最終值必定為2。

例3：

#include <pthread.h>
int counter = 0;
pthread_mutex_t mutex;
void *func(void *params)
{
    pthread_mutex_lock(&mutex);
    counter++; //處于臨界區，不會產生數據競跑
    pthread_mutex_unlock(&mutex);
}
void main()
{
    pthread_t thread1, thread2;
    pthread_mutex_init(&mutex);
    pthread_create(&thread1, 0, func, 0);
    pthread_create(&thread2, 0, func, 0);
    pthread_join(thread1, 0 );
    pthread_join(thread2, 0 );
    pthread_mutex_destroy(&mutex);
}

2. 死鎖和活鎖

然而，鎖的使用非常容易導致多線程Bug，最常見的莫過于死鎖和活鎖。從原理上講，死鎖的產生是由于兩個（或多個）線程在試圖獲取正被其他線程占有的資源時造成的線程停滯。在下例中，假設線程1在獲取mutex_a鎖之后正在嘗試獲取mutex_b鎖，而線程2此時已經獲取了mutex_b鎖并正在嘗試獲取mutex_a鎖，兩個線程就會因為獲取不到自己想要的資源、且自己正占有著對方想要的資源而停滯，從而產生死鎖。

例4：

// 線程 1                     
void func1()                    
{                       
    LOCK(&mutex_a);                      
    LOCK(&mutex_b);//線程1停滯在此         
    counter++;                       
    UNLOCK(&mutex_b);                     
    UNLOCK(&mutex_a);                    
}                       
 
// 線程 2
void func2()
{
    LOCK(&mutex_b);
    LOCK(&mutex_a);//線程2停滯在此
    counter++;
    UNLOCK(&mutex_a);
    UNLOCK(&mutex_b);
}

例4中的死鎖其實是最簡單的情形，在實際的程序中，死鎖往往發生在復雜的函數調用過程中。在下面這個例子中，線程1在func1()中獲取了mutex_a鎖，之后調用func_call1()并在其函數體中嘗試獲取mutex_b鎖；與此同時線程2在func2()中獲取了mutex_b鎖之后再在func_call2()中嘗試獲取mutex_a鎖從而造成死鎖。可以想象，隨著程序復雜度的增加，想要正確的檢測出死鎖會變得越來越困難。

例5：

// 線程 1                     
void func1()                    
{                       
LOCK(&mutex_a);                      
...                     
func_call1();                
UNLOCK(&mutex_a);                
}                       
 
func_call1()                    
{                       
   LOCK(&mutex_b);               
   ...                       
   UNLOCK(&mutex_b);                 
   ...                       
}                       
 
// 線程 2
void func2()
{
    LOCK(&mutex_b);
    ...
    func_call2()
    UNLOCK(&mutex_b);
}
 
func_call2()
{
    LOCK(&mutex_a);
    ...
    UNLOCK(&mutex_b);
    ...
}

其實避免死鎖的方法非常簡單，其基本原則就是保證各個線程加鎖操作的執行順序是全局一致的。例如，如果上例中的線程1和線程2都是先對mutex_a加鎖再對mutex_b進行加鎖就不會產生死鎖了。在實際的軟件開發中，除了嚴格遵守相同加鎖順序的原則防止死鎖之外，我們還可以使用RAII（Resource Acquisition Is Initialization，即“資源獲取即初始化”）的手段來封裝加鎖解鎖操作，從而幫助減少死鎖的發生[1]。

除死鎖外，多個線程的加鎖、解鎖操作還可能造成活鎖。在下例中，程序員為了防止死鎖的產生而做了如下處理：當線程1在獲取了mutex_a鎖之后再嘗試獲取mutex_b時，線程1通過調用一個非阻塞的加鎖操作（類似pthread_mutex_trylock）來嘗試進行獲得mutex_b：如果線程1成功獲得mutex_b，則trylock()加鎖成功并返回true，如果失敗則返回false。線程2也使用了類似的方法來保證不會出現死鎖。不幸的是，這種方法雖然防止了死鎖的產生，卻可能造成活鎖。

例如，在線程1獲得mutex_a鎖之后嘗試獲取mutex_b失敗，則線程1會釋放mutex_a并進入下一次while循環；如果此時線程2在線程1進行TRYLOCK(&mutex_b)的同時執行TRYLOCK(&mutex_a)，那么線程2也會獲取mutex_a失敗，并接著釋放mutex_b及進入下一次while循環；如此反復，兩個線程都可能在較長時間內不停的進行“獲得一把鎖、嘗試獲取另一把鎖失敗、再解鎖之前已獲得的鎖“的循環，從而產生活鎖現象。當然，在實際情況中，因為多個線程之間調度的不確定性，最終必定會有一個線程能同時獲得兩個鎖，從而結束活鎖。盡管如此，活鎖現象確實會產生不必要的性能延遲，所以需要大家格外注意。

例6：

// 線程 1                     
void func1()                    
{                       
    int done = 0;                   
    while(!done) {               
        LOCK(&mutex_a);                      
        if (TRYLOCK(&mutex_b)) {               
            counter++;                   
            UNLOCK(&mutex_b);                        
            UNLOCK(&mutex_a);                        
            done = 1;                        
        }                          
        else {                     
            UNLOCK(&mutex_a);                   
        }                          
    }                        
}                       
 
// 線程 2
void func2()
{
    int done = 0;
    while(!done) {
        LOCK(&mutex_b);
        if (TRYLOCK(&mutex_a)) {
            counter++;
            UNLOCK(&mutex_a);
            UNLOCK(&mutex_b);
            done = 1; 
        }
        else {
            UNLOCK(&mutex_b);
        }
    }
}

3. 鎖競爭性能瓶頸

在多線程程序中鎖競爭是最主要的性能瓶頸之一。在前面我們也提到過，通過使用鎖來保護共享變量能防止數據競跑，保證同一時刻只能有一個線程訪問該臨界區。但是我們也注意到，正是因為鎖造成的對臨界區的串行執行導致了并行程序的性能瓶頸。

3.1阿姆達爾法則（Amdahl’s Law）

在介紹鎖競爭引起的性能瓶頸之前，讓我們先來了解一下阿姆達爾法則。我們知道，一個并行程序是由兩部分組成的：串行執行的部分和可以并行執行的部分。假設串行部分的執行時間為S，可并行執行部分的執行時間為P，則整個并行程序使用單線程（單核）串行執行的時間為S+P。阿姆達爾法則規定，可并行執行部分的執行時間與線程數目成反比：即如果有N個線程（N核CPU）并行執行這個可并行的部分，則該部分的執行時間為P/N。由此我們可以得到并行程序總體執行時間的公式：

總體執行時間 T = S + P/N

根據這個公式，我們可以得到一些非常有意思的結論。例如，如果一個程序全部代碼都可以被并行執行，那么它的加速比會非常好，即隨著線程數（CPU核數）的增多該程序的加速比會線性遞增。換句話說，如果單線程執行該程序需要16秒鐘，用16個線程執行該程序就只需要1秒鐘。

然而，如果這個程序只有80%的代碼可以被并行執行，它的加速比卻會急劇下降。根據阿姆達爾法則，如果用16個線程并行執行次程序可并行的部分，該程序的總體執行時間T = S + P/N = (160.2) + (160.8)/16 = 4秒，這比完全并行化的情況（只需1秒）足足慢了4倍！實際上，如果該程序只有50%的代碼可以被并行執行，在使用16個線程時該程序的執行時間仍然需要8.5秒！

從阿姆達爾法則我們可以看到，并行程序的性能很大程度上被只能串行執行的部分給限制住了，而由鎖競爭引起的串行執行正是造成串行性能瓶頸的主要原因之一。

3.2鎖競爭的常用解決辦法

3.2.1 避免使用鎖

為了提高程序的并行性，最好的辦法自然是不使用鎖。從設計角度上來講，鎖的使用無非是為了保護共享資源。如果我們可以避免使用共享資源的話那自然就避免了鎖競爭造成的性能損失。幸運的是，在很多情況下我們都可以通過資源復制的方法讓每個線程都擁有一份該資源的副本，從而避免資源的共享。如果有需要的話，我們也可以讓每個線程先訪問自己的資源副本，只在程序的后講各個線程的資源副本合并成一個共享資源。例如，如果我們需要在多線程程序中使用計數器，那么我們可以讓每個線程先維護一個自己的計數器，只在程序的最后將各個計數器兩兩歸并（類比二叉樹），從而最大程度提高并行度，減少鎖競爭。

3.2.2 使用讀寫鎖

如果對共享資源的訪問多數為讀操作，少數為寫操作，而且寫操作的時間非常短，我們就可以考慮使用讀寫鎖來減少鎖競爭。讀寫鎖的基本原則是同一時刻多個讀線程可以同時擁有讀者鎖并進行讀操作；另一方面，同一時刻只有一個寫進程可以擁有寫者鎖并進行寫操作。讀者鎖和寫者鎖各自維護一份等待隊列。當擁有寫者鎖的寫進程釋放寫者鎖時，所有正處于讀者鎖等待隊列里的讀線程全部被喚醒并被授予讀者鎖以進行讀操作；當這些讀線程完成讀操作并釋放讀者鎖時，寫者鎖中的第一個寫進程被喚醒并被授予寫者鎖以進行寫操作，如此反復。

換句話說，多個讀線程和一個寫線程將交替擁有讀寫鎖以完成相應操作。這里需要額外補充的一點是鎖的公平調度問題。例如，如果在寫者鎖等待隊列中有一個或多個寫線程正在等待獲得寫者鎖時，新加入的讀線程會被放入讀者鎖的等待隊列。這是因為，盡管這個新加入的讀線程能與正在進行讀操作的那些讀線程并發讀取共享資源，但是也不能賦予他們讀權限，這樣就防止了寫線程被新到來的讀線程無休止的阻塞。

需要注意的是，并不是所有的場合讀寫鎖都具備更好的性能，大家應該根據Profling的測試結果來判斷使用讀寫鎖是否能真的提高性能，特別是要注意寫操作雖然很少但很耗時的情況。

3.2.3 保護數據而不是操作

在實際程序中，有不少程序員在使用鎖時圖方便而把一些不必要的操作放在臨界區中。例如，如果需要對一個共享數據結構進行刪除和銷毀操作，我們只需要把刪除操作放在臨界區中即可，資源銷毀操作完全可以在臨界區之外單獨進行，以此增加并行度。正是因為臨界區的執行時間大大影響了并行程序的整體性能，我們必須盡量少在臨界區中做耗時的操作，例如函數調用，數據查詢，I/O操作等。簡而言之，我們需要保護的只是那些共享資源，而不是對這些共享資源的操作，盡可能的把對共享資源的操作放到臨界區之外執行有助于減少鎖競爭帶來的性能損失。

3.2.4 盡量使用輕量級的原子操作

在例3中，我們使用了mutex鎖來保護counter++操作。實際上，counter++操作完全可以使用更輕量級的原子操作來實現，根本不需要使用mutex鎖這樣相對較昂貴的機制來實現。在今年程序員第四期的《volatile與多線程的那些事兒》中我們就有對Java和C/C++中的原子操作做過相應的介紹。

3.2.5 粗粒度鎖與細粒度鎖

為了減少串行部分的執行時間，我們可以通過把單個鎖拆成多個鎖的辦法來較小臨界區的執行時間，從而降低鎖競爭的性能損耗，即把“粗粒度鎖”轉換成“細粒度鎖”。但是，細粒度鎖并不一定更好。這是因為粗粒度鎖編程簡單，不易出現死鎖等Bug，而細粒度鎖編程復雜，容易出錯；而且鎖的使用是有開銷的（例如一個加鎖操作一般需要100個CPU時鐘周期），使用多個細粒度的鎖無疑會增加加鎖解鎖操作的開銷。在實際編程中，我們往往需要從編程復雜度、性能等多個方面來權衡自己的設計方案。

事實上，在計算機系統設計領域，沒有哪種設計是沒有缺點的，只有仔細權衡不同方案的利弊才能得到最適合自己當前需求的解決辦法。例如，Linux內核在初期使用了Big Kernel Lock（粗粒度鎖）來實現并行化。從性能上來講，使用一個大鎖把所有操作都保護起來無疑帶來了很大的性能損失，但是它卻極大的簡化了并行整個內核的難度。當然，隨著Linux內核的發展，Big Kernel Lock已經逐漸消失并被細粒度鎖而取代，以取得更好的性能。

3.2.6 使用無鎖算法、數據結構

首先要強調的是，筆者并不推薦大家自己去實現無鎖算法。為什么別去造無鎖算法的輪子呢？因為高性能無鎖算法的正確實現實在是太難了。有多難呢? Doug Lea 提到 java.util.concurrent 庫中一個Non Blocking的算法的實現大概需要1個人年，總共約500行代碼。事實上，我推薦大家直接去使用一些并行庫中已經實現好了的無鎖算法、無鎖數據結構，以提高并行程序的性能。典型的無鎖算法的庫有java.util.concurrent，Intel TBB等，它們都提供了諸如Non-blocking concurrent queue之類的數據結構以供使用。