在多线程编程中，实现线程的等待执行机制很重要，它们必须实现排斥等待以及访问一个资源，也就是在无法工作的时候进行等待。有一种方式实现线程等待----也就是使用信号量，即让相处在内核中睡眠，因此它们不再占用CPU时间。

我曾以为信号量是一个很古老的东西，它是在20世纪60年代，被Edsger Dijkstra所发明得来，在此之前，程序员都没有做太多的多线程编程，或者根本没有做任何多线程编程。我们知道信号量可以保持追踪一个资源单位的可使用情况，或者充当简陋的互斥量（mutex）。但是，随着不断学习和工作，我的观点也在改变，仅仅使用信号量和原子操作，就可以实现所有以下原语：

一个轻量级的互斥量（Mutex）一个轻量级的自动重置事件对象一个轻量级读写锁哲学家吃饭问题的另一种解法再由部分自旋的轻量级信号量

不仅如此，这些实现还具有一些理想的属性，它们都是轻量级的，某种意义上来说，它们都可以在用户空间实现完整操作，然后自旋很短时间进入内核中睡眠。我会把上述所实现的完整源码放在文末的链接里面，欢迎大家下载研究。因为C++11标准库中没有信号量，我同样提供了一个便携的Semaphore类，来直接映射到Windows、MacOS、iOS、Linux和其他POSIX系统环境中的信号量。您应该能够将所有这些原语放入几乎所有现有的C ++ 11项目中。

1.信号量就像一个保镖

想象一组排队等待的线程，就像排队进入商场或者酒吧的人群一样，信号量就像队列前的保镖，仅在有指示的情况下才允许线程继续前进。

图1

每个线程决定它们是否加入到队列中，Dijkstra将此称为P操作。P最初代表一些听起来很有趣的荷兰语单词，但是在现代的信号量实现中，您更有可能看到称为“等待（wait）”的操作。 基本上，当线程调用信号量的等待操作时，它会进入队列。

作为保镖，信号量只需要理解一个基本的指令，一开始，Dijkstra称这个指令为V操作。如今，这个操作有着不同的名称，例如：post，release，或者signal。任何正在运行的线程都可以随时调用signal，并且当它退出时，保镖会从队列中释放一个正等待的线程。 （不一定按它们到达的顺序）现在，如果有一些线程调用在没有任何线程等待排队之前发出信号怎么办？ 没问题：下一个线程进入阵容时，弹跳器将使其直接通过。 如果在空的队列中调用了信号3次，则保镖将使接下来到达的3个线程直接通过。

图2

当然，保镖需要跟踪该数字，这就是为什么所有信号量都保持整数计数器的原因。 信号（signal）递增计数器，然后等待（wait）递减。这种策略的优点在于，如果多次调用特定数量的wait，并且接着多次调用特定数量的signal，结果总是相同的：保镖将始终释放相同数量的线程，并且始终具有相同的线程数。 排队等待的线程数，无论这些等待和信号调用发生的顺序如何。

图3

2.一个轻量级互斥量（Mutex）

在以前我也写过相关信号量的文章，当时我还不知道，但是那篇文章只是可重用模式的一个例子。 诀窍是在信号灯的前面建立另一种机制，我称之为工作箱。

图4

工作箱是做出真正决定的地方。 当前线程应该排队吗？ 是否应该完全绕过队列？ 是否应该从队列中释放另一个线程？ 工作箱无法直接检查正在等待信号量的线程数，也无法检查信号量的当前信号计数。 相反，工作箱必须以某种方式跟踪自己的先前决定。 对于轻量级的互斥锁，它所需要的只是一个原子计数器。 我将其称为m_contention计数器，因为它可以跟踪同时争用互斥量的线程数。

class LightweightMutex{private:    std::atomic<int> m_contention;         // 工作箱    Semaphore m_semaphore;                 // 保镖

当一个线程决定释放互斥量，它首先访问工作箱来增加m_contention计数器。

public:    void lock()    {        if (m_contention.fetch_add(1, std::memory_order_acquire) > 0)  // 访问工作箱        {            m_semaphore.wait();     // 进入等待队列        }    }

如果先前的值为0，则意味着该互斥锁尚未竞争其他线程。 这样，当前线程会立即将自己视为新所有者，绕过该信号量，从锁中返回并进入互斥量旨在保护的任何代码。否则，如果先前的值大于0，则意味着已经考虑另一个线程拥有互斥量。 在这种情况下，当前线程必须排队等候。

图5

当前一个线程释放互斥量，它会访问工作箱来减少计数器：

    void unlock()    {        if (m_contention.fetch_sub(1, std::memory_order_release) > 1)  // Visit the box office        {            m_semaphore.signal();   // Release a waiting thread from the queue        }    }

如果先前的计数器值为1，则表示在此期间没有其他线程到达，因此无需执行其他操作。 m_contention仅保留为0。否则，如果先前的计数器值大于1，同时另一个线程试图锁定互斥锁，因此后来的线程正在队列中等待。 因此，我们提醒保镖现在可以安全释放下一个线程。 该线程将被视为新所有者。

图6

每次访问工作箱都是不可分割的，也就是一次原子操作。 因此，即使多个线程同时调用锁定和解锁，它们也总是一次访问一个工作箱。 此外，互斥量的行为完全由工作箱决定。 访问工作箱后，他们可能会以无法预测的顺序操作信号灯，但这没关系。 正如我已经解释的那样，无论这些信号量操作发生的顺序如何，结果都将保持有效。 （在最坏的情况下，某些线程可能会在线交易位置。）

此类被认为是“轻量级”，因为它在没有争用时会绕过信号量，从而避免了系统调用。我已经将其与递归版本一起作为NonRecursiveBenaphore发布到了代码中。但是，无需在实践中使用这些类。大多数可用的互斥量实现已经是轻量级的。

3.一个轻量级的自动重置事件对象

您不会经常听到关于自动重置事件对象的相关讨论，但是正如在CppCon 2014演讲中提到的那样，它们在游戏引擎中得到了广泛使用。 通常，它们用于通知其他线程（可能正在休眠）可用的工作。

图7

自动复位事件对象基本上是忽略冗余信号的信号量。 换句话说，当信号被多次调用时，事件对象的信号计数将永远不会超过1。这意味着您可以继续在某个位置发布工作单元，在每个信号单元之后盲目调用信号。 这是一种灵活的技术，即使您将工作单元发布到除队列之外的某些数据结构中，该技术也可以使用。

Windows具有对事件对象的本地支持，但是它的SetEvent函数（相当于信号）可能会有很昂贵的代价。 一台机器，即使事件已经发出信号，我们将其定为每个呼叫700 ns。 如果您要在线程之间发布数千个工作单元，则每个SetEvent的开销很快就会加起来。

幸运的是，工作箱/保镖模式大大减少了这种开销。 可以使用原子操作在工作箱实现所有自动重置事件逻辑，并且仅在绝对需要线程等待时，工作箱才会调用信号量。

图8

我已将实现发布为AutoResetEvent。 这次，工作箱采用了另一种方式来跟踪已发送了多少个线程以在队列中等待。 当m_status为负数时，其大小指示正在等待多少线程：

class AutoResetEvent{private:    // m_status == 1: Event object is signaled.    // m_status == 0: Event object is reset and no threads are waiting.    // m_status == -N: Event object is reset and N threads are waiting.    std::atomic<int> m_status;    Semaphore m_sema;

在事件对象的信号操作中，我们以原子方式递增m_status，最大限制为1：

public:    void signal()    {        int oldStatus = m_status.load(std::memory_order_relaxed);        for (;;)    // Increment m_status atomically via CAS loop.        {            assert(oldStatus <= 1);            int newStatus = oldStatus < 1 ? oldStatus + 1 : 1;            if (m_status.compare_exchange_weak(oldStatus, newStatus, std::memory_order_release, std::memory_order_relaxed))                break;            // The compare-exchange failed, likely because another thread changed m_status.            // oldStatus has been updated. Retry the CAS loop.        }        if (oldStatus < 0)            m_sema.signal();    // Release one waiting thread.    }

请注意，由于放宽了来自m_status的初始负载，因此即使m_status已经等于1，上述代码也必须调用compare_exchange_weak，这一点很重要。