Linux內核提供了3個關鍵函數供用戶來操作epoll，分別是：

• epoll_create(), 創建eventpoll對象
• epoll_ctl(), 操作eventpoll對象
• epoll_wait(), 從eventpoll對象中返回活躍的事件

而操作系統內部會用到一個名叫epoll_event_callback()的回調函數來調度epoll對象中的事件，這個函數非常重要，故本文將會對上述4個函數進行源碼分析。

源碼來源

由于epoll的實現內嵌在內核中，直接查看內核源碼的話會有一些無關代碼影響閱讀。為此在GitHub上寫的簡化版TCP/IP協議棧，里面實現了epoll邏輯。鏈接為：
https://github.com/wangbojing/NtyTcp

存放著以上4個關鍵函數的文件是[srcnty_epoll_rb.c]，本文接下來通過分析該程序的代碼來探索epoll能支持高并發連接的秘密。

兩個核心數據結構

(1)epitem

如圖所示，epitem是中包含了兩個主要的成員變量，分別是rbn和rdlink，前者是紅黑樹的節點，而后者是雙鏈表的節點，也就是說一個epitem對象即可作為紅黑樹中的一個節點又可作為雙鏈表中的一個節點。并且每個epitem中存放著一個event，對event的查詢也就轉換成了對epitem的查詢。

struct epitem {
	RB_ENTRY(epitem) rbn;
	/*  RB_ENTRY(epitem) rbn等價于
	struct {											
		struct type *rbe_left;		//指向左子樹
		struct type *rbe_right;		//指向右子樹
		struct type *rbe_parent;	//指向父節點
		int rbe_color;			    //該節點的顏色
	} rbn
	*/

	LIST_ENTRY(epitem) rdlink;
	/* LIST_ENTRY(epitem) rdlink等價于
	struct {									
		struct type *le_next;	//指向下個元素
		struct type **le_prev;	//前一個元素的地址
	}*/

	int rdy; //判斷該節點是否同時存在與紅黑樹和雙向鏈表中
	
	int sockfd; //socket句柄
	struct epoll_event event;  //存放用戶填充的事件
};

(2)eventpoll

如圖所示，eventpoll中包含了兩個主要的成員變量，分別是rbr和rdlist，前者指向紅黑樹的根節點，后者指向雙鏈表的頭結點。即一個eventpoll對象對應二個epitem的容器。對epitem的檢索，將發生在這兩個容器上（紅黑樹和雙鏈表）。

struct eventpoll {
	/*
	struct ep_rb_tree {
		struct epitem *rbh_root; 			
	}
	*/
	ep_rb_tree rbr;      //rbr指向紅黑樹的根節點
	
	int rbcnt; //紅黑樹中節點的數量（也就是添加了多少個TCP連接事件）
	
	LIST_HEAD( ,epitem) rdlist;    //rdlist指向雙向鏈表的頭節點；
	/*	這個LIST_HEAD等價于 
		struct {
			struct epitem *lh_first;
		}rdlist;
	*/
	
	int rdnum; //雙向鏈表中節點的數量（也就是有多少個TCP連接來事件了）

	// ...略...
	
};

四個關鍵函數

(1) epoll_create()

//創建epoll對象，包含一顆空紅黑樹和一個空雙向鏈表
int epoll_create(int size) {
	//與很多內核版本一樣，size參數沒有作用，只要保證大于0即可
	if (size <= 0) return -1;
	
	nty_tcp_manager *tcp = nty_get_tcp_manager(); //獲取tcp對象
	if (!tcp) return -1;
	
	struct _nty_socket *epsocket = nty_socket_allocate(NTY_TCP_SOCK_EPOLL);
	if (epsocket == NULL) {
		nty_trace_epoll("malloc failedn");
		return -1;
	}

	// 1° 開辟了一塊內存用于填充eventpoll對象
	struct eventpoll *ep = (struct eventpoll*)calloc(1, sizeof(struct eventpoll));
	if (!ep) {
		nty_free_socket(epsocket- >id, 0);
		return -1;
	}

	ep- >rbcnt = 0;

	// 2° 讓紅黑樹根指向空
	RB_INIT(&ep- >rbr);       //等價于ep- >rbr.rbh_root = NULL;

	// 3° 讓雙向鏈表的頭指向空
	LIST_INIT(&ep- >rdlist);  //等價于ep- >rdlist.lh_first = NULL;

	// 4° 并發環境下進行互斥
	// ...該部分代碼與主線邏輯無關，可自行查看...

	//5° 保存epoll對象
	tcp- >ep = (void*)ep;
	epsocket- >ep = (void*)ep;

	return epsocket- >id;
}

對以上代碼的邏輯進行梳理，可以總結為以下6步：

1. 創建eventpoll對象
2. 讓eventpoll中的rbr指向空
3. 讓eventpoll中的rdlist指向空
4. 在并發環境下進行互斥
5. 保存eventpoll對象
6. 返回eventpoll對象的句柄(id)

(2)epoll_ctl()

該函數的邏輯其實很簡單，無非就是將用戶傳入的參數封裝為一個epitem對象，然后根據傳入的op是①EPOLL_CTL_ADD、②EPOLL_CTL_MOD還是③EPOLL_CTL_DEL，來決定是①將epitem對象插入紅黑樹中，②更新紅黑樹中的epitem對象，還是③移除紅黑樹中的epitem對象。

//往紅黑樹中加每個tcp連接以及相關的事件
int epoll_ctl(int epid, int op, int sockid, struct epoll_event *event) {

	nty_tcp_manager *tcp = nty_get_tcp_manager();
	if (!tcp) return -1;

	nty_trace_epoll(" epoll_ctl -- > 1111111:%d, sockid:%dn", epid, sockid);
	struct _nty_socket *epsocket = tcp- >fdtable- >sockfds[epid];

	if (epsocket- >socktype == NTY_TCP_SOCK_UNUSED) {
		errno = -EBADF;
		return -1;
	}

	if (epsocket- >socktype != NTY_TCP_SOCK_EPOLL) {
		errno = -EINVAL;
		return -1;
	}

	nty_trace_epoll(" epoll_ctl -- > eventpolln");

	struct eventpoll *ep = (struct eventpoll*)epsocket- >ep;
	if (!ep || (!event && op != EPOLL_CTL_DEL)) {
		errno = -EINVAL;
		return -1;
	}

	if (op == EPOLL_CTL_ADD) {
		//添加sockfd上關聯的事件
		pthread_mutex_lock(&ep- >mtx);

		struct epitem tmp;
		tmp.sockfd = sockid;
		struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep- >rbr, &tmp); //先在紅黑樹上找，根據key來找，也就是這個sockid，找的速度會非?？?/span>
		if (epi) {
			//原來有這個節點，不能再次插入
			nty_trace_epoll("rbtree is existn");
			pthread_mutex_unlock(&ep- >mtx);
			return -1;
		}

		//只有紅黑樹上沒有該節點【沒有用過EPOLL_CTL_ADD的tcp連接才能走到這里】；

		//(1)生成了一個epitem對象，這個結構對象，其實就是紅黑的一個節點；
		epi = (struct epitem*)calloc(1, sizeof(struct epitem));
		if (!epi) {
			pthread_mutex_unlock(&ep- >mtx);
			errno = -ENOMEM;
			return -1;
		}
		
		//(2)把socket(TCP連接)保存到節點中；
		epi- >sockfd = sockid;  //作為紅黑樹節點的key，保存在紅黑樹中

		//(3)我們要增加的事件也保存到節點中；
		memcpy(&epi- >event, event, sizeof(struct epoll_event));

		//(4)把這個節點插入到紅黑樹中去
		epi = RB_INSERT(_epoll_rb_socket, &ep- >rbr, epi); //實際上這個時候epi的rbn成員就會發揮作用，如果這個紅黑樹中有多個節點，那么RB_INSERT就會epi- >rbi相應的值：可以參考圖來理解
		assert(epi == NULL);
		ep- >rbcnt ++;
		
		pthread_mutex_unlock(&ep- >mtx);

	} else if (op == EPOLL_CTL_DEL) {
		pthread_mutex_lock(&ep- >mtx);

		struct epitem tmp;
		tmp.sockfd = sockid;
		
		struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep- >rbr, &tmp);//先在紅黑樹上找，根據key來找，也就是這個sockid，找的速度會非?？?/span>
		if (!epi) {
			nty_trace_epoll("rbtree no existn");
			pthread_mutex_unlock(&ep- >mtx);
			return -1;
		}
		
		//只有在紅黑樹上找到該節點【用過EPOLL_CTL_ADD的tcp連接才能走到這里】；

		//從紅黑樹上把這個節點移除
		epi = RB_REMOVE(_epoll_rb_socket, &ep- >rbr, epi);
		if (!epi) {
			nty_trace_epoll("rbtree is no existn");
			pthread_mutex_unlock(&ep- >mtx);
			return -1;
		}

		ep- >rbcnt --;
		free(epi);
		
		pthread_mutex_unlock(&ep- >mtx);

	} else if (op == EPOLL_CTL_MOD) {
		struct epitem tmp;
		tmp.sockfd = sockid;
		struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep- >rbr, &tmp); //先在紅黑樹上找，根據key來找，也就是這個sockid，找的速度會非?？?/span>
		if (epi) {
			//紅黑樹上有該節點，則修改對應的事件
			epi- >event.events = event- >events;
			epi- >event.events |= EPOLLERR | EPOLLHUP;
		} else {
			errno = -ENOENT;
			return -1;
		}

	} else {
		nty_trace_epoll("op is no existn");
		assert(0);
	}

	return 0;
}

(3)epoll_wait()

//到雙向鏈表中去取相關的事件通知
int epoll_wait(int epid, struct epoll_event *events, int maxevents, int timeout) {

	nty_tcp_manager *tcp = nty_get_tcp_manager();
	if (!tcp) return -1;

	struct _nty_socket *epsocket = tcp- >fdtable- >sockfds[epid];

	struct eventpoll *ep = (struct eventpoll*)epsocket- >ep;
	
    // ...此處主要是一些負責驗證性工作的代碼...

	//(1)當eventpoll對象的雙向鏈表為空時，程序會在這個while中等待一定時間，
	//直到有事件被觸發，操作系統將epitem插入到雙向鏈表上使得rdnum >0時，程序才會跳出while循環
	while (ep- >rdnum == 0 && timeout != 0) {
		// ...此處主要是一些與等待時間相關的代碼...
	}


	pthread_spin_lock(&ep- >lock);

	int cnt = 0;

	//(1)取得事件的數量
	//ep- >rdnum：代表雙向鏈表里邊的節點數量（也就是有多少個TCP連接來事件了）
	//maxevents：此次調用最多可以收集到maxevents個已經就緒【已經準備好】的讀寫事件
	int num = (ep- >rdnum > maxevents ? maxevents : ep- >rdnum); //哪個數量少，就取得少的數字作為要取的事件數量
	int i = 0;
	
	while (num != 0 && !LIST_EMPTY(&ep- >rdlist)) { //EPOLLET

		//(2)每次都從雙向鏈表頭取得 一個一個的節點
		struct epitem *epi = LIST_FIRST(&ep- >rdlist);

		//(3)把這個節點從雙向鏈表中刪除【但這并不影響這個節點依舊在紅黑樹中】
		LIST_REMOVE(epi, rdlink); 

		//(4)這是個標記，標記這個節點【這個節點本身是已經在紅黑樹中】已經不在雙向鏈表中；
		epi- >rdy = 0;  //當這個節點被操作系統 加入到 雙向鏈表中時，這個標記會設置為1。

		//(5)把事件標記信息拷貝出來；拷貝到提供的events參數中
		memcpy(&events[i++], &epi- >event, sizeof(struct epoll_event));
		
		num --;
		cnt ++;       //拷貝 出來的 雙向鏈表 中節點數目累加
		ep- >rdnum --; //雙向鏈表里邊的節點數量減1
	}
	
	pthread_spin_unlock(&ep- >lock);

	//(5)返回 實際 發生事件的 tcp連接的數目；
	return cnt; 
}

該函數的邏輯也十分簡單，就是讓先看一下eventpoll對象的雙鏈表中是否有節點。如果有節點的話則取出節點中的事件填充到用戶傳入的指針所指向的內存中。如果沒有節點的話，則在while循環中等待一定時間，直到有事件被觸發后操作系統會將epitem插入到雙向鏈表上使得rdnum>0時(這個過程是由操作系統調用epoll_event_callback函數完成的)，程序才會跳出while循環，去雙向鏈表中取數據。

(4)epoll_event_callback()

通過跟蹤epoll_event_callback在內核中被調用的位置?？芍?，當服務器在以下5種情況會調用epoll_event_callback：

1. 客戶端connect()連入，服務器處于SYN_RCVD狀態時
2. 三路握手完成，服務器處于ESTABLISHED狀態時
3. 客戶端close()斷開連接，服務器處于FIN_WAIT_1和FIN_WAIT_2狀態時
4. 客戶端send/write()數據，服務器可讀時
5. 服務器可以發送數據時

接下來，我們來看一下epoll_event_callback的源碼：

//當發生客戶端三路握手連入、可讀、可寫、客戶端斷開等情況時，操作系統會調用這個函數，用以往雙向鏈表中增加一個節點【該節點同時 也在紅黑樹中】
int epoll_event_callback(struct eventpoll *ep, int sockid, uint32_t event) {
	struct epitem tmp;
	tmp.sockfd = sockid;

	//(1)根據給定的key【這個TCP連接的socket】從紅黑樹中找到這個節點
	struct epitem *epi = RB_FIND(_epoll_rb_socket, &ep- >rbr, &tmp);
	if (!epi) {
		nty_trace_epoll("rbtree not existn");
		assert(0);
	}

	//(2)從紅黑樹中找到這個節點后，判斷這個節點是否已經被連入到雙向鏈表里【判斷的是rdy標志】
	if (epi- >rdy) {
		//這個節點已經在雙向鏈表里，那無非是把新發生的事件標志增加到現有的事件標志中
		epi- >event.events |= event;
		return 1;
	} 

	//走到這里，表示 雙向鏈表中并沒有這個節點，那要做的就是把這個節點連入到雙向鏈表中

	nty_trace_epoll("epoll_event_callback -- > %dn", epi- >sockfd);
	
	pthread_spin_lock(&ep- >lock);

	//(3)標記這個節點已經被放入雙向鏈表中，我們剛才研究epoll_wait()的時候，從雙向鏈表中把這個節點取走的時候，這個標志被設置回了0
	epi- >rdy = 1;  

	//(4)把這個節點鏈入到雙向鏈表的表頭位置
	LIST_INSERT_HEAD(&ep- >rdlist, epi, rdlink);

	//(5)雙向鏈表中的節點數量加1，剛才研究epoll_wait()的時候，從雙向鏈表中把這個節點取走的時候，這個數量減了1
	ep- >rdnum ++;

	pthread_spin_unlock(&ep- >lock);
	pthread_mutex_lock(&ep- >cdmtx);
	pthread_cond_signal(&ep- >cond);
	pthread_mutex_unlock(&ep- >cdmtx);

	return 0;
}

以上代碼的邏輯也十分簡單，就是將eventpoll所指向的紅黑樹的節點插入到雙向鏈表中。

總結

epoll底層實現中有兩個關鍵的數據結構，一個是eventpoll另一個是epitem，其中eventpoll中有兩個成員變量分別是rbr和rdlist,前者指向一顆紅黑樹的根，后者指向雙向鏈表的頭。而epitem則是紅黑樹節點和雙向鏈表節點的綜合體，也就是說epitem即可作為樹的節點，又可以作為鏈表的節點，并且epitem中包含著用戶注冊的事件。

• 當用戶調用epoll_create()時，會創建eventpoll對象（包含一個紅黑樹和一個雙鏈表）；
• 而用戶調用epoll_ctl(ADD)時，會在紅黑樹上增加節點（epitem對象）；
• 接下來，操作系統會默默地在通過epoll_event_callback()來管理eventpoll對象。當有事件被觸發時，操作系統則會調用epoll_event_callback函數，將含有該事件的epitem添加到雙向鏈表中。
• 當用戶需要管理連接時，只需通過epoll_wait()從eventpoll對象中的雙鏈表下"摘取"epitem并取出其包含的事件即可。