I/O复用模型之epoll学习

简介：

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

优点：

1、支持一个进程打开大数目的socket描述符：select最不能忍受的是一个进程所打开的FD是有一定限制的，由FD_SETSIZE设置，默认值是1024,。对于那些需要支持上万连接数目的IM服务器来说实在是太少了。

2、IO效率不随FD数目的增加而线性下降：传统的select/poll的一个缺陷就是当你拥有一个很大的socket集合，由于网络延时，任意时间只有一部分的socket是“活跃"的，但是select/poll每次调用都会线性扫描全部的集合，导致效率呈现线性的下降。但是epoll不存在这个问题，它只会对“活跃”的socket进行操作，原因是在内核实现中epoll是根据每个fd上面的callback函数实现的，其他idle状态socket则不会。但是，如果所有的socket基本上都是活跃的，比如说一个高速的LAN网络，epoll并不比select/poll有什么效率，相反，如果过多食用epoll_ctl，效率还会稍微下降。

3、select从内核到用户空间传递文件描述符上的发送的信息都是使用内存复制的方式进行的，而epoll是采用共享内存的方式。

epoll中的三个主要的函数：

1、int epoll_create(int size)：

函数功能：生成一个epoll专用的文件描述符。

函数参数：size表示用来告诉内核这个监听的最大数目。

函数返回值：生成的文件描述符。

需要注意的地方：当创建好一个epoll句柄后，他就会占用一个fd值，所以在使用完epoll后，必须调用close()函数关闭，否则可能导致fd被耗尽。

2、int epoll_ctl(int epfd, int op, int fd, struct epoll_event * event)：

函数功能：控制某个epoll文件描述符上的事件，可以注册事件，修改时间，删除事件。

函数参数：

epfd：由epoll_create生成的epoll专用的文件描述符。

op：EPOLL_CTL_ADD注册、EPOLL_CTL_MOD修改、EPOLL_CTL_DEL删除。

fd：关联的文件描述符。

event：指向epoll_event的指针。

函数返回值：0表示成功，-1表示失败。

3、int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)：

函数功能：轮询I/O时间的发生，类似于selelct()。

函数参数：

epfd：由epoll_create生成的epoll专用的文件描述符。

events：用于回传待处理事件的数组。

maxevents：每次能处理的事件数。

timeout：等待I/O事件发生的超时值，-1相当于阻塞，0相当于非阻塞。

函数返回值：>= 0表示返回的事件数，-1表示错误。

epoll中的主要用到的数据结构：

typedef union epoll_data {
    void *ptr;
    int fd;
    _uint32_t u32;
    _uint64_t u64;
} epoll_data_t;


struct epoll_event {
    _uint32_t events; /*Epoll events */
    epoll_data_t data; /*User data variable*/
};

其中，events的类型有：

EPOLLIN：文件描述符可以读。

EPOLLOUT：文件描述符可以写。

EPOLLPRI：文件描述符有紧急的数据可读。

EPOLLERR：文件描述符发生错误。

EPOLLHUP：文件描述符被挂断。

EPOLLET：文件描述符有事件发生。

epoll的两种工作方式：

1、ET：即Edge Triggered，边缘触发。仅当状态发生变化时才会通知，需要细致的处理每个请求，否则容易发生丢失事件的情况。只支持非阻塞的socket。

2、LT：Level Triggered，水平触发（默认工作方式）。只要还有没有处理的事件就会一直通知，因此不用担心事件丢失的情况。效率会低于ET触发，尤其在大并发，大流量的情况下。支持阻塞和非阻塞的socket。

服务器端：

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

#define SERV_PORT 1234 /*服务器监听的端口号*/
#define MAX_CLINE 100 /*最大同时连接请求数*/
#define MAX_EVENTS 1000
#define BUF_SIZE 1024
#define EPOLL_RUN_TIME_OUT -1


void setnonblocking(int sock)
{
	int opts = fcntl(sock, F_GETFL);
	if (opts < 0) {
		perror("fcntl error");
		exit(1);
	}

	opts |= O_NONBLOCK;
	
	if (fcntl(sock, F_GETFL, opts) < 0) {
		perror("fcntl error");
		exit(1);
	}
}


void do_use_fd(int client_fd)
{
	char message[BUF_SIZE];
	int res = recv(client_fd, message, BUF_SIZE, 0);
	if (res < 0) {
		perror("recv error");
		exit(1);
	} else if (res == 0) {
		close(client_fd);
	} else {
		const char str[] = "hehe...
";
		if (send(client_fd, str, sizeof(str), 0) == -1) {
			perror("send error...");
			exit(1);
		}
	}
}


int main()
{
	int sock_fd;
	struct sockaddr_in server_addr, remote_addr;
	
	if ((sock_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
		perror("can not create socket");
		exit(1);
	}

	server_addr.sin_family = AF_INET; 
	server_addr.sin_port = htons(SERV_PORT);
	server_addr.sin_addr.s_addr = INADDR_ANY;
	bzero(&(server_addr.sin_zero), 8);

	setnonblocking(sock_fd); //设置监听socket为不阻塞
	if (bind(sock_fd, (struct sockaddr *)&server_addr, sizeof(struct sockaddr)) == -1) {
		perror("bind error");
		exit(1);
	}
	
	if (listen(sock_fd, MAX_CLINE) == -1) {
		perror("listen error");
		exit(1);
	}
	
	socklen_t sin_size = sizeof(struct sockaddr_in);

	struct epoll_event ev, events[MAX_EVENTS];
	int conn_sock, nfds, epollfd;

	epollfd = epoll_create(MAX_EVENTS); //创建一个epoll描述符，并将监听socket加入epoll

	if (epollfd == -1) {
		perror("epoll_erate error");
		exit(1);
	}

	puts("epoll_erate Ok...");

	ev.events = EPOLLIN | EPOLLET; //可读,边沿触发
	ev.data.fd = sock_fd;

	if (epoll_ctl(epollfd, EPOLL_CTL_ADD, sock_fd, &ev) == -1) {
		perror("epoll_ctl error");
		exit(1);
	}

	puts("epoll_ctl Ok...");

	while (true) {

		puts("epoll_wait start...");

		//类似于select()调用，函数返回的是事件数
		nfds = epoll_wait(epollfd, events, MAX_EVENTS, EPOLL_RUN_TIME_OUT);
		if (nfds == -1) {
			perror("epoll_wait error");
			exit(1);
		}

		printf("epoll_wait returns, nfds = %d
", nfds);
	
		for (int i = 0; i < nfds; ++i) {
			//新的连接到来，将连接添加到epoll中，并发送"welcome"消息
			if (events[i].data.fd == sock_fd) {
				conn_sock = accept(sock_fd, (struct sockaddr *)&remote_addr, &sin_size);
				if (conn_sock == -1) {
					perror("accept error");
					exit(1);
				}

				setnonblocking(conn_sock); //设为非阻塞的
				ev.events = EPOLLIN | EPOLLET; //可读，边缘触发
				ev.data.fd = conn_sock;

				if (epoll_ctl(epollfd, EPOLL_CTL_ADD, conn_sock, &ev) == -1) {
					perror("eppol_ctl error");
					exit(1);
				}

				char message[BUF_SIZE];
				bzero(message, BUF_SIZE);
				int res = sprintf(message, "Welcome client %d", conn_sock);
				if ((res = send(conn_sock, message, BUF_SIZE, 0)) < 0) {
					perror("send error");
					exit(1);
				} else if (res == 0) {
					close(conn_sock);
				}

			} else {
				printf("%d.....
", i);
				do_use_fd(events[i].data.fd);
			}
		}

	}

	close(sock_fd);
	close(epollfd);

	return 0;
}

客户端：

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

#define SERV_PORT 1234 /*服务器监听的端口号*/
#define MAX_EVENTS 1000 
#define BUF_SIZE 100

int main()
{
	int client_fd, res;
	struct sockaddr_in addr;

	addr.sin_family = AF_INET;
	addr.sin_port = htons(SERV_PORT);
	addr.sin_addr.s_addr = htonl(0x7f000001);
	
	for (int i = 0; i < MAX_EVENTS; ++i) {
		if((client_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
			perror("can not create socket");
			exit(1);
		}
		
		if (connect(client_fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
			perror("connect error");
			exit(1);
		}

		char message1[] = "I'm client..
";
		char message2[BUF_SIZE];
		bzero(message2, BUF_SIZE);
	
		if ((res = send(client_fd, message1, BUF_SIZE, 0)) < 0) {
			perror("send error");
			exit(1);
		}

		if ((res = recv(client_fd, message2, BUF_SIZE, 0)) < 0) {
			perror("recv error");
			exit(1);
		} else if (res == 0) {
			close(client_fd);
		} else 
			printf("%s
", message2);

	}
	return 0;
}