在无锁实现中没有互斥锁的条件变量

Posted 2023-02-22

【中文标题】在无锁实现中没有互斥锁的条件变量

【英文标题】：condition_variable without mutex in a lock-free implementation

【发布时间】：2015-10-04 16:10:59

【问题描述】：

我有一个使用 std::atomics 以类似于 Herb Sutters CPPCon2014 谈话的方式实现的无锁单生产者多消费者队列。

有时，生产者的速度太慢，无法满足所有消费者的需求，因此消费者可能会挨饿。我想防止饥饿的消费者排队等候，因此我为10ms 添加了睡眠。这个值是任意的，不是最优的。我想使用一个信号，一旦队列中再次有空闲插槽，消费者就可以发送给生产者。在基于锁的实现中，我自然会使用std::condition_variable 来完成这项任务。但是现在在我的无锁实现中，我不确定引入mutex 是否是正确的设计选择，只能使用std::condition_variable。

我只是想问你，在这种情况下，mutex 是否正确？

编辑：我有一个从不睡觉的制作人。并且有多个消费者，如果他们饿了就去睡觉。因此整个系统总是在进步，因此我认为它是无锁的。 我目前的解决方案是在消费者 GetData 函数中做到这一点：std::unique_lock<std::mutex> lk(_idleMutex); _readSetAvailableCV.wait(lk);

一旦新数据准备好，这在生产者线程中：_readSetAvailableCV.notify_all();

【问题讨论】：

一个带有信号的睡眠线程不是无锁的，几乎按照定义。你明白无锁主要是为了保证线程调度和进度，而不是“更快”，对吧？您到底需要什么无锁保证？

我想我不明白。如果是消费者挨饿，为什么队列中有空闲槽时他们会通知生产者？

但是，如果您确实需要一个具有多个写入器的变量，这意味着至少有一个写入器准备好使用更新，这听起来像是一个原子标志。

@Lorehead OP 想要一个无锁队列......但有锁可以阻止空闲读者旋转。并且想知道如何在没有锁的情况下做到这一点。

@BenjaminMenkuec 您应该使用互斥锁，因为互斥锁通过取消调度竞争线程来最小化争用。原子和无锁算法无法避免争用惩罚。

【参考方案1】：

如果您的大部分线程只是等待生产者将资源排入队列，我不确定无锁实现是否值得。大多数时候，你的线程会休眠，它们不会互相争夺队列锁。

这就是为什么我认为（根据您提供的数据量），将所有内容更改为使用 mutex + conditional_variable 就可以了。当生产者将资源加入队列时，它只通知一个线程（notify_one()）并释放队列锁。如果队列再次为空，则锁定队列的消费者将资源出列并返回睡眠状态。线程之间不应该有任何真正的“摩擦”（如果你的生产者很慢），所以我会这样做。

【讨论】：

我同意你的看法。在我用互斥锁实现它之前，线程之间没有太大的摩擦。但是我想将其转换为无锁设计，只是为了练习:) 我测量了速度，它与以前大致相同。 PS：根据硬盘的系统和速度，有时消费者更快，有时生产者更快

让代码变得更糟似乎不是有用的做法。

【参考方案2】：

我刚刚看了这个关于并发 TS 的 CPPCON 视频： Artur Laksberg @cppcon2015

在本次演讲的中间，Artur 解释了我的问题究竟是如何通过障碍和闩锁解决的。他还以我的方式展示了使用 condition_variable 的现有解决方法。他强调了用于此目的的 condition_variable 的一些弱点，例如虚假唤醒和在您进入等待之前缺少通知信号。 但是在我的应用程序中，这些限制没有问题，所以我认为现在，我将使用我在帖子编辑中提到的解决方案 - 直到闩锁/障碍可用。 谢谢大家的评论。

【参考方案3】：

只需对现有的设计进行最少的更改，您就可以简单地使用信号量。信号量开始为空，每次产品推送到队列时都会增加。消费者在从队列中弹出之前首先尝试关闭信号量。

C++11 不提供信号量实现，但可以使用互斥锁、条件变量和计数器来模拟。^†

如果你真的想要在生产者比消费者快的时候实现无锁行为，你可以使用双重检查锁定。

/* producer */
bool was_empty = q.empty_lock_free();
q.push_lock_free(x);
if (was_empty) 
    scoped_lock l(q.lock());
    if (!q.empty()) 
        q.cond().signal();
    


/* consumers */
for (;;) 
    if (q.empty_lock_free()) 
        scoped_lock l(q.lock());
        while (q.empty()) 
            q.cond().wait();
        
        x = q.pop();
        if (!q.empty()) 
            q.cond().signal();
        
     else 
        try 
            x = q.pop_lock_free();
         catch (empty_exception) 
            continue;
        
        break;
    

【参考方案4】：

pthreads 的一种可能性是，一个饥饿的线程用pause() 休眠并用SIGCONT 唤醒。每个线程都有自己的awake 标志。如果生产者发布新输入时任何线程处于休眠状态，请使用pthread_kill() 唤醒一个。

【讨论】：

C++11 等价物是condition_variable::notify_one()。