C++ 信号量（半*无锁*），在哪里可以获得？

Question

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

信号量（如在 Linux 中）是一个有用的构建块，但在 C++ 标准中没有找到，在 boost（目前）中也没有。我主要讨论的是抢占式调度程序中单个进程的线程之间的信号量。

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

信号量不在标准中是否有原因？

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

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

哪些库支持类似的东西？

是否可以通过 windows api 和其他广泛的系统来实现它？

编辑：不可能使用原子+互斥体+条件变量来实现无锁 - 您要么必须在发布中阻塞，要么在等待中旋转。如果您想要无锁的 post()，则不能在 post() 中锁定互斥体。我想在可能抢占式的调度程序上运行，并且我不希望 post() 被其他获取互斥锁并被抢占的线程阻塞。那么，这不是像C++0x has no semaphores 这样的问题的重复吗？如何同步线程？

edit2：下面的示例实现只是为了演示使用atomics+mutex+condvar可以完成的最佳操作，据我所知。post() 和 wait() 执行一次无锁比较交换，并且只有在必须时，它们才会锁定互斥锁。

然而 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”机制（不需要为unpark 锁定），您就可以执行此操作。Park 是指允许线程进入睡眠状态（操作系统阻塞），unpark 是指唤醒该线程。

您已经接近原子计数器和 condvar 方法。问题在于 condvar 互斥锁是语义的一部分。所以你必须放弃 condvars 并进入更低的水平。首先，您应该将所有状态（例如当前信号量值、是否有任何等待者（以及可能有多少个 wwaiter））打包到单个原子值中，并通过比较和交换以原子方式操作它。如果您将这些作为单独的值，这可以防止发生竞争。

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

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

然而，这里是一种机制的草图，该机制在无锁队列中使用每线程等待者对象。信号量由一个原子操作的控制字和一个具有元素类型或堆栈的无锁列表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 以原子方式进行转换，然后使用 withfutex()对FUTEX_WAIT控制字进行第二次检查并以原子方式进行块（这种原子检查和睡眠是 的力量futex）。