如何在Ubuntu22.04下利用RSS实现Raw Socket按四元组多线程分发包？

核心结论 #

完全可以利用NIC的RSS（接收端缩放）技术结合Linux内核特性，实现无需调度线程的Raw Socket四元组分发，主要有两种可靠方案：SO_INCOMING_CPU绑定RSS队列 和 SO_REUSEPORT + BPF过滤。

第一步：确认并配置NIC的RSS支持 #

首先要确保你的网卡支持RSS并正确配置：

1. 检查RSS状态：

   ethtool -k eth0 | grep rx-rss
   

   输出显示rx-rss: on即为开启状态。
2. 配置RSS hash覆盖四元组（确保同四元组的包被分到同一个RSS队列）：

   # 针对TCP IPv4设置hash因子为源IP、目的IP、源端口、目的端口
   ethtool -N eth0 rx-flow-hash tcp4 sdfn
   # 针对UDP IPv4同理
   ethtool -N eth0 rx-flow-hash udp4 sdfn
   

3. 查看RSS队列数量：

   ethtool -l eth0 | grep -A2 rx
   

   记录current count的值，后续创建的线程/套接字数量建议与这个值匹配。

方案一：SO_INCOMING_CPU绑定RSS队列 #

该方案让内核将指定RSS队列的数据包直接分发到绑定对应CPU的Raw Socket，线程只需处理自己的套接字即可。

实现步骤 #

1. 创建与RSS队列数量相同的Raw Socket。
2. 每个套接字通过SO_INCOMING_CPU选项绑定到对应CPU核心（CPU编号与RSS队列编号通常一一对应，可通过cat /proc/interrupts查看网卡中断的CPU绑定确认）。
3. 每个线程单独监听一个套接字，处理数据包。

代码示例（C语言） #

#include <sys/socket.h>
#include <linux/if_packet.h>
#include <net/ethernet.h>
#include <unistd.h>
#include <pthread.h>
#include <net/if.h>

#define RSS_QUEUE_COUNT 4 // 替换为你的网卡RSS队列数
#define NIC_NAME "eth0"   // 替换为你的网卡名

void* packet_handler(void* arg) {
    int sock_fd = *(int*)arg;
    unsigned char buf[4096];
    ssize_t recv_len;

    while (1) {
        recv_len = recvfrom(sock_fd, buf, sizeof(buf), 0, NULL, NULL);
        if (recv_len < 0) continue;
        // 此处添加数据包处理逻辑
    }
    return NULL;
}

int main() {
    int socks[RSS_QUEUE_COUNT];
    pthread_t threads[RSS_QUEUE_COUNT];
    struct sockaddr_ll bind_addr = {0};
    int if_index = if_nametoindex(NIC_NAME);

    // 初始化每个套接字并绑定CPU
    for (int i = 0; i < RSS_QUEUE_COUNT; i++) {
        // 创建Raw Socket
        socks[i] = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL));
        if (socks[i] < 0) {
            perror("socket create failed");
            return 1;
        }

        // 绑定到指定CPU
        int target_cpu = i;
        if (setsockopt(socks[i], SOL_SOCKET, SO_INCOMING_CPU, &target_cpu, sizeof(target_cpu)) < 0) {
            perror("SO_INCOMING_CPU set failed");
            return 1;
        }

        // 绑定到目标网卡
        bind_addr.sll_family = AF_PACKET;
        bind_addr.sll_ifindex = if_index;
        bind_addr.sll_protocol = htons(ETH_P_ALL);
        if (bind(socks[i], (struct sockaddr*)&bind_addr, sizeof(bind_addr)) < 0) {
            perror("bind failed");
            return 1;
        }

        // 创建线程处理该套接字
        pthread_create(&threads[i], NULL, packet_handler, &socks[i]);
    }

    // 等待所有线程结束
    for (int i = 0; i < RSS_QUEUE_COUNT; i++) {
        pthread_join(threads[i], NULL);
    }

    return 0;
}

方案二：SO_REUSEPORT + BPF过滤 #

该方案利用SO_REUSEPORT让多个Raw Socket共享同一网卡监听，再通过BPF过滤器基于四元组hash值分发数据包，无需依赖NIC的RSS（但结合RSS性能更佳）。

实现步骤 #

1. 创建与线程数相同的Raw Socket，开启SO_REUSEPORT选项。
2. 为每个套接字附加BPF过滤器，过滤器会计算数据包四元组的hash值，仅保留hash取模后等于线程编号的包。
3. 每个线程处理对应套接字的数据包。

代码示例（C语言） #

#include <sys/socket.h>
#include <linux/if_packet.h>
#include <net/ethernet.h>
#include <linux/bpf.h>
#include <linux/filter.h>
#include <unistd.h>
#include <pthread.h>
#include <net/if.h>

#define THREAD_COUNT 4    // 自定义线程数，建议与CPU核心数匹配
#define NIC_NAME "eth0"   // 替换为你的网卡名

// 生成基于四元组hash的BPF过滤器
struct sock_fprog build_bpf_filter(int thread_id) {
    struct sock_filter filter_instructions[] = {
        // 跳过以太网帧头（14字节）
        BPF_STMT(BPF_LD + BPF_B + BPF_ABS, 14),
        // 仅处理IPv4数据包（0x0800）
        BPF_JUMP(BPF_JMP + BPF_JEQ + BPF_K, 0x0800, 0, 10),
        // 获取IP头长度（IP头首字节低4位 *4）
        BPF_STMT(BPF_LDX + BPF_B + BPF_MSH, 14),
        // 加载源IP并加入hash计算
        BPF_STMT(BPF_LD + BPF_W + BPF_IND, 14 + 12),
        BPF_STMT(BPF_ST, 0),
        // 加载目的IP并异或到hash
        BPF_STMT(BPF_LD + BPF_W + BPF_IND, 14 + 16),
        BPF_STMT(BPF_XOR, 0),
        // 加载源端口并异或到hash
        BPF_STMT(BPF_LD + BPF_H + BPF_IND, 14 + 20),
        BPF_STMT(BPF_XOR, 0),
        // 加载目的端口并异或到hash
        BPF_STMT(BPF_LD + BPF_H + BPF_IND, 14 + 22),
        BPF_STMT(BPF_XOR, 0),
        // hash值取模线程数
        BPF_STMT(BPF_ALU + BPF_MOD + BPF_K, THREAD_COUNT),
        // 仅保留与线程编号匹配的包
        BPF_JUMP(BPF_JMP + BPF_JEQ + BPF_K, thread_id, 0, 1),
        BPF_STMT(BPF_RET + BPF_K, 0xFFFF), // 保留数据包
        BPF_STMT(BPF_RET + BPF_K, 0),      // 丢弃数据包
    };

    struct sock_fprog filter_prog = {
        .len = sizeof(filter_instructions) / sizeof(filter_instructions[0]),
        .filter = filter_instructions,
    };
    return filter_prog;
}

void* packet_handler(void* arg) {
    int sock_fd = *(int*)arg;
    unsigned char buf[4096];
    ssize_t recv_len;

    while (1) {
        recv_len = recvfrom(sock_fd, buf, sizeof(buf), 0, NULL, NULL);
        if (recv_len < 0) continue;
        // 此处添加数据包处理逻辑
    }
    return NULL;
}

int main() {
    int socks[THREAD_COUNT];
    pthread_t threads[THREAD_COUNT];
    struct sockaddr_ll bind_addr = {0};
    int if_index = if_nametoindex(NIC_NAME);
    int reuse_port = 1;

    for (int i = 0; i < THREAD_COUNT; i++) {
        // 创建Raw Socket
        socks[i] = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_ALL));
        if (socks[i] < 0) {
            perror("socket create failed");
            return 1;
        }

        // 开启SO_REUSEPORT
        if (setsockopt(socks[i], SOL_SOCKET, SO_REUSEPORT, &reuse_port, sizeof(reuse_port)) < 0) {
            perror("SO_REUSEPORT set failed");
            return 1;
        }

        // 绑定到目标网卡
        bind_addr.sll_family = AF_PACKET;
        bind_addr.sll_ifindex = if_index;
        bind_addr.sll_protocol = htons(ETH_P_ALL);
        if (bind(socks[i], (struct sockaddr*)&bind_addr, sizeof(bind_addr)) < 0) {
            perror("bind failed");
            return 1;
        }

        // 附加BPF过滤器
        struct sock_fprog filter = build_bpf_filter(i);
        if (setsockopt(socks[i], SOL_SOCKET, SO_ATTACH_FILTER, &filter, sizeof(filter)) < 0) {
            perror("BPF filter attach failed");
            return 1;
        }

        // 创建线程处理该套接字
        pthread_create(&threads[i], NULL, packet_handler, &socks[i]);
    }

    // 等待所有线程结束
    for (int i = 0; i < THREAD_COUNT; i++) {
        pthread_join(threads[i], NULL);
    }

    return 0;
}

注意事项 #

• 运行Raw Socket程序需要root权限，执行时需加sudo。
• 方案一依赖NIC的RSS支持，方案二则可在无RSS的网卡上运行，但性能略逊于前者。
• Ubuntu 22.04内核版本为5.15+，完全支持SO_INCOMING_CPU和SO_REUSEPORT特性。
• 若使用方案一，可通过ethtool -x eth0查看RSS队列与CPU的映射关系，确保绑定的CPU正确。

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