背景

linux 4.19

epoll()也是一种I/O多路复用的技术，但是完全不同于select()/poll()。更加高效，高效的原因其他博客也都提到了，这篇笔记主要是从源码的角度来分析一下实现过程。

作为自己的学习笔记，分析都在代码注释中，后续回顾的时候看注释好一点。

相关链接：

• Linux内核学习之 – ARMv8架构的系统调用笔记
• Linux内核学习之 – 系统调用open()和write()的实现笔记
• Linux内核学习之 – 系统调用poll()分析笔记

对于epoll()的分析建立在《Linux内核学习之 – 系统调用open()和write()的实现笔记》 + 《Linux内核学习之 – 系统调用poll()分析笔记》的基础上，因为有很多通用的步骤和环节。所以这篇文章写的没有那么细致，有些地方一笔带过。最好先瞄一眼这两篇。

同时对一些细节的处理分析没有写在博客中，有点麻烦，都写到注释中了。如epoll如何处理不同标志位如EPOLLONESHOT/EPOLLLT/EPOLLET，epoll中实现的一些检查宏等等

还有一件事，就是看了很多博客，感觉很多都是复制粘贴的，根本没看过内核真正的实现过程。就以4.19内核来讲，很多博客提到的两点感觉都不对：

1. 关于epoll_wait()休眠时，唤醒进程的到底是啥。这里很多博客没搞明白嵌套epollfd的概念。
2. 关于epoll使用共享内存来进行用户态/内核态传参这件事，看完后发现，根本就没用共享内存，起码4.19没用。

一、epoll()的使用方法

#include <stdlib.h>  
#include <stdio.h>   
#include <stdbool.h>   
#include <string.h> 
#include <unistd.h> 
#include <fcntl.h>  
#include <poll.h> 
#include <sys/epoll.h> 
#include <sys/time.h>
#include <errno.h> 

int main()  
{  
    int nfds;
    struct epoll_event event;
    struct epoll_event events[1024];
    char buff[10];
    unsigned char cnt = 0;

    int epoll_fd = -1;
    int dev_fd = -1;

    epoll_fd = epoll_create(1); // g, linux2.6之后传入参数size > 0即可，不会影响什么
    if(epoll_fd <= 0)
        printf("create err\n");

    dev_fd = open("/dev/xx",O_RDONLY);
    if(dev_fd < 0){
        printf("Failed open\n");  
 
    }
    
    event.events = EPOLLET;
    event.data.fd = dev_fd;

    // g, 添加要监听的设备
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, dev_fd, &event); // g, 参数为：epollfd, op(要执行什么操作), 要监听的fd， 事件


    while(1) {
        // g, 倒数第二个参数不能大于epoll_create()传递的参数
        nfds = epoll_wait(epoll_fd, &events, 1, 500 );
        if( nfds < 0 ) {
            printf("err\n");
            break;
        }
        for (int n = 0; n < nfds; ++n) {
            if (events[n].events & EPOLLIN) { 
                // g, 处理可读事件
            } else if(events[n].events & EPOLLOUT) {
                // g, 处理可写事件
            } 
            ...
            ...
        }
        
    }
    return 0;
}  

同样，与poll()一样，要求在驱动程序中实现dev_poll()

二、epoll()系统调用

2.1 epoll_create()

该系统调用如下：

fs/eventpoll.c:
SYSCALL_DEFINE1(epoll_create, int, size)
{
	if (size <= 0)
		return -EINVAL;

	return do_epoll_create(0);
}

所以说，你传入什么参数无所谓，只要大于0就行了，没影响。之后单纯的调用了函数do_epoll_create()：

static int do_epoll_create(int flags)
{
	int error, fd;

	// g, 这个结构体里面是有一颗红黑树的
	struct eventpoll *ep = NULL;
	struct file *file;

	/* Check the EPOLL_* constant for consistency.  */
	BUILD_BUG_ON(EPOLL_CLOEXEC != O_CLOEXEC);

	if (flags & ~EPOLL_CLOEXEC)
		return -EINVAL;
	/*
	 * Create the internal data structure ("struct eventpoll").
	 */
	error = ep_alloc(&ep);		// g, 使用kzalloc而不是slab来分配。申请内存后初始化一些参数
	if (error < 0)
		return error;
	/*
	 * Creates all the items needed to setup an eventpoll file. That is,
	 * a file structure and a free file descriptor.
	 */
	fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC));	// g, 与do_sys_open()中使用了一样同样的获取fd的函数
	if (fd < 0) {
		error = fd;
		goto out_free_ep;
	}

	// g, 创建一个struct file，初始化该file，绑定匿名anon_inode_inode，
	// g, 正常的inode都能在文件系统中找到对应的dentry，往往是inode->dentry->file->fd，但是对于匿名inode,来说不出现在文件系统中
	// g, 但是不是说没有dentry，仍然要为其创建一个struct dentry，只是该dentry目录不会出现在文件系统中，就叫"[eventpoll]"
	// g, 其中会使用 file->private_data = priv将ep_alloc()创建的eventpoll对象赋值给struct file的private_data 成员变量
	// g, 会绑定file->f_op = eventpoll_fops，其中实现了.poll
	file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,
				 O_RDWR | (flags & O_CLOEXEC));
	if (IS_ERR(file)) {
		error = PTR_ERR(file);
		goto out_free_fd;
	}
	ep->file = file;			// g, 绑定ep与刚申请的file的关系
	fd_install(fd, file);		// g，建立fd与file的联系，与do_sys_open()调用的fd_install是一样的
	return fd;

out_free_fd:
	put_unused_fd(fd);
out_free_ep:
	ep_free(ep);
	return error;
}

可以认为，该函数做了三件事：

1. 申请一个没用过的fd
2. 创建一个struct file结构体，绑定一个匿名inode。同时绑定file->private_data = 分配的struct eventpoll，该结构体贯穿三个系统调用
3. 建立fd与struct file的关系，返回该fd

可以认为，是青春版open()。不同的地方就在于第二步。

对于open()来说，我是实实在在的希望能找到一个真正的文件，或者创建一个真正的文件，所以struct file最终绑定的inode都是在文件系统中能找到的，代表一个真实存在的文件。但是epoll()不一样，不需要一个真正存在的文件，只是希望凑齐三板斧struct file/struct dentry/struct inode，毕竟其它程序都是这么做的，为了适应大环境(比如说一些系统调用，一些vfs提供的api等等)也得这么干。但是真没必要来个真实的inode，所以就出现了一种奇怪的inode：匿名inode。

关于匿名inode，可以看一下这篇文章：Linux fd 系列“匿名句柄” 是一切皆文件背后功臣

我个人的理解呢，就是你需要文件系统那一套(因为Linux下一切皆文件)，但是又不需要真正存在这个文件，只是用它那一套规则，就可以用一个匿名inode。而且还能给struct file绑定自己设置的ops，而不是像open()时那样把inode->ops绑定到file->ops。还挺方便。

同时内核中有很多使用匿名inode的模块，所以把申请匿名inode的代码给平台化了，想用就调接口就行了：

fs/anon_inodes.c：
struct file *anon_inode_getfile(const char *name,
				const struct file_operations *fops,
				void *priv, int flags)
{
	struct file *file;

	if (IS_ERR(anon_inode_inode))
		return ERR_PTR(-ENODEV);

	if (fops->owner && !try_module_get(fops->owner))
		return ERR_PTR(-ENOENT);

	/*
	 * We know the anon_inode inode count is always greater than zero,
	 * so ihold() is safe.
	 */
	ihold(anon_inode_inode); // g, 判断inode->i_count+1后是否 < 2 

	// g, 虽然匿名inode不会有"实际的"dentry，但是该函数里面仍然会分配一个struct dentry绑定给file，这不过这个dentry不会有真实的路径
	// g, 传入的name就是dentry对应的name, anon_inode_mnt是分配dentry所要使用的文件系统，要用到该文件系统的super block。看样子文件系统叫anon_inode_mnt
	// g, 传入的anon_inode_inode是要与函数中分配的dentry绑定的inode，最终会绑定dentry->d_inode = inode;
	// g, 对于这个anon_inode_inode，会在start_kenerl时就进行申请和创建了。
	// g, 最终结果：绑定file->f_path = path, file->f_op = fops， path.dentry = 新分配的dentry, path.dentry->d_inode = anon_inode_inode，file->f_inode = path->dentry->d_inode;
	file = alloc_file_pseudo(anon_inode_inode, anon_inode_mnt, name,
				 flags & (O_ACCMODE | O_NONBLOCK), fops);
	if (IS_ERR(file))
		goto err;

	file->f_mapping = anon_inode_inode->i_mapping;

	file->private_data = priv;		// g,绑定申请的file->private_data = 申请的struct eventpoll结构体，会在后续的其他系统调用中用到

	return file;

err:
	iput(anon_inode_inode);		// g，对inode->i_count - 1
	module_put(fops->owner);
	return file;
}

直接调这个函数，申请一个struct dentry和struct file，并且初始化。最终初始化的结果就是:

file->f_op = fops;
file->f_inode = anon_inode_inode;
dentry->d_inode = anon_inode_inode;

fops由调用者传入，在epoll()模块中传入的是eventpoll_fops()，这个后面再说。

这个anon_inode_inode，就是一个提前创建好的匿名inode，其创建过程：

// g, 此函数会在linux启动早期调用，因为被xx_initcall宏注册了
// g, 其实本文件下的代码相当于是一个"匿名inode"文件系统，提供一些函数，提供一个匿名inode
static int __init anon_inode_init(void)
{
	// g, 先mount一个文件系统。然后将会在anon_inode_fs_type文件系统中分配inode，算是创建了一个vfsmount实例
	anon_inode_mnt = kern_mount(&anon_inode_fs_type);
	if (IS_ERR(anon_inode_mnt))
		panic("anon_inode_init() kernel mount failed (%ld)\n", PTR_ERR(anon_inode_mnt));


	// g, 关于xx_initcall宏，移步这一篇博客：https://blog.csdn.net/weiqifa0/article/details/136795242?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522172068035616800186560008%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=172068035616800186560008&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduend~default-1-136795242-null-null.142^v100^pc_search_result_base8&utm_term=fs_initcall&spm=1018.2226.3001.4187
	// g, 分配了一个匿名inode
	anon_inode_inode = alloc_anon_inode(anon_inode_mnt->mnt_sb);	// g, 很多模块都会用到这个匿名inode，anon_inode_inode
	if (IS_ERR(anon_inode_inode))
		panic("anon_inode_init() inode allocation failed (%ld)\n", PTR_ERR(anon_inode_inode));

	return 0;
}

2.2 epoll_ctl()

当我们使用epoll_create()创建了epollfd之后，我们需要通过epoll_ctl()来添加我们需要监听的fd，已经要监听的事件。epoll_ctl()的使用方法在开头的使用示例中已经描述。

在分析之前，先声明两个别名，后面的分析我会经常用到：

1. epollfd，是第一步通过epoll_create()创建的fd
2. tfd，也就是target fd，也就是我们要监控的目标fd

接下来就分析一下该函数在内核中的实现过程：

SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,
		struct epoll_event __user *, event)
{
	int error;
	int full_check = 0;
	struct fd f, tf;
	struct eventpoll *ep;
	struct epitem *epi;
	struct epoll_event epds;
	struct eventpoll *tep = NULL;

	error = -EFAULT;
	if (ep_op_has_event(op) &&		// g, 如果op不是删除操作
	    copy_from_user(&epds, event, sizeof(struct epoll_event)))	// g, 拷贝用户的event
		goto error_return;

	error = -EBADF;
	f = fdget(epfd);			// g, 获取epfd对应的struct file，也就是在epoll_create那一步创建的fd
	if (!f.file)
		goto error_return;

	/* Get the "struct file *" for the target file */
	tf = fdget(fd);				// g, 获取传入的fd对应的struct file。
	if (!tf.file)
		goto error_fput;

	/* The target file descriptor must support poll */
	error = -EPERM;
	if (!file_can_poll(tf.file))	// g, 检查该fd对应的file->f_op->poll是否为NULL，是否支持poll。是支持的，f_op指向了eventpoll_fops()，实现了.poll()
		goto error_tgt_fput;

	/* Check if EPOLLWAKEUP is allowed */
	if (ep_op_has_event(op))
		ep_take_care_of_epollwakeup(&epds);		// g, 清空epds->events的EPOLLWAKEUP这个FLAG:

	/*
	 * We have to check that the file structure underneath the file descriptor
	 * the user passed to us _is_ an eventpoll file. And also we do not permit
	 * adding an epoll file descriptor inside itself.
	 */
	error = -EINVAL;
	if (f.file == tf.file || !is_file_epoll(f.file))	// g, 传入的epfd和fd不能是一个，并且epfd的file->ops必须得是eventpoll_fops才能通过这个检查
		goto error_tgt_fput;

	/*
	 * epoll adds to the wakeup queue at EPOLL_CTL_ADD time only,
	 * so EPOLLEXCLUSIVE is not allowed for a EPOLL_CTL_MOD operation.
	 * Also, we do not currently supported nested exclusive wakeups.
	 */
	// g, 关于这个宏EPOLLEXCLUSIVE，是为了解决epoll的惊群效应。惊群效应是啥？百度一搜就知道了
	// g, 解决方法也没那么难，就是设置的等待队列项的一个标志位，在唤醒该等待队列时，所有带有这个标志位的等待队列项，只有一个会被处理(最先添加到等待队列中的会被处理)
	// g, 具体的处理过程，后面分析的时候到了对应的代码那里会分析。
	if (ep_op_has_event(op) && (epds.events & EPOLLEXCLUSIVE)) {	// g, 此函数ep_op_has_event()判断传入的操作是否是DEL，如果不是DEL，进一步处理
		if (op == EPOLL_CTL_MOD)									// g, 如果是MOD操作，则报错。因为EPOLLEXCLUSIVE不支持MOD
			goto error_tgt_fput;
		if (op == EPOLL_CTL_ADD && (is_file_epoll(tf.file) ||		// g, 虽然EPOLLEXCLUSIVE支持ADD操作，但是不支持嵌套epollfd
				(epds.events & ~EPOLLEXCLUSIVE_OK_BITS)))
			goto error_tgt_fput;
	}

	/*
	 * At this point it is safe to assume that the "private_data" contains
	 * our own data structure.
	 */
	ep = f.file->private_data;		// g, 在epoll_create()的时候，已经把创建的struct file->private_data设置为struct eventpoll

	/*
	 * When we insert an epoll file descriptor, inside another epoll file
	 * descriptor, there is the change of creating closed loops, which are
	 * better be handled here, than in more critical paths. While we are
	 * checking for loops we also determine the list of files reachable
	 * and hang them on the tfile_check_list, so we can check that we
	 * haven't created too many possible wakeup paths.
	 *
	 * We do not need to take the global 'epumutex' on EPOLL_CTL_ADD when
	 * the epoll file descriptor is attaching directly to a wakeup source,
	 * unless the epoll file descriptor is nested. The purpose of taking the
	 * 'epmutex' on add is to prevent complex toplogies such as loops and
	 * deep wakeup paths from forming in parallel through multiple
	 * EPOLL_CTL_ADD operations.
	 */
	mutex_lock_nested(&ep->mtx, 0);
	// g, 如果要添加的文件描述符本身也代表一个epoll实例，那么有可能会造成死循环，内核对此情况做了检查，如果存在死循环则返回错误。
	if (op == EPOLL_CTL_ADD) {
		if (!list_empty(&f.file->f_ep_links) ||
						is_file_epoll(tf.file)) {
			full_check = 1;
			mutex_unlock(&ep->mtx);
			mutex_lock(&epmutex);
			if (is_file_epoll(tf.file)) {				// g, 判断target_file是不是仍然是个epoll
				error = -ELOOP;
				if (ep_loop_check(ep, tf.file) != 0) {	// g, 检查把一个epollfd添加到另外一个epollfd会不会有问题
					goto error_tgt_fput;
				}
			} else
				list_add(&tf.file->f_tfile_llink,		// g, 这个f_tfile_llink只有开启了CONFIG_EPOLL宏的时候才会出现在struct file中
							&tfile_check_list);
			mutex_lock_nested(&ep->mtx, 0);
			if (is_file_epoll(tf.file)) {
				tep = tf.file->private_data;
				mutex_lock_nested(&tep->mtx, 1);
			}
		}
	}

	/*
	 * Try to lookup the file inside our RB tree, Since we grabbed "mtx"
	 * above, we can be sure to be able to use the item looked up by
	 * ep_find() till we release the mutex.
	 */
	epi = ep_find(ep, tf.file, fd);	// g, 看一下rb树中是否已经有这个fd了

	error = -EINVAL;
	switch (op) {
	case EPOLL_CTL_ADD:
		if (!epi) {
			epds.events |= EPOLLERR | EPOLLHUP;
			// ep:f->private_data, epds:用户传入的, tf:用户传入的target file
			// g, 申请并向rb树中插入一个新的epitem，会设置epitem->event = epds.events
			// g, 这里面ep_insert还会调用ep_item_poll()，即会调用vfs_poll()最终调到驱动实现的poll，驱动又会调用poll_wait()
			// g, poll_wait()函数中会执行的函数为ep_ptable_queue_proc()，该函数新分配一个等待队列项，然后添加到epi->ffd.file(也就是驱动设备)->private_data->poll_wait表示的等待队列头中
			// g, 并且会为该等待队列项注册回调，回调函数为：ep_poll_callback()，该函数负责把检测到events的结点加入到就绪队列rdlist中
			// g, 其中epi是新创建的，会绑定epi->ffd.file = tf.file(也就是target file)
			error = ep_insert(ep, &epds, tf.file, fd, full_check);  
		} else
			error = -EEXIST;
		break;
	case EPOLL_CTL_DEL:
		if (epi)
			error = ep_remove(ep, epi);
		else
			error = -ENOENT;
		break;
	case EPOLL_CTL_MOD:
		if (epi) {
			if (!(epi->event.events & EPOLLEXCLUSIVE)) {
				epds.events |= EPOLLERR | EPOLLHUP;
				error = ep_modify(ep, epi, &epds);
			}
		} else
			error = -ENOENT;
		break;
	}
	if (tep != NULL)
		mutex_unlock(&tep->mtx);
	mutex_unlock(&ep->mtx);

error_tgt_fput:
	if (full_check) {
		clear_tfile_check_list();
		mutex_unlock(&epmutex);
	}

	fdput(tf);
error_fput:
	fdput(f);
error_return:

	return error;
}

该函数做了如下几件事情：

1. 使用copy_from_user拷贝用户传入的struct event
2. 做一些检查，其中有一步是检查是否支持嵌套调用epoll
3. 检查通过后，搜索struct eventpoll保存的红黑树中是否存在传入的fd对应的节点
4. 根据epoll_ctl()传入的控制参数，分别处理新增/删除/调整命令。

首先声明，这个函数中有一个比较复杂的事情，就是可能存在嵌套调用epoll的现象，该函数的实现中有很多代码也是为了处理这种情况而写的。在分析过程中，遇到针对嵌套epollfd的情况，我都没有进行深入分析。

其次，epoll的实现过程中会调用一些NAPI函数，如果.config中如果没开启CONFIG_NET_RX_BUSY_POLL宏，这些函数都为空。我在我用的ubuntu的内核.config中发现该宏是开启的，但是仍然把它当做没开启，不去分析。

ps：嵌套epollfd，就是把一个使用epoll_create()创建的epollfd，作为一个被监控的fd对象，加入到另外一个epoll_create()创建的epollfd的监控列表中。NAPI是linux新的网卡数据处理API(New API)，采取了综合中断方式与轮询方式的技术来处理网卡数据，这个正点的网卡驱动有讲过。

检查过程就不展开说了，在注释中也都写上了。主要分析对不同命令的执行过程，以EPOLL_CTL_ADD命令为例：

	...
	...
	epi = ep_find(ep, tf.file, fd);	// g, 看一下rb树中是否已经有这个fd了

	error = -EINVAL;
	switch (op) {
	case EPOLL_CTL_ADD:
		if (!epi) {					// g, 如果rb tree中没有这个fd对应的节点
			epds.events |= EPOLLERR | EPOLLHUP;
			// ep:f->private_data, epds:用户传入的, tf:用户传入的target file
			// g, 申请并向rb树中插入一个新的epitem，会设置epitem->event = epds.events
			// g, 这里面ep_insert还会调用ep_item_poll()，即会调用vfs_poll()最终调到驱动实现的poll，驱动又会调用poll_wait()
			// g, poll_wait()函数中会执行的函数为ep_ptable_queue_proc()，该函数新分配一个等待队列项，然后添加到epi->ffd.file(也就是驱动设备)->private_data->poll_wait表示的等待队列头中
			// g, 并且会为该等待队列项注册回调，回调函数为：ep_poll_callback()，该函数负责把检测到events的结点加入到就绪队列rdlist中
			// g, 其中epi是新创建的，会绑定epi->ffd.file = tf.file(也就是target file)
			error = ep_insert(ep, &epds, tf.file, fd, full_check);  
		} else
			error = -EEXIST;
		break;
	case EPOLL_CTL_DEL:
		if (epi)
			error = ep_remove(ep, epi);
		else
			error = -ENOENT;
		break;
	case EPOLL_CTL_MOD:
		if (epi) {
			if (!(epi->event.events & EPOLLEXCLUSIVE)) {
				epds.events |= EPOLLERR | EPOLLHUP;
				error = ep_modify(ep, epi, &epds);
			}
		} else
			error = -ENOENT;
		break;
	}
	...
	...

若没有在rb tree中找到已有的fd对应的节点，那么就会执行ep_insert():

// g, 传入的参数：ep, &epds, tf.file, fd, full_check
static int ep_insert(struct eventpoll *ep, const struct epoll_event *event,
		     struct file *tfile, int fd, int full_check)
{
	int error, pwake = 0;
	__poll_t revents;
	long user_watches;
	struct epitem *epi;
	struct ep_pqueue epq;

	lockdep_assert_irqs_enabled();

	user_watches = atomic_long_read(&ep->user->epoll_watches);
	if (unlikely(user_watches >= max_user_watches))
		return -ENOSPC;
	if (!(epi = kmem_cache_alloc(epi_cache, GFP_KERNEL)))
		return -ENOMEM;

	/* Item initialization follow here ... */
	INIT_LIST_HEAD(&epi->rdllink);
	INIT_LIST_HEAD(&epi->fllink);
	INIT_LIST_HEAD(&epi->pwqlist);
	epi->ep = ep;
	ep_set_ffd(&epi->ffd, tfile, fd);		// g, 传入的tfile是fdf对应的file，也就是要监控的fd对应的file。最终epi->ffd->file = tfile, epi->fd =fd
	epi->event = *event;
	epi->nwait = 0;
	epi->next = EP_UNACTIVE_PTR;
	if (epi->event.events & EPOLLWAKEUP) {
		error = ep_create_wakeup_source(epi);
		if (error)
			goto error_create_wakeup_source;
	} else {
		RCU_INIT_POINTER(epi->ws, NULL);
	}

	/* Add the current item to the list of active epoll hook for this file */
	spin_lock(&tfile->f_lock);
	list_add_tail_rcu(&epi->fllink, &tfile->f_ep_links);
	spin_unlock(&tfile->f_lock);

	/*
	 * Add the current item to the RB tree. All RB tree operations are
	 * protected by "mtx", and ep_insert() is called with "mtx" held.
	 */
	ep_rbtree_insert(ep, epi);		// g, 向rb树中插入一个新的epitem

	/* now check if we've created too many backpaths */
	error = -EINVAL;
	if (full_check && reverse_path_check())
		goto error_remove_epi;

	/* Initialize the poll table using the queue callback */
	epq.epi = epi;

	// g, 等价于: epq.pt->qproc = ep_ptable_queue_proc
	// g, 这个函数的作用用一句话表示：用于新建一个等待队列项并注册其回调函数为ep_poll_callback，然后我们在驱动的poll中调用poll_wait()->pt->_qproc也就是ep_poll_callback，将新注册的等待队列项加入到驱动中注册的等待队列头中
	// g, 并为该等待队列项注回调:ep_poll_callback()，当该等待队列在驱动中被某个事件唤醒时会调用该函数，该函数大有所为，会把就绪的epoll item添加到双向链表rdlist中
	init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);

	/*
	 * Attach the item to the poll hooks and get current event bits.
	 * We can safely use the file* here because its usage count has
	 * been increased by the caller of this function. Note that after
	 * this operation completes, the poll callback can start hitting
	 * the new item.
	 */
	// g, 这里面最终调用到了poll_wait()，在poll()的解析中，这个poll_wait()是我们驱动来调用的
	// g, poll_wait()最终会调用绑定的pt->ep_ptable_queue_proc(), 把当前进程加入到了等待队列&ep->poll_wait中
	// g, 这里会先进行一次判断，判断等待的事件是否已经发生。如果revents不为0则说明要监听的事件已经发生了
	revents = ep_item_poll(epi, &epq.pt, 1);

	/*
	 * We have to check if something went wrong during the poll wait queue
	 * install process. Namely an allocation for a wait queue failed due
	 * high memory pressure.
	 */
	error = -ENOMEM;
	if (epi->nwait < 0)
		goto error_unregister;

	/* We have to drop the new item inside our item list to keep track of it */
	spin_lock_irq(&ep->wq.lock);

	/* record NAPI ID of new item if present */
	ep_set_busy_poll_napi_id(epi);	// g, 空

	/* If the file is already "ready" we drop it inside the ready list */
	if (revents && !ep_is_linked(epi)) {				// g, 我还没有wait，这个事件就已经发生了。
		list_add_tail(&epi->rdllink, &ep->rdllist);		// g, 那我直接把该epi加入到ep->rdllist中就好了，等会epoll_wait直接读到了
		ep_pm_stay_awake(epi);

		/* Notify waiting tasks that events are available */
		if (waitqueue_active(&ep->wq))
			wake_up_locked(&ep->wq);
		if (waitqueue_active(&ep->poll_wait))
			pwake++;
	}

	spin_unlock_irq(&ep->wq.lock);

	atomic_long_inc(&ep->user->epoll_watches);

	/* We have to call this outside the lock */
	if (pwake)
		ep_poll_safewake(&ep->poll_wait);

	return 0;

error_unregister:
	ep_unregister_pollwait(ep, epi);
error_remove_epi:
	spin_lock(&tfile->f_lock);
	list_del_rcu(&epi->fllink);
	spin_unlock(&tfile->f_lock);

	rb_erase_cached(&epi->rbn, &ep->rbr);

	/*
	 * We need to do this because an event could have been arrived on some
	 * allocated wait queue. Note that we don't care about the ep->ovflist
	 * list, since that is used/cleaned only inside a section bound by "mtx".
	 * And ep_insert() is called with "mtx" held.
	 */
	spin_lock_irq(&ep->wq.lock);
	if (ep_is_linked(epi))
		list_del_init(&epi->rdllink);
	spin_unlock_irq(&ep->wq.lock);

	wakeup_source_unregister(ep_wakeup_source(epi));

error_create_wakeup_source:
	kmem_cache_free(epi_cache, epi);

	return error;
}

该函数会：

1. 根据用户传入的信息，创建初始化一个struct epitem对象，该对象是红黑树的中的节点对象
2. 将新创建的struct epitem对象添加到红黑树列表中
3. 通过ep_item_poll()函数，最终调用tfd对应的文件实现的poll()，如果是设备fd，那么此处就是设备驱动实现的dev_poll()

最重要的是第三步，这一步工作的实现如下：

	...
	...
	// g, 等价于: epq.pt->qproc = ep_ptable_queue_proc
	// g, 这个函数的作用用一句话表示：用于新建一个等待队列项并注册其回调函数为ep_poll_callback，然后我们在驱动的poll中调用poll_wait()->pt->_qproc也就是ep_poll_callback，将新注册的等待队列项加入到驱动中注册的等待队列头中
	// g, 并为该等待队列项注回调:ep_poll_callback()，当该等待队列在驱动中被某个事件唤醒时会调用该函数，该函数大有所为，会把就绪的epoll item添加到双向链表rdlist中
	init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);

	/*
	 * Attach the item to the poll hooks and get current event bits.
	 * We can safely use the file* here because its usage count has
	 * been increased by the caller of this function. Note that after
	 * this operation completes, the poll callback can start hitting
	 * the new item.
	 */
	// g, 这里面最终调用到了poll_wait()，在poll()的解析中，这个poll_wait()是我们驱动来调用的
	// g, poll_wait()最终会调用绑定的pt->ep_ptable_queue_proc(), 把当前进程加入到了等待队列&ep->poll_wait中
	// g, 这里会先进行一次判断，判断等待的事件是否已经发生。如果revents不为0则说明要监听的事件已经发生了
	revents = ep_item_poll(epi, &epq.pt, 1);
	...
	...

其中第一步init_poll_funcptr()的作用与poll()系统调用中poll_initwait()函数的作用一样，都是在初始化struct poll_table_struct的_qproc域，该域为一个函数指针，当tfd实现的poll()中调用poll_wait()函数时，最终会调用到pt->_qproc。

在这里可以看到，为该域注册了函数：ep_ptable_queue_proc()。该函数等会分析。执行完注册工作后，就会调用ep_item_poll()函数：

static __poll_t ep_item_poll(const struct epitem *epi, poll_table *pt,
				 int depth)
{
	struct eventpoll *ep;
	bool locked;

	pt->_key = epi->event.events;
	if (!is_file_epoll(epi->ffd.file))							// g, 如果target file的f_op不是eventpoll_fops。说明不是epollfd嵌套的情况
		return vfs_poll(epi->ffd.file, pt) & epi->event.events;	// g, 调用target file的poll，也就是驱动中实现的poll，所以！！最终我们仍然要在驱动中，执行poll_wait()!!，传入自己的等待队列头，唤醒工作也是我们在驱动中做的事情！

	/* -------- 分割线，如果走到下面几步，说明出现了epollfd嵌套的情况，就不分析这种情况了，比较复杂 -----------*/
	ep = epi->ffd.file->private_data;					
	// g, 会调用pt->_qproc()，也就是ep_ptable_queue_proc()
	// g, ep_ptable_queue_proc()会把一个新初始化的等待队列加入到ep->poll_wait这个等待队列头链表中，也就是epi->ffd.file->private_data->poll_wait，也就是target file->private_data->poll_wait
	poll_wait(epi->ffd.file, &ep->poll_wait, pt);	// g, 嵌套epollfd的情况，第二个epollfd作为被监控的fd，自然是要调用一次poll_wait()
	locked = pt && (pt->_qproc == ep_ptable_queue_proc);

	return ep_scan_ready_list(epi->ffd.file->private_data,
				  ep_read_events_proc, &depth, depth,
				  locked) & epi->event.events;
}

如果tfd对应的struct file中绑定的f_op不是eventpoll_fops(epoll_create()中会为生成epollfd对应的struct file绑定该ops)，说明tfd不是一个epollfd，也就说没有出现嵌套epollfd的情况，此时就会直接调用vfs_poll()，最终调用到tfd实现的poll()，如果是设备fd，那么最终调用到设备驱动中实现的dev_poll()。

关于vfs_xx()，虚拟文件系统提供的api，最终都会调用struct file的f_op，f_op的绑定在分析open()系统调用时已经讲过了。

对于epoll()系统调用来说，驱动中实现的dev_poll()，与poll()系统调用对驱动程序的要求是一样的，就是要初始化一个等待队列头，然后在dev_poll()中调用poll_wait()函数，从而最终调用到我们注册的pt->_qproc，也就是刚才提到的ep_ptable_queue_proc()：

static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
				 poll_table *pt)
{
	struct epitem *epi = ep_item_from_epqueue(pt);
	struct eppoll_entry *pwq;

	// ep->poll_wait
	// g,初始化一个等待队列项，其中注册的回调为ep_poll_callback
	if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {	// g,分配一个新的pwq
		init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);	// g, 给等待信号&pwq->wait添加回调ep_poll_callback()
		pwq->whead = whead;											// g, 这个就是监控的fd中创建的等待队列头(比如我们的驱动中创建的等待队列头),target file
		pwq->base = epi;
		// g, 如果传入的events有EPOLLEXCLUSIVE标志，则表示希望解决惊群效应。所谓惊群效应，就是多个进程调用epoll()监控同一个fd，当fd的事件来临时会把他们全部唤醒
		// g, 但是使用add_wait_queue_exclusive()这个函数，会为等待队列结点添加一个WQ_FLAG_EXCLUSIVE标志。
		// g, 有了该标志，会使得所有带有该标志的等待队列项，一次只有一个被唤醒。
		if (epi->event.events & EPOLLEXCLUSIVE)						
			add_wait_queue_exclusive(whead, &pwq->wait);  
		else
			add_wait_queue(whead, &pwq->wait);			// g, 加入等待队列头whead，等待队列头就是整个等待队列的链表头,这里把初始化的等待对列项插入到所监控的fd的等待对列中
		list_add_tail(&pwq->llink, &epi->pwqlist);
		epi->nwait++;
	} else {
		/* We have to signal that an error occurred */
		epi->nwait = -1;
	}
}

与poll()系统调用中为pt->_qproc注册的函数作用基本一致。但是这里根据是否传入了标志EPOLLEXCLUSIVE，会选择不同的向等待队列(该等待队列的等待队列头在驱动程序中初始化，由驱动程序调用poll_wait()函数时传入。)中添加等待队列项的接口。

add_wait_queue_exclusive这个函数的分析等到后面分析内核的调度模块再写出来吧，不放在这里写了，作用就是注释写的作用。

同时为添加到等待队列中的等待队列项注册回调函数：ep_poll_callback()，当驱动程序唤醒这个等待队列时，唤醒过程中就会执行该回调函数。该函数会放在epoll_wait()中分析，该函数会操作struct epollevent中的维护就绪事件的另一个域rdllist，是一个双向链表，这也是epoll比poll高效的原因之一。

对于其他控制命令如EPOLL_CTL_DEL或者EPOLL_CTL_MOD对应的处理过程，有兴趣的可以自行分析。因为感觉跟驱动中实现的dev_poll()联系不大，我就不分析了。

2.3 epoll_wait()

该系统调用在内核中的实现：

fs/eventpoll.c：
SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events,
		int, maxevents, int, timeout)
{
	return do_epoll_wait(epfd, events, maxevents, timeout);
}

单纯的调用了do_epoll_wait()函数：

fs/eventpoll.c：
// g, events:用于回传代处理事件的数组
static int do_epoll_wait(int epfd, struct epoll_event __user *events,
			 int maxevents, int timeout)
{
	int error;
	struct fd f;
	struct eventpoll *ep;

	/* The maximum number of event must be greater than zero */
	if (maxevents <= 0 || maxevents > EP_MAX_EVENTS)
		return -EINVAL;

	/* Verify that the area passed by the user is writeable */
	if (!access_ok(VERIFY_WRITE, events, maxevents * sizeof(struct epoll_event)))
		return -EFAULT;

	/* Get the "struct file *" for the eventpoll file */
	f = fdget(epfd);			// g, 获取epoll_create()创建的匿名inode
	if (!f.file)
		return -EBADF;

	/*
	 * We have to check that the file structure underneath the fd
	 * the user passed to us _is_ an eventpoll file.
	 */
	error = -EINVAL;
	if (!is_file_epoll(f.file))
		goto error_fput;

	/*
	 * At this point it is safe to assume that the "private_data" contains
	 * our own data structure.
	 */
	ep = f.file->private_data;		// g, 获取epfd的私有数据,也就是struct eventpoll，由epoll_create()时绑定到epollfd的private_data中，贯穿三个系统调用

	/* Time to fish for events ... */
	error = ep_poll(ep, events, maxevents, timeout);	// g, 完成主要工作的函数

error_fput:
	fdput(f);
	return error;
}

该函数做了些检查和获取struct eventpoll的工作，然后调用了ep_poll()函数，该函数做了主要的工作：

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	int res = 0, eavail, timed_out = 0;
	u64 slack = 0;
	wait_queue_entry_t wait;
	ktime_t expires, *to = NULL;

	lockdep_assert_irqs_enabled();

	// g, 这一段设置等待时间
	if (timeout > 0) {
		struct timespec64 end_time = ep_set_mstimeout(timeout);

		slack = select_estimate_accuracy(&end_time);
		to = &expires;
		*to = timespec64_to_ktime(end_time);
	} else if (timeout == 0) {		// g, 用户态传入的时间i为0，不希望阻塞，需要直接去判断是否有事件发生，所以t需要直接goto check_events
		/*
		 * Avoid the unnecessary trip to the wait queue loop, if the
		 * caller specified a non blocking operation.
		 */
		timed_out = 1;
		spin_lock_irq(&ep->wq.lock);
		goto check_events;
	}

fetch_events:

	// g, 下面这个函数检测当前epoll上是否已经收集了有效的事件(此时还没开始休眠)，那么加上!则表示 是否 当前 没有 有效数据集
	if (!ep_events_available(ep))			// g, 如果没事件发生
		// g, 依赖于宏CONFIG_NET_RX_BUSY_POLL，这个宏在.config中确实开了，但是当做它没开，因为不好分析。没开的时候为空函数
		// g, 此处涉及到NAPI技术，是linux新的网卡数据处理API(New API)，采用了中断和轮询的方式处理网卡数据。
		ep_busy_loop(ep, timed_out);		

	spin_lock_irq(&ep->wq.lock);

	if (!ep_events_available(ep)) {			// g, 没有可用的事件，ready list 和ovflist 都为空
		/*
		 * Busy poll timed out.  Drop NAPI ID for now, we can add
		 * it back in when we have moved a socket with a valid NAPI
		 * ID onto the ready list.
		 */
		ep_reset_busy_poll_napi_id(ep);		// g, 依赖于宏CONFIG_NET_RX_BUSY_POLL

		/*
		 * We don't have any available event to return to the caller.
		 * We need to sleep here, and we will be wake up by
		 * ep_poll_callback() when events will become available.
		 */
		// g, 初始化一个等待队列项,会绑定wait->p = current。default_wake_function()需要wait->p
		// g, 并且设置wait->func为默认的default_wake_function()
		init_waitqueue_entry(&wait, current);			// g, 与init_waitqueue_func_entry相比少了设置func这一项
		__add_wait_queue_exclusive(&ep->wq, &wait);		// g, 加入wait到ep->wq等待队列头，使用_exclusive保证在唤醒ep->wq等待队列时，所有使用_exclusive添加进去的等待队列项，只有一个会被处理

		for (;;) {
			/*
			 * We don't want to sleep if the ep_poll_callback() sends us
			 * a wakeup in between. That's why we set the task state
			 * to TASK_INTERRUPTIBLE before doing the checks.
			 */
			set_current_state(TASK_INTERRUPTIBLE);
			/*
			 * Always short-circuit for fatal signals to allow
			 * threads to make a timely exit without the chance of
			 * finding more events available and fetching
			 * repeatedly.
			 */
			if (fatal_signal_pending(current)) {
				res = -EINTR;
				break;
			}
			if (ep_events_available(ep) || timed_out)		// g, 如果有事件发生了，或者说超时了，就跳出循环。当有事件发生的时候驱动会唤醒等待队列，然后回调函数把节点添加到rdlist中
				break;
			if (signal_pending(current)) {
				res = -EINTR;
				break;
			}

			spin_unlock_irq(&ep->wq.lock);
			if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS))		// g, 休眠传入的时间,等待被唤醒
				timed_out = 1;

			spin_lock_irq(&ep->wq.lock);
		}

		__remove_wait_queue(&ep->wq, &wait);
		__set_current_state(TASK_RUNNING);
	}
check_events:
	/* Is it worth to try to dig for events ? */
	eavail = ep_events_available(ep);		// g, 检测事件发生的情况。能到这一步，要么是上面的循环被打破，要么是用户不希望阻塞

	spin_unlock_irq(&ep->wq.lock);

	/*
	 * Try to transfer events to user space. In case we get 0 events and
	 * there's still timeout left over, we go trying again in search of
	 * more luck.
	 */
	if (!res && eavail &&
	    !(res = ep_send_events(ep, events, maxevents)) && !timed_out)		// g, 获取结果并发送到用户空间
		goto fetch_events;

	return res;
}

总结一下该函数做了什么事情：

1. 根据用户传入的等待时间设置转换为时间结构体。同时判断用户希不希望立刻返回(传入时间为0)
2. 死循环检查就绪队列rdllist中是否有就绪的事件，如果没有，则休眠。直到被唤醒或者超时，则会跳出循环
3. 跳出循环后，再次检查事件是否真正发生，向用户空间返回结果。

对于一些检查函数和用户传入的时间的转换工作，就不展开分析了，注释中都写了。主要是分析一下第二步和第三步。首先分析一下休眠过程：

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	...
	...
		else if (timeout == 0) {		// g, 用户态传入的时间i为0，不希望阻塞，需要直接去判断是否有事件发生，所以t需要直接goto check_events
		timed_out = 1;
		spin_lock_irq(&ep->wq.lock);
		goto check_events;
	}

fetch_events:
	...
	...
	if (!ep_events_available(ep)) {			// g, 没有可用的事件，ready list 和ovflist 都为空
		ep_reset_busy_poll_napi_id(ep);		// g, 依赖于宏CONFIG_NET_RX_BUSY_POLL

		// g, 初始化一个等待队列项,会绑定wait->p = current。default_wake_function()需要wait->p
		// g, 并且设置wait->func为默认的default_wake_function()
		init_waitqueue_entry(&wait, current);			// g, 与init_waitqueue_func_entry相比少了设置func这一项
		__add_wait_queue_exclusive(&ep->wq, &wait);		// g, 加入wait到ep->wq等待队列头，使用_exclusive保证在唤醒ep->wq等待队列时，所有使用_exclusive添加进去的等待队列项，只有一个会被处理

		for (;;) {
			set_current_state(TASK_INTERRUPTIBLE);
			if (fatal_signal_pending(current)) {
				res = -EINTR;
				break;
			}
			if (ep_events_available(ep) || timed_out)		// g, 如果有事件发生了，或者说超时了，就跳出循环。当有事件发生的时候驱动会唤醒等待队列，然后回调函数把节点添加到rdlist中
				break;
			if (signal_pending(current)) {
				res = -EINTR;
				break;
			}

			spin_unlock_irq(&ep->wq.lock);
			if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS))		// g, 休眠传入的时间,等待被唤醒
				timed_out = 1;

			spin_lock_irq(&ep->wq.lock);
		}

		__remove_wait_queue(&ep->wq, &wait);
		__set_current_state(TASK_RUNNING);
	}
check_events:
	...
	...
	return res;
}

如果用户传入的等待时间为0，则不希望休眠，直接跳过休眠阶段，去检查就绪队列。否则就要进行休眠处理。这一部分的处理逻辑是：

1. 不希望休眠，跳过休眠
2. 希望休眠，则初始化一个等待队列项，使用默认的唤醒回调函数default_wake_function()。该函数在poll()系统调用介绍过了，会唤醒休眠的进程
3. 进入死循环，首先判断在到epoll_wait()期间是否已经有事件就绪了，有的话，则break，不需要休眠
4. 如果没有时间就绪，则将等待队列项加入到等待队列头ep->wq中。这是struct eventpoll中的一个域。既然是struct eventpoll中的等待队列头，那么可以预见的就是唤醒该等待队列一定是eventpoll.c下的某一处代码完成的。
5. 调用schedule_hrtimeout_range休眠，等待被唤醒或者超时

还记的上一节讲的，调用驱动程序的dev_poll()，最终会向驱动程序中实现的等待队列中添加等待队列项吗，当时注册的等待队列项的回调函数是ep_poll_callback()，这里可以展开分析一下了：

fs/eventpoll.c：
static int ep_poll_callback(wait_queue_entry_t *wait, unsigned mode, int sync, void *key)
{
	int pwake = 0;
	unsigned long flags;
	struct epitem *epi = ep_item_from_wait(wait);
	struct eventpoll *ep = epi->ep;
	__poll_t pollflags = key_to_poll(key); // g, key已经被设置为了event.events
	int ewake = 0;

	spin_lock_irqsave(&ep->wq.lock, flags);

	ep_set_busy_poll_napi_id(epi);

	/*
	 * If the event mask does not contain any poll(2) event, we consider the
	 * descriptor to be disabled. This condition is likely the effect of the
	 * EPOLLONESHOT bit that disables the descriptor when an event is received,
	 * until the next EPOLL_CTL_MOD will be issued.
	 */
	// g, 如果这个epi->event.events设置了EP_PRIVATE_BITS标志，那么就会直接跳到结尾。
	// g, 该标志表明该事件只会监听一次，依赖于用户传入的EPOLLONESHOT表示，若存在该标志，则会在第一次向用户空间返回结果时设置EP_PRIVATE_BITS。
	if (!(epi->event.events & ~EP_PRIVATE_BITS))			
		goto out_unlock;

	/*
	 * Check the events coming with the callback. At this stage, not
	 * every device reports the events in the "key" parameter of the
	 * callback. We need to be able to handle both cases here, hence the
	 * test for "key" != NULL before the event match test.
	 */
	if (pollflags && !(pollflags & epi->event.events))
		goto out_unlock;

	/*
	 * If we are transferring events to userspace, we can hold no locks
	 * (because we're accessing user memory, and because of linux f_op->poll()
	 * semantics). All the events that happen during that period of time are
	 * chained in ep->ovflist and requeued later on.
	 */
	if (ep->ovflist != EP_UNACTIVE_PTR) {
		if (epi->next == EP_UNACTIVE_PTR) {
			epi->next = ep->ovflist;
			ep->ovflist = epi;
			if (epi->ws) {
				/*
				 * Activate ep->ws since epi->ws may get
				 * deactivated at any time.
				 */
				__pm_stay_awake(ep->ws);
			}

		}
		goto out_unlock;
	}

	/* If this file is already in the ready list we exit soon */
	if (!ep_is_linked(epi)) {
		list_add_tail(&epi->rdllink, &ep->rdllist);		// g, 把当前节点添加到readylist中，当驱动唤醒该等待队列时就会这么做。
		ep_pm_stay_awake_rcu(epi);
	}

	/*
	 * Wake up ( if active ) both the eventpoll wait list and the ->poll()
	 * wait list.
	 */
	if (waitqueue_active(&ep->wq)) {			// g, epoll_wait()中会向ep->wq中添加等待队列项，所以执行过epoll_wait()这个就不为空
		if ((epi->event.events & EPOLLEXCLUSIVE) &&
					!(pollflags & POLLFREE)) {
			switch (pollflags & EPOLLINOUT_BITS) {
			case EPOLLIN:
				if (epi->event.events & EPOLLIN)		// g, 请求检测EPOLLIN就会检测EPOLLIN
					ewake = 1;
				break;
			case EPOLLOUT:
				if (epi->event.events & EPOLLOUT)		// g, 请求检测EPOLLOUT就会检测EPOLLIN
					ewake = 1;
				break;
			case 0:
				ewake = 1;
				break;
			}
		}
		wake_up_locked(&ep->wq);					// g, 在这里才是真正唤醒被epoll阻塞的进程！！！！
	}
	if (waitqueue_active(&ep->poll_wait))			// g, 嵌套epollfd相关，只有出现嵌套epollfd的时候，ep->poll_wait才不为空！！
		pwake++;

out_unlock:
	spin_unlock_irqrestore(&ep->wq.lock, flags);

	/* We have to call this outside the lock */
	if (pwake)
		ep_poll_safewake(&ep->poll_wait);			// g, 唤醒嵌套epollfd的等待队列。网上的所有博客，都认为该函数是唤醒被epoll的阻塞进程，这是不对的

	if (!(epi->event.events & EPOLLEXCLUSIVE))
		ewake = 1;

	if (pollflags & POLLFREE) {
		/*
		 * If we race with ep_remove_wait_queue() it can miss
		 * ->whead = NULL and do another remove_wait_queue() after
		 * us, so we can't use __remove_wait_queue().
		 */
		list_del_init(&wait->entry);
		/*
		 * ->whead != NULL protects us from the race with ep_free()
		 * or ep_remove(), ep_remove_wait_queue() takes whead->lock
		 * held by the caller. Once we nullify it, nothing protects
		 * ep/epi or even wait.
		 */
		smp_store_release(&ep_pwq_from_wait(wait)->whead, NULL);
	}

	return ewake;
}

比较关键的地方有三点：

• 向rdllist中添加该节点
• 通过wake_up_locked(&ep->wq)唤醒之前在死循环中休眠的进程
• 修改ep->ovflist

第一点是高效的原因之一，就是唤醒进程后不需要再去遍历所有检测的fd，只需要遍历处理就绪队列即可。

第二点要提一下，很多博客可能根本没仔细分析代码，认为下面这段代码是唤醒休眠进程的：

if (pwake)
		ep_poll_safewake(&ep->poll_wait);			

实际上这个代码是唤醒的ep->poll_wait，这个等待队列是一个epollfd作为嵌套epollfd中被监听的fd时才会有的。我们要唤醒的等待队列是ep->wq。

第三点后面会用到，因为在向用户空间返回结果时，仍然有可能出现新的就绪事件，需要依赖于ovflist域进行判断

唤醒了休眠的进程后，会在schedule_hrtimeout_range()处继续执行，然后跳出循环进入下一步，也就是通过ep_send_events()函数向用户空间返回结果：

fs/eventpoll.c：
static int ep_send_events(struct eventpoll *ep,
			  struct epoll_event __user *events, int maxevents)
{
	struct ep_send_events_data esed;

	esed.maxevents = maxevents;				// g, 数值用户空间传入的参数，也就是最大events值
	esed.events = events;								// g, 用户空间传入地址

	ep_scan_ready_list(ep, ep_send_events_proc, &esed, 0, false);
	return esed.res;
}

该函数使用用户传入的参数初始化一个struct ep_send_events_data结构体，然后调用ep_scan_ready_list()函数：

static __poll_t ep_scan_ready_list(struct eventpoll *ep,
			      __poll_t (*sproc)(struct eventpoll *,
					   struct list_head *, void *),
			      void *priv, int depth, bool ep_locked)
{
	__poll_t res;
	int pwake = 0;
	struct epitem *epi, *nepi;
	LIST_HEAD(txlist);

	lockdep_assert_irqs_enabled();

	/*
	 * We need to lock this because we could be hit by
	 * eventpoll_release_file() and epoll_ctl().
	 */

	if (!ep_locked)
		mutex_lock_nested(&ep->mtx, depth);

	/*
	 * Steal the ready list, and re-init the original one to the
	 * empty list. Also, set ep->ovflist to NULL so that events
	 * happening while looping w/out locks, are not lost. We cannot
	 * have the poll callback to queue directly on ep->rdllist,
	 * because we want the "sproc" callback to be able to do it
	 * in a lockless way.
	 */
	spin_lock_irq(&ep->wq.lock);
	list_splice_init(&ep->rdllist, &txlist);	// g, 该函数把ep->rdllist移动到txlist中，然后把ep->rdllist置为空链表
	ep->ovflist = NULL; 
	spin_unlock_irq(&ep->wq.lock);

	/*
	 * Now call the callback function.
	 */
	res = (*sproc)(ep, &txlist, priv);	// g, 执行ep_send_events_proc(ep, &txlist, &esed)

	spin_lock_irq(&ep->wq.lock);
	/*
	 * During the time we spent inside the "sproc" callback, some
	 * other events might have been queued by the poll callback.
	 * We re-insert them inside the main ready-list here.
	 * 这一段话的意思就是，在我们调用ep_send_events_proc()函数期间，可能又有新的fd就绪并发出事件了，
	 */
	for (nepi = ep->ovflist; (epi = nepi) != NULL;		// g, ep->ovflist会在等待队列项的call_back函数中设置，所以一旦不为NULL，说明有fd唤醒了某个等待队列
	     nepi = epi->next, epi->next = EP_UNACTIVE_PTR) {	/// g, ep->ovflist是发生事件的epi链表的表头
		/*
		 * We need to check if the item is already in the list.
		 * During the "sproc" callback execution time, items are
		 * queued into ->ovflist but the "txlist" might already
		 * contain them, and the list_splice() below takes care of them.
		 */
		if (!ep_is_linked(epi)) {
			list_add_tail(&epi->rdllink, &ep->rdllist);
			ep_pm_stay_awake(epi);
		}
	}
	/*
	 * We need to set back ep->ovflist to EP_UNACTIVE_PTR, so that after
	 * releasing the lock, events will be queued in the normal way inside
	 * ep->rdllist.
	 */
	ep->ovflist = EP_UNACTIVE_PTR;			// g, 将ep->ovflist合并到ep->rdllist中后，重新指向一个无效PTR

	/*
	 * Quickly re-inject items left on "txlist".
	 */
	/* ------------------- g, 下面这一段已经是为了处理下一此调用epoll_wait()所做的工作了 ------------*/
	list_splice(&txlist, &ep->rdllist);			// g, 合并两个链表，txlist可能剩下一些事件(因为可能在ep_send_events_proc()中再次确认事件的时候又不通过了)
	__pm_relax(ep->ws);

	if (!list_empty(&ep->rdllist)) {			// g, 不为空，说明又又又又有事件发生了
		/*
		 * Wake up (if active) both the eventpoll wait list and
		 * the ->poll() wait list (delayed after we release the lock).
		 */
		if (waitqueue_active(&ep->wq))			// g, 如果ep->wq有等待项，才会唤醒。说实话本次epoll_wait()不会有了，因为epoll_wait只会调用一次，只会向ep->wq中添加一次等待项
			wake_up_locked(&ep->wq);				
		if (waitqueue_active(&ep->poll_wait))
			pwake++;
	}
	spin_unlock_irq(&ep->wq.lock);

	if (!ep_locked)
		mutex_unlock(&ep->mtx);

	/* We have to call this outside the lock */
	if (pwake)
		ep_poll_safewake(&ep->poll_wait);

	return res;
}

该函数的工作有以下几点：

1. 新初始化一个新的列表txlist，把ep->rdllist移动到txlist中，然后把ep->rdllist置为空链表
2. 执行ep_send_events_proc(ep, &txlist, &esed)，完成填充用户空间地址的工作
3. 检测在调用ep_send_events_proc期间，是否又有新的事件就绪，并对其进行处理

其中向用户空间返回的工作在ep_send_events_proc()中实现：

static __poll_t ep_send_events_proc(struct eventpoll *ep, struct list_head *head,
			       void *priv)
{
	struct ep_send_events_data *esed = priv;
	__poll_t revents;
	struct epitem *epi;
	struct epoll_event __user *uevent;
	struct wakeup_source *ws;
	poll_table pt;

	init_poll_funcptr(&pt, NULL);				// g, 重新初始化pt->_qproc，指向NULL，这样在接下来调用到设备驱动的dev_poll()的时候，调用poll_wait()就没有任何效果，只会return mask；

	/*
	 * We can loop without lock because we are passed a task private list.
	 * Items cannot vanish during the loop because ep_scan_ready_list() is
	 * holding "mtx" during this call.
	 */
	for (esed->res = 0, uevent = esed->events;
	     !list_empty(head) && esed->res < esed->maxevents;) {		// g, esed->maxevents是用户传入的最大监听数量，不得超过
		epi = list_first_entry(head, struct epitem, rdllink);		// g, 遍历就绪队列rdllist中已经就绪的事件，每次都获取第一元素(因为会不断删除，所以头元素会不停更换)

		/*
		 * Activate ep->ws before deactivating epi->ws to prevent
		 * triggering auto-suspend here (in case we reactive epi->ws
		 * below).
		 *
		 * This could be rearranged to delay the deactivation of epi->ws
		 * instead, but then epi->ws would temporarily be out of sync
		 * with ep_is_linked().
		 */
		ws = ep_wakeup_source(epi);
		if (ws) {
			if (ws->active)
				__pm_stay_awake(ep->ws);
			__pm_relax(ws);
		}

		list_del_init(&epi->rdllink);						// g, 从rdllink链表中删去这一项。

		revents = ep_item_poll(epi, &pt, 1);				// g, 再次进行一遍查询，确定是否真的就绪了。最终返回的revents就是驱动dev_poll()返回的mask

		/*
		 * If the event mask intersect the caller-requested one,
		 * deliver the event to userspace. Again, ep_scan_ready_list()
		 * is holding "mtx", so no operations coming from userspace
		 * can change the item.
		 */
		if (revents) {										// g, 如果确实就绪了，有事件发生了
			if (__put_user(revents, &uevent->events) ||
			    __put_user(epi->event.data, &uevent->data)) {
				list_add(&epi->rdllink, head);
				ep_pm_stay_awake(epi);
				if (!esed->res)
					esed->res = -EFAULT;
				return 0;
			}
			esed->res++;									// g, +1
			uevent++;										// g, +1
			if (epi->event.events & EPOLLONESHOT)			// g, 判断是否是一次性触发时间
				epi->event.events &= EP_PRIVATE_BITS;		// g, 如果是的话，则设置EP_PRIVATE_BITS标志，有了该标志则不会被再次加入到rdllist中
			else if (!(epi->event.events & EPOLLET)) {		// g, 该标志ET是边缘触发，没有该标志则是LT触发，会导致持续事件监控持续触发，只要事件存在就会一直触发。默认LT触发
				/*
				 * If this file has been added with Level
				 * Trigger mode, we need to insert back inside
				 * the ready list, so that the next call to
				 * epoll_wait() will check again the events
				 * availability. At this point, no one can insert
				 * into ep->rdllist besides us. The epoll_ctl()
				 * callers are locked out by
				 * ep_scan_ready_list() holding "mtx" and the
				 * poll callback will queue them in ep->ovflist.
				 */
				list_add_tail(&epi->rdllink, &ep->rdllist);		// g, 一直触发的方式也很简单，就是再把就绪项加回到就绪队列即可。
				ep_pm_stay_awake(epi);
			}
		}
	}

	return 0;
}

该函数完成了三项工作：

1. 再次调用设备驱动的dev_poll()，确认是不是真的有事件发生了。
2. 通过__put_user向用户空间拷贝信息，说明，根本没有共享内存
3. 判断标志，是否是一次性触发事件，是否是边缘触发事件

没什么好展开说的，复习的时候注重看下注释吧。该函数返回之后，会在ep_scan_ready_list()函数中继续处理确认失败的事件(再次确认时确认失败)，和在调用ep_send_events_proc()期间就绪的事件。这些事件都会被重新加入到rdllist中，在下一次调用epoll_wait()时可以监测到