這個過程在Linux上是一筆很大的開銷，更不用說創建了切換一次上下文需要幾微秒的時間因此，為了有效地向大量用戶提供服務，一個進程必須同時處理許多tcp連接現在假設一個進程保持10000個連接，那么你怎么找出哪個連接有數據要讀，哪個連接要寫呢

當然，我們可以通過循環遍歷來發現IO事件，但是這種方法太低級了我們希望有一種更有效的機制，當IO事件發生在多個連接中的一個上時，能夠直接快速地發現這些事件其實Linux操作系統已經為我們做到了這一點，也就是我們熟悉的IO復用機制這里的重用是指流程的重用

Linux上的多路復用方案包括選擇，輪詢和epoll其中，epoll的性能最好，可以支持最大的并發量所以今天我們以epoll為拆解對象，深入揭示內核是如何實現多路IO管理的

為了討論方便，我們舉一個使用epoll的簡單例子:

intmainlisten，cfd1 =接受，cfd2 =接受，efd = epoll _ createepoll_ctl，epoll_ctl，epoll_wait

其中，與epoll相關的功能如下:

創建一個Epoll對象。

Epoll_ctl:將需要管理的連接添加到Epoll對象中。

Epoll_wait:等待它管理的連接上的IO事件。

借助這個演示，我們將深入拆解epoll的原理。相信你看懂這篇文章后，對epoll的駕馭能力會變得完美！！

友情提醒，長話，小心！！

1.接受創建新的套接字

讓我們直接從服務器端的accept開始接受后，進程將創建一個新的套接字用于與相應的客戶端通信，然后將其放入當前進程的打開文件列表中

一個連接的套接字內核對象的更具體的結構圖如下。

接下來，我們來看看接收連接時創建套接字內核對象的源代碼。Accept的系統調用代碼位于源文件net/socket.c中

//file:net/socket . csys call _ define 4 struct socket * sock，* newsock//找到sock sock = sockfd _ lookup _ light，//1.1申請并初始化一個新的socket newsock = sock _ alloc，new sock—gt，type = sock—gt，類型，new sock—gt，ops = sock—gt，ops//1.2申請一個新的file對象，設置為新的socket new file = sock _ alloc _ file，.......//1.3接收連接err = sock—gt，ops—gt，接受，//1.4在當前進程的打開文件列表fd_install中添加一個新文件，1.1初始化結構套接字對象

在上面的源代碼中，第一步是調用sock_alloc來申請一個struct socket對象然后，將處于監聽狀態的套接字對象上的協議操作功能集ops分配給新套接字

inet_stream_ops的定義如下

//file:net/IP v4/af _ inet . cconstructproto _ op sinet _ stream _ ops =...accept = inet _ accept，listen = inet _ listen，sendmsg = inet _ sendmsg，

struct socket對象中有一個重要的成員——文件內核對象指針這個指針在初始化時是空的在accept方法中，將調用sock_alloc_file來申請內存并初始化然后將新的文件對象設置為sock—gt，去吧，文件

看看sock_alloc_file的實現過程:

struct file * sock _ alloc _ filestructfile * file，file = alloc _ file，void _ _ FD _ install...fdt=files_fdtable，BUG _ ON！= NULL)，rcu _ assign _ pointer，文件)，二，epoll_create的實現

當用戶進程調用epoll_create時，內核將創建一個struct eventpoll的內核對象并將它與當前進程的打開文件列表相關聯

對于struct eventpoll對象，更詳細的結構如下。

epoll_create的源代碼比較簡單。在fs/eventpoll.c下

//file:fs/event poll . csys call _ define 1 structeventpoll * EP = NULL，//創建eventpoll對象error = EP _ alloc(amp，EP)，

struct eventpoll的定義也在這個源文件中。

//file:fs/event poll . cstrueeventpoll//sys _ epoll _ wait使用的等待隊列wait _ queue _ head _ twq//準備接收的描述符將放在這里structlist _ headrdllist//每個epoll對象中都有一個紅黑樹structrb _ rootrbr......

eventpoll結構中幾個成員的含義如下:

Wq:等待隊列鏈表當軟中斷數據準備好時，epoll對象上阻塞的用戶進程將通過wq找到

一棵紅黑相間的樹為了支持海量連接的高效搜索，插入和刪除，eventpoll內部使用了紅黑樹該樹用于管理用戶進程下添加的所有套接字連接

Rdllist:就緒描述符的鏈表當一些連接準備好時，內核會將準備好的連接放入rdllist鏈表中這樣，應用程序進程只需要通過判斷鏈表就可以找到就緒進程，而不需要遍歷整個樹

當然，在申請了這個結構之后，還需要做一點初始化工作，這些都是在ep_alloc中完成的。

//file:fs/event poll . cstaticistep _ allocstructeventpoll * EP，//申請epollevent內存EP = kzaloc (sizeof (* EP)，GFP _ kernel)，//初始化等待隊列頭init _ wait queue _ head(amp，EP—gt，wq)，//初始化就緒列表INIT _ LIST _ HEAD(amp，EP—gt，rdllist)，//初始化紅黑樹指針EP—gt，rbr = RB _ ROOT......

說起來，這些成員都是剛剛定義或初始化的，還沒有使用過它們將在下面使用

3.將套接字添加到epoll_ctl

理解這一步是理解整個epoll的關鍵。

為簡單起見，我們只考慮使用EPOLL_CTL_ADD來添加socket，忽略刪除和先更新。

讓我們假設我們已經為與客戶機和epoll內核對象的多個連接創建了套接字。當用epoll_ctl注冊每個套接字時，內核將做以下三件事

1.分配紅黑樹節點對象表項，

2.在socket的等待隊列中添加一個等待事件，其回調函數為ep_poll_callback。

3.將epitem插入epoll對象的紅黑樹中

通過epoll_ctl添加兩個socket后，這些內核數據結構在流程中的最終示意圖大致如下:

我們來詳細看看socket是如何添加到epoll對象中的，并找到epoll_ctl的源代碼。

//file:fs/event poll . csys call _ define 4 structeventpoll * EP，structfile*file，* tfile//根據epfd找到eventpoll內核對象file = fget(epfd)，EP = file—gt，private _ data//根據套接字句柄號，找到其文件內核對象tfile = fget(FD)，開關(op)caseEPOLL_CTL_ADD:if(！epi)事件

在epoll_ctl中，首先根據傳入的fd找到與eventpoll和socket相關的內核對象對于EPOLL_CTL_ADD操作，它將執行ep_insert函數的所有注冊都在此函數中完成

//file:fs/event poll . cstaticistep _ insert//3.1分配并初始化epitem//分配一個epi對象structepitem * epi如果(！(epi=kmem_cache_alloc(epi_cache，GFP _ KERNEL)))return—eno mem，//初始化分配的epi//epi—gt，句柄號和結構文件對象地址INIT _ LIST _ HEAD(amp，epi—gt，pwqlist)，epi—gt，ep = epEP _ set _ ffd(amp，epi—gt，ffd，tfile，FD)，//3.2設置套接字等待隊列//定義并初始化ep_pqueue對象structep _ pqueueepqepq.epi = epiinit _ poll _ func ptr(amp，epq.pt，EP _ ptable _ queue _ proc)，//調用ep_ptable_queue_proc注冊回調函數//實際注入的函數是EP _ poll _ callback events = EP _ item _ poll(epi，ampEPQ . pt)，.......//3.3將epi插入eventpoll對象ep_rbtree_insert中的紅黑樹(ep，epi)，.......3.1分配和初始化epitem

對于每個套接字，當調用epoll_ctl時，將為其分配一個epitem。該結構的主要數據如下:

//file:fs/event poll . cstrueitem//紅黑樹節點structrb _ noderbn//套接字文件描述符信息structepoll _ filefdffd//它所屬的eventpoll對象structeventpoll * ep//等待隊列structlist _ headpwqlist

Epitem初始化，首先在epi—gt，Ep = ep這行代碼將其Ep指針指向eventpoll對象。另外，用要添加的套接字的文件和fd填充epitem—gt，ffd .

使用的ep_set_ffd函數如下。

staticinlinevoidep _ set _ ffd ffd—file = file，ffd—FD = FD，3.2設置套接字等待隊列

創建并初始化epitem后，ep_insert中的第二件事是在socket對象上設置等待任務隊列并在文件fs/eventpoll.c下設置ep_poll_callback作為數據準備好時的回調函數

這個塊的源代碼有點復雜如果你不耐煩，直接跳到下面的粗體字我們先來看ep_item_poll

staticinlineinsignedintep _ item _ poll pt—_ key = epi—event . events，return epi—ffd . file—f _ op—poll(epi—ffd . file，pt)amp，epi—event .事件，

看，這里調用了socket下的file—gt，f _ op—gt，民意測驗.從上面第一節socket的結構圖我們知道這個函數其實就是sock_poll。

/* Nokernellockheld—perfect */staticunsignedinsock _ poll...returnsock—ops—poll(文件，sock，等待)，

也回頭看看第一節socket的結構圖，sock—gt，ops—gt，Poll實際指向tcp_poll。

//file:net/IP v4/TCP . cunsigned TCP _ pollstructsock * sk = sock—sk，sock_poll_wait(file，sk_sleep(sk)，wait)，

在傳遞sock_poll_wait的第二個參數之前，調用sk_sleep函數在這個函數中，它獲取sock對象下的等待隊列列表頭wait_queue_head_t，后面會在這里插入等待隊列項注意這里是套接字的等待隊列，而不是epoll對象

//file:include/net/sock . hstaticinlinewait _ queue _ head _ t * sk _ sleep build _ BUG _ ON(offset of(struct socket _ wq，wait)！=0)，returnamprcu _ de reference _ raw(sk—sk _ wq)—等待，

然后真的進入sock_poll_wait。

staticinlinevoidsock _ poll _ wait poll _ wait(filp，wait_address，p)，

staticinlinevoidpoll _ waitif(Pamp，ampp—_ qprocamp，ampwait_address)p—_ qpro(filp，wait _ address，p)，

這里的qproc是一個函數指針，在前面調用init_poll_funcptr時設置為ep_ptable_queue_proc函數。

靜態插入...init _ poll _ func ptr(amp，epq.pt，EP _ ptable _ queue _ proc)，...

//file:include/Linux/poll . hstaticinlinevoidinit _ poll _ funcptrpt—gt，_ qproc = qprocpt—gt，_ key = ~ 0UL/*所有事件啟用*/

敲黑板！！！注意，過了好久才說到重點！在ep_ptable_queue_proc函數中，創建了一個新的等待隊列項，并將其回調函數注冊為ep_poll_callback函數然后將這個等待項添加到套接字的等待隊列中

//file:fs/event poll . cstaticvoidep _ ptable _ queue _ procstructeppoll _ entry * pwq，f(epi—nwait = 0 amp，amp(pwq = kmem _ cache _ alloy (pwq _ cache，GFP _ kernel))//初始化回調方法init _ wait queue _ func _ entry(amp，pwq—gt，等等，EP _ poll _ callback)，//將ep_poll_callback放入socket的等待隊列，whead(注意不是epoll的等待隊列)，add_wait_queue(whead，amppwq—gt，等等)，

在之前深入了解高性能網絡開發道路上的絆腳石——同步阻塞網絡IO中阻塞系統調用recvfrom時，由于需要在數據就緒時喚醒用戶進程，所以將等待對象項的私有設置為當前用戶進程描述符current而今天的socket是由epoll管理的，所以我們不需要在一個socket準備好的時候喚醒進程，所以這里有q—gt，Private在沒用的時候設置為NULL

//file:include/Linux/wait . hstaticinlinevoidinit _ wait queue _ func _ entry q—flags = 0，q—private = NULL，//ep_poll_callback注冊在wait_queue_t對象上//數據到達時調用q—funcq—func = func，

如上，在等待隊列條目中只設置了回調函數q—gt，Func是ep_poll_callback我們將在第5節中看到，數據到來，軟中斷將數據接收到socket的接收隊列后，會通過注冊的ep_poll_callback函數回調，然后通知epoll對象

3.3插入紅黑樹

分配完epitem對象后，立即將其插入紅黑樹。插入了一些套接字描述符的epoll中的紅黑樹示意圖如下:

先說說這里為什么用紅黑樹很多人說是因為效率高其實我覺得這個解釋不夠全面不如說搜索效率樹可以和HASHTABLE相比個人認為更合理的解釋是讓epoll在搜索效率，插入效率，內存開銷等方面更加均衡最后我發現最適合這個需求的數據結構是紅黑樹

四。epoll_wait等待接收

epoll_wait的功能并不復雜當它被調用時，它觀察event poll—gt，rdllist鏈表中有數據嗎有數據就還如果沒有數據，就創建一個等待隊列項，添加到eventpoll的等待隊列中，然后自己屏蔽

注意:當epoll_ctl添加socket時，它還會創建一個等待隊列條目不同的是，這里的等待隊列項掛在epoll對象上，而前者掛在socket對象上

其源代碼如下:

//file:fs/event poll . csyscall _ define 4...error=ep_poll(ep，events，maxevents，time out)，static intep _ poll(struct event poll * EP，structepoll_event__user*events，intmaxevents，long time out)wait _ queue _ twait，......fetch_events://4.1判斷是否有事件就緒如果(！Ep_events_available(ep))//4.2定義等待事件并關聯當前進程init _ wait queue _ entry(amp，等等，當前)，//4.3將新的waitqueue添加到epoll—gt，_ _ add _ wait _ queue _ exclusive(amp，EP—gt，wq，amp等等)，for(，，)...//4.4讓CPU主動進入睡眠狀態如果(！schedule_hrtimeout_range(to，slack，HR timer _ MODE _ ABS))timed _ out = 1，...4.1判斷就緒隊列中是否有事件就緒。

首先，調用ep_events_available來確定就緒鏈表中是否有任何可處理的事件。

//file:fs/event poll . cstaticinlineintep _ events _ available return！list _ empty(amp，ep—rdllist)ep—ovflist！= EP _ UNACTIVE _ PTR4.2定義等待事件并關聯當前流程

假設沒有就緒連接，它將轉到init_waitqueue_entry來定義等待任務，并將當前添加到waitqueue。

是的，當沒有IO事件時，epoll也會阻塞當前進程這是合理的，因為沒什么事做，占用CPU也沒什么意義網上很多文章在討論堵與不堵之類的概念時，都有一個非常不好的習慣，就是不說主題這會讓你看起來云里霧里就拿epoll來說，epoll本身是阻塞的，但是一般情況下socket會設置為非阻塞這些概念只有在主體被說出時才有意義

//file:include/Linux/wait . hstaticinlinevoidinit _ wait queue _ entry q—gt，標志= 0，q—gt，private = p，q—gt，func =默認_喚醒_功能，

注意這里回調函數的名字是default_wake_function當數據到達時，這個函數將在下面的第5節中被調用

4.3加入等待隊列

staticinlinevoid _ _ add _ wait _ queue _ exclusive wait—flags

這里，上一節中定義的等待事件被添加到epoll對象的等待隊列中。

4.4放棄CPU主動進入睡眠。

通過set_current_state將當前流程設置為可中斷調用schedule_hrtimeout_range放棄CPU，主動休眠

//file:kernel/HR timer . cint _ _ sched schedule _ HR time out _ range return schedule _ HR time out _ range _ CLOCK(expires，delta，mode，CLOCK _ MONOTONIC)，int _ _ sched schedule _ HR time out _ range _ clockschedule，

在計劃中選擇下一個進程計劃。

//file:kernel/sched/core . cstaticvoid _ _ sched _ _ schedule next = pick _ next _ task(rq)，...context_switch(rq，prev，next)，第五，數據來了。

在之前執行epoll_ctl的過程中，內核為每個套接字添加了一個等待隊列條目在epoll_wait的末尾，一個等待隊列元素被添加到事件輪詢對象中在討論數據接收之前，讓我們稍微總結一下這些隊列項的內容

socket—gt，sock—gt，sk_data_ready設置的就緒處理函數是sock_def_readable。

在socket的等待隊列條目中，它的回調函數是ep_poll_callback另外，它的私有是沒有用的，指向空指針null

在eventpoll的等待隊列條目中，回調函數是default_wake_functionPrivate是指等待事件的用戶進程

本節我們將看到軟中斷是如何在數據處理后依次進入各個回調函數，最后通知用戶進程的。

5.1接收數據到任務隊列

軟中斷如何處理網絡幀，這里不做介紹，以免過于笨重如果你有興趣，可以看看《舉例說明Linux網絡包接收過程》一文今天我們就從tcp協議棧的處理入口函數tcp _ V4 _ RCV說起

在tcp_v4_rcv中，首先在本地計算機上根據接收到的網絡報文頭中的source和dest信息查詢對應的socket找到之后，我們直接去找接收到的主題函數tcp_v4_do_rcv

//file:net/IP v4/TCP _ IP v4 . CIN TCP _ v4 _ do _ rcv if(sk—sk _ state TCP _ established)//執行數據處理if (TCP _ rcv _ established (sk，skb，TCP _ HDR (skb)，sk b—gt len))rsk = sk，gotoresetreturn0//非建立狀態下的其他數據包處理......

假設我們是在established狀態下處理數據包，然后進入tcp_rcv_established函數進行處理。

//file:net/IP v4/TCP _ input . CIN TCP _ rcv _ established...//接收數據入隊列eated = TCP _ queue _ rcv(SK，SKB，TCP _ header _ len，ampfrag stocked)，//數據就緒，喚醒套接字上被阻塞的進程sk—gt，sk_data_ready(sk，0)，

在tcp_rcv_established中，通過調用tcp_queue_rcv函數將接收數據放到socket的接收隊列中。

如下面的源代碼所示

//file:net/IP v4/TCP _ input . cstaticint _ _ must _ check TCP _ queue _ rcv//將接收到的數據放在套接字的接收隊列末尾if(！吃了)_ _ skb _ queue _ tail(amp，sk—gt，sk_receive_queue，skb)，skb_set_owner_r(skb，sk)，returneaten5.2查找就緒回調函數

在調用tcp_queue_rcv進行接收之后，再調用sk_data_ready來喚醒等待套接字的用戶進程這是另一個函數指針回想一下在accept函數創建sock的過程中，上面第一節提到的sock_init_data函數在這個函數中，sk_data_ready已經被設置為sock_def_readable函數這是默認的數據就緒處理功能

當socket上的數據準備好了，內核會用Sock _ DEF _ REABLE這個函數作為入口，找到添加socket時epoll_ctl設置的回調函數ep_poll_callback。

我們來詳細看看細節:

//file:net/core/sock . cstaticvoidsock _ def _ readablestructsocket _ wq * wq，rcu _ read _ lock，wq = rcu _ de reference(sk—sk _ wq)，//這個名字起得不好并不是說有一個阻斷的過程

事實上，這里所有的函數名都很混亂。

Wq_has_sleeper，對于一個簡單的recvfrom系統調用，其實就是判斷是否有進程阻塞但是對于epoll下的socket，只是判斷等待隊列不為空，不一定有進程阻塞

Wake _ up _ interrupt _ sync _ poll，只進入socket等待隊列項上設置的回調函數，不一定有喚醒進程的操作。

那么我們就重點講一下wake _ up _ interrupt _ sync _ poll。

讓我們來看看內核是如何找到注冊在等待隊列條目中的回調函數的。

//file:include/Linux/wait . h # define wake _ up _ INTERRUPTIBLE _ sync _ poll _ _ wake _ up _ sync _ key((x)，TASK_INTERRUPTIBLE，1，(void*)(m))

//file:kernel/sched/core . cvoid _ _ wake _ up _ sync _ key...__wake_up_common(q，mode，nr_exclusive，wake_flags，key)，

然后輸入__wake_up_common。

static void _ _ wake _ up _ common wait _ queue _ t * curr，* nextlist_for_each_entry_safe(curr，next，ampq—task_list，task _ list)unsigned flags = curr—flags，if(curr—func(curr，mode，wake_flags，key)amp，amp(flagsampWQ _ FLAG _ EXCLUSIVE)amp，amp！—NR _ exclusive)break，

在__wake_up_common中，選擇一個注冊在等待隊列中的元素curr，并回調其curr—gt，功能.當我們調用ep_insert時，我們將這個函數設置為ep_poll_callback。

5.3執行套接字就緒回調函數

前面找到了socket等待隊列項中注冊的函數ep_poll_callback，然后軟中斷會調用它。

//file:fs/event poll . cstaticintip _ poll _ callback//獲取等待對應的epitemsstructureepitem * epi = EP _ item _ from _ wait(wait)，//獲取epitem對應的eventpoll結構structeventpoll * EP = epi—gt，EP，//1.將當前epitem添加到event poll list _ add _ tail(amp，epi—gt，rdllink，ampEP—gt，rdllist)，//2.檢查是否存在等待if(wait queue _ active(amp，EP—gt，wq))wake _ up _ locked(amp，EP—gt，wq)，

在ep_poll_callback中，根據等待任務隊列項上的extra base指針，可以找到epitem，然后也可以找到eventpoll對象。

它做的第一件事是將自己的epitem添加到epoll的就緒隊列中。

然后，它將檢查eventpoll對象上的等待隊列中是否有等待項。

如果你不執行軟中斷，你就完了如果有等待項，找到等待項中設置的回調函數

調用wake _ up _ locked = gt_ _ wake _ up _ locked = gt__wake_up_common .

static void _ _ wake _ up _ common wait _ queue _ t * curr，* nextlist_for_each_entry_safe(curr，next，ampq—task_list，task _ list)unsigned flags = curr—flags，if(curr—func(curr，mode，wake_flags，key)amp，amp(flagsampWQ _ FLAG _ EXCLUSIVE)amp，amp！—NR _ exclusive)break，

在__wake_up_common中，調用curr—gt，功能.這里的func是epoll_wait傳入的default_wake_function函數。

5.4執行epoll就緒通知

在default_wake_function的等待隊列項中找到進程描述符，然后喚醒它。

源代碼如下:

//file:kernel/sched/core . cint default _ wake _ functionreturntry _ to _ wake _ up(curr—private，mode，wake _ flags)，

等待隊列項目curr—gt私有指針是一個在等待epoll對象時被阻塞的進程

將epoll_wait進程推入runnable隊列，等待內核重新調度該進程然后，在對應于epoll_wait的進程重新運行后，它將從調度中恢復

當進程喚醒時，繼續執行在epoll_wait處暫停的代碼將rdlist中的就緒事件返回給用戶進程

//file:fs/event poll . cstaticintep _ poll......_ _ remove _ wait _ queue(amp，ep—wq，amp等等)，set_current_state(任務_運行)，Check_events://將就緒事件返回給用戶進程EP _ send _ events (EP，events，max events))

從用戶的角度來看，epoll_wait只是多等了一會兒，但是執行過程仍然是順序的。

摘要

下面用一張圖總結一下epoll的整個工作距離。

其中，軟中斷回調時，回調函數也被整理出來:

Sock _ def _ readable:初始化Sock對象時設置= gtEP _ poll _ callback:epoll _ CTL = gt時添加到socketdefault _ wake _ function:epoll _ wait設置為epoll。

綜上所述，epoll相關函數的內核運行環境分為兩部分:

用戶內核狀態當調用epoll_wait之類的函數時，進程將被困在內核狀態中執行這部分代碼負責檢查接收隊列和阻塞當前進程，放棄CPU

硬中斷上下文在這些組件中，從網卡接收數據包進行處理，然后放入socket的接收隊列中對于epoll，找到與socket關聯的epitem，并將其添加到epoll對象的就緒列表中這時候順便檢查一下epoll上是否有阻塞的進程，如果有就喚醒它

為了介紹每一個細節，本文涉及到很多過程，包括分塊。

但實際上，只要有足夠的工作，epoll_wait根本不會阻塞進程用戶會一直工作，一直工作，直到epoll_wait中真的沒有工作可做，才會主動放棄CPU