epoll 的EPOLLONESHOT 事件———实例程序

本实例程序是一个服务器，他接收用户数据，并输出到标准输出，他的特点是：对于每一个EPOLLIN事件，都唤醒一个线程去处理他，而主进程继续往下执行，也就是进入下一轮循环，调用epoll_wait监听其他事件。

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

#define MAX_EVENT_NUMBER 1024
#define BUFFER_SIZE 1024

struct 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.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 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);
			/* 休眠 5s，模拟数据处理过程 */
			sleep(5);
		}
	}
	printf("end thread receiving data on fd: %d\n", sockfd);
}

int main(int argc, char* argv[])
{
    
	if (argc <= 2)
	{
    
		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);
	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(10);
	assert(epollfd != -1);
	/* 注意，监听sockfd 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_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)	/* 对每个 EPOLLIN 事件开启一个线程，然后进程继续往下执行下一次 epoll_wait */
			{
    
				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("something else happened\n");
			}
		}
	}

	close(listenfd);
	return 0;
}