一文讲清I/O多路复用(select、poll和epoll)

发布时间 2024-01-02 01:07:09作者: HOracle

一、写在前面

本文尽可能使用最简单和最清楚的逻辑对比select、poll和epoll技术之间的区别,说明I/O多路复用从select发展到epoll到底优化了什么。

注意:1、本文不讲解select、poll和epoll三个系统调用的具体参数;2、未有特殊说明则本文以Linux平台作为基础;3、所有讲解仅代表作者本人理解,如有错误烦请指正。

二、概述

网络服务端处理并发连接的最简方案当属多线程设计,而多线程最大的问题就在于CPU上下文切换,且线程的资源占用较多,在高并发情况下多线程设计方案显得非常不合理,所以我们不如从单线程逻辑上进行I/O优化。

这里说明一下,我们刚才所谈及的线程是信号处理线程,这里读者可以简单理解成管理连接的线程(监听文件描述符是否有事件触发),而不是工作线程(及处理数据的线程)。下面是单线程设计伪码:

while(true)
{
    for(fd : fd[n])
    {
        if(fd有事件触发)
        {
            处理fd上的事件
        }
    }
}

单线程的问题在于if判断上,用户态程序为了判断fd是否有事件触发(例如有数据可读或可写)需要切换到内核态,这样以来,每次if判断都要进行用户态到内核态的切换,效率低下,而I/O多路复用技术主要就是对这个缺陷进行优化,它的整体思路便是:如果在用户态监听fd每次判断都要切换到内核态,那么就让内核直接帮我们监听fd,然后把触发的事件一次性返回给我们。

三、select技术

1、概述

在用户态初始化一个bitmap,将需要监听的文件描述符映射到这个bitmap中,把bitmap传入内核,内核监听对应的fd,如果fd有事件触发则对bitmap进行标记,标记完后整个返回到用户态,用户程序通过对bitmap的判断确定需要处理的fd进行处理。

2、流程

1)现有5个文件描述符fd(1~5)分别为:1、3、4、6、7;

2)用户程序创建一个位图rset,将第1位、第3位依次类推至第7位置为1;

3)将rset传入select(下面是select的处理):

① 将rset拷贝到内核,内核中的位图记为rset1(用户态到内核态,第一次切换)

② 内核阻塞(可以设置超时)监听fd(1~5),有事件发生时,内核通过修改rset1标记事件对应的fd

③ 将rset1拷贝回用户态的rset返回(内核态到用户态,第二次切换)

4)select退出,用户程序获得rset,并通过遍历rset确认需要处理的fd及对应的事件;

5)处理事件

3、编码

void main()
{
    // ...
    // 已创建服务器socket为sockfd
    listen(sockfd, 5);
    
    int fds[5] = {0};
    int max = 0;
    
    for(int i = 0; i < 5; ++i)
    {
        int fds[i] = accept(sockfd, NULL, NULL);
        if(fds[i] > max)
        {
        	max = fds[i];
        }
    }
    
    fd_set rset;
    char buf[MAXLEN] = {0};
    
    while(true)
    {
        FD_ZERO(&rset);
        for(int i = 0; i < 5; ++i)
        {
            FD_SET(fds[i], &rset);
        }

        select(max + 1, &rset, NULL, NULL, NULL);	// 阻塞

        for(int i = 0; i < 5; ++i)
        {
            if(FD_ISSET(fds[i], &rset))
            {
                memset(buf, 0, MAXLEN);
                read(fds[i], buf, MAXLEN);	// 事件处理,以读事件为例
                std::cout << buf << "\n";
            }
        }
    }
}

4、问题

1)bitmap大小受限(默认为1024)

2)rset不可重用,每次while开始需要循环初始化rset,时间复杂度O(n)

3)用户态和内核态之间切换

4)用户程序需要循环遍历rset,判断哪个fd上发生了事件,时间复杂度O(n)

四、poll技术

1、概述

struct pollfd
{
	int fd;
	short events;
	short revents;
};

在用户态初始化一个pollfd数组pfds,通过需要监听的fd和事件初始化pfds(修改pollfd结构中的fd和events),把pfds传入内核,内核监听对应的fd,如果有事件触发则修改对应的pollfd的revents,修改完成后返回pollfd数组到用户态,用户态遍历pollfd数组确认需要处理的fd进行处理。

2、流程

1)现有5个文件描述符fd(1~5);

2)用户程序创建一个长度为5的pollfd数组pfds,使用fd(1~5)和需要监听的事件初始化pfds的fd和events成员;

3)将pfds传入poll(下面是poll的处理):

① 将pfds拷贝到内核,内核中的pollfd数组记为pfds1(用户态到内核态,第一次切换)

② 内核阻塞监听pfds1中的fd,有事件发生时,内核修改pfds1的revents

③ 将pfds1拷贝回用户态的pfds返回(内核态到用户态,第二次切换)

4)poll退出,用户程序获得pfds,并通过遍历pfds确认哪些fd有事件发生;

5)处理事件

3、编码

void main()
{
    // ...
    struct pollfd pfds[5];

    for(int i = 0; i < 5; ++i)
    {
        pfds[i].fd = accept(sockfd, NULL, NULL);
        pfds[i].events = POLLIN;	// 读事件
    }

    char buf[MAXLEN] = {0};

    while(true)
    {
        poll(pfds, 5, 1000);	// 阻塞,超时时间1000ms

        for(int i = 0; i < 5; ++i)
        {
            if(pfds[i].revents & POLLIN)
            {
                pfds[i].revents = 0;
                memset(buf, 0, MAXLEN);
                read(pfds[i].fd, buf, MAXLEN);	// 事件处理,以读事件为例
                std::cout << buf << "\n";
            }
        }
    }
}

4、问题

1)bitmap大小受限(默认为1024)(使用了pollfd数组)

2)rset不可重用,每次while开始需要循环初始化rset,时间复杂度O(n)(pollfd结构因为有revents成员所以可重用)

3)用户态和内核态之间切换

4)用户程序需要循环遍历rset,判断哪个fd上发生了事件,时间复杂度O(n)

五、epoll技术

1、概述

typedef union epoll_data {
    void *ptr;
    int fd;
    __uint32_t u32;
    __uint64_t u64;
} epoll_data_t;

struct epoll_event
{
	__uint32_t events;
	epoll_data_t data;
};

用户态通过调用epoll_create函数创建一个epoll实例(底层实现为红黑树,通过文件描述符管理),将被监听fd和事件通过epoll_ctl和epoll_event注册到epoll实例,创建一个epoll_events数组用于接收epoll处理结果,用户态和内核态共享这个epoll_events数组,内核监听对应的fd,当有事件触发时内核对epoll_events数组进行重排,将已经就绪的fd放到靠前位置(水平触发),然后返回触发事件的fd总数,用户程序拿到返回后直接处理已经发生事件的fd,而不需要遍历所有fd。

2、流程

1)现有5个文件描述符fd(1~5);

2)用户程序通过epoll_create调用获取epfd;

3)循环调用epoll_ctl将fd及对应的待监听事件注册到epfd上(以epoll_event作为载体)

4)创建一个epoll_event数组events用于存储触发结果(events由用户态和内核态共享);

5)将epfd和events通过epoll_wait传入epoll(下面是epoll的处理):

① 内核阻塞监听注册到epfd上的fd,有事件发生时对events进行重排(分为水平触发和边缘触发)

② 内核返回已经触发事件的fd总数

6)epoll退出,用户程序拿到触发事件的fd总数,直接处理已经发生事件events

3、编码

void main()
{
    int epfd = epoll_create(1);

    for(int i = 0; i < 5; ++i)
    {
        static struct epoll_event ev;
        ev.data.fd = accept(sockfd, NULL, NULL);
        ev.events = EPOLLIN;	// 读事件
        epoll_ctl(epfd, EPOLL_CTL_ADD, ev.data.fd, &ev);	//注册事件
    }

    struct epoll_event events[5];
    char buf[MAXLEN] = {0};
    
    while(true)
    {
        int fds = epoll_wait(epfd, events, 5, 1000);	// 阻塞,超时时间1000ms

        for(int i = 0; i < fds; ++i)
        {
            memset(buf, 0, MAXLEN);
            read(events[i].data.fd, buf, MAXLEN);	// 事件处理,以读事件为例
            std::cout << buf << "\n";
        }
    }
}

4、问题

1)bitmap大小受限(默认为1024)(使用了pollfd数组)

2)rset不可重用,每次while开始需要循环初始化rset,时间复杂度O(n)(pollfd结构因为有revents成员所以可重用)

3)用户态和内核态之间切换(用户态和内核态共用epoll_events数组)

4)用户程序需要循环遍历rset,判断哪个fd上发生了事件,时间复杂度O(n)(直接返回已经发生事件的fd总数)