epoll源码剖析

（主要基于Linux-2.6.11.12版本进行分析。）

1. 主要数据结构

struct eventpoll {
	/* Protect the this structure access */
	rwlock_t lock;

	/*
	 * This semaphore is used to ensure that files are not removed
	 * while epoll is using them. This is read-held during the event
	 * collection loop and it is write-held during the file cleanup
	 * path, the epoll file exit code and the ctl operations.
	 */
	struct rw_semaphore sem;

	/* Wait queue used by sys_epoll_wait() */
	wait_queue_head_t wq;

	/* Wait queue used by file->poll() */
	wait_queue_head_t poll_wait;

	/* List of ready file descriptors */
	struct list_head rdllist;

	/* RB-Tree root used to store monitored fd structs */
	struct rb_root rbr;
};

struct epitem {
	/* RB-Tree node used to link this structure to the eventpoll rb-tree */
	struct rb_node rbn;

	/* List header used to link this structure to the eventpoll ready list */
	struct list_head rdllink;

	/* The file descriptor information this item refers to */
	struct epoll_filefd ffd;

	/* Number of active wait queue attached to poll operations */
	int nwait;

	/* List containing poll wait queues */
	struct list_head pwqlist;

	/* The "container" of this item */
	struct eventpoll *ep;

	/* The structure that describe the interested events and the source fd */
	struct epoll_event event;

	/*
	 * Used to keep track of the usage count of the structure. This avoids
	 * that the structure will desappear from underneath our processing.
	 */
	atomic_t usecnt;

	/* List header used to link this item to the "struct file" items list */
	struct list_head fllink;

	/* List header used to link the item to the transfer list */
	struct list_head txlink;

	/*
	 * This is used during the collection/transfer of events to userspace
	 * to pin items empty events set.
	 */
	unsigned int revents;
};

struct eppoll_entry {
	/* List header used to link this structure to the "struct epitem" */
	struct list_head llink;

	/* The "base" pointer is set to the container "struct epitem" */
	void *base;

	/*
	 * Wait queue item that will be linked to the target file wait
	 * queue head.
	 */
	wait_queue_t wait;

	/* The wait queue head that linked the "wait" wait queue item */
	wait_queue_head_t *whead;
};

文件系统结构

/**
 * 对内核支持的每一种文件系统，存在一个这样的结构对其进行描述。
 */
struct file_system_type {
	/**
	 * 文件系统类型的名称 
	 */
	const char *name;
	/**
	 * 此文件系统类型的属性 
	 */
	int fs_flags;
	/**
	 * 函数指针，当安装此类型的文件系统时，就由VFS调用此例程从设备上将此文件系统的superblock读入内存中
	 */	
	struct super_block *(*get_sb) (struct file_system_type *, int,
				       const char *, void *);
	/**
	 * 删除超级块的方法。
	 */
	void (*kill_sb) (struct super_block *);
	/**
	 * 指向实现文件系统的模块的指针。
	 */
	struct module *owner;
	/**
	 * 下一个文件系统指针。
	 */
	struct file_system_type * next;
	/**
	 * 具有相同文件系统类型的超级块对象链表的头。
	 */
	struct list_head fs_supers;
};

基本数据结构关系

2. eventpoll_init()

epoll开始的准备工作由eventpoll_init完成，

static int __init eventpoll_init(void)
{
	int error;

	init_MUTEX(&epsem);

	/* Initialize the structure used to perform safe poll wait head wake ups */
	ep_poll_safewake_init(&psw);

	/* Allocates slab cache used to allocate "struct epitem" items */
	epi_cache = kmem_cache_create("eventpoll_epi", sizeof(struct epitem),
			0, SLAB_HWCACHE_ALIGN|EPI_SLAB_DEBUG|SLAB_PANIC,
			NULL, NULL);

	/* Allocates slab cache used to allocate "struct eppoll_entry" */
	pwq_cache = kmem_cache_create("eventpoll_pwq",
			sizeof(struct eppoll_entry), 0,
			EPI_SLAB_DEBUG|SLAB_PANIC, NULL, NULL);

	/*
	 * Register the virtual file system that will be the source of inodes
	 * for the eventpoll files
	 */
	error = register_filesystem(&eventpoll_fs_type);
	if (error)
		goto epanic;

	/* Mount the above commented virtual file system */
	eventpoll_mnt = kern_mount(&eventpoll_fs_type);
	error = PTR_ERR(eventpoll_mnt);
	if (IS_ERR(eventpoll_mnt))
		goto epanic;

	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: successfully initialized.\n",
			current));
	return 0;

epanic:
	panic("eventpoll_init() failed\n");
}

2.1 kmem_cache_create()

	/* Allocates slab cache used to allocate "struct epitem" items */
	epi_cache = kmem_cache_create("eventpoll_epi", sizeof(struct epitem),
			0, SLAB_HWCACHE_ALIGN|EPI_SLAB_DEBUG|SLAB_PANIC,
			NULL, NULL);

	/* Allocates slab cache used to allocate "struct eppoll_entry" */
	pwq_cache = kmem_cache_create("eventpoll_pwq",
			sizeof(struct eppoll_entry), 0,
			EPI_SLAB_DEBUG|SLAB_PANIC, NULL, NULL);

    该函数是slab分配器接口，即创建一个新的高速缓存——内存池。数据结构类型为struct epitem和struct epoll_entry。

    epoll在被内核初始化时（操作系统启动），同时会开辟出epoll自己的内核告诉cache区，用于安置每一个我们想监控的socket，这些socket会以红黑树的形式保存在内核cache张总，以支持快速的查找、插入、删除。

    这个内核高速缓冲区，就是建立连续的物理内存页，然后在之上建立slab层，简单地说，就是物理上分配好你想要的size大小的内存对象，每次使用时都是使用空闲的已分配好的对象。

2.2 register_filesystem()

    注册文件系统，将相应的file_system_type加入到链表中。

error = register_filesystem(&eventpoll_fs_type);

    在内核中，一切皆文件。所以，epoll向内核注册了一个文件系统，用于存储上述的被监控socket。

   当调用epoll_create时，就会在这个虚拟的epoll文件系统中创建一个file结点。当然这个file不是普通文件，它只服务于epoll。

3. sys_epoll_create()

asmlinkage long sys_epoll_create(int size)
{
	int error, fd;
	struct inode *inode;
	struct file *file;

	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_create(%d)\n",
		     current, size));

	/* Sanity check on the size parameter */
	error = -EINVAL;
	if (size <= 0)
		goto eexit_1;

	/*
	 * Creates all the items needed to setup an eventpoll file. That is,
	 * a file structure, and inode and a free file descriptor.
	 */
	error = ep_getfd(&fd, &inode, &file);
	if (error)
		goto eexit_1;

	/* Setup the file internal data structure ( "struct eventpoll" ) */
	error = ep_file_init(file);
	if (error)
		goto eexit_2;


	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_create(%d) = %d\n",
		     current, size, fd));

	return fd;

eexit_2:
	sys_close(fd);
eexit_1:
	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_create(%d) = %d\n",
		     current, size, error));
	return error;
}

    epoll极其高效的原因：

    由于在调用epoll_create时，内核除了帮我们在epoll文件系统中创建了个file结点，在内核cache里建了一个红黑树用于存储以后epoll_ctl传来的socket外，还会再建立一个list链表，用于存储准备就绪的事件，当epoll_wait调用时，仅仅观察这个list链表有没有数据即可。有数据就返回，没有就sleep，等到timeout时间到后，即使链表没有数据也返回。

所以，epoll_wait非常高效。

3.1 ep_getfd()

     在第一次调用epoll_create时，是要创建新的inode、新的file、新的fd。

static int ep_getfd(int *efd, struct inode **einode, struct file **efile)
{
	struct qstr this;
	char name[32];
	struct dentry *dentry;
	struct inode *inode;
	struct file *file;
	int error, fd;

	/* Get an ready to use file */
	error = -ENFILE;
	file = get_empty_filp();
	if (!file)
		goto eexit_1;

	/* Allocates an inode from the eventpoll file system */
	inode = ep_eventpoll_inode();
	error = PTR_ERR(inode);
	if (IS_ERR(inode))
		goto eexit_2;

	/* Allocates a free descriptor to plug the file onto */
	error = get_unused_fd();
	if (error < 0)
		goto eexit_3;
	fd = error;

	/*
	 * Link the inode to a directory entry by creating a unique name
	 * using the inode number.
	 */
	error = -ENOMEM;
	sprintf(name, "[%lu]", inode->i_ino);
	this.name = name;
	this.len = strlen(name);
	this.hash = inode->i_ino;
	dentry = d_alloc(eventpoll_mnt->mnt_sb->s_root, &this);
	if (!dentry)
		goto eexit_4;
	dentry->d_op = &eventpollfs_dentry_operations;
	d_add(dentry, inode);
	file->f_vfsmnt = mntget(eventpoll_mnt);
	file->f_dentry = dentry;
	file->f_mapping = inode->i_mapping;

	file->f_pos = 0;
	file->f_flags = O_RDONLY;
	file->f_op = &eventpoll_fops;
	file->f_mode = FMODE_READ;
	file->f_version = 0;
	file->private_data = NULL;

	/* Install the new setup file into the allocated fd. */
	fd_install(fd, file);

	*efd = fd;
	*einode = inode;
	*efile = file;
	return 0;

eexit_4:
	put_unused_fd(fd);
eexit_3:
	iput(inode);
eexit_2:
	put_filp(file);
eexit_1:
	return error;
}

3.2 ep_file_init()

    设置文件内部数据结构，即struct eventpoll。

static int ep_file_init(struct file *file)
{
	struct eventpoll *ep;

	if (!(ep = kmalloc(sizeof(struct eventpoll), GFP_KERNEL)))
		return -ENOMEM;

	memset(ep, 0, sizeof(*ep));
	rwlock_init(&ep->lock);
	init_rwsem(&ep->sem);
	init_waitqueue_head(&ep->wq);
	init_waitqueue_head(&ep->poll_wait);
	INIT_LIST_HEAD(&ep->rdllist);
	ep->rbr = RB_ROOT;

	file->private_data = ep;

	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: ep_file_init() ep=%p\n",
		     current, ep));
	return 0;
}

    file->private_data = ep，这里进行初始化，为了可以在函数sys_epoll_ctl直接获取eventpoll文件中的私有数据。

4. sys_epoll_ctl()

    在函数sys_epoll_ctl中，如果增加socket句柄，则检查在红黑树中是否存在，存在就立即返回；不存在则添加到树干上，然后向内核注册回调函数，用于当中断事件来临时向准备就绪链表中插入数据。

asmlinkage long
sys_epoll_ctl(int epfd, int op, int fd, struct epoll_event __user *event)
{
	int error;
	struct file *file, *tfile;
	struct eventpoll *ep;
	struct epitem *epi;
	struct epoll_event epds;

	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_ctl(%d, %d, %d, %p)\n",
		     current, epfd, op, fd, event));

	error = -EFAULT;
	if (EP_OP_HASH_EVENT(op) &&
	    copy_from_user(&epds, event, sizeof(struct epoll_event)))
		goto eexit_1;

	/* Get the "struct file *" for the eventpoll file */
	error = -EBADF;
	file = fget(epfd);
	if (!file)
		goto eexit_1;

	/* Get the "struct file *" for the target file */
	tfile = fget(fd);
	if (!tfile)
		goto eexit_2;

	/* The target file descriptor must support poll */
	error = -EPERM;
	if (!tfile->f_op || !tfile->f_op->poll)
		goto eexit_3;

	/*
	 * 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 (file == tfile || !IS_FILE_EPOLL(file))
		goto eexit_3;

	/*
	 * At this point it is safe to assume that the "private_data" contains
	 * our own data structure.
	 */
	ep = file->private_data;

	down_write(&ep->sem);

	/* Try to lookup the file inside our hash table */
	epi = ep_find(ep, tfile, fd);

	error = -EINVAL;
	switch (op) {
	case EPOLL_CTL_ADD:
		if (!epi) {
			epds.events |= POLLERR | POLLHUP;

			error = ep_insert(ep, &epds, tfile, fd);
		} 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) {
			epds.events |= POLLERR | POLLHUP;
			error = ep_modify(ep, epi, &epds);
		} else
			error = -ENOENT;
		break;
	}

	/*
	 * The function ep_find() increments the usage count of the structure
	 * so, if this is not NULL, we need to release it.
	 */
	if (epi)
		ep_release_epitem(epi);

	up_write(&ep->sem);

eexit_3:
	fput(tfile);
eexit_2:
	fput(file);
eexit_1:
	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_ctl(%d, %d, %d, %p) = %d\n",
		     current, epfd, op, fd, event, error));

	return error;
}

4.1 ep = file->private_data;

    获取eventpoll文件中的私有数据，该数据在event_create中创建。

4.2 ep_find()

     在eventpoll中存储文件描述符信息的红黑树中查找指定fd对应的epitem实例。

    一个新创建的epoll文件带有一个struct eventpoll结构，这个结构再挂一个红黑树，而这个红黑树就是每次epoll_ctl时fd存放的地方。

    ep_find的实现，是struct eventpoll的rbr成员（strut rb_root），原来就是一个红黑树的根。而红黑树上挂的是struct epitem。

4.3 ep_insert()

    首先，进行ep_find，

如果找到了struct epitem而用户操作是ADD，那么返回-EEXIST；

如果是DEL，则ep_remove;

如果找不到struct epitem而用户操作是ADD，就ep_insert创建并插入一个。

static int ep_insert(struct eventpoll *ep, struct epoll_event *event,
		     struct file *tfile, int fd)
{
	int error, revents, pwake = 0;
	unsigned long flags;
	struct epitem *epi;
	struct ep_pqueue epq;

	error = -ENOMEM;
	if (!(epi = EPI_MEM_ALLOC()))
		goto eexit_1;

	/* Item initialization follow here ... */
	EP_RB_INITNODE(&epi->rbn);
	INIT_LIST_HEAD(&epi->rdllink);
	INIT_LIST_HEAD(&epi->fllink);
	INIT_LIST_HEAD(&epi->txlink);
	INIT_LIST_HEAD(&epi->pwqlist);
	epi->ep = ep;
	EP_SET_FFD(&epi->ffd, tfile, fd);
	epi->event = *event;
	atomic_set(&epi->usecnt, 1);
	epi->nwait = 0;

	/* Initialize the poll table using the queue callback */
	epq.epi = epi;
	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.
	 */
	revents = tfile->f_op->poll(tfile, &epq.pt);/////
	........
		
}

4.3.1 EPI_MEM_ALLOC()

    首先，申请一个epi空间。

2.3.2 进行初始化

EP_RB_INITNODE(&epi->rbn);
	INIT_LIST_HEAD(&epi->rdllink);
	INIT_LIST_HEAD(&epi->fllink);
	INIT_LIST_HEAD(&epi->txlink);
	INIT_LIST_HEAD(&epi->pwqlist);
	epi->ep = ep;

2.3.3 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;

	if (epi->nwait >= 0 && (pwq = PWQ_MEM_ALLOC())) {
		init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
		pwq->whead = whead;
		pwq->base = epi;
		add_wait_queue(whead, &pwq->wait);
		list_add_tail(&pwq->llink, &epi->pwqlist);
		epi->nwait++;
	} else {
		/* We have to signal that an error occurred */
		epi->nwait = -1;
	}
}

  函数init_waitqueue_func_entry()中定义等待队列上的唤醒函数为ep_poll_callback，并对等待队列进行初始化。

ep_poll_callback()

    把红黑树上收到event的epitem（代表每个fd）插入ep->rdlist中，

这样，当epoll_wait返回时，rdlist里就都是就绪的fd了。

static int ep_poll_callback(wait_queue_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;

	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: poll_callback(%p) epi=%p ep=%p\n",
		     current, epi->file, epi, ep));

	write_lock_irqsave(&ep->lock, flags);

	/*
	 * 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.
	 */
	if (!(epi->event.events & ~EP_PRIVATE_BITS))
		goto is_disabled;

	/* If this file is already in the ready list we exit soon */
	if (EP_IS_LINKED(&epi->rdllink))
		goto is_linked;

	list_add_tail(&epi->rdllink, &ep->rdllist);

is_linked:
	/*
	 * Wake up ( if active ) both the eventpoll wait list and the ->poll()
	 * wait list.
	 */
	if (waitqueue_active(&ep->wq))
		wake_up(&ep->wq);
	if (waitqueue_active(&ep->poll_wait))
		pwake++;

is_disabled:
	write_unlock_irqrestore(&ep->lock, flags);

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

	return 1;
}

EP_PRIVATE_BITS，即宏替换为（EPOLLONESHOT | EPOLLET）.

list_add_tail(&epi->rdlink, &ep->rdlist); 

    epi->rdlink插入到ep->rdlist之前； struct epitem放到放到struct eventpoll的rdlist中去。

4. sys_epoll_wait()

asmlinkage long sys_epoll_wait(int epfd, struct epoll_event __user *events,
			       int maxevents, int timeout)
{
	int error;
	struct file *file;
	struct eventpoll *ep;

	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_wait(%d, %p, %d, %d)\n",
		     current, epfd, events, maxevents, timeout));

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

	/* Verify that the area passed by the user is writeable */
	if ((error = verify_area(VERIFY_WRITE, events, maxevents * sizeof(struct epoll_event))))
		goto eexit_1;

	/* Get the "struct file *" for the eventpoll file */
	error = -EBADF;
	file = fget(epfd);
	if (!file)
		goto eexit_1;

	/*
	 * 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(file))
		goto eexit_2;

	/*
	 * At this point it is safe to assume that the "private_data" contains
	 * our own data structure.
	 */
	ep = file->private_data;

	/* Time to fish for events ... */
	error = ep_poll(ep, events, maxevents, timeout);

eexit_2:
	fput(file);
eexit_1:
	DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_wait(%d, %p, %d, %d) = %d\n",
		     current, epfd, events, maxevents, timeout, error));

	return error;
}

4.1 maxevents

    事件个数一定大于0，否则返回-EINVAL。

4.2 verify_area()

/**
 * 函数verify_area执行与access_ok宏类似的检查，虽然它被认为是陈旧过时的
 * 但是在源代码中仍然被广泛使用。
 */
static inline int verify_area(int type, const void __user * addr, unsigned long size)
{
	return access_ok(type,addr,size) ? 0 : -EFAULT;
}

    对系统调用所传递地址的检查是通过access_ok宏实现的。

· 它由两个分别为addr和size的参数。

· 该宏检查addr到addr+size-1之间的地址区间。

4.3 file = fget(epfd)

    获取epfd对应的file实例。

然后接着调用IS_FILE_EPOLL(file)，判断是否为eventpoll的file，

即(f)->f_op == &eventpoll_fops.

4.4 ep_poll

这个函数是epoll的核心函数，接下来进行分析。

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	int res, eavail;
	unsigned long flags;
	long jtimeout;
	wait_queue_t wait;

	/*
	 * Calculate the timeout by checking for the "infinite" value ( -1 )
	 * and the overflow condition. The passed timeout is in milliseconds,
	 * that why (t * HZ) / 1000.
	 */
	jtimeout = timeout == -1 || timeout > (MAX_SCHEDULE_TIMEOUT - 1000) / HZ ?
		MAX_SCHEDULE_TIMEOUT: (timeout * HZ + 999) / 1000;

retry:
	write_lock_irqsave(&ep->lock, flags);

	res = 0;
	if (list_empty(&ep->rdllist)) {
		/*
		 * 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.
		 */
		init_waitqueue_entry(&wait, current);
		add_wait_queue(&ep->wq, &wait);

		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);
			if (!list_empty(&ep->rdllist) || !jtimeout)
				break;
			if (signal_pending(current)) {
				res = -EINTR;
				break;
			}

			write_unlock_irqrestore(&ep->lock, flags);
			jtimeout = schedule_timeout(jtimeout);
			write_lock_irqsave(&ep->lock, flags);
		}
		remove_wait_queue(&ep->wq, &wait);

		set_current_state(TASK_RUNNING);
	}

	/* Is it worth to try to dig for events ? */
	eavail = !list_empty(&ep->rdllist);

	write_unlock_irqrestore(&ep->lock, flags);

	/*
	 * 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_events_transfer(ep, events, maxevents)) && jtimeout)
		goto retry;

	return res;
}

首先，调用list_empty(&ep->rdlist)，判断ep->rdlist是否为NULL。

eventpoll下的struct list_head rdlist，双链表中存放着将要通过epoll_wait返回给用户的满足条件的事件。

而struct rb_root rbr，是红黑树的根结点，树中存储所有添加到epoll中的需要监控的事件。

    如果没有事件到来，不会返回给调用方；

    一直在这里睡眠，直到事件发生，被ep_poll_callback()唤醒。

init_waitqueue_entry()

    初始化wait_queue_t结构的变量。

add_wait_queue()

    将wait进程插入等待队列链表的第一个位置。

4.4.1 set_current_state()

    for循环中，设置TASK_INTERRUPTIBLE状态，其原因是：如果ep_poll_callback()发生唤醒，不会去休眠。

4.4.2 signal_pending()

     如果进程描述符所表示的进程有非阻塞的挂起信号，就返回1。否则返回0。
    该函数只是通过检查进程的TIF_SIGPENDING标志。

static inline int signal_pending(struct task_struct *p)
{
	return unlikely(test_tsk_thread_flag(p,TIF_SIGPENDING));
}

4.4.3 remove_wait_queue()

    将wait进程从等待队列链表中删除。

4.4.4 ep_event_transfer()

    把rdlist中的fd拷贝到用户空间。

static int ep_events_transfer(struct eventpoll *ep,
			      struct epoll_event __user *events, int maxevents)
{
	int eventcnt = 0;
	struct list_head txlist;

	INIT_LIST_HEAD(&txlist);

	/*
	 * We need to lock this because we could be hit by
	 * eventpoll_release_file() and epoll_ctl(EPOLL_CTL_DEL).
	 */
	down_read(&ep->sem);

	/* Collect/extract ready items */
	if (ep_collect_ready_items(ep, &txlist, maxevents) > 0) {
		/* Build result set in userspace */
		eventcnt = ep_send_events(ep, &txlist, events);

		/* Reinject ready items into the ready list */
		ep_reinject_items(ep, &txlist);
	}

	up_read(&ep->sem);

	return eventcnt;
}

4.4.4.1 ep_collect_ready_items()

    把rdlist里的fd挪到txlist中（挪完后rdlist就空了）。

static int ep_collect_ready_items(struct eventpoll *ep, struct list_head *txlist, int maxevents)
{
	int nepi;
	unsigned long flags;
	struct list_head *lsthead = &ep->rdllist, *lnk;
	struct epitem *epi;

	write_lock_irqsave(&ep->lock, flags);

	for (nepi = 0, lnk = lsthead->next; lnk != lsthead && nepi < maxevents;) {
		epi = list_entry(lnk, struct epitem, rdllink);

		lnk = lnk->next;

		/* If this file is already in the ready list we exit soon */
		if (!EP_IS_LINKED(&epi->txlink)) {
			/*
			 * This is initialized in this way so that the default
			 * behaviour of the reinjecting code will be to push back
			 * the item inside the ready list.
			 */
			epi->revents = epi->event.events;

			/* Link the ready item into the transfer list */
			list_add(&epi->txlink, txlist);
			nepi++;

			/*
			 * Unlink the item from the ready list.
			 */
			EP_LIST_DEL(&epi->rdllink);
		}
	}

	write_unlock_irqrestore(&ep->lock, flags);

	return nepi;
}

4.4.4.2 ep_send_events()

    把txlist中的fd拷贝到用户空间。

static int ep_send_events(struct eventpoll *ep, struct list_head *txlist,
			  struct epoll_event __user *events)
{
	int eventcnt = 0;
	unsigned int revents;
	struct list_head *lnk;
	struct epitem *epi;

	/*
	 * We can loop without lock because this is a task private list.
	 * The test done during the collection loop will guarantee us that
	 * another task will not try to collect this file. Also, items
	 * cannot vanish during the loop because we are holding "sem".
	 */
	list_for_each(lnk, txlist) {
		epi = list_entry(lnk, struct epitem, txlink);

		/*
		 * Get the ready file event set. We can safely use the file
		 * because we are holding the "sem" in read and this will
		 * guarantee that both the file and the item will not vanish.
		 */
		revents = epi->ffd.file->f_op->poll(epi->ffd.file, NULL);

		/*
		 * Set the return event set for the current file descriptor.
		 * Note that only the task task was successfully able to link
		 * the item to its "txlist" will write this field.
		 */
		epi->revents = revents & epi->event.events;

		if (epi->revents) {
			if (__put_user(epi->revents,
				       &events[eventcnt].events) ||
			    __put_user(epi->event.data,
				       &events[eventcnt].data))
				return -EFAULT;
			if (epi->event.events & EPOLLONESHOT)
				epi->event.events &= EP_PRIVATE_BITS;
			eventcnt++;
		}
	}
	return eventcnt;
}

在ep_send_events（）中，

    revents = epi->ffd.file->f_op->poll(epi->ffd.file, NULL);

调用函数scull_p_poll，也就是其中的poll_wait（）函数，

POLL方法是poll、epoll和select这三个系统调用的后端实现。可用来查询某个或多个文件描述符上的读取或写入是否会被阻塞。

poll方式返回一个位掩码mask,用来指出非阻塞的读取或写入是否可能。并且会向内核提供将调用进程置于休眠状态直到IO变为可能时的信息，并且驱动程序中将POLL方法定义为NULL，则设备会被认为既可读也可写，并且不会阻塞。

设备先要把current(当前进程)挂在inq和outq两个队列上（这个“挂”操作是wait回调函数指针做的），然后等设备唤醒，唤醒后就能通过mask拿到事件掩码了。

这里的mask参数就是负责事件掩码。

4.4.4.3 ep_reinject_items()

    把一部分fd从txlist里“返还”给rdlist以便下次还能从rdlist里发现它。

static void ep_reinject_items(struct eventpoll *ep, struct list_head *txlist)
{
	int ricnt = 0, pwake = 0;
	unsigned long flags;
	struct epitem *epi;

	write_lock_irqsave(&ep->lock, flags);

	while (!list_empty(txlist)) {
		epi = list_entry(txlist->next, struct epitem, txlink);

		/* Unlink the current item from the transfer list */
		EP_LIST_DEL(&epi->txlink);

		/*
		 * If the item is no more linked to the interest set, we don't
		 * have to push it inside the ready list because the following
		 * ep_release_epitem() is going to drop it. Also, if the current
		 * item is set to have an Edge Triggered behaviour, we don't have
		 * to push it back either.
		 */
		if (EP_RB_LINKED(&epi->rbn) && !(epi->event.events & EPOLLET) &&
		    (epi->revents & epi->event.events) && !EP_IS_LINKED(&epi->rdllink)) {
			list_add_tail(&epi->rdllink, &ep->rdllist);
			ricnt++;
		}
	}

	if (ricnt) {
		/*
		 * Wake up ( if active ) both the eventpoll wait list and the ->poll()
		 * wait list.
		 */
		if (waitqueue_active(&ep->wq))
			wake_up(&ep->wq);
		if (waitqueue_active(&ep->poll_wait))
			pwake++;
	}

	write_unlock_irqrestore(&ep->lock, flags);

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

函数中进行判断时，

    EP_RB_LINKED(epi->rbn) && !(epi->event.events & EPOLLET) && (epi->revents & epi->event.events) && IEP_IS_LINKED(&epi->rdlink),

    是哪些“没有标上EPOLLET”（标红代码）且“事件被关注”（标蓝代码）的fd重新被放回了rdlist。

LT模式下，只要一个句柄上事件一次没有处理完，会在以后调用epoll_wait时此次返回这个句柄，从txlist拷贝到用户空间后，会返还给rdlist。

而ET模式下，仅在第一次返回。

--------------------------------------------------------------------

总结

  1. select和poll每次调用这些函数的时候都需要将监控的fd和需要监控的事件从用户空间拷贝到内核空间，非常影响效率。而epoll就是自己保存用户空间拷入的fd和需要监控的事件，只需在调用epoll_ctl的时候就把所有的fd和需要监控的事件只进行一次从用户空间到内核空间的拷贝。

  2. poll和select类似，每次调用都返回整个用户注册的事件集合（包括就绪的和未就绪的），应用程序索引就绪文件描述符的时间复杂度为O（n)。而epoll是在内核中维护一个事件表，epoll_wait的events参数返回就绪的事件，时间复杂度为O（1）.

 3. poll和epoll_wait分别用nfds和maxevents参数指定最多监听多少个文件描述符和事件个数，即65535（cat/proc/sys/fs/file-max）。而select允许监听的最大文件描述符个数为1024.

并发支持完美，不会随着socket的增加而降低效率，也不用在内核空间和用户空间之间做无效的copy操作。

 4. poll只能工作在相对低效的LT模式（电平触发），而epoll支持LT和ET模式。

 ET 边沿触发：只触发一次，无论缓冲区中是否还有剩余数据，直到有新的数据到达才会被触发，再去读取缓冲区里面的数据。

 LT 水平触发（默认）： LT(level triggered)是缺省的工作方式，并且同时支持block和no-block socket，每次缓冲区都有数据都要触发。

    epoll可以监控管道文件，任意文件，不仅仅是socket文件.

 5. poll采用轮询方式，即每次调用都要扫描整个注册文件描述符集合，并将其中就绪的文件描述符返回个用户，因此检测就绪事件的时间复杂度是O（n）。epoll则采用回调方式。内核检测到就绪的文件描述符，将触发回调函数，回调函数将该文件描述符上对应的事件插入内核就绪事件队列。内核最后将该就绪事件队列的内容拷贝到用户空间。时间复杂度为O（1）.

 6. 能处理EPOLLONESHOT事件

----------------------------------

应用场景

1. epoll_wait适用于I/O密集型，即连接数量多，但活动连接较少的情况。因为epoll则采用回调方式。内核检测到就绪的文件描述符，将触发回调函数，回调函数将该文件描述符上对应的事件插入内核就绪事件队列。内核最后将该就绪事件队列的内容拷贝到用户空间。

    但是，当活动连接较多时，epoll_wait的效率未必比select和poll高，因为此时回调函数被触发的过于频繁。

 2. 并发支持完美，不会随着socket的增加而降低效率，也不用在内核空间和用户空间之间做无效的copy操作。