在线程之间共享相同的epoll文件描述符可以吗？

Question

在多个线程之间共享相同的Epoll fd (而不是套接字fd)安全吗？如果是这样的话，每个线程是否必须将自己的事件数组传递给epoll_wait(2)，或者它们可以共享它？

例如

    void *thread_func(void *thread_args) {
      // extract socket_fd, epoll_fd, &event, &events_array from 
      //     thread_args
      // epoll_wait() using epoll_fd and events_array received from main
      // now all threads would be using same epoll_fd and events array 
    }

    void main( void ) {
      // create and bind to socket
      // create events_fd
      // allocate memory for events array
      // subscribe to events EPOLLIN and EPOLLET
      // pack the socket_fd, epoll_fd, &events, &events_array into 
      //   thread_args struct.

      // create multiple threads and pass thread_func and 
      //   same thread_args to all threads
    }

还是这样做更好：

    void *thread_func(void *socket_fd) {
      // create events_fd
      // allocate memory for events array
      // subscribe to events EPOLLIN and EPOLLET
      // epoll_wait using own epoll_fd and events_array
      // now all threads would have a separate epoll_fd with 
      //   events populated on its own array
   }

   void main(void) {
     // create and bind to socket

     //create multiple threads and pass thread_func and socket_fd to 
     //  all threads
   }

在C语言中有如何做到这一点的好例子吗？我看到的示例在main()中运行事件循环，并在检测到事件时生成一个新线程来处理请求。我想要做的是在程序开始时创建一个特定数量的线程，让每个线程运行事件循环并处理请求。

Answer 1

在多个线程之间共享相同的Epoll fd (不是套接字fd)安全吗？

是的，它是安全的-- epoll(7)接口是线程安全的--但是在这样做时应该小心，至少应该使用EPOLLET (边缘触发模式，而不是默认级别触发模式)来避免其他线程中的虚假唤醒。这是因为当一个新事件可用处理时，级别触发模式会唤醒每个线程。因为只有一个线程将处理它，这将不必要地唤醒大多数线程。

如果使用共享epfd，那么每个线程都必须将自己的事件数组或共享事件数组传递给epoll_wait()

是的，您需要在每个线程上设置一个单独的事件数组，否则就会出现争用条件，并且会发生一些不愉快的事情。例如，您可能有一个线程仍在迭代epoll_wait(2)返回的事件并处理请求时，突然间另一个线程用相同的数组调用epoll_wait(2)，然后事件在另一个线程读取它们的同时被覆盖。不太好！对于每个线程，您绝对需要一个单独的数组。

假设每个线程都有一个单独的数组，那么两种可能性--等待相同的epoll fd或对每个线程都有单独的epoll fd --将同样工作得很好，但是请注意语义是不同的。对于全局共享的epoll fd，每个线程都等待来自任何客户端的请求，因为客户端都被添加到同一个epoll fd中。对于每个线程有一个单独的epoll fd，那么每个线程基本上负责一个客户端子集(那些被该线程接受的客户端)。

这可能与您的系统无关，也可能会产生巨大的差异。例如，可能会发生这样的情况:线程不幸地得到了一组发出大量频繁请求的高级用户，从而导致线程过度工作，而其他具有不太活跃客户端的线程则几乎处于空闲状态。这不是不公平吗？另一方面，您可能希望只使用一些线程来处理特定的用户类，在这种情况下，在每个线程上设置不同的epoll fds可能是有意义的。与往常一样，您需要考虑这两种可能性，评估权衡，考虑您的具体问题，并作出决定。

下面是一个使用全局共享epoll fd的示例。我本来不打算这么做，但是一件事导致了另一件事，而且，嗯，这很有趣，我认为它可能会帮助你开始工作。它是一个回声服务器，在端口3000上侦听，并且有一个由20个线程组成的池，使用epoll并发地接受新客户端并服务请求。

#include <stdio.h>
#include <stdlib.h>
#include <inttypes.h>
#include <errno.h>
#include <string.h>
#include <pthread.h>
#include <assert.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <sys/epoll.h>

#define SERVERPORT 3000
#define SERVERBACKLOG 10
#define THREADSNO 20
#define EVENTS_BUFF_SZ 256

static int serversock;
static int epoll_fd;
static pthread_t threads[THREADSNO];

int accept_new_client(void) {

    int clientsock;
    struct sockaddr_in addr;
    socklen_t addrlen = sizeof(addr);
    if ((clientsock = accept(serversock, (struct sockaddr *) &addr, &addrlen)) < 0) {
        return -1;
    }

    char ip_buff[INET_ADDRSTRLEN+1];
    if (inet_ntop(AF_INET, &addr.sin_addr, ip_buff, sizeof(ip_buff)) == NULL) {
        close(clientsock);
        return -1;
    }

    printf("*** [%p] Client connected from %s:%" PRIu16 "\n", (void *) pthread_self(),
           ip_buff, ntohs(addr.sin_port));

    struct epoll_event epevent;
    epevent.events = EPOLLIN | EPOLLET;
    epevent.data.fd = clientsock;

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, clientsock, &epevent) < 0) {
        perror("epoll_ctl(2) failed attempting to add new client");
        close(clientsock);
        return -1;
    }

    return 0;
}

int handle_request(int clientfd) {
    char readbuff[512];
    struct sockaddr_in addr;
    socklen_t addrlen = sizeof(addr);
    ssize_t n;

    if ((n = recv(clientfd, readbuff, sizeof(readbuff)-1, 0)) < 0) {
        return -1;
    }

    if (n == 0) {
        return 0;
    }

    readbuff[n] = '\0';

    if (getpeername(clientfd, (struct sockaddr *) &addr, &addrlen) < 0) {
        return -1;
    }

    char ip_buff[INET_ADDRSTRLEN+1];
    if (inet_ntop(AF_INET, &addr.sin_addr, ip_buff, sizeof(ip_buff)) == NULL) {
        return -1;
    }

    printf("*** [%p] [%s:%" PRIu16 "] -> server: %s", (void *) pthread_self(),
           ip_buff, ntohs(addr.sin_port), readbuff);

    ssize_t sent;
    if ((sent = send(clientfd, readbuff, n, 0)) < 0) {
        return -1;
    }

    readbuff[sent] = '\0';

    printf("*** [%p] server -> [%s:%" PRIu16 "]: %s", (void *) pthread_self(),
           ip_buff, ntohs(addr.sin_port), readbuff);

    return 0;
}

void *worker_thr(void *args) {
    struct epoll_event *events = malloc(sizeof(*events)*EVENTS_BUFF_SZ);
    if (events == NULL) {
        perror("malloc(3) failed when attempting to allocate events buffer");
        pthread_exit(NULL);
    }

    int events_cnt;
    while ((events_cnt = epoll_wait(epoll_fd, events, EVENTS_BUFF_SZ, -1)) > 0) {
        int i;
        for (i = 0; i < events_cnt; i++) {
            assert(events[i].events & EPOLLIN);

            if (events[i].data.fd == serversock) {
                if (accept_new_client() == -1) {
                    fprintf(stderr, "Error accepting new client: %s\n",
                        strerror(errno));
                }
            } else {
                if (handle_request(events[i].data.fd) == -1) {
                    fprintf(stderr, "Error handling request: %s\n",
                        strerror(errno));
                }
            }
        }
    }

    if (events_cnt == 0) {
        fprintf(stderr, "epoll_wait(2) returned 0, but timeout was not specified...?");
    } else {
        perror("epoll_wait(2) error");
    }

    free(events);

    return NULL;
}

int main(void) {
    if ((serversock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) {
        perror("socket(2) failed");
        exit(EXIT_FAILURE);
    }

    struct sockaddr_in serveraddr;
    serveraddr.sin_family = AF_INET;
    serveraddr.sin_port = htons(SERVERPORT);
    serveraddr.sin_addr.s_addr = INADDR_ANY;

    if (bind(serversock, (const struct sockaddr *) &serveraddr, sizeof(serveraddr)) < 0) {
        perror("bind(2) failed");
        exit(EXIT_FAILURE);
    }

    if (listen(serversock, SERVERBACKLOG) < 0) {
        perror("listen(2) failed");
        exit(EXIT_FAILURE);
    }

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

    struct epoll_event epevent;
    epevent.events = EPOLLIN | EPOLLET;
    epevent.data.fd = serversock;

    if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, serversock, &epevent) < 0) {
        perror("epoll_ctl(2) failed on main server socket");
        exit(EXIT_FAILURE);
    }

    int i;
    for (i = 0; i < THREADSNO; i++) {
        if (pthread_create(&threads[i], NULL, worker_thr, NULL) < 0) {
            perror("pthread_create(3) failed");
            exit(EXIT_FAILURE);
        }
    }

    /* main thread also contributes as worker thread */
    worker_thr(NULL);

    return 0;
}

几个注意事项：

• main()应该返回int，而不是void (如您在示例中所示)
• 始终处理错误返回代码。忽略它们是很常见的，当事情破裂时，很难知道发生了什么。
• 代码假定没有任何请求大于511字节(如handle_request()中的缓冲区大小所示)。如果请求大于此，则可能会在套接字中保留一些数据很长时间，因为epoll_wait(2)在该文件描述符上发生新事件之前不会报告它(因为我们使用的是EPOLLET)。在最坏的情况下，客户端可能永远不会发送任何新的数据，永远等待回复。
• 打印每个请求的线程标识符的代码假定pthread_t是不透明指针类型。实际上，pthread_t是Linux中的指针类型，但在其他平台上可能是整数类型，因此这是不可移植的。但是，这可能不是什么大问题，因为epoll是Linux特定的，所以代码是不可移植的。
• 它假定，当线程仍在为来自该客户端的请求提供服务时，来自同一客户端的其他请求不会到达。如果同时到达一个新请求，而另一个线程开始为其服务，则我们有一个争用条件，客户端将不一定按照他发送的相同顺序接收回送消息(然而，write(2)是原子的，因此虽然答复可能不正常，但它们不会中断)。