Linux 知识之IO多路复用epoll

#include <sys/epoll.h>

int epoll_create(int size); /* 从Linux内核版本2.6.8起,形参size被忽略,但仍需大于0 */

#include <sys/epoll.h>

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

typedef union epoll_data {      /* 联合体,联合体内成员无偏移,既共用首地址,该联合体内存占8字节 */
	void        *ptr;	
	int          fd;
	uint32_t     u32;
	uint64_t     u64;
} epoll_data_t;

struct epoll_event {
	uint32_t     events;    /* epoll 监听的事件类型 */
	epoll_data_t data;      /* 事件就绪时返回给用户进程 */
};

// Eventpoll.c(kernel-4.9\fs)
static int ep_send_events_proc(struct eventpoll *ep, struct list_head *head,void *priv)
{
......
	for (eventcnt = 0, uevent = esed->events;
	     !list_empty(head) && eventcnt < esed->maxevents;) {
......
        list_del_init(&epi->rdllink);	/* 将事件从就绪队列中移除 */

		if (epi->event.events & EPOLLONESHOT)
			epi->event.events &= EP_PRIVATE_BITS;
		else if (!(epi->event.events & EPOLLET)) {	
        /* 当触发方式为非边缘触发时,会再次将该事件加入到就绪队列中.
         * 当用户下次调用epoll_wait时会检测事件是否处理完成,
         * 如果未处理完毕,则会再次报告有事件产生.
        */
		    list_add_tail(&epi->rdllink, &ep->rdllist);
		    ep_pm_stay_awake(epi);
		}
	}
	return eventcnt;
}

#include <sys/epoll.h>

int epoll_wait(int epfd, struct epoll_event *events,int maxevents, int timeout);

/*本例子的流程为: 
* 1. 调用socketpair得到一对双向fd
* 2. 调用fork,产生子进程.子进程每5秒向父进程发送信息
* 3. 父进程创建eventfd,用于通知终端输入情况
* 4. 父进程创建epoll,并将socketpair和eventfd加入监听列表中
* 5. 父进程调用epoll_wait等待监听事件触发
*/
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>         
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <pthread.h>

#define true 1
#define false 0
#define EPOLL_INIT 1

typedef unsigned char boolean ;

struct Transfer_Data {
	int eventFd;
	char *msg;
	char len;
	boolean exit;
};

int addFd(int epollFd, int fd, int events) {
	struct epoll_event eventItem;
	memset(&eventItem, 0, sizeof(struct epoll_event)); 
	eventItem.events = events;
	eventItem.data.fd = fd;
	if (epoll_ctl(epollFd ,EPOLL_CTL_ADD, fd, &eventItem) < 0) {
		printf("add fd fail reason: %s\n", strerror(errno));
		return -1;
	}
	return 0;
}

void childLoop(int fd) {	
	char recvMsg[128] = { 0 };
	char *sendMsg = "I am Child";

	while (true) {
		sleep(5);
		if (write(fd, sendMsg, strlen(sendMsg) + 1) < 0) {
			printf("%s write fail reason: %s\n", __func__, strerror(errno));
			return;
		}
	}
}

int awoken(int wakeFd) {
	uint64_t counter;
	if (read(wakeFd, &counter, sizeof(uint64_t)) < 0) {
		printf("read wakeFd fail reason: %s\n", strerror(errno));
		return -1;
	}
	return 0;
}

void *thread_func(void *arg) {  
	struct Transfer_Data *ptd = (struct Transfer_Data *)arg;
	uint64_t inc = 1;
	int err = 0;
	for (;ptd -> exit == false;) {
		fgets(ptd -> msg, ptd -> len, stdin);
		if ((err = write(ptd -> eventFd, &inc, sizeof(uint64_t))) < 0) {	
			printf("write eventfd fail reason: %s\n",strerror(errno));
			ptd -> exit = true;
			break;
		}
	}
	return NULL;
}  

void parentLoop(int fd) {	
	char msg[128] = { 0 };
	char recvMsg[128] = {0};
	int mEpollFd = -1;
	int eventCount = 0;
	int err = 0;
	struct epoll_event events[8];
	
	int wakeEventFd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); 
	if (wakeEventFd < 0) {
		printf("Create eventfd fail reason: %s\n",strerror(errno));
		return;
	}

	pthread_t readThread;
	struct Transfer_Data td = {wakeEventFd, msg, 128, false};
	err = pthread_create(&readThread, NULL, thread_func, &td);
	if (err != 0) {
		printf("Create thread fail reason: %s\n", strerror(err));
		goto EXIT;
	}

	if ((mEpollFd = epoll_create(EPOLL_INIT)) < 0) {
		printf("epoll_create fail reason: %s\n", strerror(errno));
		goto EXIT;
	}

	if (addFd(mEpollFd, fd, EPOLLIN) || addFd(mEpollFd, wakeEventFd, EPOLLIN)) {
		printf("addFd fail\n");
		goto EXIT;
	}
	
	while (td.exit == false) {
		printf("%s epoll_wait\n",__func__);
		eventCount = epoll_wait(mEpollFd, events, sizeof(events) / sizeof(*events), -1);		
		if (eventCount < 0) {	
			printf("epoll_wait fail reason: %s\n", strerror(errno));
			goto EXIT;
		}else if (eventCount == 0) {
			printf("epoll_wait timeout,continue wait\n");
			continue;
		}else {
			for (int i =0; i < eventCount; i++) {
				if (events[i].data.fd == wakeEventFd) {
					if (awoken(events[i].data.fd) < 0)
						goto EXIT;
					printf("%s read msg : %s\n", __func__, msg);
				}else {
					if (read(events[i].data.fd, recvMsg, sizeof(recvMsg) / sizeof(*recvMsg)) < 0) {
						printf("%s read fail reason: %s\n", __func__, strerror(errno));
						goto EXIT;
					}
					printf("%s read msg : %s\n", __func__, recvMsg);
				}
			}
		}
	}
EXIT:
	td.exit = true;
	pthread_join(readThread, NULL);
    /* 创建和close应成对出现,防止FD泄露 */
	close(wakeEventFd);
	close(mEpollFd);
	return;
}

int main(int argc __unused, char **argv __unused) {
	int pid;
	int fd[2];
	int err = 0;
	
	err = socketpair(AF_UNIX, SOCK_SEQPACKET | SOCK_CLOEXEC | SOCK_NONBLOCK, 0, fd);

	if (err != 0) {
		printf("socketpair fail reason: %s\n", strerror(errno));
		return -1;
	}

	pid = fork();

	if (pid < 0) {
		printf("fork fail reason: %s\n", strerror(errno));
		return -1;

	}

	if (pid == 0) {			//子进程
		close(fd[0]);		//关闭fd[0]
		childLoop(fd[1]);
		close(fd[1]);		//关闭fd[1]
		return 0;
	}else {				//父进程
		close(fd[1]);		//关闭fd[1]
		parentLoop(fd[0]);
		close(fd[0]);		//关闭fd[0]
		return 0;
	}
	return 0;
}