在 Linux 上等待多个条件变量而没有不必要的睡眠？

Posted 2023-03-14

【中文标题】在 Linux 上等待多个条件变量而没有不必要的睡眠？

【英文标题】：Wait on multiple condition variables on Linux without unnecessary sleeps?

【发布时间】：2011-02-20 11:04:32

【问题描述】：

我正在编写一个对延迟敏感的应用程序，它实际上想要一次等待多个条件变量。我以前读过在 Linux 上获得此功能的几种方法（显然这是在 Windows 上内置的），但它们似乎都不适合我的应用程序。我知道的方法有：

让一个线程等待您要等待的每个条件变量，当唤醒时将发出一个您等待的单个条件变量的信号。

通过定时等待循环多个条件变量。

改为将虚拟字节写入文件或管道，并轮询它们。

#1 & #2 不适合，因为它们会导致不必要的睡眠。使用#1，您必须等待虚拟线程唤醒，然后向真实线程发出信号，然后让真实线程唤醒，而不是真正的线程刚开始醒来——额外的调度程序量子花费在这对我的应用程序实际上很重要，我不想使用完整的 RTOS。 #2 更糟糕的是，您可能会花费 N * 超时时间入睡，或者您的超时时间将为 0，在这种情况下您永远不会睡觉（无休止地消耗 CPU 和饿死其他线程也很糟糕）。

对于#3，管道是有问题的，因为如果被“发出信号”的线程很忙甚至崩溃（我实际上是在处理单独的进程而不是线程——互斥锁和条件将存储在共享内存中），然后写入线程将被卡住，因为管道的缓冲区将满，任何其他客户端也是如此。文件是有问题的，因为应用运行的时间越长，文件就会无休止地增长。

有没有更好的方法来做到这一点？很好奇适合 Solaris 的答案。

【问题讨论】：

我在 C++0x 线程原语中也遇到了这个限制，它似乎在很大程度上基于 pthreads 最小公分母。

你不能用一个信号量来代替吗？一旦等待线程得到一个单元，它就可以轮询各种源以找到一个已“触发”的源（可能是一组易失性布尔值？）。

【参考方案1】：

您的 #3 选项（将虚拟字节写入文件或管道，然后轮询它们）在 Linux 上有更好的选择：eventfd。

使用eventfd，您有一个内核内无符号 64 位计数器，而不是有限大小的缓冲区（如在管道中）或无限增长的缓冲区（如在文件中）。一个 8 字节的write 向计数器添加一个数字；一个 8 字节的 read 要么将计数器归零并返回其先前的值（不带 EFD_SEMAPHORE），要么将计数器减 1 并返回 1（带 EFD_SEMAPHORE）。当计数器不为零时，文件描述符被认为对轮询函数（select、poll、epoll）是可读的。

即使计数器接近 64 位限制，如果您将文件描述符设为非阻塞，write 也会失败并显示EAGAIN。当计数器为零时，read 也会发生同样的情况。

【讨论】：

太棒了，这有很多用例。

很遗憾 eventfd 不允许“命名”。也就是说，您不能在与句柄共享没有父子关系的进程之间使用 eventfd（或任何其他 *fd 机制）。否则，这些机制将完全取代过时、功能不足、从性能角度来看效率低下的 POSIX 命名废话我们现在拥有（做类似的事情），但不能与其他 fd 等待混合或通过 epoll/poll 等待/选择。

@Michael Goldshteyn：你不能使用通过 Unix 域套接字传递的文件描述符将 eventfd 描述符（或任何其他文件描述符）传递给不相关的进程吗？

@CesarB，是的，这是一个选项，但它为应该是一个简单的功能增加了很多复杂性：基于 fd 的命名事件和计时器。

这会导致每次读取到内核的往返，即使它是“信号的”。与信号量相比，这不是“最佳”方法。

【参考方案2】：

如果您在谈论 POSIX 线程，我建议您使用单个条件变量和事件标志的数量或类似的东西。这个想法是使用对等 condvar 互斥锁来保护事件通知。无论如何，您都需要在 cond_wait() 退出后检查事件。这是我的旧代码，可以从我的培训中说明这一点（是的，我检查了它是否可以运行，但请注意它是不久前准备的，并且急于为新人准备）。

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>

static pthread_cond_t var;
static pthread_mutex_t mtx;

unsigned event_flags = 0;
#define FLAG_EVENT_1    1
#define FLAG_EVENT_2    2

void signal_1()

    pthread_mutex_lock(&mtx);
    event_flags |= FLAG_EVENT_1;
    pthread_cond_signal(&var);
    pthread_mutex_unlock(&mtx);


void signal_2()

    pthread_mutex_lock(&mtx);
    event_flags |= FLAG_EVENT_2;
    pthread_cond_signal(&var);
    pthread_mutex_unlock(&mtx);


void* handler(void*)

    // Mutex is unlocked only when we wait or process received events.
    pthread_mutex_lock(&mtx);

    // Here should be race-condition prevention in real code.

    while(1)
    
        if (event_flags)
        
            unsigned copy = event_flags;

            // We unlock mutex while we are processing received events.
            pthread_mutex_unlock(&mtx);

            if (copy & FLAG_EVENT_1)
            
                printf("EVENT 1\n");
                copy ^= FLAG_EVENT_1;
            

            if (copy & FLAG_EVENT_2)
            
                printf("EVENT 2\n");
                copy ^= FLAG_EVENT_2;

                // And let EVENT 2 to be 'quit' signal.
                // In this case for consistency we break with locked mutex.
                pthread_mutex_lock(&mtx);
                break;
            

            // Note we should have mutex locked at the iteration end.
            pthread_mutex_lock(&mtx);
        
        else
        
            // Mutex is locked. It is unlocked while we are waiting.
            pthread_cond_wait(&var, &mtx);
            // Mutex is locked.
        
    

    // ... as we are dying.
    pthread_mutex_unlock(&mtx);


int main()

    pthread_mutex_init(&mtx, NULL);
    pthread_cond_init(&var, NULL);

    pthread_t id;
    pthread_create(&id, NULL, handler, NULL);
    sleep(1);

    signal_1();
    sleep(1);
    signal_1();
    sleep(1);
    signal_2();
    sleep(1);

    pthread_join(id, NULL);
    return 0;

【讨论】：

这是一个明智的答案，但不幸的是语义不同。例如，如果我轮询一个文件，并且在我醒来之前有 10 个字节写入该文件，那么当我醒来时，我发现已写入 10 个字节。在这个方案下，如果一个事件在我醒来之前发生了十次，我只知道最后一次。

您可以尝试扩展该方案，以便有一个事件标志列表而不是事件标志，并且阅读器线程跟踪它在列表中的位置，但这不能扩展到多线程——你怎么知道什么时候可以删除列表中的元素？为了让它更快，您现在需要一个无锁链表和引用计数实现。可能仍然比等待调度程序量更快，但远非理想......

您可以执行任何事件列表“pop”，而不是复制事件标志。除非您释放 mtx，否则列表不会更改（当然只有在任何修改都在同一个互斥锁下时）。这是该方案的最大优势之一。例如，您可以使用事件队列，是的，这个队列是受保护的。当“读者”检查它并“弹出”时，任何愿意“推动”的人都会等待很短的时间。但请注意，您不应处理被锁定的事件，只能“提取”。

我最终没有使用这个答案，因为它仍然不太适合我的应用程序，但对于问题中可用的信息，这是正确的方法:)

在处理收到的事件后，这个答案是不是缺少pthread_mutex_lock(&mtx);？

【参考方案3】：

如果您希望在 POSIX 条件变量同步模型下获得最大的灵活性，您必须避免编写仅通过公开条件变量来向用户传达事件的模块。 （然后，您基本上重新发明了信号量。）

活动模块的设计应使其接口通过注册函数提供事件的回调通知：并且，如有必要，可以注册多个回调。

多个模块的客户端向每个模块注册一个回调。这些都可以被路由到一个共同的地方，在那里他们锁定同一个互斥体，改变一些状态，解锁，并点击同一个条件变量。

这种设计还提供了这样一种可能性，即如果为响应事件而完成的工作量相当小，也许它可以在回调的上下文中完成。

回调在调试方面也有一些优势。您可以在以回调形式到达的事件上放置断点，并查看其生成方式的调用堆栈。如果您在作为信号量唤醒或通过某种消息传递机制到达的事件上设置断点，则调用跟踪不会显示事件的来源。

---

话虽如此，您可以使用支持等待多个对象的互斥锁和条件变量创建自己的同步原语。这些同步原语可以在内部基于回调，以对应用程序的其余部分不可见的方式。

它的要点是，对于线程想要等待的每个对象，等待操作将与该对象的回调接口排队。当一个对象发出信号时，它会调用其所有注册的回调。被唤醒的线程将所有回调接口出列，并查看每个回调接口中的一些状态标志以查看哪些对象发出信号。

【参考方案4】：

对于等待多个条件变量，有一个 Solaris 实现，如果您有兴趣可以移植到 Linux：WaitFor API