# 检查关键配置
zgrep -E "BPF|XDP|BPF_SYSCALL|BPF_JIT|NET_CLS_BPF|NET_ACT_BPF|BPF_STREAM_PARSER" /proc/config.gz

# 基础工具
sudo apt update && sudo apt install -y 
  clang llvm libbpf-dev linux-tools-$(uname -r) 
  linux-headers-$(uname -r) bpfcc-tools iproute2 
  pkg-config gcc make git

# 或使用RHEL系
sudo dnf install -y clang llvm libbpf-devel kernel-devel-$(uname -r) 
  bpftool iproute-tc perf

# 检查bpftool版本
bpftool version
# 检查JIT状态
cat /proc/sys/net/core/bpf_jit_enable  # 应输出 1
# 测试XDP挂载能力
sudo ip link set dev eth0 xdp off 2>/dev/null && echo "XDP supported"

[网卡驱动] --> [XDP程序] --(命中媒体流)--> [XSK_MAP] --> [用户态Worker (AF_XDP)]
                    |
                    +--(非媒体流/异常包)--> [内核协议栈] --> [原有应用]

#ifndef __MEDIA_XDP_COMMON_H
#define __MEDIA_XDP_COMMON_H

#include <linux/types.h>
#include <bpf/bpf_endian.h>

// 媒体流识别规则：支持端口范围 + 协议特征
struct media_rule {
    __u16 proto;          // IPPROTO_UDP / IPPROTO_TCP
    __u16 port_min;       // 端口范围下界
    __u16 port_max;       // 端口范围上界
    __u32 flags;          // 扩展标志：BIT0=RTP, BIT1=QUIC
};

// 共享配置Map
struct {
    __uint(type, BPF_MAP_TYPE_ARRAY);
    __uint(max_entries, 64);
    __type(key, __u32);
    __type(value, struct media_rule);
} media_rules SEC(".maps");

// XSK重定向Map
struct {
    __uint(type, BPF_MAP_TYPE_XSKMAP);
    __uint(max_entries, 64); // 对应队列数
    __type(key, __u32);
    __type(value, __u32);
} xsks_map SEC(".maps");

// 统计计数器
struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
    __uint(max_entries, 8);
    __type(key, __u32);
    __type(value, __u64);
} xdp_stats SEC(".maps");

enum stat_idx {
    STAT_TOTAL_RX = 0,
    STAT_PASS_KERNEL = 1,
    STAT_REDIRECT_XSK = 2,
    STAT_DROP_INVALID = 3,
    STAT_L4_PARSE_FAIL = 4,
};

#endif

#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/udp.h>
#include <linux/tcp.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_endian.h>
#include "common.h"

SEC("xdp")
int xdp_media_bypass(struct xdp_md *ctx) {
    void *data_end = (void *)(long)ctx->data_end;
    void *data = (void *)(long)ctx->data;
    struct ethhdr *eth = data;
    __u64 *stats;
    __u32 key = 0;

    // 1. 以太网头检查
    if ((void *)(eth + 1) > data_end) return XDP_DROP;
    if (eth->h_proto != bpf_htons(ETH_P_IP)) {
        // 非IPv4直接放行内核栈
        stats = bpf_map_lookup_elem(&xdp_stats, &key);
        if (stats) __sync_fetch_and_add(&stats[STAT_PASS_KERNEL], 1);
        return XDP_PASS;
    }

    struct iphdr *ip = (void *)(eth + 1);
    if ((void *)(ip + 1) > data_end) return XDP_DROP;

    // 2. 遍历规则表匹配五元组
    for (int i = 0; i < 64; i++) {
        struct media_rule *rule = bpf_map_lookup_elem(&media_rules, &i);
        if (!rule) continue;

        if (ip->protocol != rule->proto) continue;

        void *l4 = (void *)ip + (ip->ihl << 2);
        if (rule->proto == IPPROTO_UDP) {
            struct udphdr *udp = l4;
            if ((void *)(udp + 1) > data_end) break;
            __u16 dport = bpf_ntohs(udp->dest);
            if (dport >= rule->port_min && dport <= rule->port_max) {
                // 可选：进一步校验RTP版本号/QUIC长字段
                goto redirect_xsk;
            }
        } else if (rule->proto == IPPROTO_TCP) {
            struct tcphdr *tcp = l4;
            if ((void *)(tcp + 1) > data_end) break;
            __u16 dport = bpf_ntohs(tcp->dest);
            if (dport >= rule->port_min && dport <= rule->port_max) {
                goto redirect_xsk;
            }
        }
    }

    // 3. 未命中 -> 走内核协议栈
    stats = bpf_map_lookup_elem(&xdp_stats, &key);
    if (stats) __sync_fetch_and_add(&stats[STAT_PASS_KERNEL], 1);
    return XDP_PASS;

redirect_xsk:
    // 4. 计算队列索引（RSS哈希或简单取模）
    __u32 queue_id = ctx->rx_queue_index % 64;
    stats = bpf_map_lookup_elem(&xdp_stats, &key);
    if (stats) __sync_fetch_and_add(&stats[STAT_REDIRECT_XSK], 1);
    return bpf_redirect_map(&xsks_map, queue_id, 0);
}

char _license[] SEC("license") = "GPL";

CLANG ?= clang
LLC ?= llc
BPFTOOL ?= bpftool
ARCH := $(shell uname -m | sed 's/x86_64/x86/; s/aarch64/arm64/')
VMLINUX_H := vmlinux.h

# 生成 vmlinux.h (CO-RE依赖)
$(VMLINUX_H):
    $(BPFTOOL) btf dump file /sys/kernel/btf/vmlinux format c > $@

# 编译BPF目标文件
media_xdp.o: media_xdp.c common.h $(VMLINUX_H)
    $(CLANG) -g -O2 -target bpf -D__TARGET_ARCH_$(ARCH) 
        -I/usr/include/$(ARCH)-linux-gnu 
        -I. 
        -c media_xdp.c -o $@

# 生成用户态skeleton头文件
media_xdp.skel.h: media_xdp.o
    $(BPFTOOL) gen skeleton $< > $@

all: media_xdp.o media_xdp.skel.h

clean:
    rm -f *.o *.skel.h vmlinux.h

#include "media_xdp.skel.h"
#include <signal.h>
#include <unistd.h>
#include <xdp/libxdp.h>
#include <xdp/xsk.h>

static volatile bool running = true;
static void sig_handler(int sig) { running = false; }

int main(int argc, char **argv) {
    struct media_xdp_bpf *skel;
    struct xdp_program *prog;
    struct xsk_socket_info *xsks[64] = {0};
    int ifindex = if_nametoindex("eth0"); // 替换为实际网卡
    int queue_count = 4; // 根据RSS队列数调整

    // 1. 打开并加载骨架
    skel = media_xdp_bpf__open_and_load();
    if (!skel) { fprintf(stderr, "BPF skeleton load failedn"); return 1; }

    // 2. 配置媒体流规则 (示例: UDP 5000-6000 为RTP流)
    __u32 rule_idx = 0;
    struct media_rule rule = { .proto = IPPROTO_UDP, .port_min = 5000, .port_max = 6000, .flags = 1 };
    bpf_map_update_elem(bpf_map__fd(skel->maps.media_rules), &rule_idx, &rule, BPF_ANY);

    // 3. 创建 AF_XDP Socket 并绑定到 XSK_MAP
    for (int i = 0; i < queue_count; i++) {
        struct xsk_socket_config cfg = {
            .rx_size = 4096, .tx_size = 4096,
            .libbpf_flags = XSK_LIBBPF_FLAGS__INHIBIT_PROG_LOAD,
            .xdp_flags = XDP_FLAGS_SKB_MODE // 或 XDP_FLAGS_DRV_MODE (需驱动支持)
        };
        int err = xsk_socket__create(&xsks[i], "eth0", i, 0, &cfg);
        if (err) { perror("xsk_socket__create"); goto cleanup; }
        // 将FD填入XSK_MAP
        int xsk_fd = xsk_socket__fd(xsks[i]);
        bpf_map_update_elem(bpf_map__fd(skel->maps.xsks_map), &i, &xsk_fd, BPF_ANY);
    }

    // 4. 挂载XDP程序
    prog = xdp_program__from_bpf_object(skel->obj, "xdp_media_bypass");
    xdp_program__attach(prog, ifindex, XDP_MODE_SKB, 0); // 生产建议用 DRV/NATIVE 模式

    signal(SIGINT, sig_handler);
    signal(SIGTERM, sig_handler);

    printf("eBPF Media Bypass Running on eth0 (queues: %d)...n", queue_count);

    // 5. 事件循环 (简化版，生产需用epoll/io_uring多线程)
    while (running) {
        for (int i = 0; i < queue_count; i++) {
            // 轮询接收
            int rx = xsk_ring_cons__peek(&xsks[i]->rx, 64, &idx);
            if (rx > 0) {
                // 业务处理: 转发至媒体进程、转码节点、录制模块
                // 此处仅演示收包统计
                xsk_ring_cons__release(&xsks[i]->rx, rx);
            }
            // 发送完成回收
            xsk_ring_prod__release(&xsks[i]->tx, tx_done);
        }
        // 定期读取统计Map
        sleep(1);
    }

cleanup:
    xdp_program__detach(prog, ifindex, XDP_MODE_SKB, 0);
    media_xdp_bpf__destroy(skel);
    return 0;
}

gcc -O2 -g -Wall loader.c -o media_loader 
  -lbpf -lxdp -lelf -lz 
  -I/usr/include/bpf -I.

[Unit]
Description=eBPF Media Server Bypass Accelerator
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
ExecStart=/opt/media-bypass/media_loader
Restart=on-failure
RestartSec=5
LimitMEMLOCK=infinity
AmbientCapabilities=CAP_BPF CAP_NET_ADMIN CAP_PERFMON
CapabilityBoundingSet=CAP_BPF CAP_NET_ADMIN CAP_PERFMON
NoNewPrivileges=yes

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable --now media-xdp-bypass

# 1. 查看XDP统计
sudo bpftool map dump name xdp_stats

# 2. 网卡级XDP统计
ethtool -S eth0 | grep -i xdp

# 3. AF_XDP Socket 统计
cat /proc/net/xdp

# 4. CPU亲和性与中断均衡
# 建议将XDP处理线程、网卡RSS队列、用户态Worker绑定至同一NUMA节点物理核心
taskset -c 2-7 ./media_loader  # 示例绑定核心

# 1. 查看网卡支持队列数
ethtool -l eth0
# 输出示例: Combined: 16 (当前启用 4)

# 2. 开启最大队列数 (需驱动支持)
ethtool -L eth0 combined 16

# 3. 配置RSS哈希字段 (四元组/五元组)
# 媒体流建议包含 UDP 端口，避免同一会话因端口变化跑飞
ethtool -N eth0 rx-flow-hash udp4 sdfn  # src/dst IP + src/dst Port
ethtool -N eth0 rx-flow-hash tcp4 sdfn

# 4. 绑定中断亲和性 (将队列N绑定到CPU核心N)
# 示例：16队列绑定至物理核心 2-17 (预留 0-1 给系统/内核线程)
for i in {0..15}; do
  cpu=$((i + 2))
  echo $cpu > /proc/irq/$(grep "eth0-TxRx-$i" /proc/interrupts | awk '{print $1}' | sed 's/://')/smp_affinity_list
done

// media_xdp.c 优化：显式指定队列映射，避免内核默认哈希不均
static __always_inline __u32 select_queue(struct xdp_md *ctx, __u32 max_queues) {
    // 方案A: 使用网卡RSS哈希结果 (需驱动支持 XDP_RSS_HASH)
    // return ctx->rx_hash % max_queues; 
    
    // 方案B: 五元组哈希 (通用兼容，计算开销略大)
    // 此处建议使用 bpf_jhash 或 bpf_compute_rss_hash (内核 5.10+)
    return ctx->rx_queue_index % max_queues; // 最简单，依赖网卡RSS已生效
}

#!/bin/bash
# 绑定 Worker 线程至对应 NUMA 节点核心
# 假设 2 个 NUMA 节点，网卡 eth0 位于 Node 0 (核心 0-31)，eth1 位于 Node 1

IFACE="eth0"
QUEUES=16
NUMA_NODE=0

# 获取该 NUMA 节点空闲物理核心列表 (排除超线程兄弟核)
CORES=$(lscpu -p=CPU,CORE,SOCKET,NODE | grep -v "^#" | awk -F, -v node=$NUMA_NODE '$4==node {print $1}' | sort -u | head -n $QUEUES)

idx=0
for core in $CORES; do
  taskset -c $core ./media_loader --interface $IFACE --queue $idx --cpu $core &
  echo "Started Worker queue=$idx on CPU=$core (NUMA=$NUMA_NODE)"
  idx=$((idx+1))
done
wait

// common.h 扩展
#define MAX_SESSIONS 2000000  // 200万并发会话，按内存规划调整
#define SESSION_TTL_SEC 180   // 空闲超时 3分钟 (覆盖 ICE Keepalive 间隔)

struct session_key {
    __u32 sip, dip;      // 网络序
    __u16 sport, dport;  // 网络序
    __u8 proto;          // IPPROTO_UDP/TCP
    __u8 pad[3];
};

struct session_val {
    __u32 queue_id;      // 固定转发队列 (保证乱序控制)
    __u64 last_seen;     // 最后活跃时间 (ktime_get_ns)
    __u32 client_ip;     // 记录客户端真实IP (用于日志/限流)
    __u16 client_port;
    __u8  state;         // 0=SYN/INIT, 1=ESTABLISHED, 2=CLOSING
    __u8  flags;         // BIT0: 需要回注内核栈 (如信令包)
};

// LRU Hash Map 自动淘汰旧条目
struct {
    __uint(type, BPF_MAP_TYPE_LRU_HASH);
    __uint(max_entries, MAX_SESSIONS);
    __type(key, struct session_key);
    __type(value, struct session_val);
} session_table SEC(".maps");

// media_xdp.c 片段
static __always_inline int handle_session(struct xdp_md *ctx, struct iphdr *ip, void *l4, __u16 l4_len) {
    struct session_key key = {};
    struct session_val *val, new_val = {};
    __u64 now = bpf_ktime_get_ns();
    __u32 queue_id = select_queue(ctx, MAX_QUEUES);

    // 构造 Key (网络序存储，避免字节序转换开销)
    key.sip = ip->saddr; key.dip = ip->daddr; key.proto = ip->protocol;
    if (ip->protocol == IPPROTO_UDP) {
        struct udphdr *udp = l4;
        key.sport = udp->source; key.dport = udp->dest;
    } else if (ip->protocol == IPPROTO_TCP) {
        struct tcphdr *tcp = l4;
        key.sport = tcp->source; key.dport = tcp->dest;
    } else return XDP_PASS;

    // 1. 查找会话
    val = bpf_map_lookup_elem(&session_table, &key);
    if (val) {
        // 命中：更新时间、检查状态
        val->last_seen = now;
        if (val->flags & 0x1) return XDP_PASS; // 标记需走内核栈 (如 DTLS 握手包)
        return bpf_redirect_map(&xsks_map, val->queue_id, 0);
    }

    // 2. 未命中：新建会话 (仅处理首包/信令包特征)
    // 策略：仅允许特定端口范围或包含特定 Payload 特征的包建立会话
    if (!is_media_first_packet(l4, l4_len)) return XDP_PASS; 

    new_val.queue_id = queue_id;
    new_val.last_seen = now;
    new_val.client_ip = ip->saddr;
    new_val.client_port = key.sport;
    new_val.state = 1;
    // 反向键 (用于回包快速匹配，可选优化)
    // struct session_key rev_key = { .sip=ip->daddr, .dip=ip->saddr, ... };
    // bpf_map_update_elem(&session_table, &rev_key, &new_val, BPF_NOEXIST);

    bpf_map_update_elem(&session_table, &key, &new_val, BPF_NOEXIST);
    return bpf_redirect_map(&xsks_map, queue_id, 0);
}

// loader.c 优化：创建共享 UMEM
struct xsk_umem_config umem_cfg = {
    .fill_size = 65536,   // Fill Ring 大小
    .comp_size = 65536,   // Completion Ring 大小
    .frame_size = XSK_UMEM__DEFAULT_FRAME_SIZE, // 2048B
    .frame_headroom = XDP_PACKET_HEADROOM,      // 256B 预留头部空间
    .flags = XDP_UMEM_UNALIGNED_CHUNK_FLAG,     // 支持非对齐访问
};

struct xsk_umem *umem;
xsk_umem__create(&umem, &umem_cfg, NULL, NULL); // 使用 malloc 分配内存 (或 hugetlbfs)

// 为每个队列创建 Socket 并绑定同一 UMEM
for (int i = 0; i < queue_count; i++) {
    struct xsk_socket_config sock_cfg = {
        .rx_size = 4096, .tx_size = 4096,
        .libbpf_flags = XSK_LIBBPF_FLAGS__INHIBIT_PROG_LOAD,
        .xdp_flags = XDP_FLAGS_DRV_MODE | XDP_USE_NEED_WAKEUP, // 关键：驱动模式 + 唤醒优化
        .bind_flags = XDP_USE_NEED_WAKEUP, // Tx 侧也启用
    };
    xsk_socket__create_shared(&xsks[i], iface, i, umem, &sock_cfg);
}

// worker_thread.c 核心循环
void *worker_loop(void *arg) {
    struct worker_ctx *ctx = arg;
    struct xsk_socket *xsk = ctx->xsk;
    struct xsk_ring_cons *rx = &xsk->rx;
    struct xsk_ring_prod *tx = &xsk->tx;
    struct xsk_ring_prod *fill = &xsk->umem->fill;
    struct xsk_ring_cons *comp = &xsk->umem->comp;

    const int BATCH = 64;
    __u32 idx_rx, idx_tx, idx_fill, idx_comp;
    struct xdp_desc rx_descs[BATCH], tx_descs[BATCH];

    while (running) {
        // --- 1. 批量接收 (RX) ---
        int nb_rx = xsk_ring_cons__peek(rx, BATCH, &idx_rx);
        if (nb_rx > 0) {
            // 预取描述符地址
            for (int i = 0; i < nb_rx; i++) {
                rx_descs[i] = *xsk_ring_cons__rx_desc(rx, idx_rx + i);
            }
            
            // 业务处理: 解析 RTP Header -> 转发至下游 Worker (Ring Buffer / Shared Memory)
            // 关键：此处不拷贝数据，仅传递 (addr, len, metadata) 指针
            process_media_batch(ctx, rx_descs, nb_rx); 

            // 释放 RX Ring
            xsk_ring_cons__release(rx, nb_rx);
            
            // 同步归还缓冲区到 Fill Ring (零拷贝回收)
            // 策略：业务处理完毕后统一归还，或引用计数归还
            recycle_buffers_to_fill(ctx, rx_descs, nb_rx, fill); 
        }

        // --- 2. 批量发送 (TX) ---
        // 从业务层取出待发送包元数据 (已填充好头部的帧地址)
        int nb_tx = dequeue_tx_batch(ctx, tx_descs, BATCH);
        if (nb_tx > 0) {
            int idx = xsk_ring_prod__reserve(tx, nb_tx, &idx_tx);
            if (idx == nb_tx) {
                for (int i = 0; i < nb_tx; i++) {
                    *xsk_ring_prod__tx_desc(tx, idx_tx + i) = tx_descs[i];
                }
                xsk_ring_prod__submit(tx, nb_tx);
                
                // 触发驱动发送 (仅当 Need Wakeup 时)
                if (xsk_ring_prod__needs_wakeup(tx)) {
                    sendto(xsk_socket__fd(xsk), NULL, 0, MSG_DONTWAIT, NULL, 0);
                }
            } else {
                // TX Ring 满：回压处理 / 丢包统计 / 重试
                handle_tx_backpressure(ctx, tx_descs, nb_tx);
            }
        }

        // --- 3. 处理 Completion Ring (TX 完成回收) ---
        int nb_comp = xsk_ring_cons__peek(comp, BATCH, &idx_comp);
        if (nb_comp > 0) {
            // 帧已发送完成，归还 UMEM 供 Fill Ring 复用
            for (int i = 0; i < nb_comp; i++) {
                __u64 addr = *xsk_ring_cons__comp_addr(comp, idx_comp + i);
                return_frame_to_umem_pool(ctx->pool, addr);
            }
            xsk_ring_cons__release(comp, nb_comp);
        }

        // 空闲时短暂休眠或 epoll 等待
        if (nb_rx == 0 && nb_tx == 0) usleep(100); // 或 io_uring/epoll 事件驱动
    }
}

# 旁路节点上配置：标记旁路处理过的包，回包强制走本机路由表
# 1. 标记入站旁路包 (在 XDP 程序中 bpf_redirect_map 后设置 mark，或 tc ingress 打标)
# 2. 配置策略路由
ip rule add fwmark 0x100 lookup 100 priority 100
ip route add local 0.0.0.0/0 dev lo table 100  # 本地回环处理 (配合 TPROXY)
# 或: ip route add <Backend_Subnet> via <Gateway> dev eth1 table 100

# 定义 RTP 流模板
class RTPTemplate:
    def __init__(self):
        self.base_pkt = Ether()/IP()/UDP()/Raw(load="RTP_HEADER_12BYTES" + "PAYLOAD"*100)
        # 变量: SSRC, SeqNum, Timestamp, Payload Type
    
    def create_stream(self, pps, duration):
        return STLStream(
            packet=STLPktBuilder(pkt=self.base_pkt, vm=STLScVmRaw([
                STLVmFlowVar(name="seq", min_value=0, max_value=65535, size=2, op="inc"),
                STLVmFlowVar(name="ts", min_value=0, max_value=0xFFFFFFFF, size=4, op="inc", step=160), # 20ms @ 8kHz
                STLVmWrFlowVar(fv_name="seq", pkt_offset="UDP:Raw+2"),
                STLVmWrFlowVar(fv_name="ts", pkt_offset="UDP:Raw+4"),
                STLVmFixIpv4(offset="IP") # 重算校验和
            ])),
            mode=STLTXCont(pps=pps),
            next=STLFlowLatencyStats(pg_id=10)
        )

# 1. 模拟网卡驱动 RX 队列停摆 (触发 XDP 重载/旁路降级)
# 使用 tc netem 模拟驱动级丢包/延迟
tc qdisc add dev eth0 root netem loss 10% delay 5ms

# 2. 模拟内核内存不足 (slab 耗尽)
# 限制 kmem 缓存
echo 1000000 > /sys/kernel/slab/kmalloc-2048/limit_in_bytes

# 3. 模拟 CPU 热插拔 / 核心下线 (测试 RSS 重新平衡)
echo 0 > /sys/devices/system/cpu/cpu8/online

# 1. 注入规则错误：将媒体端口范围改错，验证流量是否正确回落内核栈
bpftool map update name media_rules key 0 0 0 0 value 17 0 0 0 0 0 0 0  # 端口范围设为 0

# 2. 注入 Map 溢出：快速填满 session_table，验证 LRU 淘汰与新建会话逻辑
# 使用 pytest-bpf 或自写脚本并发发送不同五元组包

# 3. 验证优雅降级：卸载 XDP 程序 (ip link set dev eth0 xdp off)，确认业务无感切回内核栈
# 监控指标：内核协议栈 CPU 占用、Socket 队列积压、应用层错误率

单流成本 (元/月) = 
  (CPU核心单价 * 占用核心数) + 
  (内存单价 * 占用内存GB) + 
  (网卡分摊单价 * 占用带宽Gbps) + 
  (运维人力分摊)