[系统运维]从驱动开发角度理解linux epoll调用链

epoll是linux中一种处理高并发的事件查询机制，是在原有poll机制上的改进，相比原有的poll机制，epoll在处理/监控大量文件描述符时，拥有更好的性能。
网上能够找到大量关于Epoll的资料，包括epoll原理介绍，使用场景和作用，内核源码分析，从用户空间的角度如何使用epoll的进行编程等等。而本文只从linux设备驱动开发者的角度谈谈driver中实现的poll接口在epoll框架中是如何调用的，对于驱动开发者来说用户使用epoll还是poll是否有区别。

一、调用链分析

1.用户的常见用法
在linux shell中通过man epoll我们可以看到一种常见的用法

           epollfd = epoll_create1(0);
           if (epollfd == -1) {
               perror("epoll_create1");
               exit(EXIT_FAILURE);
           }

           ev.events = EPOLLIN;
           ev.data.fd = listen_sock;
           if (epoll_ctl(epollfd, EPOLL_CTL_ADD, listen_sock, &ev) == -1) {
               perror("epoll_ctl: listen_sock");
               exit(EXIT_FAILURE);
           }

           for (;;) {
               nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);
               if (nfds == -1) {
                   perror("epoll_wait");
                   exit(EXIT_FAILURE);
               }

               for (n = 0; n < nfds; ++n) {
                   if (events[n].data.fd == listen_sock) {
                       conn_sock = accept(listen_sock,
                                          (struct sockaddr *) &addr, &addrlen);
                       if (conn_sock == -1) {
                           perror("accept");
                           exit(EXIT_FAILURE);
                       }
                       setnonblocking(conn_sock);
                       ev.events = EPOLLIN | EPOLLET;
                       ev.data.fd = conn_sock;
                       if (epoll_ctl(epollfd, EPOLL_CTL_ADD, conn_sock,
                                   &ev) == -1) {
                           perror("epoll_ctl: conn_sock");
                           exit(EXIT_FAILURE);
                       }
                   } else {
                       do_use_fd(events[n].data.fd);
                   }
               }
           }

总结一下，首先需要调用epoll_create来创建一个epoll实例。

epollfd = epoll_create1(0);

通过epoll_ctl可以增加、删除一个监控对象，这里通过EPOLL_CTL_ADD的动作将需要监控的fd以及对应的监控事件注册到刚才创建的epoll实例中。

epoll_ctl(epollfd, EPOLL_CTL_ADD, listen_sock, &ev)

然后通过epoll_wait来等待监控的fd中有事件发生，此时线程会阻塞在epoll_wait中，当有关注的事件发生时，阻塞的线程会被唤醒，此后需要检查是哪些fd发生了事件，并进行相应处理

nfds = epoll_wait(epollfd, events, MAX_EVENTS, -1);

2.driver中实现的poll接口
驱动开发中，对于一个字符设备的poll功能，我们一般这样实现：

static unsigned int my_poll(struct file *filp, poll_table *wait)
{
	unsigned int event = 0;
	......
	if(err)
		return POLLERR;
	/*注册等待队列*/
	poll_wait(filp, &my_wait_queue, wait);
	if(condition) {
	    /*根据业务需求决定这里返回POLLIN，或者POLLOUT，还是其他event*/
		event = event_mask; 
	}
	return event;
}

static const struct file_operations my_fops = {
        .owner          = THIS_MODULE,
        .poll           = my_poll,
        .open           = my_open,
        .release        = my_close
}; 

//在中断或者其他函数中，我们会完成事件的通知，并唤醒调用poll接口的线程
int my_notify()
{
	......
	set_condition();
	wake_up(&my_wait_queue);
}

poll_wait本质上是在调用回调函数，这个回调函数是epoll框架里实现的，同理poll的框架也实现了另一个回调函数，二者并不一样，但是不影响driver的poll接口在两种框架中都能正常使用。

static inline void poll_wait(struct file * filp, wait_queue_head_t * wait_address, poll_table *p)
{        
         if (p && p->_qproc && wait_address)
                 p->_qproc(filp, wait_address, p);
}

3. epoll的行为逻辑

当用户调用epoll_wait的时候，线程陷入内核态后并不会轮询所有监控的fd，而是只关注一个ready list的链表,轮询ready list上的文件描述符对应的poll接口，获取事件掩码并返回给用户。这也是epoll比poll更高效的原因之一。
而epoll所监控的文件描述符是何时挂到ready list上的，则是通过epoll注册给等待队列的回调函数ep_poll_callback（）实现的，将在下文分析。

总的来说，epoll_wait开始等待事件发生时，只会轮询部分文件描述符，而不是所有的文件描述符，所有事件的通知都是事件的实际发生方，也就是驱动设备，主动通知的。

事实上，和poll框架不一样的是，epoll框架会在epoll_ctl添加监控的文件描述符时调用该文件描述符的poll接口：fop->poll,也就是驱动开发者写的poll接口。在驱动中，当事件真正发生时，通过epoll_poll_callback()可以

4.从user space到kernel space的调用栈
这里先列出调用关系栈：
当用户调用epoll_ctl时，调用栈如下：

epoll_ctl(epollfd, EPOLL_CTL_ADD, listen_sock, &ev) //用户新增一个待监控的fd
	->SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,struct epoll_event __user *, event) //通过系统调用陷入内核
		->ep_insert(ep, &epds, tf.file, fd, full_check); 
			->init_poll_funcptr(&epq.pt, ep_ptable_queue_proc); //这里注册的回调函数将会保存在poll_table中，随后传给device driver的poll接口，并在poll_wait()中被调用
			->ep_item_poll(epi, &epq.pt); // 首次添加监控时，先调用poll接口获取事件掩码
				-> epi->ffd.file->f_op->poll(...) //这里调用的就是驱动中实现的poll接口
					-> my_poll //这里就是device driver实现的poll函数
						->poll_wait //通常我们会在poll函数中调用poll_wait注册等待队列
							->ep_ptable_queue_proc //在poll_wait中，调用epoll注册的回调函数
								-> init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);//注册等待队列的回调函数为ep_poll_callback，此步很重要，决定了等待队列唤醒后执行的动作
			-> if(事件发生) list_add_tail(&epi->rdllink, &ep->rdllist); //如果有事件发生，则挂载到ready list上，这样就能被epoll_wait关注到

可以看到，在epoll_ctl首次添加一个新的需要监控的文件描述符时，会去调用device driver中的poll函数，并注册等待队列的回调函数，如果在添加文件描述符时driver中已经有事件发生了，则会直接将对应的ep_item添加到epoll实例的ready list中，从而epoll_wait函数执行时不会休眠，而是直接通过ready list来得知有事件发生。

当用户调用epoll_wait时，调用关系如下：

SYSCALL_DEFINE4(epoll_wait,...)
	-> do_epoll_wait()
		-> ep_poll()
			-> ep_events_available() // 该函数判断是否有事件发生, 本质上是查询ready list是否为空，for循环轮询直到超时，或者ready list不为空时跳出
			-> ep_send_events() // 如果有事件发生，将事件发送给用户空间
				-> ep_scan_ready_list()
					-> ep_send_events_proc()
						-> ep_item_poll() //同上文，之后会调用到fop->poll，查询事件掩码，之后发给用户。

假设在epoll_ctl()调用f_op->poll时，driver中并没有事件发生，则对应的ep_item也不会挂载到ready list上，此时如果epoll_wait被调用，则epoll_wait无法得知device driver内是否有事件发生，那么device driver要如何通知epoll有新事件发生了呢？这就和epoll_ctl中注册的回调函数有关。

回看epoll_ctl()的调用链中，通过ep_insert->init_poll_funcptr->f_op->poll->poll_wait-> ep_ptable_queue_proc -> init_waitqueue_func_entry的调用链中，将ep_poll_callback(）设定为了等待队列的回调函数，该函数会在device driver中发生事件的时候被调用，以上文提供的my_notify（）函数为例子，调用链如下：

my_notify() //在device driver中，当事件发生时，我们通常会通过有一个通知动作
 -> wake_up(&my_wait_queue); //当等待的条件满足后，通过wake_up唤醒休眠在poll等待队列的线程
   -> __wake_up_common() 
     -> ret = curr->func(curr, mode, wake_flags, key); // 在等待队列一般的用法中，这里的回调是default_wake_function()，用于唤醒休眠的线程, 但是在epoll中该回调已经被注册成了ep_poll_callback
       -> ep_poll_callback()
       	 -> if (!ep_is_linked(epi) && list_add_tail_lockless(&epi->rdllink, &ep->rdllist))  {...} //在if判断语句中，如果ep_item没有加入到ready list中，就加入到ready list里，从而该事件能被epoll_wait追踪到。

从以上调用链我们知道，只要device driver中有事件发生，并调用wake_up函数，则该事件就会被挂到ready list上，从而被epoll_wait感知到。这点和poll()的用法相比，还是有较大不同的。

二、对驱动开发者来说，epoll和poll有何不同

通过以上分析，我们发现epoll的用户行为和poll是不太一样的（需要先调用epoll_ctl注册事件），但是对于device driver来说，fop->poll接口的语义仍然是一样的，即“调用该函数时，返回此时刻的事件掩码”，因此大部分情况下，driver的poll接口无需针对epoll进行特定的适配。

但是由于epoll_ctl也会调用到fop->poll，若此时driver fop->poll走了某些if分支没有调用到poll_wait()，则会导致ep_poll_callback无法被注册到等待队列中。比如下面这个例子，函数中（ready_condition）条件可能需要device driver完成了某些初始化操作（也许是调用一个特定的ioctl，也许是等待某个状态)才会变成true，若用户在device driver初始化操作完成前就调用了epoll_ctl，则会导致ep_poll_callback无法被注册到等待队列中。

static unsigned int my_poll_func(struct file *filp, poll_table *wait)
{
	unsigned int event = 0;
	......
	if(err)
		return POLLERR;
	if (!ready_condition) // 如果driver中限制用户必须在某些初始化操作后才能调用poll，并在这里设置了拦截，则会直接返回。
		return POLLERR;
	/*注册等待队列*/
	poll_wait(filp, &my_wait_queue, wait);
	if(condition) {
	    /*根据业务需求决定这里返回POLLIN，或者POLLOUT，还是其他event*/
		event = event_mask; 
	}
	return event;
}

这样一来，即使device driver中事件发生了，也无法通知到阻塞在epoll_wait()的线程。这将导致epoll_wait永远无法唤醒或者返回超时错误（TIMEOUT)。这是驱动开发者需要小心的事情，因为我们无法保证用户行为是按照驱动预设逻辑进行的。

由于技术有限，无法深入分析更多，如有谬误，欢迎交流探讨。