epoll内核源代码剖析

女爷i 2022-08-05 08:58 218阅读 0赞

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的核心数据结构图了：

![Image 1][]![Image 1][]

作者：董昊 （要转载的同学帮忙把名字和博客链接http://donghao.org/uii/带上，多谢了！)  
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都不会有任何事件通知了。

http://oldblog.donghao.org/2009/08/linuxiapolliepollaueouaeaeeio.html

[Image 1]: