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

Question

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

信号量(如在 linux 中)是一个有用的构建 block ，在 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)。此外，我假设标准没有 promise 没有虚假唤醒。

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”机制，您就可以做到这一点，这不需要为 unpark 锁定。 Park 是指允许线程进入休眠状态(操作系统阻塞)，而 unpark 是指唤醒该线程。

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

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

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

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

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

以下是基本大纲:

等待

如果信号量可用(正)，CAS 将其递减并继续。

如果信号量不可用(零)，线程调用 ResetEvent在其waiter_node然后将事件推送到服务员列表中，检查 sem 值是否仍然为零，然后调用 WaitForObject上面是 waiter_node .当它返回时，从顶部开始等待例程。

邮政

增加控制字。弹出 waiter_node ，如有，请调用 SetEvent在上面。

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

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

在 Linux 上，替换 Event与 futex .从 futex 开始，实现“单一 futex”解决方案更容易。允许在内核内部进行原子检查和阻止操作，从而避免 Event 中固有的许多竞争。解决方案。所以一个基本的草图是一个单一的控制字，你用 CAS 原子地进行转换，然后使用 futex()与 FUTEX_WAIT对控制字进行第二次检查并自动阻止(这种原子检查和 sleep 是 futex 的力量)。