C語言執行緒互斥和原子操作

如果多個執行緒存取相同的資料，並且它們中至少有一個修改了資料，那麼對共用資料的所有存取必須同步以防止資料競爭。但是，一個正在讀取共用資料的執行緒可能中斷另一個正在修改相同共用資料的執行緒，因此，可能導致執行緒讀取到不一致的資料。

甚至，由於程式在每次執行時系統可能排程不同的執行緒，導致每次執行程式時錯誤訊息只能間歇地反映當時情況，很難在測試中復現錯誤。如例 1 所示，哪怕是自增一個計數器這樣的簡單操作，都可能產生資料競爭。

【例1】沒有同步下的並行儲存存取

#include <stdio.h>
#include <threads.h>
#define COUNT 10000000L
long counter = 0
void incFunc(void) { for (long i = 0; i < COUNT; ++i)  ++counter; }
void decFunc(void) { for (long i = 0; i < COUNT; ++I)  --counter; }

int main(void)
{
    clock_t cl = clock()
    thrd_t th1, th2;
    if (thrd_create(&th1, (thrd-start_t)incFunc, NULL) != thrd_success
      || thrd_create(&th2, (thrd_start_t)decFunc, NULL) != thrd_success)
    {
        fpintf(stderr,"Error creating thread/n"); return -1;
    }
    thrd_join(th1, NULL);
    thrd_join(th2, NULL);

    printf("Counter: %ld t", counter);
    printf("CPU time: %ld msn", (clock()-cl)*1000L/CLOCKS_PER_SEC);
    return 0;
}

在程式結束時，計數器應為 0。然而，在沒有同步的情況下，結果則不同：程式每次執行時，最終獲得計數器值都是不同的。下面是一個典型的輸出範例：

Counter: -714573              CPU time: 59 ms

為了保障同步，C 標準庫提供了互斥操作（mutex operation）和原子操作（atomic operation）。

互斥

互相排斥（mutex exclusion）技術，簡稱為互斥（mutex），它用於防止多個執行緒同時存取共用資源。互斥技術採用一個物件控制獨占存取許可權，該物件稱之為互斥。配合條件變數（condition variable），互斥可以實現廣泛的同步存取控制。例如，它們允許程式設計師為資料存取操作指定執行次序。

在 C 程式中，一個互斥採用型別為 mtx_t 的物件表示，它能在一段時間內被一個執行緒鎖定，而其他執行緒必須等待，直到它被解鎖。在標頭檔案 threads.h 中，包括了關於互斥操作的所有宣告。最重要的互斥函數有：

int mtx_init（mtx_t*mtx，int mutextype）；

建立一個互斥，該互斥的屬性由 mutextype 指定。如果成功建立了一個新互斥，函數 mtx_init（）會將新互斥寫入由引數 mtx 參照的物件，然後返回宏值 thrd_success。

引數 mutextype 的取值可以是以下 4 個：

mtx_plain
mtx_timed
mtx_plain | mtx_recursive
mtx_timed | mtx_recursive

mtx_plain 表示請求一個簡單的互斥，它既不支援超時也不支援遞回，而其他 3 個值則表示支援超時和（或）遞回。

void mtx_destroy（mtx_t*mtx）；

銷毀 mtx 參照的互斥，並釋放它的所有資源。

int mtx_lock（mtx_t*mtx）；

阻塞正在呼叫的執行緒，直到該執行緒獲得引數 mtx 參照的互斥。除該互斥支援遞回的情況以外，正在呼叫的執行緒不能是已持有該互斥的執行緒。如果呼叫成功獲得互斥，則函數返回值 thrd_success，否則，返回值 thrd_error。

int mtx_unlock（mtx_t*mtx）；

釋放引數 mtx 參照的互斥。在呼叫函數 mtx_unlock（）之前，呼叫者必須持有該互斥。如果呼叫釋放互斥成功，則函數返回值 thrd_success，否則，返回值 thrd_error。

通常情況下，在程式碼某個關鍵區間（critical section）的起始點呼叫函數 mtx_lock（），在其結束點呼叫函數 mtx_unlock（），在這段區間中只有一個執行緒執行。

函數 mtx_lock（）還有兩個替代的選擇：一個選擇是函數 mtx_trylock（），如果該互斥恰好未被其他任何執行緒獲取，它則為當前執行緒獲得互斥，如果該互斥被其他執行緒獲取，它也不會阻塞當前執行緒；另一個選擇是函數 mtx_timedlock（），它僅在指定的時間內阻塞執行緒。所有這些函數都通過其返回值表明呼叫它們後，是否成功地獲得了互斥。

例 2 中的程式是例 1 的修改版本，它展示了如何使用互斥來消除對變數 counter 的資料競爭。

【例2】在例 1 的程式中新增一個互斥

#include <stdio.h>
#include <threads.h>

#define COUNT 10000000L

long counter = 0;
mtx_t mtx;                              // 為存取counter而設立的互斥
void incFunc(void)
{
    for (long i = 0; i < COUNT; ++i)
    {  mtx_lock(&mtx); ++counter; mtx_unlock(&mtx); }
}
void decFunc(void)
{
    for (long i = 0; i < COUNT; ++i)
    {  mtx_lock(&mtx); --counter; mtx_unlock(&mtx); }
}
int main(void)
{
    if (mtx_init(&mtx, mtx_plain) != thrd_success)
    {
        fprintf(stderr, "Error initializing the mutex.n");
        return -1;
    }
    // 如例14-2所示，啟動執行緒，等待它們完成，列印輸出
    mtx_destroy(&mtx);
    return 0;
}

函數 incFunc（）和 decFunc（）將不再並行地存取 counter，因為一次只有其中一個可以鎖定互斥（為保障可讀性，省略錯誤檢查）。現在，在程式結束時，計數器具有正確的值：0。下面是一個典型的輸出範例：

Counter: 0             CPU time: 650 ms

實現同步性需要付出代價。較高的 CPU 時間表明：修改後的程式需要大約 10 倍於原來的時間來執行。其原因是，通過鎖定互斥實現同步性遠比自增和自減一個變數具有更為複雜的操作。在不需要互斥鎖定的情況下，使用原子物件可以獲得更好的效能。

原子物件

原子物件（atomic object）是一個可通過原子操作（atomic operation）被讀取或修改的物件。原子操作是指不能被並行執行緒中斷的操作。在C11標準下，可以使用型別限定符_Atomic宣告一個原子物件（如果實現版本定義了宏__STDC_NO_ATOMICS__，則表明該實現版本不支援原子操作，自然也不能宣告原子物件）。例如，在例14-2程式中的變數counter可以通過以下方式宣告它為原子物件：

_Atomic long counter = ATOMIC_VAR_INIT(0L);

上述宣告定義了原子化的 long 型別變數 counter，並將其值初始化為 0。在標頭檔案 stdatomic.h 中定義了宏 ATOMIC_VAR_INIT，以及其他所有用於原子物件的宏、型別和宣告。特別是，stdatomic.h 中還定義了對應於所有整數型別的原子型別縮寫。例如，型別 atomic_uchar 等效於 _Atomic unsigned char。

語法 _Atomic（T）也可用於為給定的非原子型別 T 指定其對應的原子型別。陣列和函數型別不能為原子型別。然而，原子型別可以具有不同於其對應的非原子型別的空間大小和對齊方式。

原子操作

讀取或寫入一個原子物件是一個原子操作，也就是說它是不能被中斷的操作。這意味著：不同的執行緒可以同時存取一個原子物件而不引起競態條件。對於每個原子物件，物件的所有修改以一個確定的全域性化次序執行，這稱為該物件的修改次序（modification order）。

具有結構或聯合型別的原子物件只能被作為一個整體讀取或寫入：為了安全地存取單個成員，原子結構或聯合應首先複製到等效的非原子物件中。

注意，無論是使用宏 ATOMIC_VAR_INIT，還是通過泛型函數 ATOMIC_INIT（），一個原子物件的初始化不是一個原子操作。

原子操作通常用於進行讀-修改-寫操作。例如，字尾自增和自減運算子 ++ 和 --，當它們應用於原子物件時，是原子化的讀-修改-寫操作。同樣，複合賦值運算子，如 +=，當其原子化使用時，它們的左運算元是一個原子物件。

例 1 中的程式可以通過宣告變數 counter 作為原子物件，在不受任何其他影響下執行正確的計數，以最終獲得 0 值。該方案計時結果顯示，使用原子型別變數 counter 比例 2 所使用的互斥方法要快兩倍多。

除了已經提到的運算子，還有許多函數可以執行原子操作，包括 atomic_store（）、atomic_exchange（）和 atomic_compare_exchange_strong（）。

原子型別具有無鎖（lock-free）屬性，它表示不使用鎖定和解鎖操作實現對一個原子物件的原子存取。該方式只需要使用型別 atomic_flag（它是一個結構型別）以確保實現無鎖，atomic_flag 有“設定”和“清除”兩種狀態。宏 ATOMIC_FLAG_INIT 將一個 atomic_flag 物件初始化為“清除”狀態，如以下範例宣告所示：

atomic_flag done = ATOMIC_FLAG_INIT;

C11 提供了函數 atomic_flag_test_and_set（）和 atomic_flag_clear（），由此對一個 atomic_flag 物件執行狀態操作。整型原子型別通常也都是無鎖的。要確定一個給定的型別是否是無鎖的，程式可以檢查宏 ATOMIC_type_LOCK_FREE，其中 type 是一個指定整數型別的大寫縮寫，如 BOOL、INT，或 LLONG。

與指標型別對應的宏是 ATOMIC_POINTER_LOCK_FREE。所有這些宏的值可能為 0、1 或 2。值為 0，表示該型別不是無鎖的；值為 1，表示該型別對特定物件是無鎖的；值為 2，表示該型別始終是無鎖的。或者，可以呼叫泛型函數來確定一個給定的原子物件是否是無鎖的：

_Bool atomic_is_lock_free(const volatile A *obj);

在函數引數宣告中的預留位置A代表任一原子型別。因此，引數 obj 為指標，它指向任一給定原子物件。