4 Linux网络编程7 poll,epoll - 《Linux服务器编程入门》

Poll API介绍

先讲下select的缺点 1 每次调用select都需要把fd_set从用户态拷贝到内核态(这样内核才直到要监听哪些文件描述符) 返回传出时需要将fd_set从内核态拷贝到用户态 2 select在内核中需要遍历传进来的置1的fd 在fd_set中有较多置1的文件描述符时开销会很大 O(n) 3 select能支持的最多的文件描述符数量为1024 太少了 4 fd_set既是传入也是传出参数不能重用，每次都需要重置

//poll就是对select的改进
//poll 的fds可重复使用且能支持的文件描述符数量无限制但是缺点1,2还是存在的
#include
struct pollfd{
int fd;//委托内核检测的文件描述符
short events;//委托内核对fd的什么事件进行检测
short revents;//文件描述符实际发生的事件 传出参数 return events
};
int poll(struct pollfd* fds,nfds_t nfds,int timeout);
图片1.png
POLLIN 检测此fd读缓冲区是否有新的数据同时检测POLLIN|POLLOUT
fds为struct pollfd结构体类型数组的首地址，保存了要检测的文件描述符的信息
nfds fds数组的最后一个有效元素的下标+1
timeout 阻塞时长传入0表示不阻塞 -1表示永久阻塞直到检测到fds中的fd的一些状态发生改变才解除阻塞函数返回 >0为阻塞的时长单位为毫秒

失败返回-1 n>0 检测到fds中有n个fd的状态发生变化

poll代码举例

//io多路复用之poll
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <string.h>
#include <poll.h>
int main()
{
    //创建监听socket
    int lfd = socket(PF_INET, SOCK_STREAM, 0);
    struct sockaddr_in saddr;
    saddr.sin_port = htons(9999);
    saddr.sin_family = AF_INET;
    saddr.sin_addr.s_addr = INADDR_ANY; //服务器要绑定的地址 服务器任一网卡的ip
    //绑定
    bind(lfd, (struct sockaddr *)&saddr, sizeof(saddr));
    //监听
    listen(lfd, 8); //等待连接的最大队列长度为8
    //原BIO accept 接收连接 + while死循环与客户端通信
    //初始化检测的文件描述符属性数组
    struct pollfd fds[1026];
    for (int i = 0; i < 1026; i++)
    {
        fds[i].fd = -1;
        fds[i].events = POLLIN; //只检测所有文件描述的读缓存区是否有新数据
    }
    fds[0].fd = lfd; //监听的文件描述符
    int nfds = 0;
    while (1)
    {
        //调用poll 让内核帮助监听文件描述符
        int ret = poll(fds, nfds + 1, -1); //只辅助监听 不检测是否能写 不检测是否有一样 永久阻塞
        if (ret == -1)
        {
            perror("poll:");
            exit(-1);
        }
        else if (ret == 0) //超时时间到了 并且在这段时间内要检测的fd的状态没有改变(没有新的可读信息)
        {
            continue;
        }
        else if (ret > 0) //ret>0 ret中保存我们监听的fd中有几个的状态发生改变(有几个fd的读缓冲区数据发生变化--即有新的可读信息)
        {
            if (fds[0].revents & POLLIN) //因为revents的结果不一定只有POLLIN
            //在同时检测读和写时revents结果可能是POLLIN|POLLOUT作为结果
            //如果我们用fds[0].revents == POLLIN 来判断是否有新数据到来
            //当同时 有新数据到来和写缓冲区有空间可写将会导致返回POLLIN|POLLOUT!=POLLIN
            //这种情况下用& POLLIN就行可以单独检测我们像检测的状态是否发生变化
            {
                //判断服务器监听描述符的读缓冲区是否有新的数据
                //表示有新的客户端连接进来了！！！！！！！
                struct sockaddr_in client_addr;
                int len = sizeof(client_addr);
                int cfd = accept(lfd, (struct sockaddr *)&client_addr, &len); //接收这个新的连接  这里因为提前直到有新连接到来 调用accept不会阻塞 而是直接接收到那个连接请求
                //将新的与客户端通信的fd 加入fd数组中 已完成poll对这个新通信连接的监听
                for (int i = 1; i < 1026; i++)
                //我们不能直接将数组下标为nfds+1的空间分给这个fd
                //因为客户端断开是随机的 在nfds+1之前如果有数组空间空闲了 但我们跳过没用不就浪费了
                //所以还是要从头检查数组是否还有空间没被占用(fd!=-1)
                {
                    if (fds[i].fd == -1)
                    {
                        fds[i].fd = cfd;
                        fds[i].events = POLLIN;
                        nfds = nfds > i ? nfds : i; //数组中的最大可用索引
                        break;
                    }
                } //从头开始找
            }
            //遍历其他集合中的文件描述符
            for (int i = 0 + 1; i <= nfds; i++)
            {
                if (fds[i].revents & POLLIN)
                {
                    //fd = i对应的连接的客户端发来了数据
                    char buf[1024] = {};
                    int len = read(fds[i].fd, buf, sizeof(buf)); //调用read 因为缓冲区有数据 所以不会阻塞可以直接读
                    if (len == -1)
                    {
                        perror("read:");
                        exit(-1);
                    }
                    else if (len == 0)
                    {
                        //对方断开连接
                        printf("client closed\n");
                        //将这个通信fd移除
                        close(fds[i].fd);
                        fds[i].fd = -1; //=-1表示检测数组中的这个位置空闲了
                    }
                    else if (len > 0) //读到了数据
                    {
                        //数据处理
                        printf("from client: %s \n", buf);
                        write(fds[i].fd, buf, strlen(buf) + 1); //回写
                    }
                }
            }
        }
    }
    close(lfd);
    return 0;
}

epoll API介绍

epoll_create在内核区实例化eventpoll(结构体) 返回epfd，在用户态对epfd操作就能影响内核区中的eventpoll，eventpoll结构体中有struct rb_root rbr 红黑树根节点(保存要检测的fd的信息) struct list_head rdlist 就绪链表头节点(epoll检测到fd的状态改变会将其信息放在这个链表中)。因为eventpoll就是在内核中的我们不需要进行用户态和内核态之间的拷贝操作
epoll_ctl函数能够完成对内核区eventpoll添加或删除要检测的fd(如果是添加还要传入指定要检测fd什么信息 ev)
epoll_wait 让内核去检测红黑树根fd节点是否有指定的事件发生，如果有则会将rdlist中的一些信息拷贝到用户区(只会返回发生指定事件的fd信息不需要我们像poll和select一样自己去遍历检查)
图片2.png

#include <sys/epoll.h>
int epoll_create(int size);
//在内核中创建一个eventpoll(结构体)实例(含 需要检测的fd信息的红黑树 和 双向就绪链表 保存检测到对应事件发生的fd的信息)
//size 目前没有什么意义了但是要大于0(以前是用哈希实现的)
//失败返回-1  大于0 为一个epfd文件描述符用于操作eventpoll实例
int epoll_ctl(int epfd,int op,int fd,struct epoll_event* event);
//对epfd对应的eventpoll实例进行添加、删除fd或修改fd对应信息的操作
//epfd eventpoll实例对应的文件描述符
//op 要进行什么操作 删除EPOLL_CTL_DEL 添加EPOLL_CTL_ADD 修改EPOLL_CTL_MOD  fd
//修改红黑树节点信息
//fd 要检测的文件描述符
//event  具体要检测fd的什么信息
struct epoll_event
{
    uint32_t events;//检测的事件 事件非常多 可以在 man epoll_ctl中看到 主要有EPOLLIN(与此fd关联的文件是否可读) EPOLLOUT(与此fd关联的文件是否可写)  EPOLLERR
    epoll_data_t data;//用户数据信息
};
//epoll_data_t 是个联合体
typedef union epoll_data{
      void* ptr;
      int fd;
      uint32_t u32;
      uint64_t u64;
}epoll_data_t;
int epoll_wait(int epfd,struct epoll_event* event,int maxevents,int timeout);
//让内核去检测红黑树根fd节点是否有指定的事件发生，如果有则会将rdlist中的一些信息拷贝到用户区(只会返回发生指定事件的fd信息 不需要我们像poll和select一样自己去遍历检查)
//epfd eventpoll实例对应的文件描述符
//event 作为传出参数 保存了发生了指定事件的文件描述符信息 是个数组 maxevents就是这个传出数组的大小 events不可以是空指针，内核只负责把数据复制到这个 events数组中，不会去帮助我们在用户态中分配内存。内核这种做法效率很高
//timeout 阻塞时长 传入0表示不阻塞 -1表示永久阻塞直到检测到红黑树中的fd的一些状态发生改变才解除阻塞函数返回   >0为阻塞的时长单位为毫秒
//成功返回发生变化的文件描述符的个数 失败返回-1

epoll举例

//io多路复用之epoll
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <string.h>
#include <sys/epoll.h>
int main()
{
    //创建监听socket
    int lfd = socket(PF_INET, SOCK_STREAM, 0);
    struct sockaddr_in saddr;
    saddr.sin_port = htons(9999);
    saddr.sin_family = AF_INET;
    saddr.sin_addr.s_addr = INADDR_ANY; //服务器要绑定的地址 服务器任一网卡的ip
    //绑定
    bind(lfd, (struct sockaddr *)&saddr, sizeof(saddr));
    //监听
    listen(lfd, 8); //等待连接的最大队列长度为8
    //原BIO accept 接收连接 + while死循环与客户端通信
    //epoll_create在内核创建event_poll结构体实例
    int epfd = epoll_create(1);
    //将监听文件描述符的相关信息加到event_poll结构体实例中的红黑树中
    struct epoll_event epev;
    epev.events = EPOLLIN; //检测是否可读的事件
    epev.data.fd = lfd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, lfd, &epev);
    struct epoll_event epevs[1026];
    int maxevents = 1026;
    // event 作为传出参数 保存了发生了指定事件的文件描述符信息 是个数组
    // maxevents就是这个传出数组的大小 events不可以是空指针，内核只负责把数据复制到这个
    // events数组中，不会去帮助我们在用户态中分配内存。内核这种做法效率很高
    while (1)
    {
        //epoll wait 去检测我们指定的fd是否发生了指定的事件
        int ret = epoll_wait(epfd, epevs, maxevents, -1);
        //返回ret有几个fd发生了指定的事件  具体是哪几个保存在epevs的data.fd中
        if (ret == -1)
        {
            perror("epoll_wait:");
            exit(-1);
        }
        else if (ret == 0) //超时时间到了 并且在这段时间内要检测的fd的状态没有改变(没有新的可读信息)
        {
            continue;
        }
        else if (ret > 0)
        {
            //ret>0 ret中保存我们监听的fd中有几个的状态发生改变(有几个fd的读缓冲区数据发生变化--即有新的可读信息)
            //具体哪几个保存在传出的结构体数组中
            for (int i = 0; i < ret; i++)
            {
                if (epevs[i].data.fd == lfd && (epevs[i].events & EPOLLIN))
                {
                    //监听描述符发生对于事件 并且这个事件是 读缓存区可读
                    //有新的客户端连入 accept创建通信fd  并将这个通信fd加入epoll实例的红黑树中
                    struct sockaddr_in client_addr;
                    int len = sizeof(client_addr);
                    int cfd = accept(lfd, (struct sockaddr *)&client_addr, &len); //接收这个新的连接  这里因为提前直到有新连接到来 调用accept不会阻塞 而是直接接收到那个连接请求
                    epev.events = EPOLLIN; //检测是否可读的事件 | EPOLLOUT
                    epev.data.fd = cfd;
                    epoll_ctl(epfd, EPOLL_CTL_ADD, cfd, &epev);
                }
                else if (epevs[i].events & EPOLLIN) //只检测读事件 非0
                {
                    //用于通信fd 客户端有数据到达
                    char buf[1024] = {};
                    int len = read(epevs[i].data.fd, buf, sizeof(buf)); //调用read 因为缓冲区有数据 所以不会阻塞可以直接读
                    if (len == -1)
                    {
                        perror("read:");
                        exit(-1);
                    }
                    else if (len == 0)
                    {
                        //对方断开连接
                        printf("client closed\n");
                        //将这个通信fd移除
                        epoll_ctl(epfd, EPOLL_CTL_DEL, epevs[i].data.fd, NULL);
                        close(epevs[i].data.fd);
                        //之前都是先关闭后移除 这里是先从红黑树中移除再关闭
                    }
                    else if (len > 0) //读到了数据
                    {
                        //数据处理
                        printf("from client: %s \n", buf);
                        write(epevs[i].data.fd, buf, strlen(buf) + 1); //回写
                    }
                }
            }
        }
    }
    close(lfd);
    close(epfd);
    return 0;
}

epoll的两种工作模式

LT(level triggered)模式状态时同时支持阻塞和非阻塞socket。，主线程正在epoll_wait等待事件时，请求到了，epoll_wait返回后没有去处理请求(recv)，那么下次epoll_wait时此请求还是会返回（立刻返回了）；
ET(Edge Triggered)模式
而ET模式状态下要求使用非阻塞socket，这次没处理，下次epoll_wait时将不返回（所以我们应该每次一定要处理），可见很大程度降低了epoll的触发次数（记住这句话先）。内核只会在fd从未就绪变为就绪通知一次(具体ET在什么情况下会触发在Edge Triggered (ET) 边沿触发有写)

没有将读缓冲区数据读完，下次epoll ET不会提醒(LT会提醒)这个fd读缓冲区还有数据。所以当epoll触发后需要一次性将读缓冲区数据读完(循环读)，如果循环读读缓冲区的过程中发生阻塞将会影响对于其他fd检测。

ET模式减少了epoll事件被重复触发的次数，必须使用非阻塞socket,以避免一个对文件的阻塞读和阻塞写将处理多个fd的任务饿死(不去处理)。
Level Triggered (LT) 水平触发socket接收缓冲区不为空有数据可读读事件一直触发socket发送缓冲区不满可以继续写入数据写事件一直触发符合思维习惯，epoll_wait返回的事件就是socket的状态
在read 无法一次性读完读缓冲区，下次调用epoll wait依然会触发通知
Edge Triggered (ET) 边沿触发socket的接收缓冲区状态变化时触发读事件，即空的接收缓冲区刚接收到数据时触发读事件socket的发送缓冲区状态变化时触发写事件，即满的缓冲区刚空出空间时触发读事件仅在状态变化时触发事件
设置边沿触发 epoll_event中的event设置为EPOLLIN | EPOLLET

//io多路复用之epoll ET 边沿触发
//socket的接收缓冲区状态变化时触发读事件，即空的接收缓冲区刚接收到数据时触发读事件
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <string.h>
#include <sys/epoll.h>
#include <fcntl.h> //设置文件描述符非阻塞
#include <errno.h>
int main()
{
    //创建监听socket
    int lfd = socket(PF_INET, SOCK_STREAM, 0);
    struct sockaddr_in saddr;
    saddr.sin_port = htons(9999);
    saddr.sin_family = AF_INET;
    saddr.sin_addr.s_addr = INADDR_ANY; //服务器要绑定的地址 服务器任一网卡的ip
    //绑定
    bind(lfd, (struct sockaddr *)&saddr, sizeof(saddr));
    //监听
    listen(lfd, 8); //等待连接的最大队列长度为8
    //原BIO accept 接收连接 + while死循环与客户端通信
    //epoll_create在内核创建event_poll结构体实例
    int epfd = epoll_create(1);
    //将监听文件描述符的相关信息加到event_poll结构体实例中的红黑树中
    struct epoll_event epev;
    epev.events = EPOLLIN; //检测是否可读的事件
    epev.data.fd = lfd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, lfd, &epev);
    struct epoll_event epevs[1026];
    int maxevents = 1026;
    // event 作为传出参数 保存了发生了指定事件的文件描述符信息 是个数组
    // maxevents就是这个传出数组的大小 events不可以是空指针，内核只负责把数据复制到这个
    // events数组中，不会去帮助我们在用户态中分配内存。内核这种做法效率很高
    while (1)
    {
        //epoll wait 去检测我们指定的fd是否发生了指定的事件
        int ret = epoll_wait(epfd, epevs, maxevents, -1);
        //返回ret有几个fd发生了指定的事件  具体是哪几个保存在epevs的data.fd中
        if (ret == -1)
        {
            perror("epoll_wait:");
            exit(-1);
        }
        else if (ret == 0) //超时时间到了 并且在这段时间内要检测的fd的状态没有改变(没有新的可读信息)
        {
            continue;
        }
        else if (ret > 0)
        {
            //ret>0 ret中保存我们监听的fd中有几个的状态发生改变(有几个fd的读缓冲区数据发生变化--即有新的可读信息)
            //具体哪几个保存在传出的结构体数组中
            for (int i = 0; i < ret; i++)
            {
                if (epevs[i].data.fd == lfd && (epevs[i].events & EPOLLIN))
                {
                    //监听描述符发生对于事件 并且这个事件是 读缓存区可读
                    //有新的客户端连入 accept创建通信fd  并将这个通信fd加入epoll实例的红黑树中
                    struct sockaddr_in client_addr;
                    int len = sizeof(client_addr);
                    int cfd = accept(lfd, (struct sockaddr *)&client_addr, &len); //接收这个新的连接  这里因为提前直到有新连接到来 调用accept不会阻塞 而是直接接收到那个连接请求
                    //通信的fd设置读非阻塞
                    int flag = fcntl(cfd, F_GETFL);         //获取cfd属性
                    fcntl(cfd, F_SETFL, flag | O_NONBLOCK); //添加非阻塞属性
                    //用于和客户端通信的fd设置为边沿触发
                    epev.events = EPOLLIN | EPOLLET; //检测是否可读的事件 | EPOLLOUT
                    epev.data.fd = cfd;
                    //边沿触发只会提醒一次 即空的接收缓冲区刚接收到数据时触发读事件
                    //我们需要在触发后将读缓冲区内的数据全部读出来(因为不会二次提醒)
                    epoll_ctl(epfd, EPOLL_CTL_ADD, cfd, &epev);
                }
                else if (epevs[i].events & EPOLLIN) //只检测读事件 非0
                {
                    //用于通信fd 客户端有数据到达
                    //循环一次性读出读缓冲区的数据
                    char buf[5] = {};
                    int len = 0; //调用read 因为缓冲区有数据 所以不会阻塞可以直接读
                    //要求一次读 sizeof(buf)个字节  实际读了len个字节
                    while ((len = read(epevs[i].data.fd, buf, sizeof(buf))) > 0)
                    {
                        //raed 返回值>0 表示读缓冲区还有数据 则继续读
                        // printf("from client: %s \n", buf);
                        write(STDOUT_FILENO, buf, len);        //直接写再标准输出中(屏幕)
                        write(epevs[i].data.fd, buf, len + 1); //回写
                    }
                    if (len == -1)
                    {
                        //read 被信号中断 会触发 EINTR错误 此时不应该退出 而是继续读
                        //raed非阻塞读当读缓冲区读完了 再一次读将会触发EGAIN 此时不应该退出而是继续工作
                        if (errno == EAGAIN)
                            printf("buff read finish");
                        else
                        {
                            perror("read:");
                            exit(-1);
                        }
                    }
                    else if (len == 0) //read 返回值为0 表示客户端断开连接
                    {
                        //对方断开连接
                        printf("client closed\n");
                        //将这个通信fd移除
                        epoll_ctl(epfd, EPOLL_CTL_DEL, epevs[i].data.fd, NULL);
                        close(epevs[i].data.fd);
                        //之前都是先关闭后移除 这里是先从红黑树中移除再关闭
                    }
                }
            }
        }
    }
    close(lfd);
    close(epfd);
    return 0;
}