您的快递（高并发服务器之poll和epoll）请签收

Posted 2021-01-08 liudw

　　前言

　　之前已经介绍过select函数，请参考这篇博客:https://www.cnblogs.com/liudw-0215/p/9661583.html,原理都是类似的，有时间先阅读下那篇博客，以便于理解这篇博客。

　　一、poll函数

　　1、函数说明

　　原型：int poll(struct pollfd *fds, nfds_t nfds, int timeout);

　　参数说明：　

　　参数fds：

　　struct pollfd {

int fd; /* 文件描述符 */

short events; /* 监控的事件 */

short revents; /* 监控事件中满足条件返回的事件 */

};

POLLIN 普通或带外优先数据可读,即POLLRDNORM | POLLRDBAND

POLLRDNORM 数据可读

POLLRDBAND 优先级带数据可读

POLLPRI 高优先级可读数据

POLLOUT 普通或带外数据可写

POLLWRNORM 数据可写

POLLWRBAND 优先级带数据可写

POLLERR 发生错误

POLLHUP 发生挂起

POLLNVAL 描述字不是一个打开的文件

 

nfds：监控数组中有多少文件描述符需要被监控

 

timeout：毫秒级等待

-1：阻塞等，#define INFTIM -1 Linux中没有定义此宏

0：立即返回，不阻塞进程

>0：等待指定毫秒数，如当前系统时间精度不够毫秒，向上取值

　　如果不再监控某个文件描述符时，可以把pollfd中，fd设置为-1，poll不再监控此pollfd，下次返回时，把revents设置为0。

　　2、程序示例

　　理解select之后，再解poll就很简单了，服务端代码如下：

　　

/* server.c */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <poll.h>
#include <errno.h>
#include "wrap.h"

#define MAXLINE 80
#define SERV_PORT 6666
#define OPEN_MAX 1024

int main(int argc, char *argv[])
{
    int i, j, maxi, listenfd, connfd, sockfd;
    int nready;
    ssize_t n;
    char buf[MAXLINE], str[INET_ADDRSTRLEN];
    socklen_t clilen;
    struct pollfd client[OPEN_MAX];
    struct sockaddr_in cliaddr, servaddr;

    listenfd = Socket(AF_INET, SOCK_STREAM, 0);

    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(SERV_PORT);

    Bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr));

    Listen(listenfd, 20);

    client[0].fd = listenfd;
    client[0].events = POLLRDNORM;                     /* listenfd监听普通读事件 */

    for (i = 1; i < OPEN_MAX; i++)
        client[i].fd = -1;                             /* 用-1初始化client[]里剩下元素 */
    maxi = 0;                                         /* client[]数组有效元素中最大元素下标 */

    for ( ; ; ) {
        nready = poll(client, maxi+1, -1);             /* 阻塞 */
        if (client[0].revents & POLLRDNORM) {         /* 有客户端链接请求 */
            clilen = sizeof(cliaddr);
            connfd = Accept(listenfd, (struct sockaddr *)&cliaddr, &clilen);
            printf("received from %s at PORT %d\\n",
                    inet_ntop(AF_INET, &cliaddr.sin_addr, str, sizeof(str)),
                    ntohs(cliaddr.sin_port));
            for (i = 1; i < OPEN_MAX; i++) {
                if (client[i].fd < 0) {
                    client[i].fd = connfd;     /* 找到client[]中空闲的位置，存放accept返回的connfd */
                    break;
                }
            }

            if (i == OPEN_MAX)
                perr_exit("too many clients");

            client[i].events = POLLRDNORM;         /* 设置刚刚返回的connfd，监控读事件 */
            if (i > maxi)
                maxi = i;                         /* 更新client[]中最大元素下标 */
            if (--nready <= 0)
                continue;                         /* 没有更多就绪事件时,继续回到poll阻塞 */
        }
        for (i = 1; i <= maxi; i++) {             /* 检测client[] */
            if ((sockfd = client[i].fd) < 0)
                continue;
            if (client[i].revents & (POLLRDNORM | POLLERR)) {
                if ((n = Read(sockfd, buf, MAXLINE)) < 0) {
                    if (errno == ECONNRESET) { /* 当收到 RST标志时 */
                        /* connection reset by client */
                        printf("client[%d] aborted connection\\n", i);
                        Close(sockfd);
                        client[i].fd = -1;
                    } else {
                        perr_exit("read error");
                    }
                } else if (n == 0) {
                    /* connection closed by client */
                    printf("client[%d] closed connection\\n", i);
                    Close(sockfd);
                    client[i].fd = -1;
                } else {
                    for (j = 0; j < n; j++)
                        buf[j] = toupper(buf[j]);
                        Writen(sockfd, buf, n);
                }
                if (--nready <= 0)
                    break;                 /* no more readable descriptors */
            }
        }
    }
    return 0;
}

View Code

 

　　程序中封装了包裹函数，有需要的请评论留言。

　　二、epoll函数

　　1、介绍　

　　epoll是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率，因为它会复用文件描述符集合来传递结果而不用迫使开发者每次等待事件之前都必须重新准备要被侦听的文件描述符集合，另一点原因就是获取事件的时候，它无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了。

　　目前epell是linux大规模并发网络程序中的热门首选模型。

　　epoll除了提供select/poll那种IO事件的电平触发（Level Triggered）外，还提供了边沿触发（Edge Triggered），这就使得用户空间程序有可能缓存IO状态，减少epoll_wait/epoll_pwait的调用，提高应用程序效率。

　　2、函数说明

　　跟select和poll不一样，epoll不是一个函数，需要三个函数一起来实现，分别为epoll_create、epoll_ctl和epoll_wait，下面分别来说明这三个函数。

　　（1）epoll_create函数

　　功能：创建一个epoll，参数size用来告诉内核监听的文件描述符的个数，跟内存大小有关。

　　原型：int epoll_create(int size)

　　又到了上图时间了，如下图：　　PS：依旧是全博客园最丑图，不接受反驳。

　　

　　epoll_create返回的epfd，其实创建了红黑树，是它的根节点。

　　（2）epoll_ctl函数

　　功能：控制某个epoll监控的文件描述符上的事件：注册、修改、删除。

　　原型：int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)

　　参数说明：　

epfd： 为epoll_creat的

op： 表示动作，用3个宏来表示：

EPOLL_CTL_ADD (注册新的fd到epfd)，

EPOLL_CTL_MOD (修改已经注册的fd的监听事件)，

EPOLL_CTL_DEL (从epfd删除一个fd)；

event： 告诉内核需要监听的事件

 

struct epoll_event {

__uint32_t events; /* Epoll events */

epoll_data_t data; /* User data variable */

};

typedef union epoll_data {

void *ptr;

int fd;

uint32_t u32;

uint64_t u64;

} epoll_data_t;

 

EPOLLIN ： 表示对应的文件描述符可以读（包括对端SOCKET正常关闭）

EPOLLOUT： 表示对应的文件描述符可以写

EPOLLPRI： 表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）

EPOLLERR： 表示对应的文件描述符发生错误

EPOLLHUP： 表示对应的文件描述符被挂断；

EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)而言的

EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听这个socket的话，需要再次把这个socket加入到EPOLL队列里

 

 

　　（3）epoll_wait函数

 

　　功能：等待所监控文件描述符上有事件的产生

　　原型：int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)

　　参数说明：　　

events： 用来存内核得到事件的集合，

maxevents： 告之内核这个events有多大，这个maxevents的值不能大于创建epoll_create()时的size，

timeout： 是超时时间

-1： 阻塞

0： 立即返回，非阻塞

>0： 指定毫秒

返回值： 成功返回有多少文件描述符就绪，时间到时返回0，出错返回-1

　　3、示例程序

　　服务端程序如下：

　　

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <errno.h>
#include "wrap.h"

#define MAXLINE 80
#define SERV_PORT 6666
#define OPEN_MAX 1024

int main(int argc, char *argv[])
{
    int i, j, maxi, listenfd, connfd, sockfd;
    int nready, efd, res;
    ssize_t n;
    char buf[MAXLINE], str[INET_ADDRSTRLEN];
    socklen_t clilen;
    int client[OPEN_MAX];
    struct sockaddr_in cliaddr, servaddr;
    struct epoll_event tep, ep[OPEN_MAX];

    listenfd = Socket(AF_INET, SOCK_STREAM, 0);

    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(SERV_PORT);

    Bind(listenfd, (struct sockaddr *) &servaddr, sizeof(servaddr));

    Listen(listenfd, 20);

    for (i = 0; i < OPEN_MAX; i++)
        client[i] = -1;
    maxi = -1;

    efd = epoll_create(OPEN_MAX);
    if (efd == -1)
        perr_exit("epoll_create");

    tep.events = EPOLLIN; tep.data.fd = listenfd;

    res = epoll_ctl(efd, EPOLL_CTL_ADD, listenfd, &tep);
    if (res == -1)
        perr_exit("epoll_ctl");

    while (1) {
        nready = epoll_wait(efd, ep, OPEN_MAX, -1); /* 阻塞监听 */
        if (nready == -1)
            perr_exit("epoll_wait");

        for (i = 0; i < nready; i++) {
            if (!(ep[i].events & EPOLLIN))
                continue;
            if (ep[i].data.fd == listenfd) {
                clilen = sizeof(cliaddr);
                connfd = Accept(listenfd, (struct sockaddr *)&cliaddr, &clilen);
                printf("received from %s at PORT %d\\n", 
                        inet_ntop(AF_INET, &cliaddr.sin_addr, str, sizeof(str)), 
                        ntohs(cliaddr.sin_port));
                for (j = 0; j < OPEN_MAX; j++) {
                    if (client[j] < 0) {
                        client[j] = connfd; /* save descriptor */
                        break;
                    }
                }

                if (j == OPEN_MAX)
                    perr_exit("too many clients");
                if (j > maxi)
                    maxi = j;         /* max index in client[] array */

                tep.events = EPOLLIN; 
                tep.data.fd = connfd;
                res = epoll_ctl(efd, EPOLL_CTL_ADD, connfd, &tep);
                if (res == -1)
                    perr_exit("epoll_ctl");
            } else {
                sockfd = ep[i].data.fd;
                n = Read(sockfd, buf, MAXLINE);
                if (n == 0) {
                    for (j = 0; j <= maxi; j++) {
                        if (client[j] == sockfd) {
                            client[j] = -1;
                            break;
                        }
                    }
                    res = epoll_ctl(efd, EPOLL_CTL_DEL, sockfd, NULL);
                    if (res == -1)
                        perr_exit("epoll_ctl");

                    Close(sockfd);
                    printf("client[%d] closed connection\\n", j);
                } else {
                    for (j = 0; j < n; j++)
                        buf[j] = toupper(buf[j]);
                    Writen(sockfd, buf, n);
                }
            }
        }
    }
    close(listenfd);
    close(efd);
    return 0;
}

View Code

 

　　

　　总结：需要包裹函数、客户端等程序的，欢迎留言