select/poll系统调用多写事件问题：公平性与EAGAIN处理问询

1. 文件描述符的公平触发问题 #

select和poll本身不提供公平性保证：内核不会自动将已处理的可写fd移到触发队列尾部。实际处理时，你会按照fd_set/pollfd数组的顺序从头遍历，先处理的fd会优先占用内核发送缓冲区，导致后面的fd还没处理就遇到缓冲区满，返回EAGAIN。

要实现所有套接字的公平处理，可以用以下两种方式：

• 轮询遍历偏移：记录上次处理到的数组索引，下次从该位置开始遍历，循环到数组末尾后再从头开始，避免每次都从第一个fd开始抢占资源。
• 动态调整fd顺序：每次处理完一个可写fd后，将其移到pollfd数组的尾部（或select的fd_set末尾），让后续的fd有优先处理的机会。

2. EAGAIN的处理与事件循环 #

出现EAGAIN时绝对不能跳过整个事件循环：

• 若EAGAIN是write/send返回的，说明当前fd的内核发送缓冲区已满，只需保留该fd的可写监听，后续缓冲区有空时内核会再次触发可写事件。
• 若EAGAIN是select/poll系统调用本身返回的（如部分UNIX系统资源不足），需像处理EINTR一样重新调用select/poll，不能终止循环。

你提到的可移植性处理是对的：跨平台代码要同时检查EAGAIN和EINTR，遇到时重新进入事件循环。

3. C语言代码示例（基于poll的公平处理） #

以下是实现轮询遍历+EAGAIN处理的示例代码：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <poll.h>
#include <errno.h>
#include <string.h>

#define MAX_CONNS 1024
#define BUF_SIZE 4096

// 存储每个连接的待发送数据
typedef struct {
    char buf[BUF_SIZE];
    size_t len;
    size_t sent;
} ConnData;

int main() {
    struct pollfd fds[MAX_CONNS];
    ConnData conn_data[MAX_CONNS];
    int conn_count = 0;
    int last_process_idx = 0; // 记录上次处理到的索引，实现轮询

    // 初始化监听fd（此处省略socket绑定监听逻辑，假设已添加到fds[0]）
    fds[0].fd = /* 替换为你的监听socket fd */;
    fds[0].events = POLLIN;
    conn_count++;

    while (1) {
        int ret = poll(fds, conn_count, -1);
        if (ret == -1) {
            // 处理系统调用错误，EAGAIN/EINTR时重新循环
            if (errno == EINTR || errno == EAGAIN) {
                continue;
            }
            perror("poll failed");
            exit(EXIT_FAILURE);
        }

        // 从上次处理的位置开始轮询，实现公平性
        int processed = 0;
        int idx = last_process_idx;
        while (processed < ret) {
            // 循环遍历数组，处理完末尾回到开头
            if (idx >= conn_count) {
                idx = 0;
            }

            struct pollfd *pfd = &fds[idx];
            ConnData *data = &conn_data[idx];

            if (pfd->revents & POLLOUT) {
                processed++;
                // 发送待发送数据
                ssize_t sent = send(pfd->fd, data->buf + data->sent, data->len - data->sent, MSG_NOSIGNAL);
                if (sent == -1) {
                    if (errno == EAGAIN || errno == EWOULDBLOCK) {
                        // 缓冲区满，保留POLLOUT监听，等待下次触发
                        continue;
                    } else {
                        // 连接错误，关闭fd并移除
                        close(pfd->fd);
                        // 将最后一个fd移到当前位置，减少数组移动开销
                        *pfd = fds[conn_count - 1];
                        *data = conn_data[conn_count - 1];
                        conn_count--;
                        // 调整last_process_idx，避免跳过元素
                        if (last_process_idx == idx) {
                            last_process_idx--;
                        }
                        idx--; // 因为当前位置被替换，需要重新检查
                        continue;
                    }
                }

                data->sent += sent;
                if (data->sent == data->len) {
                    // 数据发送完成，取消POLLOUT监听，重置发送状态
                    pfd->events &= ~POLLOUT;
                    data->len = 0;
                    data->sent = 0;
                }
            }

            if (pfd->revents & POLLIN) {
                processed++;
                // 处理读事件（此处省略复杂接收逻辑，示例为回显场景）
                char buf[BUF_SIZE];
                ssize_t recv_len = recv(pfd->fd, buf, BUF_SIZE, 0);
                if (recv_len <= 0) {
                    close(pfd->fd);
                    *pfd = fds[conn_count - 1];
                    *data = conn_data[conn_count - 1];
                    conn_count--;
                    if (last_process_idx == idx) {
                        last_process_idx--;
                    }
                    idx--;
                    continue;
                }
                // 将接收的数据存入待发送缓冲区，标记可写
                memcpy(data->buf, buf, recv_len);
                data->len = recv_len;
                data->sent = 0;
                pfd->events |= POLLOUT;
            }

            idx++;
        }

        // 更新下次处理的起始位置
        last_process_idx = idx % conn_count;
    }

    return 0;
}

代码关键点解释 #

• 轮询遍历：通过last_process_idx记录上次处理的结束位置，下次从该位置开始遍历，避免每次都优先处理数组前面的fd，保证公平性。
• EAGAIN处理：send返回EAGAIN时，保留该fd的POLLOUT监听，等待内核缓冲区有空时再次触发可写事件，不中断整个循环。
• 连接管理：当连接出错时，将数组末尾的fd移到当前位置，减少数组元素移动的开销，同时调整last_process_idx避免逻辑错误。
• 可移植性：poll返回-1时，检查EAGAIN和EINTR，遇到时重新调用poll，符合POSIX标准。

内容的提问来源于stack exchange，提问作者jorge is not ai