5.9-selcct_poll_epoll 和 reactor 的模拟实现

5.9-select_poll_epoll

本文演示 select 等 io 多路复用函数的应用方法，函数具体介绍可以参考我过去写的博客。

先绑定监听的文件描述符

int sockfd = socket(AF_INET, SOCK_STREAM, 0);
struct sockaddr_in serveraddr;
memset(&serveraddr, 0, sizeof(struct sockaddr_in));serveraddr.sin_family = AF_INET;
serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);
serveraddr.sin_port = htons(2052);if (-1 == bind(sockfd, (struct sockaddr*)&serveraddr, sizeof(struct sockaddr)))
{perror("bind");return -1;
}listen(sockfd, 10);

1. select

select 函数适用于 win/linux 平台，但是使用时每次都需要检查位图内所有客户端的状态变化情况，属于针对于每一个文件进行处理而非针对事件处理，效率较低。打个比方：如果客户端是小区内的住户，那么 selcet 作为快递员，会从快递仓库中挑选出要被配送到该小区指定住户的快递，并对于每一个住户是否有快递/寄快递。

演示如下：

fd_set rfds, rset;FD_ZERO(&rfds);
FD_SET(sockfd, &rfds);int maxfd = sockfd;
while (1)
{rset = rfds;int nready = select(maxfd + 1, &rset, NULL, NULL, NULL);if (FD_ISSET(sockfd, &rset)){struct sockaddr_in clientaddr;socklen_t len = sizeof(struct sockaddr_in);int clientfd = accept(sockfd, (struct sockaddr*)&clientaddr, &len);printf("sockfd: %d\n", clientfd);FD_SET(clientfd, &rfds);maxfd = clientfd;}int i = 0;for (i = sockfd + 1; i <= maxfd; i++){if (FD_ISSET(i, &rset)){char buffer[128] = { 0 };int count = recv(i, buffer, 128, 0);if (0 == count){printf("disconnect\n");close(i);FD_CLR(i, &rfds);break;}send(i, buffer, count, 0);printf("clientfd: %d, count: %d, buffer: %s\n", i, count, buffer);}}
}

2. poll

poll 函数适用于 Linux 平台，效率相比于 select 有所提升。如果 selcet 是快递员要对于每一个住户确定一次需求，poll 则是可以直接锁定不同客户的需求。

演示如下：

struct pollfd fds[1024] = { 0 };fds[sockfd].fd = sockfd;
fds[sockfd].events = POLLIN;int maxfd = sockfd;while (1)
{int nready = poll(fds, maxfd + 1, -1);if (fds[sockfd].revents & POLLIN){struct  sockaddr_in clientaddr;int len = sizeof(struct sockaddr_in);int clientfd = accept(sockfd, (struct sockaddr*)&clientaddr, &len);printf("hello, %d\n", clientfd);fds[clientfd].fd = clientfd;fds[clientfd].events = POLLIN;maxfd = clientfd;}int i = 0;for (i = sockfd + 1; i <= maxfd; i++){if (fds[i].revents & POLLIN){char buffer[128] = { 0 };int count = recv(fds[i].fd, buffer, 128, 0);if (count == 0){printf("disconnect\n");close(i);fds[i].fd = -1;fds[i].events = 0;continue;}send(i, buffer, count, 0);printf("clientfd: %d, sount: %d, lbuffer: %s\n", i, count, buffer);}}
}

3.1 epoll

在支持 Linux 的函数中，epoll 是最高效的。还是上面的比方，epoll 则是在一定程度上结合了前两者的优点，并且底层使用红黑树，查找速度更快。

演示如下：

int epfd = epoll_create(1);struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = sockfd;epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);struct epoll_event events[1024] = { 0 };while (1)
{int nready = epoll_wait(epfd, events, 1024, -1);int i = 0;for (i = 0; i < nready; i++){int curfd = events[i].data.fd;if (sockfd == curfd){struct sockaddr_in clientaddr;socklen_t len = sizeof(struct sockaddr_in);int clientfd = accept(sockfd, (struct sockaddr*)&clientaddr, &len);ev.events = EPOLLIN;ev.data.fd = clientfd;epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);printf("clientfd: %d\n", clientfd);}else if (events[i].events & EPOLLIN){char buffer[10] = { 0 }; // 只要有数据就会一直触发，因此会回复多次int count = recv(curfd, buffer, 10, 0);if (count == 0){printf("disconnect\n");close(curfd);epoll_ctl(epfd, EPOLL_CTL_DEL, curfd, NULL);continue;}send(curfd, buffer, count, 0);printf("clientfd: %d, sount: %d, buffer: %s\n", curfd, count, buffer);}}}

3.2 基于 epoll 模拟实现面对事件的 reactor 的底层原理

定义结构体，为了方便，直接把 epfd 定位全局变量

#define BUFFER_LENGTH	1024typedef int (* RCALLBACK)(int fd);// save buffer data
struct conn_item
{int fd;char rbuffer[BUFFER_LENGTH];int rlen;char wbuffer[BUFFER_LENGTH];int wlen;union // 联合，在后续代码中会用到{RCALLBACK accept_callback;RCALLBACK recv_callback;}recv_t;RCALLBACK send_callback;
};struct conn_item connlist[1024];#if 1
int epfd;
#elif
struct reactor
{int epfd;struct conn_item connlist[1024];
};
#endif

reactor 的模拟借助以下回调函数实现，可以简化代码。

int set_cb(int fd, int event, int flag);
int accept_cb(int fd);
int recv_cb(int fd);
int send_cb(int fd);

具体实现如下，体现出面对事件的处理思想。

/*
1. listenfd 触发 EPOLLIN 事件 -> 执行 accept_cb
2. client 触发 EPOLLIN 事件 -> recv_cb
3. client 触发 EPOLLOUT 事件 -> send_cb
*/// ADD: flag == 1 else 0
int set_cb(int fd, int event, int flag)
{if (flag){struct epoll_event ev;ev.events = event;ev.data.fd = fd;epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);}else{struct epoll_event ev;ev.events = event;ev.data.fd = fd;epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);}
}int accept_cb(int fd)
{struct sockaddr_in clientaddr;socklen_t len = sizeof(struct sockaddr_in);int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);set_cb(clientfd, EPOLLIN, 1);// set connlistconnlist[clientfd].fd = clientfd;memset(connlist[clientfd].wbuffer, 0, BUFFER_LENGTH);connlist[clientfd].wlen = 0;memset(connlist[clientfd].rbuffer, 0, BUFFER_LENGTH);connlist[clientfd].rlen = 0;// set callbackconnlist[clientfd].recv_t.recv_callback = recv_cb;connlist[clientfd].send_callback = send_cb;printf("clientfd: %d\n", clientfd);return clientfd;
}int recv_cb(int fd)
{char * buffer = connlist[fd].rbuffer;int index = connlist[fd].rlen;int count = recv(fd, buffer + index, BUFFER_LENGTH - index, 0);if (count == 0){printf("disconnect\n");epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);close(fd);return -1;}connlist[fd].rlen += count;#if ENABLE_HTTP_RESPONSE// 此处可以自行修改，使对应不同输入实现特定输出，如 http 相应：
http_response(&connlist[fd]);#else// 不做处理直接发送返回
memcpy(connlist[fd].wbuffer, connlist[fd].rbuffer, connlist[fd].rlen);
connlist[fd].wlen = connlist[fd].rlen;#endif// eventset_cb(fd, EPOLLOUT, 0);return count;
}int send_cb(int fd)
{char * buffer = connlist[fd].wbuffer;int index = connlist[fd].wlen;int count = send(fd, buffer, index, 0);// 事件来一次执行一次set_cb(fd, EPOLLIN, 0);return count;
}

mainloop 部分

int main()
{int sockfd = socket(AF_INET, SOCK_STREAM, 0);struct sockaddr_in serveraddr;memset(&serveraddr, 0, sizeof(struct sockaddr_in));serveraddr.sin_family = AF_INET;serveraddr.sin_addr.s_addr = htonl(INADDR_ANY);serveraddr.sin_port = htons(2048);if (-1 == bind(sockfd, (struct sockaddr*)&serveraddr, sizeof(struct sockaddr))){perror("bind");return -1;}listen(sockfd, 10);connlist[sockfd].fd = sockfd;connlist[sockfd].recv_t.accept_callback = accept_cb;// epoll 边缘触发/*对于 IO 处理：随着时间增多会越来越复杂if（listenfd){...}else // clientfd{...}对于事件处理 -> reactor 对事件反应更缓和if (events & EPOLLIN){...}else if (events & EPOLLOUT){...}*/epfd = epoll_create(1); // int sizeset_cb(sockfd, EPOLLIN, 1);struct epoll_event events[1024] = { 0 };while (1) // main loop{int nready = epoll_wait(epfd, events, 1024, -1);int i = 0;for (i = 0; i < nready; i++){int curfd = events[i].data.fd;// 由于结构体内 accept_callback() 与 recv_callback() 共享同一块内存，故此处 if 条件判断可以省略。// if (sockfd == curfd)// {// 	// accept_cb()// 	// int clientfd = accept_cb(sockfd);// 	int clientfd = connlist[sockfd].recv_t.accept_callback(sockfd);// 	printf("client: %d\n", clientfd);// }if (events[i].events & EPOLLIN){// int count = recv_cb(curfd);int count = connlist[curfd].recv_t.recv_callback(curfd);if (count != -1)printf("recv <- clientfd: %d, count: %d, buffer: %s\n", curfd, count, connlist[curfd].rbuffer);}else if (events[i].events & EPOLLOUT){// int count = send_cb(curfd);int count = connlist[curfd].send_callback(curfd);printf("send -> clientfd: %d, count: %d, buffer: %s\n", curfd, count, connlist[curfd].wbuffer);}}}return 0;
}

附：本文的 http_response() 函数定义，以供测试使用。

#include <time.h>typedef struct conn_item connection_t;int http_response(connection_t *conn)
{const char *html_body = "<html><head><title>chipen.com</title></head><body><h1>chipen</h1></body></html>";int content_length = strlen(html_body);// 生成符合 HTTP 标准格式的 Date 字符串time_t now = time(NULL);struct tm *gmt = gmtime(&now);char date_str[128];strftime(date_str, sizeof(date_str), "Date: %a, %d %b %Y %H:%M:%S GMT\r\n", gmt);// 构建完整的 HTTP 响应conn->wlen = snprintf(conn->wbuffer, BUFFER_LENGTH,"HTTP/1.1 200 OK\r\n""Content-Type: text/html\r\n""Content-Length: %d\r\n""Accept-Ranges: bytes\r\n""%s""\r\n"  // Header 与 Body 的分隔符"%s",   // HTML 内容content_length,date_str,html_body);return conn->wlen;
}