前言

最近學習linux的poll機制時，總感覺學習的不夠深入，僅僅停留在按照套路寫驅動的層面。覺得還是要再深入分析，因此用一篇博客來記錄這個知識點，內容主要來源于一些書籍、網絡資源、韋東山老師的視頻教程，以及自己的分析和理解。內容會隨著本人學習的深入不斷完善，如有錯漏也希望同道中人指正。

1 應用層使用poll

1.1 poll的原型

poll的函數原型如下：

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

參數與返回值的含義如下：

1.2 使用示例

	int ret;char name[20] = {0};// 要監聽兩個文件struct pollfd fds[2];...for (int i = 0; i < 2; ++i){sprintf(name, "/dev/mydev%d", i);// 填充要監聽的文件描述符fds[i].fd = open(name, O_RDWR);if (fds[i].fd == -1){perror("err is");exit(EXIT_FAILURE);}// 指定要監聽的事件fds[i].events = POLLIN | POLLRDNORM;// 初始化返回事件fds[i].revents = 0;}while (1){// 監聽2個文件（若無監聽事件返回則永久等待）ret = poll(fds, 2, -1);if (ret == -1){perror("err is");exit(EXIT_FAILURE);}// 第一個文件有監聽的事件返回if ((fds[0].revents & POLLIN) || (fds[0].revents & POLLRDNORM)){// 做相應處理}// 第二個文件有監聽的事件返回if ((fds[1].revents & POLLIN) || (fds[1].revents & POLLRDNORM)){// 做相應處理}}...

2 驅動怎么支持poll

2.1 字符設備驅動

docker pull命令？字符設備驅動通過struct file_operations結構提供一系列的操作，其中就包含poll，應用層對poll的調用最終會執行到該結構的poll成員，因此我們要支持poll操作的話就需要填充該成員，即完成驅動層的poll函數。這個函數一般可以這么寫：

static unsigned int xxx_poll(struct file *file, poll_table *wait)
{unsigned int mask = 0;struct xxx_dev *device = file->private_data;...// 將當前進程加入讀等待隊列poll_wait(file, &device->wait_queue_for_read, wait);// 將當前進程加入寫等待隊列poll_wait(file, &device->wait_queue_for_write, wait);if (xxx) // 設備文件有數據可讀mask |= POLLIN | POLLRDNORM;if (xxx) // 設備文件有數據可寫mask |= POLLOUT | POLLWRNORM;...return mask;
}

此外，如果暫時沒有監聽事件返回，poll可能會睡眠，因此我們需要在write、read等操作中在適當的時機喚醒等待隊列中的進程。比如，寫入數據之后通常設備文件可讀，此時可以喚醒等待在讀等待隊列中的進程；而讀出數據后通常設備文件可寫，此時可以喚醒等待在寫等待隊列中的進程。

3 一些更深入的分析

以下涉及到一些源碼分析，源碼的版本是linux-kernel-5.3.1

3.1 從系統調用poll到驅動的xxx_poll

我們在應用層調用poll函數之后發生了什么？它是怎樣一步步的調用到驅動層的poll函數的，此外，它還做了哪些事情？這些內容我們可以通過跟蹤poll函數得到。

3.1.1 sys_poll

應用層的poll函數會觸發80中斷（在ARM平臺下是軟中斷svc），進而根據系統調用號找到與之對應的內核函數sys_poll（一般系統調用xxx在內核中對應的函數是sys_xxx）：

SYSCALL_DEFINE3(poll, struct pollfd __user *, ufds, unsigned int, nfds,int, timeout_msecs)
{struct timespec64 end_time, *to = NULL;int ret;// 對時間參數做一些處理if (timeout_msecs >= 0) {to = &end_time;poll_select_set_timeout(to, timeout_msecs / MSEC_PER_SEC,NSEC_PER_MSEC * (timeout_msecs % MSEC_PER_SEC));}// 調用do_sys_pollret = do_sys_poll(ufds, nfds, to);if (ret == -ERESTARTNOHAND) {// 如果出錯則做一些錯誤處理...}return ret;
}

3.1.2 do_sys_poll

static int do_sys_poll(struct pollfd __user *ufds, unsigned int nfds,struct timespec64 *end_time)
{// 在棧上申請了一個struct poll_wqueues類型的數據（這個數據比較重要，后面還會見到）struct poll_wqueues table;int err = -EFAULT, fdcount, len;// 我們知道應用層向內核傳入了一個struct pollfd類型的數組，我們要將這些數據傳入內核，因此要在內核里申請空間接收這些數據；// 而在堆上申請數據代價比較大，在棧上申請較為簡單快捷，但棧上空間有限，因此這里申請了一定數量的空間，如果還是放不下后面才會在堆上再申請空間。long stack_pps[POLL_STACK_ALLOC/sizeof(long)];// struct poll_list是變長數組，即最后的成員是一個零長的數組，這個變長數組會占據棧上申請的空間，用于存放應用層傳入的struct pollfd類型的數組struct poll_list *const head = (struct poll_list *)stack_pps;struct poll_list *walk = head;unsigned long todo = nfds;// 對進程監聽的文件數量做一個限制if (nfds > rlimit(RLIMIT_NOFILE))return -EINVAL;// 應用層傳入的struct pollfd類型的數組大小為nfds// 當前在棧上申請的空間可以放入N_STACK_PPS這么多的struct pollfd類型的元素// 取兩者當中小的那個len = min_t(unsigned int, nfds, N_STACK_PPS);// 在這個循環里把應用層傳入的struct pollfd類型的數組拷貝到內核中for (;;) {walk->next = NULL;walk->len = len;if (!len)break;// 進行拷貝操作if (copy_from_user(walk->entries, ufds + nfds-todo,sizeof(struct pollfd) * walk->len))goto out_fds;// 將已經完成拷貝的數量減去todo -= walk->len;// 全部拷貝完則跳出循環if (!todo)break;// 棧上的空間不夠，需要在堆上申請內存// 申請時是一頁一頁申請的，POLLFD_PER_PAGE表示一頁內存能夠存放多少struct pollfd// 一頁一頁的申請有什么好處呢，申請和釋放的效率更高么？這個暫不清楚len = min(todo, POLLFD_PER_PAGE);walk = walk->next = kmalloc(struct_size(walk, entries, len),GFP_KERNEL);if (!walk) {err = -ENOMEM;goto out_fds;}}// 初始化struct poll_wqueues類型的變量tablepoll_initwait(&table);// 調用do_poll，do_poll會返回發生監聽事件的文件數量fdcount = do_poll(head, &table, end_time);// do_poll函數中可能會申請一些資源，且這些資源被記錄到變量table中，因此這里要釋放poll_freewait(&table);// 將返回事件記錄到用戶空間的struct pollfd元素的revents成員中（告訴應用程序返回的是哪個監聽事件）for (walk = head; walk; walk = walk->next) {struct pollfd *fds = walk->entries;int j;for (j = 0; j < walk->len; j++, ufds++)if (__put_user(fds[j].revents, &ufds->revents))goto out_fds;}err = fdcount;
out_fds:walk = head->next;// 如果在堆上申請了空間去存放應用層傳入的數據那么在函數返回之前要釋放掉while (walk) {struct poll_list *pos = walk;walk = walk->next;kfree(pos);}// 返回的就是do_poll的返回值，如果未出錯的話會返回發生監聽事件的文件數量return err;
}

上文中，我用注釋的方式大概分析了do_sys_poll所做的事情，總的來說做了這么幾件事：

1. 將應用層傳入的struct pollfd類型的數組拷貝到內核
2. 初始化了struct poll_wqueues類型的變量table
3. 調用do_poll函數
4. 將返回事件記錄到用戶空間的struct pollfd元素的revents成員
5. 最終返回發生監聽事件的文件數量

select與poll。只有上述的注釋，可能還有一些比較重要的細節沒有介紹清楚，因此這里再做一些補充：

用棧和堆來存放應用層的struct pollfd數組：
內核首先在棧上申請空間，這段空間是一個long型數組，數組名為stack_pps。然后會將該數組通過強制類型轉換，解釋為struct poll_list類型，這個類型比較有意思，它使用了變長數組這一語言特性：

struct poll_list {struct poll_list *next;int len;struct pollfd entries[0];
};

如果棧上空間不夠，就會在堆上申請內存，申請時是一頁一頁的申請的，而這些后來申請的內存與之前堆上的內存怎么聯系在一起呢？答案是struct poll_list的next成員。這種做法在內核的其他地方也有用到，這里我們使用一張圖來說明：

poll_initwait(&table)：
這個函數用來初始化struct poll_wqueues類型的變量table，這個結構的成員如下：

struct poll_wqueues {poll_table pt;struct poll_table_page *table;struct task_struct *polling_task;int triggered;int error;int inline_index;struct poll_table_entry inline_entries[N_INLINE_POLL_ENTRIES];
};

初始化函數會為每一個成員填充初始值：

void poll_initwait(struct poll_wqueues *pwq)
{init_poll_funcptr(&pwq->pt, __pollwait);pwq->polling_task = current;pwq->triggered = 0;pwq->error = 0;pwq->table = NULL;pwq->inline_index = 0;
}

其中特別需要注意的是：pt成員的_qproc成員被賦值為了__pollwait；polling_task成員被賦值為了current（前進程在調用poll）。

3.1.3 do_poll

static int do_poll(struct poll_list *list, struct poll_wqueues *wait,struct timespec64 *end_time)
{poll_table* pt = &wait->pt;ktime_t expire, *to = NULL;int timed_out = 0, count = 0;u64 slack = 0;__poll_t busy_flag = net_busy_loop_on() ? POLL_BUSY_LOOP : 0;unsigned long busy_start = 0;// 如果不需要等待（應用層調用poll函數時傳入timeout參數為0）if (end_time && !end_time->tv_sec && !end_time->tv_nsec) {pt->_qproc = NULL; // 不需要等待的話，這個函數指針（__pollwait）就不需要了timed_out = 1;     // 不需要等待則直接將超時標記置為1}// 需要等待的話這里會處理一下時間參數，看名字應該跟時間精度有關系if (end_time && !timed_out)slack = select_estimate_accuracy(end_time);for (;;) {struct poll_list *walk;bool can_busy_loop = false;// 這個循環處理應用層傳入的struct pollfd類型的數組，數組中每個元素都代表一個要監聽的文件// 聯系上一小節的內容，struct poll_list結構會以一個鏈表的形式接收應用層傳入的struct pollfd，這個循環就是遍歷上述鏈表for (walk = list; walk != NULL; walk = walk->next) {struct pollfd * pfd, * pfd_end;pfd = walk->entries;pfd_end = pfd + walk->len;// 每個struct poll_list中都有一個變長數組，這個循環就是遍歷變長數組的，這樣一來用戶層傳入的所有struct pollfd都被遍歷到了for (; pfd != pfd_end; pfd++) {// 對每一個要監聽的文件調用do_pollfdif (do_pollfd(pfd, pt, &can_busy_loop,busy_flag)) {count++;             // 如果被監聽的文件發生了監聽的事件那么計數加1pt->_qproc = NULL;   // 既然監聽事件有了，那就無需睡眠了，這個函數指針也就不需要了...}}}// 到這里所有要監聽的文件都遍歷完了pt->_qproc = NULL;  // 這個函數指針起的作用是將當前進程掛到等待隊列上，在驅動的poll函數中的poll_wait里面調用。這個動作在當前循環中只需要進行一次。if (!count) {       // 沒有監聽事件發生count = wait->error;if (signal_pending(current)) // 判斷進程是否收到信號count = -ERESTARTNOHAND;}// 如果超時、發生錯誤、進程收到信號，或者有監聽事件發生則跳出循環（返回count）if (count || timed_out)break;// 這里應該是和套接字有關的處理，暫不深究if (can_busy_loop && !need_resched()) {...}busy_flag = 0;// 如果應用層調用poll時選擇永久延時，即timeout < 0，那么這里的end_time為NULL// 因此這里的條件判斷是針對timeout >= 0的情況，且to需要是NULL，即這個循環第一次執行到這里才會執行這個分支if (end_time && !to) {// 如果需要延時，那么我們需要把延時時間轉換成ktime_t類型，后面的睡眠函數只認ktime_t類型的時間參數expire = timespec64_to_ktime(*end_time);to = &expire;}// 運行到這里說明沒有監聽事件發生，且應用層的timeout != 0，需要睡眠// 調用poll_schedule_timeout函數進行睡眠，直到超時，或被喚醒（要么被信號喚醒，要么被讀寫等驅動程序從相應的等待隊列中喚醒）if (!poll_schedule_timeout(wait, TASK_INTERRUPTIBLE, to, slack))timed_out = 1;}return count;
}

epoll和poll？上述內容比較多，因此還是做一個總結，總的來說do_poll函數做了這么幾件事情：

1. 根據是否需要延時做一些處理（不需要延時的話會把超時參數置1）
2. 在一個大循環中：遍歷每一個被監聽的文件，并調用do_pollfd處理這些被監聽的文件；對于有監聽事件的文件進行計數；如果有監聽事件/信號/超時/出錯，則跳出循環，否則進行睡眠

do_poll調用了一個非常關鍵的函數do_pollfd，接下來就分析這個函數。

3.1.4 do_pollfd

static inline __poll_t do_pollfd(struct pollfd *pollfd, poll_table *pwait,bool *can_busy_poll,__poll_t busy_flag)
{int fd = pollfd->fd;__poll_t mask = 0, filter;struct fd f;// 校驗文件描述符的合法性if (fd < 0)goto out;mask = EPOLLNVAL;// 根據文件描述符從當前進程獲取文件對象（struct file）f = fdget(fd);if (!f.file)goto out;filter = demangle_poll(pollfd->events) | EPOLLERR | EPOLLHUP;pwait->_key = filter | busy_flag;// 調用vfs_poll處理要監聽的文件，返回該文件的所有事件（如POLLIN、POLLOUT等）mask = vfs_poll(f.file, pwait);if (mask & busy_flag)*can_busy_poll = true;// 過濾掉我們不關心的事件（我們只關心需要監聽的事件，即應用層event指定的事件）mask &= filter;fdput(f);out:/* ... and so does ->revents */pollfd->revents = mangle_poll(mask);return mask;
}

可見，正真干活的函數是vfs_poll，其實也就是這個函數，調用了驅動層的poll函數，我們終于接近真相了！

3.1.5 vfs_poll

static inline __poll_t vfs_poll(struct file *file, struct poll_table_struct *pt)
{// 如果我們沒有在驅動層注冊相應的poll函數，則返回DEFAULT_POLLMASK（內核認為這個可能性不大，畢竟既然應用層有poll的需求，驅動層就應該提供相應的poll函數）if (unlikely(!file->f_op->poll))return DEFAULT_POLLMASK;// 調用驅動層的poll函數（一般被命名為xxx_poll），返回值就是xxx_poll的返回值return file->f_op->poll(file, pt);
}

終于，我們跟蹤完了從應用層的poll到驅動層的xxx_poll的整個調用棧。在xxx_poll中，一般會根據設備的情況，將相應的標志置位并返回。比如，如設備此時有數據可讀，我們會置位POLLIN；設備有數據可寫，我們會置位POLLOUT。

3.1.6 總結

以上的跟蹤過程內容比較多，有必要在對其進行一個梳理。考慮到圖比文字更有表現力，因此這里我使用一幅圖來對上述內容做一個總結：

3.2 poll_wait做了什么

上文已經介紹過，驅動程序中xxx_poll會調用poll_wait函數將當前進程掛到等待列表上，poll_wait的原型如下：

static inline void poll_wait(struct file * filp, wait_queue_head_t * wait_address, poll_table *p)

docker原理與架構？其中參數wait_address是一個等待隊列，粗看這個原型，應該是把當前進程掛到wait_address上（事實也是如此）。不過看到有些文獻上有這樣的表述“poll_wait所做的工作是把當前進程添加到wait參數指定的等待列表（poll_table）中”，個人感覺這樣的表述不妥，讓人迷惑，poll_table中只有一個函數指針，一個key字段，哪來的“等待列表”？于是決定分析一下這個函數，也為后來者提供一點資料。

3.2.1 poll_wait的定義

static inline void poll_wait(struct file * filp, wait_queue_head_t * wait_address, poll_table *p)
{if (p && p->_qproc && wait_address)p->_qproc(filp, wait_address, p);
}

3.2.2 __pollwait

我們傳給poll_wait的參數p，實際上指向的是do_sys_poll中定義的struct poll_wqueues類型的變量table的pt成員。回憶一下上文，table的pt成員的_qproc成員被賦值為了__pollwait；polling_task成員被賦值為了current，其他成員也都在poll_initwait中做了初始化（下圖紅色加粗的是各成員初始化的值）：

因此我們調用poll_wait實際上是調用__pollwait。在給出__pollwait函數的定義并對其進行分析之前，我們還要看一個結構struct poll_table_entry。這個結構對于我們理解__pollwait函數有幫助：

struct poll_table_entry {struct file *filp; // 文件對象__poll_t key;wait_queue_entry_t wait;          // 等待隊列元素wait_queue_head_t *wait_address;  // 等待隊列頭部
};

這個結構和poll的睡眠有很大的關系，從它的成員類型大概可以猜出進程會被掛到wait_address指向的等待隊列，而使用的等待隊列元素就是wait。

好了，現在我們開始分析__pollwait函數：

static void __pollwait(struct file *filp, wait_queue_head_t *wait_address, poll_table *p)
{// pwq指向的是前文所說的變量tablestruct poll_wqueues *pwq = container_of(p, struct poll_wqueues, pt);// 申請一個poll_table_entry類型的變量，并將其地址返回給entrystruct poll_table_entry *entry = poll_get_entry(pwq);if (!entry)return;// 將filp賦值給entry的filp成員，get_file起到的作用是將filp指向的文件對象的引用計數加1entry->filp = get_file(filp);// 將調用poll_wait時傳入的等待隊列賦值給entry的wait_address成員entry->wait_address = wait_address;// 將table變量的pt成員的_key成員賦值給entry的key成員entry->key = p->_key;// entry的wait成員是一個等待隊列元素// 這里會將這個等待隊列元素的各個成員做一個初始化// flags	= 0;// private	= NULL;// func		= pollwake;init_waitqueue_func_entry(&entry->wait, pollwake);// private	= &tableentry->wait.private = pwq;// 將entry的wait成員添加到調用poll_wait時傳入的等待隊列add_wait_queue(wait_address, &entry->wait);
}

可以看到，在__pollwait函數中，kernel會申請一個struct poll_table_entry類型的變量，使用這個變量的wait成員（一個等待隊列元素）記錄當前進程，并掛接到該變量的wait_address成員指向的等待隊列（掛接到等待隊列的是等待隊列元素，這個元素會通過指針或直接或間接的指向當前進程）：

3.2.3 一些補充說明

到這里我們已經說清楚了poll_wait函數做了哪些事情，在結束這一小節之前，我再做一些補充說明，主要關于：
① 使用poll_get_entry函數申請struct poll_table_entry的一些細節；
② poll_freewait函數。

按鍵poll機制，下面開始說明：

① poll_get_entry：

使用該函數申請struct poll_table_entry時，會先在table變量的inline_entries成員（struct poll_table_entry類型的數組）中尋找可用空間，如果找不到，就會在堆上申請內存，一次申請一頁，table變量的table成員（struct poll_table_page *類型）就會指向申請到的頁。用完一頁就會再申請一頁，以頭插的方式插入table變量的table成員指向的頁鏈。這個用棧空間來優化性能、用鏈表來串聯多個內存頁的做法之前已經遇到，不再多說，給一幅圖就都清楚了：

② poll_freewait：

使用poll去監聽多個文件時，最終會調用到每個文件在驅動層的xxx_poll函數，這個函數又會調用poll_wait，進而調用__pollwait，在這個函數里面會調用poll_get_entry函數申請struct poll_table_entry，并將任務掛到調用poll_wait時指定的等待隊列。假如要監聽的文件有監聽事件，poll就會返回，那么我們之前向等待隊列掛接的等待隊列元素就需要移除，申請的struct poll_table_entry就需要被釋放。這些操作都發生在poll_freewait：

void poll_freewait(struct poll_wqueues *pwq)
{struct poll_table_page * p = pwq->table;int i;for (i = 0; i < pwq->inline_index; i++)free_poll_entry(pwq->inline_entries + i); // 從等待隊列移除等待隊列元素while (p) { // 在堆內存申請了空間struct poll_table_entry * entry;struct poll_table_page *old;entry = p->entry;do {entry--;free_poll_entry(entry); // 移除等待隊列元素} while (entry > p->entries);old = p;p = p->next;free_page((unsigned long) old); // 釋放頁}
}

PS：對私有數據的使用的一些體會

至此，一定程度上，應該已經把Linux的poll機制的一些基本內容說清楚了。在閱讀源碼的過程中，我學到了一些優化性能的技巧，以及關于私有數據（private）使用的一些技巧，這里記錄一些我對私有數據的使用的一些體會：

調用poll函數超時、通常使用void *類型的指針來指向私有數據，私有數據的內含只有私有數據的提供者才知道，kernel是不知道的，因此怎么去解讀并處理私有數據，也應該由私有數據的提供者給出。換句話說，我們給指向私有數據的指針賦值為私有數據的地址，也要提供處理私有數據的函數，因為只有我們自己知道私有數據的內容。內核會在適當的時候調用我們提供的函數處理我們提供的私有數據。

這樣做的好處是可以讓kernel中的在適當的時候調用我們提供的函數處理我們提供的私有數據這一機制保持穩定，同時私有數據可以根據實際需要靈活多變，只要提供不同的處理私有數據的函數即可。

舉例：
① 在Linux的poll機制中，我們將等待隊列元素的private成員指向了變量table，同時將func成員指向了pollwake，pollwake知道等待隊列元素的private指向的是table。pollwake在等待隊列被喚醒時會被調用，它能夠正確的處理private。

② 我們使用DECLARE_WAITQUEUE來定義等待隊列元素時，private成員指向了current，func成員指向了default_wake_function，default_wake_function會把private解讀成struct task_struct *類型的指針。

4 select和poll

loading…

5 epoll又是怎么回事

loading…

6 參考文獻

poll實現，[1] 韋東山老師一期視頻教程
[2] linux-kernel-5.3.1