在Linux环境下select函数的初体验-清一色财经

在linux中, 主要的 IO复用方式中, 有epoll, poll 和select, 这次先来学习下select。select 能够同时监视多个文件描述符的变法, 也支持超时返回。

select介绍

在linux中, 主要的 IO复用方式中, 有epoll, poll 和select, 这次先来学习下select.

select 能够同时监视多个文件描述符的变法, 也支持超时返回.

先来看下select函数的定义

/*/usr/include/sys/select.h*/
externintselect(int__nfds,//最大文件描述符+1
fd_set*__restrict__readfds,//读状态文件集
fd_set*__restrict__writefds,//写状态文件集
fd_set*__restrict__exceptfds,//异常状态文件集
structtimeval*__restrict__timeout);//超时时间

如上图函数声明所示, 不管我们关注什么状态, 我们都应该把同一类状态的文件描述符存到同一个fd_set集合,以便select能够相应的位置打上标签, 以便后续我们来判断该文件描述符是否已经准备好

这些传递给select函数的参数, 将告诉内核:

我们需要监听的文件描述符
对于每个文件描述符, 我们所关心的状态 (读/写/异常)
我们要等待多长时间 (无限长/超时返回)

而内核也会通过select的返回, 告知我们一些信息:

已经准备好的文件描述符个数
那三种状态分别是哪些文件描述符

我们可以通过以下方式将关注的文件描述符加入相应的文件集:

intsocket_test;
socket_test=socket(...);//创建socket文件描述符
connent(socket_test,..);//连接服务端
FD_SET(socket_test,&rdfds);//加入读状态文件集
FD_SET(socket_test,&wdfds);//加入写状态文件集
....

select原理

select函数执行顺序是: SYSCALL_DEFINE5 (sys_select) -> core_sys_select -> do_select

我们都知道, select 支持监听三个文件集: 读文件集, 写文件集, 异常文件集;

在我们调用FD_SET(socket_test, &rdfds)时, 实际上执行的操作是: 在rdfds成员数组中, 将__FDELT (d)位置的值设成 __FDMASK (d), 直接说会有点疑惑, 先看下相关的函数,宏定义是怎样定义的吧:

/*取自:/usr/include/sys/select.h*/
#defineFD_SET(fd,fdsetp)__FD_SET(fd,fdsetp)
typedeflongint__fd_mask;
/*取自:/usr/include/bits/select.h*/
#define__NFDBITS(8*(int)sizeof(__fd_mask))
#define__FDELT(d)((d)/__NFDBITS)
#define__FDMASK(d)((__fd_mask)1<<((d)%__NFDBITS))
#define__FD_SET(d,set)(__FDS_BITS(set)[__FDELT(d)]|=__FDMASK(d))
typedefstruct
{
/*XPG4.2requiresthismembername.Otherwiseavoidthename
fromtheglobalnamespace.*/
#ifdef__USE_XOPEN
__fd_maskfds_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;
/*/usr/include/linux/posix_types.h*/
#define__FD_SETSIZE1024

举个栗子, 假设 fd=3, 当我们执行FD_SET(fd, &rdfds)时:

算出 __FDELT(d) 和 __FDMASK(d)的值, 通过上面的宏定义, 可以分别得出结果: 3/(8*8), 1<<3%(8*8), 也就是0 和二进制的 0000 0100
然后分别将值存入 rdfds.__fds_bits第0个位置, 值为十进制的8
我们可以将__fds_bit的每个索引看成是一个聚合的过程, 每个值8字节, 也就是有64位, 可以存64个fd, 在我的系统上, 算出数组的长度是__FD_SETSIZE / __NFDBITS = 1024/8=128个, 也就是大概能容纳 128*64=8192(如果理解错误请指出)

经过上面的运算, 我们将需要关注的文件描述关联到 rdfds文件集了, 对于写文件集, 异常文件集都是同样的运算, 等这些步骤都进行完了, 接下来就是进入core_sys_select函数了:

执行到 core_sys_select 时, 定义一个fd_set_bits结构体: fds.
分别为fds的成员(in, out, ex, res_in, res_out, res_ex)申请内存
将我们传给select的 rdfds, wrfds, exfds分别赋值给 in, out, ex, 这样fds就能记录三个集合的结果了
初始化那个三个成员之后, 将执行do_select(n, &fds, end_time)
在do_select中, 函数将进入死循环,其中还有两个循环, 分别是针对 "最大文件描述符数" 和 fd_set_bits数组中单个值位数. 从上面我们已经知道, 在fd_set_bits每个值都代表所关注的文件描述符, 每个值是__NFDBITS(8 *8字节)大小,也就是64位, 所以在上面循环内, 还要再循环64次
看到这里其实大家都应该有个底了, 为什么要循环那么多次, 因为我们需要通过每个文件描述符对应的file_operations结构体的接口f_op->poll来得知是否已经准备好了

简单介绍 file_operations

我们都知道,当我们打开一些设备或者文件时, 总是返回一个文件描述符, 其实通过这个文件描述符, 我们通过fget_light 来获得对应的file结构体, 为什么还要反查这个file, 因为通过这个file结构体可以得到: file_operations结构体

file_operations结构体: 用来存储驱动内核模块提供的对设备进行各种操作的函数的指针。该结构体的每个域都对应着驱动内核模块用来处理某个被请求的事务的函数的地址。

/*linux-2.6.32/include/linux/fs.h*/
structfile_operations{
...
unsignedint(*poll)(structfile*,structpoll_table_struct*);//select通过这个来获取状态
...(其他接口忽略)

do_select 循环体源码:

/*select.c/do_select()*/
for(;;){
unsignedlong*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){
unsignedlongin,out,ex,all_bits,bit=1,mask,j;
unsignedlongres_in=0,res_out=0,res_ex=0;
conststructfile_operations*f_op=NULL;
structfile*file=NULL;
in=*inp++;out=*outp++;ex=*exp++;
all_bits=in|out|ex;
if(all_bits==0){
i+=__NFDBITS;
continue;
}
for(j=0;j<__NFDBITS;++j,++i,bit<<=1){//遍历64位
intfput_needed;
if(i>=n)
break;
if(!(bit&all_bits))
continue;
//在当前进程的structfiles_struct中根据所谓的用户空间文件描述符fd来获取文件描述符
file=fget_light(i,&fput_needed);
if(file){
f_op=file->f_op;//file_operations结构体
mask=DEFAULT_POLLMASK;
if(f_op&&f_op->poll){
wait_key_set(wait,in,out,bit);
mask=(*f_op->poll)(file,wait);
}
fput_light(file,fput_needed);
if((mask&POLLIN_SET)&&(in&bit)){//判断读状态
res_in|=bit;
retval++;
wait=NULL;
}
if((mask&POLLOUT_SET)&&(out&bit)){//判断写状态
res_out|=bit;
retval++;
wait=NULL;
}
if((mask&POLLEX_SET)&&(ex&bit)){//判断异常状态
res_ex|=bit;
retval++;
wait=NULL;
}
}
}
if(res_in)
*rinp=res_in;
if(res_out)
*routp=res_out;
if(res_ex)
*rexp=res_ex;
cond_resched();
}
wait=NULL;
if(retval||timed_out||signal_pending(current))
break;
if(table.error){
retval=table.error;
break;
}
/*
*Ifthisisthefirstloopandwehaveatimeout
*given,thenweconverttoktime_tandsettheto
*pointertotheexpiryvalue.
*/
if(end_time&&!to){
expire=timespec_to_ktime(*end_time);
to=&expire;
}
if(!poll_schedule_timeout(&table,TASK_INTERRUPTIBLE,
to,slack))
timed_out=1;
}

当select经历完上面的流程, 将会有以下结果:

>0: 准备好的文件描述符个数
0: 超时
-1: 出错或者接收到信号

那我们接下来要做的就是,

通过行FD_ISSET()判断之前绑定的文件fd, 如果为真, 则进行相应操作
因为select返回后, 之前存好的rdfds, wdfds, exfds都会被清空, 所以需要用FD_SET()重新加入

select实战

上面已经学习到关于select的相关知识, 那么我们应该要来实战下:

这次我们需要实现的目标是:

一个程序, 同时连接3个socket_server, 并且将socket_server发送的消息打印出来(不需要响应, 也不需要交互)

程序代码:

/*filename:test_select.c*/
#include<stdio.h>
#include<stdlib.h>
#include<sys/socket.h>
#include<sys/select.h>
#include<sys/types.h>
#include<netinet/in.h>
#include<unistd.h>
#include<fcntl.h>
voidmain()
{
//socket1
intsocketd;
charbuffer[1025];
structsockaddr_inseraddr;
socketd=socket(AF_INET,SOCK_STREAM,0);
seraddr.sin_family=AF_INET;
seraddr.sin_port=htons(9997);
inet_pton(AF_INET,"127.0.0.1",&seraddr.sin_addr);
if(connect(socketd,(structsockaddr*)&seraddr,sizeof(seraddr))<0)
{
printf("socketd1connectfailed\n");
exit(3);
}
//socket2
intsocketd2;
charbuffer2[1025];
structsockaddr_inseraddr2;
socketd2=socket(AF_INET,SOCK_STREAM,0);
seraddr2.sin_family=AF_INET;
seraddr2.sin_port=htons(9998);
inet_pton(AF_INET,"127.0.0.1",&seraddr2.sin_addr);
if(connect(socketd2,(structsockaddr*)&seraddr2,sizeof(seraddr))<0)
{
printf("socketd2connectfailed\n");
exit(3);
}
//scoket3
intsocketd3;
charbuffer3[1025];
structsockaddr_inseraddr3;
socketd3=socket(AF_INET,SOCK_STREAM,0);
seraddr3.sin_family=AF_INET;
seraddr3.sin_port=htons(9999);
inet_pton(AF_INET,"127.0.0.1",&seraddr3.sin_addr);
if(connect(socketd3,(structsockaddr*)&seraddr3,sizeof(seraddr))<0)
{
printf("socketd3connectfailed\n");
exit(3);
}
intmaxfdp;
fd_setfds;//select需要的文件描述符集合
maxfdp=socketd3+1;//select第一个形参就是打开的最大文件描述符+1
structtimevaltimeout={3,0};//超时设置
while(1)
{
FD_ZERO(&fds);//初始化文件描述符集合
FD_SET(socketd,&fds);//分别添加以上三个需要监听的文件描述符
FD_SET(socketd2,&fds);
FD_SET(socketd3,&fds);
select(maxfdp,&fds,NULL,NULL,&timeout);
//通过FD_ISSET来分别判断监听的文件描述符在fds有没有被设置成1
if(FD_ISSET(socketd,&fds))
{
read(socketd,buffer,1024);
printf("1%s\n",buffer);
}
if(FD_ISSET(socketd2,&fds))
{
read(socketd2,buffer2,1024);
printf("2%s\n",buffer2);
}
if(FD_ISSET(socketd3,&fds))
{
read(socketd3,buffer3,1024);
printf("3%s\n",buffer3);
}
}
}

为了快速建立简单的测试服务端, 所以用python实现简单socket_server:

#socket1.py
importsocket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind(('localhost',9997))
s.listen(2)
rint'Socket1isonready!'
client,info=s.accept()
printinfo
while1:
message=raw_input('input:')
client.send(message)
s.close()
-------------------------------
#socket2.py
importsocket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind(('localhost',9998))
s.listen(2)
rint'Socket2isonready!'
client,info=s.accept()
printinfo
while1:
message=raw_input('input:')
client.send(message)
s.close()
-------------------------------
#socket3.py
importsocket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind(('localhost',9999))
s.listen(2)
rint'Socket3isonready!'
client,info=s.accept()
printinfo
while1:
message=raw_input('input:')
client.send(message)
s.close()

分别运行 socket1.py, socket2.py, socket3.py将会看到如下结果:

#运行socket1.py
[root@iZ23pynfq19Z~]#pythonsocket1.py
Socket1isonready!
----------------------
#运行socket1.py
[root@iZ23pynfq19Z~]#pythonsocket2.py
Socket2isonready!
----------------------
#运行socket1.py
[root@iZ23pynfq19Z~]#pythonsocket3.py
Socket3isonready!

当我们编译 test_select.c 并运行时, 将会看到三个服务端都出现了相应的响应:

#运行socket1.py
[root@iZ23pynfq19Z~]#pythonsocket1.py
Socket1isonready!
('127.0.0.1',55951)#客户端连接的信息,端口不一定相同
input:
----------------------
#运行socket1.py
[root@iZ23pynfq19Z~]#pythonsocket2.py
Socket2isonready!
('127.0.0.1',55921)
input:
----------------------
#运行socket1.py
[root@iZ23pynfq19Z~]#pythonsocket3.py
Socket3isonready!
('127.0.0.1',55933)
input:

那么我们来尝试三个服务端分别发送消息到select程序吧:

#socket1.py
[root@iZ23pynfq19Z~]#pythonsocket1.py
Socket1isonready!
('127.0.0.1',55951)#客户端连接的信息,端口不一定相同
input:asd
input:qwe
input:as
input:
----------------------
#socket1.py
[root@iZ23pynfq19Z~]#pythonsocket2.py
Socket2isonready!
('127.0.0.1',55921)
input:asd
input:asd
input:asd
input:as
input:s
input:
----------------------
#socket1.py
[root@iZ23pynfq19Z~]#pythonsocket3.py
Socket3isonready!
('127.0.0.1',55933)
input:asd
input:qwe
input:a
input:

将看到select程序都能输出三个socket_server发出的消息:

需要注意的是:

前面的数字是socket_server的编号, 因为server发送消息的顺序是乱的, 所以输出的编号也是乱的
这次只为验证select, 所以并没对程序的健壮性作较好的设计, 所以如果服务端/客户端刷屏了, 直接ctrl-c终止吧

经过上述的实验, 我们应该能够简单的了解select的用法和效果, 通过select实现IO多路复用, 可以让我们一定程度上避免多线程/多进程的繁琐, 在我们日常工作上, 有必要的话尝试这种方式也不失一种偷懒的方法.

©本文为清一色官方代发，观点仅代表作者本人，与清一色无关。清一色对文中陈述、观点判断保持中立，不对所包含内容的准确性、可靠性或完整性提供任何明示或暗示的保证。本文不作为投资理财建议，请读者仅作参考，并请自行承担全部责任。文中部分文字/图片/视频/音频等来源于网络，如侵犯到著作权人的权利，请与我们联系（微信/QQ:1074760229）。转载请注明出处：清一色财经