- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

linux epoll的使用

ZhjDayDayUp 发表于 2022/01/17 21:03:10 2022/01/17

【摘要】 epoll是一种I/O多路复用技术，支持高并发

epoll是一种I/O多路复用技术，支持高并发。考虑如下场景：当进程存在大量的tcp连接时（如100万个），但是某一时刻只有少量的活跃连接（如100个），在这种情况下，如何高效地处理呢？

传统的poll和select方法：当事件发生时，会去遍历所有的连接，因为它并不知道哪些连接是真正有事件发生的。

epoll方法：会维护一个红黑表用于存储需要监听的连接，维护一个双向链表用于存储真正发生事件的连接。当事件发生时，只需要遍历此链表，可以提高处理效率。

epoll的API

使用时调用如下三个主要函数：

1、调用epoll_create建立一个epoll对象

int epoll_create(int size)
// size参数（大于0）表示epoll对象会处理的事件的大致数量，而不是能够处理的事件的最大数。
// 成功，则返回epoll对象句柄；如果出错，则返回-1，并且将errno设置为指示错误。

这一步，会创建一个eventpoll结构的对象，主要结构如下：rbr成员指向一颗红黑树，存储了需要监听的连接；rdllist是一个双向链表，存储满足条件的事件

struct eventpoll {
// ...
struct rb_root rbr; // 管理 epoll 监听的事件
struct list_head rdllist; // 保存着 epoll_wait 返回满⾜条件的事件
// ...
};

// 每个红黑树节点的结构为
struct epitem {
// ...
struct rb_node rbn; // 红⿊树节点
struct list_head rdllist; // 双向链表节点
struct epoll_filefd ffd; // 事件句柄信息
struct eventpoll *ep; // 指向所属的eventpoll对象
struct epoll_event event; // 注册的事件类型
// ...
};

需要注意的是，当创建好epoll对象后，它会占用一个fd值，在linux下查看/proc/进程id/fd可以看到该fd，因此在使用完后需要close关闭，负责可能导致fd资源被耗尽。

2、调用epoll_ctl向epoll对象中添加需要监听的连接，也可以修改或者删除epoll对象中的特定连接。

int epoll_ctl(int epfd, int op, int fd, struct epoll_event* event);
// epfd表示epoll_create创建的epoll对象的句柄
// op表示操作类型，可以是：EPOLL_CTL_ADD 添加新的事件到epoll中；EPOLL_CTL_MOD 修改epoll中的事件；EPOLL_CTL_DEL 删除epoll中的事件
// fd表示连接的句柄，也就是需要监听的fd
// epoll_event表示需要监听的事件，结构如下：
typedef union epoll_data {
void *ptr; // 指向用户自定义的数据
int fd; // 注册的文件描述符
uint32_t u32;
uint64_t u64;
}epoll_data_t;

struct epoll_event {
__uint32_t events; // 表示监听的事件
epoll_data_t data;
};
// events的取值：
EPOLLIN 表示该连接上有数据可读（tcp连接远端主动关闭连接，也是可读事件，因为需要处理发送来的FIN包； FIN包就是read 返回 0）
EPOLLOUT 表示该连接上可写发送（主动向上游服务器发起⾮阻塞tcp连接，连接建⽴成功事件相当于可写事件）
EPOLLRDHUP 表示tcp连接的远端关闭或半关闭连接
EPOLLPRI 表示连接上有紧急数据需要读
EPOLLERR 表示连接发⽣错误
EPOLLHUP 表示连接被挂起
EPOLLET 将触发⽅式设置为边缘触发，系统默认为⽔平触发
EPOLLONESHOT 表示该事件只处理⼀次，下次需要处理时需重新加⼊epoll

// 返回值：0表示成功，-1表示错误。
// 错误码：
EBADF：epfd或fd不是有效的文件描述符。
EEXIST：op为EPOLL_CTL_ADD，并且提供的文件描述符fd已在该epoll实例中注册。
EINVAL：epfd不是epoll文件描述符，或者fd与epfd相同，或者此接口不支持请求的操作op。
ENOENT：op是EPOLL_CTL_MOD或EPOLL_CTL_DEL，并且fd未在该epoll实例中注册。
ENOMEM：没有足够的内存来处理请求的操作控制操作。
ENOSPC：尝试在主机上注册（EPOLL_CTL_ADD）新文件描述符时遇到了/proc/sys/fs/epoll/max_user_watches施加的限制。
EPERM：目标文件fd不支持epoll。

3、调用epoll_wait等待事件的发生，并返回事件集合（即events），如果没有时间发生，则等待timieout毫秒

int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout)
// epfd表示epoll对象句柄
// events表示分配好的epoll_event结构体数组，epoll将会把发生的事件复制到events数组中（events不可以是空指针，内核只负责把数据复制到这个 events数组中，不会去帮助我们在用户态中分配内存。内核这种做法效率很高）
// maxevents表示本次可以返回的最大事件数目，通常maxevents参数与预分配的events数组的大小是相等的。该值不能大于创建epoll_create时的size
// timeout表示超时时间，单位为毫秒。0表示立即返回，-1表示一直等待

// 返回值：实际发生事件的连接数，0表示没有事件发生，-1表示错误

epoll的触发方式

分别为水平触发和边缘触发

1、水平触发时机

对于读操作，只要缓冲区内容不为空，就会返回读就绪；

对于写操作，只要换缓冲区还不满，就会返回写就绪。

当被监控的文件描述符上有可读写事件发生时，epoll_wait()会通知处理程序去读写。如果这次没有把数据一次性全部读写完，那么下次调用epoll_wait()时，它还会通知你在上次没读写完的文件描述符上继续读写。如果一直不去读写，它会一直通知。如果系统中有大量不需要读写的就绪文件描述符，而它们每次都会返回，这样会大大降低处理就绪文件描述符的效率。

水平触发模式适合一次性读大数据。

2、边缘触发时机

对于读操作，当缓冲区由不可读变为可读时；或者有新数据到达时（即缓冲区中的待读数据变多时）；或者缓冲区由数据可读，且应用进程对相应的描述符进行EPOLL_CTL_MOD修改EPOLLIN事件时。

当被监控的文件描述符上有可读写事件发生时，epoll_wait会通知处理程序去读写。如果这次没有把数据全部读写完，那么下次调用epoll_wait时，它不会再通知你。这种模式比水平模式的效率高，适合读取少量数据的场景。

具体实例

#include <stdio.h>
#include <unistd.h>
#include <sys/time.h>
#include <stdint.h>
#include <pthread.h>
#include <sys/eventfd.h>
#include <sys/epoll.h>

int efd = -1;
pthread_t pid = 0;

#define POST_PROCESS(PID, EFD)\
do\
{\
    if ((PID) != 0)\
    {\
        pthread_join((PID), nullptr);\
        (PID) = 0;\
    }\
\
    if ((EFD) >= 0)\
    {\
        close((EFD));\
        (EFD) = -1;\
    }\
\
}while(0)

void *read_thread(void *)
{
    printf("read_func begin\n");

    uint64_t data;
    int ret;

    int epfd = epoll_create(10);
    if (epfd < 0)
    {
        printf("epoll_create fail\n");
        return nullptr;
    }

    struct epoll_event event;
    event.events = EPOLLET | EPOLLIN;
    event.data.fd = efd;
    ret = epoll_ctl(epfd, EPOLL_CTL_ADD, efd, &event);
    if (ret < 0)
    {
        printf("epoll_ctl fail\n");
        POST_PROCESS(pid, efd);
        return nullptr;
    }

    struct epoll_event events[10];
    while(true)
    {
        ret = epoll_wait(epfd, &events[0], 10, 5000);
        if (ret > 0)
        {
            for (int i = 0; i < ret; i++)
            {
                if (events[i].events & EPOLLIN)
                {
                    int event_fd = events[i].data.fd;
                    ret = read(event_fd, &data, sizeof(data));
                    if (ret < 0)
                    {
                        printf("read fail\n");
                    }
                    printf("success read from efd, read %d bytes, data is %d\n", ret, data);
                }
            }
        }
        else if (ret == 0)
        {
            printf("epoll wait timeout\n");
            break;
        }
        else
        {
            printf("epoll wait error\n");
            POST_PROCESS(pid, efd);
            return nullptr;
        }
    }
    return nullptr;
}

int main()
{
    efd = eventfd(0, 0);
    if (efd < 0)
    {
        printf("eventfd fail\n");
        return -1;
    }

    pthread_t pid = 0;
    int ret = pthread_create(&pid, nullptr, read_thread, nullptr);
    if (ret < 0)
    {
        printf("pthread_create fail\n");
        POST_PROCESS(pid, efd);
        return -1;
    }

    uint64_t data;
    for (int i = 1; i <= 5; i++)
    {
        data = i;
        ret = write(efd, &data, sizeof(data));
        if (ret < 0)
        {
            printf("write fail\n");
            POST_PROCESS(pid, efd);

            return -1;
        }
        else
        {
            printf("write success, data = %d\n", data);
        }

        sleep(1);
    }

    POST_PROCESS(pid, efd);

    return 0;
}