欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

I/O多路复用:select、poll和epoll详解

程序员文章站 2022-06-13 13:02:15
...

I/O多路复用

I/O复用使得程序能同时监听多个文件描述符,这对提高程序的性能至关重要。通常,网络程序在下列情况下需要使用I/O复用技术:

客户端程序要同事处理多个socket。比如将讨论的非阻塞connect技术。

客户端程序要同时处理用户输入和网络连接。

TCP服务器要同时处理监听socket和连接socket。这是I/O复用使用最多的场合。

I/O复用虽然能同时监听多个文件描述符,但它本身是阻塞的。并且当多个文件描述符同时就绪时,如果不采取额外的措施,程序就只能按顺序一次处理其中的每一个文件描述符,这使得服务器程序看起来像是串行工作的。如果要实现并发,只能使用多进程或多线程等编程手段。Linux下实现I/O复用的系统调用主要有select、poll和epoll。

 

select

select系统调用的用途是:在一段指定时间内,监听用户感兴趣的文件描述符上的可读、可写和异常事件。内核通过对这些参数在线修改来反馈其中的就绪事件。每次调用select都要重置这3个参数。函数原型:

#include <sys/select.h>
int select(nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, &timeout);

nfds参数指定被监听的文件描述符的总数。

select将事件分为可读(readfds)、可写(writefds)和异常(exceptfds),并将相应的事件放入对应的文件描述符集合中。

timeout参数用来设置select函数的超时时间,采用指针参数是因为内核将修改它以告诉应用程序select等待了多久。如果给timeout变量的成员都传递0,则select将立即返回。如果给timeout传递NULL,则select将一直阻塞,直到某个文件描述符就绪。

select成功时返回就绪文件描述符的总数。如果在超时时间内没有任何文件描述符就绪,select将返回0。select失败时返回-1并设置errno。使用以下函数完成对事件的操作:

void FD_CLR(int fd, fd_set *set);     //将fd从set上删除
void FD_ZERO(fd_set *set);            //将set中所有设置位清除
void FD_SET(int fd, fd_set *set);     //将fd添加到相应的集合上
int FD_ISSET(int fd, fd_set *set);    //判断fd是不是在set集合上

select工作原理:

I/O多路复用:select、poll和epoll详解

select示例代码:

FD_ZERO(&readset);              //清空一个文件描述符集合
FD_SET(new_sock, &readset);     //将一个文件描述符添加到一个指定的文件描述符集合
maxfd=new_sock+1;

while(1)  
{
    r_readset = readset;         //因为每次会修改传入的事件,所以用变量记录事件,每次调用select时需要传入变量r_readset,而不是readset。
    timeout.tv_sec = 0;
    timeout.tv_usec = 500000;
    //io复用
    if((nfound = select(maxfd,&r_readset,(fd_set *)0,(fd_set *)0,&timeout))<0){
        perror("select");
        continue;
    }
    else if(nfound==0){
        continue;
    }
    if(FD_ISSET(new_sock,&r_readset))//检查集合中指定的文件描述符是否可以读
    {
        //接收消息
        byteread=recv(new_sock,&msgs,sizeof(msgs),0);
        if(byteread<0) //recv出错
        {
            perror("recv:");
            break;
        }
        if(byteread==0) //连接中止,发送下线信息
        {
            break;
        }

    //完成相应的操作

    }//end if
}//end while

select的参数类型fd_set没有将文件描述符和事件绑定,它只是一个文件描述符集合,所以select需要提供3种类型的参数分别传入。由于内核对fd_set集合是在线修改的,应用程序下次调用select前需要重置这3个fd_set集合。工作流程:

(1)创建TCP连接,将 fd 添加到 fd_set 集合中;
(2)将 fd_set 集合从用户态拷贝到内核态;
(3)遍历这个 fd_set 集合,找出所有就绪的 fd,执行对应 fd 的相关操作;
(4)将内核态的 fd_set 集合拷贝到用户态

select的缺点:

(1)进程能够监视的文件描述符的数量存在最大限制,一般是1024,由于select采用轮询的方式扫描文件描述符,所以文件描述符数量越多,性能越差;

(2)内核 / 用户空间内存拷贝问题,select需要复制大量的句柄数据结构,产生巨大的开销;

(3)select返回的是含有整个句柄的数组,应用程序需要遍历整个数组才能发现哪些句柄发生了事件。

 

poll

poll系统调用和select类似,也是在一定时间内轮询一定数量的文件描述符,以测试其中是否有就绪事件。poll的原型如下:

#include <poll.h>
int poll(struct pollfd fds[], nfds_t nfds, int timeout);

fds:是一个struct pollfd结构类型的数组,用于存放需要检测其状态的socket文件描述符;每当调用这个函数之后,系统不会清空这个数组,操作起来比较方便;特别是对于socket连接比较多的情况下,在一定程度上可以提高处理的效率;poll()函数适合于大量socket描述符的情况;从下面的结构体可以看出,结构体pollfd将文件描述符和事件进行了绑定。

struct pollfd
{
    int fd;           //文件描述符
    short events;     //注册的事件
    short revents;    //实际发生的事件,由内核填充
}

poll事件类型:
POLLIN                有数据可读
POLLRDNORM            有普通数据可读
POLLRDBAND            有优先数据可读
POLLPRI               有紧急数据可读
POLLOUT               数据可写
POLLWRNORM            普通数据可写
POLLWRBAND            优先数据可写
POLLMSGSIGPOLL        消息可用 

nfds:nfds_t类型的参数,用于标记数组fds中的结构体元素的总数量;

timeout:是poll函数调用阻塞的时间,单位:毫秒;

返回值大于0:fds中准备好读、写或异常事件的socket文件描述符的总数量;

返回值等于0:fds中没有任何socket文件描述符准备好读、写,或异常事件;此时poll超时,超时时间是timeout毫秒;换句话说,如果所检测的socket文件描述符上没有任何事件发生的话,那么poll()函数会阻塞timeout所指定的毫秒时间长度之后返回,如果timeout==0,那么poll() 函数立即返回而不阻塞,如果timeout == INFTIM,那么poll() 函数会一直阻塞下去,直到所检测的socket文件描述符上的感兴趣的事件发生是才返回,如果感兴趣的事件永远不发生,那么poll()就会永远阻塞下去。

poll统一处理所有事件类型,因此只需一个事件集参数。用户通过pollfd.events传入感兴趣的事件,它是一系列事件的按位或,内核通过修改pollfd.revents反馈其中就绪的事件。可以使用pollfd.revents与事件类型按位与进行判断是否发生相应的事件。poll同样存在的问题:(1)内核/用户空间内存拷贝问题;(2)应用程序需要采用的轮询方式来检测就绪事件,算法时间复杂度是O(n)。相应代码会放到文章最后。

 

epoll

epoll是Linux特有的I/O复用函数。它在实现和使用上与select、poll有很大差异。首先,epoll使用一组函数来完成任务,而不是单个函数。其次,epoll把用户关心的文件描述符上的事件放在内核里的一个事件表中,从而无需像select和poll那样每次调用都要重复传入文件描述符集合或事件集合。但epoll需要使用一个额外的文件描述符,来唯一标识内核中的这个事件表。这个文件描述符使用epoll_create函数来创建。

#include <sys/epoll.h>
int epoll_create(int size);

size参数现在并不起作用,只是给内核一个提示,告诉它事件表需要多大。该函数返回的文件描述符将用作其他所有epoll系统调用的第一个参数,以指定要访问的内核事件表。

当进程调用epoll_create方法时,Linux内核会创建一个eventpoll结构体,这个结构体中有两个成员与epoll的使用方式密切相关。eventpoll结构体如下所示:

struct eventpoll{
    ....
    //红黑树的根节点,这颗树中存储着所有添加到epoll中的需要监控的事件
    struct rb_root  rbr;
    //双链表中则存放着将要通过epoll_wait返回给用户的满足条件的事件
    struct list_head rdlist;
    ....
};

 

epoll_ctl( )

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event); //用于操作epoll的内核事件表

fd参数是要操作的文件描述符,op参数则指定操作类型。操作类型有以下3种:

EPOLL_CTL_ADD,往事件表中注册fd上的事件。

EPOLL_CTL_MOD,修改fd上的注册事件。

EPOLL_CTL_DEL,删除fd上的注册事件。

event参数指定事件,它是epoll_event结构指针类型。epoll_event的定义如下:

struct epoll_event
{
    __uint32_t events;   //epoll事件
    epoll_data_t data;   //用户数据
}

其中events成员描述事件类型。epoll支持的事件类型和poll基本相同。表示epoll事件类型的宏是在poll对应的宏前加上E,比如epoll的数据可读事件是EPOLLIN。但epoll有两个额外的事件类型——EPOLLET和EPOLLONESHOT。它们对于epoll的高效运作非常键。data成员用于存储用户数据,其类型epoll_data_t的定义如下:

typedef union epoll_data
{
    void *ptr;
    int fd;
    uint32_t u32;
    uint64_t u64;
}epoll_data_t;

epoll_data_t是一个联合体,其中使用最多的成员是fd,它指定事件所从属的目标文件描述符。epoll_ctl成功时返回0,失败时返回-1并设置errno。

每一个epoll对象都有一个独立的eventpoll结构体,用于存放通过epoll_ctl方法向epoll对象中添加进来的事件。这些事件都会挂在红黑树上,如此,重复添加的事件就可以通过红黑树而高效的识别出来(红黑树的插入时间效率是O(lgn),其中n为元素个数)。

而所有添加到epoll中的事件都会与设备(网卡)驱动程序建立回调关系,也就是说,当相应的事件发生时会调用这个回调方法。这个回调方法在内核中叫ep_poll_callback,它会将发生的事件添加到eventpoll的rdlist双链表中。

在epoll中,对于每一个事件,都会建立一个epitem结构体,如下所示:

struct epitem{
    struct rb_node  rbn;           //红黑树节点
    struct list_head    rdllink;   //双向链表节点
    struct epoll_filefd  ffd;      //事件句柄信息
    struct eventpoll *ep;          //指向其所属的eventpoll对象
    struct epoll_event event;      //期待发生的事件类型
}

 

epoll_wait()

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

该函数成功时返回就绪的文件描述符的个数,失败时返回-1并设置errno。epoll_wait函数如果检测到事件,就将所有就绪的事件从内核事件表中复制到它的第二个参数events指向的数组中。这个数组只用于输出epoll_wait检测到的就绪事件,而不像select和poll的数组参数那样及用于传入用户注册的事件,又用于输出内核检测到的就绪事件。这就极大地提高了应用程序索引就绪文件描述符的效率。

当调用epoll_wait检查是否有事件发生时,只需要检查eventpoll对象中的rdlist双链表中是否有epitem元素即可。如果rdlist不为空,则把发生的事件复制到用户态,同时将事件数量返回给用户。

简单的归纳epoll的用法了:

(1)调用epoll_create(),返回一个句柄来唯一标识内核中的事件表。

(2)使用epoll_ctl()系统调用,向epoll对象中添加、删除、修改感兴趣的事件,返回0表示成功,返回-1表示失败。

(3)通过epoll_wait()系统调用获取就绪事件。

 

epoll对文件描述符的操作有两种模式:LT(Level Trigger,电平触发)模式和ET(Edge Trigger,边沿触发)模式。LT模式是默认的工作模式,这种模式下epoll相当于一个效率较高的poll。当往epoll内核事件表中注册一个文件描述符上的EPOLLET事件时,epoll将以ET模式来操作该文件描述符。ET模式是epoll的高效工作模式。

对于采用LT工作模式的文件描述符,当epoll_wait检测到其上有事件发生并将此事件通知应用程序后,应用程序可以不立即处理该事件。这样,当应用程序下一次调用epoll_wait时,epoll_wait还会再次向应用程序通告此事件,直到该事件被处理。而对于采用ET工作模式的文件描述符,当epoll_wait检测到其上有事件发生并将此事件通知应用程序后,应用程序必须立即处理该事件,因为后续的epoll_wait调用将不再向应用程序通知这一事件。可见,ET模式在很大程度上降低了同一个epoll事件被重复触发的次数,因此效率比LT高。

 

select、poll和epoll对比

这三组系统调用都能同时监听多个文件描述符。它们将等待由timeout参数指定的超时时间,直到一个或多个文件描述符上有事件发生时返回,返回值是就绪文件描述符的数量。返回0表示没有事件发生。

select的参数类型fd_set没有将文件描述符和事件绑定,它只是一个文件描述符集合,所以select需要提供3种类型的参数分别传入。由于内核对fd_set集合是在线修改的,应用程序下次调用select前需要重置这3个fd_set集合。

poll的参数类型pollfd将文件描述符和事件都定义在其中,任何事件都被统一处理,从而使得编程接口简洁得多。并且内核每次修改的是revents成员,而events成员保持不变,因此下次调用poll时不需要重置pollfd的事件集合参数。由于每次select和poll调用都返回整个用户注册的事件集合,所以应用程序找到就绪文件描述符的时间复杂度为O(n)。

epoll采用与select和poll完全不同的方式管理用户注册的事件。它在内核中维护一个事件表,并提供了一个独立的系统调用epoll_ctl来控制往其中添加、删除、修改事件。每次epoll_wait调用都直接从该内核事件表中取得用户注册的事件,而不用反复从用户空间读入这些事件。epoll_wait系统调用的events参数仅用来返回就绪的事件,这使得应用程序索引就绪文件描述符的事件复杂度为O(1)。

I/O多路复用:select、poll和epoll详解

 

代码实现

以下代码使用poll实现了一个简单聊天室的功能,该聊天室程序能让所有用户同时在线群聊,它分为客户端和服务器两个部分。其中客户端程序有两个功能:一是从标准输入读入用户数据,并将数据发送至服务器;二是往标准输出终端打印服务器发送给它的数据。服务器的功能是接收客户数据,并把客户数据发送给每一个登录到该服务器上的客户端。下面依次给出服务器端和客户端的代码。

#define _GNU_SOURCE 1   //server.c
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <fcntl.h>
#include <stdlib.h>
#include <poll.h>

#define USER_LIMIT 5
#define BUFFER_SIZE 64
#define FD_LIMIT 65535

struct client_data
{
    sockaddr_in address;
    char* write_buf;
    char buf[ BUFFER_SIZE ];
};

int setnonblocking( int fd )
{
    int old_option = fcntl( fd, F_GETFL );
    int new_option = old_option | O_NONBLOCK;
    fcntl( fd, F_SETFL, new_option );
    return old_option;
}

int main( int argc, char* argv[] )
{
    if( argc <= 2 )
    {
        printf( "usage: %s ip_address port_number\n", basename( argv[0] ) );
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi( argv[2] );

    int ret = 0;
    struct sockaddr_in address;
    bzero( &address, sizeof( address ) );
    address.sin_family = AF_INET;
    inet_pton( AF_INET, ip, &address.sin_addr );
    address.sin_port = htons( port );

    int listenfd = socket( PF_INET, SOCK_STREAM, 0 );      //1.创建socket
    printf("Listenfd:[%d]\n", listenfd);                   //每次都是3
    assert( listenfd >= 0 );
    //socket返回值是一个文件描述符,socket类型本身也是定义为int的,既然是文件描述符,那么在
    //系统中都当作是文件来对待。0,1,2分别表示标准输入、标准输出、标准错误。所以其他打开文件描述符都会大于2。

    ret = bind( listenfd, ( struct sockaddr* )&address, sizeof( address ) );   //2.bind
    printf("Bind ret:[%d]\n", ret);
    assert( ret != -1 );

    ret = listen( listenfd, 5 );                           //3.监听
    printf("Listen ret:[%d]\n", ret);
    assert( ret != -1 );

    client_data* users = new client_data[FD_LIMIT];
    pollfd fds[USER_LIMIT+1];           //文件描述符上可读、可写和异常事件
    int user_counter = 0;
    for( int i = 1; i <= USER_LIMIT; ++i )
    {
        fds[i].fd = -1;
        fds[i].events = 0;
    }
    fds[0].fd = listenfd;               //指定文件描述符
    fds[0].events = POLLIN | POLLERR;   //events告诉poll监听fd上的什么事件,这个例子就是监听可读和错误事件
    fds[0].revents = 0;                 //由内核填写,以通知fd上实际发生的事件

    while( 1 )
    {
        ret = poll( fds, user_counter+1, -1 );
        if ( ret < 0 )
        {
            printf( "poll failure\n" );
            break;  //跳出循环
        }
    
        for( int i = 0; i < user_counter+1; ++i )
        {
            if( ( fds[i].fd == listenfd ) && ( fds[i].revents & POLLIN ) )
            {   //监听的文件描述符是server创建的socket且有可读事件发生,即有客户端的连接。接下来完成accept相关工作。
                struct sockaddr_in client_address;
                socklen_t client_addrlength = sizeof( client_address );
                int connfd = accept( listenfd, ( struct sockaddr* )&client_address, &client_addrlength );    //accept
                printf("Connfd:[%d]\n", connfd);
                if ( connfd < 0 )
                {
                    printf( "errno is: %d\n", errno );
                    continue;
                }
                if( user_counter >= USER_LIMIT )
                {
                    const char* info = "too many users\n";
                    printf( "%s", info );
                    send( connfd, info, strlen( info ), 0 );
                    close( connfd );
                    continue;         //continue语句的作用是跳过本次循环体中余下未执行的语句,立即进入下一次循环条件判定。
                }
                user_counter++;
                users[connfd].address = client_address;
                setnonblocking( connfd );
                fds[user_counter].fd = connfd;
                fds[user_counter].events = POLLIN | POLLRDHUP | POLLERR;
                fds[user_counter].revents = 0;
                printf( "comes a new user, now have %d users\n", user_counter );
            } //if( ( fds[i].fd == listenfd ) && ( fds[i].revents & POLLIN ) ) end
            
            else if( fds[i].revents & POLLERR )
            {
                printf( "get an error from %d\n", fds[i].fd );
                char errors[ 100 ];
                memset( errors, '\0', 100 );
                socklen_t length = sizeof( errors );
                if( getsockopt( fds[i].fd, SOL_SOCKET, SO_ERROR, &errors, &length ) < 0 )
                {
                    printf( "get socket option failed\n" );
                }
                continue;
            }//else if( fds[i].revents & POLLERR ) end
            
            else if( fds[i].revents & POLLRDHUP )
            {//客户端退出是触发,POLLRDHUP:TCP连接被对方关闭,或者对方关闭了写操作
                users[fds[i].fd] = users[fds[user_counter].fd];
                close( fds[i].fd );
                fds[i] = fds[user_counter];
                i--;
                user_counter--;
                printf( "a client left\n" );
            }//else if( fds[i].revents & POLLRDHUP ) end
            
            else if( fds[i].revents & POLLIN )
            {
                int connfd = fds[i].fd;
                memset( users[connfd].buf, '\0', BUFFER_SIZE );
                ret = recv( connfd, users[connfd].buf, BUFFER_SIZE-1, 0 );
                printf( "get %d bytes of client data %s from %d\n", ret, users[connfd].buf, connfd );
                if( ret < 0 )
                {
                    if( errno != EAGAIN )
                    {
                        close( connfd );
                        users[fds[i].fd] = users[fds[user_counter].fd];
                        fds[i] = fds[user_counter];
                        i--;
                        user_counter--;
                    }
                }
                else if( ret == 0 )
                {
                    printf( "code should not come to here\n" );
                }
                else
                {
                    for( int j = 1; j <= user_counter; ++j )
                    {
                        if( fds[j].fd == connfd )
                        {
                            continue;
                        }
                        
                        fds[j].events |= ~POLLIN;
                        fds[j].events |= POLLOUT;
                        users[fds[j].fd].write_buf = users[connfd].buf;
                    }
                }
            }//else if( fds[i].revents & POLLIN ) end
            
            else if( fds[i].revents & POLLOUT )
            {
                int connfd = fds[i].fd;
                if( ! users[connfd].write_buf )
                {
                    continue;
                }
                ret = send( connfd, users[connfd].write_buf, strlen( users[connfd].write_buf ), 0 );
                users[connfd].write_buf = NULL;
                fds[i].events |= ~POLLOUT;
                fds[i].events |= POLLIN;
            }
            
        }//for end
    }//while end

    delete [] users;
    close( listenfd );
    return 0;
}
#define _GNU_SOURCE 1       //client.c
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <poll.h>
#include <fcntl.h>

#define BUFFER_SIZE 64

int main( int argc, char* argv[] )
{
    if( argc <= 2 )
    {
        printf( "usage: %s ip_address port_number\n", basename( argv[0] ) );
        return 1;
    }
    const char* ip = argv[1];
    int port = atoi( argv[2] );

    struct sockaddr_in server_address;
    bzero( &server_address, sizeof( server_address ) );
    server_address.sin_family = AF_INET;
    inet_pton( AF_INET, ip, &server_address.sin_addr );
    server_address.sin_port = htons( port );

    int sockfd = socket( PF_INET, SOCK_STREAM, 0 );
    printf("Sockfd:[%d]\n", sockfd);
    assert( sockfd >= 0 );
    if ( connect( sockfd, ( struct sockaddr* )&server_address, sizeof( server_address ) ) < 0 )
    {
        printf( "connection failed\n" );
        close( sockfd );
        return 1;
    }

    pollfd fds[2];
    fds[0].fd = 0;       //标准输入
    fds[0].events = POLLIN;
    fds[0].revents = 0;
    fds[1].fd = sockfd;
    fds[1].events = POLLIN | POLLRDHUP;
    fds[1].revents = 0;
    char read_buf[BUFFER_SIZE];
    int pipefd[2];
    int ret = pipe( pipefd );
    assert( ret != -1 );

    while( 1 )
    {
        ret = poll( fds, 2, -1 );
        if( ret < 0 )
        {
            printf( "poll failure\n" );
            break;
        }

        if( fds[1].revents & POLLRDHUP )
        {
            printf( "server close the connection\n" );
            break;
        }
        else if( fds[1].revents & POLLIN )
        {
            memset( read_buf, '\0', BUFFER_SIZE );
            recv( fds[1].fd, read_buf, BUFFER_SIZE-1, 0 );
            printf( "%s\n", read_buf );
        }

        if( fds[0].revents & POLLIN )
        {
            ret = splice( 0, NULL, pipefd[1], NULL, 32768, SPLICE_F_MORE | SPLICE_F_MOVE );
            ret = splice( pipefd[0], NULL, sockfd, NULL, 32768, SPLICE_F_MORE | SPLICE_F_MOVE );
        }
    }
    
    close( sockfd );
    return 0;
}

运行过程:

(1)运行服务器端代码

I/O多路复用:select、poll和epoll详解

(2)运行客户端代码并发送消息

I/O多路复用:select、poll和epoll详解

(3)服务器端转发消息的情况如下:

I/O多路复用:select、poll和epoll详解

 

参考:https://blog.csdn.net/davidsguo008/article/details/73556811