C++ 信号量 (semi *lockfree*)，我在哪里可以得到一个？

Question

编辑：这不是任何允许在 post() 中锁定互斥锁的问题的重复。请仔细阅读，我需要一个无锁的 post()！如果您没有真正的答案，请勿标记此重复项。

信号量（就像在 linux 中一样）是一个有用的构建块，在 c++ 标准中找不到，在 boost 中也没有（目前）。我主要谈论的是单个进程的线程之间的信号量，通过抢占式调度程序。

我对它们是非阻塞（即无锁）特别感兴趣，除非它确实需要阻塞。也就是说， post() 和 try_wait() 应该始终是无锁的。如果在返回足够多的 post() 之后，它们的调用强烈发生，则 wait() 调用应该是无锁的。 此外，调度程序应该阻止阻塞等待（）而不是自旋锁定。 如果我还想要一个带有超时的 wait_for 怎么办 - 它会使实现进一步复杂化，同时仍然避免饥饿？

信号量不在标准中的原因有哪些？

Edit3：所以，我不知道有一个针对标准 P0514R4 的提案可以准确处理这些问题，并且除了专门添加一个 std::semaphore 之外，还有针对此处提出的所有问题的解决方案。 http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2018/p0514r4.pdf

boost 也没有这些。具体来说，进程间的那些是自旋锁定的。

哪些库支持这样的功能？

是否可以在 windows api 和其他广泛使用的系统上实现它？

编辑：使用 atomics+mutex+condition_variable 来实现无锁是不可能的——你要么在 post 中阻塞，要么在等待中旋转。如果你想要一个无锁的 post()，你不能在 post() 中锁定一个互斥锁。我想在一个可能抢占式调度程序上运行，并且我不希望 post() 被其他使用互斥锁并被抢占的线程阻塞。 所以，这不是重复像C++0x has no semaphores? How to synchronize threads?

这样的问题

编辑2： 以下示例实现只是为了演示使用 atomics+mutex+condvar、AFAIK 可以完成的最佳操作。 post() 和 wait() 执行一次无锁 compare_exchange，并且只有在必须时才锁定互斥体。

但是 post() 不是无锁的。更糟糕的是，它可能会被锁定互斥体并被抢占的 wait() 阻塞。

为简单起见，我只实现了 post_one() 和 wait_one_for(Duration)，而不是 post(int) 和 wait_for(int,Duration)。此外，我假设标准没有承诺没有虚假唤醒。

class semaphore //provides acquire release memory ordering for the user
{
private:
    using mutex_t = std::mutex;
    using unique_lock_t = std::unique_lock<mutex_t>;
    using condvar_t = std::condition_variable;
    using counter_t = int;

    std::atomic<counter_t> atomic_count_; 
    mutex_t mutex_;
    condvar_t condvar_;
    counter_t posts_notified_pending_;
    counter_t posts_unnotified_pending_;
    counter_t waiters_running_;
    counter_t waiters_aborted_pending_;

public:
    void post_one()
    {
        counter_t start_count = atomic_count_.fetch_add(+1, mo_acq_rel);
        if (start_count < 0) {
            unique_lock_t lock(mutex_);
            if (0 < waiters_running_) {
                ++posts_notified_pending_;
                condvar_.notify_one();
            }
            else {
                if (0 == waiters_aborted_pending_) {
                    ++posts_unnotified_pending_;
                }
                else {
                    --waiters_aborted_pending_;
                }
            }
        }
    }

    template< typename Duration >
    bool wait_one_for(Duration timeout)
    {
        counter_t start_count = atomic_count_.fetch_add(-1, mo_acq_rel);
        if (start_count <= 0) {
            unique_lock_t a_lock(mutex_);

            ++waiters_running_;
            BOOST_SCOPE_EXIT(&waiters_running_) {
                --waiters_running_;
            } BOOST_SCOPE_EXIT_END

            if( ( 0 == posts_notified_pending_ ) && ( 0 < posts_unnotified_pending_ ) ) {
                --posts_unnotified_pending_;
                return true;
            }
            else {

                auto wait_result = condvar_.wait_for( a_lock, timeout);
                switch (wait_result) {
                case std::cv_status::no_timeout: {
                    --posts_notified_pending_;
                    return true;
                } break;
                case std::cv_status::timeout: {

                    counter_t abort_count = atomic_count_.fetch_add(+1, mo_acq_rel);
                    if (abort_count >= 0) {
                        /*too many post() already increased a negative atomic_count_ and will try to notify, let them know we aborted. */
                        ++waiters_aborted_pending_;
                    }

                    return false;
                } break;
                default: assert(false); return false;
                }
            }
        }
        return true;
    }


    bool try_wait_one()
    {
        counter_t count = atomic_count_.load( mo_acquire );
        while (true) {
            if (count <= 0) {
                return false;
            }
            else if (atomic_count_.compare_exchange_weak(count, count-1, mo_acq_rel, mo_relaxed )) {
                return true;
            }
        }
    }
};

Answer 1

是的，只要您的操作系统提供合适的“停放”和“取消停放”机制，您就可以执行此操作，无需为取消停放而锁定。 Park 是指允许线程进入睡眠状态（操作系统阻塞），而 unpark 是指唤醒该线程。

您的原子计数器和 condvar 方法已经很接近了。问题是 condvar a mutex 作为语义的一部分是必需的。因此，您必须放弃 condvars 并降低水平。首先，您应该将所有状态（例如当前信号量值、是否有任何等待者（可能还有多少等待者））打包成一个原子值，并通过比较和交换来原子地操作它。如果您将这些作为单独的值，这可以防止发生争用。

然后您可以绘制一个状态图，显示信号量的所有可能状态，以及所有可能的转换状态的边（例如，当服务员到达时，“没有服务员”状态将转换为“有服务员”状态）。您使用比较和交换来实现所有转换，并且当它失败时，您必须重新计算转换，因为它可能已经改变！

那么你只需要实现阻塞。在 Windows 上，您将使用 Events - 自动或手动重置。两者都有其优点和怪癖，并且有不止一种方法可以给这只猫剥皮。例如，您可能可以让它与单个共享事件和自动重置事件一起使用。

然而，这里有一个机制的草图，它在无锁队列中使用每个线程的等待者对象。信号量由一个原子操作的控制字和一个元素类型为 waiter_node 或堆栈或任何你想使用的现成的类似并发列表的东西组成的无锁列表。

我们将假设每个线程都拥有一个 waiter_node 对象，该对象只包含一个手动重置事件对象。这可以创建一次并存储在 TLS 中（可能是最有效的），或者在每次需要发生等待时按需分配并在等待完成时取消分配。

这是基本大纲：

等待

• 如果信号量可用（正），CAS 将其递减并继续。
• 如果信号量不可用（零），线程在其waiter_node 上调用ResetEvent，然后将事件推送到服务员列表中，检查sem 值是否仍然为零，然后在其上调用WaitForObject waiter_node。当它返回时，从顶部开始等待例程。

发布

• 增加控制字。弹出waiter_node（如果有），然后调用SetEvent。

这里有各种各样的“竞争”，例如 waiter_node 在等待线程甚至休眠之前被 post 操作弹出，但它们应该是良性的。

即使在这种基于服务员队列的设计上也有许多变体。例如，您可以将列表“head”和控制词整合在一起，因此它们是同一个东西。然后wait 不需要仔细检查信号量计数，因为推送操作同时验证信号量状态。您还可以实现“直接切换”，如果有服务员，posting 线程根本不会增加控制字，而只是弹出一个并用它已成功获取信号量的信息唤醒它。

在 Linux 上，您将 Event 替换为 futex。因为futex 允许在内核内部进行原子检查和阻止操作，从而避免Event 解决方案中固有的许多竞争，所以在那里实现“单一futex”解决方案更容易。因此，基本草图是单个控制字，您使用 CAS 原子地进行转换，然后使用 futex() 和 FUTEX_WAIT 对控制字进行第二次检查并原子地阻塞（这个原子检查和睡眠是futex的力量。