select、poll、epoll之间的区别

时间 2019-11-10

标签 select poll epoll 之间区别繁體版

原文原文链接

　　select，poll，epoll都是IO多路复用的机制。I/O多路复用就经过一种机制，能够监视多个描述符，一旦某个描述符就绪（通常是读就绪或者写就绪），可以通知程序进行相应的读写操做。但select，poll，epoll本质上都是同步I/O，由于他们都须要在读写事件就绪后本身负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需本身负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。关于这三种IO多路复用的用法，前面三篇总结写的很清楚，并用服务器回射echo程序进行了测试。链接以下所示：html

select：http://www.cnblogs.com/Anker/archive/2013/08/14/3258674.htmllinux

poll：http://www.cnblogs.com/Anker/archive/2013/08/15/3261006.html程序员

epoll：http://www.cnblogs.com/Anker/archive/2013/08/17/3263780.htmlweb

　　今天对这三种IO多路复用进行对比，参考网上和书上面的资料，整理以下：数据库

一、select实现编程

select的调用过程以下所示：数组

（1）使用copy_from_user从用户空间拷贝fd_set到内核空间缓存

（2）注册回调函数__pollwaittomcat

（3）遍历全部fd，调用其对应的poll方法（对于socket，这个poll方法是sock_poll，sock_poll根据状况会调用到tcp_poll,udp_poll或者datagram_poll）安全

（4）以tcp_poll为例，其核心实现就是__pollwait，也就是上面注册的回调函数。

（5）__pollwait的主要工做就是把current（当前进程）挂到设备的等待队列中，不一样的设备有不一样的等待队列，对于tcp_poll来讲，其等待队列是sk->sk_sleep（注意把进程挂到等待队列中并不表明进程已经睡眠了）。在设备收到一条消息（网络设备）或填写完文件数据（磁盘设备）后，会唤醒设备等待队列上睡眠的进程，这时current便被唤醒了。

（6）poll方法返回时会返回一个描述读写操做是否就绪的mask掩码，根据这个mask掩码给fd_set赋值。

（7）若是遍历完全部的fd，尚未返回一个可读写的mask掩码，则会调用schedule_timeout是调用select的进程（也就是current）进入睡眠。当设备驱动发生自身资源可读写后，会唤醒其等待队列上睡眠的进程。若是超过必定的超时时间（schedule_timeout指定），仍是没人唤醒，则调用select的进程会从新被唤醒得到CPU，进而从新遍历fd，判断有没有就绪的fd。

（8）把fd_set从内核空间拷贝到用户空间。

总结：

select的几大缺点：

（1）每次调用select，都须要把fd集合从用户态拷贝到内核态，这个开销在fd不少时会很大

（2）同时每次调用select都须要在内核遍历传递进来的全部fd，这个开销在fd不少时也很大

（3）select支持的文件描述符数量过小了，默认是1024

2 poll实现

　　poll的实现和select很是类似，只是描述fd集合的方式不一样，poll使用pollfd结构而不是select的fd_set结构，其余的都差很少。

关于select和poll的实现分析，能够参考下面几篇博文：

http://blog.csdn.net/lizhiguo0532/article/details/6568964#comments

http://blog.csdn.net/lizhiguo0532/article/details/6568968

http://blog.csdn.net/lizhiguo0532/article/details/6568969

http://www.ibm.com/developerworks/cn/linux/l-cn-edntwk/index.html?ca=drs-

三、epoll

　　epoll既然是对select和poll的改进，就应该能避免上述的三个缺点。那epoll都是怎么解决的呢？在此以前，咱们先看一下epoll和select和poll的调用接口上的不一样，select和poll都只提供了一个函数——select或者poll函数。而epoll提供了三个函数，epoll_create,epoll_ctl和epoll_wait，epoll_create是建立一个epoll句柄；epoll_ctl是注册要监听的事件类型；epoll_wait则是等待事件的产生。

　　对于第一个缺点，epoll的解决方案在epoll_ctl函数中。每次注册新的事件到epoll句柄中时（在epoll_ctl中指定EPOLL_CTL_ADD），会把全部的fd拷贝进内核，而不是在epoll_wait的时候重复拷贝。epoll保证了每一个fd在整个过程当中只会拷贝一次。

　　对于第二个缺点，epoll的解决方案不像select或poll同样每次都把current轮流加入fd对应的设备等待队列中，而只在epoll_ctl时把current挂一遍（这一遍必不可少）并为每一个fd指定一个回调函数，当设备就绪，唤醒等待队列上的等待者时，就会调用这个回调函数，而这个回调函数会把就绪的fd加入一个就绪链表）。epoll_wait的工做实际上就是在这个就绪链表中查看有没有就绪的fd（利用schedule_timeout()实现睡一会，判断一会的效果，和select实现中的第7步是相似的）。

　　对于第三个缺点，epoll没有这个限制，它所支持的FD上限是最大能够打开文件的数目，这个数字通常远大于2048,举个例子,在1GB内存的机器上大约是10万左右，具体数目能够cat /proc/sys/fs/file-max察看,通常来讲这个数目和系统内存关系很大。

总结：

（1）select，poll实现须要本身不断轮询全部fd集合，直到设备就绪，期间可能要睡眠和唤醒屡次交替。而epoll其实也须要调用epoll_wait不断轮询就绪链表，期间也可能屡次睡眠和唤醒交替，可是它是设备就绪时，调用回调函数，把就绪fd放入就绪链表中，并唤醒在epoll_wait中进入睡眠的进程。虽然都要睡眠和交替，可是select和poll在“醒着”的时候要遍历整个fd集合，而epoll在“醒着”的时候只要判断一下就绪链表是否为空就好了，这节省了大量的CPU时间。这就是回调机制带来的性能提高。

（2）select，poll每次调用都要把fd集合从用户态往内核态拷贝一次，而且要把current往设备等待队列中挂一次，而epoll只要一次拷贝，并且把current往等待队列上挂也只挂一次（在epoll_wait的开始，注意这里的等待队列并非设备等待队列，只是一个epoll内部定义的等待队列）。这也能节省很多的开销。

参考资料：

http://www.cnblogs.com/apprentice89/archive/2013/05/09/3070051.html

http://www.linuxidc.com/Linux/2012-05/59873p3.htm

http://xingyunbaijunwei.blog.163.com/blog/static/76538067201241685556302/

http://blog.csdn.net/kkxgx/article/details/7717125

select(poll)系统调用实现解析

上层要能使用select()和poll()系统调用来监测某个设备文件描述符，那么就必须实现这个设备驱动程序中struct file_operation结构体的poll函数，为何？

由于这两个系统调用最终都会调用驱动程序中的poll函数来初始化一个等待队列项, 而后将其加入到驱动程序中的等待队列头，这样就能够在硬件可读写的时候wake up这个等待队列头，而后等待(能够是多个)同一个硬件设备可读写事件的进程都将被唤醒。

(这个等待队列头能够包含多个等待队列项，这些不一样的等待队列项是由不一样的应用程序调用select或者poll来监测同一个硬件设备的时候调用file_operation的poll函数初始化填充的)。

下面就以select系统调用分析具体实现，源码路径：fs/select.c。

1、 select()系统调用代码走读

调用顺序以下：sys_select() à core_sys_select() à do_select() à fop->poll()

SYSCALL_DEFINE5(select, int, n, fd_set __user *, inp, fd_set __user *, outp,

fd_set __user *, exp, struct timeval __user *, tvp)

{

struct timespec end_time, *to = NULL;

struct timeval tv;

int ret;

if (tvp) {// 若是超时值非NULL

if (copy_from_user(&tv, tvp, sizeof(tv))) // 从用户空间取数据到内核空间

return -EFAULT;

to = &end_time;

// 获得timespec格式的将来超时时间

if (poll_select_set_timeout(to,

tv.tv_sec + (tv.tv_usec / USEC_PER_SEC),

(tv.tv_usec % USEC_PER_SEC) * NSEC_PER_USEC))

return -EINVAL;

}

ret = core_sys_select(n, inp, outp, exp, to); // 关键函数

ret = poll_select_copy_remaining(&end_time, tvp, 1, ret);

/*若是有超时值, 并拷贝离超时时刻还剩的时间到用户空间的timeval中*/

return ret; // 返回就绪的文件描述符的个数

}

==================================================================

core_sys_select()函数解析

int core_sys_select(int n, fd_set __user *inp, fd_set __user *outp,

fd_set __user *exp, struct timespec *end_time)

{

fd_set_bits fds;

/**

typedef struct {

unsigned long *in, *out, *ex;

unsigned long *res_in, *res_out, *res_ex;

} fd_set_bits;

这个结构体中定义的全是指针，这些指针都是用来指向描述符集合的。

**/

void *bits;

int ret, max_fds;

unsigned int size;

struct fdtable *fdt;

/* Allocate small arguments on the stack to save memory and be faster */

long stack_fds[SELECT_STACK_ALLOC/sizeof(long)];

// 256/32 = 8, stack中分配的空间

/**

@ include/linux/poll.h

#define FRONTEND_STACK_ALLOC 256

#define SELECT_STACK_ALLOC FRONTEND_STACK_ALLOC

**/

ret = -EINVAL;

if (n < 0)

goto out_nofds;

/* max_fds can increase, so grab it once to avoid race */

rcu_read_lock();

fdt = files_fdtable(current->files); // RCU ref, 获取当前进程的文件描述符表

max_fds = fdt->max_fds;

rcu_read_unlock();

if (n > max_fds)// 若是传入的n大于当前进程最大的文件描述符，给予修正

n = max_fds;

* We need 6 bitmaps (in/out/ex for both incoming and outgoing),

* since we used fdset we need to allocate memory in units of

* long-words.

size = FDS_BYTES(n);

// 以一个文件描述符占一bit来计算，传递进来的这些fd_set须要用掉多少个字

bits = stack_fds;

if (size > sizeof(stack_fds) / 6) {

// 除6，为何?由于每一个文件描述符须要6个bitmaps

/* Not enough space in on-stack array; must use kmalloc */

ret = -ENOMEM;

bits = kmalloc(6 * size, GFP_KERNEL); // stack中分配的过小，直接kmalloc

if (!bits)

goto out_nofds;

}

// 这里就能够明显看出struct fd_set_bits结构体的用处了。

fds.in = bits;

fds.out = bits + size;

fds.ex = bits + 2*size;

fds.res_in = bits + 3*size;

fds.res_out = bits + 4*size;

fds.res_ex = bits + 5*size;

// get_fd_set仅仅调用copy_from_user从用户空间拷贝了fd_set

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); // 对这些存放返回状态的字段清0

zero_fd_set(n, fds.res_out);

zero_fd_set(n, fds.res_ex);

ret = do_select(n, &fds, end_time); // 关键函数，完成主要的工做

if (ret < 0) // 有错误

goto out;

if (!ret) { // 超时返回，无设备就绪

ret = -ERESTARTNOHAND;

if (signal_pending(current))

goto out;

ret = 0;

}

// 把结果集,拷贝回用户空间

if (set_fd_set(n, inp, fds.res_in) ||

set_fd_set(n, outp, fds.res_out) ||

set_fd_set(n, exp, fds.res_ex))

ret = -EFAULT;

out:

if (bits != stack_fds)

kfree(bits); // 若是有申请空间，那么释放fds对应的空间

out_nofds:

return ret; // 返回就绪的文件描述符的个数

}

++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

do_select()函数解析：

int do_select(int n, fd_set_bits *fds, struct timespec *end_time)

{

ktime_t expire, *to = NULL;

struct poll_wqueues table;

poll_table *wait;

int retval, i, timed_out = 0;

unsigned long slack = 0;

rcu_read_lock();

// 根据已经设置好的fd位图检查用户打开的fd, 要求对应fd必须打开, 而且返回

// 最大的fd。

retval = max_select_fd(n, fds);

rcu_read_unlock();

if (retval < 0)

return retval;

n = retval;

// 一些重要的初始化:

// poll_wqueues.poll_table.qproc函数指针初始化，该函数是驱动程序中poll函数实

// 现中必需要调用的poll_wait()中使用的函数。

poll_initwait(&table);

wait = &table.pt;

if (end_time && !end_time->tv_sec && !end_time->tv_nsec) {

wait = NULL;

timed_out = 1; // 若是系统调用带进来的超时时间为0，那么设置

// timed_out = 1，表示不阻塞，直接返回。

}

if (end_time && !timed_out)

slack = estimate_accuracy(end_time); // 超时时间转换

retval = 0;

for (;;) {

unsigned long *rinp, *routp, *rexp, *inp, *outp, *exp;

inp = fds->in; outp = fds->out; exp = fds->ex;

rinp = fds->res_in; routp = fds->res_out; rexp = fds->res_ex;

// 全部n个fd的循环

for (i = 0; i < n; ++rinp, ++routp, ++rexp) {

unsigned long in, out, ex, all_bits, bit = 1, mask, j;

unsigned long res_in = 0, res_out = 0, res_ex = 0;

const struct file_operations *f_op = NULL;

struct file *file = NULL;

// 先取出当前循环周期中的32个文件描述符对应的bitmaps

in = *inp++; out = *outp++; ex = *exp++;

all_bits = in | out | ex; // 组合一下，有的fd可能只监测读，或者写，

// 或者e rr，或者同时都监测

if (all_bits == 0) { // 这32个描述符没有任何状态被监测，就跳入

// 下一个32个fd的循环中

i += __NFDBITS; //每32个文件描述符一个循环，正好一个long型数

continue;

}

// 本次32个fd的循环中有须要监测的状态存在

for (j = 0; j < __NFDBITS; ++j, ++i, bit <<= 1) {// 初始bit = 1

int fput_needed;

if (i >= n) // i用来检测是否超出了最大待监测的fd

break;

if (!(bit & all_bits))

continue; // bit每次循环后左移一位的做用在这里，用来

// 跳过没有状态监测的fd

file = fget_light(i, &fput_needed); // 获得file结构指针，并增长

// 引用计数字段f_count

if (file) { // 若是file存在

f_op = file->f_op;

mask = DEFAULT_POLLMASK;

if (f_op && f_op->poll) {

wait_key_set(wait, in, out, bit);// 设置当前fd待监测

// 的事件掩码

mask = (*f_op->poll)(file, wait);

调用驱动程序中的poll函数，以evdev驱动中的

evdev_poll()为例该函数会调用函数poll_wait(file, &evdev->wait, wait)，继续调用__pollwait()回调来分配一个poll_table_entry结构体，该结构体有一个内嵌的等待队列项，设置好wake时调用的回调函数后将其添加到驱动程序中的等待队列头中。

}

fput_light(file, fput_needed);

// 释放file结构指针，实际就是减少他的一个引用

计数字段f_count。

// mask是每个fop->poll()程序返回的设备状态掩码。

if ((mask & POLLIN_SET) && (in & bit)) {

res_in |= bit; // fd对应的设备可读

retval++;

wait = NULL; // 后续有用，避免重复执行__pollwait()

}

if ((mask & POLLOUT_SET) && (out & bit)) {

res_out |= bit; // fd对应的设备可写

retval++;

wait = NULL;

}

if ((mask & POLLEX_SET) && (ex & bit)) {

res_ex |= bit;

retval++;

wait = NULL;

}

// 根据poll的结果写回到输出位图里,返回给上级函数

if (res_in)

*rinp = res_in;

if (res_out)

*routp = res_out;

if (res_ex)

*rexp = res_ex;

这里的目的纯粹是为了增长一个抢占点。

在支持抢占式调度的内核中（定义了CONFIG_PREEMPT），

cond_resched是空操做。

cond_resched();

}

wait = NULL; // 后续有用，避免重复执行__pollwait()

if (retval || timed_out || signal_pending(current))

break;

if (table.error) {

retval = table.error;

break;

}

/*跳出这个大循环的条件有: 有设备就绪或有异常(retval!=0), 超时(timed_out

= 1), 或者有停止信号出现*/

* If this is the first loop and we have a timeout

* given, then we convert to ktime_t and set the to

* pointer to the expiry value.

if (end_time && !to) {

expire = timespec_to_ktime(*end_time);

to = &expire;

}

// 第一次循环中，当前用户进程从这里进入休眠，

// 上面传下来的超时时间只是为了用在睡眠超时这里而已

// 超时，poll_schedule_timeout()返回0；被唤醒时返回-EINTR

if (!poll_schedule_timeout(&table, TASK_INTERRUPTIBLE,

to, slack))

timed_out = 1; /* 超时后，将其设置成1，方便后面退出循环返回到上层 */

}

// 清理各个驱动程序的等待队列头，同时释放掉全部空出来

// 的page页(poll_table_entry)

poll_freewait(&table);

return retval; // 返回就绪的文件描述符的个数

}

++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

==================================================================

2、重要结构体之间关系

比较重要的结构体由四个：struct poll_wqueues、struct poll_table_page、struct poll_table_entry、struct poll_table_struct，这小节重点讨论前三个，后面一个留到后面小节。

2.一、结构体关系

每个调用select()系统调用的应用进程都会存在一个struct poll_weueues结构体，用来统一辅佐实现这个进程中全部待监测的fd的轮询工做，后面全部的工做和都这个结构体有关，因此它很是重要。

struct poll_wqueues {

poll_table pt;

struct poll_table_page *table;

struct task_struct *polling_task; //保存当前调用select的用户进程struct task_struct结构体

int triggered; // 当前用户进程被唤醒后置成1，以避免该进程接着进睡眠

int error; // 错误码

int inline_index; // 数组inline_entries的引用下标

struct poll_table_entry inline_entries[N_INLINE_POLL_ENTRIES];

};

实际上结构体poll_wqueues内嵌的poll_table_entry数组inline_entries[] 的大小是有限：

#define MAX_STACK_ALLOC 832

#define FRONTEND_STACK_ALLOC 256

#define WQUEUES_STACK_ALLOC

(MAX_STACK_ALLOC - FRONTEND_STACK_ALLOC)

#define N_INLINE_POLL_ENTRIES

(WQUEUES_STACK_ALLOC / sizeof(struct poll_table_entry))

若是空间不够用，后续会动态申请物理内存页以链表的形式挂载poll_wqueues.table上统一管理。接下来的两个结构体就和这项内容密切相关：

struct poll_table_page { // 申请的物理页都会将起始地址强制转换成该结构体指针

struct poll_table_page * next; // 指向下一个申请的物理页

struct poll_table_entry * entry; // 指向entries[]中首个待分配(空的) poll_table_entry地址

struct poll_table_entry entries[0]; // 该page页后面剩余的空间都是待分配的

// poll_table_entry结构体

};

对每个fd调用fop->poll() à poll_wait() à __pollwait()都会先从poll_wqueues. inline_entries[]中分配一个poll_table_entry结构体，直到该数组用完才会分配物理页挂在链表指针poll_wqueues.table上而后才会分配一个poll_table_entry结构体。具体用来作什么？这里先简单说说，__pollwait()函数调用时须要3个参数，第一个是特定fd对应的file结构体指针，第二个就是特定fd对应的硬件驱动程序中的等待队列头指针，第3个是调用select()的应用进程中poll_wqueues结构体的poll_table项(该进程监测的全部fd调用fop->poll函数都用这一个poll_table结构体)。

struct poll_table_entry {

struct file *filp; // 指向特定fd对应的file结构体;

unsigned long key; // 等待特定fd对应硬件设备的事件掩码，如POLLIN、

// POLLOUT、POLLERR;

wait_queue_t wait; // 表明调用select()的应用进程，等待在fd对应设备的特定事件

// (读或者写)的等待队列头上，的等待队列项;

wait_queue_head_t *wait_address; // 设备驱动程序中特定事件的等待队列头；

};

总结一下几点：

1. 特定的硬件设备驱动程序的事件等待队列头是有限个数的，一般是有读事件和写事件的等待队列头；

2. 而一个调用了select()的应用进程只存在一个poll_wqueues结构体；

3. 该应用程序能够有多个fd在进行同时监测其各自的事件发生，但该应用进程中每个fd有多少个poll_table_entry存在，那就取决于fd对应的驱动程序中有几个事件等待队列头了，也就是说，一般驱动程序的poll函数中须要对每个事件的等待队列头调用poll_wait()函数。好比，若是有读写两个等待队列头，那么就在这个应用进程中存在两个poll_table_entry结构体，在这两个事件的等待队列头中分别将两个等待队列项加入；

4. 若是有多个应用进程使用selcet()方式同时在访问同一个硬件设备，此时硬件驱动程序中加入等待队列头中的等待队列项对每个应用程序来讲都是相同数量的(一个事件等待队列头一个，数量取决于事件等待队列头的个数)。

2.二、注意项

对于第3点中，若是驱动程序中有多个事件等待队列头，那么在这种状况下，写设备驱动程序时就要特别当心了，特别是设备有事件就绪而后唤醒等待队列头中全部应用进程的时候须要使用另外的宏，唤醒使用的宏和函数源码见include/linux/wait.h:

在这以前看一看__pollwait()函数中填充poll_table_entry结构体时注册的唤醒回调函数pollwake()。

static int pollwake(wait_queue_t *wait, unsigned mode, int sync, void *key)

{

struct poll_table_entry *entry;

entry = container_of(wait, struct poll_table_entry, wait);

// 取得poll_table_entry结构体指针

if (key && !((unsigned long)key & entry->key))

/*这里的条件判断相当重要，避免应用进程被误唤醒，什么意思？*/

return 0;

return __pollwake(wait, mode, sync, key);

}

到底什么状况下会出现误唤醒呢？固然是有先决条件的。

驱动程序中存在多个事件的等待队列头，而且应用程序中只监测了该硬件的某几项事件，好比，驱动中有读写等待队里头，但应用程序中只有在监测读事件的发生。这种状况下，写驱动程序时候，若是唤醒函数用法不当，就会引发误唤醒的状况。

先来看一看咱们熟知的一些唤醒函数吧！

#define wake_up(x) __wake_up(x, TASK_NORMAL, 1, NULL)

#define wake_up_interruptible(x) __wake_up(x, TASK_INTERRUPTIBLE, 1, NULL)

void __wake_up(wait_queue_head_t *q, unsigned int mode, int nr, void *key);

注意到这个key了吗？一般咱们调用唤醒函数时key为NULL，很容易看出，若是咱们在这种状况下，使用上面两种唤醒函数，那么上面红色字体的判断条件一直都会是假，那么也就是说，只要设备的几类事件之一有发生，无论应用程序中是否对其有监测，都会在这里顺利经过将应用程序唤醒，唤醒后，从新调用一遍fop->poll(注意：第一次和第二次调用该函数时少作了一件事，后面代码详解)函数，获得设备事件掩码。假如刚好在此次唤醒后的一轮调用fop->poll()函数的循环中，没有其余硬件设备就绪，那么可想而知，从源码上看，do_select()会直接返回0。

// mask是每个fop->poll()程序返回的设备状态掩码。

if ((mask & POLLIN_SET) && (in & bit)) {

res_in |= bit; // fd对应的设备可读

retval++;

wait = NULL; // 后续有用，避免重复执行__pollwait()

}

(in & bit)这个条件就是用来确认用户程序有没有让你监测该事件的, 若是没有retval仍然是0，基于前面的假设，那么do_select()返回给上层的也是0。那又假如应用程序中调用select()的时候没有传入超时值，那岂不是和事实不相符合吗？没有传递超时值，那么select()函数会一直阻塞直到至少有1个fd的状态就绪。

因此在这种状况下，设备驱动中唤醒函数须要用另外的一组：

#define wake_up_poll(x, m) /

__wake_up(x, TASK_NORMAL, 1, (void *) (m))

#define wake_up_interruptible_poll(x, m) /

__wake_up(x, TASK_INTERRUPTIBLE, 1, (void *) (m))

这里的m值，应该和设备发生的事件相符合。设置poll_table_entry结构体的key项的函数是：

#define POLLIN_SET

(POLLRDNORM | POLLRDBAND | POLLIN | POLLHUP | POLLERR)

#define POLLOUT_SET (POLLWRBAND | POLLWRNORM | POLLOUT | POLLERR)

#define POLLEX_SET (POLLPRI)

static inline void wait_key_set(poll_table *wait, unsigned long in,

unsigned long out, unsigned long bit)

{

if (wait) {

wait->key = POLLEX_SET;

if (in & bit)

wait->key |= POLLIN_SET;

if (out & bit)

wait->key |= POLLOUT_SET;

}

这里的m值，能够参考上面的宏来设置，注意传递的不是key的指针，而就是其值自己，只不过在wake_up_poll()到pollwake()的传递过程当中是将其转换成指针的。

若是唤醒函数使用后面一组的话，再加上合理设置key值，我相信pollwake()函数中的if必定会严格把关，不让应用程序没有监测的事件唤醒应用进程，从而避免了发生误唤醒

3、讨论几个细节

3.一、fop->poll()

fop->poll()函数就是file_operations结构体中的poll函数指针项，该函数相信不少人都知道怎么写，网上大把的文章介绍其模板，可是为何要那么写，并且它作了什么具体的事情？本小节来揭开其神秘面纱，先贴一个模板上来。

static unsigned int XXX_poll(struct file *filp, poll_table *wait)

{

unsigned int mask = 0;

struct XXX_dev *dev = filp->private_data;

...

poll_wait(filp, &dev->r_wait, wait);

poll_wait(filp ,&dev->w_wait, wait);

if(...)//读就绪

｛

mask |= POLLIN | POLLRDNORM;

｝

if(...)//写就绪

｛

mask |= POLLOUT | POLLRDNORM;

｝

return mask;

}

poll_wait()只因有wait字样，常常给人误会，觉得它会停在这里等，也就是常说的阻塞。不过咱们反过来想一想，要是同一个应用进程同时监测多个fd，那么没一个fd调用xxx_poll的时候都阻塞在这里，那和不使用select()又有何区别呢？都会阻塞在当个硬件上而耽误了被的设备就绪事件的读取。

其实，这个poll_wait()函数所作的工做挺简单，就是添加一个等待等待队列项到poll_wait ()函数传递进去的第二个参数，其表明的是驱动程序中的特定事件的等待队列头。

下面以字符设备evdev为例，文件drivers/input/evdev.c。

static unsigned int evdev_poll(struct file *file, poll_table *wait)

{

struct evdev_client *client = file->private_data;

struct evdev *evdev = client->evdev;

poll_wait(file, &evdev->wait, wait);

return ((client->head == client->tail) ? 0 : (POLLIN | POLLRDNORM)) |

(evdev->exist ? 0 : (POLLHUP | POLLERR));

}

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

poll_table *p)

{

if (p && wait_address)

p->qproc(filp, wait_address, p);

}

其中wait_address是驱动程序须要提供的等待队列头，来容纳后续等待该硬件设备就绪的进程对应的等待队列项。关键结构体poll_table, 这个结构体名字也取的很差，什么table？其实其中没有table的一丁点概念，容易让人误解呀！

typedef void (*poll_queue_proc)(struct file *, wait_queue_head_t *, struct poll_table_struct *);

typedef struct poll_table_struct {

poll_queue_proc qproc;

unsigned long key;

} poll_table;

fop->poll()函数的poll_table参数是从哪里传进来的？好生阅读过代码就能够发现，do_select()函数中存在一个结构体struct poll_wqueues，其内嵌了一个poll_table的结构体，因此在后面的大循环中依次调用各个fd的fop->poll()传递的poll_table参数都是poll_wqueues.poll_table。

poll_table结构体的定义其实蛮简单，就一个函数指针，一个key值。这个函数指针在整个select过程当中一直不变，而key则会根据不一样的fd的监测要求而变化。

qproc函数初始化在函数do_select()àpoll_initwait()àinit_poll_funcptr(&pwq->pt, __pollwait)中实现，回调函数就是__pollwait()。

int do_select(int n, fd_set_bits *fds, struct timespec *end_time)

{

struct poll_wqueues table;

…

poll_initwait(&table);

…

}

void poll_initwait(struct poll_wqueues *pwq)

{

init_poll_funcptr(&pwq->pt, __pollwait);

…

}

static inline void init_poll_funcptr(poll_table *pt, poll_queue_proc qproc)

{

pt->qproc = qproc;

pt->key = ~0UL; /* all events enabled */

}

/* Add a new entry */

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

{

struct poll_wqueues *pwq = container_of(p, struct poll_wqueues, pt);

struct poll_table_entry *entry = poll_get_entry(pwq);

if (!entry)

return;

get_file(filp);

entry->filp = filp; // 保存对应的file结构体

entry->wait_address = wait_address; // 保存来自设备驱动程序的等待队列头

entry->key = p->key; // 保存对该fd关心的事件掩码

init_waitqueue_func_entry(&entry->wait, pollwake);

// 初始化等待队列项，pollwake是唤醒该等待队列项时候调用的函数

entry->wait.private = pwq;

// 将poll_wqueues做为该等待队列项的私有数据，后面使用

add_wait_queue(wait_address, &entry->wait);

// 将该等待队列项添加到从驱动程序中传递过来的等待队列头中去。

}

该函数首先经过container_of宏来获得结构体poll_wqueues的地址，而后调用poll_get_entry()函数来得到一个poll_table_entry结构体，这个结构体是用来链接驱动和应用进程的关键结构体，其实联系很简单，这个结构体中内嵌了一个等待队列项wait_queue_t，和一个等待队列头 wait_queue_head_t，它就是驱动程序中定义的等待队列头，应用进程就是在这里保存了每个硬件设备驱动程序中的等待队列头(固然每个fd都有一个poll_table_entry结构体)。

很容易想到的是，若是这个设备在别的应用程序中也有使用，又刚好别的应用进程中也是用select()来访问该硬件设备，那么在另一个应用进程的同一个地方也会调用一样的函数来初始化一个poll_table_entry结构体，而后将这个结构体中内嵌的等待队列项添加到同一份驱动程序的等待队列头中。此后，若是设备就绪了，那么驱动程序中将会唤醒这个对于等待队列头中全部的等待队列项(也就是等待在该设备上的全部应用进程，全部等待的应用进程将会获得同一份数据)。

上面红色字体的语句保存了一个应用程序select一个fd的硬件设备时候的最全的信息，方便在设备就绪的时候容易获得对应的数据。这里的entry->key值就是为了防止第二节中描述的误唤醒而准备的。设置这个key值的地方在函数do_select()中。以下：

if (file) {

f_op = file->f_op;

mask = DEFAULT_POLLMASK;

if (f_op && f_op->poll) {

wait_key_set(wait, in, out, bit); // 见第二节 mask = (*f_op->poll)(file, wait);

}

fop->poll()函数的返回值都是有规定的，例如函数evdev_poll()中的返回值：

return ((client->head == client->tail) ? 0 : (POLLIN | POLLRDNORM)) |

(evdev->exist ? 0 : (POLLHUP | POLLERR));

会根据驱动程序中特定的buffer队列标志，来返回设备状态。这里的判断条件是读循环buffer的头尾指针是否相等：client->head == client->tail。

3.二、poll_wait()函数在select()睡眠先后调用的差别

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

poll_table *p)

{

if (p && wait_address)

p->qproc(filp, wait_address, p);

}

这里有一个if条件判断，若是驱动程序中没有提供等待队列头wait_address，那么将不会往下执行p->qproc(__pollwait())，也就是不会将表明当前应用进程的等待队列项添加进驱动程序中对应的等待队列头中。也就是说，若是应用程序刚好用select来监测这个fd的这个等待队列头对应的事件时，是永远也得不到这个设备的就绪或者错误状态的。

若是select()中调用fop->poll()时传递进来的poll_table是NULL，一般状况下，只要在应用层传递进来的超时时间结构体值不为0，哪怕这个结构体指针你传递NULL，那么在函数do_select()中第一次睡眠以前的那次全部fd的大循环中调用fop->poll()函数传递的poll_table是绝对不会为NULL的，可是第一次睡眠唤醒以后的又一次全部fd的大循环中再次调用fop->poll()函数时，此时传递的poll_table是NULL，可想而知，这一次只是检查fop->poll()的返回状态值而已。最后若是从上层调用select时传递的超时值结构体赋值成0，那么do_select()函数的只会调用一次全部fd的大循环，以后再也不进入睡眠，直接返回0给上层，基本上这种状况是没有获得任何有用的状态。

为了不应用进程被唤醒以后再次调用pollwait()的时候重复地调用函数__pollwait()，那么在传递poll_table结构体指针的时候，在睡眠以前保证其为有效地址，而在唤醒以后保证传入的poll_table地址是NULL，由于在唤醒以后，再次调用fop->poll()的做用只是为了再次检查设备的事件状态而已。具体详见代码。

3.三、唤醒应用进程

第二节中已经讨论过驱动程序唤醒进程的一点注意项，但这里再次介绍睡眠唤醒的整个流程。

睡眠是调用函数poll_schedule_timeout()来实现：

int poll_schedule_timeout(struct poll_wqueues *pwq, int state,

ktime_t *expires, unsigned long slack)

{

int rc = -EINTR;

set_current_state(state);

if (!pwq->triggered) // 这个triggered在何时被置1的呢?只要有一个fd

// 对应的设备将当前应用进程唤醒后将会把它设置成1

rc = schedule_hrtimeout_range(expires, slack, HRTIMER_MODE_ABS);

__set_current_state(TASK_RUNNING);

set_mb(pwq->triggered, 0);

return rc;

}

唤醒的话会调用函数pollwake():

static int pollwake(wait_queue_t *wait, unsigned mode, int sync, void *key)

{

struct poll_table_entry *entry;

entry = container_of(wait, struct poll_table_entry, wait);

if (key && !((unsigned long)key & entry->key))

return 0;

return __pollwake(wait, mode, sync, key);

}

static int __pollwake(wait_queue_t *wait, unsigned mode, int sync, void *key)

{

struct poll_wqueues *pwq = wait->private;

DECLARE_WAITQUEUE(dummy_wait, pwq->polling_task);

* Although this function is called under waitqueue lock, LOCK

* doesn't imply write barrier and the users expect write

* barrier semantics on wakeup functions. The following

* smp_wmb() is equivalent to smp_wmb() in try_to_wake_up()

* and is paired with set_mb() in poll_schedule_timeout.

smp_wmb();

pwq->triggered = 1;

// select()用户进程只要有被唤醒过，就不可能再次进入睡眠，由于这个标志在睡眠的时候有用

return default_wake_function(&dummy_wait, mode, sync, key);

// 默认通用的唤醒函数

}

参考网址：

1. http://blogold.chinaunix.net/u2/60011/showart_1334783.html

http://yuanbor.blog.163.com/blog/static/56674620201051134748647/

http://www.cnblogs.com/hanyan225/archive/2010/10/13/1850497.html

http://hi.baidu.com/operationsystem/blog/item/208eab9821da8f0e6f068cea.html

2. fs/select.c

drivers/input/evdev.c

include/linux/poll.h

include/linux/wait.h

kernel/wait.c

使用事件驱动模型实现高效稳定的网络服务器程序

前言

事件驱动为广大的程序员所熟悉，其最为人津津乐道的是在图形化界面编程中的应用；事实上，在网络编程中事件驱动也被普遍使用，并大规模部署在高链接数高吞吐量的服务器程序中，如 http 服务器程序、ftp 服务器程序等。相比于传统的网络编程方式，事件驱动可以极大的下降资源占用，增大服务接待能力，并提升网络传输效率。

关于本文说起的服务器模型，搜索网络能够查阅到不少的实现代码，因此，本文将不拘泥于源代码的陈列与分析，而侧重模型的介绍和比较。使用 libev 事件驱动库的服务器模型将给出实现代码。

本文涉及到线程 / 时间图例，只为代表线程在各个 IO 上确实存在阻塞时延，但并不保证时延比例的正确性和 IO 执行前后的正确性；另外，本文所说起到的接口也只是笔者熟悉的 Unix/Linux 接口，并未推荐 Windows 接口，读者能够自行查阅对应的 Windows 接口。

阻塞型的网络编程接口

几乎全部的程序员第一次接触到的网络编程都是从 listen()、send()、recv() 等接口开始的。使用这些接口能够很方便的构建服务器 / 客户机的模型。

咱们假设但愿创建一个简单的服务器程序，实现向单个客户机提供相似于“一问一答”的内容服务。

图 1. 简单的一问一答的服务器 / 客户机模型

咱们注意到，大部分的 socket 接口都是阻塞型的。所谓阻塞型接口是指系统调用（通常是 IO 接口）不返回调用结果并让当前线程一直阻塞，只有当该系统调用得到结果或者超时出错时才返回。

实际上，除非特别指定，几乎全部的 IO 接口 ( 包括 socket 接口 ) 都是阻塞型的。这给网络编程带来了一个很大的问题，如在调用 send() 的同时，线程将被阻塞，在此期间，线程将没法执行任何运算或响应任何的网络请求。这给多客户机、多业务逻辑的网络编程带来了挑战。这时，不少程序员可能会选择多线程的方式来解决这个问题。

多线程的服务器程序

应对多客户机的网络应用，最简单的解决方式是在服务器端使用多线程（或多进程）。多线程（或多进程）的目的是让每一个链接都拥有独立的线程（或进程），这样任何一个链接的阻塞都不会影响其余的链接。

具体使用多进程仍是多线程，并无一个特定的模式。传统意义上，进程的开销要远远大于线程，因此，若是须要同时为较多的客户机提供服务，则不推荐使用多进程；若是单个服务执行体须要消耗较多的 CPU 资源，譬如须要进行大规模或长时间的数据运算或文件访问，则进程较为安全。一般，使用 pthread_create () 建立新线程，fork() 建立新进程。

咱们假设对上述的服务器 / 客户机模型，提出更高的要求，即让服务器同时为多个客户机提供一问一答的服务。因而有了以下的模型。

图 2. 多线程的服务器模型

在上述的线程 / 时间图例中，主线程持续等待客户端的链接请求，若是有链接，则建立新线程，并在新线程中提供为前例一样的问答服务。

不少初学者可能不明白为什么一个 socket 能够 accept 屡次。实际上，socket 的设计者可能特地为多客户机的状况留下了伏笔，让 accept() 可以返回一个新的 socket。下面是 accept 接口的原型：

int accept(int s, struct sockaddr *addr, socklen_t *addrlen);

输入参数 s 是从 socket()，bind() 和 listen() 中沿用下来的 socket 句柄值。执行完 bind() 和 listen() 后，操做系统已经开始在指定的端口处监听全部的链接请求，若是有请求，则将该链接请求加入请求队列。调用 accept() 接口正是从 socket s 的请求队列抽取第一个链接信息，建立一个与 s 同类的新的 socket 返回句柄。新的 socket 句柄便是后续 read() 和 recv() 的输入参数。若是请求队列当前没有请求，则 accept() 将进入阻塞状态直到有请求进入队列。

上述多线程的服务器模型彷佛完美的解决了为多个客户机提供问答服务的要求，但其实并不尽然。若是要同时响应成百上千路的链接请求，则不管多线程仍是多进程都会严重占据系统资源，下降系统对外界响应效率，而线程与进程自己也更容易进入假死状态。

不少程序员可能会考虑使用“线程池”或“链接池”。“线程池”旨在减小建立和销毁线程的频率，其维持必定合理数量的线程，并让空闲的线程从新承担新的执行任务。“链接池”维持链接的缓存池，尽可能重用已有的链接、减小建立和关闭链接的频率。这两种技术均可以很好的下降系统开销，都被普遍应用不少大型系统，如 websphere、tomcat 和各类数据库等。

可是，“线程池”和“链接池”技术也只是在必定程度上缓解了频繁调用 IO 接口带来的资源占用。并且，所谓“池”始终有其上限，当请求大大超过上限时，“池”构成的系统对外界的响应并不比没有池的时候效果好多少。因此使用“池”必须考虑其面临的响应规模，并根据响应规模调整“池”的大小。

对应上例中的所面临的可能同时出现的上千甚至上万次的客户端请求，“线程池”或“链接池”或许能够缓解部分压力，可是不能解决全部问题。

总之，多线程模型能够方便高效的解决小规模的服务请求，但面对大规模的服务请求，多线程模型并非最佳方案。下一章咱们将讨论用非阻塞接口来尝试解决这个问题。

非阻塞的服务器程序

以上面临的不少问题，必定程度是 IO 接口的阻塞特性致使的。多线程是一个解决方案，还一个方案就是使用非阻塞的接口。

非阻塞的接口相比于阻塞型接口的显著差别在于，在被调用以后当即返回。使用以下的函数能够将某句柄 fd 设为非阻塞状态。

fcntl( fd, F_SETFL, O_NONBLOCK );

下面将给出只用一个线程，但可以同时从多个链接中检测数据是否送达，而且接受数据。

图 3. 使用非阻塞的接收数据模型

在非阻塞状态下，recv() 接口在被调用后当即返回，返回值表明了不一样的含义。如在本例中，

recv() 返回值大于 0，表示接受数据完毕，返回值便是接受到的字节数；
recv() 返回 0，表示链接已经正常断开；
recv() 返回 -1，且 errno 等于 EAGAIN，表示 recv 操做还没执行完成；
recv() 返回 -1，且 errno 不等于 EAGAIN，表示 recv 操做遇到系统错误 errno。

能够看到服务器线程能够经过循环调用 recv() 接口，能够在单个线程内实现对全部链接的数据接收工做。

可是上述模型毫不被推荐。由于，循环调用 recv() 将大幅度推高 CPU 占用率；此外，在这个方案中，recv() 更多的是起到检测“操做是否完成”的做用，实际操做系统提供了更为高效的检测“操做是否完成“做用的接口，例如 select()。

使用 select() 接口的基于事件驱动的服务器模型

大部分 Unix/Linux 都支持 select 函数，该函数用于探测多个文件句柄的状态变化。下面给出 select 接口的原型：

 
     FD_ZERO(int fd, fd_set* fds)  
    
     FD_SET(int fd, fd_set* fds)  
    
     FD_ISSET(int fd, fd_set* fds)  
    
     FD_CLR(int fd, fd_set* fds)  
    
     int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds,  
    
     struct timeval *timeout)

这里，fd_set 类型能够简单的理解为按 bit 位标记句柄的队列，例如要在某 fd_set 中标记一个值为 16 的句柄，则该 fd_set 的第 16 个 bit 位被标记为 1。具体的置位、验证可以使用 FD_SET、FD_ISSET 等宏实现。在 select() 函数中，readfds、writefds 和 exceptfds 同时做为输入参数和输出参数。若是输入的 readfds 标记了 16 号句柄，则 select() 将检测 16 号句柄是否可读。在 select() 返回后，能够经过检查 readfds 有否标记 16 号句柄，来判断该“可读”事件是否发生。另外，用户能够设置 timeout 时间。

下面将从新模拟上例中从多个客户端接收数据的模型。

图 4. 使用 select() 的接收数据模型

上述模型只是描述了使用 select() 接口同时从多个客户端接收数据的过程；因为 select() 接口能够同时对多个句柄进行读状态、写状态和错误状态的探测，因此能够很容易构建为多个客户端提供独立问答服务的服务器系统。

图 5. 使用 select() 接口的基于事件驱动的服务器模型

这里须要指出的是，客户端的一个 connect() 操做，将在服务器端激发一个“可读事件”，因此 select() 也能探测来自客户端的 connect() 行为。

上述模型中，最关键的地方是如何动态维护 select() 的三个参数 readfds、writefds 和 exceptfds。做为输入参数，readfds 应该标记全部的须要探测的“可读事件”的句柄，其中永远包括那个探测 connect() 的那个“母”句柄；同时，writefds 和 exceptfds 应该标记全部须要探测的“可写事件”和“错误事件”的句柄 ( 使用 FD_SET() 标记 )。

做为输出参数，readfds、writefds 和 exceptfds 中的保存了 select() 捕捉到的全部事件的句柄值。程序员须要检查的全部的标记位 ( 使用 FD_ISSET() 检查 )，以肯定到底哪些句柄发生了事件。

上述模型主要模拟的是“一问一答”的服务流程，因此，若是 select() 发现某句柄捕捉到了“可读事件”，服务器程序应及时作 recv() 操做，并根据接收到的数据准备好待发送数据，并将对应的句柄值加入 writefds，准备下一次的“可写事件”的 select() 探测。一样，若是 select() 发现某句柄捕捉到“可写事件”，则程序应及时作 send() 操做，并准备好下一次的“可读事件”探测准备。下图描述的是上述模型中的一个执行周期。

图 6. 一个执行周期

这种模型的特征在于每个执行周期都会探测一次或一组事件，一个特定的事件会触发某个特定的响应。咱们能够将这种模型归类为“事件驱动模型”。

相比其余模型，使用 select() 的事件驱动模型只用单线程（进程）执行，占用资源少，不消耗太多 CPU，同时可以为多客户端提供服务。若是试图创建一个简单的事件驱动的服务器程序，这个模型有必定的参考价值。

但这个模型依旧有着不少问题。

首先，select() 接口并非实现“事件驱动”的最好选择。由于当须要探测的句柄值较大时，select() 接口自己须要消耗大量时间去轮询各个句柄。不少操做系统提供了更为高效的接口，如 linux 提供了 epoll，BSD 提供了 kqueue，Solaris 提供了 /dev/poll …。若是须要实现更高效的服务器程序，相似 epoll 这样的接口更被推荐。遗憾的是不一样的操做系统特供的 epoll 接口有很大差别，因此使用相似于 epoll 的接口实现具备较好跨平台能力的服务器会比较困难。

其次，该模型将事件探测和事件响应夹杂在一块儿，一旦事件响应的执行体庞大，则对整个模型是灾难性的。以下例，庞大的执行体 1 的将直接致使响应事件 2 的执行体迟迟得不到执行，并在很大程度上下降了事件探测的及时性。

图 7. 庞大的执行体对使用 select() 的事件驱动模型的影响

幸运的是，有不少高效的事件驱动库能够屏蔽上述的困难，常见的事件驱动库有 libevent 库，还有做为 libevent 替代者的 libev 库。这些库会根据操做系统的特色选择最合适的事件探测接口，而且加入了信号 (signal) 等技术以支持异步响应，这使得这些库成为构建事件驱动模型的不二选择。下章将介绍如何使用 libev 库替换 select 或 epoll 接口，实现高效稳定的服务器模型。

使用事件驱动库 libev 的服务器模型

Libev 是一种高性能事件循环 / 事件驱动库。做为 libevent 的替代做品，其第一个版本发布与 2007 年 11 月。Libev 的设计者声称 libev 拥有更快的速度，更小的体积，更多功能等优点，这些优点在不少测评中获得了证实。正由于其良好的性能，不少系统开始使用 libev 库。本章将介绍如何使用 Libev 实现提供问答服务的服务器。

（事实上，现存的事件循环 / 事件驱动库有不少，做者也无心推荐读者必定使用 libev 库，而只是为了说明事件驱动模型给网络服务器编程带来的便利和好处。大部分的事件驱动库都有着与 libev 库相相似的接口，只要明白大体的原理，便可灵活挑选合适的库。）

与前章的模型相似，libev 一样须要循环探测事件是否产生。Libev 的循环体用 ev_loop 结构来表达，并用 ev_loop( ) 来启动。

void ev_loop( ev_loop* loop, int flags )

Libev 支持八种事件类型，其中包括 IO 事件。一个 IO 事件用 ev_io 来表征，并用 ev_io_init() 函数来初始化：

void ev_io_init(ev_io *io, callback, int fd, int events)

初始化内容包括回调函数 callback，被探测的句柄 fd 和须要探测的事件，EV_READ 表“可读事件”，EV_WRITE 表“可写事件”。

如今，用户须要作的仅仅是在合适的时候，将某些 ev_io 从 ev_loop 加入或剔除。一旦加入，下个循环即会检查 ev_io 所指定的事件有否发生；若是该事件被探测到，则 ev_loop 会自动执行 ev_io 的回调函数 callback()；若是 ev_io 被注销，则再也不检测对应事件。

不管某 ev_loop 启动与否，均可以对其添加或删除一个或多个 ev_io，添加删除的接口是 ev_io_start() 和 ev_io_stop()。

 
     void ev_io_start( ev_loop *loop, ev_io* io )  
    
     void ev_io_stop( EV_A_* )

由此，咱们能够容易得出以下的“一问一答”的服务器模型。因为没有考虑服务器端主动终止链接机制，因此各个链接能够维持任意时间，客户端能够自由选择退出时机。

图 8. 使用 libev 库的服务器模型

上述模型能够接受任意多个链接，且为各个链接提供彻底独立的问答服务。借助 libev 提供的事件循环 / 事件驱动接口，上述模型有机会具有其余模型不能提供的高效率、低资源占用、稳定性好和编写简单等特色。

因为传统的 web 服务器，ftp 服务器及其余网络应用程序都具备“一问一答”的通信逻辑，因此上述使用 libev 库的“一问一答”模型对构建相似的服务器程序具备参考价值；另外，对于须要实现远程监视或远程遥控的应用程序，上述模型一样提供了一个可行的实现方案。

总结

本文围绕如何构建一个提供“一问一答”的服务器程序，前后讨论了用阻塞型的 socket 接口实现的模型，使用多线程的模型，使用 select() 接口的基于事件驱动的服务器模型，直到使用 libev 事件驱动库的服务器模型。文章对各类模型的优缺点都作了比较，从比较中得出结论，即便用“事件驱动模型”能够的实现更为高效稳定的服务器程序。文中描述的多种模型能够为读者的网络编程提供参考价值。

select，poll，epoll实现分析—结合内核源代码

select，poll，epoll都是IO多路复用的机制。所谓I/O多路复用机制，就是说经过一种机制，能够监视多个描述符，一旦某个描述符就绪（通常是读就绪或者写就绪），可以通知程序进行相应的读写操做。但select，poll，epoll本质上都是同步I/O，由于他们都须要在读写事件就绪后本身负责进行读写，也就是说这个读写过程是阻塞的，而异步I/O则无需本身负责进行读写，异步I/O的实现会负责把数据从内核拷贝到用户空间。关于阻塞，非阻塞，同步，异步将在下一篇文章详细说明。

select和poll的实现比较类似，目前也有不少为人诟病的缺点，epoll能够说是select和poll的加强版。

1、select实现

一、使用copy_from_user从用户空间拷贝fd_set到内核空间

二、注册回调函数__pollwait

三、遍历全部fd，调用其对应的poll方法（对于socket，这个poll方法是sock_poll，sock_poll根据状况会调用到tcp_poll,udp_poll或者datagram_poll）

四、以tcp_poll为例，其核心实现就是__pollwait，也就是上面注册的回调函数。

五、__pollwait的主要工做就是把current（当前进程）挂到设备的等待队列中，不一样的设备有不一样的等待队列，对于tcp_poll来讲，其等待队列是sk->sk_sleep（注意把进程挂到等待队列中并不表明进程已经睡眠了）。在设备收到一条消息（网络设备）或填写完文件数据（磁盘设备）后，会唤醒设备等待队列上睡眠的进程，这时current便被唤醒了。

六、poll方法返回时会返回一个描述读写操做是否就绪的mask掩码，根据这个mask掩码给fd_set赋值。

七、若是遍历完全部的fd，尚未返回一个可读写的mask掩码，则会调用schedule_timeout是调用select的进程（也就是current）进入睡眠。当设备驱动发生自身资源可读写后，会唤醒其等待队列上睡眠的进程。若是超过必定的超时时间（schedule_timeout指定），仍是没人唤醒，则调用select的进程会从新被唤醒得到CPU，进而从新遍历fd，判断有没有就绪的fd。

八、把fd_set从内核空间拷贝到用户空间。

总结：

select的几大缺点：

（1）每次调用select，都须要把fd集合从用户态拷贝到内核态，这个开销在fd不少时会很大

（2）同时每次调用select都须要在内核遍历传递进来的全部fd，这个开销在fd不少时也很大

（3）select支持的文件描述符数量过小了，默认是1024

2、poll实现

poll的实现和select很是类似，只是描述fd集合的方式不一样，poll使用pollfd结构而不是select的fd_set结构。其余的都差很少。

3、epoll实现

说明一下这个回调机制的原理，其实很简单，看一下select和epoll在把current加入fd对应的设备等待队列时使用的代码：

select:

static void __pollwait(struct file *filp, wait_queue_head_t *wait_address,
poll_table *p)
{
struct poll_table_entry *entry = poll_get_entry(p);
if (!entry)
return;
get_file(filp);
entry->filp = filp;
entry->wait_address = wait_address;
init_waitqueue_entry(&entry->wait, current);
add_wait_queue(wait_address, &entry->wait);
}

其中init_waitqueue_entry实现以下：

static inline void init_waitqueue_entry(wait_queue_t *q, struct task_struct *p)
{
q->flags = 0;
q->private = p;
q->func = default_wake_function;
}

上面的代码是说创建一个poll_table_entry结构entry，首先把current设置为entry->wait的private成员，同时把default_wake_function设为entry->wait的func成员，而后把entry->wait链入到wait_address中（这个wait_address就是设备的等待队列，在tcp_poll中就是sk_sleep）。

再看一下epoll：

/*
* This is the callback that is used to add our wait queue to the
* target file wakeup lists.
*/
static void ep_ptable_queue_proc(struct file *file, wait_queue_head_t *whead,
poll_table *pt)
{
struct epitem *epi = ep_item_from_epqueue(pt);
struct eppoll_entry *pwq;
if (epi->nwait >= 0 && (pwq = kmem_cache_alloc(pwq_cache, GFP_KERNEL))) {
init_waitqueue_func_entry(&pwq->wait, ep_poll_callback);
pwq->whead = whead;
pwq->base = epi;
add_wait_queue(whead, &pwq->wait);
list_add_tail(&pwq->llink, &epi->pwqlist);
epi->nwait++;
} else {
/* We have to signal that an error occurred */
epi->nwait = -1;
}
}

其中init_waitqueue_func_entry的实现以下：

static inline void init_waitqueue_func_entry(wait_queue_t *q,
wait_queue_func_t func)
{
q->flags = 0;
q->private = NULL;
q->func = func;
}

能够看到，整体和select的实现是相似的，只不过它是建立了一个eppoll_entry结构pwq，只不过pwq->wait的func成员被设置成了回调函数ep_poll_callback（而不是default_wake_function，因此这里并不会有唤醒操做，而只是执行回调函数），private成员被设置成了NULL。最后吧pwq->wait链入到whead中（也就是设备等待队列中）。这样，当设备等待队列中的进程被唤醒时，就会调用ep_poll_callback了。

再梳理一下，当epoll_wait时，它会判断就绪链表中有没有就绪的fd，若是没有，则把current进程加入一个等待队列(file->private_data->wq)中，并在一个while（1）循环中判断就绪队列是否为空，并结合schedule_timeout实现睡一会，判断一会的效果。若是current进程在睡眠中，设备就绪了，就会调用回调函数。在回调函数中，会把就绪的fd放入就绪链表，并唤醒等待队列(file->private_data->wq)中的current进程，这样epoll_wait又能继续执行下去了。

总结：

一、select，poll实现须要本身不断轮询全部fd集合，直到设备就绪，期间可能要睡眠和唤醒屡次交替。而epoll其实也须要调用epoll_wait不断轮询就绪链表，期间也可能屡次睡眠和唤醒交替，可是它是设备就绪时，调用回调函数，把就绪fd放入就绪链表中，并唤醒在epoll_wait中进入睡眠的进程。虽然都要睡眠和交替，可是select和poll在“醒着”的时候要遍历整个fd集合，而epoll在“醒着”的时候只要判断一下就绪链表是否为空就好了，这节省了大量的CPU时间。这就是回调机制带来的性能提高。

二、select，poll每次调用都要把fd集合从用户态往内核态拷贝一次，而且要把current往设备等待队列中挂一次，而epoll只要一次拷贝，并且把current往等待队列上挂也只挂一次（在epoll_wait的开始，注意这里的等待队列并非设备等待队列，只是一个epoll内部定义的等待队列）。这也能节省很多的开销。

select，poll，epoll比较

select，poll，epoll简介

select

select本质上是经过设置或者检查存放fd标志位的数据结构来进行下一步处理。这样所带来的缺点是：

1 单个进程可监视的fd数量被限制

2 须要维护一个用来存放大量fd的数据结构，这样会使得用户空间和内核空间在传递该结构时复制开销大

3 对socket进行扫描时是线性扫描

poll

poll本质上和select没有区别，它将用户传入的数组拷贝到内核空间，而后查询每一个fd对应的设备状态，若是设备就绪则在设备等待队列中加入一项并继续遍历，若是遍历完全部fd后没有发现就绪设备，则挂起当前进程，直到设备就绪或者主动超时，被唤醒后它又要再次遍历fd。这个过程经历了屡次无谓的遍历。

它没有最大链接数的限制，缘由是它是基于链表来存储的，可是一样有一个缺点：大量的fd的数组被总体复制于用户态和内核地址空间之间，而无论这样的复制是否是有意义。

poll还有一个特色是“水平触发”，若是报告了fd后，没有被处理，那么下次poll时会再次报告该fd。

epoll

epoll支持水平触发和边缘触发，最大的特色在于边缘触发，它只告诉进程哪些fd刚刚变为就需态，而且只会通知一次。

在前面说到的复制问题上，epoll使用mmap减小复制开销。

还有一个特色是，epoll使用“事件”的就绪通知方式，经过epoll_ctl注册fd，一旦该fd就绪，内核就会采用相似callback的回调机制来激活该fd，epoll_wait即可以收到通知

1 支持一个进程所能打开的最大链接数

select	单个进程所能打开的最大链接数有FD_SETSIZE宏定义，其大小是32个整数的大小（在32位的机器上，大小就是3232，同理64位机器上 FD_SETSIZE为3264），固然咱们能够对它进行修改，而后从新编译内核，可是性能可能会受到影响，这须要进一步的测试。
poll	poll本质上和select没有区别，可是它没有最大链接数的限制，缘由是它是基于链表来存储的
epoll	虽然链接数有上限，可是很大，1G内存的机器上能够打开10万左右的链接，2G内存的机器能够打开20万左右的链接。

2 FD剧增后带来的IO效率问题

select	由于每次调用时都会对链接进行线性遍历，因此随着FD的增长会形成遍历速度慢的“线性降低性能问题”。
poll	同上
epoll	由于epoll内核中实现是根据每一个fd上的callback函数来实现的，只有活跃的socket才会主动调用callback，因此在活跃socket较少的状况下，使用epoll没有前面二者的线性降低的性能问题，可是全部socket都很活跃的状况下，可能会有性能问题。

3 消息传递方式

select	内核须要将消息传递到用户空间，都须要内核拷贝动做
poll	同上
epoll	epoll经过内核和用户空间共享一块内存来实现的

综上，在选择select，poll，epoll时要根据具体的使用场合以及这三种方式的自身特色。表面上看epoll的性能最好，可是在链接数少而且链接都十分活跃的状况下，select和poll的性能可能比epoll好，毕竟epoll的通知机制须要不少函数回调。

select、poll、epoll使用小结

Linux上可使用不一样的I/O模型，咱们能够经过下图了解经常使用的I/O模型：同步和异步模型，以及阻塞和非阻塞模型，本文主要分析其中的异步阻塞模型。

1、select使用

这个模型中配置的是非阻塞I/O,而后使用阻塞select系统调用来肯定一个I/O描述符什么时候有操做。使用select调用能够为多个描述符提供通知，对于每一个提示符，咱们能够请求描述符的可写，可读以及是否发生错误。异步阻塞I/O的系统流程以下图所示：

使用select经常使用的几个函数以下：

[cpp] view plain copy

FD_ZERO(int fd, fd_set* fds)
FD_SET(int fd, fd_set* fds)
FD_ISSET(int fd, fd_set* fds)
FD_CLR(int fd, fd_set* fds)
int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout)

fd_set类型能够简单的理解为按bit位标记句柄的队列。具体的置位、验证可使用FD_SET,FD_ISSET等宏实现。在select函数中，readfds、writefds和exceptfds同时做为输入参数和输出参数，若是readfds标记了一个位置，则，select将检测到该标记位可读。timeout为设置的超时时间。

下面咱们来看如何使用select：

[cpp] view plain copy

SOCKADDR_IN addrSrv;
int reuse = 1;
SOCKET sockSrv,connsock;
SOCKADDR_IN addrClient;
pool pool;
int len=sizeof(SOCKADDR);
/*建立TCP*/
sockSrv=socket(AF_INET,SOCK_STREAM,0);
/*地址、端口的绑定*/
addrSrv.sin_addr.S_un.S_addr=htonl(INADDR_ANY);
addrSrv.sin_family=AF_INET;
addrSrv.sin_port=htons(port);
if(bind(sockSrv,(SOCKADDR*)&addrSrv,sizeof(SOCKADDR))<0)
{
fprintf(stderr,"Failed to bind");
return ;
}
if(listen(sockSrv,5)<0)
{
fprintf(stderr,"Failed to listen socket");
return ;
}
setsockopt(sockSrv,SOL_SOCKET,SO_REUSEADDR,(const char*)&reuse,sizeof(reuse));
init_pool(sockSrv,&pool);
while(1)
{
/*经过selete设置为异步模式*/
pool.ready_set=pool.read_set;
pool.nready=select(pool.maxfd+1,&pool.ready_set,NULL,NULL,NULL);
if(FD_ISSET(sockSrv,&pool.ready_set))
{
connsock=accept(sockSrv,(SOCKADDR *)&addrClient,&len);
//loadDeal()/*链接处理*/
//printf("test\n");
add_client(connsock,&pool);//添加到链接池
}
/*检查是否有事件发生*/
check_client(&pool);
}

上面是一个服务器代码的关键部分，设置为异步的模式，而后接受到链接将其添加到链接池中。监听描述符上使用select，接受客户端的链接请求，在check_client函数中，遍历链接池中的描述符，检查是否有事件发生。

2、poll使用

poll函数相似于select，可是其调用形式不一样。poll不是为每一个条件构造一个描述符集，而是构造一个pollfd结构体数组，每一个数组元素指定一个描述符标号及其所关心的条件。定义以下：

[cpp] view plain copy

#include <sys/poll.h>
int poll (struct pollfd *fds, unsigned int nfds, int timeout);
struct pollfd {
int fd; /* file descriptor */
short events; /* requested events to watch */
short revents; /* returned events witnessed */
};

每一个结构体的events域是由用户来设置，告诉内核咱们关注的是什么，而revents域是返回时内核设置的，以说明对该描述符发生了什么事件。这点与select不一样，select修改其参数以指示哪个描述符准备好了。在《unix环境高级编程》中有一张events取值的表，以下：

POLLIN :可读除高优级外的数据，不阻塞

POLLRDNORM：可读普通数据，不阻塞

POLLRDBAND：可读O优先数据，不阻塞

POLLPRI：可读高优先数据，不阻塞

POLLOUT ：可写普数据，不阻塞

POLLWRNORM：与POLLOUT相同

POLLWRBAND：写非0优先数据，不阻塞

其次revents还有下面取值

POLLERR ：已出错

POLLHUP：已挂起，当以描述符被挂起后，就不能再写向该描述符，可是仍能够从该描述符读取到数据。

POLLNVAL：此描述符并不引用一打开文件

对poll函数，nfds表示fds中的元素数，timeout为超时设置，单位为毫秒若为0，表示不等待，为-1表示描述符中一个已经准备好或捕捉到一个信号返回，大于0表示描述符准备好，或超时返回。函数返回值返回值若为0，表示没有事件发生，-1表示错误，并设置errno，大于0表示有几个描述符有事件。

poll的使用和select基本相似。在此再也不介绍。poll相对因而select的优点是监听的描述符数量没有限制。

3、epoll学习

epoll有两种模式,Edge Triggered(简称ET) 和 Level Triggered(简称LT).在采用这两种模式时要注意的是,若是采用ET模式,那么仅当状态发生变化时才会通知,而采用LT模式相似于原来的select/poll操做,只要还有没有处理的事件就会一直通知.

1）epoll数据结构介绍：

[cpp] view plain copy

typedef union epoll_data
{
void *ptr;
int fd;
__uint32_t u32;
__uint64_t u64;
} epoll_data_t;
struct epoll_event
{
__uint32_t events; /* Epoll events */
epoll_data_t data; /* User data variable */
};

常见的事件以下：

EPOLLIN：表示对描述符的能够读

EPOLLOUT：表示对描述符的能够写

EPOLLPRI：表示对描述符的有紧急数据能够读

EPOLLERR：发生错误

EPOLLHUP：挂起

EPOLLET：边缘触发

EPOLLONESHOT：一次性使用，当监听完此次事件以后，若是还须要继续监听这个socket的话，须要再次把这个socket加入到EPOLL队列里

2）函数介绍

epoll的三个函数

[cpp] view plain copy

int epoll_creae(int size);

功能：该函数生成一个epoll专用的文件描述符

参数：size为epoll上能关注的最大描述符数

[cpp] view plain copy

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

功能：用于控制某个epoll文件描述符时间，能够注册、修改、删除

参数：epfd由epoll_create生成的epoll专用描述符

op操做：EPOLL_CTL_ADD 注册 EPOLL_CTL_MOD修改 EPOLL_DEL删除

fd：关联的文件描述符

evnet告诉内核要监听什么事件

[cpp] view plain copy

int epoll_wait(int epfd,struct epoll_event*events,int maxevents,int timeout);

功能：该函数等待i/o事件的发生。

参数：epfd要检测的句柄

events：用于回传待处理时间的数组

maxevents：告诉内核这个events有多大，不能超过以前的size

timeout：为超时时间

使用方法参考：https://banu.com/blog/2/how-to-use-epoll-a-complete-example-in-c/epoll-example.c

epoll支持的FD上限是最大能够打开文件的数目（select面临这样的问题），IO效率不随FD数目增长而线性降低（select、poll面临的问题）使用mmap加速内核与用户空间的消息传递。如今libevent封装了几种的实现，能够经过使用libevent来实现多路复用。

本文参考：https://banu.com/blog/2/how-to-use-epoll-a-complete-example-in-c/

http://www.ibm.com/developerworks/cn/linux/l-cn-edntwk/index.html?ca=drs-

http://www.ibm.com/developerworks/cn/linux/l-async/

select、poll、epoll之间的区别

select(poll)系统调用实现解析

使用事件驱动模型实现高效稳定的网络服务器程序

前言

阻塞型的网络编程接口

图 1. 简单的一问一答的服务器 / 客户机模型

多线程的服务器程序

图 2. 多线程的服务器模型

非阻塞的服务器程序

图 3. 使用非阻塞的接收数据模型

使用 select() 接口的基于事件驱动的服务器模型

图 4. 使用 select() 的接收数据模型

图 5. 使用 select() 接口的基于事件驱动的服务器模型

图 6. 一个执行周期

图 7. 庞大的执行体对使用 select() 的事件驱动模型的影响

使用事件驱动库 libev 的服务器模型

图 8. 使用 libev 库的服务器模型

总结

相关主题

select，poll，epoll实现分析—结合内核源代码

select，poll，epoll比较

select、poll、epoll使用小结