多路转接技术

在之前学习五种IO模型时，我们认识到了IO的本质是等+拷贝，而多路转接技术可以让等的过程重叠，即同时等待多个文件描述符的就绪状态，所以今天我们就来学习下如何等待多个文件描述就绪。

本篇文章会介绍三种实现多路转接的系统调用接口，实际最常用的是epoll，其实一些老的机器上面只兼容select，而poll并不常用。

1.select

select是系统提供的一个多路转接接口。

• select系统调用可以让我们的程序同时监视多个文件描述符的上的事件是否就绪。
• select的核心工作就是等，当监视的多个文件描述符中有一个或多个事件就绪时，select才会成功返回并将对应文件描述符的就绪事件告知调用者。

1.1select系统调用及参数介绍

int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

参数说明：

• nfds：需要监视的文件描述符中，最大的文件描述符值+1（select底层使用for循环遍历实现，该值是为了界定遍历范围）。
• readfds：输入输出型参数，调用时用户告知内核需要监视哪些文件描述符的读事件是否就绪，返回时内核告知用户哪些文件描述符的读事件已经就绪。
• writefds：输入输出型参数，调用时用户告知内核需要监视哪些文件描述符的写事件是否就绪，返回时内核告知用户哪些文件描述符的写事件已经就绪。
• exceptfds：输入输出型参数，调用时用户告知内核需要监视哪些文件描述符的异常事件是否就绪，返回时内核告知用户哪些文件描述符的异常事件已经就绪。
• timeout：输入输出型参数，调用时由用户设置select的等待时间，返回时表示timeout的剩余时间。

参数timeout的取值：

• NULL/nullptr：select调用后进行阻塞等待，直到被监视的某个文件描述符上的某个事件就绪。
• 0：selec调用后t进行非阻塞等待，无论被监视的文件描述符上的事件是否就绪，select检测后都会立即返回。
• 特定的时间值：select调用后在指定的时间内进行阻塞等待，如果被监视的文件描述符上一直没有事件就绪，则在该时间后select进行超时返回。

返回值说明：

• 如果函数调用成功，则返回有事件就绪的文件描述符个数。
• 如果timeout时间耗尽，则返回0。
• 如果函数调用失败，则返回-1，同时错误码会被设置。

select调用失败时，错误码可能被设置为：

• EBADF：文件描述符为无效的或该文件已关闭。
• EINTR：此调用被信号所中断。
• EINVAL：参数nfds为负值。
• ENOMEM：核心内存不足。

---

（1）fd_set类型

fd_set类型可以理解为一个位图，每个比特位位置代表是哪个文件描述符，值代表是否就绪。

调用select函数之前就需要用fd_set结构定义出对应的文件描述符集，然后将需要监视的文件描述符添加到文件描述符集当中，这个添加的过程本质就是在进行位操作，但是这个位操作不需要用户自己进行，系统提供了一组专门的接口，用于对fd_set类型的位图进行各种操作。

void FD_CLR(int fd, fd_set *set);      //用来清除描述词组set中相关fd的位
int  FD_ISSET(int fd, fd_set *set);    //用来测试描述词组set中相关fd的位是否为真
void FD_SET(int fd, fd_set *set);      //用来设置描述词组set中相关fd的位
void FD_ZERO(fd_set *set);             //用来清除描述词组set的全部位

（2）timeval结构

传入select函数的最后一个参数timeout，就是一个指向timeval结构的指针，timeval结构用于描述一段时间长度，该结构当中包含两个成员，其中tv_sec表示的是秒，tv_usec表示的是微秒。

调用时由用户设置select的等待时间，返回时表示timeout的剩余时间。

（3）socket就绪条件

读就绪

• socket内核中，接收缓冲区中的字节数，大于等于低水位标记SO_RCVLOWAT，此时可以无阻塞的读取该文件描述符，并且返回值大于0。
• socket TCP通信中，对端关闭连接，此时对该socket读，则返回0。
• 监听的socket上有新的连接请求。
• socket上有未处理的错误。

写就绪

• socket内核中，发送缓冲区中的可用字节数，大于等于低水位标记SO_SNDLOWAT，此时可以无阻塞的写，并且返回值大于0。
• socket的写操作被关闭（close或者shutdown），对一个写操作被关闭的socket进行写操作，会触发SIGPIPE信号。=
• socket使用非阻塞connect连接成功或失败之后。
• socket上有未读取的错误。

1.2select基本工作流程

利用select多路转接实现一个简单的Echo服务器，该服务器要做的就是读取客户端发来的数据并进行打印：

1. 先初始化服务器，完成套接字的创建、绑定和监听。
2. 定义一个fd_array辅助数组用于保存监听套接字和已经与客户端建立连接的套接字，刚开始时就将监听套接字添加到fd_array数组当中。
3. 然后服务器开始循环调用select函数，检测读事件是否就绪，如果就绪则执行对应的操作。
4. 每次调用select函数之前，都需要定义一个读文件描述符集readfds，并将fd_array辅助数组当中保存的文件描述符依次设置进readfds当中表示让select帮我们监视这些文件描述符的读事件是否就绪。
5. 当select检测到数据就绪时会将读事件就绪的文件描述符设置进readfds当中，此时我们就能通过提取readfds中的信息得知哪些文件描述符已经就绪，并对这些文件描述符进行对应的操作。
   • 如果读事件就绪的是监听套接字，则调用accept函数从底层全连接队列获取已经建立好的连接，并将该连接对应的套接字添加到fd_array数组当中。
   • 如果读事件就绪的是与客户端建立连接的套接字，则调用read函数读取客户端发来的数据并进行打印输出。
6. 当然，服务器与客户端建立连接的套接字读事件就绪，也可能是因为客户端将连接关闭了，此时服务器应该调用close关闭该套接字，并将该套接字从fd_array辅助数组当中清除，因为下一次不需要再监视该文件描述符的读事件了。

为什么要有辅助数组？

• 在服务器程序中，随着客户端连接的建立和断开，需要监听的文件描述符集合会动态变化。select调用后，只有发生事件的文件描述符会被保留在集合中，未发生事件的文件描述符会被清除。

• 因此，每次调用select之前，都需要重新构建文件描述符集合，确保所有需要监听的文件描述符都被包括在内。辅助数组可以用来存储当前所有需要监听的文件描述符，方便在每次调用select之前重新构建fd_set。

说明

• 服务器刚开始运行时，fd_array数组当中只有监听套接字，因此select第一次调用时只需要监视监听套接字的读事件是否就绪，但每次调用accept获取到新连接后，都会将新连接对应的套接字添加到fd_array当中，因此后续select调用时就需要监视监听套接字和若干连接套接字的读事件是否就绪。

• 由于调用select时还需要传入被监视的文件描述符中最大文件描述符值+1，因此每次在遍历fd_array对readfds进行重新设置时，还需要记录最大文件描述符值。

这其中还有很多细节，下面我们就来实现这样一个select服务器（这里我们只对读取实现多路转接）。

1.3select技术实现echo服务器

（1）构造服务器

对于一个Tcp服务器来说，我们首先需要创建一个listen套接字，然后在该listen套接字上等待获取新连接（普通套接字），而这个等待新连接到来的行为等价于对方给我发送数据，所以我们将获取新连接的行为看作读事件，所以在构造时，创建完listen套接字后，我们还需要将listen套接字仿佛辅助数组，未来由select统一进行等待。

SelectServer(uint16_t port)
    : _port(port), _listensock(std::make_unique<TcpSocket>())
    {
        InetAddr addr("0", _port);
        _listensock->BuildListenSocket(addr);
        for (int i = 0; i < N; i++)
        {
            _fd_array[i] = defaultfd;
        }
        // listensocket 等待新连接到来，等价于对方给我发送数据！我们作为读事件统一处理
        // 新连接到来 等价于 读事件就绪！
        // 首先要将listensock添加到select中！
        _fd_array[0] = _listensock->SockFd(); // 首先将listen套接字放入辅助数组
    }

（2）服务器核心逻辑Loop

服务器首先必须要有一个Loop()方法，是服务器执行的主逻辑，服务器就是一个死循环。

我们需要利用select监视读事件，所以每次循环都需要首先要初始化出来一个fd_set结构，并利用FD_ZERO将该文件描述符集内容清空，然后将辅助数组中保存的就绪的文件描述符通过FD_SET函数赋值给文件描述符集，注意更新最大文件描述符的值。（这里我们只考虑读事件，所以写事件和异常事件我们不考虑）。

之后我们就可以填充timeval结构：

• NULL/nullptr：select调用后进行阻塞等待，直到被监视的某个文件描述符上的某个事件就绪。
• 0：selec调用后t进行非阻塞等待，无论被监视的文件描述符上的事件是否就绪，select检测后都会立即返回。
• 特定的时间值：select调用后在指定的时间内进行阻塞等待，如果被监视的文件描述符上一直没有事件就绪，则在该时间后select进行超时返回。

然后根据select的返回值打印日志，当select返回值>0时（返回值是有事件就绪的文件描述符个数），证明监视的套接字中有读事件发生，此时对读事件进行处理，这里我们封装一个HandlerEvent函数，表示对读事件处理。

void Loop()
{
    while (true)
    {
        fd_set rfds;
        FD_ZERO(&rfds); // 将文件描述符集清空
        int max_fd = defaultfd;

        for (int i = 0; i < N; i++)
        {
            if (_fd_array[i] == defaultfd)
            {
                continue;
            }
            FD_SET(_fd_array[i], &rfds); // 将所有合法的fd添加到rfds中
            if (max_fd < _fd_array[i])
            {
                max_fd = _fd_array[i]; // 更新出最大的fd的值
            }
        }

        struct timeval timeout = {0, 0};

        // select 同时等待的fd，是有上限的。因为fd_set是具体的数据类型，有自己的大小！
        // rfds是一个输入输出型参数，每次调用，都要对rfds进行重新设定!
        int n = select(max_fd + 1, &rfds, nullptr, nullptr, /*&timeout*/ nullptr);
        switch (n)
        {
            case 0:
                LOG(INFO, "timeout, %d.%d\n", timeout.tv_sec, timeout.tv_usec);
                break;
            case -1:
                LOG(ERROR, "select error...\n");
                break;
            default:
                LOG(DEBUG, "Event Happen. n : %d\n", n); 
                HandlerEvent(rfds);
                break;
        }
    }
}

（3）对读事件进行处理HandlerEvent

调用该函数时，证明此时rfds文件描述符集已经被设定（输入输出型参数），文件描述符集中就是哪些文件描述符发生读事件，需要进行处理，所以我们只需要遍历辅助数组中保存的文件描述符，检测他们在rfds中是否被设定，如果被设定证明在该文件描述符上有事件发生需要进行处理，然后只需要分成两种情况，一种是listen套接字发生读事件，此时需要获取连接，另一种就是普通套接字，此时进行IO处理即可。

void HandlerEvent(fd_set &rfds)
{
    for (int i = 0; i < N; i++)
    {
        if (_fd_array[i] == defaultfd)
            continue;
        if (FD_ISSET(_fd_array[i], &rfds))
        {
            if (_fd_array[i] == _listensock->SockFd()) // 如果是listen套接字则获取连接
            {
                AcceptClient();
            }
            else
            {
                ServiceIO(i); // 如果是普通套接字提供IO服务
            }
        }
    }
}

（4）listen套接字获取连接AcceptClient

获取新连接，证明此时会有新的套接字（文件描述符）到来，所以需要将新的文件描述符保存到辅助数组中去，先获取新连接，然后遍历辅助数组找未被占用的位置，找到后，将该新连接（普通套接字）保存到辅助数组中。

void AcceptClient()
{
    InetAddr clientaddr;
    int sockfd = _listensock->Accepter(&clientaddr); // 这里调用accept会不会阻塞呢？？不会。因为事件已经就绪了
    if (sockfd < 0)
        return;

    LOG(DEBUG, "Get new Link, sockfd: %d, client info %s:%d\n", sockfd, clientaddr.Ip().c_str(), clientaddr.Port());
    int pos = 1; //从1开始，是因为0被listen套接字占了
    for (; pos < N; pos++)
    {
        if (_fd_array[pos] == defaultfd) //当检测到空位时跳出循环
            break;
    }
    if (pos == N)
    {
        ::close(sockfd); // sockfd->Close();
        LOG(WARNING, "server is full!\n");
        return;
    }
    else
    {
        _fd_array[pos] = sockfd;
        LOG(DEBUG, "%d add to select array!\n", sockfd);
    }
    LOG(DEBUG, "curr fd_array[] fd list : %s\n", RfdsToString().c_str());
}

（5）普通套接字提供IO服务

如果是普通套接字，则直接提供IO服务，我们实现的Echo服务器，所以仅需要将读取到的数据再发送出去即可。

需要注意的是：

如果recv返回值为0，表示连接已经被对端（即发送方）正常关闭。所以作为服务器也应该相应地关闭你的套接字资源，释放相关资源，这里不仅需要close，还需要将辅助数组中对应文件描述符还原为默认值，表示该文件描述符已经不再需要被等待了。

void ServiceIO(int pos)
{
    char buffer[1024];
    ssize_t n = ::recv(_fd_array[pos], buffer, sizeof(buffer) - 1, 0); // 这里读取会不会被阻塞？不会
    if (n > 0)
    {
        buffer[n] = 0;
        std::cout << "client say# " << buffer << std::endl;
        std::string echo_string = "[server echo]# ";
        echo_string += buffer;
        ::send(_fd_array[pos], echo_string.c_str(), echo_string.size(), 0);
    }
    else if (n == 0)
    {
        LOG(DEBUG, "%d is closed\n", _fd_array[pos]);
        ::close(_fd_array[pos]);
        _fd_array[pos] = defaultfd;
        LOG(DEBUG, "curr fd_array[] fd list : %s\n", RfdsToString().c_str());
    }
    else
    {
        LOG(DEBUG, "%d recv error\n", _fd_array[pos]);
        ::close(_fd_array[pos]);
        _fd_array[pos] = defaultfd;
        LOG(DEBUG, "curr fd_array[] fd list : %s\n", RfdsToString().c_str());
    }
}

1.4select优缺点

（1）select的优点

• 可以同时等待多个文件描述符，并且只负责等待，实际的IO操作由accept、read、write等接口来完成，这些接口在进行IO操作时不会被阻塞。
• select同时等待多个文件描述符，因此可以将“等”的时间重叠，提高了IO的效率。

当然，这也是所有多路转接接口的优点。

（2）select的缺点

• 每次调用select，都需要手动设置fd集合，从接口使用角度来说也非常不便。（输入参数和输出参数混合导致）
• 每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大。（输入参数和输出参数混合导致）
• 同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大。
• 由于采用固定大小的位图来表示文件描述符集，所以select可监控的文件描述符是有上限的。

1.5select的适用场景

多路转接接口select、poll和epoll，需要在一定的场景下使用，如果场景选择的不适宜，可能会适得其反。

• 多路转接接口一般适用于多连接，且多连接中只有少部分连接比较活跃。因为少量连接比较活跃，也就意味着几乎所有的连接在进行IO操作时，都需要花费大量时间来等待事件就绪，此时使用多路转接接口就可以将这些等的事件进行重叠，提高IO效率。
• 对于多连接中大部分连接都很活跃的场景，其实并不适合使用多路转接。因为每个连接都很活跃，也就意味着任何时刻每个连接上的事件基本都是就绪的，此时根本不需要动用多路转接接口来帮我们进行等待，毕竟使用多路转接接口也是需要花费系统的时间和空间资源的。

2.poll（了解）

poll也是系统提供的一个多路转接接口。

• poll系统调用也可以让我们的程序同时监视多个文件描述符上的事件是否就绪，和select的定位是一样的，适用场景也是一样的。

2.1poll系统调用及参数介绍

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

参数说明：

• fds：一个poll函数监视的结构列表，每一个元素包含三部分内容：文件描述符、监视的事件集合、就绪的事件集合。
• nfds：表示fds数组的长度。
• timeout：表示poll函数的超时时间，单位是毫秒（ms）。

参数timeout的取值：

• -1：poll调用后进行阻塞等待，直到被监视的某个文件描述符上的某个事件就绪。
• 0：poll调用后进行非阻塞等待，无论被监视的文件描述符上的事件是否就绪，poll检测后都会立即返回。
• 特定的时间值：poll调用后在指定的时间内进行阻塞等待，如果被监视的文件描述符上一直没有事件就绪，则在该时间后poll进行超时返回。

返回值说明：

• 如果函数调用成功，则返回有事件就绪的文件描述符个数。
• 如果timeout时间耗尽，则返回0。
• 如果函数调用失败，则返回-1，同时错误码会被设置。

poll调用失败时，错误码可能被设置为：

• EFAULT：fds数组不包含在调用程序的地址空间中。
• EINTR：此调用被信号所中断。
• EINVAL：nfds值超过RLIMIT_NOFILE值。
• ENOMEM：核心内存不足。

---

（1）pollfd类型

struct pollfd结构当中包含三个成员：

• fd：特定的文件描述符，若设置为负值则忽略events字段并且revents字段返回0。
• events：需要监视该文件描述符上的哪些事件。
• revents：poll函数返回时告知用户该文件描述符上的哪些事件已经就绪。

events和revents的取值：

事件	描述	是否可作为输入	是否可作为输出
POLLIN（常用）	数据（包括普通数据和优先数据）可读	是	是
POLLRDNORM	普通数据可读	是	是
POLLRDBAND	优先级带数据可读（Linux不支持）	是	是
POLLPRI	高优先级数据可读，比如TCP带外数据	是	是
POLLOUT（常用）	数据（包括普通数据和优先数据）可写	是	是
POLLWRNORM	普通数据可写	是	是
POLLWRBAND	优先级带数据可写	是	是
POLLRDHUP	TCP连接被对方关闭，或者对方关闭了写操作，它由GNU引入	是	是
POLLERR	错误	否	是
POLLHUP	挂起。比如管道的写端被关闭后，读端描述符上将收到POLLHUP事件	否	是
POLLNVAL	文件描述符没有打开	否	是

这些取值实际都是以宏的方式进行定义的，它们的二进制序列当中有且只有一个比特位是1，且为1的比特位是各不相同的。

• 因此在调用poll函数之前，可以通过或运算符将要监视的事件添加到events成员当中。
• 在poll函数返回后，可以通过与运算符检测revents成员中是否包含特定事件，以得知对应文件描述符的特定事件是否就绪。

2.2poll技术实现echo服务器

poll与select的区别就在于每次不需要重新将输入输出型参数重置，以及可控监控fd数量。

const static int gdefaultport = 8888;
const static int gbacklog = 8;
const static int gnum = 1024;
const static int gdefaultfd = -1;
const static int gdefultevent = 0;

class PollServer
{
    public:
    PollServer(int port = gdefaultport) : _port(port), _listensock(new TcpSocket()), _isrunning(false), _timeout(-1)
    {
    }
    void HandlerEvents()
    {
        for (int i = 0; i < gnum; i++)
        {
            if (_events[i].fd == gdefaultfd)
                continue;

            int fd = _events[i].fd;
            short revents = _events[i].revents;
            if (revents & POLLIN)
            {
                if (fd == _listensock->GetSockFd())
                {
                    // 新连接到来
                    std::string clientip;
                    uint16_t clientport;
                    int sockfd = _listensock->AcceptConnection(&clientip, &clientport);
                    if (sockfd < 0)
                        continue;
                    lg.LogMessage(Debug, "get a new client, %s:%d\n", clientip.c_str(), clientport);
                    int j = 0;
                    for (; j < gnum; j++)
                    {
                        if (_events[j].fd == gdefaultfd)
                            break;
                    }
                    if (j < gnum)
                    {
                        // 给poll添加新的sockfd
                        _events[j].fd = sockfd;
                        _events[j].events = POLLIN;
                        _events[j].revents = gdefultevent;
                        lg.LogMessage(Debug, "add a new client, fd is : %d\n", sockfd);
                        PrintDebug();
                    }
                    else
                    {
                        // 扩容或者close
                    }
                }
                else
                {
                    char buffer[1024];
                    // fd上面的读事件就绪了
                    ssize_t n = recv(fd, buffer, sizeof(buffer) - 1, 0);
                    if (n > 0)
                    {
                        buffer[n] = 0;
                        lg.LogMessage(Debug, "client info# %s\n", buffer);

                        std::string echo_message = "[server echo]# ";
                        echo_message += buffer;
                        // 写回去 --- epoll会说
                        send(fd, echo_message.c_str(), echo_message.size(), 0);
                    }
                    else if (n == 0)
                    {
                        ::close(fd);
                        _events[i].fd = gdefaultfd;
                        _events[i].events = _events[i].revents = gdefultevent;
                        lg.LogMessage(Info, "client quit..., remove fd: %d\n", fd);
                        PrintDebug();
                    }
                    else
                    {
                        ::close(fd);
                        _events[i].fd = gdefaultfd;
                        _events[i].events = _events[i].revents = gdefultevent;
                        lg.LogMessage(Info, "recv error..., remove fd: %d\n", fd);
                        PrintDebug();
                    }
                }
            }
            if (revents & POLLOUT)
            {
                // TODO
            }
        }
    }
    void InitServer()
    {
        _listensock->BuildListenSocketMethod(_port, gbacklog);
        for (int i = 0; i < gnum; i++)
        {
            _events[i].fd = gdefaultfd;
            _events[i].events = gdefultevent;
            _events[i].revents = gdefultevent;
        }
        _events[0].fd = _listensock->GetSockFd();
        _events[0].events = POLLIN; // 表示对读事件关心(listensock上有新连接，就是读事件就绪)
        lg.LogMessage(Debug, "add listen socket, fd is : %d\n", _listensock->GetSockFd());
    }
    void Loop()
    {
        _isrunning = true;
        while (_isrunning)
        {
            int n = poll(_events, gnum, _timeout);
            switch (n)
            {
                case 0:
                    lg.LogMessage(Info, "poll timeout...\n");
                    break;
                case -1:
                    lg.LogMessage(Error, "poll error!!!\n");
                    break;
                default:
                    // 正常的就绪的fd
                    lg.LogMessage(Info, "get a read event!\n");
                    HandlerEvents();
                    break;
            }
        }
        _isrunning = false;
    }
    void Stop()
    {
        _isrunning = false;
    }

    void PrintDebug()
    {
        std::cout << "current poll rfds list is : ";
        for (int i = 0; i < gnum; i++)
        {
            if (_events[i].fd == gdefaultfd)
                continue;
            else
                std::cout << _events[i].fd << " ";
        }
        std::cout << std::endl;
    }
    ~PollServer()
    {
    }

    private:
    std::unique_ptr<Socket> _listensock;
    int _port;
    int _isrunning;
    struct pollfd _events[gnum];
    int _timeout;
};

2.3poll优缺点

（1）poll的优点

• struct pollfd结构当中包含了events和revents，相当于将select的输入输出型参数进行分离，因此在每次调用poll之前，不需要像select一样重新对参数进行设置。
• poll可监控的文件描述符数量没有限制，由传入poll函数的第二个参数决定。

（2）poll的缺点

• 和select函数一样，当poll返回后，需要遍历fds数组来获取就绪的文件描述符。
• 每次调用poll，都需要把大量的struct pollfd结构从用户态拷贝到内核态，这个开销也会随着poll监视的文件描述符数目的增多而增大。
• 同时每次调用poll都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大。

3.epoll

按照 man 手册的说法：epoll是为处理大批量句柄而作了改进的 poll。

句柄是指能够标定资源的结构，比如文件描述符、文件结构体、套接字等。

epoll在2.5.44内核中被引进，它几乎具备了select和poll的所有优点，被公认为Linux2.6下性能最好的多路I/O就绪通知方法。

3.1epoll系统调用及参数介绍

（1）epoll_create

epoll_create函数用于创建一个epoll模型。

int epoll_create(int size);

参数说明：

• size：自从Linux2.6.8之后，size参数是被忽略的，但size的值必须设置为大于0的值。

返回值说明：

• epoll模型创建成功返回其对应的文件描述符，否则返回-1，同时错误码会被设置。

注意： 不再使用后，必须调用close函数关闭epoll模型对应的文件描述符，当所有引用epoll实例的文件描述符都已关闭时，内核将销毁该实例并释放相关资源。

（2）epoll_ctl

epoll_ctl函数用于向指定的epoll模型（epfd）中对于指定的文件描述符（fd）上指定的监听事件（event）进行指定的操作（op）。

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

参数说明：

• epfd：指定的epoll模型。
• op：表示具体的动作，用三个宏来表示。
• fd：需要监视的文件描述符。
• event：需要监视该文件描述符上的哪些事件。（输入型参数）

第二个参数op的取值有以下三种：

• EPOLL_CTL_ADD：注册新的文件描述符到指定的epoll模型中。
• EPOLL_CTL_MOD：修改已经注册的文件描述符的监听事件。
• EPOLL_CTL_DEL：从epoll模型中删除指定的文件描述符。

返回值说明：

• 函数调用成功返回0，调用失败返回-1，同时错误码会被设置。

第四个参数对应的struct epoll_event结构如下：

struct epoll_event结构中有两个成员：

• events可以理解为一个位图结构，表示的是需要监视的事件
• data是一个联合体（union），允许用户存储与事件相关联的额外数据。它可以是以下几种类型之一：
  • void *ptr：一个指向任意数据的指针，用户可以根据需要存储任何类型的数据。
  • int fd：直接存储文件描述符本身，这在某些情况下非常有用，比如需要知道是哪个文件描述符触发了事件。
  • __uint32_t u32 和 __uint64_t u64：分别提供32位和64位的无符号整数空间，用于存储用户定义的数值数据。

events的常用取值如下：

• EPOLLIN：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）。
• EPOLLOUT：表示对应的文件描述符可以写。
• EPOLLPRI：表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据到来）。
• EPOLLERR：表示对应的文件描述符发送错误。
• EPOLLHUP：表示对应的文件描述符被挂断，即对端将文件描述符关闭了。
• EPOLLET：将epoll的工作方式设置为边缘触发（Edge Triggered）模式。
• EPOLLONESHOT：只监听一次事件，当监听完这次事件之后，如果还需要继续监听该文件描述符的话，需要重新将该文件描述符添加到epoll模型中。

（3）epoll_wait

epoll_ctl函数用于获取指定epoll模型（epfd）上已经就绪的文件描述符上的事件（events）。

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

参数说明：

• epfd：指定的epoll模型。
• events：内核会将已经就绪的事件拷贝到events数组当中（events不能是空指针，内核只负责将就绪事件拷贝到该数组中，不会帮我们在用户态中分配内存）。（输出型参数）
• maxevents：events数组中的元素个数，该值不能大于创建epoll模型时传入的size值。
• timeout：表示epoll_wait函数的超时时间，单位是毫秒（ms）。

参数timeout的取值：

• -1：epoll_wait调用后进行阻塞等待，直到被监视的某个文件描述符上的某个事件就绪。
• 0：epoll_wait调用后进行非阻塞等待，无论被监视的文件描述符上的事件是否就绪，epoll_wait检测后都会立即返回。
• 特定的时间值：epoll_wait调用后在直到的时间内进行阻塞等待，如果被监视的文件描述符上一直没有事件就绪，则在该时间后epoll_wait进行超时返回。

返回值说明：

• 如果函数调用成功，则返回有事件就绪的文件描述符个数。
• 如果timeout时间耗尽，则返回0。
• 如果函数调用失败，则返回-1，同时错误码会被设置。

epoll_wait调用失败时，错误码可能被设置为：

• EBADF：传入的epoll模型对应的文件描述符无效。
• EFAULT：events指向的数组空间无法通过写入权限访问。
• EINTR：此调用被信号所中断。
• EINVAL：epfd不是一个epoll模型对应的文件描述符，或传入的maxevents值小于等于0。

3.2epoll工作原理

当某一进程调用epoll_create函数时，Linux内核会创建一个eventpoll结构体，也就是我们所说的epoll模型，eventpoll结构体当中的成员rbr（红黑树）和rdlist（就绪队列）与epoll的使用方式密切相关。

struct eventpoll{
	...
	//红黑树的根节点，这棵树中存储着所有添加到epoll中的需要监视的事件
	struct rb_root rbr;
	//就绪队列中则存放着将要通过epoll_wait返回给用户的满足条件的事件
	struct list_head rdlist;
	...
}

所以epoll_ctl函数就是对指定的epoll模型（struct eventpoll）中的红黑树（fd为键值）进行EPOLL_CTL_ADD（添加节点）、EPOLL_CTL_MOD（修改节点）、EPOLL_CTL_DEL（删除节点）操作，红黑树中的节点代表监视哪些文件描述符上的哪些事件。

• 调用epoll_ctl向红黑树当中新增节点时，如果设置了EPOLLONESHOT选项，当监听完这次事件后，如果还需要继续监听该文件描述符则需要重新将其添加到epoll模型中，本质就是当设置了EPOLLONESHOT选项的事件就绪时，操作系统会自动将其从红黑树当中删除。
• 而如果调用epoll_ctl向红黑树当中新增节点时没有设置EPOLLONESHOT，那么该节点插入红黑树后就会一直存在，除非用户调用epoll_ctl将该节点从红黑树当中删除。

当红黑树中的某节点准备就绪时（该节点对应的文件描述符上的事件就绪），操作系统会自动将该文件描述符和事件形成一个新节点，并将该节点插入到就绪队列中。

调用epoll_wait函数定期（timeout）检查就绪队列中是否有节点，若检查时无节点就会超时或阻塞，若有就会将已经就绪节点中的数据拷贝给epoll_wait函数中的输出型参数events中，用户仅需定义一个epoll_event类型的结构体数组接收即可，此时就绪事件会严格的按照下标顺序依次放入到数组中，未来对这些已经就绪的事件进行处理时，仅需对该数组进行遍历，遍历到epoll_wait函数的返回值处即可获取所有就绪事件。

借用就绪队列这种方案，检测有没有事件就绪时间复杂度为O(1)，不需要类似select一样遍历了。

但我们发现红黑树中的节点和就绪队列中的节点大致相似，因为他们都需要文件描述符以及文件描述符上的事件，所以在操作系统中，一个节点的描述如下（句柄）：

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

即一个句柄中不仅有文件描述符以及文件描述符上的事件，还有该句柄是否连接到红黑树、是否连接到就绪队列中等信息。

• 红黑树节点（rbn）：用于将epitem插入到epoll对象所维护的红黑树中。红黑树是一种自平衡二叉查找树，它允许在O(log n)的时间复杂度内完成查找、插入和删除操作，这对于管理大量文件描述符非常有效。
• 双向链表节点（rdllink）：当文件描述符上的事件就绪时，epitem会被添加到epoll对象所维护的就绪链表中。这个链表用于快速遍历所有就绪的事件，以便epoll_wait()函数能够高效地返回给用户空间。
• 文件描述符信息（ffd）：包含被监听的文件描述符的详细信息，如文件描述符本身、所属的epoll对象等。

回调机制

这里有一个问题：当操作系统检测到文件描述符对应事件已就绪，如何将红黑树中的节点自动插入到就绪队列中呢？

• 回调。

所有添加到红黑树当中的事件，都会与设备（网卡）驱动程序建立回调方法，这个回调方法在内核中叫ep_poll_callback。

当被监视的文件描述符（如 socket）上的事件发生时（如数据可读、可写或发生错误等），内核会调用这个回调函数来处理相应的事件。而该回调函数完成的就是将节点插入到就绪队列中。

• 对于select和poll来说，操作系统在监视多个文件描述符上的事件是否就绪时，需要让操作系统主动对这多个文件描述符进行轮询检测，这一定会增加操作系统的负担。
• 而对于epoll来说，操作系统不需要主动进行事件的检测，当红黑树中监视的事件就绪时，会自动调用对应的回调方法，将就绪的事件添加到就绪队列当中。
• 当用户调用epoll_wait函数获取就绪事件时，不需要遍历，只需要关注底层就绪队列是否为空，如果不为空则将就绪队列当中的就绪事件拷贝给用户即可。

所以epoll这样的方案相较于select和poll有了很大提升。

补充：

• 当不断有监视的事件就绪时，会不断调用回调方法向就绪队列当中插入节点，而上层也会不断调用epoll_wait函数从就绪队列当中获取节点，这是典型的生产者消费者模型。
• 由于就绪队列可能会被多个执行流同时访问，因此必须要使用互斥锁对其进行保护，eventpoll结构当中的lock和mtx就是用于保护临界资源的，因此epoll本身是线程安全的。
• eventpoll结构当中的wq（wait queue）就是等待队列，当多个执行流想要同时访问同一个epoll模型时，就需要在该等待队列下进行等待。

3.3eventpoll与文件

一个进程可以调用epoll_create可以为我们创建一个epoll模型，而epoll模型其实就是红黑树、就绪队列和回调函数组成的一个模型，回调函数这里是嵌入在系统底层的，而红黑树和就绪队列是真实的维护在struct eventpoll结构体中的，这都是上面我们学习到的。

而现在我们来关心以下epfd。

epoll_create为我们创建一个epoll模型，可为什么返回了一个文件描述符呢？

也就是说epoll模型也是一个文件。对于Linux，一切皆文件。

struct file中维护了一个ptr指针，该指针指向了eventpoll结构，所以通过epfd我们就可以找到这个epoll模型中维护的所有内容，包括红黑树、就绪队列、文件描述符、文件描述符上的事件等等。

所以epoll_ctl和epoll_wait都需要传入epfd，因为他们都需要通过epfd找到epoll模型。

3.4epoll技术实现echo服务器

总结一下，epoll的使用过程就是三部曲：

• 调用epoll_create创建一个epoll模型。
• 调用epoll_ctl，将要监控的文件描述符进行注册。
• 调用epoll_wait，等待文件描述符就绪。

（1）构造服务器

对于一个Tcp服务器来说，我们首先需要创建一个listen套接字，然后在该listen套接字上等待获取新连接（普通套接字），而这个等待新连接到来的行为等价于对方给我发送数据，所以我们将获取新连接的行为看作读事件，所以在构造时，创建完listen套接字后，我们还需要调用epoll_create函数创建epoll模型，然后将listen套接字对应的文件描述符和需要监听的事件传递给epoll_ctl函数，让epoll_ctl函数进行添加节点的操作。

class EpollServer
{
    const static int gnum = 64;

    EpollServer(uint16_t port = 8888)
        : _port(port)
            ,_listensock(std::make_unique<TcpSocket>())
            ,_epfd(-1)
        {
            // 1. 创建listensock
            InetAddr addr("0", _port);
            _listensock->BuildListenSocket(addr);

            // 2. 创建epoll模型
            _epfd = ::epoll_create(128);
            if (_epfd < 0)
            {
                LOG(FATAL, "epoll_create error\n");
                exit(5);
            }
            LOG(DEBUG, "epoll_create success, epfd: %d\n", _epfd);
            // 3. 只有一个listensock, listen sock 关心的事件：读事件
            struct epoll_event ev;
            ev.events = EPOLLIN;
            ev.data.fd = _listensock->SockFd(); //epoll_event结构体中的data字段允许用户保存一部分数据（比如fd），以便最后从epoll_wait中的输出型参数*epoll_event中提取出来，这样才能在处理的时候保证知道是哪个文件描述符触发了事件。
            epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock->SockFd(), &ev);
        }
    
    ~EpollServer()
    {
        _listensock->Close();
        if (_epfd >= 0)
            ::close(_epfd);
    }

    private:
    uint16_t _port;
    std::unique_ptr<Socket> _listensock;
    int _epfd;

    struct epoll_event _revs[gnum]; //就绪事件的数组
};

（2）服务器核心逻辑Loop

服务器首先必须要有一个Loop()方法，是服务器执行的主逻辑，服务器就是一个死循环。

我们可以设置timeout以达到不同的等待策略：

• -1：epoll_wait调用后进行阻塞等待，直到被监视的某个文件描述符上的某个事件就绪。
• 0：epoll_wait调用后进行非阻塞等待，无论被监视的文件描述符上的事件是否就绪，epoll_wait检测后都会立即返回。
• 特定的时间值：epoll_wait调用后在直到的时间内进行阻塞等待，如果被监视的文件描述符上一直没有事件就绪，则在该时间后epoll_wait进行超时返回。

根据返回值决定执行执行的逻辑，当epoll_wait返回值>0时（返回值是有事件就绪的文件描述符个数），证明监视的套接字中有读事件发生，此时对读事件进行处理，这里我们封装一个HandlerEvent函数，表示对读事件处理。

调用epoll_wait函数定期（timeout）检查就绪队列中是否有节点，若检查时无节点就会超时或阻塞，若有就会将已经就绪节点中的数据拷贝给epoll_wait函数中的输出型参数events中，用户仅需定义一个epoll_event类型的结构体数组接收即可，此时就绪事件会严格的按照下标顺序依次放入到数组中，未来对这些已经就绪的事件进行处理时，仅需对该数组进行遍历，遍历到epoll_wait函数的返回值处即可获取所有就绪事件。

void Loop()
{
    int timeout = 1000; //设置不同的值有不同的策略
    while (true)
    {
        int n = ::epoll_wait(_epfd, _revs, gnum, timeout);
        switch (n)
        {
            case 0:
                LOG(DEBUG, "epoll_wait timeout...\n");
                break;
            case -1:
                LOG(DEBUG, "epoll_wait failed...\n");
                break;
            default:
                LOG(DEBUG, "epoll_wait haved event ready..., n : %d\n", n);
                HandlerEvent(n);
                break;
        }
    }
}
private:
uint16_t _port;
std::unique_ptr<Socket> _listensock;
int _epfd;

struct epoll_event _revs[gnum]; //就绪事件的数组

（3）对读事件进行处理HandlerEvent

调用该函数时，由于epoll_wait的返回值是有事件就绪的文件描述符个数，所以我们仅需遍历就绪事件数组到这之前的事件即可。

而后我们取出就绪事件数组的内容，即文件描述符和文件描述符上发生的事件，然后通过与操作符判断发生了什么事件（代码中以读事件为例），然后只需要分成两种情况，一种是listen套接字发生读事件，此时需要获取连接，另一种就是普通套接字，此时进行IO处理即可。

• 如果是listen套接字，则获取新连接，然后将新的文件描述符和事件传递给epoll_ctl函数，让epoll模型对他们进行管理即可。

• 如果是普通套接字，则直接提供IO服务，我们实现的Echo服务器，所以仅需要将读取到的数据再发送出去即可。

需要注意的是：

如果recv返回值为0，表示连接已经被对端（即发送方）正常关闭。所以作为服务器也应该相应地关闭你的套接字资源，释放相关资源，这里不仅需要close，还需要使用epoll_ctl函数执行EPOLL_CTL_DEL操作将节点移除出epoll模型，表示该文件描述符已经不再需要被等待了，更重要的是一定要先移除再调用close关闭文件描述符，因为要删除的fd，必须是合法的！

void HandlerEvent(int num)
{
    for (int i = 0; i < num; i++) 
    {
        uint32_t revents = _revs[i].events;
        int sockfd = _revs[i].data.fd; //获取事件是在哪个文件描述符上发生了

        // 读事件就绪
        if (revents & EPOLLIN)
        {
            if (sockfd == _listensock->SockFd())
            {
                InetAddr clientaddr;
                int newfd = _listensock->Accepter(&clientaddr); // 会不会被阻塞？不会!
                if (newfd < 0)
                    continue;

                // 获取新链接成功
                // 读取或者写入？不能！
                // 将新获取的fd添加到红黑树节点中去，让epoll等待
                struct epoll_event ev;
                ev.events = EPOLLIN;
                ev.data.fd = newfd;
                epoll_ctl(_epfd, EPOLL_CTL_ADD, newfd, &ev);
                LOG(DEBUG, "_listensock ready, accept done, epoll_ctl done, newfd is: %d\n", newfd);
            }
            else
            {
                char buffer[1024];
                ssize_t n = ::recv(sockfd, buffer, sizeof(buffer), 0); //如何保证读取到完整报文？？
                if (n > 0)
                {
                    LOG(DEBUG, "normal fd %d ready, recv begin...\n", sockfd);
                    buffer[n] = 0;
                    std::cout << "client say# " << buffer << std::endl;

                    std::string echo_string = "server echo# ";
                    echo_string += buffer;
                    ::send(sockfd, echo_string.c_str(), echo_string.size(), 0);
                }
                else if (n == 0)
                {
                    LOG(DEBUG, "normal fd %d close, me too!\n", sockfd);
                    // 对端连接关闭了
                    ::epoll_ctl(_epfd, EPOLL_CTL_DEL, sockfd, nullptr); // 要删除的fd，必须是合法的！
                    ::close(sockfd);
                }
                else
                {
                    ::epoll_ctl(_epfd, EPOLL_CTL_DEL, sockfd, nullptr); // 要删除的fd，必须是合法的！
                    ::close(sockfd);
                }
            }
        }
    }
}

epoll相较于select和poll的优势：

• 在使用select和poll时，都需要借助第三方数组来维护历史上的文件描述符以及需要监视的事件，这个第三方数组是由用户自己维护的，对该数组的增删改操作都需要用户自己来进行。
• 而使用epoll时，不需要用户自己维护所谓的第三方数组，epoll底层的红黑树就充当了这个第三方数组的功能，并且该红黑树的增删改操作都是由内核维护的，用户只需要调用epoll_ctl让内核对该红黑树进行对应的操作即可。
• 在使用多路转接接口时，数据流都有两个方向，一个是用户告知内核，一个是内核告知用户。select和poll将这两件事情都交给了同一个函数来完成，而epoll在接口层面上就将这两件事进行了分离，epoll通过调用epoll_ctl完成用户告知内核，通过调用epoll_wait完成内核告知用户。

3.5epoll优点

• 接口使用方便：虽然拆分成了三个函数，但是反而使用起来更方便高效。
• 数据拷贝轻量：只在新增监视事件的时候调用epoll_ctl将数据从用户拷贝到内核，而select和poll每次都需要重新将需要监视的事件从用户拷贝到内核。此外，调用epoll_wait获取就绪事件时，只会拷贝就绪的事件(struct epoll_event)，不会进行不必要的拷贝操作。
• 事件回调机制：避免操作系统主动轮询检测事件就绪，而是采用回调函数的方式，将就绪的文件描述符结构加入到就绪队列中。调用epoll_wait时直接访问就绪队列就知道哪些文件描述符已经就绪，检测是否有文件描述符就绪的时间复杂度是O(1) ，因为本质只需要判断就绪队列是否为空即可。
• 没有数量限制：监视的文件描述符数目无上限，只要内存允许，就可以一直向红黑树当中新增节点。

3.6epoll工作方式

在上述代码中，调用recv时有一个明显问题：recv能保证读取到的报文是完整的么？

不能，想要确保报文是完整的，就必须需要应用层协议来实现，那假设我们一次只读取到了完整报文的一半，下次再调用recv读取。但是此时buffer中的内容已经重置了（生命周期），第一次读取的数据已经被覆盖了。

那怎么保证数据读全呢？也就是怎么保证在多路转接技术下实现正确的读写操作呢？

（1）LT水平触发工作模式（Level Triggered）

• 只要底层有事件就绪，epoll就会一直通知用户。
• 就像数字电路当中的电平触发一样，只要一直处于高电平，则会一直触发。

epoll默认状态下就是LT工作模式。

• 由于在LT工作模式下，只要底层有事件就绪就会一直通知用户，因此当epoll检测到底层读事件就绪时，可以不立即进行处理，或者只处理一部分，因为只要底层数据没有处理完，下一次epoll还会通知用户事件就绪。
• select和poll其实就是工作是LT模式下的。
• 支持阻塞读写和非阻塞读写。

（2）ET边缘触发工作模式（Edge Triggered）

• 只有底层就绪事件数量由无到有或由有到多发生变化的时候，epoll才会通知用户。
• 就像数字电路当中的上升沿触发一样，只有当电平由低变高的那一瞬间才会触发。

如果要将epoll改为ET工作模式，则需要在添加事件时设置EPOLLET选项。

• 由于在ET工作模式下，只有底层就绪事件无到有或由有到多发生变化的时候才会通知用户，因此当epoll检测到底层读事件就绪时，必须立即进行处理，而且必须全部处理完毕，因为有可能此后底层再也没有事件就绪，那么epoll就再也不会通知用户进行事件处理，此时没有处理完的数据就相当于丢失了。
• ET工作模式下用户一定会以最快的速度把内核TCP中的数据读走（倒逼程序员），因此ET的性能一般比LT性能更高，Nginx就是默认采用ET模式使用epoll的。
• 只支持非阻塞的读写。

（3）回答几个问题

1）那么在ET工作模式下，如何保证数据读取完了呢？—循环读取。

当底层读事件就绪时，循环调用recv函数进行读取，直到某次调用recv读取时，实际读取到的字节数小于用户缓冲区字节数了，则说明本次底层数据已经读取完毕了。

但是循环读取又面临一个问题，可能会引发阻塞，因为我们的epoll只能保证一次就绪，下一次就不能保证了（比如一共有500字节数据，每次读100字节，那么第6次缓冲区就没有数据了），就会引发阻塞，那如何解决呢？—设置非阻塞。

所以，ET工作模式下，recv和send操作的文件描述符必须设置为非阻塞状态。

2）但是LT模式如果也能每次都把数据全部处理，那么也就不会出现反复通知了，此时LT模式和ET模式效率不就没差了么？

是的，但是LT模式下不强制你尽快读取（不会出错），ET模式下你不读取走下次很可能读取不到了，所以ET模式下可以倒逼着程序员必须尽快处理（程序员水平参差不齐，如果是ET证明一定效率高，可如果是LT可能效率高可能不高）。

3）TCP协议中的PSH标记位

当PSH被置1时，该报文会告知对方尽快将你的接收缓冲区当中的数据交付给上层。

操作系统的策略是当缓冲区当中的数据量达到一定标准时才能进行读取，当PSH被置1后，此时，就不会根据这个标准，而是催促操作系统尽快将缓冲区中的数据向上交付。

实际上当PSH被置1后，就是告诉上层：事件已经就绪。哪怕再小的数据，也要放到就绪队列里。

---

那些过去，从未过去。 —毕业了，我们一无所有