epoll的使用实例

Posted 2020-10-19 xcywt

　　在网络编程中通常需要处理很多个连接，可以用select和poll来处理多个连接。但是select都受进程能打开的最大文件描述符个数的限制。并且select和poll效率会随着监听fd的数目增多而下降。

解决方法就是用epoll

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

2.epoll、select、poll的区别：
1）相比于select和poll，epoll最大的好处在于不会随着监听fd数目的增加而降低效率
2）内核中的select与poll的实现是采用轮询来处理的，轮询数目越多耗时就越多
3）epoll的实现是基于回调的，如果fd有期望的事件发生就会通过回调函数将其加入epoll就绪队列中。也就是说它只关心“活跃”的fd，与fd数目无关。
4）内核空间用户空间数据拷贝问题，如何让内核把fd消息通知给用户空间？select和poll采取了内存拷贝的方法，而epoll采用的是共享内存的方式。速度快
5）epoll不仅会告诉应用程序有I/o事件的到来，还会告诉应用程序相关的信息，这写信息是应用程序填充的，因此根据这写信息应用程序就能直接定位到事件，而不必遍历整个fd集合。
6）poll和select受进程能打开的最大文件描述符的限制。select还受FD_SETSIZE的限制。但是epoll的限制的最大可以打开文件的数目（cat /proc/sys/fs/file-max进行查看）。

3.epoll的工作模式
有下面两种工作模式，默认是水平触发。
EPOLLLT：水平触发（Level Triggered），事件没有处理完也会触发。完全靠kernel epoll驱动，应用程序只需要处理从epoll_wait返回的fds。这些fds我们认为他们处于就绪状态。
LT模式支持阻塞fd和非阻塞fd。

EPOLLET：边沿触发（Edge Triggered）。只有空闲->就绪才会触发。应用程序需要维护一个就绪队列。
此模式下，系统仅仅通知应用程序哪些fds变成了就绪状态，一旦fd变成了就绪状态，epoll将不再关注这个fd的任何状态信息（从epoll队列移除）。
直到应用程序通过读写操作触发EAGAIN状态，epoll认为这个fd又变成空闲状态，那么epoll又重新关注这个fd的状态变化（重新加入epoll队列中）。
随着epoll_wait的返回，队列中的fds是减少的，所以在大并发的系统中，EPOLLET更有优势。但是对程序员的要求也更高。
ET模式只支持non-block socket，以避免由于一个文件句柄的阻塞读/阻塞写把处理多个文件描述符的任务饿死。

4.如何使用

主要是下面几个函数和结构体。

　　 #include <sys/epoll.h>
       int epoll_create(int size);
       int epoll_create1(int flags);
       int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
       int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
       The struct epoll_event is defined as :
           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 */
           };

1）int epoll_create(int size);
创建一个epoll的句柄，
size：告诉内核这个句柄可以监听的数目一共多大。
注意这里返回一个句柄，也是一个文件描述符，后面还是要关闭的。

2）int epoll_create1(int flags);
上面那个的加强版，flags可以是EPOLL_CLOEXEC，表示具有执行后关闭的特性

       EPOLL_CLOEXEC
              Set the close-on-exec (FD_CLOEXEC) flag on the new file descrip‐
              tor.  See the description of the O_CLOEXEC flag in  open(2)  for
              reasons why this may be useful

 

3） int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
注册函数，用来对创建的epollfd进行操作的。
epfd：create出来的fd
op：执行的动作，由3个宏来表示
　　EPOLL_CTL_ADD：注册新的fd到epfd中；
　　EPOLL_CTL_MOD：修改已经注册的fd的监听事件；
　　EPOLL_CTL_DEL：从epfd中删除一个fd；
fd：要监听的fd
event：表示需要监听的事件
结构见上面。
events可以是以下几个宏的集合：
EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
EPOLLOUT：表示对应的文件描述符可以写；
EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）；
EPOLLERR：表示对应的文件描述符发生错误；
EPOLLHUP：表示对应的文件描述符被挂断；
EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。
EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

4）  int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);
用来等待事件的产生
epfd：create出来的epollfd
events：从内核得到事件的集合。一般的一个数组
maxevents：用来告诉内核events有多大，不能大于epoll_create()时的size。
timeout是超时时间：单位的毫秒，为0表示立即返回，-1表示阻塞。
返回：0表示超时，>0表示需要处理的事件数目。<0表示出错

5.实例：
server端是一个回射服务器：

#include<sys/types.h>
#include<sys/socket.h>
#include<sys/select.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<poll.h>
#include<stdlib.h>
#include<stdio.h>
#include<string.h>
#include<errno.h>
#include<unistd.h>
#include <fcntl.h>
#include<sys/epoll.h>
#include<vector>
#include<algorithm>
// #include"comm.h"
void setfdisblock(int fd, bool isblock)
{
    int flags = fcntl(fd, F_GETFL);
    if(flags < 0)
        return;
    if(isblock) // 阻塞
    {
        flags &= ~O_NONBLOCK;
    }
    else // 非阻塞
    {
        flags |= O_NONBLOCK;
    }    
    
    if(fcntl(fd, F_SETFL, flags)<0)
        perror("fcntl set");
}
typedef std::vector<struct epoll_event> EventList;
#define CLIENTCOUNT 2048
#define MAX_EVENTS 2048
int main(int argc, char **argv)
{
    int listenfd = socket(AF_INET, SOCK_STREAM, 0);
    if(listenfd < 0)
    {
        perror("socket");
        return -1;
    }
    
    unsigned short sport = 8080;
    if(argc == 2)
    {
        sport = atoi(argv[1]);
    }
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    printf("port = %d\\n", sport);
    addr.sin_port = htons(sport);
    addr.sin_addr.s_addr = inet_addr("127.0.0.1");
    
    if(bind(listenfd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
    {
        perror("bind");
        return -2;
    }
    if(listen(listenfd, 20) < 0)
    {
        perror("listen");
        return -3;
    }
        
    struct sockaddr_in connaddr;
    socklen_t len = sizeof(connaddr);
    
    int i = 0, ret = 0;
    std::vector<int> clients; // 客户端存储的迭代器
    int epollfd = epoll_create1(EPOLL_CLOEXEC);
    //int epollfd = epoll_create(MAX_EVENTS);// 设置连接数
    
    struct epoll_event event;
    event.events = EPOLLIN|EPOLLET;
    event.data.fd = listenfd;
    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, listenfd, &event) < 0)
    {
        perror("epoll_ctl");
        return -2;
    }
    EventList events(16);
    int count = 0;
    int nready = 0;
    char buf[1024] = {0};
    int conn  = 0;
    while(1)
    {
        
        nready = epoll_wait(epollfd, &*events.begin(), static_cast<int>(events.size()), -1);
        if(nready == -1)
        {
            perror("epoll_wait");
                        return -3;
        }
        if(nready == 0) // 肯定不会走到这里，因为上面没设置超时时间
        {
            continue;
        }
        if((size_t)nready == events.size()) // 对clients进行扩容
            events.resize(events.size() * 2);
        for(i = 0; i < nready; i++)
        {
            if(events[i].data.fd == listenfd)
            {
                conn = accept(listenfd, (struct sockaddr*)&connaddr, &len);
                if(conn < 0)
                {
                    perror("accept");
                    return -4;
                }
                char strip[64] = {0};
                char *ip = inet_ntoa(connaddr.sin_addr);
                strcpy(strip, ip);
                printf("client connect, conn:%d,ip:%s, port:%d, count:%d\\n", conn, strip,ntohs(connaddr.sin_port), ++count);
        
                clients.push_back(conn);
                // 设为非阻塞
                setfdisblock(conn, false);
                // add fd in events
                event.data.fd = conn;// 这样在epoll_wait返回时就可以直接用了
                event.events = EPOLLIN|EPOLLET;
                epoll_ctl(epollfd, EPOLL_CTL_ADD, conn, &event);
                
            }
            else if(events[i].events & EPOLLIN)
            {
                conn = events[i].data.fd;
                if(conn < 0)
                    continue;
                ret = read(conn, buf, sizeof(buf));
                if(ret == -1)
                {
                    perror("read");
                    return -5;
                }
                else if(ret == 0)
                {
                    printf("client close remove:%d, count:%d\\n", conn, --count);
                    close(conn);
                    event = events[i];
                    epoll_ctl(epollfd, EPOLL_CTL_DEL, conn, &event);
                    clients.erase(std::remove(clients.begin(), clients.end(), conn), clients.end());
                }
                write(conn, buf, sizeof(buf));
                memset(buf, 0, sizeof(buf));
            }
        }    
    }
    close(listenfd);
    return 0;
}

client：连接服务器，通过终端发送数据给server，并接收server发来的数据。

#include<sys/types.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<sys/select.h>
#include<stdlib.h>
#include<stdio.h>
#include<string.h>
void select_test(int conn)
{
    int ret = 0;
    fd_set rset;
    FD_ZERO(&rset);
    int nready;
    int maxfd = conn;
    int fd_stdin = fileno(stdin);
    if(fd_stdin > maxfd)
    {
        maxfd = fd_stdin;
    }
    int len = 0;
    char readbuf[1024] = {0};
    char writebuf[1024] = {0};
    while(1)
    {
        FD_ZERO(&rset);
        FD_SET(fd_stdin, &rset);
        FD_SET(conn, &rset);
        nready = select(maxfd+1, &rset, NULL, NULL, NULL);
        if(nready == -1)
        {
            perror("select");
            exit(0);
        }
        else if(nready == 0)
        {
            continue;    
        }
        if(FD_ISSET(conn, &rset))
        {
            ret = read(conn, readbuf, sizeof(readbuf));
            if(ret == 0)
            {
                printf("server close1\\n");
                break;
            }
            else if(-1 == ret)
            {
                perror("read1");
                break;
            }    
            fputs(readbuf, stdout);
            memset(readbuf, 0, sizeof(readbuf));
        }    
        
        if(FD_ISSET(fd_stdin, &rset))
        {    
            read(fd_stdin, writebuf, sizeof(writebuf));        
            len = strlen(writebuf);
            ret = write(conn, writebuf, len);
            if(ret == 0)
            {
                printf("server close3\\n");
                break;
            }
            else if(-1 == ret)
            {
                perror("write");
                break;
            }
            memset(writebuf, 0, sizeof(writebuf));    
        }
    }
}
int sockfd = 0;
int main(int argc, char **argv)
{
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if(sockfd < 0)
    {
        perror("socket");
        return -1;
    }
        
    unsigned short sport = 8080;
    if(argc == 2)
    {
        sport = atoi(argv[1]);
    }
    struct sockaddr_in addr;
    addr.sin_family = AF_INET;
    printf("port = %d\\n", sport);
    addr.sin_port = htons(sport);
    addr.sin_addr.s_addr = inet_addr("127.0.0.1");
    if(connect(sockfd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
    {
        perror("connect");
        return -2;
    }
    struct sockaddr_in addr2;
    socklen_t len = sizeof(addr2);
    if(getpeername(sockfd, (struct sockaddr*)&addr2, &len) < 0)
    {
        perror("getsockname");
        return -3;
    }
    printf("Server: port:%d, ip:%s\\n", ntohs(addr2.sin_port), inet_ntoa(addr2.sin_addr));
    select_test(sockfd);
    close(sockfd);
    return 0;
}

还可以进行暴力连接测试。可以参考： http://www.cnblogs.com/xcywt/p/8120166.html 这个例子的客户端就是暴力连接测试的。
测试结果可以和select实现的server进行比较，发现在虚拟机上epollserver好像没有快多少（我也暂时没找到原因）。
但是在服务器上，还是有很快的，效率提升了很多。