2

从这个问题我知道我可以epoll_ctl(2)在另一个线程阻塞时调用epoll_wait(2)。不过我还有一个问题。

当epoll与EPOLLONESHOT标志一起使用时，只会触发一个事件，并且 fd 必须使用epoll_ctl(2). 这是必要的，因此只有一个线程将从 fd 中读取并适当地处理结果。

以下是在某种程度上可视化我假设的问题的时间线：

Thread1:                       Thread2:                  Kernel:
-----------------------------------------------------------------------
epoll_wait();
                                                         Receives chunk
dispatch chunk to thread 2
epoll_wait();                  Handle chunk
                               Still handle chunk        Receives chunk
                               Rearm fd for epoll
?

当 fd 在接收到一个块后重新配置时，问号上会发生什么？会epoll触发一个EPOLLIN事件，还是会无限期地阻塞，尽管套接字是可读的？我的架构完全合理吗？

4

1 回答 1

3

您的架构是明智的，它会起作用：epoll将文件描述符标记为可读并触发EPOLLIN事件。

这方面的文档很少而且很微妙。的 Q/A 部分man 7 epoll简要提到了这一点：

Q8 对文件描述符的操作是否会影响已收集但尚未报告的事件？

A8 您可以对现有文件描述符执行两个操作。对于这种情况，删除将毫无意义。修改将重新读取可用的 I/O。

您可以对现有文件描述符（现有文件描述符是过去已添加到 epoll 集中的文件描述符 - 这包括等待重新配置的文件描述符）执行的两个操作是删除和修改。正如手册页所述， delete 在这里毫无意义，而 modify 将重新评估文件描述符中的条件。

不过，没有什么能比得上真实世界的实验。以下程序测试这种边缘情况：

#include <stdio.h>
#include <pthread.h>
#include <signal.h>
#include <stdlib.h>
#include <assert.h>
#include <semaphore.h>
#include <sys/epoll.h>
#include <unistd.h>

static pthread_t tids[2];
static int epoll_fd;
static char input_buff[512];
static sem_t chunks_sem;

void *dispatcher(void *arg) {
    struct epoll_event epevent;

    while (1) {
        printf("Dispatcher waiting for more chunks\n");
        if (epoll_wait(epoll_fd, &epevent, 1, -1) < 0) {
            perror("epoll_wait(2) error");
            exit(EXIT_FAILURE);
        }

        ssize_t n;
        if ((n = read(STDIN_FILENO, input_buff, sizeof(input_buff)-1)) <= 0) {
            if (n < 0)
                perror("read(2) error");
            else
                fprintf(stderr, "stdin closed prematurely\n");
            exit(EXIT_FAILURE);
        }

        input_buff[n] = '\0';
        sem_post(&chunks_sem);
    }

    return NULL;
}

void *consumer(void *arg) {
    sigset_t smask;
    sigemptyset(&smask);
    sigaddset(&smask, SIGUSR1);

    while (1) {
        sem_wait(&chunks_sem);
        printf("Consumer received chunk: %s", input_buff);
        /* Simulate some processing... */
        sleep(2);
        printf("Consumer finished processing chunk.\n");
        printf("Please send SIGUSR1 after sending more data to stdin\n");

        int signo;
        if (sigwait(&smask, &signo) < 0) {
            perror("sigwait(3) error");
            exit(EXIT_FAILURE);
        }

        assert(signo == SIGUSR1);

        struct epoll_event epevent;
        epevent.events = EPOLLIN | EPOLLONESHOT;
        epevent.data.fd = STDIN_FILENO;

        if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, STDIN_FILENO, &epevent) < 0) {
            perror("epoll_ctl(2) error when attempting to readd stdin");
            exit(EXIT_FAILURE);
        }

        printf("Readded stdin to epoll fd\n");
    }
}

int main(void) {

    sigset_t sigmask;
    sigfillset(&sigmask);
    if (pthread_sigmask(SIG_SETMASK, &sigmask, NULL) < 0) {
        perror("pthread_sigmask(3) error");
        exit(EXIT_FAILURE);
    }

    if ((epoll_fd = epoll_create(1)) < 0) {
        perror("epoll_create(2) error");
        exit(EXIT_FAILURE);
    }

    struct epoll_event epevent;
    epevent.events = EPOLLIN | EPOLLONESHOT;
    epevent.data.fd = STDIN_FILENO;

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, STDIN_FILENO, &epevent) < 0) {
        perror("epoll_ctl(2) error");
        exit(EXIT_FAILURE);
    }

    if (sem_init(&chunks_sem, 0, 0) < 0) {
        perror("sem_init(3) error");
        exit(EXIT_FAILURE);
    }

    if (pthread_create(&tids[0], NULL, dispatcher, NULL) < 0) {
        perror("pthread_create(3) error on dispatcher");
        exit(EXIT_FAILURE);
    }

    if (pthread_create(&tids[1], NULL, consumer, NULL) < 0) {
        perror("pthread_create(3) error on consumer");
        exit(EXIT_FAILURE);
    }

    size_t i;
    for (i = 0; i < sizeof(tids)/sizeof(tids[0]); i++) {
        if (pthread_join(tids[i], NULL) < 0) {
            perror("pthread_join(3) error");
            exit(EXIT_FAILURE);
        }
    }

    return 0;
}

它的工作原理如下：调度程序线程添加stdin到一个 epoll 集，然后在它变得可读时使用它epoll_wait(2)来获取输入。stdin当输入到达时，调度程序唤醒工作线程，工作线程打印输入并通过休眠 2 秒来模拟一些处理时间。与此同时，调度程序返回主循环并epoll_wait(2)再次阻塞。

stdin在您通过发送它告诉它之前，工作线程不会重新武装SIGUSR1。因此，我们只需将更多内容写入stdin，然后发送SIGUSR1到进程。工作线程接收到信号，然后才重新配置stdin- 到那时它已经可读，并且调度程序已经在等待epoll_wait(2)。

您可以从输出中看到调度程序已正确唤醒，并且一切都像魅力一样工作：

Dispatcher waiting for more chunks
testing 1 2 3 // Input
Dispatcher waiting for more chunks // Dispatcher notified worker and is waiting again
Consumer received chunk: testing 1 2 3
Consumer finished processing chunk.
Please send SIGUSR1 after sending more data to stdin
hello world // Input
Readded stdin to epoll fd // Rearm stdin; dispatcher is already waiting
Dispatcher waiting for more chunks // Dispatcher saw new input and is now waiting again
Consumer received chunk: hello world
Consumer finished processing chunk.
Please send SIGUSR1 after sending more data to stdin
于 2015-07-21T11:56:26.600 回答