如果你需要在用户态拿到网卡上的高速数据包,又不想像DPDK那样把整块网卡摁在用户态驱动里,AF_XDP基本就是Linux原生生态里最值得上手的一条路。它借助XDP程序把包直接从驱动层重定向到一个特殊套接字,配合libbpf负责BPF程序的加载、libxdp负责AF_XDP socket的管理,一套代码就能在两三千行以内实现一个高性能收包/发包通道。这篇文章用一个最小可运行的程序,把从UMEM初始化、XSK Socket创建到XDP程序挂载、收包循环为止的所有环节完整过一遍。适合有C语言基础、知道BPF是什么但还没真正动过AF_XDP的人,跟着敲完,你大概就能理解这套东西是怎么转起来,也踩不到我当初踩过的那些坑。
1. 别急着写代码,先弄懂AF_XDP的数据流与选型
很多新手拿到AF_XDP第一反应是翻API,然后被一堆Ring、Queue、UMEM的术语绕晕。我建议反过来,先搞清楚一个包从网卡进来之后,是怎么一路走到你用户态程序里的。数据流顺了,API无非就是对应数据流上的几个动作。
1.1 包是怎么从网卡“零拷贝”到用户态的
网卡驱动在收到帧之后,会先走XDP hook。如果网卡支持XDP,驱动会在NAPI上下文里把缓冲信息封装成xdp_buff,交给我们挂载的BPF程序。BPF程序看一眼rx_queue_index,决定调用bpf_redirect_map把它送进XSKMAP对应的AF_XDP socket。驱动随后把整个packet的缓冲区信息整理成一个xdp_desc——里面是UMEM偏移和长度——塞进该socket的RX Ring。用户态进程只要poll一下socket fd,就能知道有没有新包,再从RX Ring里拿出描述符,用xsk_umem__get_data拿到指向数据的内存地址。
整个过程里,数据本身没有复制,搬来搬去的只是描述符。AF_XDP说的“零拷贝”,指的是数据从网卡DMA进UMEM之后,到用户态读完为止,不再有第二次拷贝。用户态和内核态共享同一块内存池,内核把包放进去,用户态直接读。这一点和AF_PACKET那种“内核skb + recvfrom拷贝”是完全不同的设计。
这里要特别提醒一句:零拷贝指的是数据面,但描述符本身还是要通过Ring来交换。Ring就是用户态和内核态共享的环形队列,内核写、用户读,或者反过来。它们都是无锁的,靠内存屏障和缓存指针来同步,所以效率很高,但也意味着你代码里对Ring的消费和释放时机必须严格正确,否则就会出现数据覆盖或者丢描述符的诡异问题。
1.2 为什么不用裸bpf而要用libbpf/libxdp
理论上你可以通过bpf()系统调用自己加载XDP程序,自己创建AF_XDP socket,自己mmap所有Ring。但真这么做的话,你要面对一堆内核结构体、版本兼容性、BPF指令校验问题,而且容易在细节上翻车。libbpf解决了BPF程序从编译产物到加载、校验、map操作的整套流程,是目前BPF生态的事实标准基础库。libxdp则进一步把XDP和AF_XDP的开发体验拉高了一个台阶,像xsk_socket__create、xsk_umem__create、xsk_socket__update_xskmap这些接口,把原本几十行甚至上百行的样板代码压缩成了几个函数调用。
和AF_PACKET比,AF_XDP绕过了内核协议栈,拿包更直接,延迟和CPU开销都低得多;和DPDK比,AF_XDP不需要把整块网卡接管成用户态驱动,内核协议栈还能照常处理没有被重定向的流量,程序可以随时attach和detach,开发和调试成本低很多。当然,代价是上限性能和DPDK那种彻底绕过内核的模型还有差距,但对于绝大多数需要高性能收包、负载均衡、旁路监控的场景来说,AF_XDP在可维护性和性能之间已经是非常好的平衡点。
用libbpf/libxdp还有个实际好处:它们会帮你处理很多内核版本差异。比如不同内核上XDP程序attach的方式、Ring size的限制、NEED_WAKEUP语义的差异,库都做了兼容。你只要跟着库的API走,基本不会踩到“换个内核代码就跑不了”的坑。
1.3 环境准备:内核、依赖、权限
我建议直接用内核5.10以上的发行版做实验,比如Ubuntu 22.04、CentOS Stream 9这类。太老的内核虽然也能跑AF_XDP,但像NEED_WAKEUP、XDP multi-buffer这些特性会缺失,学习时对比起来很麻烦。
依赖方面需要三样东西:
- libbpf-dev:BPF程序的加载库。
- libxdp-dev:AF_XDP socket和XDP程序的封装库。
- clang/llvm:把C源码编译成BPF目标文件。
Ubuntu 22.04上直接安装:
apt install -y clang llvm libbpf-dev libxdp-dev bpftool linux-tools-generic如果你的发行版没有libxdp,可以自己去xdp-tools仓库编译安装:
git clone https://github.com/xdp-project/xdp-tools cd xdp-tools make -C lib/libxdp make -C lib/libxdp install安装完先确认内核开了AF_XDP支持:
grep XDP_SOCKETS /boot/config-$(uname -r)能看到CONFIG_XDP_SOCKETS=y就说明内核支持。
权限上,加载BPF程序和attach XDP需要root或者CAP_NET_ADMIN、CAP_SYS_ADMIN,创建AF_XDP socket和注册UMEM默认要求root或者CAP_NET_RAW。平时做实验我建议直接sudo跑,并且把内存锁限制放开,因为UMEM注册时需要锁定内存,默认的ulimit -l太小会导致注册失败:
ulimit -l unlimited2. 最小闭环需要哪些零件:UMEM、Queue与XSKMAP
一个能收包的AF_XDP程序,跑起来之后你会看到它同时牵扯着内存池、四个Ring、一个socket、一个BPF map和一段XDP程序。每个零件都有明确的分工,缺一个都转不动。我按依赖关系一个个说。
2.1 UMEM:用户态和内核共享的内存池
UMEM是整个AF_XDP的数据仓库。你在用户态分配一块大内存,注册给内核,然后这块内存就被拆成等长的chunk。内核收到网络包时,会从Fill Ring拿一个空闲chunk地址,把包DMA进去;用户态处理完这个包之后,再把chunk地址还回Fill Ring,让内核下次继续用。
UMEM有两个关键参数:frame_size和frame_headroom。frame_size决定每个chunk多大,至少要能放得下一个完整的数据帧,一般设2048字节,跑超大MTU可能要4096甚至更大。frame_headroom是在chunk开头预留的一段空间,常见的场景是给包处理元数据留位置,不需要就设0。
分配UMEM时有两个坑我特别提醒:
- 内存必须按页对齐,建议用mmap分配,并且加上MAP_POPULATE把物理页预分配出来,避免运行时缺页影响收包性能。
- 你在Fill Ring里放的是chunk的“偏移地址”,不是绝对地址。比如第i个chunk的偏移是i * frame_size,不是
(unsigned long)umem_base + i * frame_size。这个搞反了,xsk_umem__get_data拿到的数据位置会完全不对。
2.2 四个Ring谁进谁出
AF_XDP socket上一共有四个Ring,名字和方向是这样的:
| Ring | 生产者 | 消费者 | 内容 | 做什么 |
|---|---|---|---|---|
| Fill Ring (FQ) | 用户态 | 内核 | 空闲chunk的偏移 | 用户态把可用chunk交给内核 |
| RX Ring | 内核 | 用户态 | xdp_desc描述符 | 内核把收到的包交给用户态 |
| TX Ring | 用户态 | 内核 | xdp_desc描述符 | 用户态把要发的包交给内核 |
| Completion Ring (CQ) | 内核 | 用户态 | 已完成发送的chunk偏移 | 内核通知用户态发完了 |
填包的是Fill,收包的是RX,发包的是TX,回收的是Completion,这四个方向的语义如果记混,后面代码基本没法写。很多初学者会以为RX Ring里直接放着包数据,其实不是,Ring里只有16字节的desc,包数据始终在UMEM里,desc里的addr字段告诉你去UMEM的哪个偏移读。
Ring size必须是2的幂,这是内核无锁队列的硬性要求。常见的选择是1024、2048、4096,我示例里用4096。太小容易在大流量下来不及消费,太大则占用内存和缓存。
2.3 XSK Socket与XSKMAP:队列到应用的映射
AF_XDP socket对应的不是一个“连接”,而是一对“网卡队列号 + 用户态通道”。你创建一个socket时,需要告诉内核网卡名和queue_id,绑定之后,这个socket就只负责接收某个网卡队列上被XDP程序重定向过来的包。
XSKMAP是一个BPF_MAP_TYPE_XSKMAP类型的map,key是queue_id,value是xsk socket的fd。XDP程序在驱动收包路径上做判决时,用bpf_redirect_map从map里查这个queue_id对应的socket fd,查到就把包重定向进去。
所以整个映射关系是:网卡队列 -> XSKMAP -> XSK Socket -> UMEM chunk。如果网卡有多队列,你可以为每个队列创建一个socket,分别放进XSKMAP,就实现了多队列并行收包。这也是AF_XDP能横向扩展吞吐的核心机制。
2.4 XDP程序:判决在哪里放行
XDP程序是整个数据面的“入口闸门”。它跑在驱动收到包之后、进入内核协议栈之前,我们在这个位置决定包是送到AF_XDP socket、丢掉还是继续走协议栈。
最小可用的XDP程序甚至不需要几行:
#include <linux/bpf.h> #include <bpf/bpf_helpers.h> struct { __uint(type, BPF_MAP_TYPE_XSKMAP); __uint(max_entries, 64); __type(key, __u32); __type(value, __u32); } xsks_map SEC(".maps"); SEC("xdp") int xsk_demo_prog(struct xdp_md *ctx) { __u32 queue_id = ctx->rx_queue_index; return bpf_redirect_map(&xsks_map, queue_id, XDP_PASS); } char _license[] SEC("license") = "GPL";这个程序做的事情很直白:从上下文拿到当前包来自哪个队列,然后调bpf_redirect_map把这个队列的包送进XSKMAP里对应的socket。第三个参数XDP_PASS是“如果map里查不到这个队列”时的兜底动作,这里选择让包继续走内核协议栈。实际项目里可以根据需要改成XDP_DROP,或者对特定五元组做分流。
3. 动手写一个可运行的实例
前面理论讲完,接下来是实操。我会给出一份尽可能精简但五脏俱全的示例,目标是把eth0第0号队列的收包全部重定向到用户态,程序启动后不断打印收到的包。
用户态程序公共部分是这样的:
#include <stdio.h> #include <stdlib.h> #include <string.h> #include <errno.h> #include <poll.h> #include <unistd.h> #include <sys/mman.h> #include <net/if.h> #include <linux/bpf.h> #include <bpf/libbpf.h> #include <bpf/bpf.h> #include <xdp/xsk.h> #define FRAME_SIZE 2048 #define NUM_FRAMES 4096 #define RING_SIZE 4096 #define BATCH_SIZE 64 static void *umem_area; static struct xsk_umem *umem; static struct xsk_ring_prod fill_r; static struct xsk_ring_cons comp_r; static struct xsk_ring_cons rx_r; static struct xsk_ring_prod tx_r; static struct xsk_socket *xsk; static struct bpf_link *link; static int xskmap_fd;3.1 第一步:初始化UMEM并填满Fill Queue
UMEM初始化是整个程序的地基,必须在创建socket之前完成。代码分三件事:mmap出内存、用xsk_umem__create注册、把所有chunk地址塞进Fill Queue。
static void setup_umem(void) { struct xsk_umem_config cfg = { .frame_size = FRAME_SIZE, .frame_headroom = 0, .flags = 0, }; size_t umem_size = FRAME_SIZE * NUM_FRAMES; unsigned int i, idx; umem_area = mmap(NULL, umem_size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0); if (umem_area == MAP_FAILED) { perror("mmap umem"); exit(1); } if (xsk_umem__create(&umem, umem_area, umem_size, &fill_r, &comp_r, &cfg)) { fprintf(stderr, "xsk_umem__create failed: %s\n", strerror(errno)); exit(1); } if (xsk_ring_prod__reserve(&fill_r, NUM_FRAMES, &idx) != NUM_FRAMES) { fprintf(stderr, "fill ring reservation failed\n"); exit(1); } for (i = 0; i < NUM_FRAMES; i++) { *xsk_ring_prod__fill_addr(&fill_r, idx + i) = i * FRAME_SIZE; } xsk_ring_prod__submit(&fill_r, NUM_FRAMES); }注意这里Fill Ring的capacity就是RING_SIZE=4096,NUM_FRAMES也是4096,正好能一次性放满。每个chunk的地址就是简单的i * FRAME_SIZE,这个值是相对于UMEM起始的偏移量,不是绝对指针。这个约定是整个AF_XDP地址体系的核心,读包时用xsk_umem__get_data(umem, addr),填包时往fill_addr里写offset,两边都是同一个坐标系。
3.2 第二步:加载XDP程序并拿到XSKMAP
XDP程序需要先编译成.o文件,再用libbpf加载。加载成功之后,我们从BPF object里找到xsks_map的fd,后面要把xsk socket fd登记进去。
static void load_xdp_prog(const char *ifname) { struct bpf_object *obj; struct bpf_program *prog; int ifindex = if_nametoindex(ifname); obj = bpf_object__open_file("xsk_demo_prog.o", NULL); if (libbpf_get_error(obj)) { fprintf(stderr, "open xdp program failed\n"); exit(1); } if (bpf_object__load(obj)) { fprintf(stderr, "load xdp program failed\n"); exit(1); } prog = bpf_object__find_program_by_name(obj, "xsk_demo_prog"); xskmap_fd = bpf_object__find_map_fd_by_name(obj, "xsks_map"); if (!prog || xskmap_fd < 0) { fprintf(stderr, "cannot find prog/map in object\n"); exit(1); } link = bpf_program__attach_xdp(prog, ifindex); if (libbpf_get_error(link)) { fprintf(stderr, "attach xdp failed: %s\n", strerror(errno)); exit(1); } }一个很多人会忽略的点:attach之前,如果网卡上已经有别的XDP程序,attach可能失败。实验环境里最省事的做法是先用bpftool net detach xdp dev eth0清掉旧的,再启动我们的程序。生产环境则应该用更完善的XDP dispatcher机制来管理多个程序的共存,libxdp在这块做了很多工作。
3.3 第三步:创建XSK Socket并登记到Map
创建socket这步看似简单,但它把前面所有零件串起来了:UMEM、Fill/RX/TX Ring、网卡队列,全部在这个调用里生效。
static void setup_xsk(const char *ifname, __u32 queue_id) { struct xsk_socket_config cfg = { .rx_size = RING_SIZE, .tx_size = RING_SIZE, }; if (xsk_socket__create(&xsk, ifname, queue_id, umem, &rx_r, &tx_r, &cfg)) { fprintf(stderr, "xsk_socket__create failed: %s\n", strerror(errno)); exit(1); } if (xsk_socket__update_xskmap(xsk, xskmap_fd)) { fprintf(stderr, "update xskmap failed: %s\n", strerror(errno)); exit(1); } }xsk_socket__create内部其实做了很多事:创建socket fd、setsockopt注册UMEM、配置RX/TX Ring、bind到指定网卡队列。xsk_socket__update_xskmap则等价于执行了一次bpf_map_update_elem,把当前socket fd塞进XSKMAP对应的queue_id位置。做完这步,XDP程序才能把包重定向到我们这个socket。
这里提醒一下:创建socket时传的是网卡名而不是ifindex,因为libxdp内部会自己转。bind_flags默认是copy模式,也就是内核先把包拷贝到UMEM再给用户态,驱动不支持零拷贝时也能跑。想切到零拷贝,可以在xsk_socket_config里设bind_flags = XDP_ZEROCOPY,但要求网卡驱动支持,不然会报错。
3.4 第四步:收包循环怎么转
收包循环就是整个AF_XDP程序的核心循环,逻辑上是四步:poll等可读、从RX Ring批量peek、处理数据、把用过的chunk还回Fill Ring。
static void rx_process(void) { struct pollfd pfd = { .fd = xsk_socket__fd(xsk), .events = POLLIN, }; unsigned int rcvd, i, idx_rx, idx_fq; __u64 addrs[BATCH_SIZE]; if (poll(&pfd, 1, -1) <= 0) return; rcvd = xsk_ring_cons__peek(&rx_r, BATCH_SIZE, &idx_rx); if (rcvd == 0) return; for (i = 0; i < rcvd; i++) { const struct xdp_desc *desc = xsk_ring_cons__rx_desc(&rx_r, idx_rx + i); void *pkt = xsk_umem__get_data(umem, desc->addr); addrs[i] = desc->addr; printf("got packet len=%u\n", desc->len); for (int j = 0; j < 16 && j < desc->len; j++) printf("%02x ", ((unsigned char *)pkt)[j]); printf("\n"); } xsk_ring_cons__release(&rx_r, rcvd); if (xsk_ring_prod__reserve(&fill_r, rcvd, &idx_fq) == rcvd) { for (i = 0; i < rcvd; i++) *xsk_ring_prod__fill_addr(&fill_r, idx_fq + i) = addrs[i]; xsk_ring_prod__submit(&fill_r, rcvd); } }最关键的是两个配对的API:peek和release是消费RX Ring的一对,reserve和submit是生产Fill Ring的一对。很多刚上手的人只peek不release,或者只reserve不submit,结果是Ring指针越走越乱,最终程序卡死或者收不到包。
处理完包之后必须把用过的chunk还回Fill Ring。如果不还,UMEM里的chunk只会越来越少,内核没有空闲chunk可用来收包,很快整个收包通道就会停摆。这个归还动作是AF_XDP编程里最容易漏掉的一环。
还回去的时机最好在release之后统一做,因为这样我们只调用一次reserve/submit,吞吐更高。如果BATCH_SIZE设置得大,比如256,一次处理256个包再统一归还,比每包单独归还要快不少。
3.5 第五步(可选):发送路径与Completion处理
接收路径通了,发送路径其实只多三个动作:从TX Ring拿一个desc、往UMEM对应chunk里写数据、提交并唤醒内核。发完之后,通过Completion Ring回收chunk。
static void send_one_pkt(__u64 addr, int len) { unsigned int idx; struct xdp_desc *desc; if (xsk_ring_prod__reserve(&tx_r, 1, &idx) != 1) return; desc = xsk_ring_prod__tx_desc(&tx_r, idx); desc->addr = addr; desc->len = len; xsk_ring_prod__submit(&tx_r, 1); /* 让内核知道TX Ring有新包 */ sendto(xsk_socket__fd(xsk), NULL, 0, MSG_DONTWAIT); /* 回收发送完成的chunk */ unsigned int done; if ((done = xsk_ring_cons__peek(&comp_r, 1, &idx))) { /* 这里可以把chunk重新交给Fill Ring,或者放进自己的空闲池 */ xsk_ring_cons__release(&comp_r, done); } }这里有一层很多人第一次没意识到的逻辑:发送时用的chunk必须是你自己“占住”的,不能跟接收路径共用同一个chunk,否则会出现一边读一边写的数据竞争。真实项目里通常要在用户态维护一个chunk的空闲池和引用计数,收包用完归还、发送完成归还,这个所有权管理是AF_XDP开发里最费心思的部分。示例里我们只说明骨架,不展开完整的池管理。
发送完之后的sendto就是“踢一下”内核,让内核对TX Ring做处理。如果设置了XDP_USE_NEED_WAKEUP,则还要用poll或者检查xsk_ring_prod__needs_wakeup来判断是否需要踢;没设置的话,NAPI也会周期处理,但主动踢一次更靠谱。
3.6 编译、运行与验证
XDP程序编译成BPF目标文件:
clang -O2 -g -Wall -target bpf -c xsk_demo_prog.c -o xsk_demo_prog.o用户态程序编译:
gcc -O2 -g -Wall xsk_user.c -o xsk_user -lxdp -lbpf运行:
ulimit -l unlimited sudo ./xsk_user eth0 0如果你手头没有支持AF_XDP的物理网卡,用veth也能做功能验证。veth在较新的内核上支持XDP重定向,性能一般,但验证逻辑没问题。先建一对veth:
ip link add veth0 type veth peer name veth1 ip link set veth0 up ip link set veth1 up ip addr add 10.0.0.1/24 dev veth0 ip addr add 10.0.0.2/24 dev veth1然后一个终端跑sudo ./xsk_user veth0 0,另一个终端ping:
ping -I veth1 10.0.0.1你会看到程序不断打印收到的包。注意,因为所有进入veth0的包都被XDP重定向到了用户态,内核协议栈收不到这些包,ping很可能会超时。这是AF_XDP的正常行为——流量被“接管”了。如果希望协议栈还能继续处理,需要在XDP程序里用bpf_clone_redirect复制一份,或者在用户态自己处理回包。
4. 实战中的坑与排查技巧
AF_XDP的资料不算多,很多问题只有在实操里才会暴露。这一节我把跑代码时最容易踩的坑和排查思路整理出来,按照报错/现象来组织,你可以当成速查表用。
4.1 权限与内存锁问题
最常见的启动失败是各种Operation not permitted。
创建AF_XDP socket报EPERM,一般是用户态没有CAP_NET_RAW;加载BPF程序报EPERM,需要CAP_BPF和CAP_SYS_ADMIN;注册UMEM报EPERM或者EAGAIN,最有可能的是mlock内存超过限额。内核在注册UMEM时要锁定用户态内存,ulimit -l默认值往往不够,尤其是你分配了8MB甚至更大的UMEM。
解决办法很直接:
ulimit -l unlimited如果用的是容器或者systemd unit,还需要在对应配置里放开MemoryLockLimit。
还有一个容易被忽略的点:如果你用root跑程序但报的还是EPERM,检查一下是不是容器缺cap-add。我见过有人在docker里折腾半天,最后加上--cap-add=NET_ADMIN --cap-add=SYS_ADMIN就好了。
4.2 “怎么收不到包”的排查路径
程序起来了,但在那干等着没输出。我的排查顺序是这样的:
第一,确认XDP程序确实挂上去了:
bpftool net show dev eth0如果这里没显示程序,说明attach那步出了问题,或者被其他程序顶掉了。
第二,确认XSKMAP里确实有值:
bpftool map dump id <map_id>能看到key=0、value是某个非零fd,说明socket登记成功了。如果value是0,说明xsk_socket__update_xskmap没生效,或者run的时候网卡队列号对不上。
第三,确认流量确实经过了目标队列。如果你绑定的是queue 0,但网卡RSS哈希把包都分到了queue 1,那自然收不到。测试时最省事的是把多队列合并成一个:
ethtool -L eth0 combined 1这样所有包都走queue 0,XSKMAP也能覆盖到。生产环境多队列场景下,你需要为每个队列创建一个socket,而不是只绑一个。
第四,确认XDP程序的判决逻辑。可以把返回值临时改成return XDP_PASS,然后看协议栈能不能正常收包,这能快速判断驱动XDP hook本身有没有问题。如果PASS也没用,多半是驱动不支持,或者veth这类接口根本不走这条路径。
4.3 参数选型与常见配置错误
Ring size必须设为2的幂,这是内核无锁环形队列的硬约束。你设个1000、2047,内核会直接拒绝或者行为异常。示例里的4096对多数场景都够用。
frame_size太小会导致大包丢包。如果你的MTU是1500,2048够;但如果你在跑9000的jumbo frame,frame_size必须至少是帧大小加一些头部余量,否则包会被截断或者触发XDP multi-buffer逻辑。简单起见,先设2048,遇到大包问题再往上调。
fill ring的容量最好大于等于rx ring,这样初始化时能一次性把所有chunk都交给内核。如果fill ring比chunk总数小,那么初始化时只能放一部分chunk进去,剩下的chunk要等收包消费后再归还,逻辑上更绕,也容易出现“fill里没有空闲chunk”的窗口期。
还有一个小坑:XDP程序和用户态程序的加载顺序。如果XDP程序已经先attach了,但xskmap里还没有任何socket,那么流量进入XDP hook时查不到目标,会走XDP_PASS兜底,正常走协议栈,所以你暂时看不到报错。等到socket创建、xskmap更新后,流量才切到用户态。这个“中间态”有时候会让人误以为程序没生效。
4.4 用bpftool和xdpdump排查现场
bpftool是排查BPF问题的瑞士军刀,三个命令我几乎必用:
# 看某个网卡上的XDP程序 bpftool net show dev eth0 # 看XSKMAP里的内容,确认fd是否正确 bpftool map dump id <map_id> # 看BPF程序的运行状态 bpftool prog show另外,xdp-tools带了一个xdpdump工具,可以在XDP层抓包,它不需要写任何代码就能看到网卡进入XDP hook的包。如果你自己写的程序收不到包,先用xdpdump抓一下:
xdpdump -i eth0 -w /tmp/out.pcap能抓到说明驱动XDP路径没问题,那就是我们程序的问题;抓不到说明流量压根没到XDP hook,或者驱动不支持。这个对照排查法能省掉大量瞎猜的时间。
4.5 性能上几个值得留意的方向
如果你把它用在真实的高性能场景,以下几点每一条都能带来肉眼可见的收益:
从copy模式切到zero-copy。默认bind_flags=0是copy模式,驱动会先把包拷贝到UMEM,再让用户态读。zero-copy需要设置XDP_ZEROCOPY,驱动直接把包DMA进UMEM,省掉一次拷贝,CPU占用会明显下降。但前提是网卡驱动支持。
使用NEED_WAKEUP。设置XDP_USE_NEED_WAKEUP之后,内核不会频繁轮询你的Ring,而是在需要用户态配合时唤醒你。这能减少无谓的软中断开销,但代价是用户态必须正确判断什么时候该踢内核。典型例子:填充Fill Ring之后,要检查
xsk_ring_prod__needs_wakeup(&fill_r),需要就poll一次。批量消费和批量归还。一次处理64个包、256个包,比一次处理一个包吞吐高得多,因为Ring操作和系统调用都被摊薄了。
UMEM分配尽量靠近网卡所在的NUMA node。如果网卡在node 0,UMEM就分配在node 0的内存上,否则跨NUMA访问会有明显的性能损耗。用libnuma或者hugepage分配都可以。
IRQ affinity和RSS队列绑定。把网卡队列的中断绑到固定CPU核,用户态收包线程也绑到同一个核,减少cache miss。这个优化在DPDK里是标配,AF_XDP同样适用。
busy-poll。如果接受持续占用CPU的模型,可以开SO_BUSY_POLL,让用户态在没有新包时短暂自旋而不是立刻sleep,延迟能进一步降低,但代价是CPU占用上升。
我自己在xsk这个方向折腾了不短时间,最大的体会是:AF_XDP真正难的不是某个API不会用,而是你要始终清楚“这块chunk当前归谁所有”。是归内核还是归用户态?是在Fill Ring、RX Ring、TX Ring还是自己的空闲池里?一旦这个所有权模型在你脑子里建立起来,剩下的无非是围绕Ring做正确的生产消费动作。这个模型想通了,后面加队列、做发送路径、上性能优化,都是顺着同一根藤往上摸。希望这篇文章能帮你少走一点弯路。