C++ 併發編程（六）：信號量（Semaphore）

下面這段介紹，修改自 wxWidgets 官方文檔（詳見：wxSemaphore Class Reference）。

Semaphore is a counter limiting the number of threads concurrently accessing a shared resource.

This counter is always between 0 and the maximum value specified during the semaphore creation. When the counter is strictly greater than 0, a call to Wait returns immediately and decrements the counter. As soon as it reaches 0, any subsequent calls to Semaphore::Wait block and only return when the semaphore counter becomes strictly positive again as the result of calling Signal which increments the counter.

In general, semaphores are useful to restrict access to a shared resource which can only be accessed by some fixed number of clients at the same time. For example, when modeling a hotel reservation system a semaphore with the counter equal to the total number of available rooms could be created. Each time a room is reserved, the semaphore should be acquired by calling Wait and each time a room is freed it should be released by calling Signal.

C++11 和 Boost.Thread 都沒有提供信號量。對此 Boost 是這樣解釋的（Why has class semaphore disappeared?）：

Semaphore was removed as too error prone. The same effect can be achieved with greater safety by the combination of a mutex and a condition variable. Dijkstra (the semaphore's inventor), Hoare, and Brinch Hansen all depreciated semaphores and advocated more structured alternatives. In a 1969 letter to Brinch Hansen, Wirth said "semaphores ... are not suitable for higher level languages." [Andrews-83] summarizes typical errors as "omitting a P or a V, or accidentally coding a P on one semaphore and a V on on another", forgetting to include all references to shared objects in critical sections, and confusion caused by using the same primitive for "both condition synchronization and mutual exclusion".

簡單來講，就是信號量太容易出錯了（too error prone），經過組合互斥鎖（mutex）和條件變量（condition variable）能夠達到相同的效果，且更加安全。實現以下：

class Semaphore {
public:
  explicit Semaphore(int count = 0) : count_(count) {
  }

  void Signal() {
    std::unique_lock<std::mutex> lock(mutex_);
    ++count_;
    cv_.notify_one();
  }

  void Wait() {
    std::unique_lock<std::mutex> lock(mutex_);
    cv_.wait(lock, [=] { return count_ > 0; });
    --count_;
  }

private:
  std::mutex mutex_;
  std::condition_variable cv_;
  int count_;
};

下面建立三個工做線程（Worker），來測試這個信號量。

int main() {
  const std::size_t SIZE = 3;

  std::vector<std::thread> v;
  v.reserve(SIZE);

  for (std::size_t i = 0; i < SIZE; ++i) {
    v.emplace_back(&Worker);
  }

  for (std::thread& t : v) {
    t.join();
  }
  
  return 0;
}

每一個工做線程先等待信號量，而後輸出線程 ID 和當前時間，輸出操做以互斥鎖同步以防止錯位，睡眠一秒是爲了模擬線程處理數據的耗時。

std::mutex g_io_mutex;

void Worker() {
  g_semaphore.Wait();

  std::thread::id thread_id = std::this_thread::get_id();

  std::string now = FormatTimeNow("%H:%M:%S");
  {
    std::lock_guard<std::mutex> lock(g_io_mutex);
    std::cout << "Thread " << thread_id << ": wait succeeded" << " (" << now << ")" << std::endl;
  }

  // Sleep 1 second to simulate data processing.
  std::this_thread::sleep_for(std::chrono::seconds(1));

  g_semaphore.Signal();
}

信號量自己是一個全局對象，count 爲 1，一次只容許一個線程訪問：

Semaphore g_semaphore(1);

輸出爲：

Thread 1d38: wait succeeded (13:10:10)
Thread 20f4: wait succeeded (13:10:11)
Thread 2348: wait succeeded (13:10:12)

可見每一個線程相隔一秒，即一次只容許一個線程訪問。若是把 count 改成 3：

Semaphore g_semaphore(3);

那麼三個線程輸出的時間應該同樣：

Thread 19f8: wait succeeded (13:10:57)
Thread 2030: wait succeeded (13:10:57)
Thread 199c: wait succeeded (13:10:57)

最後附上 FormatTimeNow 函數的實現：

std::string FormatTimeNow(const char* format) {
  auto now = std::chrono::system_clock::now();
  std::time_t now_c = std::chrono::system_clock::to_time_t(now);
  std::tm* now_tm = std::localtime(&now_c);

  char buf[20];
  std::strftime(buf, sizeof(buf), format, now_tm);
  return std::string(buf);
}

參考：

• C++0x has no semaphores? How to synchronize threads?
• Semaphore (programming)