鎖開銷優化以及 CAS 簡單說明
鎖
互斥鎖是用來保護一個臨界區,即保護一個訪問共用資源的程序片段,而這些共用資源又無法同時被多個線程訪問的特性。當有線程進入臨界區段時,其他線程或是進程必須等待。
在談及鎖的性能開銷,一般都會說鎖的開銷很大,那鎖的開銷有多大,主要耗在哪,怎么提高鎖的性能。
鎖的開銷
現在鎖的機制一般使用 futex(fast Userspace mutexes),內核態和用戶態的混合機制。還沒有futex的時候,內核是如何維護同步與互斥的呢?系統內核維護一個對象,這個對象對所有進程可見,這個對象是用來管理互斥鎖並且通知阻塞的進程。如果進程A要進入臨界區,先去內核查看這個對象,有沒有別的進程在占用這個臨界區,出臨界區的時候,也去內核查看這個對象,有沒有別的進程在等待進入臨界區,然后根據一定的策略喚醒等待的進程。這些不必要的系統調用(或者說內核陷入)造成了大量的性能開銷。為了解決這個問題,Futex就應運而生。
Futex是一種用戶態和內核態混合的同步機制。首先,同步的進程間通過mmap共享一段內存,futex變量就位於這段共享的內存中且操作是原子的,當進程嘗試進入互斥區或者退出互斥區的時候,先去查看共享內存中的futex變量,如果沒有競爭發生,則只修改futex,而不用再執行系統調用了。當通過訪問futex變量告訴進程有競爭發生,則還是得執行系統調用去完成相應的處理(wait 或者 wake up)。簡單的說,futex就是通過在用戶態的檢查,(motivation)如果了解到沒有競爭就不用陷入內核了,大大提高了low-contention時候的效率。
mutex 是在 futex 的基礎上用的內存共享變量來實現的,如果共享變量建立在進程內,它就是一個線程鎖,如果它建立在進程間共享內存上,那么它是一個進程鎖。pthread_mutex_t 中的 _lock 字段用於標記占用情況,先使用CAS判斷_lock是否占用,若未占用,直接返回。否則,通過__lll_lock_wait_private 調用SYS_futex 系統調用迫使線程進入沉睡。 CAS是用戶態的 CPU 指令,若無競爭,簡單修改鎖狀態即返回,非常高效,只有發現競爭,才通過系統調用陷入內核態。所以,FUTEX是一種用戶態和內核態混合的同步機制,它保證了低競爭情況下的鎖獲取效率。
所以如果鎖不存在沖突,每次獲得鎖和釋放鎖的處理器開銷僅僅是CAS指令的開銷。
確定一件事情最好的方法是實際測試和觀測它,讓我們寫一段代碼來測試無沖突時鎖的開銷:
#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <time.h>
static inline long long unsigned time_ns(struct timespec* const ts) {
if (clock_gettime(CLOCK_REALTIME, ts)) {
exit(1);
}
return ((long long unsigned) ts->tv_sec) * 1000000000LLU
+ (long long unsigned) ts->tv_nsec;
}
int main()
{
int res = -1;
pthread_mutex_t mutex;
//初始化互斥量,使用默認的互斥量屬性
res = pthread_mutex_init(&mutex, NULL);
if(res != 0)
{
perror("pthread_mutex_init failed\n");
exit(EXIT_FAILURE);
}
long MAX = 1000000000;
long c = 0;
struct timespec ts;
const long long unsigned start_ns = time_ns(&ts);
while(c < MAX)
{
pthread_mutex_lock(&mutex);
c = c + 1;
pthread_mutex_unlock(&mutex);
}
const long long unsigned delta = time_ns(&ts) - start_ns;
printf("%f\n", delta/(double)MAX);
return 0;
}
說明:以下性能測試在騰訊雲 Intel(R) Xeon(R) CPU E5-26xx v4 1核 2399.996MHz 下進行。
運行了 10 億次,平攤到每次加鎖/解鎖操作大概是 2.2ns 每次加鎖/解鎖(扣除了循環耗時 2.7ns)
在鎖沖突的情況下,開銷就沒有這么小了。
首先pthread_mutex_lock會真正的調用sys_futex來進入內核來試圖加鎖,被鎖住以后線程會進入睡眠,這帶來了上下文切換和線程調度的開銷。
可以寫兩個互相解鎖的線程來測試這個過程的開銷:
// Copyright (C) 2010 Benoit Sigoure
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#include <pthread.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/syscall.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#include <linux/futex.h>
static inline long long unsigned time_ns(struct timespec* const ts) {
if (clock_gettime(CLOCK_REALTIME, ts)) {
exit(1);
}
return ((long long unsigned) ts->tv_sec) * 1000000000LLU
+ (long long unsigned) ts->tv_nsec;
}
static const int iterations = 500000;
static void* thread(void* restrict ftx) {
int* futex = (int*) ftx;
for (int i = 0; i < iterations; i++) {
sched_yield();
while (syscall(SYS_futex, futex, FUTEX_WAIT, 0xA, NULL, NULL, 42)) {
// retry
sched_yield();
}
*futex = 0xB;
while (!syscall(SYS_futex, futex, FUTEX_WAKE, 1, NULL, NULL, 42)) {
// retry
sched_yield();
}
}
return NULL;
}
int main(void) {
struct timespec ts;
const int shm_id = shmget(IPC_PRIVATE, sizeof (int), IPC_CREAT | 0666);
int* futex = shmat(shm_id, NULL, 0);
pthread_t thd;
if (pthread_create(&thd, NULL, thread, futex)) {
return 1;
}
*futex = 0xA;
const long long unsigned start_ns = time_ns(&ts);
for (int i = 0; i < iterations; i++) {
*futex = 0xA;
while (!syscall(SYS_futex, futex, FUTEX_WAKE, 1, NULL, NULL, 42)) {
// retry
sched_yield();
}
sched_yield();
while (syscall(SYS_futex, futex, FUTEX_WAIT, 0xB, NULL, NULL, 42)) {
// retry
sched_yield();
}
}
const long long unsigned delta = time_ns(&ts) - start_ns;
const int nswitches = iterations << 2;
printf("%i thread context switches in %lluns (%.1fns/ctxsw)\n",
nswitches, delta, (delta / (float) nswitches));
wait(futex);
return 0;
}
編譯使用 gcc -std=gnu99 -pthread context_switch.c。
運行的結果是 2003.4ns/ctxsw,所以鎖沖突的開銷大概是不沖突開銷的 910 倍了,相差出乎意料的大。
另外一個c程序可以用來測試“純上下文切換”的開銷,線程只是使用sched_yield來放棄處理器,並不進入睡眠。
// Copyright (C) 2010 Benoit Sigoure
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
#include <sched.h>
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <string.h>
#include <errno.h>
static inline long long unsigned time_ns(struct timespec* const ts) {
if (clock_gettime(CLOCK_REALTIME, ts)) {
exit(1);
}
return ((long long unsigned) ts->tv_sec) * 1000000000LLU
+ (long long unsigned) ts->tv_nsec;
}
static const int iterations = 500000;
static void* thread(void*ctx) {
(void)ctx;
for (int i = 0; i < iterations; i++)
sched_yield();
return NULL;
}
int main(void) {
struct sched_param param;
param.sched_priority = 1;
if (sched_setscheduler(getpid(), SCHED_FIFO, ¶m))
fprintf(stderr, "sched_setscheduler(): %s\n", strerror(errno));
struct timespec ts;
pthread_t thd;
if (pthread_create(&thd, NULL, thread, NULL)) {
return 1;
}
long long unsigned start_ns = time_ns(&ts);
for (int i = 0; i < iterations; i++)
sched_yield();
long long unsigned delta = time_ns(&ts) - start_ns;
const int nswitches = iterations << 2;
printf("%i thread context switches in %lluns (%.1fns/ctxsw)\n",
nswitches, delta, (delta / (float) nswitches));
return 0;
}
“純上下文切換” 消耗了大概381.2ns/ctxsw。
這樣我們大致可以把鎖沖突的開銷分成三部分,“純上下文切換”開銷,大概是 381.2ns,調度器開銷(把線程從睡眠變成就緒或者反過來)大概是1622.2ns,在多核系統上,還存在跨處理器調度的開銷,那部分開銷很大。在真實的應用場景里,還要考慮上下文切換帶來的cache不命中和TLB不命中的開銷,開銷只會進一步加大。
鎖的優化
從上面可以知道,真正消耗時間的不是上鎖的次數,而是鎖沖突的次數。減少鎖沖突的次數才是提升性能的關鍵。使用更細粒度的鎖,可以減少鎖沖突。這里說的粒度包括時間和空間,比如哈希表包含一系列哈希桶,為每個桶設置一把鎖,空間粒度就會小很多--哈希值相互不沖突的訪問不會導致鎖沖突,這比為整個哈希表維護一把鎖的沖突機率低很多。減少時間粒度也很容易理解,加鎖的范圍只包含必要的代碼段,盡量縮短獲得鎖到釋放鎖之間的時間,最重要的是,絕對不要在鎖中進行任何可能會阻塞的操作。使用讀寫鎖也是一個很好的減少沖突的方式,讀操作之間不互斥,大大減少了沖突。
假設單向鏈表中的插入/刪除操作很少,主要操作是搜索,那么基於單一鎖的方法性能會很差。在這種情況下,應該考慮使用讀寫鎖,即 pthread_rwlock_t,這么做就允許多個線程同時搜索鏈表。插入和刪除操作仍然會鎖住整個鏈表。假設執行的插入和搜索操作數量差不多相同,但是刪除操作很少,那么在插入期間鎖住整個鏈表是不合適的,在這種情況下,最好允許在鏈表中的分離點(disjoint point)上執行並發插入,同樣使用基於讀寫鎖的方式。在兩個級別上執行鎖定,鏈表有一個讀寫鎖,各個節點包含一個互斥鎖,在插入期間,寫線程在鏈表上建立讀鎖,然后繼續處理。在插入數據之前,鎖住要在其后添加新數據的節點,插入之后釋放此節點,然后釋放讀寫鎖。刪除操作在鏈表上建立寫鎖。不需要獲得與節點相關的鎖;互斥鎖只建立在某一個操作節點之上,大大減少鎖沖突的次數。
鎖本身的行為也存在進一步優化的可能性,sys_futex系統調用的作用在於讓被鎖住的當前線程睡眠,讓出處理器供其它線程使用,既然這個過程的消耗很高,也就是說如果被鎖定的時間不超過這個數值的話,根本沒有必要進內核加鎖——釋放的處理器時間還不夠消耗的。sys_futex的時間消耗夠跑很多次 CAS 的,也就是說,對於一個鎖沖突比較頻繁而且平均鎖定時間比較短的系統,一個值得考慮的優化方式是先循環調用 CAS 來嘗試獲得鎖(這個操作也被稱作自旋鎖),在若干次失敗后再進入內核真正加鎖。當然這個優化只能在多處理器的系統里起作用(得有另一個處理器來解鎖,否則自旋鎖無意義)。在glibc的pthread實現里,通過對pthread_mutex設置PTHREAD_MUTEX_ADAPTIVE_NP屬性就可以使用這個機制。
CAS
鎖產生的一些問題:
- 等待互斥鎖會消耗寶貴的時間,鎖的開銷很大。
- 低優先級的線程可以獲得互斥鎖,因此阻礙需要同一互斥鎖的高優先級線程。這個問題稱為優先級倒置(priority inversion )
- 可能因為分配的時間片結束,持有互斥鎖的線程被取消調度。這對於等待同一互斥鎖的其他線程有不利影響,因為等待時間現在會更長。這個問題稱為鎖護送(lock convoying)
無鎖編程的好處之一是一個線程被掛起,不會影響到另一個線程的執行,避免鎖護送;在鎖沖突頻繁且平均鎖定時間較短的系統,避免上下文切換和調度開銷。
CAS (comapre and swap 或者 check and set),比較並替換,引用 wiki,它是一種用於線程數據同步的原子指令。
CAS 核心算法涉及到三個參數,即內存值,更新值和期望值;CAS 指令會先檢查一個內存位置是否包含預期的值;如果是這樣,就把新的值復制到這個位置,返回 true;如果不是則返回 false。
CAS 對應一條匯編指令 CMPXCHG,因此是原子性的。
bool compare_and_swap (int *accum, int *dest, int newval)
{
if ( *accum == *dest ) {
*dest = newval;
return true;
}
return false;
}
一般,程序會在循環里使用 CAS 不斷去完成一個事務性的操作,一般包含拷貝一個共享的變量到一個局部變量,然后再使用這個局部變量執行任務計算得到新的值,最后再使用 CAS 比較保存再局部變量的舊值和內存值來嘗試提交你的修改,如果嘗試失敗,會重新讀取一遍內存值,再重新計算,最后再使用 CAS 嘗試提交修改,如此循環。比如:
void LockFreeQueue::push(Node* newHead)
{
for (;;)
{
// 拷貝共享變量(m_Head) 到一個局部變量
Node* oldHead = m_Head;
// 執行任務,可以不用關注其他線程
newHead->next = oldHead;
// 下一步嘗試提交更改到共享變量
// 如果共享變量沒有被其他線程修改過,仍為 oldHead,則 CAS 將 newHead 賦值給共享變量 m_Head 並返回
// 否則繼續循環重試
if (_InterlockedCompareExchange(&m_Head, newHead, oldHead))
return;
}
}
上面的數據結構設置了一個共享的頭節點 m_Head,當 push 一個新的節點時,會把新節點加在頭節點后面;不要相信程序的執行是連續的,CPU 的執行是多線程並發。在 _InterlockedCompareExchange 即 CAS 之前,線程可能因為時間片用完被調度出去,新調度進來的線程執行完了 push 操作,多個線程共享了 m_Head 變量,此時 m_Head 已被修改了,如果原來線程繼續執行,把 oldHead 覆蓋到 m_Head,就會丟失其他線程 push 進來的節點。所以需要比較 m_Head 是不是還等於 oldHead,如果是,說明頭節點不變,可以使用 newHead 覆蓋 m_Head;如果不是,說明有其他線程 push 了新的節點,那么需要使用最新的 m_Head 更新 oldHead 的值重新走一下循環,_InterlockedCompareExchange 會自動把 m_Head 賦值給 oldHead。
ABA 問題
因為 CAS 需要在提交修改時檢查期望值和內存值有沒有發生變化,如果沒有則進行更新,但是如果原來一個值從 A 變成 B 又變成 A,那么使用 CAS 檢查的時候發現值沒有發生變化,但實際上已經發生了一系列變化。
內存的回收利用會導致 CAS 出現嚴重的問題:
T* ptr1 = new T(8, 18);
T* old = ptr1;
delete ptr1;
T* ptr2 = new T(0, 1);
// 我們不能保證操作系統不會重新使用 ptr1 內存地址,一般的內存管理器都會這樣子做
if (old1 == ptr2) {
// 這里表示,剛剛回收的 ptr1 指向的內存被用於后面申請的 ptr2了
}
ABA問題是無鎖結構實現中常見的一種問題,可基本表述為:
- 進程P1讀取了一個數值A
- P1被掛起(時間片耗盡、中斷等),進程P2開始執行
- P2修改數值A為數值B,然后又修改回A
- P1被喚醒,比較后發現數值A沒有變化,程序繼續執行。
對於P1來說,數值A未發生過改變,但實際上A已經被變化過了,繼續使用可能會出現問題。在CAS操作中,由於比較的多是指針,這個問題將會變得更加嚴重。試想如下情況:
有一個堆(先入后出)中有top和節點A,節點A目前位於堆頂top指針指向A。現在有一個進程P1想要pop一個節點,因此按照如下無鎖操作進行
pop()
{
do{
ptr = top; // ptr = top = NodeA
next_prt = top->next; // next_ptr = NodeX
} while(CAS(top, ptr, next_ptr) != true);
return ptr;
}
而進程P2在執行CAS操作之前打斷了P1,並對堆進行了一系列的pop和push操作,使堆變為如下結構:
進程P2首先pop出NodeA,之后又Push了兩個NodeB和C,由於內存管理機制中廣泛使用的內存重用機制,導致NodeC的地址與之前的NodeA一致。
這時P1又開始繼續運行,在執行CAS操作時,由於top依舊指向的是NodeA的地址(實際上已經變為NodeC),因此將top的值修改為了NodeX,這時堆結構如下:
經過CAS操作后,top指針錯誤的指向了NodeX而不是NodeB。
ABA 解決方法
Tagged state reference,增加額外的 tag bits 位,它像一個版本號;比如,其中一種算法是在內存地址的低位記錄指針的修改次數,在指針修改時,下一次 CAS 會返回失敗,即使因為內存重用機制導致地址一樣。有時我們稱這種機制位 ABA‘,因為我們使第二個 A 稍微有點不同於第一個。tag 的位數長度會影響記錄修改的次數,在現有的 CPU 下,使用 60 bit tag,在不重啟程序10年才會產生溢出問題;在 X64 CPU,趨向於支持 128 bit 的 CAS 指令,這樣更能保證避免出現 ABA 問題。
下面參考 liblfds 庫代碼說明下 Tagged state reference 的實現過程。
我們想要避免 ABA 問題的方法之一是使用更長的指針,這樣便需要一個支持 dword 長度的 CAS 指令。liblfds 是怎么跨平台實現 128 bit 指令的呢?
在 liblfds 下,CAS 指令為 LFDS710_PAL_ATOMIC_DWCAS 宏,它的完整形式是:
LFDS710_PAL_ATOMIC_DWCAS( pointer_to_destination, pointer_to_compare, pointer_to_new_destination, cas_strength, result)
- pointer_to_destination: [in, out],指向目標的指針,是一個由兩個 64 bit 整數組成的數組;
- pointer_to_compare: [in, out],用於和目標指針比較的指針,同樣是一個由兩個 64 bit 整數組成的數組;
- pointer_to_new_destination: [in],和目標指針交換的新指針;
- result: [out],如果 128 bit 的 pointer_to_compare 與 pointer_to_destination 相等,則使用 pointer_to_new_destination 覆蓋 pointer_to_destination,result 返回 1;如果不相等,則 pointer_to_destination 不變,且 pointer_to_compare 的值變為 pointer_to_destination。
從上面可以看出,liblfds 庫使用一個由兩個元素組成的一維數組來表示 128 bit 指針。
Linux 提供了 cmpxchg16b 用於實現 128 bit 的 CAS 指令,而在 Windows,使用 _InterlockedCompareExchange128。只有 128 位指針完全相等的情況下,才視為相等。
參考 liblfds/liblfds7.1.0/liblfds710/inc/liblfds710/lfds710_porting_abstraction_layer_compiler.h 下關於 CAS 的 windows 實現:
#define LFDS710_PAL_ATOMIC_DWCAS( pointer_to_destination, pointer_to_compare, pointer_to_new_destination, cas_strength, result ) \
{ \
LFDS710_PAL_BARRIER_COMPILER_FULL; \
(result) = (char unsigned) _InterlockedCompareExchange128( (__int64 volatile *) (pointer_to_destination), (__int64) (pointer_to_new_destination[1]), (__int64) (pointer_to_new_destination[0]), (__int64 *) (pointer_to_compare) ); \
LFDS710_PAL_BARRIER_COMPILER_FULL; \
}
再重點研究 new_top 的定義和提交修改過程。
new_top 是一個具有兩個元素的一維數組,元素是 struct lfds710_stack_element 指針,兩個元素分別使用 POINTER 0 和 COUNTER 1 標記。COUNTER 相當於前面說的 tag 標記,POINTER 保存的時真正的節點指針。在 X64 下,指針長度是 64 bit,所以這里使用的是 64 bit tag 記錄 pointer 修改記錄。
liblfds 用原 top 的 COUNTER + 1來初始化 new top COUNTER,即使用 COUNTER 標記 ss->top 的更換次數,這樣每一次更換 top,top 里的 COUNTER 都會變。
只有在 ss->top 和 original_top 的 POINTER 和 COUNTER 完全相等的情況下,new_top 才會覆蓋到 ss->top,否則會使用 ss->top 覆蓋 original_top,下次循環用最新的 original_top 再次操作和比較。
參考 liblfds/liblfds7.1.0/liblfds710/src/lfds710_stack/lfds710_stack_push.c,無鎖堆棧的實現:
void lfds710_stack_push( struct lfds710_stack_state *ss,
struct lfds710_stack_element *se )
{
char unsigned
result;
lfds710_pal_uint_t
backoff_iteration = LFDS710_BACKOFF_INITIAL_VALUE;
struct lfds710_stack_element LFDS710_PAL_ALIGN(LFDS710_PAL_ALIGN_DOUBLE_POINTER)
*new_top[PAC_SIZE],
*volatile original_top[PAC_SIZE];
LFDS710_PAL_ASSERT( ss != NULL );
LFDS710_PAL_ASSERT( se != NULL );
new_top[POINTER] = se;
original_top[COUNTER] = ss->top[COUNTER];
original_top[POINTER] = ss->top[POINTER];
do
{
se->next = original_top[POINTER];
LFDS710_MISC_BARRIER_STORE;
new_top[COUNTER] = original_top[COUNTER] + 1;
LFDS710_PAL_ATOMIC_DWCAS( ss->top, original_top, new_top, LFDS710_MISC_CAS_STRENGTH_WEAK, result );
if( result == 0 )
LFDS710_BACKOFF_EXPONENTIAL_BACKOFF( ss->push_backoff, backoff_iteration );
}
while( result == 0 );
LFDS710_BACKOFF_AUTOTUNE( ss->push_backoff, backoff_iteration );
return;
}
CAS 原理應用
- 無鎖數據結構,參考 https://github.com/liblfds/liblfds
- 高性能內存隊列disruptor中的CAS,參考 http://ifeve.com/disruptor/
- 數據庫樂觀鎖
參考
[wiki Compare-and-swap] https://en.wikipedia.org/wiki/Compare-and-swap
[wiki ABA problem] https://en.wikipedia.org/wiki/ABA_problem
[左耳朵耗子無鎖隊列的實現] https://coolshell.cn/articles/8239.html
[IBM 設計不使用互斥鎖的並發數據結構] https://www.ibm.com/developerworks/cn/aix/library/au-multithreaded_structures2/index.html#artrelatedtopics
[ABA problem] https://lumian2015.github.io/lockFreeProgramming/aba-problem.html
[_InterlockedCompareExchange128] https://docs.microsoft.com/en-us/cpp/intrinsics/interlockedcompareexchange128?view=vs-2019
[Linux 互斥鎖的實現原理(pthread_mutex_t)] https://www.bbsmax.com/A/x9J2WXvW56/
[futex機制介紹] https://blog.csdn.net/y33988979/article/details/82252266
[an-introduction-to-lock-free-programming] https://preshing.com/20120612/an-introduction-to-lock-free-programming/
[多進程、多線程與多處理器計算平台的性能問題] https://blog.csdn.net/Jmilk/article/details/81049623
[Implement Lock-Free Queue] http://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.53.8674&rep=rep1&type=pdf
[上下文切換和線程調度性能測試] https://github.com/tsuna/contextswitch/blob/master/timetctxsw.c
[純上下文切換性能測試] https://github.com/tsuna/contextswitch/blob/master/timetctxsw2.c
[鎖的開銷] http://xbay.github.io/2015/12/31/鎖的開銷/
[pthread包的mutex實現分析] https://blog.csdn.net/tlxamulet/article/details/79047717
[IBM通用線程:POSIX 線程詳解] https://www.ibm.com/developerworks/cn/linux/thread/posix_thread2/index.html