如何在 epoll 上使用具有级别触发行为的 eventfd？

Question

epoll_ctl当不递减 eventfd 计数器时，注册级别触发的 eventfd只会触发一次。总结这个问题，我观察到 epoll 标志（EPOLLET，EPOLLONESHOT或None用于级别触发行为）的行为相似。或者换句话说：没有效果。

你能确认这个错误吗？

我有一个多线程的应用程序。每个线程等待epoll_wait具有相同 epollfd 的新事件。如果要优雅地终止应用程序，则必须唤醒所有线程。我的想法是您使用 eventfd 计数器 ( EFD_SEMAPHORE|EFD_NONBLOCK) 为此（具有级别触发的 epoll 行为）一起唤醒。（不管少数文件描述符的雷鸣般的羊群问题。）

例如，对于 4 个线程，您将 4 个写入 eventfd。我期待epoll_wait立即一次又一次地返回，直到计数器递减（读取）4 次。epoll_wait每次写入仅返回一次。

是的，我仔细阅读了所有相关手册；)

#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/types.h>
#include <unistd.h>
#include <pthread.h>

static int event_fd = -1;
static int epoll_fd = -1;

void *thread(void *arg)
{
    (void) arg;

    for(;;) {
       struct epoll_event event;
       epoll_wait(epoll_fd, &event, 1, -1);

       /* handle events */
       if(event.data.fd == event_fd && event.events & EPOLLIN) {
           uint64_t val = 0;
           eventfd_read(event_fd, &val);
           break;
       }
    }

    return NULL;
}

int main(void)
{
    epoll_fd = epoll_create1(0);
    event_fd = eventfd(0, EFD_SEMAPHORE| EFD_NONBLOCK);

    struct epoll_event event;
    event.events = EPOLLIN;
    event.data.fd = event_fd;
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, event_fd, &event);

    enum { THREADS = 4 };
    pthread_t thrd[THREADS];

    for (int i = 0; i < THREADS; i++)
        pthread_create(&thrd[i], NULL, &thread, NULL);

    /* let threads park internally (kernel does readiness check before sleeping) */
    usleep(100000);
    eventfd_write(event_fd, THREADS);

    for (int i = 0; i < THREADS; i++)
        pthread_join(thrd[i], NULL);
}

Answer 1

当您写入 时，会调用eventfd一个函数。eventfd_signal它包含以下行来唤醒：

wake_up_locked_poll(&ctx->wqh, EPOLLIN);

作为wake_up_locked_poll一个宏：

#define wake_up_locked_poll(x, m)                       \
    __wake_up_locked_key((x), TASK_NORMAL, poll_to_key(m))

被__wake_up_locked_key定义为：

void __wake_up_locked_key(struct wait_queue_head *wq_head, unsigned int mode, void *key)
{
    __wake_up_common(wq_head, mode, 1, 0, key, NULL);
}

最后，__wake_up_common被声明为：

/*
 * The core wakeup function. Non-exclusive wakeups (nr_exclusive == 0) just
 * wake everything up. If it's an exclusive wakeup (nr_exclusive == small +ve
 * number) then we wake all the non-exclusive tasks and one exclusive task.
 *
 * There are circumstances in which we can try to wake a task which has already
 * started to run but is not in state TASK_RUNNING. try_to_wake_up() returns
 * zero in this (rare) case, and we handle it by continuing to scan the queue.
 */
static int __wake_up_common(struct wait_queue_head *wq_head, unsigned int mode,
            int nr_exclusive, int wake_flags, void *key,
            wait_queue_entry_t *bookmark)

注意这个nr_exclusive参数，你会发现写入一个eventfd只能唤醒一个独占服务员。

独家是什么意思？阅读epoll_ctl手册页给了我们一些见解：

EPOLLEXCLUSIVE（自 Linux 4.5 起）：

为附加到目标文件描述符 fd 的 epoll 文件描述符设置独占唤醒模式。当唤醒事件发生并且多个 epoll 文件描述符使用 附加到同一个目标文件时EPOLLEXCLUSIVE，一个或多个 epoll 文件描述符将接收一个带有 的事件epoll_wait(2)。

添加事件时不使用EPOLLEXCLUSIVE，但要等待epoll_wait每个线程必须将其自身放入等待队列。函数do_epoll_wait通过调用 来执行等待ep_poll。通过下面的代码，您可以看到它将当前线程添加到第 #1903 行的等待队列中：

__add_wait_queue_exclusive(&ep->wq, &wait);

这是对正在发生的事情的解释 - epoll 等待者是独占的，因此只有一个线程被唤醒。此行为已在v2.6.22-rc1中引入，相关更改已在此处讨论。

对我来说，这看起来像是函数中的一个错误：在信号量模式下，它应该执行等于写入值的eventfd_signal唤醒。nr_exclusive

所以你的选择是：

• 为每个线程创建一个单独的 epoll 描述符（可能不适用于您的设计 - 扩展问题）
• 在它周围放置一个互斥体（缩放问题）
• poll可能在eventfdepoll 和 epoll上使用
• 通过写入 1 和evenfd_write4 次来分别唤醒每个线程（可能是您能做的最好的事情）。