https://www.cnblogs.com/Solstice/archive/2010/02/10/dtor_meets_threads.html
陳碩 (giantchen_AT_gmail)
本文 PDF 下載： http://www.cppblog.com/Files/Solstice/dtor_meets_mt.pdf
摘要
編寫線程安全的類不是難事，用同步原語保護內部狀態即可。但是對象的生與死不能由對象自身擁有的互斥器來保護。如何保證即將析構對象 x 的時候，不會有另一個線程正在調用 x 的成員函數？或者說，如何保證在執行 x 的成員函數期間，對象 x 不會在另一個線程被析構？如何避免這種競態條件是 C++ 多線程編程面臨的基本問題，可以借助 boost 的 shared_ptr 和 weak_ptr 完美解決。這也是實現線程安全的 Observer 模式的必備技術。
本文源自我在 2009 年 12 月上海 C++ 技術大會的一場演講《當析構函數遇到多線程》，內容略有增刪。原始 PPT 可從 http://download.csdn.net/source/1982430 下載。
本文讀者應具有 C++ 多線程編程經驗，熟悉互斥器、競態條件等概念，了解智能指針，知道 Observer 設計模式。
目錄
1 多線程下的對象生命期管理................................................................................... 2
一個線程安全的 Counter 示例............................................................................ 3
7 插曲：系統地避免各種指針錯誤........................................................................... 10
與其他面向對象語言不同，C++ 要求程序員自己管理對象的生命期，這在多線程環境下顯得尤為困難。當一個對象能被多個線程同時看到，那么對象的銷毀時機就會變得模糊不清，可能出現多種競態條件：
- 在即將析構一個對象時，從何而知是否有另外的線程正在執行該對象的成員函數？
- 如何保證在執行成員函數期間，對象不會在另一個線程被析構？
- 在調用某個對象的成員函數之前，如何得知這個對象還活着？它的析構函數會不會剛執行到一半？
解決這些 race condition 是 C++ 多線程編程面臨的基本問題。本文試圖以 shared_ptr 一勞永逸地解決這些問題，減輕 C++ 多線程編程的精神負擔。
依據《Java 並發編程實踐》/《Java Concurrency in Practice》一書，一個線程安全的 class 應當滿足三個條件：
- 從多個線程訪問時，其表現出正確的行為
- 無論操作系統如何調度這些線程，無論這些線程的執行順序如何交織
- 調用端代碼無需額外的同步或其他協調動作
依據這個定義，C++ 標准庫里的大多數類都不是線程安全的，無論 std::string 還是 std::vector 或 std::map，因為這些類通常需要在外部加鎖。
與 MutexLock
為了便於后文討論，先約定兩個工具類。我相信每個寫C++ 多線程程序的人都實現過或使用過類似功能的類，代碼從略。
Mutex 封裝臨界區（Critical secion），這是一個簡單的資源類，用 RAII 手法 [CCS:13]封裝互斥器的創建與銷毀。臨界區在 Windows 上是 CRITICAL_SECTION，是可重入的；在 Linux 下是 pthread_mutex_t，默認是不可重入的。Mutex 一般是別的 class 的數據成員。
MutexLock 封裝臨界區的進入和退出，即加鎖和解鎖。MutexLock 一般是個棧上對象，它的作用域剛好等於臨界區域。它的構造函數原型如下。
|
1
|
explicit
MutexLock::MutexLock(Mutex& m);
|
這兩個類都不允許拷貝構造和賦值。
示例
編寫單個的線程安全的 class 不算太難，只需用同步原語保護其內部狀態。例如下面這個簡單的計數器類 Counter：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
class
Counter : boost::noncopyable
{
// copy-ctor and assignment should be private by default for a class.
public
:
Counter(): value_(0) {}
int64_t value()
const
;
int64_t increase();
int64_t decrease();
private
:
int64_t value_;
mutable
Mutex mutex_;
}
int64_t Counter::value()
const
{
MutexLock lock(mutex_);
return
value_;
}
int64_t Counter::increase()
{
MutexLock lock(mutex_);
int64_t ret = value_++;
return
ret;
}
// In a real world, atomic operations are perferred.
// 當然在實際項目中，這個 class 用原子操作更合理，這里用鎖僅僅為了舉例。
|
這個 class 很直白，一看就明白，也容易驗證它是線程安全的。注意到它的 mutex_ 成員是 mutable 的，意味着 const 成員函數如 Counter::value() 也能直接使用 non-const 的 mutex_。
盡管這個 Counter 本身毫無疑問是線程安全的，但如果 Counter 是動態創建的並透過指針來訪問，前面提到的對象銷毀的 race condition 仍然存在。
對象構造要做到線程安全，惟一的要求是在構造期間不要泄露 this 指針，即
- 不要在構造函數中注冊任何回調
- 也不要在構造函數中把 this 傳給跨線程的對象
- 即便在構造函數的最后一行也不行
之所以這樣規定，是因為在構造函數執行期間對象還沒有完成初始化，如果 this 被泄露 (escape) 給了其他對象（其自身創建的子對象除外），那么別的線程有可能訪問這個半成品對象，這會造成難以預料的后果。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
// 不要這么做 Don't do this.
class
Foo :
public
Observer
{
public
:
Foo(Observable* s) {
s->
register
(
this
);
// 錯誤
}
virtual
void
update();
};
// 要這么做 Do this.
class
Foo :
public
Observer
{
// ...
void
observe(Observable* s) {
// 另外定義一個函數，在構造之后執行
s->
register
(
this
);
}
};
Foo* pFoo =
new
Foo;
Observable* s = getIt();
pFoo->observe(s);
// 二段式構造
|
這也說明，二段式構造——即構造函數+initialize()——有時會是好辦法，這雖然不符合 C++ 教條，但是多線程下別無選擇。另外，既然允許二段式構造，那么構造函數不必主動拋異常，調用端靠 initialize() 的返回值來判斷對象是否構造成功，這能簡化錯誤處理。
即使構造函數的最后一行也不要泄露 this，因為 Foo 有可能是個基類，基類先於派生類構造，執行完 Foo::Foo() 的最后一行代碼會繼續執行派生類的構造函數，這時 most-derived class 的對象還處於構造中，仍然不安全。
相對來說，對象的構造做到線程安全還是比較容易的，畢竟曝光少，回頭率為 0。而析構的線程安全就不那么簡單，這也是本文關注的焦點。
對象析構，這在單線程里不會成為問題，最多需要注意避免空懸指針（和野指針）。而在多線程程序中，存在了太多的競態條件。對一般成員函數而言，做到線程安全的辦法是讓它們順次執行，而不要並發執行，也就是讓每個函數的臨界區不重疊。這是顯而易見的，不過有一個隱含條件或許不是每個人都能立刻想到：函數用來保護臨界區的互斥器本身必須是有效的。而析構函數破壞了這一假設，它會把互斥器銷毀掉。悲劇啊！
不是辦法
Mutex 只能保證函數一個接一個地執行，考慮下面的代碼，它試圖用互斥鎖來保護析構函數：
| Foo::~Foo() { MutexLock lock(mutex_); // free internal state (1) } |
void Foo::update() { MutexLock lock(mutex_); // (2) // make use of internal state } |
| extern Foo* x; // visible by all threads |
|
| // thread A delete x; x = NULL; // helpless |
// thread B if (x) { x->update(); } |
有 A 和 B 兩個線程，線程 A 即將銷毀對象 x，而線程 B 正准備調用 x->update()。盡管線程 A 在銷毀對象之后把指針置為了 NULL，盡管線程 B 在調用 x 的成員函數之前檢查了指針 x 的值，還是無法避免一種 race condition：
1. 線程 A 執行到了析構函數的 (1)處，已經持有了互斥鎖，即將繼續往下執行
2. 線程 B 通過了 if (x) 檢測，阻塞在 (2) 處
接下來會發生什么，只有天曉得。因為析構函數會把 mutex_ 銷毀，那么 (2) 處有可能永遠阻塞下去，有可能進入“臨界區”然后 core dump，或者發生其他更糟糕的情況。
這個例子至少說明 delete 對象之后把指針置為 NULL 根本沒用，如果一個程序要靠這個來防止二次釋放，說明代碼邏輯出了問題。
前面的例子說明，作為 class 數據成員的 Mutex 只能用於同步本 class 的其他數據成員的讀和寫，它不能保護安全地析構。因為成員 mutex 的生命期最多與對象一樣長，而析構動作可說是發生在對象身故之后（或者身亡之時）。另外，對於基類對象，那么調用到基類析構函數的時候，派生類對象的那部分已經析構了，那么基類對象擁有的 mutex 不能保護整個析構過程。再說，析構過程本來也不需要保護，因為只有別的線程都訪問不到這個對象時，析構才是安全的，否則會有第 1 節談到的競態條件發生。
另外如果要同時讀寫本 class 的兩個對象，有潛在的死鎖可能，見 PPT 第 12 頁的 swap() 和 operator=()。
有多難？
一個動態創建的對象是否還活着，光看指針（引用也一樣）是看不出來的。指針就是指向了一塊內存，這塊內存上的對象如果已經銷毀，那么就根本不能訪問 [CCS:99]（就像 free 之后的地址不能訪問一樣），既然不能訪問又如何知道對象的狀態呢？換句話說，判斷一個指針是不是野指針沒有高效的辦法。（萬一原址又創建了一個新的對象呢？再萬一這個新的對象的類型異於老的對象呢？）
在面向對象程序設計中，對象的關系主要有三種：composition, aggregation 和 association。Composition（組合/復合）關系在多線程里不會遇到什么麻煩，因為對象 x 的生命期由其惟一的擁有者 owner 控制，owner 析構的時候會把 x 也析構掉。從形式上看，x 是 owner 的直接數據成員，或者 scoped_ptr 成員，或者 owner 持有的容器的元素。
后兩種關系在 C++ 里比較難辦，處理不好就會造成內存泄漏或重復釋放。Association（關聯/聯系）是一種很寬泛的關系，它表示一個對象 a 用到了另一個對象 b，調用了后者的成員函數。從代碼形式上看，a 持有 b 的指針（或引用），但是 b 的生命期不由 a 單獨控制。Aggregation（聚合）關系從形式上看與 association 相同，除了 a 和 b 有邏輯上的整體與部分關系。如果 b 是動態創建的並在整個程序結束前有可能被釋放，那么就會出現第 1 節談到的競態條件。
那么似乎一個簡單的解決辦法是：只創建不銷毀。程序使用一個對象池來暫存用過的對象，下次申請新對象時，如果對象池里有存貨，就重復利用現有的對象，否則就新建一個。對象用完了，不是直接釋放掉，而是放回池子里。這個辦法當然有其自身的很多缺點，但至少能避免訪問失效對象的情況發生。
這種山寨辦法的問題有：
- 對象池的線程安全，如何安全地完整地把對象放回池子里，不會出現“部分放回”的競態？（線程 A 認為對象 x 已經放回了，線程 B 認為對象 x 還活着）
- thread contention，這個集中化的對象池會不會把多線程並發的操作串行化？
- 如果共享對象的類型不止一種，那么是重復實現對象池還是使用類模板？
- 會不會造成內存泄露與分片？因為對象池占用的內存只增不減，而且多個對象池不能共享內存（想想為何）。
回到正題上來，如果對象 x 注冊了任何非靜態成員函數回調，那么必然在某處持有了指向 x 的指針，這就暴露在了 race condition 之下。
一個典型的場景是 Observer 模式。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
class
Observer
{
public
:
virtual
~Observer() { }
virtual
void
update() = 0;
};
class
Observable
{
public
:
void
register
(Observer* x);
void
unregister(Observer* x);
void
notifyObservers() {
foreach Observer* x {
// 這行是偽代碼
x->update();
// (3)
}
}
// ...
}
|
當 Observable 通知每一個 Observer 時 (3)，它從何得知 Observer 對象 x 還活着？
要不在 Observer 的析構函數里解注冊 (unregister)？恐難奏效。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
struct
Observer
{
virtual
~Observer() { }
virtual
void
update() = 0;
void
observe(Observable* s) {
s->
register
(
this
);
subject_ = s;
}
virtual
~Observer() {
// (4)
subject_->unregister(
this
);
}
Observable* subject_;
};
|
我們試着讓 Observer 的析構函數去 unregister(this)，這里有兩個 race conditions。其一：(4)處如何得知 subject_ 還活着？其二：就算 subject_ 指向某個永久存在的對象，那么還是險象環生：
- 線程 A 執行到 (4)處，還沒有來得及 unregister 本對象
- 線程 B 執行到 (3) 處，x 正好指向是 (4)處正在析構的對象
那么悲劇又發生了，既然 x 所指的 Observer 對象正在析構，調用它的任何非靜態成員函數都是不安全的，何況是虛函數（C++ 標准對在構造函數和析構函數中調用虛函數的行為有明確的規定，但是沒有考慮並發調用的情況）。更糟糕的是，Observer 是個基類，執行到 (4)處時，派生類對象已經析構掉了，這時候整個對象處於將死未死的狀態，core dump 恐怕是最幸運的結果。
這些 race condition 似乎可以通過加鎖來解決，但在哪兒加鎖，誰持有這些互斥鎖，又似乎不是那么顯而易見的。要是有什么活着的對象能幫幫我們就好了，它提供一個 isAlive() 之類的程序函數，告訴我們那個對象還在不在。可惜指針和引用都不是對象，它們是內建類型。
指向對象的原始指針 (raw pointer) 是壞的，尤其當暴露給別的線程時。Observable 應當保存的不是原始的 Observer*，而是別的什么東西，能分別 Observer 對象是否存活。類似地，如果 Observer 要在析構函數里解注冊（這雖然不能解決前面提到的 race condition，但是在析構函數里打掃戰場還是應該的），那么 subject_ 的類型也不能是原始的 Observable*。
有經驗的 C++ 程序員或許會想到用智能指針，沒錯，這是正道，但也沒那么簡單，有些關竅需要注意。這兩處直接使用 shared_ptr 是不行的，會形成循環引用，直接造成資源泄漏。別着急，后文會一一講到。
有兩個指針 p1 和 p2，指向堆上的同一個對象 Object，p1 和 p2 位於不同的線程中（左圖）。假設線程 A 透過 p1 指針將對象銷毀了（盡管把 p1 置為了 NULL），那么 p2 就成了空懸指針（右圖）。這是一種典型的 C/C++ 內存錯誤。
要想安全地銷毀對象，最好讓在別人（線程）都看不到的情況下，偷偷地做。
一個解決空懸指針的辦法是，引入一層間接性，讓 p1 和 p2 所指的對象永久有效。比如下圖的 proxy 對象，這個對象，持有一個指向 Object 的指針。（從 C 語言的角度，p1 和 p2 都是二級指針。）
當銷毀 Object 之后，proxy 對象繼續存在，其值變為 0。而 p2 也沒有變成空懸指針，它可以通過查看 proxy 的內容來判斷 Object 是否還活着。要線程安全地釋放 Object 也不是那么容易，race condition 依舊存在。比如 p2 看第一眼的時候 proxy 不是零，正准備去調用 Object 的成員函數，期間對象已經被 p1 給銷毀了。
問題在於，何時釋放 proxy 指針呢？
為了安全地釋放 proxy，我們可以引入引用計數，再把 p1 和 p2 都從指針變成對象 sp1 和 sp2。proxy 現在有兩個成員，指針和計數器。
1. 一開始，有兩個引用，計數值為 2，
2. sp1 析構了，引用計數的值減為 1，
3. sp2 也析構了，引用計數的值為 0，可以安全地銷毀 proxy 和 Object 了。
慢着！這不就是引用計數型智能指針嗎？
引入另外一層間接性，another layer of indirection，用對象來管理共享資源（如果把 Object 看作資源的話），亦即 handle/body 手法 (idiom)。當然，編寫線程安全、高效的引用計數 handle 的難度非凡，作為一名謙卑的程序員，用現成的庫就行。
萬幸，C++ 的 tr1 標准庫里提供了一對神兵利器，可助我們完美解決這個頭疼的問題。
shared_ptr 是引用計數型智能指針，在 boost 和 std::tr1 里都有提供，現代主流的 C++ 編譯器都能很好地支持。shared_ptr<T> 是一個類模板 (class template)，它只有一個類型參數，使用起來很方便。引用計數的是自動化資源管理的常用手法，當引用計數降為 0 時，對象（資源）即被銷毀。weak_ptr 也是一個引用計數型智能指針，但是它不增加引用次數，即弱 (weak) 引用。
shared_ptr 的基本用法和語意請參考手冊或教程，本文從略，這里談幾個關鍵點。
- shared_ptr 控制對象的生命期。shared_ptr 是強引用（想象成用鐵絲綁住堆上的對象），只要有一個指向 x 對象的 shared_ptr 存在，該 x 對象就不會析構。當指向對象 x 的最后一個 shared_ptr 析構或 reset 的時候，x 保證會被銷毀。
- weak_ptr 不控制對象的生命期，但是它知道對象是否還活着（想象成用棉線輕輕拴住堆上的對象）。如果對象還活着，那么它可以提升 (promote) 為有效的 shared_ptr；如果對象已經死了，提升會失敗，返回一個空的 shared_ptr。“提升”行為是線程安全的。
- shared_ptr/weak_ptr 的“計數”在主流平台上是原子操作，沒有用鎖，性能不俗。
- shared_ptr/weak_ptr 的線程安全級別與 string 等 STL 容器一樣，后面還會講。
我同意孟岩說的“大部分用 C 寫的上規模的軟件都存在一些內存方面的錯誤，需要花費大量的精力和時間把產品穩定下來。”內存方面的問題在 C++ 里很容易解決，我第一次也是最后一次見到別人的代碼里有內存泄漏是在 2004 年實習那會兒，自己寫的C++ 程序從來沒有出現過內存方面的問題。
C++ 里可能出現的內存問題大致有這么幾個方面：
- 緩沖區溢出
- 空懸指針/野指針
- 重復釋放
- 內存泄漏
- 不配對的 new[]/delete
- 內存碎片
正確使用智能指針能很輕易地解決前面 5 個問題，解決第 6 個問題需要別的思路，我會另文探討。
- 緩沖區溢出 ⇒ 用 vector/string 或自己編寫 Buffer 類來管理緩沖區，自動記住用緩沖區的長度，並通過成員函數而不是裸指針來修改緩沖區。
- 空懸指針/野指針 ⇒ 用 shared_ptr/weak_ptr，這正是本文的主題
- 重復釋放 ⇒ 用 scoped_ptr，只在對象析構的時候釋放一次
- 內存泄漏 ⇒ 用 scoped_ptr，對象析構的時候自動釋放內存
- 不配對的 new[]/delete ⇒ 把 new[] 統統替換為 vector/scoped_array
正確使用上面提到的這幾種智能指針並不難，其難度大概比學習使用 vector/list 這些標准庫組件還要小，與 string 差不多，只要花一周的時間去適應它，就能信手拈來。我認為，在現代的 C++ 程序中一般不會出現 delete 語句，資源（包括復雜對象本身）都是通過對象（智能指針或容器）來管理的，不需要程序員還為此操心。
需要注意一點：scoped_ptr/shared_ptr/weak_ptr 都是值語意，要么是棧上對象，或是其他對象的直接數據成員，或是標准庫容器里的元素。幾乎不會有下面這種用法：
shared_ptr<Foo>* pFoo = new shared_ptr<Foo>(new Foo); // WRONG semantic
還要注意，如果這幾種智能指針是對象 x 的數據成員，而它的模板參數 T 是個 incomplete 類型，那么 x 的析構函數不能是默認的或內聯的，必須在 .cpp 文件里邊顯式定義，否則會有編譯錯或運行錯。（原因請見 boost::checked_delete）
上
既然透過 weak_ptr 能探查對象的生死，那么 Observer 模式的競態條件就很容易解決，只要讓 Observable 保存 weak_ptr<Observer> 即可：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
class
Observable
// not 100% thread safe!
{
public
:
void
register
(weak_ptr<Observer> x);
void
unregister(weak_ptr<Observer> x);
// 可用 std::remove/vector::erase 實現
void
notifyObservers()
{
MutexLock lock(mutex_);
Iterator it = observers_.begin();
while
(it != observers_.end()) {
shared_ptr<Observer> obj(it->lock());
// 嘗試提升，這一步是線程安全的
if
(obj) {
// 提升成功，現在引用計數值至少為 2 （想想為什么？）
obj->update();
// 沒有競態條件，因為 obj 在棧上，對象不可能在本作用域內銷毀
++it;
}
else
{
// 對象已經銷毀，從容器中拿掉 weak_ptr
it = observers_.erase(it);
}
}
}
private
:
std::vector<weak_ptr<Observer> > observers_;
// (5)
mutable
Mutex mutex_;
};
|
就這么簡單。前文代碼 (3) 處的競態條件已經彌補了。
把 Observer* 替換為 weak_ptr<Observer> 部分解決了 Observer 模式的線程安全，但還有幾個疑點：
不靈活，強制要求 Observer 必須以 shared_ptr 來管理；
不是完全線程安全，Observer 的析構函數會調用 subject_->unregister(this)，萬一 subject_ 已經不復存在了呢？為了解決它，又要求 Observable 本身是用 shared_ptr 管理的，並且 subject_ 是個 weak_ptr<Observable>；
線程爭用 (thread contention)，即 Observable 的三個成員函數都用了互斥器來同步，這會造成 register 和 unregister 等待 notifyObservers，而后者的執行時間是無上限的，因為它同步回調了用戶提供的 update() 函數。我們希望 register 和 unregister 的執行時間不會超過某個固定的上限，以免即便殃及無辜群眾。
死鎖，萬一 update() 虛函數中調用了 (un)register 呢？如果 mutex_ 是不可重入的，那么會死鎖；如果 mutex_ 是可重入的，程序會面臨迭代器失效（core dump 是最好的結果），因為 vector observers_ 在遍歷期間被無意識地修改了。這個問題乍看起來似乎沒有解決辦法，除非在文檔里做要求。（一個辦法是：用可重入的 mutex_，把容器換為 std::list，並把 ++it 往前挪一行。）
這些問題留到本文附錄中去探討，每個問題都是能解決的。
我個人傾向於使用不可重入的 Mutex，例如 pthreads 默認提供的那個，因為“要求 Mutex 可重入”本身往往意味着設計上出了問題。Java 的 intrinsic lock 是可重入的，因為要允許 synchronized 方法相互調用，我覺得這也是無奈之舉。
思考：如果把 (5) 處改為 vector<shared_ptr<Observer> > observers_;，會有什么后果？
的線程安全
雖然我們借 shared_ptr 來實現線程安全的對象釋放，但是 shared_ptr 本身不是 100% 線程安全的。它的引用計數本身是安全且無鎖的，但對象的讀寫則不是，因為 shared_ptr 有兩個數據成員，讀寫操作不能原子化。
根據文檔，shared_ptr 的線程安全級別和內建類型、標准庫容器、string 一樣，即：
- 一個 shared_ptr 實體可被多個線程同時讀取；
- 兩個的 shared_ptr 實體可以被兩個線程同時寫入，“析構”算寫操作；
- 如果要從多個線程讀寫同一個 shared_ptr 對象，那么需要加鎖。
請注意，這是 shared_ptr 對象本身的線程安全級別，不是它管理的對象的線程安全級別。
要在多個線程中同時訪問同一個 shared_ptr，正確的做法是：
|
1
2
3
|
shared_ptr<Foo> globalPtr;
Mutex mutex;
// No need for ReaderWriterLock
void
doit(
const
shared_ptr<Foo>& pFoo);
|
globalPtr 能被多個線程看到，那么它的讀寫需要加鎖。注意我們不必用讀寫鎖，而只用最簡單的互斥鎖，這是為了性能考慮，因為臨界區非常小，用互斥鎖也不會阻塞並發讀。
|
1
2
3
4
5
6
7
8
9
10
11
|
void
read()
{
shared_ptr<Foo> ptr;
{
MutexLock lock(mutex);
ptr = globalPtr;
// read globalPtr
}
// use ptr since here
doit(ptr);
}
|
寫入的時候也要加鎖：
|
1
2
3
4
5
6
7
8
9
10
11
|
void
write()
{
shared_ptr<Foo> newptr(
new
Foo);
{
MutexLock lock(mutex);
globalPtr = newptr;
// write to globalPtr
}
// use newptr since here
doit(newptr);
}
|
注意到 read() 和 write() 在臨界區之外都沒有再訪問 globalPtr，而是用了一個指向同一 Foo 對象的棧上 shared_ptr local copy。下面會談到，只要有這樣的 local copy 存在，shared_ptr 作為函數參數傳遞時不必復制，用 reference to const 即可。
技術與陷阱
意外延長對象的生命期。shared_ptr 是強引用（鐵絲綁的），只要有一個指向 x 對象的 shared_ptr 存在，該對象就不會析構。而 shared_ptr 又是允許拷貝構造和賦值的（否則引用計數就無意義了），如果不小心遺留了一個拷貝，那么對象就永世長存了。例如前面提到如果把 (5) 處 observers_ 的類型改為 vector<shared_ptr<Observer> >，那么除非手動調用 unregister，否則 Observer 對象永遠不會析構。即便它的析構函數會調用 unregister，但是不去 unregister 就不會調用析構函數，這變成了雞與蛋的問題。這也是 Java 內存泄露的常見原因。
另外一個出錯的可能是 boost::bind，因為 boost::bind 會把參數拷貝一份，如果參數是個 shared_ptr，那么對象的生命期就不會短於 boost::function 對象：
|
1
2
3
4
5
6
7
8
|
class
Foo
{
void
doit();
};
boost::function<
void
()> func;
shared_ptr<Foo> pFoo(
new
Foo);
func = bind(&Foo::doit, pFoo);
// long life foo
|
這里 func 對象持有了 shared_ptr<Foo> 的一份拷貝，有可能會不經意間延長倒數第二行創建的 Foo 對象的生命期。
函數參數。因為要修改引用計數（而且拷貝的時候通常要加鎖），shared_ptr 的拷貝開銷比拷貝原始指針要高，但是需要拷貝的時候並不多。多數情況下它可以以 reference to const 方式傳遞，一個線程只需要在最外層函數有一個實體對象，之后都可以用 reference to const 來使用這個 shared_ptr。例如有幾個個函數都要用到 Foo 對象：
|
1
2
3
4
5
6
7
8
|
void
save(
const
shared_ptr<Foo>& pFoo);
void
validateAccount(
const
Foo& foo);
bool
validate(
const
shared_ptr<Foo>& pFoo)
{
// ...
validateAccount(*pFoo);
// ...
}
|
那么在通常情況下，
|
1
2
3
4
5
6
7
|
void
onMessage(
const
string& buf)
{
shared_ptr<Foo> pFoo(
new
Foo(buf));
// 只要在最外層持有一個實體，安全不成問題
if
(validate(pFoo)) {
save(pFoo);
}
}
|
遵照這個規則，基本上不會遇到反復拷貝 shared_ptr 導致的性能問題。另外由於 pFoo 是棧上對象，不可能被別的線程看到，那么讀取始終是線程安全的。
析構動作在創建時被捕獲。這是一個非常有用的特性，這意味着：
- 虛析構不再是必須的。
- shared_ptr<void> 可以持有任何對象，而且能安全地釋放
- shared_ptr 對象可以安全地跨越模塊邊界，比如從 DLL 里返回，而不會造成從模塊 A 分配的內存在模塊 B 里被釋放這種錯誤。
- 二進制兼容性，即便 Foo 對象的大小變了，那么舊的客戶代碼任然可以使用新的動態庫，而無需重新編譯（這要求 Foo 的頭文件中不出現訪問對象的成員的 inline函數）。
- 析構動作可以定制。
這個特性的實現比較巧妙，因為 shared_ptr<T> 只有一個模板參數，而“析構行為”可以是函數指針，仿函數 (functor) 或者其他什么東西。這是泛型編程和面向對象編程的一次完美結合。有興趣的同學可以參考 Scott Meyers 的文章。
這個技術在后面的對象池中還會用到。
析構所在的線程。對象的析構是同步的，當最后一個指向 x 的 shared_ptr 離開其作用域的時候，x 會同時在同一個線程析構。這個線程不一定是對象誕生的線程。這個特性是把雙刃劍：如果對象的析構比較耗時，那么可能會拖慢關鍵線程的速度（如果最后一個 shared_ptr 引發的析構發生在關鍵線程）；同時，我們可以用一個單獨的線程來專門做析構，通過一個 BlockingQueue<shared_ptr<void> > 把對象的析構都轉移到那個專用線程，從而解放關鍵線程。
現成的 RAII handle。我認為 RAII （資源獲取即初始化）是 C++ 語言區別與其他所有編程語言的最重要的手法，一個不懂 RAII 的 C++ 程序員不是一個合格的 C++ 程序員。原來的 C++ 教條是“new 和 delete 要配對，new 了之后要記着 delete”，如果使用 RAII，要改成“每一個明確的資源配置動作（例如 new）都應該在單一語句中執行，並在該語句中立刻將配置獲得的資源交給 handle 對象（如 shared_ptr），程序中一般不出現 delete”（出處見腳注 1）。shared_ptr 是管理共享資源的利器，需要注意避免循環引用，通常的做法是 owner 持有指向 A 的 shared_ptr，A 持有指向 owner 的 weak_ptr。
假設有 Stock 類，代表一只股票的價格。每一只股票有一個惟一的字符串標識，比如 Google 的 key 是 "NASDAQ:GOOG"，IBM 是 "NYSE:IBM"。Stock 對象是個主動對象，它能不斷獲取新價格。為了節省系統資源，同一個程序里邊每一只出現的股票只有一個 Stock 對象，如果多處用到同一只股票，那么 Stock 對象應該被共享。如果某一只股票沒有再在任何地方用到，其對應的 Stock 對象應該析構，以釋放資源，這隱含了“引用計數”。
為了達到上述要求，我們可以設計一個對象池 StockFactory。它的接口很簡單，根據 key 返回 Stock 對象。我們已經知道，在多線程程序中，既然對象可能被銷毀，那么返回 shared_ptr 是合理的。自然地，我們寫出如下代碼。（可惜是錯的）
|
1
2
3
4
5
6
7
8
9
10
|
class
StockFactory : boost::noncopyable
{
// questionable code
public
:
shared_ptr<Stock> get(
const
string& key);
private
:
std::map<string, shared_ptr<Stock> > stocks_;
mutable
Mutex mutex_;
};
|
get() 的邏輯很簡單，如果在 stocks_ 里找到了 key，就返回 stocks_[key]；否則新建一個 Stock，並存入 stocks_[key]。
細心的讀者或許已經發現這里有一個問題，Stock 對象永遠不會被銷毀，因為 map 里存的是 shared_ptr，始終有鐵絲綁着。那么或許應該仿照前面 Observable 那樣存一個 weak_ptr？比如
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
class
StockFactory : boost::noncopyable
{
public
:
shared_ptr<Stock> get(
const
string& key)
{
shared_ptr<Stock> pStock;
MutexLock lock(mutex_);
weak_ptr<Stock>& wkStock = stocks_[key];
// 如果 key 不存在，會默認構造一個
pStock = wkStock.lock();
// 嘗試把棉線提升為鐵絲
if
(!pStock) {
pStock.reset(
new
Stock(key));
wkStock = pStock;
// 這里更新了 stocks_[key]，注意 wkStock 是個引用
}
return
pStock;
}
private
:
std::map<string, weak_ptr<Stock> > stocks_;
mutable
Mutex mutex_;
};
|
這么做固然 Stock 對象是銷毀了，但是程序里卻出現了輕微的內存泄漏，為什么？
因為 stocks_ 的大小只增不減，stocks_.size() 是曾經存活過的 Stock 對象的總數，即便活的 Stock 對象數目降為 0。或許有人認為這不算泄漏，因為內存並不是徹底遺失不能訪問了，而是被某個標准庫容器占用了。我認為這也算內存泄漏，畢竟是戰場沒有打掃干凈。
其實，考慮到世界上的股票數目是有限的，這個內存不會一直泄漏下去，大不了把每只股票的對象都創建一遍，估計泄漏的內存也只有幾兆。如果這是一個其他類型的對象池，對象的 key 的集合不是封閉的，內存會一直泄漏下去。
解決的辦法是，利用 shared_ptr 的定制析構功能。shared_ptr 的構造函數可以有一個額外的模板類型參數，傳入一個函數指針或仿函數 d，在析構對象時執行 d(p)。shared_ptr 這么設計並不是多余的，因為反正要在創建對象時捕獲釋放動作，始終需要一個 bridge。
|
1
2
3
|
template
<
class
Y,
class
D> shared_ptr::shared_ptr(Y* p, D d);
template
<
class
Y,
class
D>
void
shared_ptr::reset(Y* p, D d);
|
那么我們可以利用這一點，在析構 Stock 對象的同時清理 stocks_。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
class
StockFactory : boost::noncopyable
{
// in get(), change
// pStock.reset(new Stock(key));
// to
// pStock.reset(new Stock(key),
// boost::bind(&StockFactory::deleteStock, this, _1)); (6)
private
:
void
deleteStock(Stock* stock)
{
if
(stock) {
MutexLock lock(mutex_);
stocks_.erase(stock->key());
}
delete
stock;
// sorry, I lied
}
// assuming FooCache lives longer than all Foo's ...
// ...
|
這里我們向 shared_ptr<Stock>::reset() 傳遞了第二個參數，一個 boost::function，讓它在析構 Stock* p 時調用本 StockFactory 對象的 deleteStock 成員函數。
警惕的讀者可能已經發現問題，那就是我們把一個原始的 StockFactory this 指針保存在了 boost::function 里 (6)，這會有線程安全問題。如果這個 StockFactory 先於 Stock 對象析構，那么會 core dump。正如 Observer 在析構函數里去調用 Observable::unregister()，而那時 Observable 對象可能已經不存在了。
當然這也是能解決的，用到下一節的技術。
StockFactory::get() 把原始指針 this 保存到了 boost::function 中 (6)，如果 StockFactory 的生命期比 Stock 短，那么 Stock 析構時去回調 StockFactory::deleteStock 就會 core dump。似乎我們應該祭出慣用的 shared_ptr 大法來解決對象生命期問題，但是 StockFactory::get() 本身是個成員函數，如何獲得一個 shared_ptr<StockFactory> 對象呢？
有辦法，用 enable_shared_from_this。這是一個模板基類，繼承它，this 就能變身為 shared_ptr。
|
1
2
3
|
class
StockFactory :
public
boost::enable_shared_from_this<StockFactory>,
boost::noncopyable
{
/* ... */
};
|
為了使用 shared_from_this()，要求 StockFactory 對象必須保存在 shared_ptr 里。
|
1
|
shared_ptr<StockFactory> stockFactory(
new
StockFactory);
|
萬事俱備，可以讓 this 搖身一變，化為 shared_ptr<StockFactory> 了。
|
1
2
3
4
5
6
7
8
9
10
11
|
shared_ptr<Stock> StockFactory::get(
const
string& key)
{
// change
// pStock.reset(new Stock(key),
// boost::bind(&StockFactory::deleteStock, this, _1));
// to
pStock.reset(
new
Stock(key),
boost::bind(&StockFactory::deleteStock,
shared_from_this(),
_1));
// ...
|
這樣一來，boost::function 里保存了一份 shared_ptr<StockFactory>，可以保證調用 StockFactory::deleteStock 的時候那個 StockFactory 對象還活着。
注意一點，shared_from_this() 不能在構造函數里調用，因為在構造 StockFactory 的時候，它還沒有被交給 shared_ptr 接管。
最后一個問題，StockFactory 的生命期似乎被意外延長了。
把 shared_ptr 綁 (boost::bind) 到 boost:function 里，那么回調的時候 StockFactory 對象始終存在，是安全的。這同時也延長了對象的生命期，使之不短於綁得的 boost:function 對象。
有時候我們需要“如果對象還活着，就調用它的成員函數，否則忽略之”的語意，就像 Observable::notifyObservers() 那樣，我稱之為“弱回調”。這也是可以實現的，利用 weak_ptr，我們可以把 weak_ptr 綁到 boost::function 里，這樣對象的生命期就不會被延長。然后在回調的時候先嘗試提升為 shared_ptr，如果提升成功，說明接受回調的對象還健在，那么就執行回調；如果提升失敗，就不必勞神了。
使用這一技術的完整 StockFactory 代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
class
StockFactory :
public
boost::enable_shared_from_this<StockFactory>,
boost::noncopyable
{
public
:
shared_ptr<Stock> get(
const
string& key)
{
shared_ptr<Stock> pStock;
MutexLock lock(mutex_);
weak_ptr<Stock>& wkStock = stocks_[key];
pStock = wkStock.lock();
if
(!pStock) {
pStock.reset(
new
Stock(key),
boost::bind(&StockFactory::weakDeleteCallback,
boost::weak_ptr<StockFactory>(shared_from_this()),
_1));
// 上面必須強制把 shared_from_this() 轉型為 weak_ptr，才不會延長生命期
wkStock = pStock;
}
return
pStock;
}
private
:
static
void
weakDeleteCallback(boost::weak_ptr<StockFactory> wkFactory,
Stock* stock)
{
shared_ptr<StockFactory> factory(wkFactory.lock());
// 嘗試提升
if
(factory) {
// 如果 factory 還在，那就清理 stocks_
factory->removeStock(stock);
}
delete
stock;
// sorry, I lied
}
void
removeStock(Stock* stock)
{
if
(stock) {
MutexLock lock(mutex_);
stocks_.erase(stock->key());
}
}
private
:
std::map<string, weak_ptr<Stock> > stocks_;
mutable
Mutex mutex_;
};
|
兩個簡單的測試：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
void
testLongLifeFactory()
{
shared_ptr<StockFactory> factory(
new
StockFactory);
{
shared_ptr<Stock> stock = factory->get(
"NYSE:IBM"
);
shared_ptr<Stock> stock2 = factory->get(
"NYSE:IBM"
);
assert
(stock == stock2);
// stock destructs here
}
// factory destructs here
}
void
testShortLifeFactory()
{
shared_ptr<Stock> stock;
{
shared_ptr<StockFactory> factory(
new
StockFactory);
stock = factory->get(
"NYSE:IBM"
);
shared_ptr<Stock> stock2 = factory->get(
"NYSE:IBM"
);
assert
(stock == stock2);
// factory destructs here
}
// stock destructs here
}
|
這下完美了，無論 Stock 和 StockFactory 誰先掛掉都不會影響程序的正確運行。
當然，通常 Factory 對象是個 singleton，在程序正常運行期間不會銷毀，這里只是為了展示弱回調技術，這個技術在事件通知中非常有用。
除了使用 shared_ptr/weak_ptr，要想在 C++ 里做到線程安全的對象回調與析構，可能的辦法有：
- 用一個全局的 facade 來代理 Foo 類型對象訪問，所有的 Foo 對象回調和析構都通過這個 facade 來做，也就是把指針替換為 objId/handle。這樣理論上能避免 race condition，但是代價很大。因為要想把這個 facade 做成線程安全，那么必然要用互斥鎖。這樣一來，從兩個線程訪問兩個不同的 Foo 對象也會用到同一個鎖，讓本來能夠並行執行的函數變成了串行執行，沒能發揮多核的優勢。當然，可以像 Java 的 ConcurrentHashMap 那樣用多個 buckets，每個 bucket 分別加鎖，以降低 contention。
- 第 4 節提到的“只創建不銷毀”手法，實屬無奈之舉。
- 自己編寫引用計數的智能指針。本質上是重新發明輪子，把 shared_ptr 實現一遍。正確實現線程安全的引用計數智能指針不是一件容易的事情，而高效的實現就更加困難。既然 shared_ptr 已經提供了完整的解決方案，那么似乎沒有理由抗拒它。
- 將來在 C++ 0x 里有 unique_ptr，能避免引用計數的開銷，或許能在某些場合替換shared_ptr。
有垃圾回收就好辦。Google 的 Go 語言教程明確指出，沒有垃圾回收的並發編程是困難的（Concurrency is hard without garbage collection）。但是由於指針算術的存在，在 C/C++里實現全自動垃圾回收更加困難。而那些天生具備垃圾回收的語言在並發編程方面具有明顯的優勢，Java 是目前支持並發編程最好的主流語言，它的 util.concurrent 庫和內存模型是 C++ 0x 效仿的對象。
學習多線程程序設計遠遠不是看看教程了解 API 怎么用那么簡單，這最多“主要是為了讀懂別人的代碼，如果自己要寫這類代碼，必須專門花時間嚴肅認真系統地學習，嚴禁半桶水上陣”（孟岩）。一般的多線程教程上都會提到要讓加鎖的區域足夠小，這沒錯，問題是如何找出這樣的區域並加鎖，本文第 9 節舉的安全讀寫 shared_ptr 可算是一個例子。
據我所知，目前 C++ 沒有好的多線程領域專著，C 語言有，Java 語言也有。《Java Concurrency in Practice》是我讀過的寫得最好的書，內容足夠新，可讀性和可操作性俱佳。C++ 程序員反過來要向 Java 學習，多少有些諷刺。除了編程書，操作系統教材也是必讀的，至少要完整地學習一本經典教材的相關章節，可從《操作系統設計與實現》、《現代操作系統》、《操作系統概念》任選一本，了解各種同步原語、臨界區、競態條件、死鎖、典型的 IPC 問題等等，防止閉門造車。
分析可能出現的 race condition 不僅是多線程編程基本功，也是設計分布式系統的基本功，需要反復歷練，形成一定的思考范式，並積累一些經驗教訓，才能少犯錯誤。這是一個快速發展的領域，要不斷吸收新知識，才不會落伍。單 CPU 時代的多線程編程經驗到了多 CPU 時代不一定有效，因為多 CPU 能做到真正的並發執行，每個 CPU 看到的事件發生順序不一定完全相同。正如狹義相對論所說的每個觀察者都有自己的時鍾，在不違反因果律的前提下，可能發生十分違反直覺的事情。
盡管本文通篇在講如何安全地使用（包括析構）跨線程的對象，但我建議盡量減少使用跨線程的對象，我贊同縉大師說的“用流水線，生產者-消費者，任務隊列這些有規律的機制，最低限度地共享數據。這是我所知最好的多線程編程的建議了。”
不用跨線程的對象，自然不會遇到本文描述的各種險態。如果迫不得已要用，我希望本文能對您有幫助。
- 原始指針暴露給多個線程往往會造成 race condition 或額外的簿記負擔；
- 統一用 shared_ptr/scoped_ptr 來管理對象的生命期，在多線程中尤其重要；
- shared_ptr 是值語意，當心意外延長對象的生命期。例如 boost::bind 和容器；
- weak_ptr 是 shared_ptr 的好搭檔，可以用作弱回調、對象池等；
- 認真閱讀一遍 boost::shared_ptr 的文檔，能學到很多東西： http://www.boost.org/doc/libs/release/libs/smart_ptr/shared_ptr.htm
- 保持開放心態，留意更好的解決辦法，比如 unique_ptr。忘掉已被廢棄的 auto_ptr。
shared_ptr 是 tr1 的一部分，即 C++ 標准庫的一部分，值得花一點時間去學習掌握，對編寫現代的 C++ 程序有莫大的幫助。我個人的經驗是，一周左右就能基本掌握各種用法與常見陷阱，比學 STL 還快。網絡上有一些對 shared_ptr 的批評，那可以算作故意誤用的例子，就好比故意訪問失效的迭代器來證明 vector 不安全一樣。
正確使用標准庫（含 shared_ptr）作為自動化的內存/資源管理器，解放大腦，從此告別內存錯誤。
之謬
本文第 8 節把 shared_ptr/weak_ptr 應用到 Observer 模式中，部分解決了其線程安全問題。我用 Observer 舉例，因為這是一個廣為人知的設計模式，但是它有本質的問題。
Observer 模式的本質問題在於其面向對象的設計。換句話說，我認為正是面向對象 (OO) 本身造成了 Observer 的缺點。Observer 是基類，這帶來了非常強的耦合，強度僅次於友元。這種耦合不僅限制了成員函數的名字、參數、返回值，還限制了成員函數所屬的類型（必須是 Observer 的派生類）。
Observer 是基類，這意味着如果 Foo 想要觀察兩個類型的事件（比如時鍾和溫度），需要使用多繼承。這還不是最糟糕的，如果要重復觀察同一類型的事件（比如 1 秒鍾一次的心跳和 30 秒鍾一次的自檢），就要用到一些伎倆來 work around，因為不能從一個 Base class 繼承兩次。
現在的語言一般可以繞過 Observer 模式的限制，比如 Java 可以用匿名內部類，Java 7 用 Closure，C# 用 delegate，C++ 用 boost::function/ boost::bind，我在另外一篇博客《以 boost::function 和 boost:bind 取代虛函數》里有更多的講解。
在 C++ 里為了替換 Observer，可以用 Signal/Slots，我指的不是 QT 那種靠語言擴展的實現，而是完全靠標准庫實現的 thread safe、race condition free、thread contention free 的 Signal/Slots，並且不強制要求 shared_ptr 來管理對象，也就是說完全解決了第 8 節列出的 Observer 遺留問題。不過這篇文章已經夠長了，留作下次吧。有興趣的同學可以先預習一下《借 shared_ptr 實現線程安全的 copy-on-write》。
《C++ 沉思錄》/《Runminations on C++》中文版的附錄是王曦和孟岩對作者夫婦二人的采訪，在被問到“請給我們三個你們認為最重要的建議”時，Koenig 和 Moo 的第一個建議是“避免使用指針”。我 2003 年讀到這段時，理解不深，覺得固然使用指針容易造成內存方面的問題，但是完全不用也是做不到的，畢竟 C++ 的多態要透過指針或引用來起效。6 年之后重新拾起來，發現大師的觀點何其深刻，不免掩卷長嘆。
這本書詳細地介紹了 handle/body idiom，這是編寫大型 C++ 程序的必備技術，也是實現物理隔離的法寶，值得細讀。
目前來看，用 shared_ptr 來管理資源在國內 C++ 界似乎並不是一種主流做法，很多人排斥智能指針，視其為洪水猛獸（這或許受了 auto_ptr 的垃圾設計的影響）。據我所知，很多 C++ 項目還是手動管理內存和資源，因此我覺得有必要把我認為好的做法分享出來，讓更多的人嘗試並采納。我覺得 shared_ptr 對於編寫線程安全的 C++ 程序是至關重要的，不然就得土法煉鋼，自己重新發明輪子。這讓我想起了 2001 年前后 STL 剛剛傳入國內，大家也是很猶豫，覺得它性能不高，使用不便，還不如自己造的容器類。近十年過去了，現在 STL 已經是主流，大家也適應了迭代器、容器、算法、適配器、仿函數這些“新”名詞“新”技術，開始在項目中普遍使用（至少用 vector 代替數組嘛）。我希望，幾年之后人們回頭看這篇文章，覺得“怎么講的都是常識”，那我這篇文章的目的也就達到了。
.全文完 2010/Jan/22初稿 Jan 27 修訂.
《C++ 編程規范》/《C++ Coding Standards》, by Herb Sutter and Andrei Alexandrescu, 2005. 條款 13.
http://www.boost.org/doc/libs/1_41_0/libs/smart_ptr/shared_ptr.htm
http://www.artima.com/cppsource/top_cpp_aha_moments.html
要是我在文章的一開始就談垃圾回收的好處，估計很多人直接就不看了。
孟岩《快速掌握一個語言最常用的 50%》博客，這篇博客的其他文字也很有趣味：“粗粗看看語法，就擼起袖子開干，邊查 Google 邊學習”這種路子也有問題，在對於這種語言的脾氣秉性還沒有了解的情況下大刀闊斧地拼湊代碼，寫出來的東西肯定不入流。說穿新鞋走老路，新瓶裝舊酒，那都是小問題，真正嚴重的是這樣的程序員可以在短時間內堆積大量充滿缺陷的垃圾代碼。由於通常開發階段的測試完備程度有限，這些垃圾代碼往往能通過這個階段，從而潛伏下來，在后期成為整個項目的毒瘤，反反復復讓后來的維護者陷入西西弗斯困境。……其實真正寫程序不怕完全不會，最怕一知半解的去攢解決方案。因為你完全不會，就自然會去認真查書學習，如果學習能力好的話，寫出來的代碼質量不會差。而一知半解，自己動手土法煉鋼，那搞出來的基本上都是廢銅爛鐵。
http://blog.csdn.net/Solstice/archive/2008/10/13/3066268.aspx
http://blog.csdn.net/Solstice/archive/2008/11/22/3351751.aspx