1. I/O 复用功能

I/O 复用能同时监听多个文件描述符。
I/O 复用本身是阻塞的。

当有多个文件描述符同时就绪时：若不采取额外措施，程序就只能按顺序一次处理其中的每一个文件描述符，这使得服务器程序看起来是串行工作的。若要实现并发：只能通过多进程或多线程等并发手段。

Linux 实现 I/O 复用的系统调用主要有 select、poll 和 epoll。

2.文件描述符就绪条件

2.1 下列情况下文件描述符 可读

1. socket 内核接收缓存区中的字节数大于或等于其低水位标记 SO_RCVLOWAT.此时可以无阻塞的读该 socket，并且读操作返回的字节数 大于 0.2. socket 通信的对方关闭连接。此时对该 socket 读操作将返回 0.3.监听 socket 上有新的连接请求4.	socket 上有未处理的错误。此时我们可以使用 getsockopt 来读取和清除该错误。

2.2 下列情况下文件描述符 可写

1. socket 内核发送缓存区中的可用字节数大于或等于其低水位标记 SO_SNDLOWAT.此时可以无阻塞的 写 该 socket，并且写操作返回的字节数大于 0.2. socket 的写操作被关闭。对 写操作被关闭的 socket 执行写操作将触发一个 SIGPIPE 信号。3. socket 使用非阻塞 connect 连接成功或失败（超时）后。4.  socket 上有未处理的错误。此时我们可以使用 getsockopt 来读取和清除该错误。

2.3 下列情况下文件描述符 出现异常

select 能处理的异常只有一种：1. socket 上接收到带外数据。详见下文 3.2 select 实例。

3. select 系统调用

select 系统调用用途在一段时间内，监听用户感兴趣的文件描述符上的 可读、可写 和 时间。

3.1 select API

原型如下：

man 2 select#include <sys/select.h>
int select(int nfds,fd_set * readfds, fd_set * writefds, fd_set * execptfds, struct timeval * timeout);
参数：nfds：指定被监听的文件描述符的参数。通常被设置为 select 监听的所有文件描述符中的最大值 加 1，因为文件描述符是从 0 开始计数的。readfds：指向 可读事件对应的文件描述符集合。writefds：指向 可写事件对应的文件描述符集合。exceptfds：指向 异常等事件对应的文件描述符集合。timeout：用来设置 select 函数的超时时间。timeval 结构指针类型，采用指针参数是因为内核将修改它以告诉应用程序 select 等待了多久。调用失败时，timeout的值是不确定的。返回值：成功：返回就绪（可读、可写、异常）文件描述符总数。若在超时事件内没有任何文件描述符就绪，select 返回 0.失败：返回 -1并设置errno。扩展：select 等待期间，程序收到信号，则select 立即返回 -1，并设置 errno为 EINTR.fd_set  结构体定义：
#include <typesizes.h>
#define __FD_SETSIZE 1024#include <sys/select.h>
#define FD_SETSIZE __FD_SETSIZE
typedef long int __fd_mask;#undef __NFDBITS
#define _NFBDITS ( 8 * (int) sizeof(__fd_mask))typedef struct
{#ifdef __USE_XOPEN__fd_mask fds_bits[ __FD_SETSIZE / __NFDBITS ];#define __FDS_BITS(set) ((set)->fds_bits)#else__fd_mask __fds_bits[ __FD_SETSIZE / __NFDBITS ];#define __FDS_BITS(set) ((set)->__fds_bits)
}fd_set;详解fd_set：1.由结构体定义可见，仅包含一个整型数组，该数组每个元素每一位（bit）标记一个文件描述符。2.fd_set 能容纳的文件描述符数量由 FD_SETSIZE 指定，这就限制了 select 能同时处理的文件描述符总量。因为对位操作繁琐，故使用如下函数操作 fd_set ：#include <sys/select.h>FD_ZERO(fd_set * fdset);			//清除 fdset 的所有位FD_SET(int fd,fd_set * fdset);		//设置 fdset 的位 fdFD_CLR(int fd,fd_set * fdset);		//清除 fdset 的位 fdint FD_ISSET(int fd, fd_set * fdset);//测试 fdset 的位 fd 是否被设置timeval 结构体定义：struct timeval{long tv_sec;	//秒数long tv_usec;	//微秒数}timeval 结构体传值：1.若给 timeval 结构体两成员都传 0，则 select 将立即返回。2.若给 timeout 传 NULL，则 select 将一直堵塞，直到某个文件描述符就绪。

3.2 select 实例

socket 上接收到 普通数据 和 带外数据 都将使 select 返回，
但 socket 处于不同的就绪状态，前者 处于可读状态，后者 处于异常状态。
下例实现：

//server.c
#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]); //字符串转换为一个整数（类型为 int 型）。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);  //最后一个参数是 NULL，故此处会一直阻塞if(ret < 0){printf(" selection failure\n");break;;}//对于可读事件，采用普通的 recv 函数读取数据if( FD_ISSET(connfd, &read_fds)) //如果connfd 是 read_fds 中的一员返回非0，否则返回0   {ret = recv(connfd, buf, sizeof(buf)-1, 0);if(ret <= 0){break;;}printf("get %d bytes of normal data: %s\n",ret , buf);}else if(FD_ISSET(connfd, &exception_fds))   //如果connfd 是 exception_fds 中的一员返回非0，否则返回0 {ret = recv(connfd, buf, sizeof(buf)-1, MSG_OOB);if(ret <= 0){break;}printf("get %d bytes of oob data: %s\n",ret, buf);}}close(connfd);close(listenfd);return 0;
}

4. poll 系统调用

#include <poll.h>int poll(struct pollfd * fds, nfds_t nfds, int timeout );fds:是一个 pollfd 结构类型数组，指定所有感兴趣的文件描述符上发生的 可读、可写、异常等事件。pollfd ：struct pollfd{int fd;			//文件描述符short events;	//注册的事件short revents;	//实际发生的事件，由内核填充}fd：指定文件描述符events：告诉 poll 监听 fd 上的哪些事件，他是一系列事件的按位或。（可能发生的事件见下表 9-1）revents:由内核修改，以通知应用程序 fd 上实际发生了哪些事件。（可能发生的事件见下表 9-1）nfd:指定被监听事件集合 fds 的大小。nfds_t 定义：typedef unsigned long int nfds_t;timeout:指定 poll 的超时值，单位是 毫秒。当 设为 -1 时，poll 调用将永远阻塞，直到某个事件发生。当 设为 0 时，poll 调用将立即返回。返回（和 select 相同）：成功：返回就绪（可读、可写、异常）文件描述符总数。若在超时事件内没有任何文件描述符就绪，poll 返回 0.失败：返回 -1并设置errno。	扩展：通常，应用程序要根据 recv 调用的返回值来区分 socket 上接收到的有效数据还是对方关闭连接请求，并做相应处理。Linux 内核 2.6.17 开始， GNU为 poll 增加了一个 POLLRDHUP 事件，在 socket 上接收到对方关闭连接请求后触发。

表 9-1 注意点：POLLRDNORM、POLLRDBAND、POLLWRNORM、POLLWRBAND 由 XOPEN 规范定义。它们实际上是将 POLLIN 事件和 POLLOUT 事件分的更细致，以区别对待普通数据和优先数据。但 Linux 并不完全支持它们。

5. epoll 系统调用

与 select、poll 差异很大。epoll 使用一系列函数来完成任务。

//1.	epoll 需要使用一个额外文件描述符，来唯一表示内核中这个事件表，创建方法如下函数：
#include <sys/epoll.h>
int epoll_create(int size);size：给内核一个提示，告诉内核事件表需要多大。返回:文件描述符，用于其他所有 epoll 系统调用的第一个参数，以指定要访问的内核事件表。//2. 下面函数 操作 epoll 内核事件表
#include <sys/epoll.h>
int epoll_ctl(int epfd, int op, int fd, struct epoll_event * event);fd：要操作的文件描述符。op：指定操作类型，可操作性类型如下三种：EPOLL_CTL_ADD:	往事件表上 注册 fd上的事件。EPOLL_CTL_MOD:	修改 fd 上的注册事件。EPOLL_CTL_DEL:	删除 fd 上的注册事件。event：指定事件。epoll_event 结构体指针：struct epoll_event{__uint32_t events;		//epoll 事件epoll_data_t data;		//用户数据}events：描述事件类型。epoll 事件的类型宏 和 poll 基本相同（详见下表 9-1）。epoll 事件类型宏是在 poll 类型宏前加上 “E”，如 可读事件 EPOLLIN。epoll 有两个额外的事件类型————EPOLLET 和 EPOLLONESHOT .data：用于存储用户数据。epoll_data_t 是一个联合体，定义如下：typedef union epoll_data{void * ptr;int fd;unit32_t u32;unit64_t u64;} epoll_data_t;fd:其 4 个成员中使用最多的是 fd,它指定事件所从属的目标文件描述符。ptr：可用来指定和 fd 相关的用户数据。扩展：因为是联合体，故不能同时 得到 fd 和 ptr，若要实现 文件描述符 和 用户数据关联起来，以实现快速数据访问，只能用其他手段。如，放弃 fd成员，在 ptr 指向的用户数据中包含 fd。返回：成功：0失败：-1 并设置 errno。//3.在超时时间内等待一组文件描述符上的事件
#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event * events, int maxevents, int timeout);epfd：内核事件表。events:只用于接收输出 epoll_wait 检测到的就绪事件。注意：不像 select 和 poll 的数组参数既用于传入用户注册的事件，又用于输出内核检测到的就绪事件（详见下图 9-2）。epoll_wait 函数如果检测到事件，就将所有就绪事件从内核事件表（epfd参数指定）中复制到它的第二个参数 events 指向的数组中。	maxevents:指定最多监听多少个事件，必须大于 0.timeout:与 poll 接口的 timeout 参数相同。指定 epoll 的超时值，单位是 毫秒。当 设为 -1 时，epoll_wait 调用将永远阻塞，直到某个事件发生。当 设为 0 时，epoll_wait 调用将立即返回。返回：成功：返回就绪的文件描述符的个数。失败：-1，并设置 errno。

表 9-1 注意点：POLLRDNORM、POLLRDBAND、POLLWRNORM、POLLWRBAND 由 XOPEN 规范定义。它们实际上是将 POLLIN 事件和 POLLOUT 事件分的更细致，以区别对待普通数据和优先数据。但 Linux 并不完全支持它们。

5.1 LT 和 ET 模式

epoll 对文件描述符操作有两种模式： 
1.LT（Level Trigger，电平触发）默认工作模式。这个模式下 epoll 相当于一个效率较高的 poll。扩展:此模式下，当 epoll_wait 检测到有事件发生并将此事件通知应用程序后，应用程序可以 不立即处理该事件，这样会导致 应用程序下一次调用 epoll_wait 时，还会再次向应用程序 通告该事件，直到该事件被处理。					2.ET（Edge Trigger，边沿触发）设置方法：需手动往 epoll 内核事件表中注册一个文件描述符上的 EPOLLET 事件，epoll 将以 ET模式操作该文件描述符。ET 模式是 epoll 的高效工作模式。扩展：1.	当 epoll_wait 检测到其上有事件发生并将此事件通知应用程序后，应用程序必须立即处理该事件，因为后续的 epoll_wait 调用将不再向应用程序通知这一事件。2.	每个使用 ET 模式的文件描述符都应该是 非阻塞的。若是阻塞的，读或写操作将会因为没有后续的事件而一直处于阻塞状态。
总结：ET 模式降低了同一个 epoll 事件被重复触发的 次数，故效率比  LT 高。

LT 和 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 * event, int number, int epollfd, int listenfd)
{char buf[BUFFER_SIZE];for (int i = 0; i < number; i++){int sockfd = event[i].data.fd;if (sockfd == listenfd){struct sockaddr_in address_client;socklen_t addresslen_client = sizeof(address_client);int connfd = accept(listenfd, (struct sockaddr*)&address_client, &addresslen_client);addfd(epollfd, connfd, false);  //对 connfd 禁用 ET}else if( event[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 happend\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 address_client;socklen_t addrlen_client = sizeof(address_client);int connfd = accept(listenfd, (struct sockaddr*)&address_client, &addrlen_client);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){//对于非阻塞 I/O，下面的条件成立表示数据已经全部读取完毕，//此后 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 butes of content: %s\n", read, buf);}}}else{printf("something else happend \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;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;
}

5.2 EPOLLONESHOT 事件

ET模式下，一个 socket 上某个事件还有可能被触发 多次：
场景：如一个 线程（或进程）在读取完某个 socket 上的数据后开始处理这些数据，而在数据处理过程中该连接上又有新的数据可读（EPOLLIN 再次被触发），此时另一个线程被唤醒来读取这些新的数据。故，出现了两个线程同时操作一个 socket 的局面。期望：任何时刻 同一个 socket 连接都只被一个线程（或进程）处理。实现方式：使用 epoll 的 EPOLLONESHOT 事件实现。对注册了 EPOLLONESHOT 事件的文件描述符：1.	操作系统最多触发其注册的一个可读、可写或异常事件，且只触发一次，除非我们使用 epoll_ctl 寒暑表重置该文件描述符上注册的 EPOLLONESHOT 事件。例：一旦被某个线程处理完毕，该线程就应该立即重置这个 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_ADD, 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 closed 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);//休眠 5 s,模拟输出处理过程sleep(5);}} printf("end thread recving data on fa: %d\n", 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 ret =0;struct sockaddr_in address;address.sin_family = AF_INET;address.sin_port = htons(port);inet_pton(AF_INET, ip, &address.sin_addr);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){struct sockaddr_in client_address;socklen_t client_addrlen = sizeof(client_address);int connfd = accept(listenfd, (struct sockaddr*)&client_address, &client_addrlen);//对每个非监听文件描述符都注册 EPOLLONESHOT 事件addfd(epollfd, connfd, true);}else if(events[i].events & EPOLLIN){pthread_t thread;fds fds_for_new_worker;fds_for_new_worker.epollfd = epollfd;fds_for_new_worker.sockfd = sockfd;pthread_create(&thread , NULL, worker,(void *)&fds_for_new_worker);}else{printf("something else happend\n");}}}close(listenfd);return 0;
}

6.select 、poll、epoll 异同

7. 利用 I/O 复用 同时处理 TCP 和 UDP 服务

实际应用中，有的服务器程序能`同时监听多个端口`。
如，超级服务 inetd 和 android 的调试服务 adbd。由 bind 系统调用来看，一个 socket 只能与一个 socket 地址绑定，即一个端口只能监听一个端口。
故，服务器要监听多个端口就必须创建多个 socket，并绑定到相应的端口。
这就可以用 I/O 复用来实现。若服务器要同时监听同一个端口上的 TCP 和 UDP请求，也需要创建两个 socket：
1.	流socket （TCP）
2.	数据报 socket	（UDP）
最后将它们绑定到一个端口上。例，如下所示回射服务器可同时处理一个端口的 TCP 和 UDP 请求。

//回射服务器源码，同时处理一个端口的 TCP 和 UDP 请求