linux 高并发事件触发处理 — epoll

小鱼儿 2022-05-15 00:25 502阅读 0赞

## 一. 概述 ##

epoll 是 Linux 内核为处理大批量文件描述符而作了改进的 poll，是 Linux 下多路复用 IO接口 select/poll 的增强版本

在 linux 的网络编程中，很长时间都在使用 select 来做事件触发。在 2.6 内核中，有一种替换它的机制，就是 epoll。

##### select 与 epoll 区别概述 #####

(1) 函数使用上：epoll 使用一组函数来完成任务，而不是单个函数

(2) 效率：select 使用轮询来处理，随着监听 fd 数目的增加而降低效率。而 epoll 把用户关心的文件描述符事件放在内核里的一个事件表中，只需要一个额外的文件描述符来标识内核中的这个事件表即可。

## 二. epoll 接口 ##

epoll 事件触发并发处理操作过程总共需要三个接口，下面详细说明这三个接口的使用。

##### 头文件 #####

#include <sys/epoll.h>

### 1 int epoll\_create(int size); ###

前面说到 epoll 使用内核事件表来实现 I/O 复用，所以需要一个额外的文件描述符来标识使用的内核事件表。

epoll\_create 函数就是用来获取内核事件表的特殊文件描述符，该函数返回的文件描述符将用作其他 epoll 系统调用的第一个参数，以指定要访问的内核事件表。

### 2 int epoll\_ctl(int epfd, int op, int fd, struct epoll\_event \*event); ###

epoll 的事件注册函数，用来操作内核事件表。它不同与 select() 是在监听事件时告诉内核要监听什么类型的事件，而是在这里先注册要监听的事件类型。

##### 参数含义: #####

1. epfd： 要操作的内核事件表的文件描述符，即 epoll_create 的返回值
    2. op：指定操作类型，操作类型有三种：
        -> EPOLL_CTL_ADD：往内核事件表中注册指定fd 相关的事件     -> EPOLL_CTL_MOD：修改指定 fd 上的注册事件     -> EPOLL_CTL_DEL：删除指定 fd 的注册事件 3. fd：所要操作的文件描述符，也就是要内核事件表中监听的 fd
    4. event：指定所要监听的事件类型，epoll_event 结构指针类型。

##### struct epoll\_even 结构如下： #####

typedef union epoll_data {
        void *ptr;
        int fd;
        __uint32_t u32;
        __uint64_t u64;
     } epoll_data_t;
    
     struct epoll_event {
        __uint32_t events; /* Epoll events */
        epoll_data_t data; /* User data variable */
    };

其中 events 成员描述事件类型，可以是以下几种类型宏的集合：

EPOLLIN：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
    EPOLLOUT：表示对应的文件描述符可以写；
    EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；
    EPOLLERR：表示对应的文件描述符发生错误；
    EPOLLHUP：表示对应的文件描述符被挂断；
    EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。
    EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

##### 返回值 #####

epoll\_ctl 成功时返回 0，失败则返回 -1，并设置 errno

### 3 int epoll\_wait(int epfd, struct epoll\_event\* events, int maxevents, int timeout); ###

等待事件的发生，它在一段超时时间之内等待一组文件描述符上的事件，epoll\_wait 函数如果检测到事件，就将所有就绪的事件从内核事件表(epfd 参数决定)中复制到第二个参数 events 指向的数组中。

##### 参数 #####

##### (1) epfd #####

要操作的内核事件表的文件描述符，即 epoll\_create 的返回值

##### (2) events #####

内核事件表中得到的检测事件集合

##### (3) maxevents & timeout #####

maxevents 告诉内核 events 的最大 size，timeout 指定超时时间

##### 返回值 #####

成功时返回就绪的文件描述符的个数，失败返回 -1 并设置 errno

## 三. epoll 工作模式 ##

epoll 对文件描述符的操作有两种模式：LT（level trigger）和 ET（edge trigger）。LT 模式是默认模式，LT 模式与 ET 模式的区别如下：

LT模式：电平触发，当 epoll\_wait 检测到描述符事件发生并将此事件通知应用程序，应用程序可以不立即处理该事件。下次调用 epoll\_wait 时，会再次响应应用程序并通知此事件。

ET模式：边沿触发，当 epoll\_wait 检测到描述符事件发生并将此事件通知应用程序，应用程序必须立即处理该事件。如果不处理，下次调用 epoll\_wait 时，不会再次响应应用程序并通知此事件。

ET 模式在很大程度上减少了 epoll 事件被重复触发的次数，因此效率要比 LT 模式高。epoll 工作在 ET 模式的时候，必须使用非阻塞套接口，以避免由于一个文件句柄的阻塞读/阻塞写操作把处理多个文件描述符的任务饿死。

## 四. 测试程序 ##

下面是一个服务器回射程序 echo，用来帮助理解 epoll 过程。

##### 服务器代码 #####

#include <stdio.h>
    #include <stdlib.h>
    #include <string.h>
    #include <errno.h>
    
    #include <netinet/in.h>
    #include <sys/socket.h>
    #include <arpa/inet.h>
    #include <sys/epoll.h>
    #include <unistd.h>
    #include <sys/types.h>
    
    #define IPADDRESS "127.0.0.1"
    #define PORT 8787
    #define MAXSIZE 1024
    #define LISTENQ 5
    #define FDSIZE 1000
    #define EPOLLEVENTS 100
    
    //函数声明
    //创建套接字并进行绑定
    static int socket_bind(const char* ip,int port);
    //IO多路复用epoll
    static void do_epoll(int listenfd);
    //事件处理函数
    static void
    handle_events(int epollfd,struct epoll_event *events,int num,int listenfd,char *buf);
    //处理接收到的连接
    static void handle_accpet(int epollfd,int listenfd);
    //读处理
    static void do_read(int epollfd,int fd,char *buf);
    //写处理
    static void do_write(int epollfd,int fd,char *buf);
    //添加事件
    static void add_event(int epollfd,int fd,int state);
    //修改事件
    static void modify_event(int epollfd,int fd,int state);
    //删除事件
    static void delete_event(int epollfd,int fd,int state);
    
    int main(int argc,char *argv[])
    {
        int  listenfd;
        listenfd = socket_bind(IPADDRESS,PORT);
        listen(listenfd,LISTENQ);
        do_epoll(listenfd);
        return 0;
    }
    
    static int socket_bind(const char* ip,int port)
    {
        int  listenfd;
        struct sockaddr_in servaddr;
        listenfd = socket(AF_INET,SOCK_STREAM,0);
        if (listenfd == -1)
        {
            perror("socket error:");
            exit(1);
        }
        bzero(&servaddr,sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        inet_pton(AF_INET,ip,&servaddr.sin_addr);
        servaddr.sin_port = htons(port);
        if (bind(listenfd,(struct sockaddr*)&servaddr,sizeof(servaddr)) == -1)
        {
            perror("bind error: ");
            exit(1);
        }
        return listenfd;
    }
    
    static void do_epoll(int listenfd)
    {
        int epollfd;
        struct epoll_event events[EPOLLEVENTS];
        int ret;
        char buf[MAXSIZE];
        memset(buf,0,MAXSIZE);
        //创建一个描述符
        epollfd = epoll_create(FDSIZE);
        //添加监听描述符事件
        add_event(epollfd,listenfd,EPOLLIN);
        for ( ; ; )
        {
            //获取已经准备好的描述符事件
            ret = epoll_wait(epollfd,events,EPOLLEVENTS,-1);
            handle_events(epollfd,events,ret,listenfd,buf);
        }
        close(epollfd);
    }
    
    static void
    handle_events(int epollfd,struct epoll_event *events,int num,int listenfd,char *buf)
    {
        int i;
        int fd;
        //进行选好遍历
        for (i = 0;i < num;i++)
        {
            fd = events[i].data.fd;
            //根据描述符的类型和事件类型进行处理
            if ((fd == listenfd) &&(events[i].events & EPOLLIN))
                handle_accpet(epollfd,listenfd);
            else if (events[i].events & EPOLLIN)
                do_read(epollfd,fd,buf);
            else if (events[i].events & EPOLLOUT)
                do_write(epollfd,fd,buf);
        }
    }
    static void handle_accpet(int epollfd,int listenfd)
    {
        int clifd;
        struct sockaddr_in cliaddr;
        socklen_t  cliaddrlen;
        clifd = accept(listenfd,(struct sockaddr*)&cliaddr,&cliaddrlen);
        if (clifd == -1)
            perror("accpet error:");
        else
        {
            printf("accept a new client: %s:%d\n",inet_ntoa(cliaddr.sin_addr),cliaddr.sin_port);
            //添加一个客户描述符和事件
            add_event(epollfd,clifd,EPOLLIN);
        }
    }
    
    static void do_read(int epollfd,int fd,char *buf)
    {
        int nread;
        nread = read(fd,buf,MAXSIZE);
        if (nread == -1)
        {
            perror("read error:");
            close(fd);
            delete_event(epollfd,fd,EPOLLIN);
        }
        else if (nread == 0)
        {
            fprintf(stderr,"client close.\n");
            close(fd);
            delete_event(epollfd,fd,EPOLLIN);
        }
        else
        {
            printf("read message is : %s",buf);
            //修改描述符对应的事件，由读改为写
            modify_event(epollfd,fd,EPOLLOUT);
        }
    }
    
    static void do_write(int epollfd,int fd,char *buf)
    {
        int nwrite;
        nwrite = write(fd,buf,strlen(buf));
        if (nwrite == -1)
        {
            perror("write error:");
            close(fd);
            delete_event(epollfd,fd,EPOLLOUT);
        }
        else
            modify_event(epollfd,fd,EPOLLIN);
        memset(buf,0,MAXSIZE);
    }
    
    static void add_event(int epollfd,int fd,int state)
    {
        struct epoll_event ev;
        ev.events = state;
        ev.data.fd = fd;
        epoll_ctl(epollfd,EPOLL_CTL_ADD,fd,&ev);
    }
    
    static void delete_event(int epollfd,int fd,int state)
    {
        struct epoll_event ev;
        ev.events = state;
        ev.data.fd = fd;
        epoll_ctl(epollfd,EPOLL_CTL_DEL,fd,&ev);
    }
    
    static void modify_event(int epollfd,int fd,int state)
    {
        struct epoll_event ev;
        ev.events = state;
        ev.data.fd = fd;
        epoll_ctl(epollfd,EPOLL_CTL_MOD,fd,&ev);
    }

##### 客户端也用 epoll 实现，控制 STDIN\_FILENO、STDOUT\_FILENO 和 sockfd 三个描述符，程序如下所示： #####

#include <netinet/in.h>
    #include <sys/socket.h>
    #include <stdio.h>
    #include <string.h>
    #include <stdlib.h>
    #include <sys/epoll.h>
    #include <time.h>
    #include <unistd.h>
    #include <sys/types.h>
    #include <arpa/inet.h>
    
    #define MAXSIZE 1024
    #define IPADDRESS "127.0.0.1"
    #define SERV_PORT 8787
    #define FDSIZE 1024
    #define EPOLLEVENTS 20
    
    static void handle_connection(int sockfd);
    static void
    handle_events(int epollfd,struct epoll_event *events,int num,int sockfd,char *buf);
    static void do_read(int epollfd,int fd,int sockfd,char *buf);
    static void do_read(int epollfd,int fd,int sockfd,char *buf);
    static void do_write(int epollfd,int fd,int sockfd,char *buf);
    static void add_event(int epollfd,int fd,int state);
    static void delete_event(int epollfd,int fd,int state);
    static void modify_event(int epollfd,int fd,int state);
    
    int main(int argc,char *argv[])
    {
        int                 sockfd;
        struct sockaddr_in  servaddr;
        sockfd = socket(AF_INET,SOCK_STREAM,0);
        bzero(&servaddr,sizeof(servaddr));
        servaddr.sin_family = AF_INET;
        servaddr.sin_port = htons(SERV_PORT);
        inet_pton(AF_INET,IPADDRESS,&servaddr.sin_addr);
        connect(sockfd,(struct sockaddr*)&servaddr,sizeof(servaddr));
        //处理连接
        handle_connection(sockfd);
        close(sockfd);
        return 0;
    }
    
    
    static void handle_connection(int sockfd)
    {
        int epollfd;
        struct epoll_event events[EPOLLEVENTS];
        char buf[MAXSIZE];
        int ret;
        epollfd = epoll_create(FDSIZE);
        add_event(epollfd,STDIN_FILENO,EPOLLIN);
        for ( ; ; )
        {
            ret = epoll_wait(epollfd,events,EPOLLEVENTS,-1);
            handle_events(epollfd,events,ret,sockfd,buf);
        }
        close(epollfd);
    }
    
    static void
    handle_events(int epollfd,struct epoll_event *events,int num,int sockfd,char *buf)
    {
        int fd;
        int i;
        for (i = 0;i < num;i++)
        {
            fd = events[i].data.fd;
            if (events[i].events & EPOLLIN)
                do_read(epollfd,fd,sockfd,buf);
            else if (events[i].events & EPOLLOUT)
                do_write(epollfd,fd,sockfd,buf);
        }
    }
    
    static void do_read(int epollfd,int fd,int sockfd,char *buf)
    {
        int nread;
        nread = read(fd,buf,MAXSIZE);
            if (nread == -1)
        {
            perror("read error:");
            close(fd);
        }
        else if (nread == 0)
        {
            fprintf(stderr,"server close.\n");
            close(fd);
        }
        else
        {
            if (fd == STDIN_FILENO)
                add_event(epollfd,sockfd,EPOLLOUT);
            else
            {
                delete_event(epollfd,sockfd,EPOLLIN);
                add_event(epollfd,STDOUT_FILENO,EPOLLOUT);
            }
        }
    }
    
    static void do_write(int epollfd,int fd,int sockfd,char *buf)
    {
        int nwrite;
        nwrite = write(fd,buf,strlen(buf));
        if (nwrite == -1)
        {
            perror("write error:");
            close(fd);
        }
        else
        {
            if (fd == STDOUT_FILENO)
                delete_event(epollfd,fd,EPOLLOUT);
            else
                modify_event(epollfd,fd,EPOLLIN);
        }
        memset(buf,0,MAXSIZE);
    }
    
    static void add_event(int epollfd,int fd,int state)
    {
        struct epoll_event ev;
        ev.events = state;
        ev.data.fd = fd;
        epoll_ctl(epollfd,EPOLL_CTL_ADD,fd,&ev);
    }
    
    static void delete_event(int epollfd,int fd,int state)
    {
        struct epoll_event ev;
        ev.events = state;
        ev.data.fd = fd;
        epoll_ctl(epollfd,EPOLL_CTL_DEL,fd,&ev);
    }
    
    static void modify_event(int epollfd,int fd,int state)
    {
        struct epoll_event ev;
        ev.events = state;
        ev.data.fd = fd;
        epoll_ctl(epollfd,EPOLL_CTL_MOD,fd,&ev);
    }