解析Linux源碼之epoll

int init_reactor(int listen_fd,int worker_count){
	......
	// 創(chuàng)建多個(gè)epoll fd，以充分利用多核
	for(i=0;i<worker_count;i++){
		reactor->worker_fd = epoll_create(EPOLL_MAX_EVENTS);
	}
	/* epoll add listen_fd and accept */
	// 將accept后的事件加入到對應(yīng)的epoll fd中
	int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len)));
	// 將連接描述符注冊到對應(yīng)的worker里面
	epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event);
}
// reactor的worker線程
static void* rw_thread_func(void* arg){
	......

	for(;;){
		  // epoll_wait等待事件觸發(fā)
        int retval = epoll_wait(epfd,events,EPOLL_MAX_EVENTS,500);
        if(retval > 0){
        	for(j=0; j < retval; j++){
        		// 處理讀事件
        	   if(event & EPOLLIN){
                 handle_ready_read_connection(conn);
                 continue;
             }
             /* 處理其它事件 */
        	}
        }
	}
	......
}

SYSCALL_DEFINE1(epoll_create, int, size)
{
	if (size <= 0)
		return -EINVAL;

	return sys_epoll_create1(0);
}

SYSCALL_DEFINE1(epoll_create1, int, flags)
{
	// kzalloc(sizeof(*ep), GFP_KERNEL),用的是內(nèi)核空間
	error = ep_alloc(&ep);
	// 獲取尚未被使用的文件描述符，即描述符數(shù)組的槽位
	fd = get_unused_fd_flags(O_RDWR | (flags & O_CLOEXEC));
	// 在匿名inode文件系統(tǒng)中分配一個(gè)inode,并得到其file結(jié)構(gòu)體
	// 且file->f_op = &eventpoll_fops
	// 且file->private_data = ep;
	file = anon_inode_getfile("[eventpoll]", &eventpoll_fops, ep,
				 O_RDWR | (flags & O_CLOEXEC));
	// 將file填入到對應(yīng)的文件描述符數(shù)組的槽里面
	fd_install(fd,file);			 
	ep->file = file;
	return fd;
}

/*
 * 此結(jié)構(gòu)體存儲在file->private_data中
 */
struct eventpoll {
	// 自旋鎖，在kernel內(nèi)部用自旋鎖加鎖，就可以同時(shí)多線(進(jìn))程對此結(jié)構(gòu)體進(jìn)行操作
	// 主要是保護(hù)ready_list
	spinlock_t lock;
	// 這個(gè)互斥鎖是為了保證在eventloop使用對應(yīng)的文件描述符的時(shí)候，文件描述符不會被移除掉
	struct mutex mtx;
	// epoll_wait使用的等待隊(duì)列，和進(jìn)程喚醒有關(guān)
	wait_queue_head_t wq;
	// file->poll使用的等待隊(duì)列，和進(jìn)程喚醒有關(guān)
	wait_queue_head_t poll_wait;
	// 就緒的描述符隊(duì)列
	struct list_head rdllist;
	// 通過紅黑樹來組織當(dāng)前epoll關(guān)注的文件描述符
	struct rb_root rbr;
	// 在向用戶空間傳輸就緒事件的時(shí)候，將同時(shí)發(fā)生事件的文件描述符鏈入到這個(gè)鏈表里面
	struct epitem *ovflist;
	// 對應(yīng)的user
	struct user_struct *user;
	// 對應(yīng)的文件描述符
	struct file *file;
	// 下面兩個(gè)是用于環(huán)路檢測的優(yōu)化
	int visited;
	struct list_head visited_list_link;
};

SYSCALL_DEFINE4(epoll_ctl, int, epfd, int, op, int, fd,
		struct epoll_event __user *, event)
{
	/* 校驗(yàn)epfd是否是epoll的描述符 */
	// 此處的互斥鎖是為了防止并發(fā)調(diào)用epoll_ctl,即保護(hù)內(nèi)部數(shù)據(jù)結(jié)構(gòu)
	// 不會被并發(fā)的添加修改刪除破壞
	mutex_lock_nested(&ep->mtx, 0);
	switch (op) {
		case EPOLL_CTL_ADD:
			...
			// 插入到紅黑樹中
			error = ep_insert(ep, &epds, tfile, fd);
			...
			break;
		......
	}
	mutex_unlock(&ep->mtx);	
}		

static int ep_insert(struct eventpoll *ep, struct epoll_event *event,
		     struct file *tfile, int fd)
{
	/* 初始化epitem */
	// &epq.pt->qproc = ep_ptable_queue_proc
	init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);
	// 在這里將回調(diào)函數(shù)注入
	revents = tfile->f_op->poll(tfile, &epq.pt);
	// 如果當(dāng)前有事件已經(jīng)就緒，那么一開始就會被加入到ready list
	// 例如可寫事件
	// 另外，在tcp內(nèi)部ack之后調(diào)用tcp_check_space,最終調(diào)用sock_def_write_space來喚醒對應(yīng)的epoll_wait下的進(jìn)程
	if ((revents & event->events) && !ep_is_linked(&epi->rdllink)) {
		list_add_tail(&epi->rdllink, &ep->rdllist);
		// wake_up ep對應(yīng)在epoll_wait下的進(jìn)程
		if (waitqueue_active(&ep->wq)){
			wake_up_locked(&ep->wq);
		}
		......
	}	
	// 將epitem插入紅黑樹
	ep_rbtree_insert(ep, epi);
	......
}

// 將accept后的事件加入到對應(yīng)的epoll fd中
int client_fd = accept(listen_fd,(struct sockaddr *)&client_addr,&client_len)));
// 將連接描述符注冊到對應(yīng)的worker里面
epoll_ctl(reactor->client_fd,EPOLL_CTL_ADD,epifd,&event);

static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
				 poll_table *pt)
{
	// 取出當(dāng)前client_fd對應(yīng)的epitem
	struct epitem *epi = ep_item_from_epqueue(pt);
	// &pwq->wait->func=ep_poll_callback，用于回調(diào)喚醒
	// 注意，這邊不是init_waitqueue_entry,即沒有將當(dāng)前KSE(current,當(dāng)前進(jìn)程/線程)寫入到
	// wait_queue當(dāng)中，因?yàn)椴灰欢ㄊ菑漠?dāng)前安裝的KSE喚醒，而應(yīng)該是喚醒epoll\_wait的KSE
	init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
	// 這邊的whead是sk->sk_sleep,將當(dāng)前的waitqueue鏈入到socket對應(yīng)的sleep列表
	add_wait_queue(whead, &pwq->wait);	
}	

SYSCALL_DEFINE4(epoll_wait, int, epfd, struct epoll_event __user *, events,
		int, maxevents, int, timeout)
{
	/* 檢查epfd是否是epoll\_create創(chuàng)建的fd */
	// 調(diào)用ep_poll
	error = ep_poll(ep, events, maxevents, timeout);
	...
}

static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
		   int maxevents, long timeout)
{
	......
retry:
	// 獲取spinlock
	spin_lock_irqsave(&ep->lock, flags);
	// 將當(dāng)前task_struct寫入到waitqueue中以便喚醒
	// wq_entry->func = default_wake_function;
	init_waitqueue_entry(&wait, current);
	// WQ_FLAG_EXCLUSIVE，排他性喚醒，配合SO_REUSEPORT從而解決accept驚群問題
	wait.flags |= WQ_FLAG_EXCLUSIVE;
	// 鏈入到ep的waitqueue中
	__add_wait_queue(&ep->wq, &wait);
	for (;;) {
		// 設(shè)置當(dāng)前進(jìn)程狀態(tài)為可打斷
		set_current_state(TASK_INTERRUPTIBLE);
		// 檢查當(dāng)前線程是否有信號要處理，有則返回-EINTR
		if (signal_pending(current)) {
			res = -EINTR;
			break;
		}
		spin_unlock_irqrestore(&ep->lock, flags);
		// schedule調(diào)度，讓出CPU
		jtimeout = schedule_timeout(jtimeout);
		spin_lock_irqsave(&ep->lock, flags);
	}
	// 到這里，表明超時(shí)或者有事件觸發(fā)等動作導(dǎo)致進(jìn)程重新調(diào)度
	__remove_wait_queue(&ep->wq, &wait);
	// 設(shè)置進(jìn)程狀態(tài)為running
	set_current_state(TASK_RUNNING);
	......
	// 檢查是否有可用事件
	eavail = !list_empty(&ep->rdllist) || ep->ovflist != EP_UNACTIVE_PTR;
	......
	// 向用戶空間拷貝就緒事件
	ep_send_events(ep, events, maxevents)
}		   

static int ep_scan_ready_list(struct eventpoll *ep,
			      int (*sproc)(struct eventpoll *,
					   struct list_head *, void *),
			      void *priv,
			      int depth)
{
	...
	// 將epfd的rdllist鏈入到txlist
	list_splice_init(&ep->rdllist, &txlist);
	...
	/* sproc = ep_send_events_proc */
	error = (*sproc)(ep, &txlist, priv);
	...
	// 處理ovflist,即在上面sproc過程中又到來的事件
	...
}

static int ep_send_events_proc(struct eventpoll *ep, struct list_head *head,
			       void *priv)
{
	for (eventcnt = 0, uevent = esed->events;
	     !list_empty(head) && eventcnt < esed->maxevents;) {
	   // 遍歷ready list 
		epi = list_first_entry(head, struct epitem, rdllink);
		list_del_init(&epi->rdllink);
		// readylist只是表明當(dāng)前epi有事件，具體的事件信息還是得調(diào)用對應(yīng)file的poll
		// 這邊的poll即是tcp_poll,根據(jù)tcp本身的信息設(shè)置掩碼(mask)等信息 & 上興趣事件掩碼，則可以得知當(dāng)前事件是否是epoll_wait感興趣的事件
		revents = epi->ffd.file->f_op->poll(epi->ffd.file, NULL) &
			epi->event.events;
		if(revents){
			/* 將event放入到用戶空間 */
			/* 處理ONESHOT邏輯 */
			// 如果不是邊緣觸發(fā)，則將當(dāng)前的epi重新加回到可用列表中，這樣就可以下一次繼續(xù)觸發(fā)poll,如果下一次poll的revents不為0，那么用戶空間依舊能感知 */
			else if (!(epi->event.events & EPOLLET)){
				list_add_tail(&epi->rdllink, &ep->rdllist);
			}
			/* 如果是邊緣觸發(fā)，那么就不加回可用列表，因此只能等到下一個(gè)可用事件觸發(fā)的時(shí)候才會將對應(yīng)的epi放到可用列表里面*/
			eventcnt++
		}
		/* 如poll出來的revents事件epoll_wait不感興趣(或者本來就沒有事件)，那么也不會加回到可用列表 */
		......
	}
	return eventcnt;
}		

static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
{
	// 獲取wait對應(yīng)的epitem	
	struct epitem *epi = ep_item_from_wait(wait);
	// epitem對應(yīng)的eventpoll結(jié)構(gòu)體
	struct eventpoll *ep = epi->ep;
	// 獲取自旋鎖，保護(hù)ready_list等結(jié)構(gòu)
	spin_lock_irqsave(&ep->lock, flags);
	// 如果當(dāng)前epi沒有被鏈入ep的ready list,則鏈入
	// 這樣，就把當(dāng)前的可用事件加入到epoll的可用列表了
	if (!ep_is_linked(&epi->rdllink))
		list_add_tail(&epi->rdllink, &ep->rdllist);
	// 如果有epoll_wait在等待的話，則喚醒這個(gè)epoll_wait進(jìn)程
	// 對應(yīng)的&ep->wq是在epoll_wait調(diào)用的時(shí)候通過init_waitqueue_entry(&wait, current)而生成的
	// 其中的current即是對應(yīng)調(diào)用epoll_wait的進(jìn)程信息task_struct
	if (waitqueue_active(&ep->wq))
		wake_up_locked(&ep->wq);
}

void sk_stream_write_space(struct sock *sk)
{
	// 即有1/3可寫空間的時(shí)候才觸發(fā)可寫事件
	if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk) && sock) {
		clear_bit(SOCK_NOSPACE, &sock->flags);

		if (sk->sk_sleep && waitqueue_active(sk->sk_sleep))
			wake_up_interruptible_poll(sk->sk_sleep, POLLOUT |
						POLLWRNORM | POLLWRBAND)
		......
	}
}