I/O多路转接之epoll

按照man手册的说法：是为了处理大批量句柄而作了改进的poll。
这句话对我而言，说和不说没什么区别，太抽象了，所以要弄清楚什么是epoll，还是要从底层剖析！

epoll的三个相关系统调用

一：epoll_create：创建一个epoll模型（也是文件）

参数解释：
size：指定生成文件描述符的最大范围。

返回值解释：
返回一个文件描述符，该fd标识创建的epoll模型。

二：epoll_ctl：向epoll模型添加/修改/删除对应文件描述符的对应事件

参数解释：
①epfd:epoll_create的返回值。
②op：表示处理的方式，有添加，修改或删除，用三个宏表示：

③fd：要监听的对应文件描述符
④event：要关心什么事件
struct epoll_event的结构：

至于events，我还是只介绍两个：
EPOLLIN：对应fd可读
EPOLLOUT：对应fd可写

返回值解释
成功返回0，失败返回-1。

三： epoll_wait

参数解释：
①epfd:epoll_create的返回值。
②events：是一个数组，为输出型，epoll会将发生的事件复制到该数组，所以不能为空。
③maxevents：poll_wait可以处理的连接事件的最大限度值
④timeout：和poll一样。

要理解epoll，就要知道epoll模型到底是什么！

调用epoll_create,创建一个epoll模型，要做三件事：
1.在操作系统底层构建回调机制；
2.在操作系统底层构建红黑树（而监听的文件描述符正好作为红黑树的键值）；
3.在操作系统底层构建就绪队列；

调用epoll_ctl，就是对红黑树的节点（对应监听的文件描述符）进行添加/修改/删除操作。

调用epoll_wait,当事件就绪时：
把对应节点拷贝一份至就绪队列，而epoll看的就是就绪队列，所以为O(1)的时间复杂度。

epoll有两种工作模式：LT（水平触发）和ET（边缘触发）

现在假设一个场景：
已经把一个socke的读事件t添加到epoll模型；
此时socket的对端写入了10kb的数据；
调用epoll_wait，会返回，说明已经就绪；
然后调用read读取了5kb数据；
继续调用epoll_wait

LT工作模式：

epoll默认为LT工作模式，在该模式下，当socket上事件就绪，可以不立即处理或者只处理一部分；
比如上面的场景，我只读了5kb的数据，缓冲区还剩5kb，当我再次调用epoll_wait，它还是会返回，告诉我socket的读事件就绪；
直到缓冲区的所有数据全部处理完，epoll_wait才不会返回；
LT支持非阻塞读写与阻塞读写。

举个例子：张三是个快递派送员，有一天他给我派送快递，打电话叫我下楼来拿，但是我当时很忙，叫他在楼下等一会，张三就在楼下等；过了一会我下楼了，但是快递太多，我一次拿不完，我就给张三说，我先拿一部分上楼，你在这等着，我待会再拿剩下的，张三就接着等，直到我拿完快递。

LT模式下的epoll服务器

int main(int argc,char* argv[])
{
    //建立监听套接字
    if(argc != 3){
        printf("./server [ip] [port]\n");
        return 1;
    }

    int listen_sock = socket(AF_INET,SOCK_STREAM,0);
    if(listen_sock < 0){
        perror("socket");
        return 2;
    }

    struct sockaddr_in local;
    local.sin_family = AF_INET;
    local.sin_addr.s_addr = inet_addr(argv[1]);
    local.sin_port = htons(atoi(argv[2]));

    if(bind(listen_sock,(struct sockaddr*)&local,sizeof(local)) < 0){
        perror("bind");
        return 3;
    }

    if(listen(listen_sock,5) < 0){
        perror("listen");
        return 4;
    }

    //创建epoll模型
    int epoll_fd = epoll_create(100);
    if(epoll_fd < 0){
        perror("epoll_create");
        return 5;
    }

    //listen_sock的读事件
    struct epoll_event event;
    event.events = EPOLLIN;
    event.data.fd = listen_sock;

    //将listen_sock的读事件添加进epoll模型
    if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,listen_sock,&event) < 0){
        perror("epoll_ctl");
        return 6;
    }

    while(1){
        struct epoll_event events[10];//作epoll_wait的参数，输出型
        int size = epoll_wait(epoll_fd,events,sizeof(events)/sizeof(events[0]),-1);//阻塞式等待
        if(size == 0){
            printf("超时！");
            continue;
        }
        if(size < 0){
            perror("epoll_wait");
            continue;
        }
        //else size > 0,有事件就绪
        int i = 0;
        for( ; i<size; i++){
            if(!(events[i].events & EPOLLIN))//必须是读就绪
                continue;

            if(events[i].data.fd == listen_sock){//listen_sock就绪，该建立连接了
                struct sockaddr_in addr;
                socklen_t len = sizeof(addr);
                int new_sock = accept(listen_sock,(struct sockaddr*)&addr,&len);
                if(new_sock < 0){
                    perror("new_sock");
                    continue;
                }

                //有了新的fd，所以将new_sock的读事件加入模型
                struct epoll_event ev;
                ev.data.fd = new_sock;
                ev.events = EPOLLIN;
                if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,new_sock,&ev) < 0){
                    perror("epoll_ctl");
                    continue;
                }
            }
            else{//普通sock就绪，开始读
                char buf[1024];
                ssize_t s = read(events[i].data.fd,buf,sizeof(buf)-1);
                if(s < 0){
                    perror("read");
                    continue;
                }
                printf("client >%s\n",buf);
            }
        }
    }
    return 0;
}

ET工作模式
在设置epoll_ctl的第二个参数时的事件时，加上EPOLLET标志，epoll变为ET工作模式：

接着拿上面的场景举例，ET模式下，虽然只读了5kb的数据，缓冲区还有5kb数据没有处理，但第二次调用epoll_wait，就不会再返回了；
换句话说，ET模式下，socket事件就绪，只有一次处理机会，必须马上处理；
所以ET的效率比LT更高；
ET只支持非阻塞的读写。

举个例子，李四也是个快递派送员，他也给我派送快递，打电话叫我下楼来拿，我还是忙啊，就叫他等一会；忙完我下楼了，快递还是太多一次拿不完，我给李四说：我先拿一部分上楼，你在这等着，我待会再拿剩下的，于是我拿着一部分快递上楼了，但是李四不管我那么多，直接就走了。
所以如果是李四这种派送员，我必须要一次拿完快递！

ET为什么只支持非阻塞读写？
ET模式下数据就绪只会返回一次，所以当数据就绪时，就要一直read，直到读完或出错（必须一次性拿完快递）。
如果当前fd是阻塞，当读完缓冲区数据，如果对端不关闭写，那么read函数会一直阻塞，影响后续逻辑。
至于如何将文件描述符设为非阻塞，以及非阻塞情况下如何读的问题，我在另一篇文章《非阻塞IO》中有详细说明，附链接：
https://blog.csdn.net/han8040laixin/article/details/81232464

ET模式下的epoll服务器

void SetNonBlock(int sock)//把该文件描述符设为非阻塞
{
    int f1 = fcntl(sock,F_GETFD);//获取sock
    if(f1 < 0){
        perror("fcntl");
        return;
    }
    fcntl(f1,F_SETFL,f1|O_NONBLOCK);//将当前属性按位与O_NONBLOCK,其实就是把对应为设为1，使其变为非阻塞
}

int main(int argc,char* argv[])
{
    if(argc != 3){
        printf("./server [ip] [port]\n");
        return 1;
    }

    //建立监听套接字
    int listen_sock = socket(AF_INET,SOCK_STREAM,0);
    if(listen_sock < 0){
        perror("socket");
        return 2;
    }

    struct sockaddr_in local;
    local.sin_family = AF_INET;
    local.sin_addr.s_addr = inet_addr(argv[1]);
    local.sin_port = htons(atoi(argv[2]));

    if(bind(listen_sock,(struct sockaddr*)&local,sizeof(local)) < 0){
        perror("bind");
        return 3;
    }

    if(listen(listen_sock,5) < 0){
        perror("listen");
        return 4;
    }

    //ET模式下只支持非阻塞读写，将listen_sock设为非阻塞
    SetNonBlock(listen_sock);

    //建立epoll模型
    int epoll_fd = epoll_create(100);
    if(epoll_fd < 0){
        perror("epoll");
        return 5;
    }

    //listen_sock的读事件
    struct epoll_event event;
    event.events = EPOLLIN|EPOLLET;//把工作模式变为ET模式
    event.data.fd = listen_sock;
    //将listen_sock（非阻塞）的读事件添加进ET模式下的epoll模型
    if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,listen_sock,&event) < 0){
        perror("epoll_ctl");
        return 6;
    }

    while(1){
        struct epoll_event events[15];//输出型
        int size = epoll_wait(epoll_fd,events,sizeof(events)/sizeof(events[0]),-1);//非阻塞等待
        if(size < 0){
            printf("出错!\n");
            continue;
        }
        if(size == 0){
            printf("超时!\n");
                continue;
        }
        //else size > 0 有事件就绪
        int i = 0;
        for( ; i<size; i++ ){
            if(!(events[i].events & EPOLLIN))//必须是读事件
                continue;

            if(events[i].data.fd == listen_sock){//listen_sock就绪，可以建立连接了
                struct sockaddr_in addr;
                socklen_t len = sizeof(addr);
                int new_sock = accept(listen_sock,(struct sockaddr*)&addr,&len);
                if(new_sock < 0){
                    perror("accept");
                    continue;
                }
                SetNonBlock(new_sock);//设为非阻塞
                //new_sock的读事件
                struct epoll_event ev;
                ev.events = EPOLLIN|EPOLLET;//ET
                ev.data.fd = new_sock;
                if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,new_sock,&ev) < 0){
                    perror("epoll_ctl");
                    continue;
                }
            }
            else{//普通sock就绪，可以读了，但在ET模式下要一次读完
                //非阻塞调用：\
                  1.不会阻塞式等待\
                  2.条件不满足时直接以出错形式返回，错误码errno被设为EAGAIN(11号)
                  char buf[1024];
                  ssize_t total_size = 0;
                  while(1){
                      ssize_t s = read(events[i].data.fd,buf+total_size,1024);
                      total_size = total_size + s;
                      if(s < 1024 || errno == EAGAIN)
                          break;
                  }
                  buf[total_size] = '\0';
                  printf("client> %s\n",buf);
            }
        }
    }

客户端和select，poll都一样。

epoll的优缺点：
优点：
1.文件描述符无上限，操作系统使用红黑树管理fd。
2.维护就绪队列：当fd就绪，操作系统会把它从红黑树拷贝一份放到就绪队列，这样使用epoll_wait获取就绪fd时，只用看就绪队列，它是O(1)的时间复杂度。
3.事件就绪通知机制：一旦被监听的fd就绪，会有回调机制迅速**该fd，这样即使fd的数量增加，也不会影响判断就绪的性能。
epoll并没有所谓的内存映射机制，操作系统不相信任何人！而且epoll_wait里传了缓冲区，如果都内存映射了，根本不需要缓冲区，所以不存在内存映射机制！
缺点：
epoll的高性能是有使用场景的，如果在不合适的场景下使用epoll，会适得其反。
epoll适合多连接且只有一部分连接比较活跃的场景，如果只是少数几个连接，调用epoll显然太重了。