F-Stack之kqueue封装为epoll介绍

F-Stack

发布于 2018-01-31 10:28:40

2.1K0

发布于 2018-01-31 10:28:40

文章被收录于专栏：F-Stack的专栏

F-Stack是一个全用户态的高性能的网络接入开发包，基于DPDK、FreeBSD协议栈、微线程接口等，适用于各种需要网络接入的业务，用户只需要关注业务逻辑，简单的接入F-Stack即可实现高性能的网络服务器。

F-Stack中使用的FreeBSD协议栈的高性能异步事件通知的API是kqueue，而Linux系统上则是我们熟悉的epoll，大量的Linux网络server都是基于epoll事件通知机制，为降低已有服务器接入F-Stack的修改难度，F-Stack协议栈实现了把kqueue封装为epoll接口，提供的API如下：

int ff_epoll_create(int size):创建epoll fd，底层实际调用FreeBSD协议栈的kern_kqueue()接口
int ff_epoll_ctl(int epfd, int op, int fd, struct epoll_event *event):epoll事件操作函数，添加关心的fd和事件到epoll fd中，底层实际调用了FreeBSD协议栈的kern_kevent()接口
int ff_epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout):epoll事件操作通知函数，底层实际调用了FreeBSD协议栈的kern_kevent()接口
int ff_epoll_close(int epfd):epoll fd的关闭函数，底层调用kern_close()

比较核心的ff_epoll_ctl()和ff_epoll_wait()代码实现如下：

	int ff_epoll_ctl(int epfd, int op, int fd, struct epoll_event *event)
	{
		if (!event && op != EPOLL_CTL_DEL) {
	        ff_os_errno(ff_EINVAL);
			return -1;
		}
	
		struct kevent kev[3];
		if (op == EPOLL_CTL_ADD){
			EV_SET(&kev[0], fd, EVFILT_READ,
				EV_ADD | (event->events & EPOLLIN ? 0 : EV_DISABLE), 0, 0, NULL);
			EV_SET(&kev[1], fd, EVFILT_WRITE,
				EV_ADD | (event->events & EPOLLOUT ? 0 : EV_DISABLE), 0, 0, NULL);
			EV_SET(&kev[2], fd, EVFILT_USER, EV_ADD,
				    event->events & EPOLLRDHUP ? 1 : 0, 0, NULL);		
		} else if (op == EPOLL_CTL_DEL) {
			EV_SET(&kev[0], fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);
			EV_SET(&kev[1], fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
			EV_SET(&kev[2], fd, EVFILT_USER, EV_DELETE, 0, 0, NULL);
		} else if (op == EPOLL_CTL_MOD) {
			EV_SET(&kev[0], fd, EVFILT_READ,
			    event->events & EPOLLIN ? EV_ENABLE : EV_DISABLE, 0, 0, NULL);
			EV_SET(&kev[1], fd, EVFILT_WRITE,
			    event->events & EPOLLOUT ? EV_ENABLE : EV_DISABLE, 0, 0, NULL);
			EV_SET(&kev[2], fd, EVFILT_USER, 0,
			    NOTE_FFCOPY | (event->events & EPOLLRDHUP ? 1 : 0), 0, NULL);		
		} else {
			ff_os_errno(ff_EINVAL);
			return -1;
		}
	
		return ff_kevent(epfd, kev, 3, NULL, 0, NULL);
	}

ff_epoll_ctl()核心是把Linux epoll的事件EPOLLIN、EPOLLOUT(其他的暂未支持)转成成FreeBSD的事件标EVFILT_READ、EVFILT_WRITE、EVFILT_USER。

	int ff_epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout)
	{
		if (!events || maxevents < 1) {
			ff_os_errno(ff_EINVAL);
			return -1;
		}
		
		struct kevent *evlist = malloc(sizeof(struct kevent)*maxevents, M_DEVBUF, M_ZERO|M_NOWAIT);
		if(NULL == evlist){
			ff_os_errno(ff_EINVAL);
			return -1;		
		}
		memset(evlist, 0, sizeof(struct kevent)*maxevents);
		
		int ret = ff_kevent(epfd, NULL, 0, evlist, maxevents, NULL);
		if (ret == -1) {
			free(evlist, M_DEVBUF);
			return ret;
		}
	
		unsigned int event_one = 0;
		for (int i = 0; i < ret; ++i) {
			event_one = 0;
			if (evlist[i].filter & EVFILT_READ) {
				event_one |= EPOLLIN;
			} 
			if (evlist[i].filter & EVFILT_WRITE) {
				event_one |= EPOLLOUT;
			}
	
			if (evlist[i].flags & EV_ERROR) {
				event_one |= EPOLLERR;
			}
	
			if (evlist[i].flags & EV_EOF) {
				event_one |= EPOLLIN;		
			}
			events[i].events   = event_one;
			events[i].data.fd  = evlist[i].ident;
		}
		
		free(evlist, M_DEVBUF);
		return ret;
	}

ff_epoll_wait()的核心就是struct kevent结构和struct epoll_event的转换，把kqueue返回的fd和事件都封装到struct epoll_event结构中，返回给调用者。

一个实际的Server DEMO代码如下:

	int loop(void *arg)
	{
	    /* Wait for events to happen */
	
	    int nevents = ff_epoll_wait(epfd,  events, MAX_EVENTS, 0);
	    int i;
	
	    for (i = 0; i < nevents; ++i) {	
	        /* Handle new connect */
	        if (events[i].data.fd == sockfd) {
	            int nclientfd = ff_accept(sockfd, NULL, NULL);
	            assert(nclientfd > 0);
	            /* Add to event list */
	    	    ev.data.fd = nclientfd;
	    	    ev.events  = EPOLLIN;
	    	    assert(ff_epoll_ctl(epfd, EPOLL_CTL_ADD, nclientfd, &ev) == 0);
	            //fprintf(stderr, "A new client connected to the server..., fd:%d\n", nclientfd);
	        } else { 
	            if (events[i].events & EPOLLERR ) {
	                /* Simply close socket */
	        	ff_epoll_ctl(epfd, EPOLL_CTL_DEL,  events[i].data.fd, NULL);
	                ff_close(events[i].data.fd);
	                //fprintf(stderr, "A client has left the server...,fd:%d\n", events[i].data.fd);
	            } else if (events[i].events & EPOLLIN) {
	                char buf[256];
	                size_t readlen = ff_read( events[i].data.fd, buf, sizeof(buf));
	                //fprintf(stderr, "bytes are available to read..., readlen:%d, fd:%d\n", readlen,  events[i].data.fd);
	        	if(readlen > 0){
	                    ff_write( events[i].data.fd, html, sizeof(html));
	        	} else {
	        	    ff_epoll_ctl(epfd, EPOLL_CTL_DEL,  events[i].data.fd, NULL);
	                    ff_close( events[i].data.fd);
	                    //fprintf(stderr, "A client has left the server...,fd:%d\n", events[i].data.fd);		
	        	}
	            } else {
	                fprintf(stderr, "unknown event: %8.8X\n", events[i].events);
	            }
	        }
	    }
	}

实际运行结果：