Linux select 一网打尽
奇技指南
select, 你可以不用它，但你不能不了解它。通过阅读本文，可以帮你理清select的来龙去脉， 让你从中了解到：
我们常说的select的1024限制指的是什么N 4 V ? ？怎么会有这样的限制？
都说selej 4 F 9 p e z ) cct效率不高，是这样吗？为什么 ？
select使用中有坑吗？
注：本文的所有内容均指针对 Linux Kernel, 当前使用的源码版本是 5.3.0
原型
int select (int __nfds, fd_set *__restrict __readfds,
fd_set *__restri k , n S C : 2ct __writefds,
fd_set *__restrict __except; P 7 U , o $fds,
struct timeval *__restrict __timeout);select是IO多种复用的一种实现，它将需要监控的fd分为读、写、异常三类~ : X # R _，使用fd_set表示，当其返回时要么是超时，要么是有至少一种读、写或异常事件发生。
相关数据结构
FD_SET
FD_SET是select最重要的数据结构了，其V S &在内核中] P Y U k的定义g ! } S z如下：
typedef __kernel_fd_set fd_set;
#undef __FD_SETSIZE
#defineg ) j K H 2 l L __FD_SETSIZE 102* k X $ O 94
typR O @ H , 4 tedef struct {
unsigned long fds_bits;
} __kernel_fd_set;我们来简化下，fd_set是一个struct, 其内部只有一个由16个元素组成的unsigne= 6 o e ( @ ? Jd long数组，这个数组一共可以表示16 64 = 1024位， 每c $ h B B : L t o一位用来表示一个 fd, 这也就是 select针对读、写或1 _ z D m X异常每一类最多只能有 1024个fd 限制的由来。
相关宏
下面这些宏定义在内核代码fs/select.c中
- FDS_BITPERLONG: 返回每个long有多少位，通常是64biH c { ~ Yts
#define FDS_BITPERLONG (8*sizeof(long))- FDS_LONGS(nr): 获取 nr 个fd 需要用几个long来表示
#define FDS_LONGS(nr) (((nr)+FDS_BITPERLONG-1)/FDS_BITPERLONG)- FD_BYTES(nr): 获取 nr 个fd 需要用 多少个字节来表示
#define FDS_BYTES(nr) (FDS_LONGS(nr)*sizeof(long))下面这些宏m j n g : ?可以在gcc源码中找到
- FDB F , I A ) M x_ZERO: 初始化一个fd_set
#define _H O L M_FD_ZERO(s) \\
do { \\
unsigned int __i; \\Z # 7 h o 8
fd_set *__arF S - e R R ! zr = (s); \\
for (__i = 0; __& Q Ii < sizeof (fd_set) / sizeof (__fd_mask); ++__i) \\
__FDS_BITS (__arr) = 0; \\
} while (0)R i o _ B r # 5 d将上面所/ 8 F P说的由16个元素组成的unsigned long数组每一个元素都设为 0;
- __FD_SET(d, s): 将一个fd 赋值到 一个 fd_se_ K r A 8 v jt
#define __FD_SET(d, s) \\
((void) (__FDS_BITS (s) |= __FD_MASK(d)))分三步：
1、_9 | X 4 q U % Y_FD_ELT(d)：确定赋值到数组的哪一个元素
#defiC r 1 o E A ) Hne __FD_ELT(d) \\
__extE $ x Z Uension__ \\
({ long int __d = (d); \\
(__builtin) : w G ~_constG ! I o n M X !ant_p (__d) \\
? (0 <= __d && __d < __FD_SE] Q _ Y tTSIZE \\
? (__d / __NFDBITS) \\
: _/ P Y P ( 2 m_fdelt_warn (__d)) \\
: __fdelt_chk (__d) # P 8 q); })其中 #define __NFDBITS (8 * (int) sizeof (__fd_mask)) ， 即__NFDBITS = 64
这里实现使用了__builtin_constant_p针对常量作了优化，我也没有太理解h Q ` U常量与非常量实现方案有什么不同，我们暂时忽略这个细节看本质。
本质就是 一个 unsigned long有64位，直接 __d / __NF, C 5 nDBITS取模就可以确定用数组的哪一个元素了；
2、__FD_MASK(d)：确定赋值到一个 uq T X R F 1nsigned lo| 5 ~ Png的哪一位
#define __FD_MASK(d) ((__fd_mask) (1ULs I W << ((d) % __NFDBITS)))直接 （d) % __NFDBITS)取余后作为 1 左移的位数即可
3、|= ：用位或赋值即可
在内核中的实现
调用层级
系统调用入w 5 q j 3 ^口位置
位于f@ 4 . C 2s/select.c中
SYSCALL_DEFINE5(select, int, n, fd_set __user *, inp, fd_set __us0 p $ C 3 x i ] Der *, outp,
fd_set __user *, exp, struct timevalY H N 3 E , ~ __user *, tvp)
{
retb z . u y C 8 Durn kern_select(n, inp, outp, exp, tvp);
}入口函数 kern_select
static intg ? $ X { h kern_select(int n, fd_set __user *inp, fd + l g T t o H 1_set __user *outp,
fd_set __user *exp, struct timeval __userM z r *tvp)Y 1 ~ j O
{
struct timespec64 end_time, *to = NULL;
struct timeval tv;
int ret;
if (tvp) {
if (copy_from_$ 1 & f ; 0 Tuser(&tv, tvp, sizeof(tv)))
rJ S ~ Meturn -EFAULT;
to = &end_time;
if (poll1 Q * , % O ` ? O_select_set_timeout(to,
tv.tv_sec + (tv.tv_usec / USEC_PERj ) &_SEC),
(tv.tv_usec % USEC_PER_SEC) * NSEC_PER_USEC))
return -EINVAL;
}
ret = core_sys_select(nK ? s R O R s = n, inp, outp, exp, to);
return poll_select_fin0 C 1 G q )ish(&eO Q T ~nd_time, tvp, PT_TIMEVAf c @，这个是具体的实现，我们下面会重点介绍c. poll_select_finish：做的主要工作就是更新用户调用sele4 = S k 4 Ect时传进来的超时参数tvp，我列一下关键代码：
ktime_get_ts64(&rts);
rts = timespec64_sub(*end_timK / d W & c 1 R Be, rts);
iR % F f ( p f 1f (rts.tv_H X }sec < 0)
rts.tv_sec = rts.tv_nsec = 0;
...
struct timeval rt n - z ?v;m J 5 #
if (sizeof(rtv) > sizeof(rtv.tv_seQ L :c9 = P * n - z o e) + sizeof(rtv.tv_usec))
memset(&rtv, 0, sizeof(rtv));
rtv.tv_sec = rts.c t 3 x Qtv_sec;
rtv.tv_usec = rts.tv_nsecS A q i L | y = / NSEC_PER_USEC;
if (!copy_to_user(p, &rtv, sizeoa E Cf(rtv)))
return ret;
可以看到先获取当前的时间戳，然后通过timespec64_sub和传入的时间戳（接中传入的是超时时间，实现时会转化为时间戳）求出差值，将此差值传回给用户，即返回了剩余的超时时间。所以这个地方是个小陷阱，用户在调用select时，需要每次重新初始化这个超时时间。
通过 ca ( * =ore_sys_select 实现
这个函数主要功能是在实现真正的select功能前，准备好 fd_set ，即从用户空B ` u W Q 6间将所需的三类 fd_set 复制到内核空间。从下面的代码中你会看到对于每次的 select系统调用，都需要从用户空间将所需的三类 fd_set 复制到内核空间，这里存在性能上的损耗。
int cor~ , X l ~ l x Le_sys_s4 l 1 $ ^elect(int n, fd_set __user *iI R P = M m np, fd_set __user *outp,
fd_set __user *exp, struct timespec64 *end_time)
{
fd_set_bits fds;
voidx m s X 6 C *bits;
int ret, max_fds;
size_t size, alloc_size;
struct fdtable *y 2 o + Y % rfdt;
/* Allocate smV _ o = 2 / b Mall arguments on the stack to save memory and be faster */
long stack_fds;
ret = -EINVAL;
if (n < 0)
goto out_nofds;
/* max_fdss . * N O can increase, so grg F ) ! % u X lab it once to avoid race */
rcu_re% x H ` H ~ z 7ad_lock();
fdt = files_fdtable(current->files);
mg 0 h d y 8ax_fds = fdt->max_fds;
rcu_read_unlock();
if (n > max_fds)
n = max_fds;
/*: s R v e M y B
* We need 6 bitmaps (in/out/ex for both incoming and outgoing),
* since we used fdset we need to alld 2 Uocate memory- u f : 1 8 _ k in units of
* long-words.
*/
size = FDS_BYTES(n);
bits = stack_fds;
if (size > sizeof(stack_fds) / 6) {
/* Not enough space in on-s/ b b l D Vtack array; must use kmalloc */
ref | n B B $t = -ENOMEM;
if (sio B Y m = r R pze > (SIZE_MAX / 6))
goto out_nofds;
alloc_size = 6 * size;
bits = kvmallocF 3 = 0 (. B y 0 salloc_size, GFP_KERNEL);
if (!bits)
goto out_nofe E Fds;
}
fds.in = bits;
fds.out = b! _ G 2its + size;
fds.eD S E Lx = bits + 2*size;
fds.res_in = bits + 3*size;
fds.res_out =F W = I * A 3清晰，我们这里简单过一下：
分五步：
1、规范化select系统调用传入的第一个参数 n
/* max_fds ca9 ; ! 5 b J N Vn increase, so grab it once to avoid race */
rcu_reB * 9 X Z { O )ad_lock();
fdt = files_fdtable(current->files);
max_fds = fR 9 x E / D v gdt->max_fds;
rcu% & ) 1 7 5 +_read_unlt K T Iock();
if (n > max_fds)
n = max_fds;
这个n是三类不同的fd_set中所包括的fd数值的最大值 + 1， linux task打开句柄从0开始，不加1的话可能会少; U a监控fd.
用户在使| B ^ g I B用时可以有个偷懒的做法，就是将这个n设置t P / G K e k :为 FD_SETSIZE，2 3 e U B C f R通常是k d 1 / G 0 V @ o S Wt to the highesG W P j 9 # ~t-numbered file deN : ? a P h X y =scriptor in any of the three sN i ( v 6 * V bets, plus 1. The indicated file descriptors in each set are checked, up to this limit (but se# } J I j % (e BUGS).
2、计算内核空间所需要的fd_set的空间, 内核态需要三个C M rfd_set来容纳用户态传递过来的参数，还需要三个fd_set来L P T容纳select调用返回后生成的三灰fd_set,` | m u V 1 ) 即一共是6个fdZ @ o ~ * [_set
long stack_fds[SEq ;
. . .
size = FDS_BYTES(n);
bits = stack_fds;
if (size >a w w p j; sizeof(stack_fds) / 6) {
/A ) * 3 %* Not enough space in on-stack array; must use kmalloS c D ^c */
ret = -ENOMEM;
if (se f zize > (SIZE_MAX / 6))
goto out_nofds;
alloc_size = 6 * size;
bits = kvmalloc(alloc_size, GFP_KERNEL);
if (!bits)
goto out_nofds;
}
fds.in = bits;
fds.out = bits + size;
fds.ex = bits + 2*sizN E v - . we;
fds.res_in = bits + 3*size;
fds.res_out = bits + 4*size;
fds.res_ex = bits + 5*size;
这g S t U _ y里有个小技巧，先从O u R o ^ h内核栈上分配空间，如果不够用，才使用 k6 e @ jvmalloc分配。
通过 size = FDS_BYTES(n);计算出单一一种fd_set所需字节数，
再能过 allc b X zoc_sizl C 4 / ze = 6 * size; 即可计算出所需的全部字节数。
3、初始化用作参数的和用作返回值的两类fd_set
if ((ret = get_fd_set(n, in{ W } $ ` X | V 5p, fds.in)) ||
(ret = get_fd_set(n, outp,J o $ j + G fds.out)) ||- 3 . W X w
(ret = get_fd_set(n, exp, fds.eS O @x)))
goto out;
zero_fd_set(n, fds.res_in);
zero_fd_set(n, fds.res_out);
zero_fd_set(n S u p ; o G e, fds.res_ex);
4、真正实现部分 do_select， 我们在下面详讲
5、返回结果复制回用户! L = L空间
if (set_fd_set(n, inp, fds.res( % z 4 b k_in) ||
set_fd_set(n, outp, fds.res_out) ||
set_fd_set(n, exp, fds.res_ex))
ret = -EFc P m 9 DAULT;
这里又多了一次内核空间到用户空间的copy，而且? D $ O M _ x * `我们看到返回值也是用fd_set结构来表示，这意味着我们在用户空间处理里也需W J y要遍历每一位。
- 精华所在 do_select
wait queue
这里用到了Linux里一个很重要的数据结构 wait queue, 我们暂不打算展开来讲，先简单来说下其用法，比如我们在进程中read时经常要D ] E 4 %等待数据准备好，我们用伪码来写些流程：
// Read 代码
for (true) {
if 数据准备好 {
拷贝数据到用户空间buffer
return
} else {
创建一个 wait_queue_entry_t wait_entry;
wait_entry.f_ % j a z 5 9 o #unc = 自定义函数，被唤醒时会调用
wait_entry.private = 自定义的数据结构
将此 wait_entry 加入要读取数据的 设备的等待队列
set_currY W - % d y _ent_stata ; 4 : c v 4 *e(TASK_INTERRUPTIBLE) // 将当前进程状态设置为 TASK_INTERRUPTIBLE
schC V R - h 3 M # Bedulf l d 7 ie() // 将当前进程调度走，进行进程v D o % p `切换
}
}
// 设备驱动端代码
if 设备有数据可读 {
for ( 遍历其wait_queue ) {
唤醒 每一个 wait_queue_entry
调用 wait_entry.func {
将上面读取进程状态设置为 TASK_RUNNING，并加入CPUY f m O x J / c核的运行队列，被再次调度后，将读取到数据
}
}
}
do_select源码走读
- 获取当前三类fd_set中最大的fd
rcu_read_lock();
retval = max_select_fd(n, fds);
rcu_read_unlock();
n = retval;
上面 n = retval中的 n, 即为三类fd_set中最大的fd, 也是下面要介绍的循环体的上限
- 初始化( . F , $ T # e用作wait queue中的 wait e5 Y Tntry 的privat- D Qe数据结构
poll_initwait(&table);
- 核心循环体
我们讲注释写在这个代码块里
// 最外面是一个无限循环，它只有在poll到有效的事件，或者超时，或者有中断发生时，才会退出
fq z 7 5 v 5 J 5or (;;) {
u| 8 , # 5 V T gnsigned long *rinp, *routp, *rexp, *inp, *outp, *exp;
bool can_busy_loop = false;
// 首先获取需要监控的三类fd_set, 其实是 unsigned long 数组
inp = fds->in; outp = fds->out; exp = fds->ex;
// 初始化用于保F ? Q 7 p c存返回值的三类 fd_set对应的unsigned long 数组
rinp = fds->res_in; routp = fds->res_out; rexp = fds->res_ex;
// 开始循环遍历覆盖的所$ H f O w G有fd, 以上面得到的 n 上限
for (i = 0; i < n; ++J y 9 S H i 3 0 @rinp} r k u h, ++routp, ++rexp) {
unsigned long in, out, ex, all_bits, bit = 1, j;
unsigned long res_in = 0, res_out = 0, res_ex = 0;
__poll_t mask;
inx l 4 @ = *inp++; out = *outp++; ex = *exp++;
all_bits = in | out | ex;
if (all_bits == 0) {
// 如果走到这里，说明在三类fd_set的数组中，与当前下标对应的三个unsigned long的每一位均为0, 即当前
// 不存在任1 9 a , K D h T何监控的fd, 开始下一次循环
i += BITS_PER_LONG;
continue;
}
// 前z X W & N D面介绍过 fd_set的数组元素是 unsigned long, 即一个元素可表示64个fd, 这里依次遍历这64个bits
for (j = 0; j < BITS_PER_LONG; ++j, ++i, bit <<= 1) {
struct fd f;
if (d & h D ? * Z _ 7i >= n)E F ( v y a b H
break;
if (!(bit & all_bits))
co{ s @ . R | Y _ntinue;
// 走到这里，说明当前bit位上有监控的fd
f = fdget(i);
if (f.file) {t 5 { P _ v
// 针对当前fd, 设置其需要监控的事件
wE s W l w j q Fait_key_set(wait, in, ou6 8 + # c L 5 ot, bit,
busy_fD M N Mlag);
// vfs_pos V 7 a ` b X }ll 是重中之重，我们下一节会单独讲解，这里先说明它所作的事
// 1. 初始化wait entry, 将其加入到这个fd对应的soo X 7 C 6cket的等待队列中
// 2. 获取当前O 1 F Tsocket是否有读，写，异常等事件并返回
mask = vfs_poll(f.file, wai , L 4 p h t ft);
fdput(f);
// 按位与，看是否有相关事件
if ((mask & P+ : N 9OLLIN_SET? r q @ w p 4 y)l 2 * $ _ && (in & bit)) {
res_in |= bit;
retval++;
wait->_qproc = NULL;
}
if ((mask} ` R & POLLOUT_SET) && (out & bit)) {
res_out |= bit;
retval++;
wait->_qproc = NULL;
}
if ((mask & POLLEX_SET) && (ex & bit)Q ! ? f r) {
res_ex |= bit;
retval++;
wait-&G B 0 F :gt;_qproc = NULL;
}
/* got s0 P w omething, stop busy polling */
if (retval) {
can_busy_loop = false;
busy_flag = 0;
/*
* only remember a reL F % @ = m q mask)
can_busy_loop = true;
}
}
// 按unsigned lL N O ( |ong赋值给返回值数组元素
if (res_in)
*rinp = res_in;
if (res_out)
*routp = res_out;
if (res_ex)
*rexp = res_ex;
// 这里有两层for循环，这里主动出让CPU, 进行一V k Y 2 s d w次调度
cond_resched();
}
wait->_qprocn ( | . T ) C 0 o = NULL;
// 四种情况下会返回
//h a S 5 1. 任意监t y (控的fd上有事件发生
// 2C ^ U. 超时
// 3. 有中断发生
// 4. wait queue相关操作发生错误
if (retval || timed_out || signal_pending(currentZ W O H @ g))
break;
if (table.error) {
retval = table.error;
break* D V s *;
}
......
//3 5 ? a 当前监控的fd上没有事件发生，也没有超时或中断发生，
// 将当前进程设置为 TASK_INTERRUPTIBLE， 并调用 schedule
// 等待事件发生时，对应的socket将当前进程唤醒后，从 这里
// 继续s 8 d X 6 j L S运行
if (x 5 R e m W c e!pollr : y M i O D 3_schedule_timeout(&table, TASK_INTERRUPTI@ g : A d ?BLE,
to, slack))
timed_out = 1;
}
简单总结如下：
a. 循环遍历每一个监控的fd；
b. 有下列情况之一v r n S p则返回：
1. 任意监控的fd上有事件发生
2. 超时
3. 有中断发生
4. wait queue相关操作发生错误} 5 p ( V 5
c. 如查无上述情况，将当前进程设置为 TASK_Ib 1 Y ( lNTERRUPTIBLE， 并调用 schedule作进程切换；
d. 等待socket 事件发生，对应的socket将当前进程唤醒~ : b n ~ u Y后，当前进程被再次调度切换回来，3 6 y W j S |继续运行；
细心的你可能已经发现，这个有个影响效率的问题：即使只有一个监控中的fd有事件发e s生，当P 3 h , ] $ d K p前进F # -程就会被唤醒，然后要将所有监控的fd都遍历一边，依次调用vfQ D ms_poll来获取其有效事件，好麻烦啊～～～
vfs_poll 讲解
- 调用层级
- 作用：
1. 初始化wait entry, 将其加入到这个fd对应的socket的等待队列中
2. 获取当前socket是否有读，写，异常等事件并返回
- 加入等待队列时，最终会调用 fs_select.c中的 __pollwait
static void __pollwait(struct file *filpQ 7 k ] P U, wait_quE D * R J ` ^eue_head_t *wait_address,
poll_table *p)u 5 5 f o
{
struct poll_wqueues *pwq = container_/ Q m H ~ v iof(p, struct poll_wqR | rueues, p% , r Z / O :t);
struct poll_table_entry *entb ! e D Rry = poll_get_entry(pwq);
if (!entry)
return;
entry->filp = get_file(filp);
entry->wait_R Q K j T Raddress = wait_address;
entry->key = p->_key;
init_waitqueue_func_entry(&entry->wait, pollwake);
entry->wait.private = pwq;
// 加入到socket的等待队列
add_waitj E A_queue(wait_address, &entry->wait);
}
总结
- select调用中每类fd_set中最多容纳1024个f K . 7 Vd；
- 每次调用select都需要将三类fd_set从用户空间复制到内核空间；
- wait queue是个好东西，select会被当前进程task加入到每一个监控的socket的等待队列；
- select进程被唤醒后V & k d N U &即使只有一个被监控的fd有事件发生，也会再次将所有的监控fd遍历一次；
- 在遍历fd的过程中会调用cond_resched（）来主动出让CPU, 作进程切换；
关于360技术：360技术是360技术团队打造的技术分享公众号，每天推送技术干货内容，更多技术信息欢迎关注“360技术”微信公众号