当前位置：首页 > news >正文

Linux高性能服务器编程——ch9笔记

news 来源：原创 2024/5/21 1:15:26

第9章 I/O复用

同时监听多个文件描述符，但本身是阻塞的。

9.1 select系统调用

在一段指定时间内，监听用户感兴趣的文件描述符上的可读、可写和异常等事件是否就绪。
:::tips
int select(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, struct timeval* timeout);
:::
socket可读情况：
1）socket内核接收缓存区中的字节数大于或等于其低水位标记SO_RCVLOWAT。此时可以无阻塞地读该socket，并且读操作返回的字节数大于0。
2）socket通信的对方关闭连接。此时对该socket的读操作将返回0。
3）监听socket上有新的连接请求。
4）socket上有未处理的错误。此时可以使用getsockopt来读取和清除该错误。
socket可写情况：
1）socket内核发送缓存区中的可用字节数大于或等于其低水位标记SO_SNDLOWAT。此时可以无阻塞地写该socket，并且写操作返回的字节数大于0。2）socket的写操作被关闭。对写操作被关闭的socket执行写操作将触发一个SIGPIPE信号。
3）socket使用非阻塞connect连接成功或者失败（超时）之后。
4）socket上有未处理的错误。此时可以使用getsockopt来读取和清除该错误。
select能处理的异常情况只有一种：socket上接收到带外数据。

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>int main(int argc, char *argv[])
{if (argc <= 2){printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);printf("ip is %s and port is %d\n", ip, port);int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int listenfd = socket(PF_INET, SOCK_STREAM, 0);assert(listenfd >= 0);ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));assert(ret != -1);ret = listen(listenfd, 5);assert(ret != -1);struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);int connfd = accept(listenfd, (struct sockaddr *)&client_address, &client_addrlength);if (connfd < 0){printf("errno is %d\n", errno);close(listenfd);}char buf[1024];fd_set read_fds;fd_set exception_fds;FD_ZERO(&read_fds);FD_ZERO(&exception_fds);while(1){memset(buf, '\0', sizeof(buf));/* 每次调用select前都要重新再read_fds和exception_fds中设置文件描述符* connfd, 因为事件发生之后，文件描述符集合将被内核修改 */FD_SET(connfd, &read_fds);FD_SET(connfd, &exception_fds);ret = select(connfd + 1, &read_fds, NULL, &exception_fds, NULL);if (ret < 0){printf("selection failure\n");break;}/* 对于可读事件，采用普通的recv函数读取数据 */if (FD_ISSET(connfd, &read_fds)){ret = recv(connfd, buf, sizeof(buf) - 1, 0);if (ret <= 0){break;}printf("get %d bytes of normal data: %s\n", ret, buf);}/* 对于异常事件，采用MSG_OOB标志的recv函数读取带外数据 */else if (FD_ISSET(connfd, &exception_fds)){ret = recv(connfd, buf, sizeof(buf) - 1, MSG_OOB);if (ret <= 0){break;}printf("get %d bytes if oob data: %s\n", ret, buf);}}close(connfd);close(listenfd);return 0;
}

9.2 poll系统调用

指定时间内轮询一定数量的文件描述符，以测试其中是否有就绪者。
:::tips
int poll(struct pollfd* fds, nfds_t nfds, int timeout);
:::

9.3 epoll系列系统调用

epoll使用一组函数来完任务，把用户关心的文件描述符上的事件放在内核里的事件表中，而不是传参。需要使用一个额外的文件描述符（epoll_create函数创建），来唯一标识内核中的事件表（用epoll_ctl函数操作事件表）。
epoll系列系统调用的主要接口是epoll_wait函数，它在一段超时时间内等待一组文件描述符上的事件，返回就绪文件描述符个数。
:::tips
int epoll_wait(int epfd, struct epoll_event* events, int maxevents, int timeout);
:::
events数组只用于输出epoll_wait检测到的就绪事件，而不像select和poll的数组参数那样既用于传入用户注册的事件，又用于输出内核检测到的就绪事件。这就极大地提高了应用程序索引就绪文件描述符的效率。
epoll对文件描述符操作的模式：LT（电平触发，应用程序可以不立即处理事件，epoll_wait会一直通告直至事件被处理）和ET（边沿触发，应用程序必须立即处理，后续不再通知，效率高）。

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <pthread.h>#define MAX_EVENT_NUMBER 1024
#define BUFFER_SIZE 10/* 将文件描述符设置成为非阻塞的 */
int setnonblocking(int fd)
{int old_option = fcntl(fd, F_GETFL);int new_option = old_option | O_NONBLOCK;fcntl(fd, F_SETFL, new_option);return old_option;
}/* 将文件描述符fd上的EPOLLIN注册到epollfd指示的epoll内核事件表中，* 参数enable_et指定是否对fd启用ET模式 */
void addfd(int epollfd, int fd, bool enable_et)
{epoll_event event;event.data.fd = fd;event.events = EPOLLIN;if (enable_et){event.events |= EPOLLET;}epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);setnonblocking(fd);
}/* LT模式的工作流程 */
void lt(epoll_event *events, int number, int epollfd, int listenfd)
{char buf[BUFFER_SIZE];for (int i = 0; i < number; i++){int sockfd = events[i].data.fd;if (sockfd == listenfd){struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);int connfd = accept(listenfd, (struct sockaddr *)&client_address,&client_addrlength);addfd(epollfd, connfd, false);  /* 对connfd禁用ET模式 */}else if (events[i].events & EPOLLIN){/* 只要socket读缓存中还有未读出的数据， 这段代码就被触发 */printf("event trigger once\n");memset(buf, '\0', BUFFER_SIZE);int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);if (ret <= 0){close(sockfd);continue;}printf("get %d bytes of content: %s\n", ret, buf);}else{printf("something else happened\n");}}
}/* ET模式的工作流程 */
void et(epoll_event *events, int number, int epollfd, int listenfd)
{char buf[BUFFER_SIZE];for (int i = 0; i < number; i++){int sockfd = events[i].data.fd;if (sockfd == listenfd){struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);int connfd = accept(listenfd, (struct sockaddr *)&client_address,&client_addrlength);addfd(epollfd, connfd, true);   /* 对connfd开启ET模式 */}else if (events[i].events & EPOLLIN){/* 这段代码不会被重复触发，所以我们循环读取数据，以确保把* socket读缓存中的所有数据读出 */printf("event trigger once\n");while(1){memset(buf, '\0', BUFFER_SIZE);int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);if (ret < 0){/* 对于非阻塞IO，下面的条件成立表示数据已经全部读取完毕。* 此后，epoll就能再次触发sockfd上的EPOLLIN事件,* 以驱动下一次读操作 */if ((errno == EAGAIN) || (errno == EWOULDBLOCK)){printf("read later\n");break;}close(sockfd);break;}else if (ret == 0){close(sockfd);}else{printf("get %d bytes of content: %s\n", ret, buf);}}}else{printf("something else happened\n");}}
}int main(int argc, char *argv[])
{if (argc <= 2){printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int listenfd = socket(PF_INET, SOCK_STREAM, 0);assert(listenfd >= 0);ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));assert(ret != -1);ret = listen(listenfd, 5);assert(ret != -1);epoll_event events[MAX_EVENT_NUMBER];int epollfd = epoll_create(5);assert(epollfd != -1);addfd(epollfd, listenfd, true);while(1){int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);if (ret < 0){printf("epoll failure\n");break;}//lt(events, ret, epollfd, listenfd);     /* 使用LT模式 */et(events, ret, epollfd, listenfd);   /* 使用ET模式 */}close(listenfd);return 0;
}

EPOLLONESHOT事件：期望的是一个socket连接在任一时刻都只被一个线程处理（避免竞争，线程切换，死锁，数据有序）。
对于注册了EPOLLONESHOT事件的文件描述符，操作系统最多触发其上注册的一个可读、可写或者异常事件，且只触发一次。除非使用epoll_ctl函数重置该文件描述符上注册的EPOLLONESHOT事件。注册了EPOLLONESHOT事件的socket一旦被某个线程处理完毕，该线程就应该立即重置这个socket上的EPOLLONESHOT事件，以确保这个socket下一次可读时，其EPOLLIN事件能被触发，进而让其他工作线程有机会继续处理这个socket。

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <pthread.h>#define MAX_EVENT_NUMBER 1024
#define BUFFER_SIZE 1024struct fds
{int epollfd;int sockfd;
};int setnonblocking(int fd)
{int old_option = fcntl(fd, F_GETFL);int new_option = old_option | O_NONBLOCK;fcntl(fd, F_SETFL, new_option);return old_option;
}/* 将fd上的EPOLLIN和EPOLLET事件注册到epollfd指示的epoll内核事件表中，* 参数oneshot指定是否注册fd上的EPOLLONESHOT事件 */
void addfd(int epollfd, int fd, bool oneshot)
{epoll_event event;event.data.fd = fd;event.events = EPOLLIN | EPOLLET;if (oneshot){event.events |= EPOLLONESHOT;}epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);setnonblocking(fd);
}/* 重置fd上的事件, 这样操作之后，尽管fd上的EPOLLONESHOT事件被注册，* 但是操作系统仍然会触发fd上的EPOLLIN事件，且只触发一次 */
void reset_oneshot(int epollfd, int fd)
{epoll_event event;event.data.fd = fd;event.events = EPOLLIN | EPOLLET | EPOLLONESHOT;epoll_ctl(epollfd, EPOLL_CTL_MOD, fd, &event);
}/* 工作线程 */
void *worker(void *arg)
{int sockfd = ((fds *)arg)->sockfd;int epollfd = ((fds *)arg)->epollfd;printf("start new thread to receive data on fd: %d\n", sockfd);char buf[BUFFER_SIZE];memset(buf, '\0', BUFFER_SIZE);/* 循环读取sockfd上的数据，直到遇到EAGAIN错误 */while (1){int ret = recv(sockfd, buf, BUFFER_SIZE - 1, 0);if (ret == 0){close(sockfd);printf("foreiner closefd the connection\n");break;}else if (ret < 0){if (errno == EAGAIN){reset_oneshot(epollfd, sockfd);printf("read later\n");break;}}else{printf("get content: %s\n", buf);/* 休眠5s，模拟数据处理过程 */sleep(5);}}printf("end thread receiving data on fd:%d\n", sockfd);return NULL;
}int main(int argc, char *argv[])
{if (argc <= 2){printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int listenfd = socket(PF_INET, SOCK_STREAM, 0);assert(listenfd >= 0);ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));assert(ret != -1);ret = listen(listenfd, 5);assert(ret != -1);epoll_event events[MAX_EVENT_NUMBER];int epollfd = epoll_create(5);assert(epollfd != -1);/* 注意，监听socket listenfd上是不能注册EPOLLONESHOT事件的，* 否则应用程序只能处理一个客户连接，因为后续的客户连接请求将* 不再触发listenfd上的EPOLLIN事件 */addfd(epollfd, listenfd, false);while (1){int ret = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);if (ret < 0){printf("epoll failure\n");break;}for (int i = 0; i < ret; i++){int sockfd = events[i].data.fd;if (sockfd == listenfd){printf("new client connection ....\n");struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);int connfd = accept(listenfd, (struct sockaddr *)&client_address,&client_addrlength);/* 对每个非监听文件描述符都注册EPOLLONESHOT事件 */addfd(epollfd, connfd, true);}else if (events[i].events & EPOLLIN){printf("new data from %d clients ....\n", sockfd);pthread_t thread;fds fds_for_new_worker;fds_for_new_worker.epollfd = epollfd;fds_for_new_worker.sockfd = sockfd;/* 新启动一个工作线程为sockfd服务 */pthread_create(&thread, NULL, worker, (void *)&fds_for_new_worker);}else{printf("somethinf else happened\n");}}}close(listenfd);return 0;
}

9.4 三组I/O复用函数的比较

回调：内核检测到就绪的文件描述符时，将触发回调函数，将该文件描述符上对应的事件插入内核就绪事件队列，内核最后在适当的时机将该就绪事件队列中的内容拷贝到用户空间。因此epdl_wait无须轮询整个文件描述符集合来检测哪些事件已经就绪。

9.5 I/O复用的高级应用一：非阻塞connect

EINPROGRESS错误发生在对非阻塞的socket调用connect，而连接又没有立即建立时。可以调用select、poll等函数来监听这个连接失败的socket上的可写事件，当函数返回后，再利用getsockopt来读取错误码并清除该socket上的错误，如果错误码是0，表示连接成功建立，否则连接失败。通过上面描述的非阻塞connect方式，就能同时发起多个连接并一起等待。

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <assert.h>
#include <stdio.h>
#include <time.h>
#include <errno.h>
#include <fcntl.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <string.h>#define BUFFER_SIZE 1024int setnonblocking(int fd)
{int old_option = fcntl(fd, F_GETFL);int new_option = old_option | O_NONBLOCK;fcntl(fd, F_SETFL, new_option);return old_option;
}/* 超时连接函数，参数分别是服务器IP地址、端口号和超时时间(毫秒).* 函数成功时返回已经处于连接状态的socket, 失败则返回-1 */
int unblock_connect(const char *ip, int port, int time)
{int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int sockfd = socket(PF_INET, SOCK_STREAM, 0);int fdopt = setnonblocking(sockfd);ret = connect(sockfd, (struct sockaddr *)&address, sizeof(address));if (ret == 0){/* 如果连接成功，则恢复sockfd的属性， 并立即返回 */printf("connect with server immediately\n");fcntl(sockfd, F_SETFL, fdopt);return sockfd;}else if (errno != EINPROGRESS){/* 如果连接没有立即建立, 那么只有当errno是EINPROGRESS时才* 表示连接还在进行，否则出错返回 */printf("unblock connect not support\n");return -1;}fd_set readfds;fd_set writefds;struct timeval timeout;FD_ZERO(&readfds);FD_SET(sockfd, &writefds);timeout.tv_sec = time;timeout.tv_usec = 0;ret = select(sockfd + 1, NULL, &writefds, NULL, &timeout);if (ret <= 0){/* select 超时或者出错，立即返回 */printf("connection time out\n");close(sockfd);return -1;}if (!FD_ISSET(sockfd, &writefds)){printf("no events on sockfd found\n");close(sockfd);return -1;}int error = 0;socklen_t length = sizeof(error);/* 调用getsockopt来获取并清除sockfd上的错误 */if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &length) < 0){printf("get socket option failed\n");close(sockfd);return -1;}/* 错误号不为0表示连接出错 */if (error != 0){printf("connection failed after select with the error: %d\n", error);close(sockfd);return -1;}/* 连接成功 */printf("connection ready after select with the socker: %d\n", sockfd);fcntl(sockfd, F_SETFL, fdopt);return sockfd;
}int main(int argc, char *argv[])
{if (argc <= 2){printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);int sockfd = unblock_connect(ip, port, 10);if (sockfd < 0){return 1;}close(sockfd);return 0;
}

9.6 I/O复用的高级应用二：聊天室程序

I/O复用同时处理网络连接和用户输入。
客户端程序使用poll同时监听用户输入和网络连接，并利用splice函数将用户输入内容直接定向到网络连接上以发送之，从而实现数据零拷贝，提高了程序执行效率，

#define _GNU_SOURCE 1
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <poll.h>
#include <fcntl.h>#define BUFFER_SIZE 64int main(int argc, char *argv[])
{if (argc <= 2){printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);struct sockaddr_in server_address;bzero(&server_address, sizeof(server_address));server_address.sin_family = AF_INET;inet_pton(AF_INET, ip, &server_address.sin_addr);server_address.sin_port = htons(port);int sockfd = socket(PF_INET, SOCK_STREAM, 0);assert(sockfd >= 0);if (connect(sockfd, (struct sockaddr *)&server_address, sizeof(server_address)) < 0){printf("connection failed\n");close(sockfd);return 1;}pollfd fds[2];/* 注册文件描述符0(标准输入)和文件描述符sockfd上的可读事件 */fds[0].fd = 0;fds[0].events = POLLIN;fds[0].revents = 0;fds[1].fd = sockfd;fds[1].events = POLLIN | POLLRDHUP;fds[1].revents = 0;char read_buf[BUFFER_SIZE];int pipefd[2];int ret = pipe(pipefd);assert(ret != -1);while (1){ret = poll(fds, 2, -1);if (ret < 0){printf("poll failure\n");break;}if (fds[1].revents & POLLRDHUP){printf("server close the connection\n");break;}else if (fds[1].revents & POLLIN){memset(read_buf, '\0', BUFFER_SIZE);recv(fds[1].fd, read_buf, BUFFER_SIZE - 1, 0);printf("%s\n", read_buf);}if (fds[0].revents & POLLIN){/* 使用splice将用户输入的数据直接写到sockfd上(零拷贝) */ret = splice(0, NULL, pipefd[1], NULL, 32768,SPLICE_F_MORE | SPLICE_F_MOVE);ret = splice(pipefd[0], NULL, sockfd, NULL, 32768,SPLICE_F_MORE | SPLICE_F_MOVE);}}close(sockfd);return 0;
}

服务器程序使用poll同时管理监听socket和连接socket，并且使用牺牲空间换取时间的策略来提高服务器性能。

#define _GNU_SOURCE 1
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <poll.h>#define USER_LIMIT 5    /* 最大用户数量 */
#define BUFFER_SIZE 64  /* 读缓冲区的大小 */
#define FD_LIMIT 65535  /* 文件描述符数量限制 *//* 客户数据：客户端socket地址、待写到客户端的数据的位置、从客户端读入的数据 */
struct client_data
{sockaddr_in address;char *write_buf;char buf[BUFFER_SIZE];
};int setnonblocking(int fd)
{int old_option = fcntl(fd, F_GETFL);int new_option = old_option | O_NONBLOCK;fcntl(fd, F_SETFL, new_option);return old_option;
}int main(int argc, char *argv[])
{if (argc <= 2){printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int listenfd = socket(PF_INET, SOCK_STREAM, 0);assert(listenfd >= 0);ret = bind(listenfd, (struct sockaddr *)&address, sizeof(address));assert(ret != -1);ret = listen(listenfd, 5);assert(ret != -1);/* 创建users数据，分配FD_LIMIT个client_data对象，可以预期：每个可能的socket连接* 都可以获得一个这样的对象，并且socket的值可以直接用来索引(作为数组的下标)socket* 连接对应的client_data对象，这是将socket和客户端关联的简单而高效的方式 */client_data *users = new client_data[FD_LIMIT];/*尽管我们分配了足够多的client_data对象，但为了提高poll的性能，仍然有必要限制用户的数据 */pollfd fds[USER_LIMIT + 1];int user_counter = 0;for (int i = 0; i <= USER_LIMIT; ++i){fds[i].fd = -1;fds[i].events = 0;}fds[0].fd = listenfd;fds[0].events = POLLIN | POLLERR;fds[0].revents = 0;while (1){ret = poll(fds, user_counter + 1, -1);if (ret < 0){printf("poll failure\n");break;}for (int i = 0; i < user_counter + 1; ++i){if ((fds[i].fd == listenfd) && (fds[i].revents & POLLIN)){struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);int connfd = accept(listenfd, (struct sockaddr *)&client_address,&client_addrlength);if (connfd < 0){printf("errno is : %d\n", errno);continue;}/* 如果请求太多，则关闭新到的连接 */if (user_counter >= USER_LIMIT){const char *info = "too many users\n";printf("%s", info);send(connfd, info, strlen(info), 0);continue;}/* 对于新的连接，同时修改fds和users数组，前文已经提到，users[connfd]* 对应于新连接文件描述符connfd的客户数据 */user_counter++;users[connfd].address = client_address;setnonblocking(connfd);fds[user_counter].fd = connfd;fds[user_counter].events = POLLIN | POLLRDHUP | POLLERR;fds[user_counter].revents = 0;printf("comes a new user, now have %d users\n", user_counter);}else if (fds[i].revents & POLLERR){printf("get error from %d\n", fds[i].fd);char errors[100];memset(errors, '\0', 100);socklen_t length = sizeof(errors);if (getsockopt(fds[i].fd, SOL_SOCKET, SO_ERROR, &errors,&length) < 0){printf("get socket option failed\n");}continue;}else if (fds[i].revents & POLLRDHUP){/* 如果客户端关闭连接，则服务器也关闭对应的连接，并将用户总数减1 */users[fds[i].fd] = users[fds[user_counter].fd];close(fds[i].fd);fds[i] = fds[user_counter];i--;user_counter--;printf("a client left\n");}else if (fds[i].revents & POLLIN){int connfd = fds[i].fd;memset(users[connfd].buf, '\0', BUFFER_SIZE);ret = recv(connfd, users[connfd].buf, BUFFER_SIZE - 1, 0);printf("get %d bytes of client data %s from %d\n", ret,users[connfd].buf, connfd);if (ret < 0){/* 如果读操作出错，则关闭连接 */if (errno != EAGAIN){close(connfd);users[fds[i].fd] = users[fds[user_counter].fd];fds[i] = fds[user_counter];i--;user_counter--;}}else if (ret == 0){}else{/* 如果接收到客户数据，则通知其他socket连接准备写数据 */for (int j = 1; j <= user_counter; ++j){if (fds[j].fd == connfd){continue;}fds[j].events |= ~POLLIN;fds[j].events |= POLLOUT;users[fds[j].fd].write_buf = users[connfd].buf;}}}else if (fds[i].revents & POLLOUT){int connfd = fds[i].fd;if (!users[connfd].write_buf){continue;}ret = send(connfd, users[connfd].write_buf,strlen(users[connfd].write_buf), 0);users[connfd].write_buf = NULL;/* 写完数据后需要重新注册fds[i]上的可读事件 */fds[i].events |= ~POLLOUT;fds[i].events |= POLLIN;}}}delete[] users;close(listenfd);return 0;
}

9.7 I/O复用的高级应用三：同时处理TCP和UDP服务

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include <pthread.h>#define MAX_EVENT_NUMBER 1024
#define TCP_BUFFER_SIZE 512
#define UDP_BUFFER_SIZE 1024int setnonblocking(int fd)
{int old_option = fcntl(fd, F_GETFL);int new_option = old_option | O_NONBLOCK;fcntl(fd, F_SETFL, new_option);return old_option;
}void addfd(int epollfd, int fd)
{epoll_event event;event.data.fd = fd;event.events = EPOLLIN | EPOLLET;epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event);setnonblocking(fd);
}int main(int argc, char *argv[])
{if (argc <= 2){printf("usage: %s ip_address port_number\n", basename(argv[0]));return 1;}const char *ip = argv[1];int port = atoi(argv[2]);int ret = 0;struct sockaddr_in address;bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);/* 创建TCP socket，并将其绑定到端口port上 */int tcpfd = socket(PF_INET, SOCK_STREAM, 0);assert(tcpfd >= 0);ret = bind(tcpfd, (struct sockaddr *)&address, sizeof(address));assert(ret != -1);ret = listen(tcpfd, 5);assert(ret != -1);/* 创建UDP socket, 并将其绑定到端口port上 */bzero(&address, sizeof(address));address.sin_family = AF_INET;inet_pton(AF_INET, ip, &address.sin_addr);address.sin_port = htons(port);int udpfd = socket(PF_INET, SOCK_DGRAM, 0);assert(udpfd > 0);ret = bind(udpfd, (struct sockaddr *)&address, sizeof(address));assert(ret != -1);epoll_event events[MAX_EVENT_NUMBER];int epollfd = epoll_create(5);assert(epollfd != -1);/* 注册TCP socket 和UDP socket上的可读事件 */addfd(epollfd, tcpfd);addfd(epollfd, udpfd);while (1){int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1);if (number < 0){printf("epoll failure\n");break;}for (int i = 0; i < number; ++i){int sockfd = events[i].data.fd;if (sockfd == tcpfd)  // 新的TCP连接请求{struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);int connfd = accept(tcpfd, (struct sockaddr *)&client_address,&client_addrlength);addfd(epollfd, connfd);}else if (sockfd == udpfd)  // UDP套接字可读{char buf[UDP_BUFFER_SIZE];memset(buf, '\0', UDP_BUFFER_SIZE);struct sockaddr_in client_address;socklen_t client_addrlength = sizeof(client_address);ret = recvfrom(udpfd, buf, UDP_BUFFER_SIZE - 1, 0,(struct sockaddr *)&client_address, &client_addrlength);if (ret > 0){sendto(udpfd, buf, UDP_BUFFER_SIZE - 1, 0,(struct sockaddr *)&client_address, client_addrlength);}}else if (events[i].events & EPOLLIN)  // TCP套接字可读{char buf[TCP_BUFFER_SIZE];while (1){memset(buf, '\0', TCP_BUFFER_SIZE);ret = recv(sockfd, buf, TCP_BUFFER_SIZE - 1, 0);if (ret < 0){if ((errno == EAGAIN) || (errno == EWOULDBLOCK)){break;}close(sockfd);break;}else if (ret == 0){close(sockfd);}else{send(sockfd, buf, ret, 0);}}}else{printf("something else happened\n");}}}close(tcpfd);close(udpfd);return 0;
}

在UDP中，因为它是无连接的，socket通常会被不断标记为可读，因为随时都可能会接收到数据包。因此，在这个条件下，程序在接收到UDP数据包时会执行相应的操作。

9.8 超级服务xinetd

同时管理着多个子服务（通过子配置文件设置），即监听多个端口。

文件描述符0、1、2：标准输入、标准输出和标准错误，防止任何从父进程（xinetd）继承下来的不必要的输入输出流干扰telnet会话的正常通信。
将socket文件描述符dup（复制）到它们上面：telnet服务器程序将网络连接上的输入当作标准输入，井把标准输出定向到同一个网络连接上。