第三節 線程互斥
本節介紹如下內容
1.
主動對象
2.
調度與原子操作
3.
競爭條件和數據一致性
4.
為何需要互斥
5.
互斥類接口定義
6.
示例程序
7.
互斥類的Unix和Windows實現
1. 主動對象(Active Object)
第二節介紹Thread類的時候曾經提到，每個子線程一旦被創建，就被賦予了自己的生命。當一個線程類創建并啟動之后，它將會以自己的步調主動執行其獨立線程，它和其他線程（包括主線程）的執行是并行關系。
為了詳細說明這個問題，先介紹一下什么是控制流(control
flow)。計算機科學中，控制流指指令式(imperative)或函數式(functional)編程語言中語句、指令、函
數調用的執行（或求值）序列。
單線程程序中，控制流始終是一線串珠，一旦控制流達到某個對象，由程序主動調用對象函數，在函數執行完畢后，控制流返回主程序繼續執行。對于被訪問對象來 說，訪問、修改、成員函數調用都是一個被動的過程，此過程受控于該程序的控制流。所以，稱單線程程序的對象為被動對象(Passive Object)。
與被動對象相對應的是主動對象。主動對象和被動對象的區別在于，主動對象可以發起自己的線程，創建新的執行序列，產生獨立于主線程的控制流分支。簡而言 之，主動對象可以獨立于主線程，自主制定自己的控制流。如果愿意，你可以實現一個和主線程沒有任何協調關系，天馬行空式的獨立線程。當然，這樣的線程往往 也意味著毫無用處（娛樂作用除外）。
2. 調度和原子操作
從理論模型上說，多線程環境中，線程之間是獨立、對等關系。一個線程完全可以無視其他線程的存在。但實際多線程執行環境中，線程數往往遠大于CPU數量。 為了共享和分配資源，線程必需遵守一定規則，進行必要的協調。操作系統中具體的規則執行者是調度程序。因此，要想掌握多線程編程，就必需了解線程調度的基 本概念和特點。在此基礎上，能了解為什么需要在線程之間進行協調，進而才能透徹理解如何協調多線程的并發執行。
現代操作系統中，存在許多不同的線程調度模型，這些調度模型的基本共同點就是線程執行順序具有隨機性和不確定性。調度模型既無法事先知道某一時刻會存在多 少線程，也無法知道哪個線程會正在運行，甚至也不知道某個線程確切會到什么時刻執行結束，更不用說預先安排在特定時刻執行特定線程的特定語句。
前面提到，控制流就是語句，指令或函數調用的順序序列。反映到時間軸上，就是一系列離散分布的執行序列。線程調度作用于多個控制流的客觀效果，就是多個控 制流按同一條時間軸排列時，是一個近似隨機的執行序列；按每個控制流各自的時間軸排列時，則是一個具有先后順序的執行序列。
在具體的程序執行上，這個特點就表現為線程A的一個語句執行完畢后，緊接著執行的可能是另一個線程B的一個語句。甚至可能是線程A的一個語句還沒有執行完畢，就接著執行線程B的語句。
對于后一點不用感到奇怪。因為操作系統中，調度程序作為系統底層實現的一部分，參與調度的操作指令可能比高級編程語言的基本語句要底層的多。同一個調度程 序，其調度的眾多線程可能由多種高級語言編寫，所以調度程序基本上不可能以某種高級編程語言的單條語句為單位安排執行序列。
通常，一條高級語言語句可能對應多條匯編指令，而一條匯編指令可能對應多條CPU微碼指令。而一條微碼指令，則可能對應邏輯電路的一個具體電路邏輯。順便 提一下，這也是Verilog, VHDL等高級語言能夠綜合出具體的邏輯電路的基本原理。所不同的是，高級語言編譯器編譯的最終單位是匯編，而硬件描述語言綜合的最終單位是和微碼對應的 電路器件單元（集成電路前端設計的內容。記得以前暑期實習，我窩在學校做的就是這么一個元件庫，做的很不像樣子居然還得到了一定的肯定，多年后想起來還是 汗顏）。
至于系統調度程序具體會定義怎樣的原子操作集合，會以什么粒度的指令為調度基本單位，這就是系統設計者各顯神通的地方了。個人認為，作為高級語言的編程 者，記住什么操作是原子操作意義并不明顯。大多數場合，只要認為，多線程的控制流在同一條時間軸上看來是完全隨機的就可以了。要記住，墨菲定律生效的時 候，看似不可能的事情都能成為現實。
多線程程序設計中，一種樸素的設計思想是將線程具體化為主動對象，主動對象之間通過共享環境（全局資源）來維護當前應用的運行狀態；主動對象間通過一套約 定的互斥和同步規則來實現通信；線程的具體執行時序則由調度程序安排。主動對象之間，應盡量避免一個主動對象直接調用另一個主動對象的內部方法（例如 suspend, stop, exit）直接控制另一個線程的執行步調。主動對象的行為應該完全由其本身控制，和其他主動對象的交互盡量只通過環境以及同步語句來實現。
實際實現中，如果多個主動對象都可以直接調用某個主動對象的stop方法終止其運行，這個程序是非常脆弱的，因為在調度程序之外，不借助同步機制，主觀假 定線程執行時序，人為更改線程執行狀態是非常不明智的。即使這樣的程序可用，對于維護者來說其難度也不亞于維護goto語句。
這也就是前一篇文章所定義線程類中detach, start, stop沒有加鎖的原因，因為并不希望在多個線程中調用同一個線程對象的這些方法。
3.
競爭條件和數據一致性
共享環境是進程全局資源的同義詞。在多線程并發執行中，最常遇到的問題就是共享環境被污染。具體體現就是全局數據被破壞，全局文件內容被破壞 …
。
例如：
有一個64位全局變量long globalVar = 0;主動對象A想把它置為0x000A000B000C000D；假設這個操作分兩步執行，先將高32位置為000A000B，再把低32位置為
000C000D。但不巧的是，對象A剛剛將高位置位，調度程序安排另一主動對象B執行。這時，全局變量globalVar內部的值是一個非法值，它是無 意義的。在B拿著這個值做進一步處理時，得出的也是錯誤結果。這時，稱為數據的一致性被破壞。
線程調度的不確定性也可能導致程序執行結果的不確定性。有時候這種不確定性會導致程序得到錯誤的運行結果。
例如：
為了對Thread類所生成的對象總數計數，定義一個全局變量Unsigned int counter = 0; 在Thread類的構造函數中，執行++counter。現在假設有2個Thread對象objA和objB并發運行，考慮如下兩個場景：
Scenario1.操作序列如下：
1. counter = 0;
2. objA將counter值0從內存讀入寄存器A；
3. objA將寄存器A值加1；
4. objA將寄存器A值寫回內存，counter值為1；
5. objB將counter值1從內存讀入寄存器B；
6. objB將寄存器B值加1；
7. objA將寄存器B值寫回內存，counter值為2；
8. 最終counter值為2。
Scenario2.操作序列如下：
1. counter = 0;
2. objA將counter值0從內存讀入寄存器A；
3. objB將counter值0從內存讀入寄存器B
4. objA將寄存器A值加1；
5. objB將寄存器B值加1；
6. objA將寄存器A值寫回內存，counter值為1；
7. objA將寄存器B值寫回內存，counter值為1；
8. 最終counter值為1。
場景1的結果是設計的本意，場景2的結果對我們而言是錯誤的。
一個線程的執行結果正確性取決于于其他線程執行時序的條件，稱為競爭條件。中文這樣翻譯不倫不類，但從英文字面理解非常容易。Race一般是計時賽，某位選手跑得快一點，競賽結果就有變化，最終結果由race condition決定，還是非常形象的。
4. 為何需要互斥
線程間協作問題，通常可以歸為互斥和同步兩類。其中互斥又主要解決兩類問題：維護數據一致性、避免競爭條件的出現。
解決一致性問題，通俗說就是，修改數據的線程通告其他線程“我正在修改你要訪問的對象X，操作過程中不能保證這個數據的有效性，請不要使用此對象”。
避免競爭條件，通俗說就是，某線程通告其他線程“我將要進行涉及某對象的一系列操作A，在我未完成這一系列操作之前，如果有人和我同時執行涉及此對象的操作序列B（B也可能就是A），將會影響我執行結果的正確性，請不要進行涉及此對象的操作”。
這種操作序列A有時候也被稱為“原子性操作”，因為它不允許操作序列B在時間軸上和它交叉分布，必需保證在時間軸上看來，操作序列A是一個不可分割的整體。（物理學早期，原子也被認為是不可分割的）。
以上冗長的解釋，精簡成一句話，就是“線程間需要互斥執行”。需要互斥的操作對應的代碼也有一個很學術的名稱－“關鍵域（或關鍵區）”。
那么如何實現互斥呢？一個簡單的思路就是，設立一個權威仲裁者，給那些需要互斥執行的線程頒發一個共用的通行證。某個線程想要進入一個關鍵域執行，需要先 申請可以進入該關鍵域的通行證，如果別的線程已經拿走了該通行證，則本線程等待，進入休眠狀態，退出調度。如果本線程的通行證使用完畢，則應該將它歸還給 仲裁者，重新喚醒等待線程，參與調度，競爭此通行證。
比如，下列偽碼中，threadFuncA和threadFuncB就需要申請同一張通行證：
例一：
int globalCounter = 0;
void threadFuncA (通行證類型
* 通行證)
{
獲取通行證；
globalCounter++;
歸還通行證；
}
Void threadFuncB (通行證類型
* 通行證)
{
獲取通行證；
globalCounter *= 2;
歸還通行證；
}
又比如，下列偽碼中，需要為ResourceClass類的對象引用計數器制定一張通行證
。
例二：
class ResourceClass
{
public:
resource & reference()
{
獲取通行證；
++refcounter;
printf(“當前對象被引用了%u次”, refCounter);
釋放通行證；
}
private:
通行證類型
通行證；
unsigned int refCounter;
};
ResourceClass rescObj
Void threadFuncA()
{
rescObj-> reference();
}
Void threadFuncB()
{
rescObj-> reference();
}
最后一個例子，是為ResourceClass類的對象計數器制定一張通行證。
例三：
class ResourceClass
{
public:
ResourceClass ()
{
獲取通行證；
++objcounter;
printf(“當前類創建了%u個對象”, objCounter);
釋放通行證；
}
private:
static通行證類型
通行證；
unsigned int objCounter;
};
Void threadFuncA()
{
ResourceClass * rescObj = new ResourceClass ();
}
Void threadFuncB()
{
ResourceClass * rescObj = new ResourceClass ();
}
這三個例子中，例一是不同函數之間互斥，所以通行證的作用域要對兩個函數都可見。例二是同一個對象的內部函數多次調用之間的互斥，所以只要保證該函數多次 調用時共用的都是當前對象內部同一份通行證即可。例三是同一個類的所有對象在創建時都要互斥，所以必需保證這個類的所有對象構造時共用的時同一份通行證， 從而通行證被聲明為靜態成員。
這里所說的“通行證”在多線程編程中對應一個專門的術語mutex，由“mutual exclusion”拼接而來。為什么不直接用“鎖”的概念呢？因為“鎖”并不能很好的表達互斥的含義。鎖是指一定條件下不允許當前代碼段執行的概念。如 上述例二或例三，不允許多個線程同時執行同一函數，這時說這個函數被鎖定是很形象。但在例一中，A函數被鎖定，為什么B函數不能執行呢？這就較難理解了。
而且經常有人感到疑惑，為什么“加鎖后，被鎖定的關鍵域只能串行執行”。這個其實是指在各自的時間軸上，并行的控制流在經過互斥執行的代碼段時，必需以先 后順序串行執行。在今后的介紹中，mutex的申請，用acquire()操作表示，mutex的歸還，用release()表示。舍棄lock(), unlock()的表示。
為了深入理解，先來看一段使用忙等待實現互斥的代碼，用的是系統內核中使用較多的“spin
lock”互斥方法。
例4．忙等待實現互斥
//聲明為volatile，防止被編譯器優化。
volatile bool dataProcessNotDone = true;
int criticalData = 0;
unsigned threadFuncA( void* para )
{
//如果編譯器不支持volatile關鍵字，
//打開優化選項時，此句可能直接變成死循環。
while (dataProcessNotDone); // spin lock,鎖定的是后續數據
//被鎖定的代碼區
printf("critical data is %d\n", CriticalData);
return 0;
}
unsigned threadFuncB( void* para )
{
sleep(1000);
criticalData++;
dataProcessNotDone = false; //修改互斥變量
return 0;
}
在高級語言中，利用spin lock實現復雜互斥條件非常困難，單單處理競爭條件就令人望而生畏。Spin lock在每次等待解鎖的時間都很短時，具有無需線程切換，無需再調度等獨特優勢。但是在絕大多數應用中，由于互斥等待時間不確定（可能很長），多個線程等待spin lock解鎖的過程中，spinning的行為可能導致系統處于半癱瘓狀態，會嚴重影響程序性能。
除了忙等待之外，很多操作系統或線程庫都提供了互斥原語來實現互斥。如果有可能，應當盡量使用系統提供的接口實現互斥。否則，要考慮編譯器優化中的常量代入，語句執行順序重排，cpu指令序列優化等依賴于具體軟硬件環境的復雜問題（關鍵是這樣的付出沒有太大意義）。
下面根據上述概念，抽象出互斥類的概念，定義如下Mutex類接口
5. 互斥類接口定義
文件mutex.h
#ifndef __MUTEX_H__
#define __MUTEX_H__
// C++標準中以_或__開頭的名字不符合標準，
// 這一特點可以讓這樣定義的宏不會錯誤覆蓋其他名字。
class Mutex
{
public:
Mutex();
~Mutex();
bool acquire (bool block = true);
void release();
private:
//依賴于具體實現，后面再說。
};
#endif
其中，
Mutex::acquire()，獲取互斥量，有阻塞和非阻塞兩種調用方式，阻塞方式，獲取互斥量失敗時線程休眠。非阻塞方式下，獲取失敗時直接返回。
Mutex::release()，釋放互斥量。
6.
示例程序
下面的例子說明了如何實現多線程環境下的Singleton模式。
文件Singleton.h：
#ifndef __SINGLETON_H__
#define __SINGLETON_H__
#include <stdio.h>
#include "thread.h"
#include "mutex.h"
// Dummy class.
class Helper {};
// A wrapper class for Mutex class
// It is exception safe.
class Guard
{
public:
Guard(Mutex & lock):mutex(lock)
{
mutex.acquire();
}
~Guard()
{
mutex.release();
}
private:
Mutex & mutex;
};
// Correct but possibly expensive multithreaded version
class Singleton1
{
public:
Helper * getInstance() {
Guard guard(mutex);
if (helper == NULL)
{
helper = new Helper();
}
return helper;
}
private:
static Mutex mutex;
static Helper * helper;
};
// Broken multithreaded version
// "Double-Checked Locking" idiom
class Singleton2
{
public:
Helper * getInstance() {
if (helper == NULL)
{
Guard guard(mutex);
if (helper == NULL)
{
helper = new Helper();
}
}
return helper;
}
private:
static Mutex mutex;
static Helper * helper;
};
//Thread class for test.
template <typename T>
class TestThread: public Thread
{
public:
TestThread<typename T>(T & resource, Helper *&
res, Helper *&
res2Cmp)
:singleton(resource), instance(res), instance2Cmp(res2Cmp)
{}
protected:
void * run (void *)
{
for (int i=0; i<100000; i++)
{
instance = singleton.getInstance();
if (instance != instance2Cmp
&& instance != NULL
&&instance2Cmp != NULL
)
{
printf("Fail! %p <> %p.\n",
instance, instance2Cmp);
}
}
return NULL;
}
private:
T & singleton;
Helper * & instance;
Helper * & instance2Cmp;
};
#endif
文件main.cpp
#include <stdio.h>
#include "singleton.h"
#define SINGLETON Singleton1
Mutex SINGLETON::mutex;
Helper * SINGLETON::helper = NULL;
int main(int argc, char** argv)
{
Helper * instance1= NULL;
Helper * instance2 = NULL;
SINGLETON singleton;
TestThread<SINGLETON> thread1(singleton, instance1,
instance2);
TestThread<SINGLETON> thread2(singleton, instance2,
instance1);
thread1.start();
thread2.start();
thread1.wait();
thread2.wait();
printf("Finished!\n");
return 0;
}
對此示例程序，說明如下幾點。
1．定義了一個新的Guard類，這樣做的好處是做到異常安全。比如：
try
{
Mutex mutex;
mutex.acquire();
// Some operations
if (errorHappened)
throw Exception();
mutex.release();
}
catch (Exception & e)
{
// Print error message;
}
這段代碼中，拋出異常時，互斥量不能釋放，容易造成死鎖。使用Guard類重寫，
可以實現異常安全：
try
{
Mutex mutex;
Guard guard(mutex);
// Some operations
if (errorHappened)
throw Exception();
}
catch (Exception & e)
{
// Print error message;
}
2.Singleton1的實現可以確保多線程安全。但它是一個低效實現。假如有100次訪問，只有1次會修改instance指針，其余99次都是只讀操作。但是每次訪問都需要進行互斥量的獲取和釋放操作。
取決于系統實現方式，互斥操作可能比整型變量的++操作慢一個數量級。有的實現提供的mutex其實是進程級別的互斥，一次互斥操作，會進入內核態，然后再返回用戶態。而有的線程庫提供的是Process local mutex，要稍快一些。但無論那種實現，代價都較大。
因此，為了改進Singleton1的實現效率，"Double-Checked Locking" idiom被提了出來。其思路是如果instance指針為空再進入互斥操作。由于獲取互斥量過程中，可能別的線程已經將instance指針賦值，所以需要在獲得互斥量所有權之后，再次檢查instance指針值。這就是所謂”double
check”中double的來歷。
Double check的設計很聰明，但可惜無論在C++中還是在Java中，這么做其實都不能保證線程安全。考慮如下序列：
Step1. 線程A檢查instance指針，發現其為空。獲取互斥量成功，運行至語句helper = new Helper();
在打開優化選項時，這個語句在優化后可能變成2步子操作, 而且編譯器自動調整了原語句的執行順序（reordering）：
1) 分配內存，將地址賦值給helper變量。此時helper值非空。
2) 開始初始化此內存。
在運行完子語句1后，線程發生切換，此時內存尚未初始化。
Step2. 線程B檢查instance指針，發現其非空。對instance所指對象進行進一步操作，由于此時對象是初始化還未完成無效數據，程序崩潰。
那么如何實現安全的double check呢？vc2005以后的版本，以及java
1.6以后的版本中，可以通過為helper加上volatile限定符，防止編譯優化時調整指令執行順序。最新的g++對volatile如何處理，沒 有查到相關資料。不過在C++中，只要本地匯編支持memoryBarrier指令，也可以通過在C++代碼中內嵌匯編指令實現線程安全。
在此不再詳細討論。
除此之外，instance類型是否是基本類型，是否多核環境，都對不安全的double check版本運行結果有微妙影響。
3．無論編譯器如何實現，無論硬件環境如何，即使它最慢，也應該盡量使用系統提供的互斥原語。只有它是能夠確保安全的。通過系統接口實現互斥，可以避免考慮編譯優化等復雜情況。一種觀點說volatile可以確保上述double
check有效，但是intel有技術人員專門從硬件的角度批駁了這個說法，他告訴大家，即使編譯器不做這個reordering, 處理器也可能在指令級別做。唯一能確保安全的，還是由系統實現的互斥接口。（照此說法，MS 和
Intel結成wintel聯盟還是必要的，呵呵）雙方說法似乎都一樣權威時，作為程序開發者，很難取舍，因此在此類問題上還是應該適當保守。
4．Mutex, volatile變量，普通變量在使用中的具體效率對比，是一個非常復雜的問題。涉及到內存，緩存，寄存器之間各種情況的同步問題。不過大家可以針對一些簡單的例子，測試一下執行時間上的差異。
7. Unix實現
下面是借助pthread的Mutex類實現。
文件Mutex.h
#ifndef __MUTEX_H__
#define __MUTEX_H__
#include <pthread.h>
class Mutex
{
public:
Mutex();
virtual ~Mutex();
virtual bool acquire (bool block = true);
virtual void release();
private:
pthread_mutex_t handle;
};
#endif
文件 Mutex.cpp
#include "mutex.h"
Mutex::Mutex()
{
pthread_mutex_init(&handle, NULL);
}
Mutex::~Mutex()
{
pthread_mutex_destroy(&handle);
}
bool Mutex::acquire(bool block)
{
if (block)
{
return pthread_mutex_lock(&handle) == 0;
}
else
{
return pthread_mutex_trylock(&handle) == 0;
}
}
void Mutex::release()
{
ReleaseMutex(handle);
}
Windows實現
文件mutex.h
#ifndef __MUTEX_H__
#define __MUTEX_H__
#include <windows.h>
class Mutex
{
public:
Mutex();
virtual ~Mutex();
virtual bool acquire (bool block = true);
virtual void release();
private:
HANDLE handle;
};
#endif
文件mutex.cpp
#include "mutex.h"
Mutex::Mutex()
{
handle = CreateMutex(NULL, false, NULL);
}
Mutex::~Mutex()
{
CloseHandle(handle);
}
bool Mutex::acquire(bool block)
{
//Use caution when calling the wait functions
//and code that directly or indirectly creates windows.
return WaitForSingleObject(handle, block ? INFINITE : 0) ==
WAIT_OBJECT_0;
}
void Mutex::release()
{
ReleaseMutex(handle);
}
小結
本節從控制流的角度進一步介紹了什么是多線程執行中的并發，在此基礎上介紹了主動對象的概念。在多線程編程中，需要考慮線程協作的地方，如果執行順序理不清，為每一個線程畫一條時間軸，標出各自時序，對分析問題往往能有幫助。
本節也介紹了多線程設計的一個基本思想，就是主動對象要具有一定的獨立性，和其他線程的交互盡量只通過進程環境、系統或線程庫提供的同步原語實現。
為什么要互斥，這個基本問題不能透徹理解的話，鎖的概念很容易把自己弄糊涂。互斥的粒度大小也就根本無法談起。
互斥的效率問題，在實際設計中是一個值得考慮的問題。但是程序的正確性問題是一個更重要的問題。不能保證正確性，就不要用。保證正確性到實現高效性的過程，類似于學會走路到能夠飛奔的過程。對于初學者來說，欲速則不達。
為了對互斥量的使用，“通行證”所有權的轉換，以及不同系統中Mutex的實現效率等有一個充分的感性認識，大家請多動手實現才能真正有所收益，最終超越我本人的一己知見。
最后，歡迎提出各種意見，以便這個系列能夠錯誤少一些，解釋準確一些，幫助也更大一些。