LinuxI/O多路转接(select、poll、epoll)

---

I/O 多路转接：select、poll、epoll 全面解析

在 Linux 网络编程中，I/O 多路转接技术是实现高并发服务器的核心手段。它允许程序同时监视多个文件描述符的事件状态（如读、写、异常），从而提高 I/O 效率，避免为每个连接创建独立线程的开销。本文将深入探讨三种常见的 I/O 多路转接机制：select、poll 和 epoll，从原理到实现，结合代码示例，全面剖析其功能、优缺点及适用场景。

---

一、I/O 多路转接之 select

select 初识

select 是 Linux 系统提供的经典 I/O 多路转接接口。它允许程序同时监视多个文件描述符的事件是否就绪，其核心功能是“等待”——当至少一个文件描述符的事件（如读、写或异常）就绪时，select 返回并告知调用者具体就绪的事件。

select 的设计初衷是解决单线程阻塞 I/O 的问题。例如，在一个简单的服务器中，如果使用 accept 或 read 直接等待事件，可能因某个描述符未就绪而阻塞整个程序。而 select 提供了一种统一等待多个描述符的方式，从而实现非阻塞的多客户端服务。

select 函数

select 函数的原型如下：

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

参数说明

• nfds：需要监视的文件描述符中最大值加 1，用于限定内核检测范围。
• readfds：输入输出型参数，用户传入时指定需要监视读事件的文件描述符集合，返回时内核标记哪些描述符的读事件已就绪。
• writefds：输入输出型参数，类似 readfds，用于监视写事件。
• exceptfds：输入输出型参数，用于监视异常事件。
• timeout：输入输出型参数，设置等待时间，返回时表示剩余时间。
  • NULL：阻塞等待，直到有事件就绪。
  • 0：非阻塞检测，立即返回。
  • 特定时间值（如 {5, 0} 表示 5 秒）：超时返回。

返回值说明

• 成功：返回就绪的文件描述符个数。
• 超时：返回 0。
• 失败：返回 -1，并设置错误码：
  • EBADF：无效或已关闭的文件描述符。
  • EINTR：被信号中断。
  • EINVAL：nfds 为负值。
  • ENOMEM：内存不足。

fd_set 结构

fd_set 是一个位图结构，每个位表示一个文件描述符。系统提供以下宏操作：

• FD_ZERO(fd_set *set)：清空集合。
• FD_SET(int fd, fd_set *set)：设置某位。
• FD_CLR(int fd, fd_set *set)：清除某位。
• FD_ISSET(int fd, fd_set *set)：测试某位是否置位。

timeval 结构

timeout 参数指向 timeval 结构：

struct timeval {
    time_t tv_sec;       // 秒
    suseconds_t tv_usec; // 微秒
};

socket 就绪条件

在网络编程中，socket 的就绪条件决定了 select 的触发时机：

• 读就绪：
  • 接收缓冲区字节数 ≥ 低水位标记 SO_RCVLOWAT，可无阻塞读取。
  • 对端关闭连接（读返回 0）。
  • 监听 socket 上有新连接请求。
  • 有未处理的错误。
• 写就绪：
  • 发送缓冲区可用字节数 ≥ 低水位标记 SO_SNDLOWAT，可无阻塞写入。
  • 写操作被关闭（触发 SIGPIPE）。
  • 非阻塞 connect 成功或失败。
  • 有未读取的错误。
• 异常就绪：
  • 收到带外数据（与 TCP 的紧急模式相关，通过 URG 标志和紧急指针实现）。

select 基本工作流程

以一个简单的服务器为例（功能：读取并打印客户端数据），select 的工作流程如下：

1. 初始化服务器：创建监听 socket，完成绑定和监听。
2. 定义 fd_array：用数组保存监听 socket 和客户端连接 socket，初始只加入监听 socket。
3. 循环调用 select：
   • 每次调用前，重置 readfds，遍历 fd_array 设置需要监视的描述符，并记录最大值 maxfd。
   • 调用 select，等待事件就绪。
4. 处理就绪事件：
   • 若监听 socket 就绪，调用 accept 获取新连接，将新 socket 加入 fd_array。
   • 若客户端 socket 就绪，调用 read 读取数据并打印，若连接关闭则关闭 socket 并从 fd_array 中移除。

由于 readfds、writefds 和 exceptfds 是输入输出型参数，select 返回后会被修改，因此每次调用前需重新设置。timeout 也是类似逻辑。

select 服务器实现

以下是一个完整的 select 服务器实现：

Socket 类

#pragma once
#include <iostream>
#include <unistd.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <cstring>

class Socket {
public:
    static int SocketCreate() {
        int sock = socket(AF_INET, SOCK_STREAM, 0);
        if (sock < 0) { std::cerr << "socket error" << std::endl; exit(2); }
        int opt = 1;
        setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
        return sock;
    }
    static void SocketBind(int sock, int port) {
        struct sockaddr_in local;
        memset(&local, 0, sizeof(local));
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = INADDR_ANY;
        if (bind(sock, (struct sockaddr*)&local, sizeof(local)) < 0) {
            std::cerr << "bind error" << std::endl; exit(3);
        }
    }
    static void SocketListen(int sock, int backlog) {
        if (listen(sock, backlog) < 0) {
            std::cerr << "listen error" << std::endl; exit(4); }
    }
};

SelectServer 类

#pragma once
#include "socket.hpp"
#include <sys/select.h>

#define BACK_LOG 5
#define NUM (sizeof(fd_set)*8)
#define DFL_FD -1

class SelectServer {
private:
    int _listen_sock;
    int _port;
public:
    SelectServer(int port) : _port(port), _listen_sock(-1) {}
    void InitSelectServer() {
        _listen_sock = Socket::SocketCreate();
        Socket::SocketBind(_listen_sock, _port);
        Socket::SocketListen(_listen_sock, BACK_LOG);
    }
    void Run() {
        fd_set readfds;
        int fd_array[NUM];
        ClearFdArray(fd_array, NUM, DFL_FD);
        fd_array[0] = _listen_sock;

        for (;;) {
            FD_ZERO(&readfds);
            int maxfd = DFL_FD;
            for (int i = 0; i < NUM; i++) {
                if (fd_array[i] == DFL_FD) continue;
                FD_SET(fd_array[i], &readfds);
                if (fd_array[i] > maxfd) maxfd = fd_array[i];
            }

            struct timeval timeout = {5, 0}; // 超时 5 秒测试
            int ret = select(maxfd + 1, &readfds, nullptr, nullptr, &timeout);
            switch (ret) {
                case 0:
                    std::cout << "timeout: " << timeout.tv_sec << std::endl;
                    break;
                case -1:
                    std::cerr << "select error" << std::endl;
                    break;
                default:
                    std::cout << "有事件发生... timeout: " << timeout.tv_sec << std::endl;
                    HandlerEvent(readfds, fd_array, NUM);
                    break;
            }
        }
    }
private:
    void ClearFdArray(int fd_array[], int num, int default_fd) {
        for (int i = 0; i < num; i++) fd_array[i] = default_fd;
    }
    bool SetFdArray(int fd_array[], int num, int fd) {
        for (int i = 0; i < num; i++) {
            if (fd_array[i] == DFL_FD) { fd_array[i] = fd; return true; }
        }
        return false;
    }
    void HandlerEvent(const fd_set& readfds, int fd_array[], int num) {
        for (int i = 0; i < num; i++) {
            if (fd_array[i] == DFL_FD) continue;
            if (fd_array[i] == _listen_sock && FD_ISSET(fd_array[i], &readfds)) {
                struct sockaddr_in peer;
                socklen_t len = sizeof(peer);
                int sock = accept(_listen_sock, (struct sockaddr*)&peer, &len);
                if (sock < 0) { std::cerr << "accept error" << std::endl; continue; }
                std::cout << "get a new link[" << inet_ntoa(peer.sin_addr) << ":" << ntohs(peer.sin_port) << "]" << std::endl;
                if (!SetFdArray(fd_array, num, sock)) {
                    close(sock);
                    std::cout << "select server is full, close fd: " << sock << std::endl;
                }
            } else if (FD_ISSET(fd_array[i], &readfds)) {
                char buffer[1024];
                ssize_t size = read(fd_array[i], buffer, sizeof(buffer)-1);
                if (size > 0) {
                    buffer[size] = '\0';
                    std::cout << "echo# " << buffer << std::endl;
                } else if (size == 0) {
                    std::cout << "client quit" << std::endl;
                    close(fd_array[i]);
                    fd_array[i] = DFL_FD;
                } else {
                    std::cerr << "read error" << std::endl;
                    close(fd_array[i]);
                    fd_array[i] = DFL_FD;
                }
            }
        }
    }
    ~SelectServer() { if (_listen_sock >= 0) close(_listen_sock); }
};

主函数

#include "select_server.hpp"
#include <string>

static void Usage(std::string proc) {
    std::cerr << "Usage: " << proc << " port" << std::endl;
}

int main(int argc, char* argv[]) {
    if (argc != 2) { Usage(argv[0]); exit(1); }
    int port = atoi(argv[1]);
    SelectServer* svr = new SelectServer(port);
    svr->InitSelectServer();
    svr->Run();
    return 0;
}

测试与说明

• timeout 测试：
  • 设置 timeout = {5, 0}，若 5 秒内无事件就绪，打印剩余时间（通常为 0）。
  • 若有客户端连接（如通过 telnet），select 返回并处理事件，剩余时间可能为 4 秒。
• 行为：
  • 初始只监视监听 socket，客户端连接后加入新 socket。
  • 支持多客户端并发，单进程处理所有连接。
• 问题：
  • 未响应客户端（需监视写事件）。
  • 未定制协议，可能粘包。
  • 无缓冲区，直接操作原始数据。

select 的优点

• 多描述符等待：同时监视多个文件描述符，等待时间重叠，提高 I/O 效率。
• 分离等待与操作：仅负责等待，I/O 操作由 accept、read 等完成，无阻塞。

select 的缺点

• 参数重置：每次调用需手动重置 readfds 等，使用不便。
• 拷贝开销：fd_set 从用户态到内核态的全量拷贝，开销随描述符数量增加。
• 遍历开销：内核需遍历所有监视描述符，复杂度 O(n)。
• 数量限制：受 fd_set 位图大小限制（通常 1024，可通过 sizeof(fd_set)*8 查看），远低于进程最大文件描述符数（如默认 65535，可用 ulimit -n 查看）。

select 的适用场景

select 适用于多连接但活跃连接少的场景（如聊天服务器），不适合活跃连接多（如数据备份）或高并发场景，因其效率随描述符数量增加而下降。

---

二、I/O 多路转接之 poll

poll 初识

poll 是 select 的改进版本，同样用于多路转接，但通过 pollfd 结构分离输入输出参数，解决了 select 的一些局限性。它的定位与适用场景与 select 一致。

poll 函数

poll 函数原型如下：

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

参数说明

• fds：pollfd 结构数组，每个元素包含文件描述符、监视事件和就绪事件。
• nfds：fds 数组长度。
• timeout：超时时间（毫秒）。
  • -1：阻塞等待。
  • 0：非阻塞立即返回。
  • 特定值：超时返回。

返回值说明

• 成功：返回就绪描述符个数。
• 超时：返回 0。
• 失败：返回 -1，错误码如 EFAULT（地址空间错误）、EINTR（信号中断）等。

pollfd 结构

struct pollfd {
    int fd;       // 文件描述符，负值时忽略
    short events; // 监视事件
    short revents;// 就绪事件（返回时填充）
};

常用事件：

• POLLIN：数据可读。
• POLLOUT：数据可写。
• POLLERR：错误。
• POLLHUP：挂起（如管道写端关闭）。
• 可通过位运算组合（如 events |= POLLIN）。

poll 服务器实现

以下是一个 poll 服务器实现：

PollServer 类

#pragma once
#include "socket.hpp"
#include <poll.h>

#define BACK_LOG 5
#define NUM 1024
#define DFL_FD -1

class PollServer {
private:
    int _listen_sock;
    int _port;
public:
    PollServer(int port) : _port(port), _listen_sock(-1) {}
    void InitPollServer() {
        _listen_sock = Socket::SocketCreate();
        Socket::SocketBind(_listen_sock, _port);
        Socket::SocketListen(_listen_sock, BACK_LOG);
    }
    void Run() {
        struct pollfd fds[NUM];
        ClearPollfds(fds, NUM, DFL_FD);
        SetPollfds(fds, NUM, _listen_sock);
        for (;;) {
            switch (poll(fds, NUM, -1)) {
                case 0:
                    std::cout << "timeout..." << std::endl;
                    break;
                case -1:
                    std::cerr << "poll error" << std::endl;
                    break;
                default:
                    HandlerEvent(fds, NUM);
                    break;
            }
        }
    }
private:
    void ClearPollfds(struct pollfd fds[], int num, int default_fd) {
        for (int i = 0; i < num; i++) {
            fds[i].fd = default_fd;
            fds[i].events = 0;
            fds[i].revents = 0;
        }
    }
    bool SetPollfds(struct pollfd fds[], int num, int fd) {
        for (int i = 0; i < num; i++) {
            if (fds[i].fd == DFL_FD) {
                fds[i].fd = fd;
                fds[i].events |= POLLIN;
                return true;
            }
        }
        return false;
    }
    void UnSetPollfds(struct pollfd fds[], int pos) {
        fds[pos].fd = DFL_FD;
        fds[pos].events = 0;
        fds[pos].revents = 0;
    }
    void HandlerEvent(struct pollfd fds[], int num) {
        for (int i = 0; i < num; i++) {
            if (fds[i].fd == DFL_FD) continue;
            if (fds[i].fd == _listen_sock && (fds[i].revents & POLLIN)) {
                struct sockaddr_in peer;
                socklen_t len = sizeof(peer);
                int sock = accept(_listen_sock, (struct sockaddr*)&peer, &len);
                if (sock < 0) { std::cerr << "accept error" << std::endl; continue; }
                std::cout << "get a new link[" << inet_ntoa(peer.sin_addr) << ":" << ntohs(peer.sin_port) << "]" << std::endl;
                if (!SetPollfds(fds, NUM, sock)) {
                    close(sock);
                    std::cout << "poll server is full, close fd: " << sock << std::endl;
                }
            } else if (fds[i].revents & POLLIN) {
                char buffer[1024];
                ssize_t size = read(fds[i].fd, buffer, sizeof(buffer)-1);
                if (size > 0) {
                    buffer[size] = '\0';
                    std::cout << "echo# " << buffer << std::endl;
                } else if (size == 0) {
                    std::cout << "client quit" << std::endl;
                    close(fds[i].fd);
                    UnSetPollfds(fds, i);
                } else {
                    std::cerr << "read error" << std::endl;
                    close(fds[i].fd);
                    UnSetPollfds(fds, i);
                }
            }
        }
    }
    ~PollServer() { if (_listen_sock >= 0) close(_listen_sock); }
};

主函数

#include "poll_server.hpp"
#include <string>

static void Usage(std::string proc) {
    std::cerr << "Usage: " << proc << " port" << std::endl;
}

int main(int argc, char* argv[]) {
    if (argc != 2) { Usage(argv[0]); exit(1); }
    int port = atoi(argv[1]);
    PollServer* svr = new PollServer(port);
    svr->InitPollServer();
    svr->Run();
    return 0;
}

测试与说明

• 设置 timeout = -1，若无客户端连接，poll 阻塞等待。
• 使用 telnet 连接后，服务器打印客户端 IP 和端口，并回显数据。
• 支持多客户端并发，客户端退出后关闭连接并清理 fds。

poll 的优点

• 参数分离：events 和 revents 分离，无需每次重置。
• 无数量限制：监视描述符数量仅受内存限制（fds 数组可动态调整）。
• 效率提升：与 select 类似，可重叠等待时间。

poll 的缺点

• 遍历开销：返回后需遍历 fds 获取就绪描述符。
• 拷贝开销：每次调用需拷贝整个 fds 数组到内核。
• 轮询负担：内核需遍历所有监视描述符，复杂度 O(n)。

---

三、I/O 多路转接之 epoll

epoll 初识

epoll 是 Linux 2.6 引入的高性能 I/O 多路转接接口，专为大规模并发设计。它改进了 select 和 poll 的缺陷，被公认为 Linux 下最佳的多路转接方案。命名中的 “e” 可理解为 “extend”，即对 poll 的扩展。

epoll 相关系统调用

1. epoll_create：

   int epoll_create(int size);
   

   • 创建 epoll 实例，返回文件描述符。size 自 2.6.8 后被忽略，但需 > 0。
   • 使用完毕需用 close 关闭。

2. epoll_ctl：

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

   • 操作 epoll 实例：
     • EPOLL_CTL_ADD：添加事件。
     • EPOLL_CTL_MOD：修改事件。
     • EPOLL_CTL_DEL：删除事件。
   • epoll_event 结构：

     struct epoll_event {
         uint32_t events;    // 事件类型
         epoll_data_t data;  // 用户数据，通常用 data.fd
     };
     

   • 常用事件：EPOLLIN（可读）、EPOLLOUT（可写）、EPOLLET（边缘触发）等。

3. epoll_wait：

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

   • 等待就绪事件，返回就绪个数并填充 events。
   • timeout：-1（阻塞）、0（非阻塞）、特定值（超时）。

epoll 工作原理

红黑树与就绪队列

epoll 通过 eventpoll 结构管理：

struct eventpoll {
    struct rb_root rbr;     // 红黑树，存储监视事件
    struct list_head rdlist;// 就绪队列，存储就绪事件
    // ...
};

• 红黑树 (rbr)：记录需要监视的文件描述符及其事件，通过 epoll_ctl 增删改。
• 就绪队列 (rdlist)：存储已就绪的事件，epoll_wait 从中获取。

每个事件对应一个 epitem 结构：

struct epitem {
    struct rb_node rbn;      // 红黑树节点
    struct list_head rdllink;// 就绪队列节点
    struct epoll_filefd ffd; // 文件描述符
    struct epoll_event event;// 监视/就绪事件
};

回调机制

• 监视事件与底层驱动建立回调关系（如 ep_poll_callback）。
• 事件就绪时，回调函数自动将事件加入就绪队列，无需内核轮询。
• epoll_wait 只需检查就绪队列是否为空，复杂度 O(1)。

epoll 三部曲

1. epoll_create 创建模型。
2. epoll_ctl 注册事件。
3. epoll_wait 等待就绪事件。

epoll 服务器实现

EpollServer 类

#pragma once
#include "socket.hpp"
#include <sys/epoll.h>

#define BACK_LOG 5
#define SIZE 256
#define MAX_NUM 64

class EpollServer {
private:
    int _listen_sock;
    int _port;
    int _epfd;
public:
    EpollServer(int port) : _port(port), _listen_sock(-1), _epfd(-1) {}
    void InitEpollServer() {
        _listen_sock = Socket::SocketCreate();
        Socket::SocketBind(_listen_sock, _port);
        Socket::SocketListen(_listen_sock, BACK_LOG);
        _epfd = epoll_create(SIZE);
        if (_epfd < 0) { std::cerr << "epoll_create error" << std::endl; exit(5); }
    }
    void Run() {
        AddEvent(_listen_sock, EPOLLIN);
        for (;;) {
            struct epoll_event revs[MAX_NUM];
            int num = epoll_wait(_epfd, revs, MAX_NUM, -1);
            if (num < 0) { std::cerr << "epoll_wait error" << std::endl; continue; }
            if (num == 0) { std::cout << "timeout..." << std::endl; continue; }
            HandlerEvent(revs, num);
        }
    }
private:
    void AddEvent(int sock, uint32_t event) {
        struct epoll_event ev;
        ev.events = event;
        ev.data.fd = sock;
        epoll_ctl(_epfd, EPOLL_CTL_ADD, sock, &ev);
    }
    void DelEvent(int sock) {
        epoll_ctl(_epfd, EPOLL_CTL_DEL, sock, nullptr);
    }
    void HandlerEvent(struct epoll_event revs[], int num) {
        for (int i = 0; i < num; i++) {
            int fd = revs[i].data.fd;
            if (fd == _listen_sock && (revs[i].events & EPOLLIN)) {
                struct sockaddr_in peer;
                socklen_t len = sizeof(peer);
                int sock = accept(_listen_sock, (struct sockaddr*)&peer, &len);
                if (sock < 0) { std::cerr << "accept error" << std::endl; continue; }
                std::cout << "get a new link[" << inet_ntoa(peer.sin_addr) << ":" << ntohs(peer.sin_port) << "]" << std::endl;
                AddEvent(sock, EPOLLIN);
            } else if (revs[i].events & EPOLLIN) {
                char buffer[64];
                ssize_t size = recv(fd, buffer, sizeof(buffer)-1, 0);
                if (size > 0) {
                    buffer[size] = '\0';
                    std::cout << "echo# " << buffer << std::endl;
                } else if (size == 0) {
                    std::cout << "client quit" << std::endl;
                    close(fd);
                    DelEvent(fd);
                } else {
                    std::cerr << "recv error" << std::endl;
                    close(fd);
                    DelEvent(fd);
                }
            }
        }
    }
    ~EpollServer() {
        if (_listen_sock >= 0) close(_listen_sock);
        if (_epfd >= 0) close(_epfd);
    }
};

主函数

#include "epoll_server.hpp"
#include <string>

static void Usage(std::string proc) {
    std::cerr << "Usage: " << proc << " port" << std::endl;
}

int main(int argc, char* argv[]) {
    if (argc != 2) { Usage(argv[0]); exit(1); }
    int port = atoi(argv[1]);
    EpollServer* svr = new EpollServer(port);
    svr->InitEpollServer();
    svr->Run();
    return 0;
}

测试与说明

• 设置 timeout = -1，无连接时阻塞等待。
• 使用 telnet 测试，服务器支持多客户端并发，打印客户端数据。
• 查看文件描述符：ls /proc/PID/fd，包含监听 socket、epoll 实例和客户端连接。
  代码链接：I/O多路转接

epoll 的优点

• 接口清晰：三函数分离职责，使用方便。
• 轻量拷贝：仅 epoll_ctl 拷贝数据，epoll_wait 只返回就绪事件。
• 高效检测：回调机制，复杂度 O(1)。
• 无数量限制：红黑树支持任意数量描述符（仅受内存限制）。

epoll 工作方式

水平触发（LT，Level Triggered）

• 默认模式，只要事件就绪就持续通知。
• 支持阻塞和非阻塞 I/O。
• 类似 select 和 poll，可延迟处理数据。

边缘触发（ET，Edge Triggered）

• 设置 EPOLLET，仅在事件状态变化（如无到有）时通知。
• 需一次性处理所有数据，仅支持非阻塞 I/O。
• 示例（非阻塞读取）：

  int flags = fcntl(fd, F_GETFL, 0);
  fcntl(fd, F_SETFL, flags | O_NONBLOCK);
  char buffer[64];
  while (true) {
      ssize_t size = recv(fd, buffer, sizeof(buffer)-1, 0);
      if (size <= 0) break; // 处理完或出错
      buffer[size] = '\0';
      std::cout << buffer << std::endl;
  }
  

LT 与 ET 对比

• LT：简单，适合延迟处理，但可能重复通知。
• ET：高效（通知次数少，如 Nginx 默认使用），但需一次性处理，编程复杂。

---

四、总结与对比

特性	select	poll	epoll
数据结构	fd_set 位图	pollfd 数组	红黑树 + 就绪队列
数量限制	1024（fd_set 大小）	无（内存限制）	无（内存限制）
数据拷贝	每次全量拷贝	每次全量拷贝	仅 epoll_ctl 拷贝
事件检测	轮询 O(n)	轮询 O(n)	回调 O(1)
接口复杂度	单函数，需重置参数	单函数，参数分离	三函数，职责清晰
工作模式	LT	LT	LT / ET

• select：适合小规模、低活跃连接场景，简单但效率低。
• poll：改进了数量限制，适合中等规模，但仍受限于轮询。
• epoll：高性能，适合大规模高并发，推荐现代服务器开发。