欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

epoll源码剖析

程序员文章站 2022-06-14 11:19:18
...

(主要基于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;
};

基本数据结构关系

epoll源码剖析

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。

epoll源码剖析

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()函数,

epoll源码剖析

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操作。













相关标签: epoll linux