用户态下select系统调用
select多路I/O转接服务器demo：select_server.c
select函数原型
/usr/include/sys/select.h
/* According to POSIX.1-2001 */
#include <sys/select.h>
int select(int nfds, fd_set *readfds, fd_set *writefds,
fd_set *exceptfds, struct timeval *timeout);
void FD_CLR(int fd, fd_set *set); //从fdset中删除fd
int FD_ISSET(int fd, fd_set *set); //判断fd是否已存在fdset
void FD_SET(int fd, fd_set *set); //将fd添加到fdset
void FD_ZERO(fd_set *set); //fdset所有位清0
select参数：
nfds：监控的文件描述符集中，待测试的最大描述符+1
readfds：监控有读数据到达文件描述符集合，传入传出参数
writefds：监控有写数据到达文件描述符集合，传入传出参数
exceptfds：监控异常发生达文件描述符集合，传入传出参数
a）带外数据到达
b）某个已置为分组模式的伪终端存在可从其主机端读取的控制状态信息
timeout：定时阻塞监控时间，3种情况：
1）NULL，永远等下去
2）设置 timeval ，等待固定时间
3）设置 timeval 里时间均为0，检查描述字后立即返回，轮询
注意：FD_为前缀的函数并非系统调用，而是几个对fd_set进行相关位操作的宏
fd_set结构体的定义实际包含的是fds_bits位数组，其大小固定，由FD_SETSIZE指定，因此每次select系统调用可监听处理的文件描述符最大数量为FD_SETSIZE：
/* fd_set for select and pselect. */
typedef struct
{
/* XPG4.2 requires this member name. Otherwise avoid the name
from the global namespace. */
#ifdef __USE_XOPEN
__fd_mask fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->fds_bits)
#else
__fd_mask __fds_bits[__FD_SETSIZE / __NFDBITS];
# define __FDS_BITS(set) ((set)->__fds_bits)
#endif
} fd_set;
select函数作用
select函数允许进程指示内核等待多个事件中的任何一个发生，并只在有一个或多个事件发生或经历一段指定的时间后才唤醒它。
readfds、writefds、exceptfds三个参数中，若不关注某个参数，可将其设为NULL。三个参数均为NULL相当于一个定时器。
select函数局限
- select能监听的文件描述符个数受限于
FD_SETSIZE，一般为1024，单纯改变进程打开的文件描述符个数并不能改变select监听文件个数。
/usr/include/sys/select.h
/usr/include/bits/typesizes.h
/* Maximum number of file descriptors in `fd_set'. */
#define FD_SETSIZE __FD_SETSIZE
/* Number of descriptors that can fit in an `fd_set'. */
#define __FD_SETSIZE 1024
-
描述符集内任何与未就绪描述符对应的位返回时均清成0。为此，每次重新调用select函数时，我们都得再次把所有描述符集内所关心的位均置为1。
-
解决1024以下客户端时使用select是很合适的，但如果链接客户端过多，select采用的是轮询模型，会大大降低服务器响应效率，不应在select上投入更多精力。
套接字就绪条件
select源码剖析
内核源码：linux-3.10
seelct源码：fs/select.c
内核源码链接：https://mirrors.edge.kernel.org/pub/linux/kernel/v3.0/linux-3.10.tar.gz
select系统调用入口
- 若设置超时时间，用户空间（微秒量级）拷贝到内核空间（纳秒量级）
-
core_sys_select真正执行入口 - 传出剩余时间差，返回给用户空间
- 正常结束，返回满足条件的描述符个数
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;
/* 1. 永久等待 tvp == NULL
* 2. 不等待 tvp->tv_sec == 0 && tvp->tc_nsec == 0
* 3. 等待指定时间 tvp->tv_sec != 0 || tvp->tc_nsec != 0 */
if (tvp) {
/** 在设置超时情况下，拷贝用户空间下的相对超时时间（微秒量级）到内核 **/
if (copy_from_user(&tv, tvp, sizeof(tv)))
return -EFAULT;
to = &end_time;
/** poll_select_set_timeout 设定成绝对的超时时间（纳秒量级） **/
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;
}
/** select 真正执行入口 **/
ret = core_sys_select(n, inp, outp, exp, to);
/* 将此次调用完成剩余的时间差值通过 tvp 指向的 timeval 结构返回给用户空间 */
ret = poll_select_copy_remaining(&end_time, tvp, 1, ret);
return ret;
}
core_sys_select
- 先创建一个默认大小栈缓冲区（加快访问，但缓冲区大小可能不够）。
- 检查最大fd是否超出进程文件描述符位图所容量的最大值（默认1024），超过部分不监听（修正）。
- 计算前面创建的栈缓冲区是否足够存储输入、输出6个集合，若缓冲区大小不足则使用
kmalloc分配内核空间。 - 将缓冲区分成6段，从用户空间拷贝输入集到内核空间，并将内核空间结果集清0。
- 执行主线 do_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;
void *bits;
int ret, max_fds;
unsigned int size;
struct fdtable *fdt;
/* 需要将用户空间传进来的inset、outset、exset拷贝到内核空间，并且
* 需要等容量的空间来存储结果集，之后会将结果集的内容写回到用户空间。
* 定义一个SELECT_STACK_ALLOC(256字节)大小的栈上缓冲区，用于缓存输入输出结果集。
* 如果缓存的空间大小不够，那么再使用kmalloc()动态分配，
* 优先使用栈缓存而不用动态内存可以加快访问 */
long stack_fds[SELECT_STACK_ALLOC/sizeof(long)];
ret = -EINVAL;
if (n < 0)
goto out_nofds;
/* rdlock加锁：保护struct files的访问 */
rcu_read_lock();
fdt = files_fdtable(current->files);
max_fds = fdt->max_fds;
rcu_read_unlock();
/* 基于current宏，检查传入的最大fd对应参数n是否超出当前进程打开的
* 文件描述符表内所示位图容量的最大数值，超出修正*/
if (n > max_fds)
n = max_fds;
/*
* n个bits至少需要size个long才能装下（long表示bits段），
* 为了存储输入输出集，我们需要6*size个long的存储空间
* 栈上数组空间不足以存放本次select要处理的fd集合所需总计内存，
* 则使用kmalloc从内核空间分配所需的连续物理内存
*/
size = FDS_BYTES(n);
bits = stack_fds;
if (size > sizeof(stack_fds) / 6) {
/* Not enough space in on-stack array; must use kmalloc */
ret = -ENOMEM;
bits = kmalloc(6 * size, GFP_KERNEL);
if (!bits)
goto out_nofds;
}
/* 将数组分成6段 */
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;
/* 依次从用户空间拷贝输入集数据 */
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;
/* 将输出集清0 */
zero_fd_set(n, fds.res_in);
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);
out_nofds:
return ret;
}
do_select（最核心）
相关结构体
poll_table_entry
struct poll_table_entry {
struct file *filp;
unsigned long key;
wait_queue_t wait; //wait等待队列项
wait_queue_head_t *wait_address; //wait的等待队列头
};
poll_table_page
struct poll_table_page {
struct poll_table_page * next;
struct poll_table_entry * entry;
struct poll_table_entry entries[0];
};
poll_table
typedef struct poll_table_struct {
poll_queue_proc _qproc;
unsigned long _key;
} poll_table;
poll_wqueues
struct poll_wqueues {
poll_table pt; //
struct poll_table_page *table; //
struct task_struct *polling_task; //正在轮询的进程
int triggered;
int error;
int inline_index;
//记录poll信息的数组
struct poll_table_entry inline_entries[N_INLINE_POLL_ENTRIES];
};
do_select源码剖析
- 初始化内核数据，再次检查并修正最大文件描述符
- 等待队列构建，并初始化等待队列
- 无穷循环开始轮询事件监测，核心调用转poll
- 结果写入内核空间的结果集
- 释放内核相应数据结构空间
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; /* timed_out指示是否已经超时，超时1，未超时0 */
unsigned long slack = 0;
/* 借助当前进程已打开的文件描述符表检查传入且合法的已打开最大fd，并修正传入的n */
rcu_read_lock();
retval = max_select_fd(n, fds);
rcu_read_unlock();
if (retval < 0)
return retval;
n = retval;
/* poll_initwait初始化poll_wqueues结构体table，这一结构体用于本次select调用
* 对所有传入的待监听fd进行【轮询工作】，每个fd对应一个poll_table_entry。
* 初始化poll_wqueues：
* 1. 初始化poll_wqueues中的poll_table：
* * 设置监听注册函数为 __pollwait
* 2. 设置polling_task指向当前进程PCB。*/
poll_initwait(&table);
wait = &table.pt;
/* 根本不等待情况处理 */
if (end_time && !end_time->tv_sec && !end_time->tv_nsec) {
wait->_qproc = NULL; /* 注意：设为NULL了！！！ */
timed_out = 1; /* 还没开始就已经超时，这样就实现了根本不等待... */
}
/* 重新估算相对超时时间... */
if (end_time && !timed_out)
slack = select_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;
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;
in = *inp++; out = *outp++; ex = *exp++;
all_bits = in | out | ex;
if (all_bits == 0) { /* 跳过我们不关心的bits段 */
i += BITS_PER_LONG;
continue;
}
/* 以BITS_PER_LONG个为一组依次挂载到等待队列，并对事件进行检测，
* 如果没有事件到来，仅有第一次循环完成挂载，后续循环只监测事件。*/
for (j = 0; j < BITS_PER_LONG; ++j, ++i, bit <<= 1) {
struct fd f;
if (i >= n) /* 超出了关心的文件描述符范围[0, n)，那么跳出... */
break;
if (!(bit & all_bits)) /* 跳过我们不关心的bit */
continue;
f = fdget(i);
/* 因为没有rdlock加锁，因此当前进程中描述符i对应的文件可能已经
* 被异步关闭。这就是为什么需要判断file是否为空的原因 */
if (f.file) {
const struct file_operations *f_op;
f_op = f.file->f_op;
/* 注意：mask = POLLIN | POLLOUT | POLLRDNORM | POLLWRNORM; */
mask = DEFAULT_POLLMASK;
/* 如果这个文件支持poll()，那么我们就向这个文件注册监听函数；
* 如果不支持，那么我们就忽略掉这个文件描述符 */
if (f_op && f_op->poll) {
wait_key_set(wait, in, out, bit); /* 设置poll_table中想要监听的事件 */
/* #核心调用#
* 在这里会根据fd的不同创建类别调用真正的poll函数，
* socket下对应是sock_poll，如ipv4/tcp下会继续调用tcp_poll，
* 在这里完成调用poll_table注册的函数指针__poll_wait挂载等待队列操作（实际借助poll_wait封装调用）
*/
mask = (*f_op->poll)(f.file, wait); /* 对文件注册监听函数，并返回资源的当前状态 */
}
fdput(f);
/* 完成检测操作获取事件mask结果。events验证，其中retval表示就绪的资源数 */
if ((mask & POLLIN_SET) && (in & bit)) {
res_in |= bit;
retval++;
/* 保证仅在第一次循环时，完成本次fd对应挂载等待队列，
* 不论是否收到设备事件通知，本次调用仅挂载一次，因此置空poll_table注册的poll */
wait->_qproc = NULL; //
}
if ((mask & POLLOUT_SET) && (out & bit)) {
res_out |= bit;
retval++;
wait->_qproc = NULL; //
}
if ((mask & POLLEX_SET) && (ex & bit)) {
res_ex |= bit;
retval++;
wait->_qproc = NULL; //
}
}
}
/* 结果写入内核空间的结果集 */
if (res_in)
*rinp = res_in;
if (res_out)
*routp = res_out;
if (res_ex)
*rexp = res_ex;
cond_resched();
}
wait->_qproc = NULL; /* 将wait->_qproc设为NULL，表示我们不希望再进行监听注册 */
/* 事件发生、超时、中断，跳出死循环 */
if (retval || timed_out || signal_pending(current))
break;
if (table.error) { /* 发生了错误，我们也跳出死循环 */
retval = table.error;
break;
}
/*
* 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_wqueues是否已触发，如果还没有触发，那就
设置当前运行状态为可中断阻塞并进行睡眠，等待被唤醒。
* 被唤醒之后重新进行迭代，获取资源就绪情况。
* 在向资源注册监听与判断poll_wqueues是否已触发在这段时间内，可能资源异步就绪了，
如果没有触发标志，那么可能就会丢失资源就绪这个事件，可能导致select()永久沉睡。
* 这就是为什么需要poll_wqueues.triggered字段的原因。 */
if (!poll_schedule_timeout(&table, TASK_INTERRUPTIBLE,
to, slack))
timed_out = 1;
}
/* 1. 卸载安装到资源监听队列上的poll_table_entry
* 2. 释放poll_wqueues占用的资源 */
poll_freewait(&table);
return retval;
}
总结
select负责将超时时间从用户空间（微秒级）拷贝到内核空间（纳秒级），接着转真正入口core_sys_select。
core_sys_select函数调用树如下（图片来源网络）：
select注意点
- select函数第一个参数
nfds为监控的文件描述符集中，待测试的最大描述符+1。 - 每次调用select均需要将用户空间数据拷贝到内核空间
- select能监听的文件描述符个数受限于
FD_SETSIZE，一般为1024，单纯改变进程打开的文件描述符个数并不能改变select监听文件个数。修改后需重新编译内核。 - 每次select调用都要轮询完成将所有
fd的挂载到等待队列，以及对事件进行检测。 - 内核轮询检测监听集合中每一个描述符是否有事件发生，有事件到来时，不知道是哪些文件描述符有数据可以读写，需要把所有的文件描述符都轮询一遍才能知道。
- 有事件发生，轮询完一遍，将内核空间中的整个结果集
bitmap拷贝到用户空间。 - 用户进程仅知道有多少满足条件的描述符，需要遍历监听集合去查询。
- select函数
readfds、writefds、exceptfds三个集合参数均为传入传出参数，在每次调用select时需要对三个参数进行赋值。 - 解决1024以下客户端时使用select是很合适的，但如果链接客户端过多，select 采用的是轮询模型，会大大降低服务器响应效率。
参考
《UNIX网络编程 卷一：套接字联网API》
https://www.jianshu.com/p/da6642369ef0