[系统运维]I/O 复用函数，poll，epoll

I/O 复用使得程序能同时监听多个文件描述符，这对于提高程序的性能至关重要。通常， 网络程序在下列情况下需要使用 I/O 复用技术：

? TCP 服务器同时要处理监听套接字和连接套接字。

? 服务器要同时处理 TCP 请求和 UDP 请求。

? 程序要同时处理多个套接字。

? 客户端程序要同时处理用户输入和网络连接。

? 服务器要同时监听多个端口。

需要指出的是，I/O 复用虽然能同时监听多个文件描述符，但它本身是阻塞的。并且当多个文件描述符同时就绪时，如果不采取额外的措施，程序就只能按顺序依处理其中的每一个文件描述符，这使得服务器看起来好像是串行工作的。如果要提高并发处理的能力，可以配合使用多线程或多进程等编程方法。

select

1.1 select 的接口介绍

select 系统调用的用途是：在一段指定时间内，监听用户感兴趣的文件描述符的可读、可写和异常等事件。

#include <sys/select.h>

int select(int maxfd, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct ti
meval *timeout);
/*
	select 成功时返回就绪(可读、可写和异常)文件描述符的总数。如果在超时时间内没有任何文件描述符就绪，select 将返回 0。select 失败是返回-1.如果在 select 等待
期间，程序接收到信号，则 select 立即返回-1，并设置 errno 为EINTR。
	maxfd 参数指定的被监听的文件描述符的总数。它通常被设置为 select 监听的所
有文件描述符中的最大值+1
   readfds、writefds 和exceptfds 参数分别指向可读、可写和异常等事件对应的文件描述符集合。应用程序调用 select 函数时，通过这 3 个参数传入自己感兴趣的文件描述符。select 返回时，内核将修改它们来通知应用程序哪些文件描述符已经就绪
fd_set 结构如下：
	#define   FD_SETSIZE 1024
	typedef long int    fd_mask;
	#define   NFDBITS (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)
	#endif
	} fd_set;
	通过下列宏可以访问 fd_set 结构中的位：
	FD_ZERO(fd_set *fdset);	// 清除 fdset 的所有位
	FD_SET(int fd, fd_set *fdset);	// 设置 fdset 的位 fd
	FD_CLR(int fd, fd_set *fdset);	// 清除 fdset 的位 fd
	int FD_ISSET(int fd, fd_set *fdset);// 测试 fdset 的位 fd 是否被设置
	timeout 参数用来设置 select 函数的超时时间。它是一个timeval 结构类型的指
针，采用指针参数是因为内核将修改它以告诉应用程序 select 等待了多久。timeval结构的定义如下：
	struct timeval
	{
	long tv_sec; //秒数
	long tv_usec; // 微秒数
	};
	如果给timeout 的两个成员都是 0，则 select 将立即返回。如果timeout 传递NULL，则 select 将一直阻塞，直到某个文件描述符就绪
*/

socket可读的情况：

?socket内核接受缓存区中的字节数大于或等于其低水位标记SO_RCVLOWAT。此时我们可以无阻塞的读取该socket，并且读操作返回的字节数大于0。

?socket通信的对方关闭链接。此时对该socket的读操作将返回0。

?监听socket上有新的连接请求。

?socket上有未处理的错误。此时我们可以使用getsockopt来读取和清除该错误。

socket可写的情况：

?socket内核发送缓存区中的可用字节数大于或等于其低水位标记SO_RCVLOWAT。此时我们可以无阻塞的写该socket，并且写操作返回的字节数大于0。

?socket的写操作被关闭。对写操作被关闭的socket执行写操作将触发一个SIGPIPE信号。

?socket使用非阻塞connect连接成功或失败（超时）之后。

?socket上有未处理的错误。此时我们可以使用getsockopt来读取和清除该错误。

编码示例

服务端：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/select.h>

#define MAX_FD 128
#define DATALEN 1024

// 初始化服务器端的 sockfd 套接字
int InitSocket()
{
	int sockfd = socket(AF_INET, SOCK_STREAM, 0);
	if(sockfd == -1) return -1;

	struct sockaddr_in saddr;
	memset(&saddr, 0, sizeof(saddr));
	saddr.sin_family = AF_INET;
	saddr.sin_port = htons(6000);
	saddr.sin_addr.s_addr = inet_addr("127.0.0.1");

	int res = bind(sockfd, (struct sockaddr*)&saddr, sizeof(saddr));
	if(res == -1) return -1;

	res = listen(sockfd, 5);
	if(res == -1) return -1;

	return sockfd;
}

// 初始化记录服务器套接字的数组
void InitFds(int fds[], int n)
{
	int i = 0;
	for (; i < n; ++i)
	{
	fds[i] = -1;
	}
}

// 将套接字描述符添加到数组中
void AddFdToFds(int fds[], int fd, int n)
{
	int i = 0;
	for (; i < n; ++i)
	{
    	if (fds[i] == -1)
		{
			fds[i] = fd;
			break;
		}
	}
}

// 删除数组中的套接字描述符
void DelFdFromFds(int fds[], int fd, int n)
{
	int i = 0;
	for (; i < n; ++i)
	{
		if (fds[i] == fd)
		{
			fds[i] = -1;
			break;
		}
	}
}

//将数组中的套接字描述符设置到 fd_set 变量中,并返回当前最大的文件
描述符值
int SetFdToFdset(fd_set *fdset, int fds[], int n)
{
	FD_ZERO(fdset);
	int i = 0, maxfd = fds[0];
	for (; i < n; ++i)
	{
		if (fds[i] != -1)
		{
			FD_SET(fds[i], fdset);
			if (fds[i] > maxfd)
			{
				maxfd = fds[i];
			}
		}
	}

	return maxfd;
}

void GetClientLink(int sockfd, int fds[], int n)
{
	struct sockaddr_in caddr;
	memset(&caddr, 0, sizeof(caddr));
	socklen_t len = sizeof(caddr);
	int c = accept(sockfd, (struct sockaddr*)&caddr, &len);
	if (c < 0)
	{
		return;
	}

	printf("A client connection was successful\n");
	AddFdToFds(fds, c, n);
}

// 处理客户端数据
void DealClientData(int fds[], int n, int clifd)
{
	char data[DATALEN] = { 0 };
	int num = recv(clifd, data, DATALEN - 1, 0);
	if (num <= 0)
	{
		DelFdFromFds(fds, clifd, n);
		close(clifd);
		printf("A client disconnected\n");
	}
	else
	{
		printf("%d: %s\n", clifd, data);
		send(clifd, "OK", 2, 0);
	}
}

// 处理 select 返回的就绪事件
void DealReadyEvent(int fds[], int n, fd_set *fdset, int sockfd)
{
	int i = 0;
	for (; i < n; ++i)
	{
		if (fds[i] != -1 && FD_ISSET(fds[i], fdset))
		{
			if (fds[i] == sockfd)
			{
				GetClientLink(sockfd, fds, n);
			}
			else
			{
				DealClientData(fds, n, fds[i]);
			}
		}
	}
}

int main()
{
	int sockfd = InitSocket();
	assert(sockfd != -1);

	fd_set readfds;
    int fds[MAX_FD];
	InitFds(fds, MAX_FD);
	AddFdToFds(fds, sockfd, MAX_FD);

	while ( 1 )
	{
		int maxfd = SetFdToFdset(&readfds, fds, MAX_FD);
		struct timeval timeout;
		timeout.tv_sec = 2; // 秒数
		timeout.tv_usec = 0; //微秒数
		int n = select(maxfd + 1, &readfds, NULL, NULL, &timeout);
		if (n < 0)
		{
			printf("select error\n");
			break;
		}
		else if (n == 0)
		{
			printf("time out\n");
            continue;
        }
        
        DealReadyEvent(fds,MAX_FD,&readfds,sockfd);
    }
    
    exit(0);
}

客户端：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/select.h>

#define MAX_FD 128
#define DATALEN 1024

int InitSocket()
{
	int sockfd = socket(AF_INET, SOCK_STREAM, 0);
	if(sockfd == -1) return -1;

	struct sockaddr_in saddr;
	memset(&saddr, 0, sizeof(saddr));
	saddr.sin_family = AF_INET;

    saddr.sin_port = htons(6000);
	saddr.sin_addr.s_addr = inet_addr("127.0.0.1");

    int res = connect(sockfd, (struct	sockaddr*)&saddr,	sizeof(saddr));
	if(res == -1) return -1;

    return sockfd;
}

int main()
{
	int sockfd = InitSocket();
	assert(sockfd != -1);

	while ( 1 )
	{
		printf("please input: ");
		char buff[128] = { 0 };
		fgets(buff, 127, stdin);

		if (strncmp(buff, "bye", 3) == 0)
		{
			break;
		}
       
		int n = send(sockfd, buff, strlen(buff) - 1, 0);
		if (n <= 0)
		{
			printf("send error\n");
			break;
		}

		memset(buff, 0, 128);
		n = recv(sockfd, buff, 127, 0);
		if (n <= 0)
		{
			printf("recv error\n");
			break;
		}
		printf("%s\n", buff);
	}

	close(sockfd);
	exit(0);
}

poll

poll 系统调用和 select 类似，也是在指定时间内轮询一定数量的文件描述符，以测试其中是否有就绪者。

#include <poll.h>

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

/*
	poll 系统调用成功返回就绪文件描述符的总数，超时返回 0，失败返回-1
	nfds参数指定被监听事件集合fds的大小。
    timeout参数指定poll的超时值，单位是毫秒,timeout为-1时，poll 调用将永久阻塞，直到某个事件发生，timeout 为0时，poll 调用将立即返回。

	fds参数是一一个struct polfd结构类型的数组，它指定所有用户感兴趣的文件描述符上发生的可读、可写和异常等事件。pollfd 结构体定义如下:
		struct pollfd
		{
			int fd;//文件描述符
			short events; // 注册的关注事件类型
			short revents; //实际发生的事件类型，由内核填充
		};
	其中，fd成员指定文件描述符，events成员告诉poll 监听fd上的哪些事件类型。它是一系列事件的按位或，revents 成员则有内核修改，通知应用程序fd上实际发生了哪些事情。poll支持的事件类型如下：
*/

服务器端：

#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <poll.h>

#define DATALEN 1024
#define MAX_FD 128

// 初始化服务器端的 sockfd 套接字
int InitSocket()
{
	int sockfd = socket(AF_INET, SOCK_STREAM, 0);
	if(sockfd == -1) return -1;

	struct sockaddr_in saddr;
	memset(&saddr, 0, sizeof(saddr));
	saddr.sin_family = AF_INET;
	saddr.sin_port = htons(6000);
	saddr.sin_addr.s_addr = inet_addr("127.0.0.1");

	int res = bind(sockfd, (struct sockaddr*)&saddr, sizeof(saddr));
	if(res == -1) return -1;

	res = listen(sockfd, 5);
	if(res == -1) return -1;

	return sockfd;
}

void InitPollFds(struct pollfd fds[])
{
	int i = 0;
	for (; i < MAX_FD; ++i)
	{
		fds[i].fd = -1;
	}
}

void InsertFdToPollfds(struct pollfd fds[], int fd, short events)
{
	int i = 0;
	for (; i < MAX_FD; ++i)
	{
		if (fds[i].fd == -1)
		{
			fds[i].fd = fd;
			fds[i].events = events;
			break;
		}
	}
}

void GetClinetLink(struct pollfd fds[], int sockfd)
{
	struct sockaddr_in caddr;
	memset(&caddr, 0, sizeof(caddr));
	socklen_t len = sizeof(caddr);
	int c = accept(sockfd, (struct sockaddr*)&caddr, &len);
	if (c < 0)
	{
		return;
	}

	printf("A client connection was successful\n");

	InsertFdToPollfds(fds, c, POLLIN | POLLRDHUP);
}

// 处理客户端数据
int DealClientData(int clifd)
{
	char data[DATALEN] = { 0 };
	int n = recv(clifd, data, DATALEN - 1, 0);
	if(n <= 0)
	{
		return -1;
	}

	printf("%d: %s\n", clifd, data);
	send(clifd, "OK", 2, 0);
	return 0;
}

// 处理 poll 返回的就绪事件
void DealReadyEvent(struct pollfd fds[], int sockfd)
{
	int i = 0;
	for (; i < MAX_FD; ++i)
	{
		if (fds[i].fd == -1)
		{
			continue;
		}
/* POLLRDHUP 用于判断客户端程序是否关闭
	else if (fds[i].revents & POLLRDHUP)
	{
		close(fds[i].fd);
		fds[i].fd = -1;
		printf("A client disconnected\n");
		continue;
	}*/
		else if (fds[i].revents & POLLIN)
		{
			if (fds[i].fd == sockfd)
			{
				GetClinetLink(fds, sockfd);
			}
		else
		{
			if(-1 == DealClientData(fds[i].fd))
			{
				close(fds[i].fd);
				fds[i].fd = -1;
				printf("A client disconnected\n");
			}
		}
	}
	else
	{
		printf("error\n");
	}
	}
}

int main()
{
	int sockfd = InitSocket();
	assert(sockfd != -1);

	struct pollfd fds[MAX_FD];
	InitPollFds(fds);
	InsertFdToPollfds(fds, sockfd, POLLIN);

	while (1)
	{
		int n = poll(fds, MAX_FD, 2000);
		if (n < 0)
		{
			printf("poll error\n");
			continue;
		}
		else if (n == 0)
		{
			printf("timeout\n");
			continue;
		}
		else
		{
			DealReadyEvent(fds, sockfd);
		}
	}
	exit(0);
}

epoll

epoll 是 Linux 特有的 I/O 复用函数。它在实现和使用上与 select、poll 有很大差异。首先，epoll 使用一组函数来完成任务，而不是单个函数。其次，epoll 把用户关心的文件描述符上的事件放在内核里的一个事件表中。从而无需像 select 和 poll 那样每次调用都要重复传入文件描述符或事件集。但 epoll 需要使用一个额外的文件描述符，来唯一标识内核中的这个事件表。

? epoll_create()用于创建内核事件表

? epoll_ctl()用于操作内核事件表

?epoll_wait()用于在一段超时时间内等待一组文件描述符上的事件

#include <sys/epoll.h>

int epoll_create(int size);
/*
	epoll_create()成功返回内核事件表的文件描述符，失败返回-1
	size 参数现在并不起作用，只是给内核一个提示，告诉它事件表需要多大。
*/

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
/*
	epoll_ctl()成功返回 0，失败返回-1
	epfd 参数指定要操作的内核事件表的文件描述符
	fd 参数指定要操作的文件描述符
	op 参数指定操作类型：
	EPOLL_CTL_ADD 往内核事件表中注册 fd 上的事件
	EPOLL_CTL_MOD 修改 fd 上的注册事件
	EPOLL_CTL_DEL 删除 fd 上的注册事件
	event 参数指定事件，它是 epoll_event 结构指针类型，epoll_event 的定义如下：
	struct epoll_event
	{
	_uint32_t events; // epoll 事件
	epoll_data_t data; // 用户数据
	};
	其中，events 成员描述事件类型，epoll 支持的事件类型与 poll 基本相同，表示epoll 事件的宏是在 poll 对应的宏前加上‘E’，比如 epoll 的数据可读事件是EPOLLIN。但是 epoll 有两个额外的事件类型--EPOLLET 和EPOLLONESHOT。
data 成员用于存储用户数据，是一个联合体，其定义如下：
	typedef union epoll_data
	{
	void *ptr;
	int fd;
	uint32_t u32;
	uint64_t u64;
	}epoll_data_t;
	其中 fd 成员使用的最多，它指定事件所从属的目标文件描述符。
*/

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
/*
	epoll_wait()成功返回就绪的文件描述符的个数，失败返回-1，超时返回 0
	epfd 参数指定要操作的内核事件表的文件描述符
	events 参数是一个用户数组，这个数组仅仅在epoll_wait 返回时保存内核检测到的所有就绪事件，而不像 select 和 poll 的数组参数那样既用于传入用户注册的事件，又用于输出内核检测到的就绪事件。这就极大地提高了应用程序索引就绪文件描述符的效率。
	maxevents 参数指定用户数组的大小，即指定最多监听多少个事件，它必须大于0
	timeout 参数指定超时时间，单位为毫秒，如果timeout 为 0，则epoll_wait 会立即返回，如果 timeout 为-1，则 epoll_wait 会一直阻塞，直到有事件就绪。
*/

LT&ET

epoll 对文件描述符有两种操作模式：LT（Level Trigger，电平触发）模式和 ET（Edge Trigger，边沿触发）模式。LT 模式是默认的工作模式。当往epoll 内核事件表中注册一个文件描述符上的EPOLLET 事件时，epoll 将以高效的ET 模式来操作该文件描述符。

对于 LT 模式操作的文件描述符，当 epoll_wait 检测到其上有事件发生并将此事件通知应用程序后，应用程序可以不立即处理该事件。这样，当应用程序下一次调用 epoll_wait 时，还会再次向应用程序通告此事件，直到该事件被处理。

对于 ET 模式操作的文件描述符，当 epoll_wait 检测到其上有事件发生并将此事件通知应用程序后，应用程序必须立即处理该事件，因为后续的 epoll_wait 调用将不再向应用程序通知这一事件。所以 ET 模式在很大程度上降低了同一个 epoll 事件被重复触发的次数，因此效率比 LT 模式高。

服务端：

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <fcntl.h>
#include <errno.h>

#define MAX_FD 128
#define DATALEN 1024
#define EPOLLSIZE 5
#define LT 0
#define ET 1

// 初始化服务器端的 sockfd 套接字
int InitSocket()
{
	int sockfd = socket(AF_INET, SOCK_STREAM, 0);
	if(sockfd == -1) return -1;

	struct sockaddr_in saddr;
	memset(&saddr, 0, sizeof(saddr));
	saddr.sin_family = AF_INET;
	saddr.sin_port = htons(6000);
	saddr.sin_addr.s_addr = inet_addr("127.0.0.1");

	int res = bind(sockfd, (struct sockaddr*)&saddr, sizeof(saddr));
	if(res == -1) return -1;

	res = listen(sockfd, 5);
	if(res == -1) return -1;

	return sockfd;
}

// 设置文件为非阻塞模式
void SetNoWait(int fd)
{
	int old_option = fcntl(fd, F_GETFL);
	int new_option = old_option | O_NONBLOCK;
	fcntl(fd, F_SETFL, new_option);
}

//关闭客户端连接
void CloseClient(int epfd, int fd)
{
	close(fd);
	printf("A Client disconnected\n");
	if (epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL) == -1)
	{
		printf("epoll_ctl del error\n");
	}
}

//获取一个新的客户端连接，如果 flag 为 ET，则以 ET 模式处理此客户端
void GetClientLink(int sockfd, int epfd, int flag)
{
	struct sockaddr_in caddr;
	socklen_t len = sizeof(caddr);
	int c = accept(sockfd, (struct sockaddr*)&caddr, &len);
	if (c < 0)
	{
		printf("Client Link error\n");
		return;
	}

	struct epoll_event ev;
	ev.data.fd = c;
	if (flag)
	{
		ev.events = EPOLLIN | EPOLLRDHUP | EPOLLET;
		SetNoWait(c);
	}
	else
	{
		ev.events = EPOLLIN | EPOLLRDHUP;
	}

	if (epoll_ctl(epfd, EPOLL_CTL_ADD, c, &ev) == -1)
	{
		printf("epoll_ctl add error\n");
	}
}

//LT 模式的客户端数据处理方式
void LTDealClientData(int epfd, int fd)
{
	char buff[DATALEN] = { 0 };
	int n = recv(fd, buff, DATALEN - 1, 0);
	if (n <= 0)
	{
		CloseClient(epfd, fd);
		return;
	}

	printf("%d: %s\n", fd, buff);
	send(fd, "OK", 2, 0);
}

//ET 模式的客户端数据处理方式
void ETDealClientData(int epfd, int fd)
{
	while (1)
	{
		char buff[DATALEN] = { 0 };
		int n = recv(fd, buff, DATALEN - 1, 0);
		if (n == -1)
		{
			if (errno == EAGAIN || errno == EWOULDBLOCK)
			{
				printf("read later\n");
				break;
			}
			else
			{
				CloseClient(epfd, fd);
				break;
			}
		}
		else if (n == 0)
		{
			CloseClient(epfd, fd);
			break;
		}
		else
		{
			printf("%d: %s\n", fd, buff);
			send(fd, "OK", 2, 0);
		}
	}
}
	
// 处理就绪的文件描述符上的数据
void DealReadyEvent(struct epoll_event *events,int n,int sockfd,int epfd)
{
	int i = 0;
	for (; i < n; ++i)	
	{
		int fd = events[i].data.fd;
		if (fd == sockfd)
		{
			GetClientLink(sockfd, epfd, LT); // 设置为 LT 模式
		// GetClientLink(sockfd, epfd, ET); // 设置为 ET 模式
		}
		else if (events[i].events & EPOLLRDHUP)
		{
			CloseClient(epfd, fd);
		}
		else if (events[i].events & EPOLLIN)
		{
			LTDealClientData(epfd, fd); // 以 LT 模式处理客户端数据
		//ETDealClientData(epfd, fd); //以 ET 模式处理客户端数据
		}
		else
		{
			printf("error\n");
		}
	}
}
	
int main()
{
	int sockfd = InitSocket();
	assert(sockfd != -1);
	int epfd = epoll_create(EPOLLSIZE);
	assert(epfd != -1);
	
	struct epoll_event ev;
	ev.data.fd = sockfd;
	ev.events = EPOLLIN;
	
	if (epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev) == -1)
	{
		printf("epoll_ctl add error\n");
		exit(0);
	}
	while (1)
	{
		struct epoll_event events[MAX_FD];
		int n = epoll_wait(epfd, events, MAX_FD, 2000);
		if (n < 0)
		{
			printf("epoll_wait error\n");
			continue;
		}
		else if (n == 0)
		{
			printf("timeout\n");
			continue;
		}
		else
		{
			DealReadyEvent(events, n, sockfd, epfd);
		}
	}
}