图解深入揭秘epoll是如何实现IO多路复用的！

这个过程在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

。