┌─────────────────────────────────────────────────────────────┐
│                      用户态控制面                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │ 配置下发    │  │ 统计采集    │  │ 策略动态调整 │          │
│  │ (libbpf)    │  │ (perf/ring) │  │ (Map更新)    │          │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘          │
└─────────┼────────────────┼────────────────┼──────────────────┘
          │                │                │
          ▼                ▼                ▼
┌─────────────────────────────────────────────────────────────┐
│                      内核态数据面 (eBPF)                     │
│  ┌─────────────────────────────────────────────────────┐    │
│  │ XDP Layer (入向)                                     │    │
│  │  • 早期识别 RTP/RTCP 流                              │    │
│  │  • 丢包检测与序列号重排序缓存                        │    │
│  │  • 选择性确认 (SACK) 生成                            │    │
│  └─────────────────────────────────────────────────────┘    │
│  ┌─────────────────────────────────────────────────────┐    │
│  │ TC Egress / Socket Ops (出向)                        │    │
│  │  • 快速重传队列管理                                  │    │
│  │  • 拥塞感知的重传节奏控制                            │    │
│  │  • FEC 冗余包注入 (可选)                             │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘

// flow_context.h
#pragma once
#include <linux/types.h>
#include <bpf/bpf_helpers.h>

#define MAX_FLOWS 65536
#define RTP_WINDOW_SIZE 256  // 滑动窗口大小，覆盖 256 个序列号

struct rtp_flow_key {
    __u32 saddr;      // 源 IP
    __u32 daddr;      // 目的 IP
    __u16 sport;      // 源端口
    __u16 dport;      // 目的端口
    __u32 ssrc;       // RTP SSRC
} __attribute__((packed));

struct rtp_flow_ctx {
    __u32 base_seq;           // 窗口基准序列号
    __u32 max_seq;            // 当前最大序列号
    __u64 bitmap;             // 256-bit 位图，标记已收包 (高 64 位扩展可用数组)
    __u64 bitmap_high;        // 扩展位图高 64 位
    __u32 last_nack_seq;      // 最近发送 NACK 的序列号
    __u32 nack_count;         // NACK 发送计数
    __u64 last_ts;            // 最后收包时间戳 (ns)
    __u32 rtt_est;            // 估算 RTT (us)
    __u32 min_rtt;            // 最小 RTT (us)
    __u8  fec_enabled;        // 是否启用 FEC
    __u8  reserved[3];
    __u64 rx_pkts;            // 接收包计数
    __u64 rx_bytes;           // 接收字节计数
    __u64 lost_pkts;          // 累计丢包计数
    __u64 retrans_pkts;       // 重传包计数
} __attribute__((packed));

// BPF Map 定义
struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_HASH);
    __uint(max_entries, MAX_FLOWS);
    __type(key, struct rtp_flow_key);
    __type(value, struct rtp_flow_ctx);
    __uint(map_flags, BPF_F_NO_PREALLOC);
} flow_ctx_map SEC(".maps");

// retrans_queue.h
#define MAX_RETRANS_ENTRIES 32768

struct retrans_key {
    __u32 ssrc;
    __u16 seq;        // 待重传序列号
    __u16 pad;
} __attribute__((packed));

struct retrans_val {
    __u32 flow_hash;      // 流哈希，用于快速定位 flow_ctx
    __u64 enqueue_ts;     // 入队时间
    __u16 payload_len;    // 载荷长度
    __u8  priority;       // 优先级：关键帧 > 普通帧 > FEC
    __u8  attempts;       // 重传次数
    __u8  data[];         // 柔性数组存载荷 (最大 MTU)
} __attribute__((packed));

struct {
    __uint(type, BPF_MAP_TYPE_LRU_HASH);
    __uint(max_entries, MAX_RETRANS_ENTRIES);
    __type(key, struct retrans_key);
    __type(value, struct retrans_val);
    __uint(map_flags, BPF_F_NO_PREALLOC);
} retrans_queue SEC(".maps");

// xdp_rtp_parser.c
#include "flow_context.h"
#include <linux/ip.h>
#include <linux/udp.h>
#include <bpf/bpf_endian.h>

// RTP 固定头部 (12 字节)
struct rtp_hdr {
    __u8  version_p_x_cc;   // V=2, P, X, CC
    __u8  m_pt;             // M, PT
    __u16 seq;              // 序列号
    __u32 ts;               // 时间戳
    __u32 ssrc;             // 同步源标识符
} __attribute__((packed));

static __always_inline int parse_rtp_header(void *data, void *data_end, struct rtp_hdr *rtp) {
    if (data + sizeof(*rtp) > data_end) return -1;
    __builtin_memcpy(rtp, data, sizeof(*rtp));
    rtp->seq = bpf_ntohs(rtp->seq);
    rtp->ts  = bpf_ntohl(rtp->ts);
    rtp->ssrc = bpf_ntohl(rtp->ssrc);
    // 校验版本号 = 2
    if ((rtp->version_p_x_cc >> 6) != 2) return -1;
    return 0;
}

static __always_inline void update_loss_bitmap(struct rtp_flow_ctx *ctx, __u16 seq) {
    __u32 delta = seq - ctx->base_seq;
    if (delta >= RTP_WINDOW_SIZE) {
        // 序列号超出窗口，滑动窗口
        __u32 shift = delta - RTP_WINDOW_SIZE + 1;
        ctx->bitmap >>= shift;
        ctx->bitmap_high >>= shift;
        // 高位补低位溢出
        ctx->bitmap_high |= (ctx->bitmap << (64 - shift));
        ctx->base_seq = seq - RTP_WINDOW_SIZE + 1;
        delta = RTP_WINDOW_SIZE - 1;
    }
    // 标记已收到
    if (delta < 64) {
        ctx->bitmap |= (1ULL << delta);
    } else {
        ctx->bitmap_high |= (1ULL << (delta - 64));
    }
    if (seq > ctx->max_seq) ctx->max_seq = seq;
}

static __always_inline __u32 detect_gaps(struct rtp_flow_ctx *ctx, __u16 *gap_seqs, __u32 max_gaps) {
    __u32 gaps = 0;
    __u32 expected = ctx->base_seq;
    for (int i = 0; i < RTP_WINDOW_SIZE && gaps < max_gaps; i++) {
        __u64 bit = (i < 64) ? (ctx->bitmap >> i) & 1 : (ctx->bitmap_high >> (i - 64)) & 1;
        if (!bit) {
            gap_seqs[gaps++] = expected;
        }
        expected++;
    }
    return gaps;
}

SEC("xdp")
int xdp_rtp_loss_detect(struct xdp_md *ctx) {
    void *data = (void *)(long)ctx->data;
    void *data_end = (void *)(long)ctx->data_end;

    // 解析以太网 + IP + UDP
    struct ethhdr *eth = data;
    if ((void *)eth + sizeof(*eth) > data_end) return XDP_PASS;
    if (eth->h_proto != bpf_htons(ETH_P_IP)) return XDP_PASS;

    struct iphdr *ip = data + sizeof(*eth);
    if ((void *)ip + sizeof(*ip) > data_end) return XDP_PASS;
    if (ip->protocol != IPPROTO_UDP) return XDP_PASS;

    struct udphdr *udp = (void *)ip + (ip->ihl * 4);
    if ((void *)udp + sizeof(*udp) > data_end) return XDP_PASS;

    // 仅处理目标端口为媒体平面配置端口范围的包 (示例: 10000-20000)
    __u16 dport = bpf_ntohs(udp->dest);
    if (dport < 10000 || dport > 20000) return XDP_PASS;

    struct rtp_hdr rtp;
    if (parse_rtp_header((void *)udp + sizeof(*udp), data_end, &rtp) < 0) return XDP_PASS;

    // 构造流键
    struct rtp_flow_key key = {
        .saddr = ip->saddr,
        .daddr = ip->daddr,
        .sport = udp->source,
        .dport = udp->dest,
        .ssrc  = rtp.ssrc,
    };

    // 查找/创建流上下文
    struct rtp_flow_ctx *flow = bpf_map_lookup_elem(&flow_ctx_map, &key);
    if (!flow) {
        struct rtp_flow_ctx new_ctx = {0};
        new_ctx.base_seq = rtp.seq;
        new_ctx.max_seq  = rtp.seq;
        new_ctx.last_ts  = bpf_ktime_get_ns();
        new_ctx.min_rtt  = 0xFFFFFFFF;
        bpf_map_update_elem(&flow_ctx_map, &key, &new_ctx, BPF_ANY);
        flow = bpf_map_lookup_elem(&flow_ctx_map, &key);
        if (!flow) return XDP_PASS;
    }

    // 更新统计
    flow->rx_pkts++;
    flow->rx_bytes += bpf_ntohs(udp->len) - sizeof(*udp);
    flow->last_ts = bpf_ktime_get_ns();

    // 更新位图并检测丢包
    update_loss_bitmap(flow, rtp.seq);

    // 丢包检测：每 32 个包或收到关键帧 (PT=96/97/98 且 M=1) 触发一次
    bool trigger_nack = ((flow->rx_pkts & 31) == 0) || (rtp.m_pt & 0x80);
    if (trigger_nack) {
        __u16 gap_seqs[16];
        __u32 gaps = detect_gaps(flow, gap_seqs, 16);
        if (gaps > 0) {
            flow->lost_pkts += gaps;
            // 这里可生成 NACK 回包 (需配合 TC Egress 或用户态发送)
            // 简化示例：仅记录最后 NACK 序列号
            flow->last_nack_seq = gap_seqs[0];
            flow->nack_count++;
        }
    }

    return XDP_PASS;
}

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

// tc_retrans_scheduler.c
#include "flow_context.h"
#include "retrans_queue.h"
#include <linux/pkt_cls.h>
#include <bpf/bpf_helpers.h>

#define MAX_RETRANS_PER_BATCH 8
#define RETRANS_TIMEOUT_US 5000  // 5ms 重传超时
#define MAX_RETRANS_ATTEMPTS 3

SEC("classifier")
int tc_retrans_egress(struct __sk_buff *skb) {
    // 仅处理出向媒体流量 (可按标记/端口/钩子区分)
    // 这里演示：周期性扫描重传队列，构造重传包
    // 实际生产建议配合 bpf_timer (内核 5.17+) 或用户态定时器触发

    __u64 now = bpf_ktime_get_ns() / 1000; // us
    __u32 processed = 0;

    // 遍历重传队列 (LRU Hash 无序遍历，生产建议配合 Ring Buffer 或用户态调度)
    // 此处为演示逻辑，实际需配合 bpf_map_iter 或用户态协同
    void *cursor = NULL;
    struct retrans_key *key;
    struct retrans_val *val;

    while (processed < MAX_RETRANS_PER_BATCH &&
           (key = bpf_map_get_next_key(&retrans_queue, cursor, &key)) == 0) {
        val = bpf_map_lookup_elem(&retrans_queue, key);
        if (!val) { cursor = key; continue; }

        // 超时与重传次数检查
        if (now - val->enqueue_ts < RETRANS_TIMEOUT_US) { cursor = key; continue; }
        if (val->attempts >= MAX_RETRANS_ATTEMPTS) {
            bpf_map_delete_elem(&retrans_queue, key);
            cursor = key; continue;
        }

        // TODO: 此处需构造重传 skb 并重新注入协议栈
        // 受限于 TC 无法直接构造新包，生产方案通常：
        // 1. 用户态从 Ring Buffer 读取重传任务，经 AF_XDP/XSK 回注
        // 2. 或使用 Socket Ops 标记 SO_TXTIME 配合 TCP/UDP 协议栈
        // 3. 内核 6.2+ 支持 bpf_redirect_neigh / bpf_clone_redirect

        val->attempts++;
        val->enqueue_ts = now;
        processed++;
        cursor = key;
    }

    return TC_ACT_OK;
}

// sock_ops_cc.c
#include <linux/bpf.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_tracing.h>

struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 65536);
    __type(key, __u32);   // socket cookie
    __type(value, __u32); // min_rtt_us
} sock_rtt_map SEC(".maps");

SEC("sockops")
int sock_ops_rtt_track(struct bpf_sock_ops *skops) {
    __u32 op = skops->op;
    __u32 cookie = bpf_get_socket_cookie(skops);

    switch (op) {
    case BPF_SOCK_OPS_RTT_CB:
        // 内核主动回调 RTT 样本
        {
            __u32 rtt_us = skops->args[0]; // RTT 单位 us
            __u32 *min_rtt = bpf_map_lookup_elem(&sock_rtt_map, &cookie);
            if (!min_rtt || rtt_us < *min_rtt) {
                bpf_map_update_elem(&sock_rtt_map, &cookie, &rtt_us, BPF_ANY);
            }
        }
        break;
    case BPF_SOCK_OPS_TCP_CONNECT_CB:
    case BPF_SOCK_OPS_UDP_CONNECT_CB:
        // 连接建立时初始化
        {
            __u32 init_rtt = 20000; // 20ms 默认值
            bpf_map_update_elem(&sock_rtt_map, &cookie, &init_rtt, BPF_NOEXIST);
        }
        break;
    }
    return 0;
}

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

# 依赖安装 (Ubuntu 22.04 示例)
sudo apt update && sudo apt install -y clang llvm libbpf-dev linux-headers-$(uname -r) 
    bpfcc-tools linux-tools-$(uname -r) pkg-config

# 编译 (以 xdp_rtp_loss_detect.c 为例)
clang -O2 -target bpf -D__TARGET_ARCH_x86 
    -I/usr/include/x86_64-linux-gnu 
    -I/usr/include/linux 
    -c xdp_rtp_loss_detect.c -o xdp_rtp_loss_detect.o

# 加载到网卡 (假设网卡为 eth0)
sudo ip link set dev eth0 xdp obj xdp_rtp_loss_detect.o sec xdp

# 挂载 TC Egress
sudo tc qdisc add dev eth0 clsact
sudo tc filter add dev eth0 egress bpf da obj tc_retrans_scheduler.o sec classifier

# 加载 Socket Ops (需 cgroup v2)
sudo mount -t cgroup2 none /sys/fs/cgroup
sudo bpftool prog load sock_ops_rtt_track.o /sys/fs/bpf/sock_ops_rtt_track 
    type sock_ops
sudo bpftool cgroup attach /sys/fs/cgroup/ sock_ops pinned /sys/fs/bpf/sock_ops_rtt_track

# stats_collector.py
from bcc import BPF
import time
import json

bpf = BPF(src_file="xdp_rtp_loss_detect.c")
flow_map = bpf["flow_ctx_map"]

def collect_stats():
    stats = {}
    for key, val in flow_map.items():
        k = key
        v = val
        # PERCPU map 需聚合所有 CPU
        total_pkts = sum(v.rx_pkts)
        total_lost = sum(v.lost_pkts)
        loss_rate = total_lost / total_pkts if total_pkts else 0
        stats[f"{k.saddr}:{k.sport}->{k.daddr}:{k.dport}#{k.ssrc}"] = {
            "rx_pkts": total_pkts,
            "lost_pkts": total_lost,
            "loss_rate": round(loss_rate, 4),
            "rtt_est": v.rtt_est,
            "nack_count": v.nack_count,
        }
    return stats

if __name__ == "__main__":
    while True:
        data = collect_stats()
        print(json.dumps(data, indent=2))
        time.sleep(5)

# config_manager.py
from bcc import BPF

bpf = BPF(src_file="xdp_rtp_loss_detect.c")
config_map = bpf["config_map"]  # 假设定义了配置 Map

def update_fec_policy(ssrc, enable):
    key = ssrc
    val = 1 if enable else 0
    config_map[key] = val

def update_nack_interval(interval_ms):
    # 通过 Map 更新 NACK 触发周期
    pass

# 使用 pktgen 或 MoonGen 生成模拟 RTP 流量
# 示例: MoonGen 脚本发送 1000 并发 RTP 流，每流 50pps，模拟 2% 随机丢包
# 观测内核态统计与应用层 QoE 指标对比

# 开启内核态性能分析
sudo bpftool prog profile name xdp_rtp_loss_detect duration 10

// timer_retrans.c
#include "flow_context.h"
#include "retrans_queue.h"
#include <linux/bpf.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_timers.h>

// 定时器 Map：Key 为流哈希，Value 为定时器实例
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, MAX_FLOWS);
    __type(key, __u32);           // flow_hash
    __type(value, struct bpf_timer);
    __uint(map_flags, BPF_F_NO_PREALLOC);
} retrans_timers SEC(".maps");

// 重传回调函数
static void retrans_timer_cb(struct bpf_timer *timer, void *ctx) {
    __u32 flow_hash = *(__u32 *)ctx;
    
    // 1. 查找流上下文
    struct rtp_flow_ctx *flow = bpf_map_lookup_elem(&flow_ctx_map, &flow_hash);
    if (!flow) return;

    // 2. 扫描该流的重传队列 (简化：实际需二级索引 Map: flow_hash -> retrans_keys[])
    // 此处演示单包重传逻辑，生产建议批量处理
    struct retrans_key rkey = { .ssrc = flow_hash & 0xFFFFFFFF, .seq = flow->last_nack_seq };
    struct retrans_val *rval = bpf_map_lookup_elem(&retrans_queue, &rkey);
    if (!rval) return;

    // 3. 拥塞感知：读取 Socket Ops 反馈的 min_rtt
    __u32 *min_rtt = bpf_map_lookup_elem(&sock_rtt_map, &flow_hash);
    __u32 rto = min_rtt ? (*min_rtt * 4) : 20000; // 简化 RTO 计算: 4x min_rtt
    
    // 4. 重传次数与超时保护
    if (rval->attempts >= MAX_RETRANS_ATTEMPTS) {
        bpf_map_delete_elem(&retrans_queue, &rkey);
        return;
    }

    // 5. 关键路径：构造重传包并回注
    // 注：内核态构包受限，需借助 bpf_clone_redirect / bpf_redirect_neigh (需内核 6.2+)
    // 或配合 XDP_TX (驱动级回发) 实现真正内核态闭环
    // 示例：标记 skb 并重定向回同网卡 TX 队列 (需驱动支持 XDP_TX)
    // struct xdp_md *xdp = ...; // 需从上下文获取或预分配
    // bpf_xdp_adjust_head(xdp, -eth_hdr_len); // 还原 L2 头
    // bpf_redirect_map(&tx_port_map, 0, BPF_F_BROADCAST); // 广播回发或指定队列

    // 6. 更新状态并重置定时器
    rval->attempts++;
    bpf_timer_set_callback(timer, retrans_timer_cb);
    bpf_timer_start(timer, rto * 1000); // 单位 ns
}

// XDP 入向：首次检测到丢包时初始化/重置定时器
SEC("xdp")
int xdp_loss_trigger_timer(struct xdp_md *ctx) {
    // ... 解析 RTP、查找 flow_ctx (同基础篇) ...
    struct rtp_flow_ctx *flow = bpf_map_lookup_elem(&flow_ctx_map, &key);
    if (!flow) return XDP_PASS;

    // 检测到新丢包 (gap_seqs[0] != flow->last_nack_seq)
    if (new_gap_detected) {
        __u32 flow_hash = bpf_get_hash_recalc((void *)&key, sizeof(key));
        
        // 获取或创建定时器
        struct bpf_timer *timer = bpf_map_lookup_elem(&retrans_timers, &flow_hash);
        if (!timer) {
            struct bpf_timer new_timer = {};
            bpf_timer_init(&new_timer, &retrans_timers, CLOCK_MONOTONIC);
            bpf_map_update_elem(&retrans_timers, &flow_hash, &new_timer, BPF_NOEXIST);
            timer = bpf_map_lookup_elem(&retrans_timers, &flow_hash);
        }
        if (timer) {
            // 启动定时器，回调携带 flow_hash 上下文
            bpf_timer_set_callback(timer, retrans_timer_cb);
            bpf_timer_start(timer, 5000 * 1000); // 5ms 后首次重传
        }
    }
    return XDP_PASS;
}

// rtp_ext_parse.h
// 支持 RFC 8285 (One-Byte/Two-Byte Header Extensions) 解析
// 典型扩展: MID (Stream ID), RID (RTP Stream Identifier), LID (Layer ID)

static __always_inline int parse_rtp_extensions(void *data, void *data_end, 
                                                __u8 *layer_id, __u8 *priority) {
    struct rtp_hdr *rtp = data;
    if ((void *)(rtp + 1) > data_end) return -1;
    
    // 跳过 CSRC 列表
    void *ptr = (void *)(rtp + 1) + rtp->cc * 4;
    if (ptr > data_end) return -1;

    // 检查扩展位 (X bit)
    if (!(rtp->version_p_x_cc & 0x10)) return 0; // 无扩展头

    // 解析 One-Byte Header Extension (RFC 5285)
    __u16 *ext_hdr = ptr;
    if ((void *)(ext_hdr + 1) > data_end) return -1;
    __u16 profile = bpf_ntohs(*ext_hdr);
    if (profile != 0xBEDE) return 0; // 仅处理 One/Two Byte 扩展

    ptr = ext_hdr + 1;
    __u16 ext_len = bpf_ntohs(*(__u16 *)ptr); // 扩展长度(4字节为单位)
    ptr += 2;
    void *ext_end = ptr + ext_len * 4;
    if (ext_end > data_end) ext_end = data_end;

    // 遍历扩展项
    while (ptr + 1 <= ext_end) {
        __u8 id_len = *(__u8 *)ptr;
        __u8 ext_id = id_len >> 4;
        __u8 len = (id_len & 0xF) + 1; // 长度 = 值 + 1 字节
        ptr += 1;
        if (ptr + len > ext_end) break;

        // 常见扩展 ID 映射 (需与信令协商一致)
        // MID=1, RID=2, LID=3, PRIORITY=4 (示例值)
        switch (ext_id) {
            case 3: // LID (Layer ID)
                if (len >= 1) *layer_id = *(__u8 *)ptr;
                break;
            case 4: // PRIORITY
                if (len >= 1) *priority = *(__u8 *)ptr;
                break;
        }
        ptr += len;
    }
    return 0;
}

// policy_config.h
struct layer_policy {
    __u8  fec_redundancy;   // FEC 冗余度: BL=50%, EL=10%
    __u8  max_retrans;      // 最大重传次数: BL=5, EL=1
    __u16 rto_factor;       // RTO 系数: BL=3x, EL=5x (更激进)
    __u8  nack_priority;    // NACK 优先级: BL=7 (最高), EL=1
    __u8  drop_threshold;   // 拥塞时主动丢弃阈值: BL=0 (不丢), EL=50%
};

struct {
    __uint(type, BPF_MAP_TYPE_ARRAY);
    __uint(max_entries, 8); // 支持 8 个层级 (LID 0-7)
    __type(key, __u32);     // layer_id
    __type(value, struct layer_policy);
} layer_policy_map SEC(".maps");

// 默认策略初始化 (用户态下发)
/*
LID 0 (Base):     fec=50, retrans=5, rto=3, prio=7, drop=0
LID 1 (Enhance1): fec=20, retrans=3, rto=4, prio=4, drop=30
LID 2 (Enhance2): fec=10, retrans=1, rto=5, prio=1, drop=50
*/

// 在 retrans_timer_cb 或 tc_retrans_egress 中应用
static __always_inline void apply_layer_policy(struct retrans_val *rval, __u8 layer_id) {
    struct layer_policy *pol = bpf_map_lookup_elem(&layer_policy_map, &layer_id);
    if (!pol) return; // 无策略则用默认

    // 1. 重传次数上限
    if (rval->attempts >= pol->max_retrans) {
        rval->drop_flag = 1; // 标记放弃
        return;
    }

    // 2. 动态调整定时器 RTO
    // 需配合 bpf_timer_set_callback 重新计算超时
    // rto_us = base_rto * pol->rto_factor;

    // 3. 优先级标记 (用于 TC/Qdisc 优先队列或 XDP_TX 元数据)
    rval->priority = pol->nack_priority;

    // 4. FEC 触发逻辑 (在 XDP 入向或单独 FEC 程序中)
    // if (random() % 100 < pol->fec_redundancy) inject_fec_packet(...);
}

# 1. 开启 JIT 编译与 kallsyms 暴露
echo 1 > /proc/sys/net/core/bpf_jit_enable
echo 1 > /proc/sys/kernel/kptr_restrict

# 2. 使用 bpftool 生成程序火焰图 (需内核 5.15+)
sudo bpftool prog profile name xdp_rtp_loss_detect duration 30 > /tmp/profile.data

# 3. 转换为 pprof 格式分析
go tool pprof -http=:8080 /tmp/profile.data
# 重点关注：map_lookup_elem 耗时、位图操作分支预测失败率、verifier 生成的边界检查指令

// trace_correlation.c
// 定义 Trace Context 结构 (放入 skb->cb 或 metadata)
struct trace_ctx {
    __u64 trace_id;      // 全局唯一 TraceID (生成于首包 XDP)
    __u64 span_id;       // 当前 SpanID
    __u32 flow_hash;     // 流标识
    __u8  hop;           // 经过节点: 1=XDP_RX, 2=TC_EGRESS, 3=SOCK_OPS, 4=USERSPACE
    __u8  flags;         // 采样标记
    __u16 pad;
};

// Per-CPU Array 存储当前包的 Trace Context (避免 Map 查找开销)
struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
    __uint(max_entries, 1);
    __type(key, __u32);
    __type(value, struct trace_ctx);
} trace_ctx_scratch SEC(".maps");

// XDP 入向：生成/提取 TraceID
SEC("xdp")
int xdp_trace_ingress(struct xdp_md *ctx) {
    __u32 key = 0;
    struct trace_ctx *tctx = bpf_map_lookup_elem(&trace_ctx_scratch, &key);
    if (!tctx) return XDP_PASS;

    // 尝试从 RTP 扩展头读取 TraceID (业务层协商注入)
    // 如果无，则基于 (flow_hash + timestamp) 生成
    __u64 trace_id = extract_trace_id_from_rtp(ctx); // 实现略
    if (!trace_id) trace_id = bpf_ktime_get_ns() ^ flow_hash;

    tctx->trace_id = trace_id;
    tctx->span_id = bpf_get_prandom_u64();
    tctx->flow_hash = flow_hash;
    tctx->hop = 1;
    tctx->flags = (trace_id & 0xFF) < 100 ? 1 : 0; // 1% 全量采样

    // 存入 skb->cb 传递给后续 TC/SockOps (需驱动支持 XDP_METADATA)
    // bpf_xdp_set_metadata(ctx, tctx, sizeof(*tctx)); 
    return XDP_PASS;
}

// TC Egress：续写 Span
SEC("classifier")
int tc_trace_egress(struct __sk_buff *skb) {
    struct trace_ctx *tctx = (void *)skb->cb; // 从 cb 读取
    if (!tctx || tctx->hop != 1) return TC_ACT_OK;

    tctx->hop = 2;
    tctx->span_id = bpf_get_prandom_u64(); // 新 Span
    // 导出到 Ring Buffer (用户态 Jaeger/Zipkin Collector 消费)
    bpf_ringbuf_output(&trace_ringbuf, tctx, sizeof(*tctx), 0);
    return TC_ACT_OK;
}

┌────────────────────────────────────────────────────────────────┐
│                     Host CPU (x86/ARM)                         │
│  ┌─────────────┐  ┌─────────────┐  ┌────────────────────────┐  │
│  │ Control Plane│  │ Slow Path   │  │ eBPF Maps (Shared Mem) │  │
│  │ (Policy/Stats)│  │ (New Flow)  │  │ (Flow Ctx / ReTx Queue)│  │
│  └──────┬──────┘  └──────┬──────┘  └───────────┬────────────┘  │
└─────────┼────────────────┼──────────────────────┼───────────────┘
          │ PCIe / CXL     │                      │
          ▼                ▼                      ▼
┌────────────────────────────────────────────────────────────────┐
│                    SmartNIC / DPU (NVIDIA BlueField,            │
│                    Intel IPU, Pensando, Netronome)             │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │  Fast Path (P4 / eBPF Offload / Microcode)               │  │
│  │  • XDP Offload: RTP Parse, Loss Detect, Bitmap Update    │  │
│  │  • Stateful Flow Table (SRAM/TCAM): 1M+ Flows @ Line Rate│  │
│  │  • Hardware Timer Wheel: μs-level Retrans Scheduling     │  │
│  │  • Inline FEC Encode/Decode (Reed-Solomon Engine)        │  │
│  │  • Packet Replication (Multicast/Simulcast)              │  │
│  └──────────────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────────────┘

// common_pragma.h
#if defined(__TARGET_NFP__) || defined(__TARGET_BPF_DPA__)
#define OFFLOAD_MODE 1
#define NO_BPF_TIMER 1
#define NO_RINGBUF 1
// 硬件原子操作映射
#define ATOMIC_INC(ptr) __sync_fetch_and_add(ptr, 1)
#else
#define OFFLOAD_MODE 0
#define ATOMIC_INC(ptr) (*(ptr)++) // Host 单线程 Per-CPU 无需原子
#endif

// flow_context.c (统一代码库)
static __always_inline void update_flow_stats(struct rtp_flow_ctx *ctx, __u32 bytes) {
    ctx->rx_pkts++;
    ctx->rx_bytes += bytes;
    
#if OFFLOAD_MODE
    // 卸载模式：Map 可能在远程内存，需显式写回或使用硬件原子
    // 部分硬件要求显式 flush
    bpf_map_update_elem(&flow_ctx_map, &key, ctx, BPF_EXIST); 
#else
    // Host 模式：PerCPU Map 直接修改即可
#endif
}

// control_sync.c (Host 侧用户态代理)
void sync_flow_state_to_nic() {
    // 1. 批量读取 Host 侧流表变更 (脏页标记机制)
    // 2. 通过 DOCA/IPDK API 批量下发到 NIC 片上存储
    // 3. 关键一致性点：
    //    - 新建流：Host 分配 Flow ID -> 下发 NIC 创建流表项
    //    - 删除流：Host 标记 Tombstone -> NIC 延迟回收 (防止包乱序命中)
    //    - 策略变更：版本号机制 (Policy Version), NIC 原子切换
}

// NIC 侧 (P4/eBPF) 处理未知流包
action handle_miss() {
    // 1. 解析五元组 + SSRC
    // 2. 生成 "New Flow Notification" 发送给 Host CPU (专用 Queue)
    // 3. 临时按默认策略转发 (或 Drop 等待 Host 下发)
    // 4. Host 收到通知 -> 分配资源 -> 下发流表项 -> NIC 安装 -> 恢复正常处理
}

# bpf-manifest.yaml (GitOps 管理)
version: "v1.2.3"
programs:
  - name: xdp_rtp_loss_detect
    type: xdp
    attach: eth0
    bytecode: artifacts/xdp_rtp_loss_detect_v1.2.3.o
    maps:
      - flow_ctx_map (pinned: /sys/fs/bpf/flow_ctx_map)
      - layer_policy_map (pinned: /sys/fs/bpf/layer_policy_map)
    config:
      max_flows: 65536
      window_size: 256
  - name: tc_retrans_scheduler
    type: classifier
    attach: eth0:egress
    bytecode: artifacts/tc_retrans_scheduler_v1.2.3.o

#!/bin/bash
# hot_reload.sh - 秒级无损热更新

NEW_PROG="xdp_rtp_loss_detect_v1.2.4.o"
PIN_PATH="/sys/fs/bpf/xdp_rtp_loss_detect"
IFACE="eth0"

# 1. 预加载新程序到内核 (验证器校验，不挂载)
bpftool prog load $NEW_PROG /sys/fs/bpf/new_prog type xdp 
    map name flow_ctx_map pinned $PIN_PATH/flow_ctx_map 
    map name layer_policy_map pinned $PIN_PATH/layer_policy_map

# 2. 原子替换 (内核保证原子性，无包丢失)
bpftool prog pin /sys/fs/bpf/new_prog $PIN_PATH/prog_new
bpftool netdev xdp replace dev $IFACE pinned $PIN_PATH/prog_new

# 3. 验证新程序运行状态 (run_cnt 增长)
sleep 2
bpftool prog show pinned $PIN_PATH/prog_new | grep run_cnt

# 4. 清理旧程序
bpftool prog unpinned $PIN_PATH/prog_old 2>/dev/null || true
mv $PIN_PATH/prog_new $PIN_PATH/prog_old

# .gitlab-ci.yml / .github/workflows/chaos.yml
stages:
  - build
  - unit_test
  - integration_test
  - chaos_test
  - canary_deploy

chaos_test:
  stage: chaos_test
  image: ubuntu:22.04
  services:
    - name: moon-gen:latest  # 高性能流量生成
    - name: kind:latest      # Kubernetes in Docker
  script:
    - make build-bpf
    - make deploy-kind       # 部署测试集群 (含 eBPF DaemonSet)
    - python3 chaos/run_suite.py --suite=media_plane --duration=300
    - python3 chaos/verify_slo.py --target=loss_concealment>0.8,retrans_latency_p99<5ms
  artifacts:
    reports:
      junit: chaos/results.xml
    when: always

# chaos/verify_slo.py
import json, sys

SLO_THRESHOLDS = {
    "loss_concealment_rate": 0.80,      # 丢包隐匿率
    "retrans_latency_p99_us": 5000,     # 重传延迟 P99 (us)
    "cpu_overhead_pct": 15.0,           # 单核 CPU 开销上限
    "zero_packet_loss_on_reload": True, # 热更期间零丢包
    "kernel_crash_count": 0,            # 内核崩溃容忍度
}

def verify(metrics_file):
    with open(metrics_file) as f:
        m = json.load(f)
    
    failed = []
    for k, threshold in SLO_THRESHOLDS.items():
        val = m.get(k)
        if val is None: continue
        if isinstance(threshold, bool):
            if val != threshold: failed.append(f"{k}: expected {threshold}, got {val}")
        elif val > threshold:
            failed.append(f"{k}: {val} > {threshold}")
    
    if failed:
        print("❌ SLO VIOLATION:", "; ".join(failed))
        sys.exit(1)
    print("✅ All SLOs met")

// 编译时剥离不必要 Helper (通过 BTF 类型信息验证)
// 仅保留：map_lookup/update, ktime_get, get_prandom, redirect, csum_level
// 禁止：bpf_probe_read_kernel, bpf_trace_printk, bpf_override_return 等高危 Helper

// 验证命令
bpftool prog load xdp.o /sys/fs/bpf/test type xdp --dev eth0 2>&1 | grep -i "unknown func"

# 1. 生成签名密钥对
openssl genrsa -out bpf_private.pem 2048
openssl rsa -in bpf_private.pem -pubout -out bpf_public.pem

# 2. 签名字节码
openssl dgst -sha256 -sign bpf_private.pem -out xdp.o.sig xdp.o

# 3. 部署时验证 (systemd service / initramfs hook)
verify_bpf() {
    local prog=$1
    openssl dgst -sha256 -verify bpf_public.pem -signature ${prog}.sig $prog
    if [ $? -ne 0 ]; then
        echo "BPF PROGRAM SIGNATURE MISMATCH: $prog"
        exit 1
    fi
}

单核最大并发流数 ≈ (可用指令预算 - 固定开销) / (每流处理指令数)
每流处理指令 ≈ 80 (解析) + 20 (位图) + 30 (Map查找) + 20 (统计) = 150 指令
单核极限 ≈ (200 * 0.7) / 150 ≈ 0.93 → 建议 **1 核处理 1Mpps (约 600 并发 50pps 流)**
生产建议：预留 50% 余量 → **1 核支撑 300 并发流 @ 10Gbps**

// tiny_nn_infer.c (伪代码，需内核支持 bpf_loop 或尾调用链)
#include <bpf/bpf_helpers.h>

// 量化后的 2 层 MLP: Input(5) -> Hidden(8) -> Output(3: action=drop/retrans/fec)
// 权重固化在 .rodata 或 Map 中
struct nn_weights {
    __s8 w1[5][8];  // int8 量化权重
    __s8 b1[8];
    __s8 w2[8][3];
    __s8 b2[3];
} __attribute__((aligned(64)));

static __always_inline __s32 dot_product(__s8 *a, __s8 *b, int n) {
    __s32 sum = 0;
    #pragma unroll
    for (int i = 0; i < n; i++) sum += a[i] * b[i];
    return sum;
}

SEC("xdp")
int xdp_ai_policy(struct xdp_md *ctx) {
    // 1. 特征提取 (5 维)
    __s8 features[5] = {
        current_loss_rate * 100,   // 0-100
        rtt_gradient,              // -128~127
        flow_priority,             // 0-7
        queue_depth_pct,           // 0-100
        time_since_last_loss       // 归一化
    };

    // 2. 推理 (约 200 指令，< 100ns)
    __s32 hidden[8];
    for (int i=0; i<8; i++) hidden[i] = relu(dot_product(features, w1[i], 5) + b1[i]);
    __s32 output[3];
    for (int j=0; j<3; j++) output[j] = dot_product(hidden, w2[j], 8) + b2[j];

    // 3. 动作决策
    int action = argmax(output, 3); // 0=Pass, 1=Retrans, 2=FEC
    return execute_action(ctx, action);
}

# /etc/sysctl.d/99-bpf-media.conf
# 网络栈优化
net.core.netdev_max_backlog = 250000
net.core.rmem_max = 67108864
net.core.wmem_max = 67108864
net.ipv4.udp_mem = 25600 51200 102400
net.ipv4.udp_rmem_min = 8192
net.ipv4.udp_wmem_min = 8192

# BPF 相关
kernel.bpf_stats_enabled = 1
kernel.unprivileged_bpf_disabled = 1  # 生产建议关闭非特权加载

# 巨页预留 (用于 AF_XDP / 大 Map)
vm.nr_hugepages = 4096
vm.hugetlb_shm_group = <bpf_user_gid>

# NUMA 亲和
kernel.numa_balancing = 0

# /etc/systemd/system/bpf-media-loader.service
[Unit]
Description=Media Plane eBPF Programs Loader
After=network-online.target
Wants=network-online.target

[Service]
Type=oneshot
RemainAfterExit=yes
ExecStartPre=/usr/local/bin/verify_bpf_signatures.sh
ExecStart=/usr/local/bin/bpf_loader.sh --manifest /etc/bpf/media-manifest.yaml
ExecReload=/usr/local/bin/bpf_loader.sh --hot-reload --manifest /etc/bpf/media-manifest.yaml
ExecStop=/usr/local/bin/bpf_loader.sh --unload --manifest /etc/bpf/media-manifest.yaml
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target