高频交易系统中的WebSocket网络缓冲区优化技术

// 这些是TCP协议相关的选项
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));
setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPALIVE, &flag, sizeof(flag));

// 创建套接字
int sockfd = socket(AF_INET, SOCK_STREAM, 0); // SOCK_STREAM指定TCP协议

// 设置套接字选项
setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf));

// 连接、发送、接收数据
connect(sockfd, ...);
send(sockfd, ...);
recv(sockfd, ...);

# 启用TCP Fast Open
sysctl -w net.ipv4.tcp_fastopen=3

# 持久化设置
echo "net.ipv4.tcp_fastopen=3" >> /etc/sysctl.conf

// 在套接字上禁用Nagle算法
int flag = 1;
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));

# 设置BBR拥塞控制算法
sysctl -w net.ipv4.tcp_congestion_control=bbr

# 持久化设置
echo "net.ipv4.tcp_congestion_control=bbr" >> /etc/sysctl.conf

# 禁用TCP缓冲区自动调整
sysctl -w net.ipv4.tcp_moderate_rcvbuf=0

# 设置全局TCP缓冲区参数(最小,默认,最大)
sysctl -w net.ipv4.tcp_rmem="4096 131072 8388608"
sysctl -w net.ipv4.tcp_wmem="4096 65536 4194304"

# 设置内存压力点
sysctl -w net.ipv4.tcp_mem="8388608 8388608 8388608"

// 优化发送缓冲区(小型交易指令)
int snd_buf = 16 * 1024;  // 16KB
setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, &snd_buf, sizeof(snd_buf));

// 优化接收缓冲区(市场数据)
int rcv_buf = 4 * 1024 * 1024;  // 4MB
setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &rcv_buf, sizeof(rcv_buf));

// 交易指令连接(优化延迟)
int trade_snd_buf = 16 * 1024;  // 16KB
setsockopt(trade_sock, SOL_SOCKET, SO_SNDBUF, &trade_snd_buf, sizeof(trade_snd_buf));

// 市场数据连接(优化吞吐量)
int market_rcv_buf = 4 * 1024 * 1024;  // 4MB
setsockopt(market_sock, SOL_SOCKET, SO_RCVBUF, &market_rcv_buf, sizeof(market_rcv_buf));

# 分配大页内存
echo 1024 > /proc/sys/vm/nr_hugepages  # 分配1024个2MB大页

# 禁用透明大页(避免不可预测的延迟)
echo never > /sys/kernel/mm/transparent_hugepage/enabled
echo never > /sys/kernel/mm/transparent_hugepage/defrag

# 注意：Linux内核TCP栈的sk_buff分配器不支持通过sysctl直接使用大页内存。
# 不存在 net.ipv4.tcp_use_hugepages 这个内核参数。
# 若需在网络路径中利用大页优势，需使用DPDK等用户态协议栈绕过内核。

#include <sys/mman.h>
#include <fcntl.h>

// 分配大页内存用于WebSocket缓冲区
int fd = open("/dev/hugepages/ws_buffer", O_CREAT | O_RDWR, 0755);
void* buffer = mmap(NULL, BUFFER_SIZE, PROT_READ | PROT_WRITE, 
                   MAP_SHARED, fd, 0);

// 初始化缓冲区
ws_buffer_init(buffer, BUFFER_SIZE);

# 设置网卡中断亲和性
echo "f" > /proc/irq/$(cat /proc/interrupts | grep eth0 | awk '{print $1}' | tr -d :)/smp_affinity

# 调整中断合并参数
ethtool -C eth0 rx-usecs 0 rx-frames 1

# 配置RSS将处理分散到多核
ethtool -L eth0 combined 8

# 将WebSocket客户端绑定到特定CPU核心和NUMA节点
taskset -c 0,2,4,6 numactl --membind=0 ./ws_client

// 锁定内存，防止页面交换
mlockall(MCL_CURRENT | MCL_FUTURE);

// 初始化高性能WebSocket客户端
void initHighPerformanceClient() {
    // 1. 配置系统参数
    system("sysctl -w net.ipv4.tcp_fastopen=3");
    system("sysctl -w net.ipv4.tcp_congestion_control=bbr");
    system("echo 1024 > /proc/sys/vm/nr_hugepages");
    
    // 2. 创建套接字
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    
    // 3. 优化TCP参数
    int flag = 1;
    setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));
    
    // 4. 优化Socket缓冲区
    int snd_buf = 64 * 1024;  // 交易指令发送缓冲区
    int rcv_buf = 2 * 1024 * 1024;  // 市场数据接收缓冲区
    setsockopt(sockfd, SOL_SOCKET, SO_SNDBUF, &snd_buf, sizeof(snd_buf));
    setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &rcv_buf, sizeof(rcv_buf));
    
    // 5. 分配大页内存用于WebSocket处理
    int fd = open("/dev/hugepages/ws_buffer", O_CREAT | O_RDWR, 0755);
    void* buffer = mmap(NULL, BUFFER_SIZE, PROT_READ | PROT_WRITE, 
                       MAP_SHARED, fd, 0);
    
    // 6. 锁定内存，防止页面交换
    mlockall(MCL_CURRENT | MCL_FUTURE);
    
    // 7. 连接到服务器
    // connect(sockfd, ...);
    
    // 8. 使用优化的内存缓冲区处理WebSocket协议
    // websocket_init(sockfd, buffer, BUFFER_SIZE);
}

function measureRTT() {
    const timestamps = new Map();
    const messageId = Date.now().toString();
    
    // 记录发送时间(高精度时间戳)
    timestamps.set(messageId, performance.now());
    
    // 发送消息
    socket.send(JSON.stringify({
        id: messageId,
        type: "ping",
        timestamp: Date.now()
    }));
    
    // 接收响应
    socket.onmessage = (event) => {
        const data = JSON.parse(event.data);
        if (data.type === "pong" && data.replyTo === messageId) {
            const endTime = performance.now();
            const startTime = timestamps.get(data.replyTo);
            const rtt = endTime - startTime;
            
            console.log(`RTT: ${rtt.toFixed(3)} ms`);
        }
    };
}

function analyzeLatency(samples) {
    // 基本统计
    const avg = samples.reduce((a, b) => a + b, 0) / samples.length;
    const sorted = [...samples].sort((a, b) => a - b);
    const median = sorted[Math.floor(sorted.length / 2)];
    const min = sorted[0];
    const max = sorted[sorted.length - 1];
    
    // 百分位数分析
    const p99 = sorted[Math.floor(sorted.length * 0.99)];
    const p95 = sorted[Math.floor(sorted.length * 0.95)];
    const p50 = median;
    
    // 抖动计算
    let jitter = 0;
    for (let i = 1; i < samples.length; i++) {
        jitter += Math.abs(samples[i] - samples[i-1]);
    }
    jitter /= (samples.length - 1);
    
    return { avg, median, min, max, p50, p95, p99, jitter };
}