如何实现两个TCP Socket间的高速数据传输？

原代码中的核心问题 #

• 逻辑错误：send_non_block_tcp_socket_data的循环条件while (i < n)中，n未初始化就使用，属于未定义行为，直接导致传输逻辑混乱，影响效率
• 文件标志位覆盖：set_tcp_socket_non_blocking直接将文件标志位设为O_NONBLOCK，丢失了原有的O_RDWR等标志，可能引发Socket操作异常
• 低效事件处理：transfer_data_tcp_sockets嵌套循环冗余，无数据时进入死循环等待，且每次就绪只处理一次读写，未批量处理数据，浪费select调用开销
• 冗余初始化：send_non_block_tcp_socket_data每次循环重复初始化fd_set和timeval，超时后重置timeval的操作可简化

针对性优化方案 #

1. 修复核心逻辑错误 #

修正发送函数循环条件

bool send_non_block_tcp_socket_data(int socket_id, void *data, int data_size)
{
    if (socket_id == -1) return false;
    char *ptr = (char*)data;
    int i = 0;

    while (i < data_size) {
        fd_set write_fds, err_fds;
        struct timeval tv_delay = {10, 0}; // 每次select前重新初始化超时

        FD_ZERO(&write_fds);
        FD_SET(socket_id, &write_fds);
        FD_ZERO(&err_fds);
        FD_SET(socket_id, &err_fds);

        int ret = select(socket_id + 1, NULL, &write_fds, &err_fds, &tv_delay);
        if (ret < 0) {
            // 信号中断错误，重试
            if (errno == EINTR) continue;
            return false;
        } else if (ret == 0) {
            // 超时返回失败
            return false;
        }

        if (FD_ISSET(socket_id, &err_fds)) return false;
        if (FD_ISSET(socket_id, &write_fds)) {
            ssize_t sent = send(socket_id, ptr + i, data_size - i, 0);
            if (sent == -1) {
                // 非阻塞发送暂时不可写，重试
                if (errno == EAGAIN || errno == EWOULDBLOCK) continue;
                return false;
            }
            i += sent;
        }
    }
    return true;
}

修复非阻塞Socket设置函数

bool set_tcp_socket_non_blocking(int socket_id)
{
    if (socket_id == -1) return false;
    int op = fcntl(socket_id, F_GETFL);
    if (op == -1) return false;
    // 保留原有标志位，仅添加非阻塞标志
    if (fcntl(socket_id, F_SETFL, op | O_NONBLOCK) == -1) return false;
    return true;
}

2. 重构传输循环，提升事件处理效率 #

将transfer_data_tcp_sockets改为批量处理就绪数据，避免冗余select调用：

bool transfer_data_tcp_sockets(int socket_id, int socket_id2)
{
    char buffer[TRANSFER_BUFF_SIZE];
    fd_set read_fds, write_fds, err_fds;
    int max_fd = (socket_id > socket_id2) ? socket_id : socket_id2;
    ssize_t n;

    if (!set_tcp_socket_non_blocking(socket_id) || !set_tcp_socket_non_blocking(socket_id2)) {
        return false;
    }

    // 开启TCP_NODELAY，避免Nagle算法延迟
    int flag = 1;
    setsockopt(socket_id, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));
    setsockopt(socket_id2, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));

    while (true) {
        FD_ZERO(&read_fds);
        FD_SET(socket_id, &read_fds);
        FD_SET(socket_id2, &read_fds);
        FD_ZERO(&err_fds);
        FD_SET(socket_id, &err_fds);
        FD_SET(socket_id2, &err_fds);

        // 无限等待直到有事件发生
        int ret = select(max_fd + 1, &read_fds, NULL, &err_fds, NULL);
        if (ret < 0) {
            if (errno == EINTR) continue;
            fprintf(stderr, "select error\n");
            return false;
        }

        // 处理错误事件
        if (FD_ISSET(socket_id, &err_fds) || FD_ISSET(socket_id2, &err_fds)) {
            return false;
        }

        // 批量处理socket_id的可读数据
        if (FD_ISSET(socket_id, &read_fds)) {
            while ((n = recv(socket_id, buffer, TRANSFER_BUFF_SIZE, 0)) > 0) {
                ssize_t sent = 0;
                while (sent < n) {
                    ssize_t s = send(socket_id2, buffer + sent, n - sent, 0);
                    if (s == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK) {
                            // 等待socket_id2可写
                            FD_ZERO(&write_fds);
                            FD_SET(socket_id2, &write_fds);
                            ret = select(socket_id2 + 1, NULL, &write_fds, &err_fds, NULL);
                            if (ret <= 0) return false;
                            continue;
                        }
                        return false;
                    }
                    sent += s;
                }
            }
            if (n == 0) {
                // 连接关闭，关闭对端写通道
                shutdown(socket_id2, SHUT_WR);
                return true;
            }
            if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
                return false;
            }
        }

        // 批量处理socket_id2的可读数据
        if (FD_ISSET(socket_id2, &read_fds)) {
            while ((n = recv(socket_id2, buffer, TRANSFER_BUFF_SIZE, 0)) > 0) {
                ssize_t sent = 0;
                while (sent < n) {
                    ssize_t s = send(socket_id, buffer + sent, n - sent, 0);
                    if (s == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK) {
                            FD_ZERO(&write_fds);
                            FD_SET(socket_id, &write_fds);
                            ret = select(socket_id + 1, NULL, &write_fds, &err_fds, NULL);
                            if (ret <= 0) return false;
                            continue;
                        }
                        return false;
                    }
                    sent += s;
                }
            }
            if (n == 0) {
                shutdown(socket_id, SHUT_WR);
                return true;
            }
            if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
                return false;
            }
        }
    }
}

3. 缓冲区与TCP参数优化 #

• 匹配系统缓冲区：你发现TRANSFER_BUFF_SIZE=1000效果最佳，可能是因为该大小适配内核TCP分段。可通过以下代码获取系统默认缓冲区大小，以此为基准调整：

int optval;
socklen_t optlen = sizeof(optval);
getsockopt(socket_id, SOL_SOCKET, SO_RCVBUF, &optval, &optlen);
// 可将TRANSFER_BUFF_SIZE设置为optval的1-2倍，减少recv调用次数

• 增大TCP缓冲区：调整Socket的发送/接收缓冲区大小，提升吞吐量：

int buf_size = 64 * 1024; // 64KB，可根据网络环境调整
setsockopt(socket_id, SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof(buf_size));
setsockopt(socket_id, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof(buf_size));
setsockopt(socket_id2, SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof(buf_size));
setsockopt(socket_id2, SOL_SOCKET, SO_RCVBUF, &buf_size, sizeof(buf_size));

4. 细节优化 #

• 处理EINTR错误：select被信号中断时直接重试，不要返回失败
• 避免不必要的超时：除非业务需要，否则select使用NULL超时参数，无限等待事件
• 减少fd_set重复初始化：每次select前重新构建fd_set即可，无需冗余操作

内容的提问来源于stack exchange，提问作者CharlieB