併發程式設計的兩種基本模型：message passing 和 shared memory。

使用message passing 可以跨機器，分散式系統的架構更具有一致性，擴容起來也較容易。

執行緒同步的四項原則

按重要性排序：

1. 首要原則是儘量最低限度地共享物件，減少需要同步的場合。一個物件能不暴露給別的執行緒就不要暴露；如果要暴露，優先考慮immutable物件；實在不行才暴露可修改的物件，並用同步措施來充分保護它。
2. 其次是使用高階的併發程式設計構件，如TaskQueue、Producer-Consumer Queue、CountDownLatch等等。
3. 最後不得已必須使用底層同步原語（primitives）時，只用非遞迴的互斥器和條件變數，慎用讀寫鎖，不要用訊號量。
4. 除了使用atomic整數之外，不自己編寫lock-free程式碼，也不要用“核心級”同步原語。不憑空猜測“哪種做法效能會更好”，比如spin lock vs. mutex。

下面著重講第3條：底層同步原語的使用。

互斥器

互斥器（mutex）保護了臨界區，任何一個時刻最多隻能有一個執行緒在此mutex劃出的臨界區內活動。單獨使用mutex時，主要為了保護共享資料。

一些原則：

1. 用RAII手法封裝mutex的建立、銷燬、加鎖、解鎖這四個操作。
2. 只用非遞迴的mutex（即不可重入的mutex）。
3. 不手工呼叫lock()和unlock()函式，一切交給棧上的Guard物件的構造和解構函式負責。Guard物件的生命期正好等於臨界區。這樣保證了始終在同一個函式同一個scope內對某個mutex加鎖和解鎖。避免在foo()內加鎖，然後跑到bar()內解鎖，也避免在不同的語句分支中分別加鎖、解鎖。這種做法稱為“Scoped Locking”。
4. 在每次構造Guard物件的時候，思考一路上（呼叫棧）已經持有的鎖，防止因加鎖順序不同而導致死鎖。由於Guard物件是棧上物件，看函式呼叫棧就能分析用鎖的情況，非常便利。

注：所謂“重入”，常見的情況是，程式執行到某個函式foo()時，收到訊號，於是暫停目前正在執行的函式，轉到訊號處理函式，而這個訊號處理函式的執行過程中，又恰恰也會進入到剛剛執行的函式foo()，這樣便發生了所謂的重入。此時如果foo()能夠正確的執行，而且處理完成後，之前暫停的foo()也能夠正確執行，則說明它是可重入的。

次要原則有：

1. 不使用跨程序的mutex，程序間通訊只用TCP sockets。
2. 加鎖、解鎖在同一個執行緒，執行緒a不能去unlock執行緒b已經鎖住的mutex（RAII自動保證）。
3. 別忘了解鎖（RAII自動保證）。
4. 不重複解鎖（RAII自動保證）。
5. 必要的時候可以考慮用PTHREAD_MUTEX_ERRORCHECK來排錯。

只使用非遞迴的mutex

mutex分為遞迴（可重入，reentrant）和非遞迴（不可重入），它們的唯一區別：同一執行緒可以重複對遞迴mutex加鎖，但是不能重複對非遞迴mutex加鎖。

遞迴mutex不用考慮一個執行緒會把自己鎖死，但是卻隱藏了一些問題，典型情況是你以為拿到一個鎖就能修改物件了，但是可能外層程式碼也已經拿到了鎖，正在修改或讀取同一個物件。這時將會造成意向不到的後果。

而如果使用非遞迴mutex，則程式將會死鎖——把程式的邏輯錯誤儘早暴露出來，而且死鎖更容易debug。

條件變數（condition variable）

互斥器是加鎖原語，用來排他性地訪問共享資料，它不是等待原語。

如果需要等待某個條件成立，我們應該使用條件變數——一個或多個執行緒等待某個布林表示式為真，即等待別的執行緒“喚醒”它，其學名叫做“管程（monitor）”。

對條件變數的使用包括兩個動作：
1. 執行緒等待某個條件, 條件為真則繼續執行，條件為假則將自己掛起(避免busy wait,節省CPU資源)；
2. 執行緒執行某些處理之後，條件成立；則通知等待該條件的執行緒繼續執行。

對於wait端：
1. 必須與mutex一起使用（防止race-condition），該布林表示式的讀寫需受此mutex保護。
2. 在mutex已上鎖的時候才能呼叫wait()。
3. 把判斷布林條件和wait()放到while迴圈中。

程式碼：

muduo::MutexLock mutex;
muduo::Condition cond(mutex);
std::deque<int> queue;

int dequeue()
{
    MutexLockGuard lock(mutex);
    while(queue.empty()) //queue.empty()為布林表示式，必須用迴圈；必須在判斷之後再wait()
    {
        cond.wait(); //原子地unlock mutex並進入等待，不會與enqueue死鎖
        // wait()執行完畢時會自動重新加鎖
    }
    assert(!queue.empty());
    int top = queue.front();
    queue.pop_front();
    return top;
}

為什麼是while迴圈來等待條件變數而不是if語句來判斷：
這是因為可能會存在虛假喚醒（spurious wakeup）的情況。
也就是說，即使沒有執行緒呼叫condition_signal, 原先呼叫condition_wait的函式也可能會返回。此時執行緒被喚醒了，但是條件並不滿足，這個時候如果不對條件進行檢查而往下執行，就可能會導致後續的處理出現錯誤。

虛假喚醒在linux的多處理器系統中，在程式接收到訊號時可能會發生。在Windows系統和JAVA虛擬機器上也存在。在系統設計時應該可以避免虛假喚醒，但是這會影響條件變數的執行效率，而既然通過while迴圈就能避免虛假喚醒造成的錯誤，因此程式的邏輯就變成了while迴圈的情況。

對於signal/broadcast端：
1. 不一定要在mutex已上鎖的情況下呼叫signal（理論上）。
2. 在signal之前一般要修改布林表示式。
3. 修改布林表示式通常要用mutex保護。
4. 注意區分signal與broadcast：broadcast通常用於表明狀態變化，signal通常用於表明資源可用。

程式碼：

ivoid enqueue(int x)
{
    MutexLockGuard lock(mutex);
    queue.push_back(x);
    cond.notify(); //可以移出臨界區之外
}

CountDownLatch（倒計時）是一種常用且易用的同步手段，其用途有二：
1. 主執行緒發起多個子執行緒，等這些子執行緒各自都完成一定的任務之後，主執行緒才繼續執行。通常用於主執行緒等待多個子執行緒完成初始化。
2. 主執行緒發起多個子執行緒，子執行緒等待主執行緒，主執行緒完成其他一些任務之後，通知所有子執行緒開始執行。通常用於多個子執行緒等待主執行緒發出“起跑”命令。

class CountDownLatch : boost::noncopyable
{
public:
    explicit CountDownLatch(int count); //倒數幾次
    void wait(); //等待計數值變為0
    void countDown(); //計數減1

private:
    mutable MutexLock mutex_; //順序很重要，先mutex後condition
    Condition condition_;
    int count_;
};

CountDownLatch::CountDownLatch(int count)
    : mutex_(),
      condition_(mutex_),
      count_(count)
{ }

void CountDownLatch::wait()
{
    MutexLockGuard lock(mutex_);
    while (count_)
    {
        condition_.wait();
    }
}

void CountDownLatch::countDown()
{
    MutexLockGuard lock(mutex_);
    --count_;
    if (count_ == 0)
    {
        condition_notifyAll();
    }
}

封裝MutexLock、MutexLockGuard、Condition

// Use as data member of a class, eg.
//
// class Foo
// {
//  public:
//   int size() const;
//
//  private:
//   mutable MutexLock mutex_;
//   std::vector<int> data_; // GUARDED BY mutex_
// };
class MutexLock : boost::noncopyable
{
 public:
  MutexLock()
    : holder_(0)
  {
    MCHECK(pthread_mutex_init(&mutex_, NULL));
  }

  ~MutexLock()
  {
    assert(holder_ == 0);
    MCHECK(pthread_mutex_destroy(&mutex_));
  }

  // must be called when locked, i.e. for assertion
  bool isLockedByThisThread() const
  {
    return holder_ == CurrentThread::tid();
  }

  void assertLocked() const
  {
    assert(isLockedByThisThread());
  }

  // internal usage

  void lock() //僅供MuetexLockGuard呼叫，嚴禁使用者程式碼呼叫
  {
    MCHECK(pthread_mutex_lock(&mutex_)); //兩行順序不能反
    assignHolder();
  }

  void unlock() //僅供MuetexLockGuard呼叫，嚴禁使用者程式碼呼叫
  {
    unassignHolder(); //兩行順序不能反
    MCHECK(pthread_mutex_unlock(&mutex_));
  }

  pthread_mutex_t* getPthreadMutex() /* non-const */
  {
    return &mutex_;
  }

 private:
  friend class Condition;

  class UnassignGuard : boost::noncopyable
  {
   public:
    UnassignGuard(MutexLock& owner)
      : owner_(owner)
    {
      owner_.unassignHolder();
    }

    ~UnassignGuard()
    {
      owner_.assignHolder();
    }

   private:
    MutexLock& owner_;
  };

  void unassignHolder()
  {
    holder_ = 0;
  }

  void assignHolder()
  {
    holder_ = CurrentThread::tid();
  }

  pthread_mutex_t mutex_;
  pid_t holder_;
};



class MutexLockGuard : boost::noncopyable
{
 public:
  explicit MutexLockGuard(MutexLock& mutex)
    : mutex_(mutex)
  {
    mutex_.lock();
  }

  ~MutexLockGuard()
  {
    mutex_.unlock();
  }

 private:

  MutexLock& mutex_;
};

// Prevent misuse like:
// MutexLockGuard(mutex_); 
// 以上將產生一個臨時物件又馬上銷燬了
// 正確寫法：MutexLockGuard lock(mutex_);
#define MutexLockGuard(x) error "Missing guard object name"

下面這個muduo::Condition class簡單封裝了Pthreads condition variable（boost、C++的執行緒庫中，同步原語過於龐雜。如果你不需要太高的靈活性，可以自己封裝幾個簡簡單單一看就明白的class來用——提供靈活性固然是本身，然而在不需要靈活性的地方把程式碼寫死，更需要大智慧）。

class Condition : boost::noncopyable
{
 public:
  explicit Condition(MutexLock& mutex)
    : mutex_(mutex)
  {
    MCHECK(pthread_cond_init(&pcond_, NULL));
  }

  ~Condition()
  {
    MCHECK(pthread_cond_destroy(&pcond_));
  }

  void wait()
  {
    MutexLock::UnassignGuard ug(mutex_);
    MCHECK(pthread_cond_wait(&pcond_, mutex_.getPthreadMutex()));
  }

  // returns true if time out, false otherwise.
  bool waitForSeconds(double seconds);

  void notify()
  {
    MCHECK(pthread_cond_signal(&pcond_));
  }

  void notifyAll()
  {
    MCHECK(pthread_cond_broadcast(&pcond_));
  }

 private:
  MutexLock& mutex_;
  pthread_cond_t pcond_;
};

// returns true if time out, false otherwise.
bool muduo::Condition::waitForSeconds(double seconds)
{
  struct timespec abstime;
  // FIXME: use CLOCK_MONOTONIC or CLOCK_MONOTONIC_RAW to prevent time rewind.
  clock_gettime(CLOCK_REALTIME, &abstime);

  const int64_t kNanoSecondsPerSecond = 1e9;
  int64_t nanoseconds = static_cast<int64_t>(seconds * kNanoSecondsPerSecond);

  abstime.tv_sec += static_cast<time_t>((abstime.tv_nsec + nanoseconds) / kNanoSecondsPerSecond);
  abstime.tv_nsec = static_cast<long>((abstime.tv_nsec + nanoseconds) % kNanoSecondsPerSecond);

  MutexLock::UnassignGuard ug(mutex_);
  return ETIMEDOUT == pthread_cond_timedwait(&pcond_, mutex_.getPthreadMutex(), &abstime);
}

mutex和condition都是非常底層的同步原語，主要用來實現更高階的併發程式設計工具，並不鼓勵到處使用。

執行緒安全的Singleton實現

使用pthread_once：

#include <boost/noncopyable.hpp>
#include <assert.h>
#include <stdlib.h> // atexit
#include <pthread.h>

namespace muduo
{

namespace detail
{
// This doesn't detect inherited member functions!
// http://stackoverflow.com/questions/1966362/sfinae-to-check-for-inherited-member-functions
template<typename T>
struct has_no_destroy
{
  template <typename C> static char test(typeof(&C::no_destroy)); // or decltype in C++11
  template <typename C> static int32_t test(...);
  const static bool value = sizeof(test<T>(0)) == 1;
};
}

template<typename T>
class Singleton : boost::noncopyable
{
 public:
  static T& instance()
  {
    pthread_once(&ponce_, &Singleton::init);
    assert(value_ != NULL);
    return *value_;
  }

 private:
  Singleton();
  ~Singleton();

  static void init()
  {
    value_ = new T();
    if (!detail::has_no_destroy<T>::value)
    {
      ::atexit(destroy); //註冊銷燬函式
    }
  }

  static void destroy()
  {
    typedef char T_must_be_complete_type[sizeof(T) == 0 ? -1 : 1];
    T_must_be_complete_type dummy; (void) dummy;

    delete value_;
    value_ = NULL;
  }

 private:
  static pthread_once_t ponce_;
  static T*             value_;
};

//靜態變數初始化
template<typename T>
pthread_once_t Singleton<T>::ponce_ = PTHREAD_ONCE_INIT;

template<typename T>
T* Singleton<T>::value_ = NULL;

}

它用pthread_once_t來保證lazy-initialization的執行緒安全，執行緒安全性由Pthreads庫保證。另外，我們通過atexit(3)來提供銷燬功能。

使用方法：

Foo& foo = Singleton<Foo>::instance();

關於noncopyable

boost::noncopyable 用於防止複製，如果是自己實現，需要把建構函式、拷貝構造，複製構造都私有。

關於pthread_once

有時候我們需要對一些posix變數只進行一次初始化。如果我們進行多次初始化程式就會出現錯誤。

在傳統的順序程式設計中，一次性初始化經常通過使用布林變數來管理。控制變數被靜態初始化為0，而任何依賴於初始化的程式碼都能測試該變數。如果變數值仍然為0，則它能實行初始化，然後將變數置為1。以後檢查的程式碼將跳過初始化。

但是在多執行緒程式設計中，事情就變的複雜的多。如果多個執行緒併發地執行初始化序列程式碼，可能有2個執行緒發現控制變數為0，並且都實行初始化，而該過程本該僅僅執行一次。

如果我們需要對一個posix變數靜態的初始化，可使用的方法是用一個互斥量對該變數的初始化進行控制。但有時候我們需要對該變數進行動態初始化，pthread_once就會方便的多。

原型：

int pthread_once(pthread_once_t *once_control, void (*init_routine) (void))；

功能：本函式使用初值為PTHREAD_ONCE_INIT的once_control變數保證init_routine()函式在本程序執行序列中僅執行一次。

在多執行緒程式設計環境下，儘管pthread_once()呼叫會出現在多個執行緒中，init_routine()函式僅執行一次，究竟在哪個執行緒中執行是不定的，是由核心排程來決定。

Linux Threads使用互斥鎖和條件變數保證由pthread_once()指定的函式執行且僅執行一次，而once_control表示是否執行過。

如果once_control的初值不是PTHREAD_ONCE_INIT（Linux Threads定義為0），pthread_once() 的行為就會不正常。

在Linux Threads中，實際”一次性函式”的執行狀態有三種：NEVER（0）、IN_PROGRESS（1）、DONE （2），如果once初值設為1，則由於所有pthread_once()都必須等待其中一個激發”已執行一次”訊號，因此所有pthread_once ()都會陷入永久的等待中；如果設為2，則表示該函式已執行過一次，從而所有pthread_once()都會立即返回0。

關於destroy

防止delete一個不完全物件

    typedef char T_must_be_complete_type[sizeof(T) == 0 ? -1 : 1];
    T_must_be_complete_type dummy; (void) dummy;

typedef定義一個char陣列型別 T_MUST_BE_COMPELET_TYPE ：
char[-1]：如果T只宣告沒有定義，為不完全型別, 沒定義就沒有解構函式，delete就不會呼叫析構函數了；
char[1]：T是完全型別，即有定義，有delete操作，可以呼叫解構函式。

由於sizeof不能用於不完全型別，所以事實上如果T是不完全型別，編譯的時候將會在sizeof處報錯。

作者使用下面這兩句的意圖：

// 如果編譯時加Werror=unused-local-typedefs這個選項
// 則檢查到未使用的local-typedefs會報錯，因此我們下面使用一下...
T_must_be_complete_type dummy; 

// 如果編譯時加Wno-unused-parameter這個選項
// 則檢查到未使用的parameter會報錯，因此我們下面使用一下...
(void) dummy;

關於has_no_destroy

has_no_destroy用於判斷一個型別是否需要銷燬。
當且僅當一個型別是類型別，而且包含有no_destroy資料成員時，才不需要銷燬。

template<typename T>
struct has_no_destroy
{
  template <typename C>
  static char test(typeof(&C::no_destroy)); // or decltype in C++11
  template <typename C>
  static int32_t test(...);
  const static bool value = sizeof(test<T>(0)) == 1;
};

// 在Singleton的init函式中使用
static void init()
{
  value_ = new T();
  if (!detail::has_no_destroy<T>::value)
  {
    ::atexit(destroy);
  }
}

這裡我還是簡單分析下這裡的邏輯吧：

當我們寫下下面程式碼的時候：

Foo& foo = Singleton<Foo>::instance();

編譯器將會用Foo來例項化 Singleton< T >，從而產生Singleton< Foo >例項。在instance()中，會呼叫init()，在inti函式中，將會例項化出detail::has_no_destroy< Foo >，並訪問value靜態變數的值，其值如下：

const static bool value = sizeof(test<T>(0)) == 1;

我們知道，sizeof是一個運算子，其值會在編譯的時候計算出來，對一個函式sizeof，實際上是對該函式的返回型別進行sizeof，因此sizeof可能對兩個test函式的返回型別char或者int32_t進行操作。如果是char，則sizeof結果為1，value為true，如果是int32_t，sizeof結果不為1，value為false。

因此要達到我們的目的，需要讓“包含有no_destroy資料成員的類型別”匹配到第一個test函式，而其他情況匹配到第二個test函式上。

假設T是Foo，則實際上為：

sizeof(test<Foo>(0));

這個時候編譯器會進行匹配（匹配規則見上述的博文），當匹配第一個test函式的時候，編譯器發現Foo是一個類型別，因此使用::是正確的，如果Foo包含no_destroy資料成員，則這個函式匹配成功，value為true，如果Foo不包含no_destroy資料成員，則這個函式匹配失敗——由於“匹配失敗不是錯誤”，所以編譯器並不會報錯，而是匹配下一個test函式，這一次匹配成功了，value為false。

如果T是一個內建型別，則::的使用時錯誤的，因此第一個test匹配失敗，編譯器轉而匹配第二個test，value為false。

以上的這些操作都是在編譯期間完成的——也就是說，編譯完成之後，程式碼中實際上只剩下value=true 或者value=false這一行程式碼了。