什么是多路复用IO

(资料图片)

多路复用IO （IO multiplexing） 是指内核一旦发现进程指定的一个或者多个IO条件准备读取，它就通知该进程。在Linux系统中，常用的 多路复用IO 手段有 select、poll 和 epoll。

多路复用IO 主要用于处理网络请求，例如可以把多个请求句柄添加到 select 中进行监听，当有请求可进行IO的时候就会告知进程，并且把就绪的请求句柄保存下来，进程只需要对这些就绪的请求进行IO操作即可。下面通过一幅图来展示 select 的使用方式（图片来源于网络）：

多路复用IO实现原理

为了更简明的解释 多路复用IO 的原理，这里使用 select 系统调用作为分析对象。因为 select 的实现比较简单，而现在流行的 epoll 由于处于性能考虑，实现则比较复杂，不便于理解 多路复用IO 的原理，当然当理解了 select 的实现原理后，对 epoll 的实现就能应刃而解了。

select系统调用的使用

要使用 select 来监听socket是否可以进行IO，首先需要把其添加到一个类型为 fd＿set 的结构中，然后通过调用 select（） 系统调用来进行监听，下面代码介绍了怎么使用 select 来对socket进行监听的：

int socket＿can＿read（int fd）｛int retval；fd＿set set；struct timeval tv；FD＿ZERO（＆set）；FD＿SET（fd， ＆set）； tv．tv＿sec ＝ tv．tv＿usec ＝ 0； retval ＝ select（fd＋1， ＆set， NULL， NULL， ＆tv）；if （retval ＜ 0） ｛return －1； ｝return FD＿ISSET（fd， ＆set） ？ 1 ： 0；｝

通过上面的函数，可以监听一个socket句柄是否可读。

select系统调用的实现

接下来我们分析一下 select 系统调用的实现，用户程序通过调用 select 系统调用后会进入到内核态并且调用 sys＿select（） 函数，sys＿select（） 函数的实现如下：

asmlinkage longsys＿select（int n， fd＿set ＊inp， fd＿set ＊outp， fd＿set ＊exp， struct timeval ＊tvp）｛ fd＿set＿bits fds；char ＊bits；long timeout；int ret， size； timeout ＝ MAX＿SCHEDULE＿TIMEOUT；if （tvp） ｛time＿t sec， usec； ．．．if （（unsigned long） sec ＜ MAX＿SELECT＿SECONDS） ｛ timeout ＝ ROUND＿UP（usec， 1000000／HZ）； timeout ＋＝ sec ＊ （unsigned long） HZ； ｝ ｝if （n ＞ current－＞files－＞max＿fdset） n ＝ current－＞files－＞max＿fdset； ret ＝ －ENOMEM； size ＝ FDS＿BYTES（n）； bits ＝ select＿bits＿alloc（size）； fds．in ＝ （unsigned long ＊）bits； fds．out ＝ （unsigned long ＊）（bits ＋ size）； fds．ex ＝ （unsigned long ＊）（bits ＋ 2＊size）； fds．res＿in ＝ （unsigned long ＊）（bits ＋ 3＊size）； fds．res＿out ＝ （unsigned long ＊）（bits ＋ 4＊size）； fds．res＿ex ＝ （unsigned long ＊）（bits ＋ 5＊size）；if （（ret ＝ get＿fd＿set（n， inp， fds．in）） ｜｜ （ret ＝ get＿fd＿set（n， outp， fds．out）） ｜｜ （ret ＝ get＿fd＿set（n， exp， fds．ex）））goto out；zero＿fd＿set（n， fds．res＿in）；zero＿fd＿set（n， fds．res＿out）；zero＿fd＿set（n， fds．res＿ex）； ret ＝ do＿select（n， ＆fds， ＆timeout）； ．．．set＿fd＿set（n， inp， fds．res＿in）；set＿fd＿set（n， outp， fds．res＿out）；set＿fd＿set（n， exp， fds．res＿ex）；out：select＿bits＿free（bits， size）；out＿nofds：return ret；｝

sys＿select（） 函数主要把用户态的参数复制到内核态，然后再通过调用 do＿select（） 函数进行监听操作， do＿select（）函数实现如下（由于实现有点复杂，所以我们分段来分析）：

int do＿select（int n， fd＿set＿bits ＊fds， long ＊timeout）｛ poll＿table table， ＊wait；int retval， i， off；long ＿＿timeout ＝ ＊timeout； ．．．poll＿initwait（＆table）；wait ＝ ＆table；if （！＿＿timeout）wait ＝ NULL； retval ＝ 0；

上面这段代码主要通过调用 poll＿initwait（） 函数来初始化类型为 poll＿table 结构的变量 table。要理解 poll＿table结构的作用，我们先来看看下面的知识点：

因为每个socket都有个等待队列，当某个进程需要对socket进行读写的时候，如果发现此socket并不能读写， 那么就可以添加到此socket的等待队列中进行休眠，当此socket可以读写时再唤醒队列中的进程。

而 poll＿table 结构就是为了把进程添加到socket的等待队列中而创造的，我们先跳过这部分，后面分析到socket相关的知识点再来说明。

我们接着分析 do＿select（） 函数的实现：

for （；；） ｛set＿current＿state（TASK＿INTERRUPTIBLE）；for （i ＝ 0 ； i ＜ n； i＋＋） ｛ ．．． file ＝ fget（i）； mask ＝ POLLNVAL；if （file） ｛ mask ＝ DEFAULT＿POLLMASK；if （file－＞f＿op ＆＆ file－＞f＿op－＞poll） mask ＝ file－＞f＿op－＞poll（file， wait）；fput（file）； ｝

这段代码首先通过调用文件句柄的 poll（） 接口来检查文件是否能够进行IO操作，对于socket来说，这个 poll（） 接口就是 sock＿poll（），所以我们来看看 sock＿poll（） 函数的实现：

static unsigned int sock＿poll（struct file ＊file， poll＿table ＊ wait）｛struct socket ＊sock； sock ＝ socki＿lookup（file－＞f＿dentry－＞d＿inode）；return sock－＞ops－＞poll（file， sock， wait）；｝

sock＿poll（） 函数的实现很简单，首先通过 socki＿lookup（） 函数来把文件句柄转换成socket结构，接着调用socket结构的 poll（） 接口，而对应 TCP 类型的socket，这个接口对应的是 tcp＿poll（） 函数，实现如下：

unsigned int tcp＿poll（struct file ＊ file， struct socket ＊sock， poll＿table ＊wait）｛unsigned int mask；struct sock ＊sk ＝ sock－＞sk；struct tcp＿opt ＊tp ＝ ＆（sk－＞tp＿pinfo．af＿tcp）；poll＿wait（file， sk－＞sleep， wait）； ／／ 把文件添加到sk－＞sleep队列中进行等待 ．．．return mask；｝

tcp＿poll（） 函数通过调用 poll＿wait（） 函数把进程添加到socket的等待队列中。然后检测socket是否可读写，并通过mask返回可读写的状态。所以在 do＿select（） 函数中的 mask ＝ file－＞f＿op－＞poll（file， wait）； 这行代码其实调用的是 tcp＿poll（） 函数。

接着分析 do＿select（） 函数：

if （（mask ＆ POLLIN＿SET） ＆＆ ISSET（bit， ＿＿IN（fds，off））） ｛SET（bit， ＿＿RES＿IN（fds，off））； retval＋＋；wait ＝ NULL； ｝if （（mask ＆ POLLOUT＿SET） ＆＆ ISSET（bit， ＿＿OUT（fds，off））） ｛SET（bit， ＿＿RES＿OUT（fds，off））； retval＋＋；wait ＝ NULL； ｝if （（mask ＆ POLLEX＿SET） ＆＆ ISSET（bit， ＿＿EX（fds，off））） ｛SET（bit， ＿＿RES＿EX（fds，off））； retval＋＋；wait ＝ NULL； ｝

因为 mask 变量保存了socket的可读写状态，所以上面这段代码主要通过判断socket的可读写状态来把socket放置到合适的返回集合中。如果socket可读，那么就把socket放置到可读集合中，如果socket可写，那么就放置到可写集合中。

wait ＝ NULL；if （retval ｜｜ ！＿＿timeout ｜｜ signal＿pending（current））break；if（table．error） ｛ retval ＝ table．error；break； ｝ ＿＿timeout ＝ schedule＿timeout（＿＿timeout）； ｝ current－＞state ＝ TASK＿RUNNING；poll＿freewait（＆table）； ＊timeout ＝ ＿＿timeout；return retval；｝

最后这段代码的作用是，如果监听的socket集合中有可读写的socket，那么就直接返回（retval不为0时）。另外，如果调用 select（） 时超时了，或者进程接收到信号，也需要返回。

否则，通过调用 schedule＿timeout（） 来进行一次进程调度。因为前面把进程的运行状态设置成 TASK＿INTERRUPTIBLE，所以进行进程调度时就会把当前进程从运行队列中移除，进程进入休眠状态。那么什么时候进程才会变回运行状态呢？

前面我们说过，每个socket都有个等待队列，所以当socket可读写时便会把队列中的进程唤醒。这里分析一下当socket变成可读时，怎么唤醒等待队列中的进程的。

网卡接收到数据时，会进行一系列的接收数据操作，对于TCP协议来说，接收数据的调用链是： tcp＿v4＿rcv（） －＞ tcp＿data（） －＞ tcp＿data＿queue（） －＞ sock＿def＿readable（），我们来看看 sock＿def＿readable（） 函数的实现：

void sock＿def＿readable（struct sock ＊sk， int len）｛read＿lock（＆sk－＞callback＿lock）；if （sk－＞sleep ＆＆ waitqueue＿active（sk－＞sleep））wake＿up＿interruptible（sk－＞sleep）；sk＿wake＿async（sk，1，POLL＿IN）；read＿unlock（＆sk－＞callback＿lock）；｝

可以看出 sock＿def＿readable（） 函数最终会调用 wake＿up＿interruptible（） 函数来把等待队列中的进程唤醒，这时调用 select（） 的进程从休眠状态变回运行状态。