深入分析select&poll&epoll原理

// 其中cmd = schedule(), 即一个调用schedule函数的指针
#define ___wait_event(wq_head, condition, state, exclusive, ret, cmd)
({										
	__label__ __out;							
	struct wait_queue_entry __wq_entry;					
	long __ret = ret;	/* explicit shadow */				
	// 初始化过程(内部代码这里省略,直接说明)
	// 1. 设置独占标志到当前节点entry
	// 2. 将当前任务task指向节点的private
	// 3. 同时为当前entry节点传递一个唤醒的回调函数autoremove_wake_function,一旦唤醒将会自动被删除
	init_wait_entry(&__wq_entry, exclusive ? WQ_FLAG_EXCLUSIVE : 0);	
	for (;;) {
		// 防止队列中没有entry产生不断的轮询,主要处理wait_queue与entry节点添加或者删除
		long __int = prepare_to_wait_event(&wq_head, &__wq_entry, state);
		// 事件轮询检查是否事件有被唤醒							
		if (condition)							
		      break;							
										
		if (___wait_is_interruptible(state) && __int) {			
			__ret = __int;						
			goto __out;						
		}								
		
		// 调用schedule()方法						
		cmd;								
	}
	// 事件被唤醒,将当前的entry从队列中移除				
	finish_wait(&wq_head, &__wq_entry);					
__out:	__ret;									
})

static int __wake_up_common(struct wait_queue_head *wq_head, unsigned int mode,
			int nr_exclusive, int wake_flags, void *key,
			wait_queue_entry_t *bookmark)
{
	// 省略其他非核心代码...
	// 循环遍历整个等待队列
	list_for_each_entry_safe_from(curr, next, &wq_head->head, entry) {
		unsigned flags = curr->flags;
		int ret;

		if (flags & WQ_FLAG_BOOKMARK)
			continue;

		//执行回调函数
		ret = curr->func(curr, mode, wake_flags, key);
		if (ret < 0)
			break;
		
		// 检查当前节点是否为独占节点,即互斥锁,只能执行一个task,因此需要退出循环
		if (ret && (flags & WQ_FLAG_EXCLUSIVE) && !--nr_exclusive)
			break;

		if (bookmark && (++cnt > WAITQUEUE_WALK_BREAK_CNT) &&
				(&next->entry != &wq_head->head)) {
			bookmark->flags = WQ_FLAG_BOOKMARK;
			list_add_tail(&bookmark->entry, &next->entry);
			break;
		}
	}
	return nr_exclusive;
}

struct wait_queue_entry {
	unsigned int		flags;
	void			*private;
	// 这里的func就是上述休眠的时候在init_wait_entry传递autoremove_wake_function
	wait_queue_func_t	func;	
	struct list_head	entry;
};

// 唤醒函数
int autoremove_wake_function(struct wait_queue_entry *wq_entry, unsigned mode, int sync, void *key)
{
	// 公用的唤醒函数逻辑
	// 内部执行try_to_wake_up, 也就是将wq_entry的private(当前进程)添加到cpu的执行队列中,让cpu能够调度task执行
	int ret = default_wake_function(wq_entry, mode, sync, key);
	
	// 其他为当前唤醒函数私有逻辑
	if (ret)
		list_del_init(&wq_entry->entry);

	return ret;
}
EXPORT_SYMBOL(autoremove_wake_function);

for(;;){
    res = 0;    for(i=0; i<maxfds,i++){         // 检测当前fd是否就绪        if(fd[i].poll()){
           // 更新事件状态,让用户进程知道当前socket状态是可读状态
            fd_sock.event |= POLLIN;
        }
    }
        if(res | tiemout | expr){        break;    }
    schdule();
}

foreach(entry as waiter_queues){
        // 唤醒通知并将任务task加入cpu就绪队列中
    res = callback();
    
    // 说明当前节点为独占节点,只能唤醒一次,因此需要退出循环
    if(res && current == EXCLUSIVE){
        break;
    }
    
}

int select(int maxfd1,			// 最大文件描述符个数,传输的时候需要+1
		   fd_set *readset,	// 读描述符集合
		   fd_set *writeset,	// 写描述符集合
		   fd_set *exceptset,	// 异常描述符集合
		   const struct timeval *timeout);// 超时时间

// 现在很多Liunx版本使用pselect函数,最新版本(5.6.2)的select已经弃用
// 其定义如下
int pselect(int maxfd1,		        // 最大文件描述符个数,传输的时候需要+1
		   fd_set *readset,	// 读描述符集合
		   fd_set *writeset,	// 写描述符集合
		   fd_set *exceptset,	// 异常描述符集合
		   const struct timespec *timeout,    // 超时时间
		   const struct sigset_t *sigmask);  // 信号掩码指针		

// 基于POSIX协议
// posix_type.h
#define __FD_SETSIZE	1024  // 最大文件描述符为1024
// 这里只关注socket可读状态,以下主要是休眠逻辑
static int do_select(int n, fd_set_bits *fds, struct timespec64 *end_time)
{
	struct poll_wqueues table;
	poll_table *wait;
	
	// ...
	// 与上述休眠逻辑初始化等待节点操作类似(查看下面的唤醒逻辑)
	poll_initwait(&table);	
	wait = &table.pt;// 获取创建之后的等待节点
	
	rcu_read_lock();
	retval = max_select_fd(n, fds);
	rcu_read_unlock();
	n = retval;
	
	// ...
	// 操作返回值
	retval = 0;
	for (;;) {
		//...
		// 监控可读的描述符socket
		inp = fds->in;
		for (i = 0; i < n; ++rinp, ++routp, ++rexp) {
			bit = 1;
			// BITS_PER_LONG若处理器为32bit则BITS_PER_LONG=32,否则BITS_PER_LONG=64;
			for (j = 0; j < BITS_PER_LONG; ++j, ++i, bit <<= 1) {
				f = fdget(i);
				wait_key_set(wait, in, out, bit,
						     busy_flag);
				// 检测当前等待节点是否可读
				mask = vfs_poll(f.file, wait);
				fdput(f);
				if ((mask & POLLIN_SET) && (in & bit)) {
					res_in |= bit;
					retval++;
					wait->_qproc = NULL;
				}
				// ...
				
			}
		}
		// 说明有存在可读节点退出节点遍历
		if (retval || timed_out || signal_pending(current))
			break;
		// ...
		// 调度带有超时事件的schedule
		if (!poll_schedule_timeout(&table, TASK_INTERRUPTIBLE,
					   to, slack))
			timed_out = 1;
	}
	// 移除队列中的等待节点
	poll_freewait(&table);
}

// 在poll_initwait ->  __pollwait --> pollwake 的方法,主要关注pollwake方法
static int __pollwake(wait_queue_entry_t *wait, unsigned mode, int sync, void *key)
{
	struct poll_wqueues *pwq = wait->private;
	DECLARE_WAITQUEUE(dummy_wait, pwq->polling_task);

	smp_wmb();	// 内存屏障,保证数据可见性
	pwq->triggered = 1;
	
	// 与linux内核中的唤醒机制一样,下面的方法是内核执行的,不过多关心,有兴趣可以看源码core.c下面定义
	// 就是polling_task也就是read_process添加到cpu就绪队列中,让cpu能够进行调度
	return default_wake_function(&dummy_wait, mode, sync, key);
} 

// poll已经被弃用
int poll(struct pollfd *fds, 	        // fd的文件集合改成自定义结构体,不再是数组的方式,不受限于FD_SIZE
		 unsigned long nfds,     // 最大描述符个数
				int timeout);// 超时时间

struct pollfd {
	int fd;			// fd索引值
	short events;		// 输入事件
	short revents;		// 结果输出事件
};

// 当前查看的linux版本(5.6.2)使用ppoll方式,与pselect差不多,其他细节不多关注
int ppoll(struct pollfd *fds, 	                // fd的文件集合改成自定义结构体,不再是数组的方式,不受限于FD_SIZE
		 unsigned long nfds, 	       // 最大描述符个数
		 struct timespec timeout,        // 超时时间,与pselect一样
		 const struct sigset_t sigmask,	 // 信号指针掩码
		 struct size_t sigsetsize);	 // 信号大小

// 关于poll与select实现的机制差不多,因此不过多贴代码,只简单列出核心点即可
static int do_sys_poll(struct pollfd __user *ufds, unsigned int nfds,
		struct timespec64 *end_time)
{
	// ...
	for (;;) {
		// ...
		// 从用户空间将fdset拷贝到内核中
		if (copy_from_user(walk->entries, ufds + nfds-todo,
					sizeof(struct pollfd) * walk->len))
			goto out_fds;
		// ...
		// 和select一样,初始化等待节点的操作
		poll_initwait(&table);
		// do_poll的处理逻辑与do_select逻辑基本一致,只是这里用链表的方式遍历,do_select用数组的方式
		// 链表可以无限增加节点,数组有指定大小,受到FD_SIZE的限制
		fdcount = do_poll(head, &table, end_time);
		// 从等待队列移除等待节点
		poll_freewait(&table);
	}
}

// 创建保存epoll文件描述符的空间,该空间也称为“epoll例程”
int epoll_create(int size);    // 使用链表,现在已经弃用
int epoll_create(int flag);    // 使用红黑树的数据结构

// epoll注册/修改/删除 fd的操作
long epoll_ctl(int epfd,                        // 上述epoll空间的fd索引值
               int op,                         // 操作识别，EPOLL_CTL_ADD |  EPOLL_CTL_MOD  |  EPOLL_CTL_DEL
               int fd,                          // 注册的fd
               struct epoll_event *event);      // epoll监听事件的变化
struct epoll_event {
	__poll_t events;
	__u64 data;
} EPOLL_PACKED;

// epoll等待，与select/poll的逻辑一致
epoll_wait(int epfd,                            // epoll空间
           struct epoll_event *events,           // epoll监听事件的变化
           int maxevents,                        // epoll可以保存的最大事件数
        int timeout);                         // 超时时间

// 摘取核心代码
int do_epoll_ctl(int epfd, int op, int fd, struct epoll_event *epds,
		 bool nonblock)
{
	// ...

	// 在红黑树中查找存储file对应的epitem，添加的时候会将epitem加到红黑树节点中
	epi = ep_find(ep, tf.file, fd);
	
	// 对于EPOLL_CTL_ADD模式,使用mtx加锁添加到wakeup队列中
	switch (op) {
	case EPOLL_CTL_ADD:
	    // fd注册操作
		// epds->events |= EPOLLERR | EPOLLHUP;
		// error = ep_insert(ep, epds, tf.file, fd, full_check);
		break;
	case EPOLL_CTL_DEL:
	    // // 删除操作：存储epitem容器移除epitem信息
		break;
	// 对注册的fd进行修改,但epoll的模式为EPOLLEXCLUSIVE是无法进行操作的
	case EPOLL_CTL_MOD:
	    // 修改操作,内核监听到事件变化执行修改
            //error = ep_modify(ep, epi, epds);			
		break;
	}
	
	// 释放资源逻辑
}

// 添加逻辑
static int ep_insert(struct eventpoll *ep, const struct epoll_event *event,
		     struct file *tfile, int fd, int full_check)
{
    // ...
	struct epitem *epi;
	struct ep_pqueue epq;
	
	// 将fd包装在epitem的epollfile中
	epi->ep = ep;
	ep_set_ffd(&epi->ffd, tfile, fd);
	epi->event = *event;
	epi->nwait = 0;
	epi->next = EP_UNACTIVE_PTR;
	
	// 如果当前监听到事件变化，那么创建wakeup执行的source
	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);
	}

	// 添加回调函数
	epq.epi = epi;
	init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);

       // 轮询检测epitem中的事件
	revents = ep_item_poll(epi, &epq.pt, 1);

       // 将epitem插入到红黑树中	
	ep_rbtree_insert(ep, epi);

	// 如果有ready_list 则执行唤醒逻辑wakeup，这个是linux内核的唤醒机制，会将read_process添加到就绪队列中让cpu调度执行
	if (revents && !ep_is_linked(epi)) {
		list_add_tail(&epi->rdllink, &ep->rdllist);
		ep_pm_stay_awake(epi);

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

    // ....	

	// 存在预唤醒，则唤醒轮询等待节点
	if (pwake)
	    ep_poll_safewake(&ep->poll_wait);

	return 0;

// goto statement code ...
}

// epoll_wait -> do_epoll_wait -> ep_poll, 我们关注核心方法ep_poll
static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	// ...
	fetch_events: // 检测epoll是否有事件就绪
		// ...
		for (;;) {
		// ...
		// 检测当前ep空间是否有fd事件就绪
		eavail = ep_events_available(ep);
		if (eavail)
			// 是的跳出循环
			break;
		if (signal_pending(current)) {
			res = -EINTR;
			break;
		}
		// 执行休眠方法 schedule()
		if (!schedule_hrtimeout_range(to, slack, HRTIMER_MODE_ABS)) {
			timed_out = 1;
			break;
		}
	}
	// ...
	send_events: // ep有事件就绪,将event_poll转换到用户空间中
		//...
		ep_send_events(ep, events, maxevents);
	
}