【网络工程】XDP 与 AF_XDP：eBPF 驱动的早期包处理

/* XDP 动作码 */
enum xdp_action {
    XDP_ABORTED = 0,  /* 程序异常，丢弃并记录错误 */
    XDP_DROP    = 1,  /* 丢弃包（最快的丢包方式） */
    XDP_PASS    = 2,  /* 交给正常网络栈处理 */
    XDP_TX      = 3,  /* 从收到包的网卡原路发回 */
    XDP_REDIRECT = 4, /* 重定向到其他网卡/CPU/AF_XDP */
};

# 1. Native XDP（原生模式）- 性能最高
# 需要网卡驱动支持 XDP hook
# 在驱动的 NAPI 收包回调中执行
ip link set dev eth0 xdpdrv obj xdp_prog.o sec xdp

# 2. Offloaded XDP（卸载模式）- 最快
# XDP 程序卸载到网卡硬件执行（如 Netronome）
# CPU 零开销，但支持的指令集有限
ip link set dev eth0 xdpoffload obj xdp_prog.o sec xdp

# 3. Generic XDP（通用模式）- 任何网卡都支持
# 在内核网络栈的 netif_receive_skb() 之后执行
# 性能最差，主要用于开发和测试
ip link set dev eth0 xdpgeneric obj xdp_prog.o sec xdp

# 查看当前 XDP 程序
ip link show dev eth0
# 输出示例：
# 2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> ...
#     prog/xdp id 42 tag abc123

# 卸载 XDP 程序
ip link set dev eth0 xdp off

# 安装开发依赖
# Ubuntu/Debian
apt install -y clang llvm libelf-dev libbpf-dev \
    linux-headers-$(uname -r) bpftool

# 编译 XDP 程序
# XDP 程序用 C 编写，通过 clang 编译为 eBPF 字节码
clang -O2 -g -target bpf -c xdp_prog.c -o xdp_prog.o

/* xdp_drop.c - 丢弃所有数据包 */
#include <linux/bpf.h>
#include <bpf/bpf_helpers.h>

SEC("xdp")
int xdp_drop_all(struct xdp_md *ctx) {
    /* ctx->data: 包数据起始指针
     * ctx->data_end: 包数据结束指针
     * ctx->ingress_ifindex: 入接口索引
     * ctx->rx_queue_index: 接收队列索引
     */
    return XDP_DROP;
}

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

/* xdp_synflood.c - SYN Flood 防御 */
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/tcp.h>
#include <linux/in.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_endian.h>

/* 每个源 IP 的 SYN 计数器 */
struct {
    __uint(type, BPF_MAP_TYPE_LRU_HASH);
    __uint(max_entries, 1000000);
    __type(key, __u32);    /* 源 IP */
    __type(value, __u64);  /* SYN 计数 */
} syn_count SEC(".maps");

/* 黑名单 */
struct {
    __uint(type, BPF_MAP_TYPE_LRU_HASH);
    __uint(max_entries, 100000);
    __type(key, __u32);    /* 源 IP */
    __type(value, __u64);  /* 封禁时间戳 */
} blacklist SEC(".maps");

#define SYN_THRESHOLD 100  /* 每秒 SYN 包阈值 */

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

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

    /* 解析 IP 头 */
    struct iphdr *ip = (void *)(eth + 1);
    if ((void *)(ip + 1) > data_end)
        return XDP_PASS;
    if (ip->protocol != IPPROTO_TCP)
        return XDP_PASS;

    /* 解析 TCP 头 */
    struct tcphdr *tcp = (void *)ip + (ip->ihl * 4);
    if ((void *)(tcp + 1) > data_end)
        return XDP_PASS;

    __u32 src_ip = ip->saddr;

    /* 检查黑名单 */
    __u64 *blocked = bpf_map_lookup_elem(&blacklist, &src_ip);
    if (blocked) {
        __u64 now = bpf_ktime_get_ns();
        /* 封禁 60 秒 */
        if (now - *blocked < 60000000000ULL)
            return XDP_DROP;
        /* 封禁过期，移除 */
        bpf_map_delete_elem(&blacklist, &src_ip);
    }

    /* 只检查 SYN 包（非 SYN-ACK） */
    if (!tcp->syn || tcp->ack)
        return XDP_PASS;

    /* 计数 */
    __u64 *count = bpf_map_lookup_elem(&syn_count, &src_ip);
    if (count) {
        (*count)++;
        if (*count > SYN_THRESHOLD) {
            /* 超过阈值，加入黑名单 */
            __u64 now = bpf_ktime_get_ns();
            bpf_map_update_elem(&blacklist, &src_ip, &now, BPF_ANY);
            return XDP_DROP;
        }
    } else {
        __u64 init = 1;
        bpf_map_update_elem(&syn_count, &src_ip, &init, BPF_ANY);
    }

    return XDP_PASS;
}

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

/* xdp_lb.c - 基于 XDP 的简单 L4 负载均衡器 */
#include <linux/bpf.h>
#include <linux/if_ether.h>
#include <linux/ip.h>
#include <linux/tcp.h>
#include <linux/udp.h>
#include <bpf/bpf_helpers.h>
#include <bpf/bpf_endian.h>

/* 后端服务器列表 */
struct backend {
    __u32 ip;
    unsigned char mac[ETH_ALEN];
};

struct {
    __uint(type, BPF_MAP_TYPE_ARRAY);
    __uint(max_entries, 16);
    __type(key, __u32);
    __type(value, struct backend);
} backends SEC(".maps");

/* VIP 配置 */
struct {
    __uint(type, BPF_MAP_TYPE_HASH);
    __uint(max_entries, 64);
    __type(key, __u32);    /* VIP */
    __type(value, __u32);  /* 后端数量 */
} vips SEC(".maps");

/* 统计 */
struct {
    __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
    __uint(max_entries, 4);
    __type(key, __u32);
    __type(value, __u64);
} stats SEC(".maps");

#define STAT_RX    0
#define STAT_TX    1
#define STAT_DROP  2
#define STAT_PASS  3

static __always_inline void update_stat(__u32 key) {
    __u64 *val = bpf_map_lookup_elem(&stats, &key);
    if (val) (*val)++;
}

/* 简单的哈希函数选择后端 */
static __always_inline __u32 hash_flow(struct iphdr *ip, __u32 n_backends) {
    return (ip->saddr ^ ip->daddr) % n_backends;
}

/* 重新计算 IP 校验和（增量更新） */
static __always_inline void update_ip_csum(struct iphdr *ip,
                                           __u32 old_ip, __u32 new_ip) {
    __u32 csum = (~ip->check & 0xFFFF);
    csum += (~old_ip & 0xFFFF) + (~(old_ip >> 16) & 0xFFFF);
    csum += (new_ip & 0xFFFF) + ((new_ip >> 16) & 0xFFFF);
    csum = (csum >> 16) + (csum & 0xFFFF);
    csum += csum >> 16;
    ip->check = ~csum;
}

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

    update_stat(STAT_RX);

    struct ethhdr *eth = data;
    if ((void *)(eth + 1) > data_end)
        return XDP_DROP;
    if (eth->h_proto != bpf_htons(ETH_P_IP))
        return XDP_PASS;

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

    /* 检查目标 IP 是否是 VIP */
    __u32 *n_backends = bpf_map_lookup_elem(&vips, &ip->daddr);
    if (!n_backends || *n_backends == 0) {
        update_stat(STAT_PASS);
        return XDP_PASS;
    }

    /* 选择后端 */
    __u32 idx = hash_flow(ip, *n_backends);
    struct backend *be = bpf_map_lookup_elem(&backends, &idx);
    if (!be) {
        update_stat(STAT_DROP);
        return XDP_DROP;
    }

    /* 修改目标 IP */
    __u32 old_daddr = ip->daddr;
    ip->daddr = be->ip;
    update_ip_csum(ip, old_daddr, be->ip);

    /* 修改 MAC 地址 */
    __builtin_memcpy(eth->h_dest, be->mac, ETH_ALEN);
    /* 源 MAC 设为 LB 的 MAC（需要从 map 中获取或硬编码） */

    update_stat(STAT_TX);
    return XDP_TX;  /* 从同一网卡发回 */
}

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

/* xdp_loader.c - 加载和管理 XDP 程序 */
#include <bpf/libbpf.h>
#include <bpf/bpf.h>
#include <net/if.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <unistd.h>
#include <errno.h>

static int ifindex;
static int prog_fd = -1;

void cleanup(int sig) {
    (void)sig;
    if (prog_fd >= 0) {
        bpf_xdp_detach(ifindex, 0, NULL);
        printf("XDP program detached\n");
    }
    exit(0);
}

int main(int argc, char *argv[]) {
    if (argc < 3) {
        fprintf(stderr, "Usage: %s <ifname> <xdp_prog.o>\n", argv[0]);
        return 1;
    }

    const char *ifname = argv[1];
    const char *progfile = argv[2];

    ifindex = if_nametoindex(ifname);
    if (!ifindex) {
        fprintf(stderr, "Interface %s not found\n", ifname);
        return 1;
    }

    signal(SIGINT, cleanup);
    signal(SIGTERM, cleanup);

    /* 打开 BPF 对象文件 */
    struct bpf_object *obj = bpf_object__open(progfile);
    if (!obj) {
        fprintf(stderr, "Failed to open %s\n", progfile);
        return 1;
    }

    /* 加载到内核 */
    if (bpf_object__load(obj)) {
        fprintf(stderr, "Failed to load BPF object\n");
        return 1;
    }

    /* 找到 XDP 程序 */
    struct bpf_program *prog = bpf_object__find_program_by_name(
        obj, "xdp_loadbalancer");
    if (!prog) {
        fprintf(stderr, "XDP program not found\n");
        return 1;
    }

    prog_fd = bpf_program__fd(prog);

    /* 附加到网卡 */
    struct bpf_xdp_attach_opts opts = {
        .sz = sizeof(opts),
    };
    if (bpf_xdp_attach(ifindex, prog_fd, 0, &opts)) {
        fprintf(stderr, "Failed to attach XDP: %s\n", strerror(errno));
        return 1;
    }

    printf("XDP program attached to %s (ifindex %d)\n", ifname, ifindex);

    /* 配置后端（示例） */
    int backends_fd = bpf_object__find_map_fd_by_name(obj, "backends");
    int vips_fd = bpf_object__find_map_fd_by_name(obj, "vips");

    /* 添加 VIP */
    __u32 vip = 0x0A000001;  /* 10.0.0.1 */
    __u32 n_backends = 2;
    bpf_map_update_elem(vips_fd, &vip, &n_backends, BPF_ANY);

    /* 添加后端 */
    struct backend {
        __u32 ip;
        unsigned char mac[6];
    };
    struct backend be0 = {
        .ip = 0x0A000064,   /* 10.0.0.100 */
        .mac = {0x00, 0x11, 0x22, 0x33, 0x44, 0x55}
    };
    __u32 idx = 0;
    bpf_map_update_elem(backends_fd, &idx, &be0, BPF_ANY);

    /* 定期打印统计 */
    int stats_fd = bpf_object__find_map_fd_by_name(obj, "stats");
    while (1) {
        sleep(1);
        __u64 rx = 0, tx = 0, drop = 0;
        __u32 key;
        /* PERCPU map 需要读取所有 CPU 的值 */
        int ncpus = libbpf_num_possible_cpus();
        __u64 values[ncpus];

        key = 0; /* STAT_RX */
        bpf_map_lookup_elem(stats_fd, &key, values);
        for (int i = 0; i < ncpus; i++) rx += values[i];

        key = 1; /* STAT_TX */
        bpf_map_lookup_elem(stats_fd, &key, values);
        for (int i = 0; i < ncpus; i++) tx += values[i];

        key = 2; /* STAT_DROP */
        bpf_map_lookup_elem(stats_fd, &key, values);
        for (int i = 0; i < ncpus; i++) drop += values[i];

        printf("RX: %llu  TX: %llu  DROP: %llu\n", rx, tx, drop);
    }

    return 0;
}

/* af_xdp_recv.c - AF_XDP 收包示例 */
#include <linux/if_xdp.h>
#include <linux/if_link.h>
#include <bpf/xsk.h>
#include <bpf/bpf.h>
#include <net/if.h>
#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>

#define NUM_FRAMES    4096
#define FRAME_SIZE    XSK_UMEM__DEFAULT_FRAME_SIZE  /* 4096 */
#define BATCH_SIZE    64

struct xsk_socket_info {
    struct xsk_ring_cons rx;
    struct xsk_ring_prod tx;
    struct xsk_ring_prod fq;     /* Fill Queue */
    struct xsk_ring_cons cq;     /* Completion Queue */
    struct xsk_socket *xsk;
    struct xsk_umem *umem;
    void *umem_area;
};

/* 初始化 UMEM */
int setup_umem(struct xsk_socket_info *xsk_info) {
    size_t umem_size = NUM_FRAMES * FRAME_SIZE;

    /* 分配页对齐的内存 */
    if (posix_memalign(&xsk_info->umem_area, getpagesize(), umem_size)) {
        return -1;
    }

    struct xsk_umem_config cfg = {
        .fill_size = XSK_RING_PROD__DEFAULT_NUM_DESCS,
        .comp_size = XSK_RING_CONS__DEFAULT_NUM_DESCS,
        .frame_size = FRAME_SIZE,
        .frame_headroom = XSK_UMEM__DEFAULT_FRAME_HEADROOM,
        .flags = 0,
    };

    int ret = xsk_umem__create(&xsk_info->umem, xsk_info->umem_area,
                                umem_size, &xsk_info->fq, &xsk_info->cq,
                                &cfg);
    return ret;
}

/* 创建 AF_XDP Socket */
int setup_xsk_socket(struct xsk_socket_info *xsk_info,
                     const char *ifname, int queue_id) {
    struct xsk_socket_config cfg = {
        .rx_size = XSK_RING_CONS__DEFAULT_NUM_DESCS,
        .tx_size = XSK_RING_PROD__DEFAULT_NUM_DESCS,
        .libbpf_flags = XSK_LIBBPF_FLAGS__INHIBIT_PROG_LOAD,
        .xdp_flags = 0,
        .bind_flags = XDP_USE_NEED_WAKEUP,
    };

    int ret = xsk_socket__create(&xsk_info->xsk, ifname, queue_id,
                                  xsk_info->umem, &xsk_info->rx,
                                  &xsk_info->tx, &cfg);
    if (ret) return ret;

    /* 初始填充 Fill Queue：告诉内核哪些 frame 可以用来收包 */
    __u32 idx;
    ret = xsk_ring_prod__reserve(&xsk_info->fq, NUM_FRAMES / 2, &idx);
    for (__u32 i = 0; i < NUM_FRAMES / 2; i++) {
        *xsk_ring_prod__fill_addr(&xsk_info->fq, idx + i) =
            i * FRAME_SIZE;
    }
    xsk_ring_prod__submit(&xsk_info->fq, NUM_FRAMES / 2);

    return 0;
}

/* 接收循环 */
void rx_loop(struct xsk_socket_info *xsk_info) {
    struct pollfd fds = {
        .fd = xsk_socket__fd(xsk_info->xsk),
        .events = POLLIN,
    };

    unsigned long rx_packets = 0;

    while (1) {
        /* 检查是否需要唤醒 */
        if (xsk_ring_prod__needs_wakeup(&xsk_info->fq)) {
            poll(&fds, 1, 1000);
        }

        __u32 idx_rx = 0;
        unsigned int rcvd = xsk_ring_cons__peek(&xsk_info->rx,
                                                 BATCH_SIZE, &idx_rx);
        if (!rcvd) continue;

        for (__u32 i = 0; i < rcvd; i++) {
            const struct xdp_desc *desc =
                xsk_ring_cons__rx_desc(&xsk_info->rx, idx_rx + i);

            /* 直接访问包数据（零拷贝） */
            void *pkt = xsk_umem__get_data(xsk_info->umem_area,
                                            desc->addr);
            __u32 len = desc->len;

            /* 处理数据包 */
            process_raw_packet(pkt, len);
        }
        xsk_ring_cons__release(&xsk_info->rx, rcvd);

        /* 回收 frame 到 Fill Queue */
        __u32 idx_fq;
        if (xsk_ring_prod__reserve(&xsk_info->fq, rcvd, &idx_fq) == rcvd) {
            for (__u32 i = 0; i < rcvd; i++) {
                const struct xdp_desc *desc =
                    xsk_ring_cons__rx_desc(&xsk_info->rx, idx_rx + i);
                *xsk_ring_prod__fill_addr(&xsk_info->fq, idx_fq + i) =
                    desc->addr;
            }
            xsk_ring_prod__submit(&xsk_info->fq, rcvd);
        }

        rx_packets += rcvd;
        if (rx_packets % 1000000 == 0) {
            printf("Received %lu packets\n", rx_packets);
        }
    }
}

void process_raw_packet(void *data, __u32 len) {
    struct ethhdr *eth = data;
    (void)len;
    /* 解析并处理原始以太网帧 */
    /* 用户在这里做任意复杂的处理 */
    (void)eth;
}

# AF_XDP 有两种数据传输模式：

# 1. 拷贝模式（默认）：
#    驱动将包数据拷贝到 UMEM 中
#    任何网卡都支持
bind_flags = 0;

# 2. 零拷贝模式：
#    驱动直接 DMA 到 UMEM（需要驱动支持）
#    性能最高，但支持的网卡有限
bind_flags = XDP_ZEROCOPY;

# 支持零拷贝的驱动（Linux 6.x）：
# - Intel: i40e, ice, ixgbe, igc
# - Mellanox: mlx5
# - 虚拟: veth

/*
 * Maglev 哈希的核心思想：
 *
 * 1. 为每个后端生成一个偏好列表（permutation）
 * 2. 轮流让每个后端"填入"查找表
 * 3. 结果是一个固定大小的查找表，
 *    添加/删除后端时变化最小
 *
 * 查找表大小选择质数（如 65537）
 * 保证均匀分布
 */

/* 简化的 Maglev 查找表构建 */
#define TABLE_SIZE 65537  /* 质数 */

struct maglev_table {
    int lookup[TABLE_SIZE];  /* 后端索引 */
    int n_backends;
};

void build_maglev_table(struct maglev_table *t, int n_backends) {
    /* 每个后端的偏好排列 */
    int *permutation = calloc(n_backends * TABLE_SIZE, sizeof(int));
    int *next = calloc(n_backends, sizeof(int));
    memset(t->lookup, -1, sizeof(t->lookup));

    /* 生成偏好排列（使用两个哈希函数） */
    for (int i = 0; i < n_backends; i++) {
        uint32_t offset = hash1(i) % TABLE_SIZE;
        uint32_t skip = (hash2(i) % (TABLE_SIZE - 1)) + 1;
        for (int j = 0; j < TABLE_SIZE; j++) {
            permutation[i * TABLE_SIZE + j] =
                (offset + j * skip) % TABLE_SIZE;
        }
    }

    /* 轮流填入查找表 */
    int filled = 0;
    while (filled < TABLE_SIZE) {
        for (int i = 0; i < n_backends && filled < TABLE_SIZE; i++) {
            int pos = permutation[i * TABLE_SIZE + next[i]];
            while (t->lookup[pos] != -1) {
                next[i]++;
                pos = permutation[i * TABLE_SIZE + next[i]];
            }
            t->lookup[pos] = i;
            next[i]++;
            filled++;
        }
    }

    t->n_backends = n_backends;
    free(permutation);
    free(next);
}

/* 查找：O(1) */
int maglev_lookup(struct maglev_table *t, uint32_t hash) {
    return t->lookup[hash % TABLE_SIZE];
}

# 列出已加载的 XDP 程序
bpftool prog list
# 42: xdp  name xdp_syn_filter  tag abc123  gpl
#     loaded_at 2025-07-25T10:00:00+0000  uid 0
#     xlated 1234B  jited 890B  memlock 4096B

# 查看 XDP 程序的 JIT 代码
bpftool prog dump jited id 42

# 查看 BPF Map 内容
bpftool map list
bpftool map dump id 5

# 查看接口上的 XDP 程序
bpftool net list
# xdp:
# eth0(2) driver id 42

# 查看 XDP 统计
bpftool prog show id 42
# 查看运行次数和平均耗时

# XDP 相关的 tracepoints
# 列出所有 XDP tracepoints
perf list 'xdp:*'
# xdp:xdp_exception      XDP 程序返回异常
# xdp:xdp_bulk_tx        批量发送
# xdp:xdp_redirect        重定向事件
# xdp:xdp_redirect_err   重定向错误
# xdp:xdp_redirect_map   Map 重定向

# 监控 XDP 丢弃
bpftrace -e '
tracepoint:xdp:xdp_exception {
    @drops[args->act] = count();
}
interval:s:1 { print(@drops); clear(@drops); }
'

# XDP 程序性能分析
bpftrace -e '
kprobe:bpf_prog_run_xdp {
    @start[tid] = nsecs;
}
kretprobe:bpf_prog_run_xdp /@start[tid]/ {
    @ns = hist(nsecs - @start[tid]);
    delete(@start[tid]);
}
'

# 查看 XDP 包处理统计（使用 ethtool）
ethtool -S eth0 | grep xdp
# rx_xdp_aborted: 0
# rx_xdp_drop: 1523456
# rx_xdp_pass: 9876543
# rx_xdp_tx: 0
# rx_xdp_tx_errors: 0
# rx_xdp_redirect: 456789

# 问题 1：XDP 程序加载失败
# "invalid BPF program" / "verifier error"
# 原因：eBPF 验证器拒绝了程序

# 查看详细的验证器错误
bpftool prog load xdp_prog.o /sys/fs/bpf/xdp_prog 2>&1
# 常见原因：
# - 未检查边界（data + offset > data_end）
# - 循环次数不确定
# - 栈空间超过 512 字节
# - 使用了不允许的帮助函数

# 问题 2：性能低于预期
# 检查是否在 generic 模式运行
ip link show dev eth0 | grep xdp
# "xdpgeneric" 表示通用模式，性能差
# "xdp" 或 "xdpdrv" 表示原生模式

# 检查驱动是否支持 native XDP
ethtool -i eth0
# driver: i40e  → 支持
# driver: r8169 → 不支持

# 问题 3：AF_XDP 收不到包
# 检查 XDP 程序是否正确重定向
bpftool map dump name xsks_map
# 确认 queue_id 映射正确

# 检查 FILL Ring 是否有可用 frame
# 如果 FILL Ring 为空，内核无处放收到的包