作者：董昊 博客鏈接http://donghao.org/uii/

poll和epoll的使用應該不用再多說了。當fd很多時，使用epoll比poll效率更高。

我們通過內核源碼分析來看看到底是為什么。

poll剖析
poll系統調用：
int poll(struct pollfd *fds, nfds_t nfds, int timeout);
內核2.6.9對應的實現代碼為：
[fs/select.c -->sys_poll]
456 asmlinkage long sys_poll(struct pollfd __user * ufds, unsigned int nfds, long timeout)
457 {
458 struct poll_wqueues table;
459 int fdcount, err;
460 unsigned int i;
461 struct poll_list *head;
462 struct poll_list *walk;
463
464 /* Do a sanity check on nfds ... */ /* 用戶給的nfds數不可以超過一個struct file結構支持
的最大fd數（默認是256）*/
465 if (nfds > current->files->max_fdset && nfds > OPEN_MAX)
466 return -EINVAL;
467
468 if (timeout) {
469 /* Careful about overflow in the intermediate values */
470 if ((unsigned long) timeout < MAX_SCHEDULE_TIMEOUT / HZ)
471 timeout = (unsigned long)(timeout*HZ+999)/1000+1;
472 else /* Negative or overflow */
473 timeout = MAX_SCHEDULE_TIMEOUT;
474 }
475
476 poll_initwait(&table);
其中poll_initwait較為關鍵，從字面上看，應該是初始化變量table，注意此處table在整個執行poll的過
程中是很關鍵的變量。
而struct poll_table其實就只包含了一個函數指針:
[fs/poll.h]
16 /*
17 * structures and helpers for f_op->poll implementations
18 */
19 typedef void (*poll_queue_proc)(struct file *, wait_queue_head_t *, struct
poll_table_struct *);
20
21 typedef struct poll_table_struct {
22 poll_queue_proc qproc;
23 } poll_table;
現在我們來看看poll_initwait到底在做些什么
[fs/select.c]
57 void __pollwait(struct file *filp, wait_queue_head_t *wait_address, poll_table *p);
58
59 void poll_initwait(struct poll_wqueues *pwq)
60 {
61 &(pwq->pt)->qproc = __pollwait; /*此行已經被我“翻譯”了，方便觀看*/
62 pwq->error = 0;
63 pwq->table = NULL;
64 }
很明顯，poll_initwait的主要動作就是把table變量的成員poll_table對應的回調函數置為__pollwait。這
個__pollwait不僅是poll系統調用需要，select系統調用也一樣是用這個__pollwait，說白了，這是個操
作系統的異步操作的“御用”回調函數。當然了，epoll沒有用這個，它另外新增了一個回調函數，以達到其
高效運轉的目的，這是后話，暫且不表。
我們先不討論__pollwait的具體實現，還是繼續看sys_poll:
[fs/select.c -->sys_poll]
478 head = NULL;
479 walk = NULL;
480 i = nfds;
481 err = -ENOMEM;
482 while(i!=0) {
483 struct poll_list *pp;
484 pp = kmalloc(sizeof(struct poll_list)+
485 sizeof(struct pollfd)*
486 (i>POLLFD_PER_PAGE?POLLFD_PER_PAGE:i),
487 GFP_KERNEL);
488 if(pp==NULL)
489 goto out_fds;
490 pp->next=NULL;
491 pp->len = (i>POLLFD_PER_PAGE?POLLFD_PER_PAGE:i);
492 if (head == NULL)
493 head = pp;
494 else
495 walk->next = pp;
496
497 walk = pp;
498 if (copy_from_user(pp->entries, ufds + nfds-i,
499 sizeof(struct pollfd)*pp->len)) {
500 err = -EFAULT;
501 goto out_fds;
502 }
503 i -= pp->len;
504 }
505 fdcount = do_poll(nfds, head, &table, timeout);
這一大堆代碼就是建立一個鏈表，每個鏈表的節點是一個page大小（通常是4k），這鏈表節點由一個指向
struct poll_list的指針掌控，而眾多的struct pollfd就通過struct_list的entries成員訪問。上面的循環就
是把用戶態的struct pollfd拷進這些entries里。通常用戶程序的poll調用就監控幾個fd，所以上面這個鏈
表通常也就只需要一個節點，即操作系統的一頁。但是，當用戶傳入的fd很多時，由于poll系統調用每次都
要把所有struct pollfd拷進內核，所以參數傳遞和頁分配此時就成了poll系統調用的性能瓶頸。
最后一句do_poll，我們跟進去：
[fs/select.c-->sys_poll()-->do_poll()]
395 static void do_pollfd(unsigned int num, struct pollfd * fdpage,
396 poll_table ** pwait, int *count)
397 {
398 int i;
399
400 for (i = 0; i < num; i++) {
401 int fd;
402 unsigned int mask;
403 struct pollfd *fdp;
404
405 mask = 0;
406 fdp = fdpage+i;
407 fd = fdp->fd;
408 if (fd >= 0) {
409 struct file * file = fget(fd);
410 mask = POLLNVAL;
411 if (file != NULL) {
412 mask = DEFAULT_POLLMASK;
413 if (file->f_op && file->f_op->poll)
414 mask = file->f_op->poll(file, *pwait);
415 mask &= fdp->events | POLLERR | POLLHUP;
416 fput(file);
417 }
418 if (mask) {
419 *pwait = NULL;
420 (*count)++;
421 }
422 }
423 fdp->revents = mask;
424 }
425 }
426
427 static int do_poll(unsigned int nfds, struct poll_list *list,
428 struct poll_wqueues *wait, long timeout)
429 {
430 int count = 0;
431 poll_table* pt = &wait->pt;
432
433 if (!timeout)
434 pt = NULL;
435
436 for (;;) {
437 struct poll_list *walk;
438 set_current_state(TASK_INTERRUPTIBLE);
439 walk = list;
440 while(walk != NULL) {
441 do_pollfd( walk->len, walk->entries, &pt, &count);
442 walk = walk->next;
443 }
444 pt = NULL;
445 if (count || !timeout || signal_pending(current))
446 break;
447 count = wait->error;
448 if (count)
449 break;
450 timeout = schedule_timeout(timeout); /* 讓current掛起，別的進程跑，timeout到了
以后再回來運行current*/
451 }
452 __set_current_state(TASK_RUNNING);
453 return count;
454 }
注意438行的set_current_state和445行的signal_pending，它們兩句保障了當用戶程序在調用poll后
掛起時，發信號可以讓程序迅速推出poll調用，而通常的系統調用是不會被信號打斷的。
縱覽do_poll函數，主要是在循環內等待，直到count大于0才跳出循環，而count主要是靠do_pollfd函數
處理。
注意標紅的440-443行，當用戶傳入的fd很多時（比如1000個），對do_pollfd就會調用很多次，poll效
率瓶頸的另一原因就在這里。
do_pollfd就是針對每個傳進來的fd，調用它們各自對應的poll函數，簡化一下調用過程，如下：
struct file* file = fget(fd);
file->f_op->poll（file, &(table->pt));
如果fd對應的是某個socket，do_pollfd調用的就是網絡設備驅動實現的poll；如果fd對應的是某個ext3文
件系統上的一個打開文件，那do_pollfd調用的就是ext3文件系統驅動實現的poll。一句話，這個file-
>f_op->poll是設備驅動程序實現的，那設備驅動程序的poll實現通常又是什么樣子呢？其實，設備驅動
程序的標準實現是：調用poll_wait，即以設備自己的等待隊列為參數（通常設備都有自己的等待隊列，不
然一個不支持異步操作的設備會讓人很郁悶）調用struct poll_table的回調函數。
作為驅動程序的代表，我們看看socket在使用tcp時的代碼：
[net/ipv4/tcp.c-->tcp_poll]
329 unsigned int tcp_poll(struct file *file, struct socket *sock, poll_table *wait)
330 {
331 unsigned int mask;
332 struct sock *sk = sock->sk;
333 struct tcp_opt *tp = tcp_sk(sk);
334
335 poll_wait(file, sk->sk_sleep, wait);
代碼就看這些，剩下的無非就是判斷狀態、返回狀態值，tcp_poll的核心實現就是poll_wait，而
poll_wait就是調用struct poll_table對應的回調函數，那poll系統調用對應的回調函數就是
__poll_wait，所以這里幾乎就可以把tcp_poll理解為一個語句：
__poll_wait(file, sk->sk_sleep, wait);
由此也可以看出，每個socket自己都帶有一個等待隊列sk_sleep，所以上面我們所說的“設備的等待隊列”
其實不止一個。
這時候我們再看看__poll_wait的實現:
[fs/select.c-->__poll_wait()]
89 void __pollwait(struct file *filp, wait_queue_head_t *wait_address, poll_table *_p)
90 {
91 struct poll_wqueues *p = container_of(_p, struct poll_wqueues, pt);
92 struct poll_table_page *table = p->table;
93
94 if (!table || POLL_TABLE_FULL(table)) {
95 struct poll_table_page *new_table;
96
97 new_table = (struct poll_table_page *) __get_free_page(GFP_KERNEL);
98 if (!new_table) {
99 p->error = -ENOMEM;
100 __set_current_state(TASK_RUNNING);
101 return;
102 }
103 new_table->entry = new_table->entries;
104 new_table->next = table;
105 p->table = new_table;
106 table = new_table;
107 }
108
109 /* Add a new entry */
110 {
111 struct poll_table_entry * entry = table->entry;
112 table->entry = entry+1;
113 get_file(filp);
114 entry->filp = filp;
115 entry->wait_address = wait_address;
116 init_waitqueue_entry(&entry->wait, current);
117 add_wait_queue(wait_address,&entry->wait);
118 }
119 }

?

__poll_wait的作用就是創建了上圖所示的數據結構（一次__poll_wait即一次設備poll調用只創建一個
poll_table_entry），并通過struct poll_table_entry的wait成員，把current掛在了設備的等待隊列
上，此處的等待隊列是wait_address，對應tcp_poll里的sk->sk_sleep。
現在我們可以回顧一下poll系統調用的原理了：先注冊回調函數__poll_wait，再初始化table變量（類型
為struct poll_wqueues)，接著拷貝用戶傳入的struct pollfd（其實主要是fd），然后輪流調用所有fd對
應的poll（把current掛到各個fd對應的設備等待隊列上）。在設備收到一條消息（網絡設備）或填寫完文
件數據（磁盤設備）后，會喚醒設備等待隊列上的進程，這時current便被喚醒了。current醒來后離開
sys_poll的操作相對簡單，這里就不逐行分析了。

epoll原理簡介

通過上面的分析，poll運行效率的兩個瓶頸已經找出，現在的問題是怎么改進。首先，每次poll都要把
1000個fd 拷入內核，太不科學了，內核干嘛不自己保存已經拷入的fd呢？答對了，epoll就是自己保存拷
入的fd，它的API就已經說明了這一點——不是 epoll_wait的時候才傳入fd，而是通過epoll_ctl把所有fd
傳入內核再一起"wait"，這就省掉了不必要的重復拷貝。其次，在 epoll_wait時，也不是把current輪流
的加入fd對應的設備等待隊列，而是在設備等待隊列醒來時調用一個回調函數（當然，這就需要“喚醒回
調”機制），把產生事件的fd歸入一個鏈表，然后返回這個鏈表上的fd。
epoll剖析
epoll是個module，所以先看看module的入口eventpoll_init
[fs/eventpoll.c-->evetpoll_init()]
1582 static int __init eventpoll_init(void)
1583 {
1584 int error;
1585
1586 init_MUTEX(&epsem);
1587
1588 /* Initialize the structure used to perform safe poll wait head wake ups */
1589 ep_poll_safewake_init(&psw);
1590
1591 /* Allocates slab cache used to allocate "struct epitem" items */
1592 epi_cache = kmem_cache_create("eventpoll_epi", sizeof(struct epitem),
1593 0, SLAB_HWCACHE_ALIGN|EPI_SLAB_DEBUG|SLAB_PANIC,
1594 NULL, NULL);
1595
1596 /* Allocates slab cache used to allocate "struct eppoll_entry" */
1597 pwq_cache = kmem_cache_create("eventpoll_pwq",
1598 sizeof(struct eppoll_entry), 0,
1599 EPI_SLAB_DEBUG|SLAB_PANIC, NULL, NULL);
1600
1601 /*
1602 * Register the virtual file system that will be the source of inodes
1603 * for the eventpoll files
1604 */
1605 error = register_filesystem(&eventpoll_fs_type);
1606 if (error)
1607 goto epanic;
1608
1609 /* Mount the above commented virtual file system */
1610 eventpoll_mnt = kern_mount(&eventpoll_fs_type);
1611 error = PTR_ERR(eventpoll_mnt);
1612 if (IS_ERR(eventpoll_mnt))
1613 goto epanic;
1614
1615 DNPRINTK(3, (KERN_INFO "[%p] eventpoll: successfully initialized.\n",
1616 current));
1617 return 0;
1618
1619 epanic:
1620 panic("eventpoll_init() failed\n");
1621 }
很有趣，這個module在初始化時注冊了一個新的文件系統，叫"eventpollfs"（在eventpoll_fs_type結
構里），然后掛載此文件系統。另外創建兩個內核cache（在內核編程中，如果需要頻繁分配小塊內存，
應該創建kmem_cahe來做“內存池”）,分別用于存放struct epitem和eppoll_entry。如果以后要開發新
的文件系統，可以參考這段代碼。
現在想想epoll_create為什么會返回一個新的fd？因為它就是在這個叫做"eventpollfs"的文件系統里創建
了一個新文件！如下：
[fs/eventpoll.c-->sys_epoll_create()]
476 asmlinkage long sys_epoll_create(int size)
477 {
478 int error, fd;
479 struct inode *inode;
480 struct file *file;
481
482 DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_create(%d)\n",
483 current, size));
484
485 /* Sanity check on the size parameter */
486 error = -EINVAL;
487 if (size <= 0)
488 goto eexit_1;
489
490 /*
491 * Creates all the items needed to setup an eventpoll file. That is,
492 * a file structure, and inode and a free file descriptor.
493 */
494 error = ep_getfd(&fd, &inode, &file);
495 if (error)
496 goto eexit_1;
497
498 /* Setup the file internal data structure ( "struct eventpoll" ) */
499 error = ep_file_init(file);
500 if (error)
501 goto eexit_2;
函數很簡單，其中ep_getfd看上去是“get”，其實在第一次調用epoll_create時，它是要創建新inode、
新的file、新的fd。而ep_file_init則要創建一個struct eventpoll結構，并把它放入file-
>private_data，注意，這個private_data后面還要用到的。
看到這里，也許有人要問了，為什么epoll的開發者不做一個內核的超級大map把用戶要創建的epoll句柄
存起來，在epoll_create時返回一個指針？那似乎很直觀呀。但是，仔細看看，linux的系統調用有多少是
返回指針的？你會發現幾乎沒有！（特此強調，malloc不是系統調用，malloc調用的brk才是）因為linux
做為unix的最杰出的繼承人，它遵循了unix的一個巨大優點——一切皆文件，輸入輸出是文件、socket也
是文件，一切皆文件意味著使用這個操作系統的程序可以非常簡單，因為一切都是文件操作而已！（unix
還不是完全做到，plan 9才算）。而且使用文件系統有個好處：epoll_create返回的是一個fd，而不是該
死的指針，指針如果指錯了，你簡直沒辦法判斷，而fd則可以通過current->files->fd_array[]找到其真
偽。
epoll_create好了，該epoll_ctl了，我們略去判斷性的代碼：
[fs/eventpoll.c-->sys_epoll_ctl()]
524 asmlinkage long
525 sys_epoll_ctl(int epfd, int op, int fd, struct epoll_event __user *event)
526 {
527 int error;
528 struct file *file, *tfile;
529 struct eventpoll *ep;
530 struct epitem *epi;
531 struct epoll_event epds;
....
575 epi = ep_find(ep, tfile, fd);
576
577 error = -EINVAL;
578 switch (op) {
579 case EPOLL_CTL_ADD:
580 if (!epi) {
581 epds.events |= POLLERR | POLLHUP;
582
583 error = ep_insert(ep, &epds, tfile, fd);
584 } else
585 error = -EEXIST;
586 break;
587 case EPOLL_CTL_DEL:
588 if (epi)
589 error = ep_remove(ep, epi);
590 else
591 error = -ENOENT;
592 break;
593 case EPOLL_CTL_MOD:
594 if (epi) {
595 epds.events |= POLLERR | POLLHUP;
596 error = ep_modify(ep, epi, &epds);
597 } else
598 error = -ENOENT;
599 break;
600 }
原來就是在一個大的結構（現在先不管是什么大結構）里先ep_find，如果找到了struct epitem而用戶操
作是ADD，那么返回-EEXIST；如果是DEL，則ep_remove。如果找不到struct epitem而用戶操作是
ADD，就ep_insert創建并插入一個。很直白。那這個“大結構”是什么呢？看ep_find的調用方式，ep參數
應該是指向這個“大結構”的指針，再看ep = file->private_data，我們才明白，原來這個“大結構”就是那
個在epoll_create時創建的struct eventpoll，具體再看看ep_find的實現，發現原來是struct eventpoll
的rbr成員（struct rb_root），原來這是一個紅黑樹的根！而紅黑樹上掛的都是struct epitem。
現在清楚了，一個新創建的epoll文件帶有一個struct eventpoll結構，這個結構上再掛一個紅黑樹，而這
個紅黑樹就是每次epoll_ctl時fd存放的地方！
現在數據結構都已經清楚了，我們來看最核心的:
[fs/eventpoll.c-->sys_epoll_wait()]
627 asmlinkage long sys_epoll_wait(int epfd, struct epoll_event __user *events,
628 int maxevents, int timeout)
629 {
630 int error;
631 struct file *file;
632 struct eventpoll *ep;
633
634 DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_wait(%d, %p, %d, %d)\n",
635 current, epfd, events, maxevents, timeout));
636
637 /* The maximum number of event must be greater than zero */
638 if (maxevents <= 0)
639 return -EINVAL;
640
641 /* Verify that the area passed by the user is writeable */
642 if ((error = verify_area(VERIFY_WRITE, events, maxevents * sizeof(struct
epoll_event))))
643 goto eexit_1;
644
645 /* Get the "struct file *" for the eventpoll file */
646 error = -EBADF;
647 file = fget(epfd);
648 if (!file)
649 goto eexit_1;
650
651 /*
652 * We have to check that the file structure underneath the fd
653 * the user passed to us _is_ an eventpoll file.
654 */
655 error = -EINVAL;
656 if (!IS_FILE_EPOLL(file))
657 goto eexit_2;
658
659 /*
660 * At this point it is safe to assume that the "private_data" contains
661 * our own data structure.
662 */
663 ep = file->private_data;
664
665 /* Time to fish for events ... */
666 error = ep_poll(ep, events, maxevents, timeout);
667
668 eexit_2:
669 fput(file);
670 eexit_1:
671 DNPRINTK(3, (KERN_INFO "[%p] eventpoll: sys_epoll_wait(%d, %p, %d, %d) =
%d\n",
672 current, epfd, events, maxevents, timeout, error));
673
674 return error;
675 }
故伎重演，從file->private_data中拿到struct eventpoll，再調用ep_poll
[fs/eventpoll.c-->sys_epoll_wait()->ep_poll()]
1468 static int ep_poll(struct eventpoll *ep, struct epoll_event __user *events,
1469 int maxevents, long timeout)
1470 {
1471 int res, eavail;
1472 unsigned long flags;
1473 long jtimeout;
1474 wait_queue_t wait;
1475
1476 /*
1477 * Calculate the timeout by checking for the "infinite" value ( -1 )
1478 * and the overflow condition. The passed timeout is in milliseconds,
1479 * that why (t * HZ) / 1000.
1480 */
1481 jtimeout = timeout == -1 || timeout > (MAX_SCHEDULE_TIMEOUT - 1000) / HZ ?
1482 MAX_SCHEDULE_TIMEOUT: (timeout * HZ + 999) / 1000;
1483
1484 retry:
1485 write_lock_irqsave(&ep->lock, flags);
1486
1487 res = 0;
1488 if (list_empty(&ep->rdllist)) {
1489 /*
1490 * We don't have any available event to return to the caller.
1491 * We need to sleep here, and we will be wake up by
1492 * ep_poll_callback() when events will become available.
1493 */
1494 init_waitqueue_entry(&wait, current);
1495 add_wait_queue(&ep->wq, &wait);
1496
1497 for (;;) {
1498 /*
1499 * We don't want to sleep if the ep_poll_callback() sends us
1500 * a wakeup in between. That's why we set the task state
1501 * to TASK_INTERRUPTIBLE before doing the checks.
1502 */
1503 set_current_state(TASK_INTERRUPTIBLE);
1504 if (!list_empty(&ep->rdllist) || !jtimeout)
1505 break;
1506 if (signal_pending(current)) {
1507 res = -EINTR;
1508 break;
1509 }
1510
1511 write_unlock_irqrestore(&ep->lock, flags);
1512 jtimeout = schedule_timeout(jtimeout);
1513 write_lock_irqsave(&ep->lock, flags);
1514 }
1515 remove_wait_queue(&ep->wq, &wait);
1516
1517 set_current_state(TASK_RUNNING);
1518 }
....
又是一個大循環，不過這個大循環比poll的那個好，因為仔細一看——它居然除了睡覺和判斷ep->rdllist
是否為空以外，啥也沒做！
什么也沒做當然效率高了，但到底是誰來讓ep->rdllist不為空呢？
答案是ep_insert時設下的回調函數：
[fs/eventpoll.c-->sys_epoll_ctl()-->ep_insert()]
923 static int ep_insert(struct eventpoll *ep, struct epoll_event *event,
924 struct file *tfile, int fd)
925 {
926 int error, revents, pwake = 0;
927 unsigned long flags;
928 struct epitem *epi;
929 struct ep_pqueue epq;
930
931 error = -ENOMEM;
932 if (!(epi = EPI_MEM_ALLOC()))
933 goto eexit_1;
934
935 /* Item initialization follow here ... */
936 EP_RB_INITNODE(&epi->rbn);
937 INIT_LIST_HEAD(&epi->rdllink);
938 INIT_LIST_HEAD(&epi->fllink);
939 INIT_LIST_HEAD(&epi->txlink);
940 INIT_LIST_HEAD(&epi->pwqlist);
941 epi->ep = ep;
942 EP_SET_FFD(&epi->ffd, tfile, fd);
943 epi->event = *event;
944 atomic_set(&epi->usecnt, 1);
945 epi->nwait = 0;
946
947 /* Initialize the poll table using the queue callback */
948 epq.epi = epi;
949 init_poll_funcptr(&epq.pt, ep_ptable_queue_proc);
950
951 /*
952 * Attach the item to the poll hooks and get current event bits.
953 * We can safely use the file* here because its usage count has
954 * been increased by the caller of this function.
955 */
956 revents = tfile->f_op->poll(tfile, &epq.pt);
我們注意949行，其實就是
&(epq.pt)->qproc = ep_ptable_queue_proc;
緊接著 tfile->f_op->poll(tfile, &epq.pt)其實就是調用被監控文件（epoll里叫“target file”)的poll方
法，而這個poll其實就是調用poll_wait（還記得poll_wait嗎？每個支持poll的設備驅動程序都要調用
的），最后就是調用ep_ptable_queue_proc。這是比較難解的一個調用關系，因為不是語言級的直接調
用。
ep_insert還把struct epitem放到struct file里的f_ep_links連表里，以方便查找，struct epitem里的
fllink就是擔負這個使命的。
[fs/eventpoll.c-->ep_ptable_queue_proc()]
883 static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
884 poll_table *pt)
885 {
886 struct epitem *epi = EP_ITEM_FROM_EPQUEUE(pt);
887 struct eppoll_entry *pwq;
888
889 if (epi->nwait >= 0 && (pwq = PWQ_MEM_ALLOC())) {
890 init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
891 pwq->whead = whead;
892 pwq->base = epi;
893 add_wait_queue(whead, &pwq->wait);
894 list_add_tail(&pwq->llink, &epi->pwqlist);
895 epi->nwait++;
896 } else {
897 /* We have to signal that an error occurred */
898 epi->nwait = -1;
899 }
900 }
上面的代碼就是ep_insert中要做的最重要的事：創建struct eppoll_entry，設置其喚醒回調函數為
ep_poll_callback，然后加入設備等待隊列（注意這里的whead就是上一章所說的每個設備驅動都要帶的
等待隊列）。只有這樣，當設備就緒，喚醒等待隊列上的等待著時，ep_poll_callback就會被調用。每次
調用poll系統調用，操作系統都要把current（當前進程）掛到fd對應的所有設備的等待隊列上，可以想
象，fd多到上千的時候，這樣“掛”法很費事；而每次調用epoll_wait則沒有這么羅嗦，epoll只在epoll_ctl
時把current掛一遍（這第一遍是免不了的）并給每個fd一個命令“好了就調回調函數”，如果設備有事件
了，通過回調函數，會把fd放入rdllist，而每次調用epoll_wait就只是收集rdllist里的fd就可以了
——epoll巧妙的利用回調函數，實現了更高效的事件驅動模型。
現在我們猜也能猜出來ep_poll_callback會干什么了——肯定是把紅黑樹上的收到event的epitem（代表
每個fd）插入ep->rdllist中，這樣，當epoll_wait返回時，rdllist里就都是就緒的fd了！
[fs/eventpoll.c-->ep_poll_callback()]
1206 static int ep_poll_callback(wait_queue_t *wait, unsigned mode, int sync, void *key)
1207 {
1208 int pwake = 0;
1209 unsigned long flags;
1210 struct epitem *epi = EP_ITEM_FROM_WAIT(wait);
1211 struct eventpoll *ep = epi->ep;
1212
1213 DNPRINTK(3, (KERN_INFO "[%p] eventpoll: poll_callback(%p) epi=%p
ep=%p\n",
1214 current, epi->file, epi, ep));
1215
1216 write_lock_irqsave(&ep->lock, flags);
1217
1218 /*
1219 * If the event mask does not contain any poll(2) event, we consider the
1220 * descriptor to be disabled. This condition is likely the effect of the
1221 * EPOLLONESHOT bit that disables the descriptor when an event is received,
1222 * until the next EPOLL_CTL_MOD will be issued.
1223 */
1224 if (!(epi->event.events & ~EP_PRIVATE_BITS))
1225 goto is_disabled;
1226
1227 /* If this file is already in the ready list we exit soon */
1228 if (EP_IS_LINKED(&epi->rdllink))
1229 goto is_linked;
1230
1231 list_add_tail(&epi->rdllink, &ep->rdllist);
1232
1233 is_linked:
1234 /*
1235 * Wake up ( if active ) both the eventpoll wait list and the ->poll()
1236 * wait list.
1237 */
1238 if (waitqueue_active(&ep->wq))
1239 wake_up(&ep->wq);
1240 if (waitqueue_active(&ep->poll_wait))
1241 pwake++;
1242
1243 is_disabled:
1244 write_unlock_irqrestore(&ep->lock, flags);
1245
1246 /* We have to call this outside the lock */
1247 if (pwake)
1248 ep_poll_safewake(&psw, &ep->poll_wait);
1249
1250 return 1;
1251 }
真正重要的只有1231行的只一句，就是把struct epitem放到struct eventpoll的rdllist中去。現在我們
可以畫出epoll的核心數據結構圖了：

epoll獨有的EPOLLET

EPOLLET是epoll系統調用獨有的flag，ET就是Edge Trigger（邊緣觸發）的意思，具體含義和應用大家
可google之。有了EPOLLET，重復的事件就不會總是出來打擾程序的判斷，故而常被使用。那EPOLLET
的原理是什么呢？
上篇我們講到epoll把fd都掛上一個回調函數，當fd對應的設備有消息時，就把fd放入rdllist鏈表，這樣
epoll_wait只要檢查這個rdllist鏈表就可以知道哪些fd有事件了。我們看看ep_poll的最后幾行代碼：
[fs/eventpoll.c->ep_poll()]
1524
1525 /*
1526 * Try to transfer events to user space. In case we get 0 events and
1527 * there's still timeout left over, we go trying again in search of
1528 * more luck.
1529 */
1530 if (!res && eavail &&
1531 !(res = ep_events_transfer(ep, events, maxevents)) && jtimeout)
1532 goto retry;
1533
1534 return res;
1535 }
把rdllist里的fd拷到用戶空間，這個任務是ep_events_transfer做的：
[fs/eventpoll.c->ep_events_transfer()]
1439 static int ep_events_transfer(struct eventpoll *ep,
1440 struct epoll_event __user *events, int maxevents)
1441 {
1442 int eventcnt = 0;
1443 struct list_head txlist;
1444
1445 INIT_LIST_HEAD(&txlist);
1446
1447 /*
1448 * We need to lock this because we could be hit by
1449 * eventpoll_release_file() and epoll_ctl(EPOLL_CTL_DEL).
1450 */
1451 down_read(&ep->sem);
1452
1453 /* Collect/extract ready items */
1454 if (ep_collect_ready_items(ep, &txlist, maxevents) > 0) {
1455 /* Build result set in userspace */
1456 eventcnt = ep_send_events(ep, &txlist, events);
1457
1458 /* Reinject ready items into the ready list */
1459 ep_reinject_items(ep, &txlist);
1460 }
1461
1462 up_read(&ep->sem);
1463
1464 return eventcnt;
1465 }
代碼很少，其中ep_collect_ready_items把rdllist里的fd挪到txlist里（挪完后rdllist就空了），接著
ep_send_events把txlist里的fd拷給用戶空間，然后ep_reinject_items把一部分fd從txlist里“返還”給
rdllist以便下次還能從rdllist里發現它。
其中ep_send_events的實現：
[fs/eventpoll.c->ep_send_events()]
1337 static int ep_send_events(struct eventpoll *ep, struct list_head *txlist,
1338 struct epoll_event __user *events)
1339 {
1340 int eventcnt = 0;
1341 unsigned int revents;
1342 struct list_head *lnk;
1343 struct epitem *epi;
1344
1345 /*
1346 * We can loop without lock because this is a task private list.
1347 * The test done during the collection loop will guarantee us that
1348 * another task will not try to collect this file. Also, items
1349 * cannot vanish during the loop because we are holding "sem".
1350 */
1351 list_for_each(lnk, txlist) {
1352 epi = list_entry(lnk, struct epitem, txlink);
1353
1354 /*
1355 * Get the ready file event set. We can safely use the file
1356 * because we are holding the "sem" in read and this will
1357 * guarantee that both the file and the item will not vanish.
1358 */
1359 revents = epi->ffd.file->f_op->poll(epi->ffd.file, NULL);
1360
1361 /*
1362 * Set the return event set for the current file descriptor.
1363 * Note that only the task task was successfully able to link
1364 * the item to its "txlist" will write this field.
1365 */
1366 epi->revents = revents & epi->event.events;
1367
1368 if (epi->revents) {
1369 if (__put_user(epi->revents,
1370 &events[eventcnt].events) ||
1371 __put_user(epi->event.data,
1372 &events[eventcnt].data))
1373 return -EFAULT;
1374 if (epi->event.events & EPOLLONESHOT)
1375 epi->event.events &= EP_PRIVATE_BITS;
1376 eventcnt++;
1377 }
1378 }
1379 return eventcnt;
1380 }
這個拷貝實現其實沒什么可看的，但是請注意1359行，這個poll很狡猾，它把第二個參數置為NULL來調
用。我們先看一下設備驅動通常是怎么實現poll的：
static unsigned int scull_p_poll(struct file *filp, poll_table *wait)
{
struct scull_pipe *dev = filp->private_data;
unsigned int mask = 0;
/*
* The buffer is circular; it is considered full
* if "wp" is right behind "rp" and empty if the
* two are equal.
*/
down(&dev->sem);
poll_wait(filp, &dev->inq, wait);
poll_wait(filp, &dev->outq, wait);
if (dev->rp != dev->wp)
mask |= POLLIN | POLLRDNORM; /* readable */
if (spacefree(dev))
mask |= POLLOUT | POLLWRNORM; /* writable */
up(&dev->sem);
return mask;
}
上面這段代碼摘自《linux設備驅動程序（第三版）》，絕對經典，設備先要把current（當前進程）掛在
inq和outq兩個隊列上（這個“掛”操作是wait回調函數指針做的），然后等設備來喚醒，喚醒后就能通過
mask拿到事件掩碼了（注意那個mask參數，它就是負責拿事件掩碼的）。那如果wait為NULL，
poll_wait會做些什么呢？
[include/linux/poll.h->poll_wait]
25 static inline void poll_wait(struct file * filp, wait_queue_head_t * wait_address,
poll_table *p)
26 {
27 if (p && wait_address)
28 p->qproc(filp, wait_address, p);
29 }
喏，看見了，如果poll_table為空，什么也不做。我們倒回ep_send_events，那句標紅的poll，實際上
就是“我不想休眠，我只想拿到事件掩碼”的意思。然后再把拿到的事件掩碼拷給用戶空間。
ep_send_events完成后，就輪到ep_reinject_items了：
[fs/eventpoll.c->ep_reinject_items]
1389 static void ep_reinject_items(struct eventpoll *ep, struct list_head *txlist)
1390 {
1391 int ricnt = 0, pwake = 0;
1392 unsigned long flags;
1393 struct epitem *epi;
1394
1395 write_lock_irqsave(&ep->lock, flags);
1396
1397 while (!list_empty(txlist)) {
1398 epi = list_entry(txlist->next, struct epitem, txlink);
1399
1400 /* Unlink the current item from the transfer list */
1401 EP_LIST_DEL(&epi->txlink);
1402
1403 /*
1404 * If the item is no more linked to the interest set, we don't
1405 * have to push it inside the ready list because the following
1406 * ep_release_epitem() is going to drop it. Also, if the current
1407 * item is set to have an Edge Triggered behaviour, we don't have
1408 * to push it back either.
1409 */
1410 if (EP_RB_LINKED(&epi->rbn) && !(epi->event.events & EPOLLET) &&
1411 (epi->revents & epi->event.events) && !EP_IS_LINKED(&epi->rdllink)) {
1412 list_add_tail(&epi->rdllink, &ep->rdllist);
1413 ricnt++;
1414 }
1415 }
1416
1417 if (ricnt) {
1418 /*
1419 * Wake up ( if active ) both the eventpoll wait list and the ->poll()
1420 * wait list.
1421 */
1422 if (waitqueue_active(&ep->wq))
1423 wake_up(&ep->wq);
1424 if (waitqueue_active(&ep->poll_wait))
1425 pwake++;
1426 }
1427
1428 write_unlock_irqrestore(&ep->lock, flags);
1429
1430 /* We have to call this outside the lock */
1431 if (pwake)
1432 ep_poll_safewake(&psw, &ep->poll_wait);
1433 }
ep_reinject_items把txlist里的一部分fd又放回rdllist，那么，是把哪一部分fd放回去呢？看上面1410行
的那個判斷——是哪些“沒有標上EPOLLET”（標紅代碼）且“事件被關注”（標藍代碼）的fd被重新放回了
rdllist。那么下次epoll_wait當然會又把rdllist里的fd拿來拷給用戶了。
舉個例子。假設一個socket，只是connect，還沒有收發數據，那么它的poll事件掩碼總是有POLLOUT的
（參見上面的驅動示例），每次調用epoll_wait總是返回POLLOUT事件（比較煩），因為它的fd就總是被
放回rdllist；假如此時有人往這個socket里寫了一大堆數據，造成socket塞住（不可寫了），那么1411行
里標藍色的判斷就不成立了（沒有POLLOUT了），fd不會放回rdllist，epoll_wait將不會再返回用戶
POLLOUT事件。現在我們給這個socket加上EPOLLET，然后connect，沒有收發數據，此時，1410行標
紅的判斷又不成立了，所以epoll_wait只會返回一次POLLOUT通知給用戶（因為此fd不會再回到rdllist
了），接下來的epoll_wait都不會有任何事件通知了。

總結

以上是生活随笔為你收集整理的linux下poll和epoll内核源代码剖析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。