struct Message
{
// 用來將多個訊息組成串列而需要的資料結構
struct Message *prev;
struct Message *next;
// 記錄訊息內容以及產生訊息的時間
unsigned ctime;
char str[128];
};執行緒同步 6：互斥子
上集介紹了應用在執行緒同步用途當中最簡單快速便利的原子變數(Atomic Variable)。 然而原子變數的使用彈性相當有限， 在實際的應用用途中但凡我們程式的資料運算邏輯稍微複雜一點，原子變數很容易就不敷使用了。 那麼本篇我就來介紹一種在多緒同步用途中使用最廣泛、出鏡率最高的一種同步工具：互斥子(Mutex)。
使用情境
在解釋互斥子之前，先讓我們看一段程式碼，以先了解適合引入互斥子的使用情境。
這次的測試範例是一個典型的「生產者/消費者」模式， 範例程式建立了一個稱為「工廠」的執行緒，在一分鐘的時間內每隔 50 毫秒就會產生一筆消息出來， 然後另外 4 個訂閱了訊息的「消費者」執行緒就會不斷的嘗試去拿取工廠發送出來的資料，拿到之後就去處理那份資料。 當然作為一個測試用的範例，所謂的處理資料也就是將訊息內所記錄的字串給輸出出來而已。
那麼首先，下面這個就是本次範例所圍繞的中心，也就是上面描述的那個「消息」的資料格式：
這個消息採用雙向鏈結串列的設計， 這樣萬一消費者來不及取走資料的話，工廠不斷生產出來的資料就會組成一個串列一直暫存在那裡等待被取走， 而不會被覆蓋或流失掉。 不過關於鏈結串列的細節部份並不是本篇的重點，本篇只是借用這套演算法而已，所以本篇不會過度注重在相關的內容上。 如果您不知道什麼是鏈結串列且又有興趣想要了解學習，那麼可以參考其它的說明資源 [1]。
有了消息本身的資料結構以後，下面就是用來產生消息的執行緒：
struct Factory
{
// 記錄要產生的消息數量和時間間隔
unsigned period;
unsigned total_num;
// 用來堆放消息的鏈表頭尾指標
struct Message *first;
struct Message *last;
// 執行緒本身有關的資源
thrd_t thrd;
};
int FactoryThread(struct Factory *data)
{
for(unsigned i = 0; i < data->total_num; ++i)
{
struct Message *msg = malloc(sizeof(*msg));
msg->ctime = GetMsTime();
snprintf(msg->str, sizeof(msg->str), "message-%u", i);
printf("Factory: Generate \"%s\"\n", msg->str);
msg->prev = data->last;
msg->next = NULL;
if(!data->first)
data->first = msg;
data->last = msg;
SleepMs(data->period);
}
return 0;
}如上所見，工廠執行緒內容就是一個簡單的迴圈，定時不斷產生訊息並將訊息推送到鏈表緩存，等待消費者稍後取走。
完了之後再看下面的消費者執行緒都在做什麼？ 反正執行緒的函式外殼都是一樣的，這裡就不再重複展示了， 下面只展示消費者執行緒函式裡面的重點迴圈部份：
while(!data->go_term)
{
struct Factory *src = data->msg_src;
struct Message *msg = src->first;
if(msg)
{
src->first = msg->next;
if(!src->first)
src->last = NULL;
if(msg->next)
msg->next->prev = NULL;
unsigned delay = GetMsTime() - msg->ctime;
printf("Consumer[%u]: Process \"%s\", delay=%u\n",
data->idx, msg->str, delay);
free(msg);
}
}如上所見，消費者執行緒就是不斷探測工廠有沒有存在可用的消息？ 如果有的話就把消息從鏈表裡取走，然後處理消息，也就是顯示消息的內容。 其中注意在執行緒的迴圈裡就使用到了前篇介紹過的用來 正常結束執行緒的方法。 每次迴圈都會檢查一個旗標值，只要旗標值沒被設定的話就繼續不斷反覆執行迴圈內的工作，執行緒就會永遠處在工作中不會返回； 而外面的控制者 ，在範例中就是我們的主函式，則可以透過設定這個旗標值來讓執行緒知道該結束工作了。
上面這個範例的完整程式碼可以從這裡下載。 然而光看上面的示例說明，眼尖的讀者可能已經發現不對勁的地方了！ 上面兩個執行緒(其實是 5 個執行緒，因為消費者執行緒一共建立了 4 條)彼此共同存取了好多變數， 而這些變數的存取完全沒有任何保護，肯定會發生執行緒資料競爭的問題。 果不其然，這個範例從一執行開始就發生了許多奇異現象，執行沒有幾步更是直接當掉了！ 下面就是這個範例在我電腦上執行的結果：
Factory: Generate "message-0"
Consumer[0]: Process "message-0", delay=0
Factory: Generate "message-1"
Consumer[0]: Process "message-1", delay=0
Consumer[2]: Process "message-1", delay=0
Consumer[3]: Process "message-1", delay=0
Consumer[1]: Process "message-1", delay=0
Factory: Generate "message-2"
Consumer[2]: Process "message-2", delay=0
Consumer[3]: Process "message-2", delay=0
Consumer[1]: Process "message-2", delay=0
Consumer[0]: Process "message-2", delay=0
Factory: Generate "message-3"
Consumer[3]: Process "message-3", delay=0
Consumer[2]: Process "message-3", delay=0
Consumer[1]: Process "message-3", delay=0
Consumer[0]: Process "message-3", delay=0
Factory: Generate "message-4"
Consumer[2]: Process "message-4", delay=0
Consumer[3]: Process "message-4", delay=0
Consumer[1]: Process "message-4", delay=0
Consumer[0]: Process "message-4", delay=0
Factory: Generate "message-5"
Consumer[2]: Process "message-5", delay=0
Consumer[1]: Process "message-5", delay=0
Consumer[3]: Process "message-5", delay=0
Consumer[0]: Process "message-5", delay=0
Factory: Generate "message-6"
Consumer[2]: Process "message-6", delay=0
Consumer[1]: Process "message-6", delay=0
Consumer[3]: Process "message-6", delay=0
Consumer[0]: Process "message-6", delay=0
Factory: Generate "message-7"
Consumer[2]: Process "message-7", delay=0
Consumer[1]: Process "message-7", delay=0
Consumer[3]: Process "message-7", delay=0
Consumer[0]: Process "message-7", delay=0
Factory: Generate "message-8"
Consumer[2]: Process "message-8", delay=0
Consumer[3]: Process "message-8", delay=0
double free or corruption (out)
Consumer[0]: Process "message-8", delay=0
Aborted (core dumped)既然會發生執行緒競爭的問題，那麼就要來想辦法避免競爭。
那將上範例中的那些共同存取的變數改成使用原子變數可不可以呢？這個方法在本篇的範例裡顯然是解決不了問題的！
首先這並不是因為本次的問題出在指標上而不能保護，
因為 C11 定義中也存在 atomic_uintptr 這樣的原子變數型態，
而之所以無法解決衝突的問題是因為這次牽涉其中的變數太多了！
原子變數只能夠保證對那個變數本身的存取與修改是無法被中途插隊的原子操作，
可是本次範例中暴露在多緒共同編修的變數卻有好多個，
就算每一個都改使用原子變數，也沒有辦法阻止在一個接一個執行原子變數的操作中間被別的執行緒插隊的問題。
這就是光使用原子變數也對於像本篇範例這樣的使用情境只能雙手一攤的原因所在了。 那麼為了解決在這樣的使用場景中所面臨的執行緒衝突問題，我們就需要更加強大的保護機制(或稱同步機制)。
互斥子(Mutex)
先不多做解釋，先來看看給上面的範例加上互斥子的樣子。 首先是消費者執行緒的部份：
while(!data->go_term)
{
struct Factory *src = data->msg_src;
mtx_lock(&src->mutex); // 加了這一行
struct Message *msg = src->first;
if(msg)
{
src->first = msg->next;
if(!src->first)
src->last = NULL;
if(msg->next)
msg->next->prev = NULL;
}
mtx_unlock(&src->mutex); // 加了這一行
if(msg)
{
unsigned delay = GetMsTime() - msg->ctime;
printf("Consumer[%u]: Process \"%s\", delay=%u\n",
data->idx, msg->str, delay);
free(msg);
}
}然後是工廠執行緒的部份：
for(unsigned i = 0; i < data->total_num; ++i)
{
struct Message *msg = malloc(sizeof(*msg));
msg->ctime = GetMsTime();
snprintf(msg->str, sizeof(msg->str), "message-%u", i);
printf("Factory: Generate \"%s\"\n", msg->str);
mtx_lock(&data->mutex); // 加了這一行
msg->prev = data->last;
msg->next = NULL;
if(!data->first)
data->first = msg;
data->last = msg;
mtx_unlock(&data->mutex); // 加了這一行
SleepMs(data->period);
}而當然，幾個執行緒必需要操作的是同一個互斥子才能夠發揮效果， 於是這邊選擇將互斥子的實體放在工廠執行緒的資料裡面，就在鏈結串列相關變數的旁邊：
struct Factory
{
unsigned period;
unsigned total_num;
mtx_t mutex; // 加了這一行
struct Message *first;
struct Message *last;
thrd_t thrd;
};上面的程式修改內容其實就是簡單的在所有進行那些會導致衝突的「關鍵程式碼」
前後分別加上 mtx_lock() 和 mtx_unlock() 呼叫。
修改完成後，這個範例程式的執行就沒有問題了，從頭到尾一氣呵成也無錯誤歧異。
範例的完整程式碼可以從這裡下載，
整個範例程式的輸出有兩千多行，下面只節錄其中一小片段作為代表：
Factory: Generate "message-71"
Consumer[0]: Process "message-71", delay=0
Factory: Generate "message-72"
Consumer[1]: Process "message-72", delay=0
Factory: Generate "message-73"
Consumer[3]: Process "message-73", delay=0
Factory: Generate "message-74"
Consumer[0]: Process "message-74", delay=0
Factory: Generate "message-75"
Consumer[2]: Process "message-75", delay=0
Factory: Generate "message-76"
Consumer[2]: Process "message-76", delay=0
Factory: Generate "message-77"
Consumer[0]: Process "message-77", delay=0
Factory: Generate "message-78"
Consumer[2]: Process "message-78", delay=0
Factory: Generate "message-79"
Consumer[3]: Process "message-79", delay=0
Factory: Generate "message-80"
Consumer[0]: Process "message-80", delay=0從上面結果大致可以窺探，訊息的產生與消化、鏈結串列的各種操作都沒有意外，
沒有訊息被多拿也沒有被少拿，指標變數的操作看似也沒有任何異常或錯亂，
4 個等待索取訊息的執行緒也大致都能平均拿到訊息而無明顯的不平衡。
一切問題都解決了，就只是在關鍵操作的前後加上 mtx_lock() 和 mtx_unlock() 呼叫而已。
互斥子的使用
細說互斥子的使用，其使用與呼叫的基本結構如下：
// 這裡執行那些沒有競爭問題的程式碼
...
...
mtx_t *mutex = ...; // 不知從哪裡拿來的互斥子
mtx_lock(mutex); // 向互斥子要求一個鎖，如果沒有鎖的話那就等待到有為止
// 這裡執行那些原本會產生衝突競爭的程式碼
...
...
mtx_unlock(mutex); // 將鎖歸還給互斥子
// 繼續執行其它沒有競爭問題的程式碼
...
...|
Note
|
由於不同的執行緒函式庫都存在互斥子且操作方式基本相同，
因此下面將以更加概念通用且簡潔的名稱 lock() 和 unlock() 來表示互斥子的上鎖與解鎖呼叫，
而不寫成 C11 標準裡的 mtx_lock() 和 mtx_unlock()。
|
互斥子就像是一個守門員，而 lock() 的呼叫其實就是向守門員索取一張通行牌，也就是鎖(lock)。
互斥子給你一張通行牌，你就可以通行，這時 lock() 的呼叫就會正常返回，而你也就能繼續執行下面的程式碼。
不過互斥子只有一張通行牌，如果同時有很多人都呼叫了 lock() 的話，
那麼互斥子只能選擇其中一個幸運兒(通常就是最早呼叫的那個人)並把通行牌給他，而其它人就只能夠等待放行，
具體就是除了那個幸運者之外的其他人都會被卡在 lock() 呼叫裡面不會返回。
於是如果所有需要操作那些原本能導致衝突的程式碼的執行緒都自律的在執行操作前先呼叫 lock() 的話，
那麼大部份執行緒都得等待，只有也只會有一個執行緒能夠得到繼續往下執行的機會；
既然只有一個執行緒在執行，自然就不會有衝突的問題，
它可以從容的把所有操作處理完而無需擔心被別的執行緒從中插隊而導致錯亂。
等那個幸運抽中通行牌的執行緒執行完那些關鍵的操作之後，就可以呼叫 unlock() 作結，
這個呼叫的意義就是將手上持有的通行牌歸還給互斥子。
於是拿回了通行牌的互斥子就會再從剛才呼叫了 lock() 而進入等待的那些人之中挑選一個幸運兒，
然後將通行牌發給他，於是他的 lock() 呼叫就會返回，並能像前一個執行緒那樣繼續執行下面的關鍵程式碼，
並且只會有他這一個執行緒能夠執行那段程式碼，而其它執行緒依然繼續等待。
如此類推反覆，互斥子透過守門的方式讓所有執行緒排隊等待放行，一次只準一個放行通過，如此就解決了多緒共同競爭的問題。
並且因為互斥子的保護範圍是從 lock() 到 unlock() 之間的所有內容，
因此在那中間可以寫上任何複雜的操作都無所謂，操作空間和彈性非常大，完全不像原子變數那樣在用途上相當受限。
這就是為什麼互斥子是在所有解決多緒問題當中最受廣泛使用的手段了，
事實上在前篇曾描述過的那些 CPU 指令並不支援的原子變數，
或甚至像 C++ 可以把任何自訂型態都變成原子變數的那些「假原子變數」，其內部都是使用這樣的互斥子來實現的。
而那一段介於 lock() 呼叫到 unlock() 呼叫之間受保護的程式碼區域就有一個特別的名詞用來稱呼，
那段受到互斥子保呼的程式碼區域就叫作「臨界區段」(Critical Section)。
互斥子的代價
既然互斥子這麼好用，使用彈性大，臨界區段能包容一切程式碼，那麼是不是所有的問題都用互斥子來解決就好了 (事實上確實大部份的多緒衝突問題都用互斥子來處理)，什麼原子變數什麼的其它方法手段毫無價值也無學習的必要了呢？ 當然也不是這樣，因為互斥子的使用本身也是附帶相當大的代價的！ 從前面的說明內容來看，互斥子能夠解決多緒衝突的核心方法就是讓執行緒排隊，一次只放行一個執行緒通過， 其本質就是把多緒條件退化為單緒條件從而解決問題。 那麼這有什麼問題呢？問題就在於大家得「等」！ 說的誇張一點，我們甚至可以把執行緒函式內的所有內容全都包在臨界區段內， 但結果就會變成所有的執行緒都得排隊等著一個一個獲得執行的機會，活生生把多緒又回歸到單緒去了。
因此為了儘量減少執行緒之間大量的互相等待時間，我們在使用上就應該要儘量縮減臨界區段的內容，
基本就是把所有其實不會衝突、不需要受到保護的內容移出到臨界區段外面去執行，讓那些程式碼可以在不同的執行緒中去平行進行，
只保留最少量的那些真正可能會產生衝突而必須被保護的程式碼在臨界區段內，使執行緒需要排隊等待的範圍縮減到愈小愈好。
在本篇前面的互斥子範例中就已經展現出這樣的設計原則，
只將對鏈結串列的操作內容包裹在 lock() 和 unlock() 中間的臨界區段內，
至於配置訊息記憶體空間以及填寫其內容的部份、以及處理和輸出訊息內容的部份，則放在臨界區段之外，
不讓那些實際上不需要受到保護的程式碼去拖慢大家等待的時間。
然而即便已經試圖最大程度的壓縮臨界區段，互斥子的使用依然會帶來相當的性能開銷。 其原因在於前面所說的「等待」通行牌的過程其實就是讓執行緒進入睡眠狀態，而睡眠的喚醒並不是即時的， 具體得看作業系統的排程安排(可參閱前篇說明)。 總之就是當一個執行緒歸還通行牌後，互斥子將通行牌派發給下一個執行緒， 到下一個執行緒從休眠狀態回復過來真正開始執行之間存在一定程度的時間延遲。 而具體到底能差到多少呢？讓我們把前篇那個計算累加的範例 改用互斥子來做：
for(int i = 0; i < n; ++i)
{
mtx_lock(&data->mutex);
data->value += x;
mtx_unlock(&data->mutex);
}跑出來的結果如下：
Finished, value=800000000/800000000, spend=31544可看到這個測試在我的電腦上一共花費了 30 多秒完成， 對比使用原子變數的方案花費了大約 5 秒多，使用互斥子的方案大約使用了 6 倍左右的時間！ (當然連原子變數都不用而只用普通無保護變數的程式版本只花了 1 秒多的時間， 但是因為計算結果完全不正確，所以僅作為參考而已。) 總之結算下來，使用原子變數作為多緒之間的確保手段已經付出了 5 倍的效能劣化代價， 而使用互斥子則對比原子變數又是 6 倍的效能劣化代價， 足見互斥子(以及其它所有後續介紹的高級鎖)的副作用不容小覷，也足見原子變數屬實高效快速。
總結
本篇介紹了在解決執行緒競爭問題上最經常被使用的互斥子，學會了它的使用結構與邏輯， 也認識到互斥子的使用所可能帶來的性能開銷。 上鎖與解鎖互斥子由於牽涉到用戶態與核心態的切換、以及作業系統排程的參與，因此可能產生較長的時間延遲。 這就是為什麼雖然互斥子的使用彈性相當大，開鎖與解鎖中間的程式碼可以包山包海， 但人們有經常還是會在特定的地方更加青睞於只需要 CPU 指令排他、以及些許快取性能損失的原子變數的原因了。 然而也同樣因為原子變數的使用限制大的因素， 使得我們總還是經常不得不使用那些更高級的執行緒鎖(例如本篇介紹的互斥鎖)。 因此在必須使用互斥鎖的情況下， 通常我們會傾向於儘量限縮上鎖與解鎖中間的程式碼規模，只在臨界區域裡面執行真正需要受到保護的衝突內容， 在得到通行權之後儘快處理玩事情並歸還通行權，減少其它執行緒不必要的等待時間。
下一篇：「執行緒同步 7：自旋鎖」