出現原因

如果需要從一個檔案描述符中讀取資料，然後將資料寫入到另一個檔案描述符時，可以按照如下的阻塞 IO ：

while ((n = read(STDIN_FILENO, buf, BUFFER_SIZE)) > 0) {
    if (write(STDOUT, buf, n) != n) {
        fprintf(stderr, "write error");
    }
}

這種方式在只有一個讀 fd 和一個寫 fd 的情況下，這種方式能夠正常工作，不會有什麼問題。但是如果存在多個檔案描述符需要被讀取，那麼在這種情況下，如果一直阻塞等待某個檔案描述符讀取完成，那麼剩下的待讀取檔案描述符即使能夠被讀取，也會一直等待。為了解決這個問題，引入多路複用（多路轉接）技術來進行處理

假設現在讓我們自己設計 telnet ，在這裡我們主要考慮一下 telnet 和遠端主機之間的通訊問題：telnet 從終端（標準輸入）中讀取輸入，將讀取到的輸入資料寫入到網路連線（fd）上，同時從網路連線中讀取資料，將讀取到的資料寫回到終端上；在網路連線的另一端，telneted 守護程序讀取使用者輸入的命令，並將其返回到終端，具體情況如下圖所示：

由於 telnet 有兩個輸入，因此傳統的阻塞讀的方式是不可取的，因為無法知道什麼時候讀取哪個輸入。

如果沒有多路複用技術，那麼可以考慮以下幾種方式解決這個問題：

• 將一個程序通過 fork，變成兩個程序

  由於變成了兩個子程序，可以單獨地對每個輸入執行阻塞讀的操作。但是這樣又會產生新的問題：如果 telnet

  斷開了連線，那麼需要將對應的程序關閉，這個操作可以通過訊號量來進行操作，但是使得程式變得更加複雜

• 不使用程序，而是使用兩個執行緒

  通過建立兩個執行緒來分別維護兩個輸入的讀取，避免了由於程序間通訊帶來的複雜性，但是由於需要同步這兩個執行緒，因此在複雜性這一方面不見得會比使用兩個程序的方式更好

• 依舊使用一個程序來進行處理，但是使用非阻塞 IO

  將兩個輸入都變成非阻塞的，對第一個輸入傳送一個 read，如果該輸入上有資料，則讀取資料並處理它；如果沒有資料可讀，則直接返回，然後對第二個輸入執行類似的操作，在此之後，等待一定的時間，再次執行相同的處理。這種方式被成為 “輪詢”，大部分情況下都是無資料可讀的，浪費了 CPU 的處理時間，因此應該避免使用

• 還是使用一個程序，但是資料的讀取採用非同步 IO

  採用非同步 IO 的方式來進行處理，每當有準備好的 IO 可以進行時，傳送訊號通知程序進行處理。這種訊號對於每個程序來講都只有一個，因此當多個 IO 準備好的情況下，無法正確判斷到底是那個 IO 準備好了，特別是，能夠使用訊號量的數量是有限的，因此當檔案描述符變多時將會存在問題。

傳統的方式都未能很好地處理 telnet 存在的問題，IO 多路複用技術可能是解決該問題比較好的一種方案。

IO 多路複用描述如下：首先構造一個檔案描述符列表，然後呼叫一個函式，直到這個列表中至少存在一個 IO 已經準備好的情況下，該函式才返回，在從該函式返回時，程序可以得到已經準備好 IO 的檔案描述符號

多路複用函式

select 和 pselect

select：呼叫 select 函式需要以下幾個引數：

• 待檢查的 fd 集合
• 對於每個 fd 我們所關心的操作：讀、寫以及異常操作
• 希望等待多長時間（可以永遠等待、等待一個固定時間或者根本不等待）

呼叫 select 之後，可以通過 select 得到以下內容：

• 已經準備好的 fd 的數量
• 對於讀、寫或異常這三個條件中的每一個，哪些 fd 已經準備好

select 函式的原型如下所示：

#include <sys/select.h>

int select(int nfds, fd_set *readfds, fd_set *writefds,
           fd_set *exceptfds, struct timeval *timeout);

函式的最後一個引數 \(timeout\) 表示 select 函式願意等待的時間長度，有以下三種情況：

1. tvptr == NULL：表示永遠等待，如果捕捉到一個訊號則中斷此狀態
2. tvptr->tv_sec == 0 && tvptr->tv_usec == 0 ：表示不等待
3. tvptr->tv_sec != 0 || tvptr->tv_usec != 0 ：表示等待指定的秒數和微秒數

對於中間的三個引數 \(readfds\)、\(writefds\)、\(exceptfds\) 表示指向 fd 集合的指標，具體的狀態如下所示：

具體的集合實現可以不同，這裡假設只是一個簡單的位元組陣列。對於 fd_set 資料型別，可以通過呼叫以下幾個函式：

#include <sys/select.h>

void FD_CLR(int fd, fd_set *set); // 清除 set 中的某一位 fd
int  FD_ISSET(int fd, fd_set *set); // 如果 fd 在 set 中，返回非 0 值，否則返回 0 值
void FD_SET(int fd, fd_set *set); // 開啟 set 中的 fd
void FD_ZERO(fd_set *set); // 將 set 的所有位都設定為 0

對應的示例如下：

#include <stdio.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

int main(int argc, char ** argv) {
    fd_set rfds; 
    struct timeval tv;
    int retval;

    FD_ZERO(&rfds); // 預設 fd 問終端輸入，fd 為 0
    FD_SET(0, &rfds);

    /* 等待 5 秒 */
    tv.tv_sec = 5;
    tv.tv_usec = 0;
    retval = select(1, &rfds, NULL, NULL, &tv);

    if (retval == -1) {
        perror("select()");
    } else if (retval) {
        printf("Data is available now.\n");
        /* FD_ISSET(0, &rfds) will be true. */
    } else {
        printf("No data within five seconds.\n");
    }

    return 0;
}

poll

poll 介面類似於 select，和 select 最大的不同之處在於 poll 可以支援任意型別的檔案描述符，函式的原型如下所示：

#include <poll.h>
int poll(struct pollfd *fds, nfds_t nfds, int timeout);

其中，struct pollfd 的具體結構如下所示：

struct pollfd {
    int   fd;         /* 檔案描述符號 */
    short events;     /* 對當前 fd 關心的事件 */
    short revents;    /* 在該 fd 上發生的時間 */
}

epoll

epoll 是現代多路複用中使用最為廣泛的多路複用介面，效能想比較於 select有 和 poll 有很大的改進，這裡重點分析一下 epoll 以及它的實現原理

連線的建立

首先，對於每一個新建立的 socket連線，都會生成對應的 fd，這個 fd 將會儲存到當前程序持有的 “開啟檔案列表” 中，具體的情況如下圖所示：

建立 eventpoll

當呼叫 epoll_create 時，會在核心中生成一個 struct eventpoll 的核心物件，並同時將這個物件放入到程序開啟的檔案描述符列表中，此時的情況可能如下圖所示：

對於 struct eventpoll，在這裡我們主要關心的欄位屬性如下：

對上圖的欄位解釋如下：

• wq：等待佇列，當呼叫 epoll 時阻塞的程序會放入這個佇列，當資料準備就緒時，在這個佇列上找到對應的阻塞程序
• rblist：已經就緒的 fd 連結串列。當有的連線已經準備就緒時，會呼叫對應的回撥函式，將這個連線對應的 fd 當如這個連結串列中，這樣程序只需要在這個連結串列中獲取就緒的 fd，而不需要遍歷整個 fd 列表
• rbr：為了支援大量連線的高效查詢、插入和刪除，在 eventpoll 中會維護一棵紅黑樹，通過這顆樹來管理已經建立的所有的 socket 連線

新增 socket

當建立一個 socket 連線之後，需要將這個連線對應的 fd 註冊到 eventpoll 中，註冊時，核心會做以下幾件事情：

1. 建立一個新的紅黑樹節點 epollitem
2. 新增等待事件到該 socket 的等待佇列中，並註冊回撥函式 ep_poll_callback
3. 將 epitem 插入到 epoll 物件的紅黑樹中

以上文的例子為例，將原有的兩個 socket 註冊到 epoll 之後的情況如下圖所示：

epoll_wait 等待接收

如果程序 A 在呼叫 epoll_wait 時發現有 socket 可用，那麼將會從 eventpoll 的 rdlist 中獲取一個 socket 進行處理，如果不存在可用的 socket，那麼需要按照以下的步驟進行處理：

1. 呼叫 epoll_wait，檢查就緒佇列中是否存在就緒的 socket
2. 如果不存在就緒的 socket，那麼首先需要定義一個等待佇列節點，準備新增到 eventpoll 的等待佇列中
3. 將準備好的等待佇列節點插入到 eventpoll 的等待佇列中
4. 程序掛起，讓出當前持有的 CPU

具體流程如下圖所示：

喚醒程序

當 socket 接收資料完成之後，會通過呼叫已經註冊到等待佇列節點中的回撥函式，在 socket 的等待佇列中，這個回撥函式是 ep_poll_callback，該函式對應的原始碼如下：

static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
{
    //獲取 wait 對應的 epitem
    struct epitem *epi = ep_item_from_wait(wait);

    //獲取 epitem 對應的 eventpoll 結構體
    struct eventpoll *ep = epi->ep;

    //1. 將當前epitem 新增到 eventpoll 的就緒佇列中
    list_add_tail(&epi->rdllink, &ep->rdllist);

    //2. 檢視 eventpoll 的等待佇列上是否有在等待
    if (waitqueue_active(&ep->wq))
        wake_up_locked(&ep->wq);
    
    // 省略部分原始碼
}

喚醒之後的程序會發現 eventepoll 的就緒佇列中已經存在就緒的 socket，因此會正常執行

水平觸發和邊沿觸發

當 socket 準備好時，在喚醒 eventpoll 中等待佇列的程序時，有兩種觸發模式：水平出發和邊沿觸發。

邊沿觸發：僅在新的事件被首次加入到 eventepoll 的就緒佇列中時才觸發，比如：當 socket buffer 從空變為非空、buffer 資料增多、程序對 buffer 修改、buffer 資料減少等都會執行一次觸發

水平觸發：在事件狀態未變更之前，將會不斷觸發喚醒事件，由於這個觸發模式涵蓋了大部分的場景，因此這是 epoll 的預設觸發模式

舉兩個例子：

• 假設現在註冊到 eventpoll 的一個 socket 的緩衝區已經可讀了，那麼在水平觸發的模式下，只要該事件沒有被處理完畢（緩衝區不為空），那麼每次呼叫 epoll_wait 時都會包含該事件，直到該事件被處理完成；而如果是在邊沿觸發的模式下，只會觸發一次讀事件，不會反覆通知
• 假設現在註冊到 eventpoll 的一個 socket 的緩衝區已經可寫了，在水平觸發的模式下，只要該 socket 對應的緩衝區沒有被寫滿，就會一直觸發 “可寫” 事件；如果是在邊沿觸發的模式下，只會在初始時觸發一次 “可寫” 事件

以下兩種情況下的 fd 推薦使用 “邊沿觸發”：

• read 或者 write 系統呼叫返回了 EAGAIN
• 非阻塞的檔案描述符

邊沿觸發的模式可能會有以下的問題：

• 如果一次可讀的 IO 很大，由於你不得不一次性將這些 IO 處理完成，因此很可能會導致此時你無法處理其它的 fd

參考：

^[1] 《Unix 環境高階程式設計》（第三版）

^[2]https://mp.weixin.qq.com/s?__biz=MzI3NzA5MzUxNA==&mid=2664609790&idx=1&sn=1e8db814b07314f11987d05a2d39eff4

^[3]https://mp.weixin.qq.com/s?__biz=MzkzMTIyNzM5NA==&mid=2247486775&idx=1&sn=c77e367a5c5284ce970f89afaa1fecad

^[4]https://zh.wikipedia.org/wiki/Epoll