Linux內(nèi)核提供了3個(gè)關(guān)鍵函數(shù)供用戶來操作epoll，分別是：

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

而操作系統(tǒng)內(nèi)部會用到一個(gè)名叫epoll_event_callback()的回調(diào)函數(shù)來調(diào)度epoll對象中的事件，這個(gè)函數(shù)非常重要，故本文將會對上述4個(gè)函數(shù)進(jìn)行源碼分析。

源碼來源

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

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

兩個(gè)核心數(shù)據(jù)結(jié)構(gòu)

(1)epitem

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

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

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

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

(2)eventpoll

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

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

	// ...略...
	
};

四個(gè)關(guān)鍵函數(shù)

(1) epoll_create()

//創(chuàng)建epoll對象，包含一顆空紅黑樹和一個(gè)空雙向鏈表
int epoll_create(int size) {
	//與很多內(nèi)核版本一樣，size參數(shù)沒有作用，只要保證大于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° 開辟了一塊內(nèi)存用于填充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);       //等價(jià)于ep- >rbr.rbh_root = NULL;

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

	// 4° 并發(fā)環(huán)境下進(jìn)行互斥
	// ...該部分代碼與主線邏輯無關(guān)，可自行查看...

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

	return epsocket- >id;
}

對以上代碼的邏輯進(jìn)行梳理，可以總結(jié)為以下6步：

1. 創(chuàng)建eventpoll對象
2. 讓eventpoll中的rbr指向空
3. 讓eventpoll中的rdlist指向空
4. 在并發(fā)環(huán)境下進(jìn)行互斥
5. 保存eventpoll對象
6. 返回eventpoll對象的句柄(id)

(2)epoll_ctl()

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

//往紅黑樹中加每個(gè)tcp連接以及相關(guān)的事件
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上關(guān)聯(lián)的事件
		pthread_mutex_lock(&ep- >mtx);

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

		//只有紅黑樹上沒有該節(jié)點(diǎn)【沒有用過EPOLL_CTL_ADD的tcp連接才能走到這里】；

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

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

		//(4)把這個(gè)節(jié)點(diǎn)插入到紅黑樹中去
		epi = RB_INSERT(_epoll_rb_socket, &ep- >rbr, epi); //實(shí)際上這個(gè)時(shí)候epi的rbn成員就會發(fā)揮作用，如果這個(gè)紅黑樹中有多個(gè)節(jié)點(diǎn)，那么RB_INSERT就會epi- >rbi相應(yīng)的值：可以參考圖來理解
		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);//先在紅黑樹上找，根據(jù)key來找，也就是這個(gè)sockid，找的速度會非?？?/span>
		if (!epi) {
			nty_trace_epoll("rbtree no existn");
			pthread_mutex_unlock(&ep- >mtx);
			return -1;
		}
		
		//只有在紅黑樹上找到該節(jié)點(diǎn)【用過EPOLL_CTL_ADD的tcp連接才能走到這里】；

		//從紅黑樹上把這個(gè)節(jié)點(diǎn)移除
		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); //先在紅黑樹上找，根據(jù)key來找，也就是這個(gè)sockid，找的速度會非常快
		if (epi) {
			//紅黑樹上有該節(jié)點(diǎn)，則修改對應(yīng)的事件
			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()

//到雙向鏈表中去取相關(guān)的事件通知
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;
	
    // ...此處主要是一些負(fù)責(zé)驗(yàn)證性工作的代碼...

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


	pthread_spin_lock(&ep- >lock);

	int cnt = 0;

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

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

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

		//(4)這是個(gè)標(biāo)記，標(biāo)記這個(gè)節(jié)點(diǎn)【這個(gè)節(jié)點(diǎn)本身是已經(jīng)在紅黑樹中】已經(jīng)不在雙向鏈表中；
		epi- >rdy = 0;  //當(dāng)這個(gè)節(jié)點(diǎn)被操作系統(tǒng) 加入到 雙向鏈表中時(shí)，這個(gè)標(biāo)記會設(shè)置為1。

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

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

該函數(shù)的邏輯也十分簡單，就是讓先看一下eventpoll對象的雙鏈表中是否有節(jié)點(diǎn)。如果有節(jié)點(diǎn)的話則取出節(jié)點(diǎn)中的事件填充到用戶傳入的指針?biāo)赶虻膬?nèi)存中。如果沒有節(jié)點(diǎn)的話，則在while循環(huán)中等待一定時(shí)間，直到有事件被觸發(fā)后操作系統(tǒng)會將epitem插入到雙向鏈表上使得rdnum>0時(shí)(這個(gè)過程是由操作系統(tǒng)調(diào)用epoll_event_callback函數(shù)完成的)，程序才會跳出while循環(huán)，去雙向鏈表中取數(shù)據(jù)。

(4)epoll_event_callback()

通過跟蹤epoll_event_callback在內(nèi)核中被調(diào)用的位置?？芍?dāng)服務(wù)器在以下5種情況會調(diào)用epoll_event_callback：

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

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

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

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

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

	//走到這里，表示 雙向鏈表中并沒有這個(gè)節(jié)點(diǎn)，那要做的就是把這個(gè)節(jié)點(diǎn)連入到雙向鏈表中

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

	//(3)標(biāo)記這個(gè)節(jié)點(diǎn)已經(jīng)被放入雙向鏈表中，我們剛才研究epoll_wait()的時(shí)候，從雙向鏈表中把這個(gè)節(jié)點(diǎn)取走的時(shí)候，這個(gè)標(biāo)志被設(shè)置回了0
	epi- >rdy = 1;  

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

	//(5)雙向鏈表中的節(jié)點(diǎn)數(shù)量加1，剛才研究epoll_wait()的時(shí)候，從雙向鏈表中把這個(gè)節(jié)點(diǎn)取走的時(shí)候，這個(gè)數(shù)量減了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所指向的紅黑樹的節(jié)點(diǎn)插入到雙向鏈表中。

總結(jié)

epoll底層實(shí)現(xiàn)中有兩個(gè)關(guān)鍵的數(shù)據(jù)結(jié)構(gòu)，一個(gè)是eventpoll另一個(gè)是epitem，其中eventpoll中有兩個(gè)成員變量分別是rbr和rdlist,前者指向一顆紅黑樹的根，后者指向雙向鏈表的頭。而epitem則是紅黑樹節(jié)點(diǎn)和雙向鏈表節(jié)點(diǎn)的綜合體，也就是說epitem即可作為樹的節(jié)點(diǎn)，又可以作為鏈表的節(jié)點(diǎn)，并且epitem中包含著用戶注冊的事件。

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