一、背景AF_XDPAddress Family XDP是一种XDP的网络socket(简称XSK)的协议族名称类似于AF_INET。作为eBPF生态的关键组件AF_XDP通过创新的内存共享机制实现了真正的零拷贝数据传输。其核心突破在于内核态与用户态的直接内存映射数据包无需经过传统协议栈的层层处理即可直达应用程序。上一篇博客介绍了eBPF和XDP技术并提及了其与传统TCP/IP协议栈的路径差异。在此基础之上本篇博客我将结合AF_XDP实现一条不同于传统TCP/IP协议栈send()的主从同步路径装包 - 零拷贝推送 - UDP广播 - DMA直取。如果你已经熟悉KV存储的原理本篇博客也可以看作是eBPF AF_XDP的一次实践参考。前置知识AF_XDP实现零拷贝收发的方法四个RingUMEMUMEMUser-Space Memory和4 Rings——fill ring、TX ring、RX ring、completion ring是mmap()映射出来的用户态-内核态共享内存是沟通网卡与用户程序的关键。下图展示了4 Rings和UMEM的使用流程之后我们主从同步的实现就是基于这个流程原理图。用户程序分配一块和内核态共享的UMEM内存池内存池由chunks来组织并且请求内核程序分配四个ring用于传递索引。用户态将空闲的chunks索引推入fill ring内核程序得以了解哪些chunks空闲可写。内核程序驱动DMA将数据直写进UMEM的空闲chunks并且将写好的chunks索引推入RX ring用户程序得以了解哪些chunks已经填好数据。用户程序将需要发送的chunks索引推入TX ring内核程序得以了解哪些chunks需要发送并直接驱动网卡发送chunks内核程序将已经发送完毕的chunks索引填入completion ring用户程序得以回收利用。总结: AF_XDP的本质是越过内核协议栈网卡和用户态程序的直接对话。二、用 eBPF 实现主从同步初始化kv程序调posix_memalign在用户态分配一块准备用于共享的内存区域。if(posix_memalign(x-umem_buf,(size_t)getpagesize(),(size_t)KVS_EBPF_NUM_FRAMES*KVS_EBPF_FRAME_SIZE)!0){...}在刚才分配的内存区域上创建UMEM这会初始化 fill ring 和 complete ring。顺便切割UMEM内存区域为chunks。xsk_umem__create(x-umem,x-umem_buf,...x-fq,x-cq,umem_cfg);...// 切割为chunksfor(i0;iKVS_EBPF_NUM_FRAMES;i)x-chunks[x-nfree](uint64_t)i*KVS_EBPF_FRAME_SIZE;创建AF_XDP socket并且将网卡的一个硬件队列queue_id指定用于收发数据绑定到此socket并指派好用于控制数据收发的RX ring和TX ringxsk_socket__create(x-xsk,iface,queue_id,x-umem,x-rx,x-tx,xsk_cfg);2.1. 主机kv存储主从同步中我们可以将同步的行为两端简化为主机和从机两端——从机单方面接收来自主机的数据同步。所以我们使用 AF_XDP 的时候对于主机只需要考虑数据推送组包、广播。先将检查空闲的 chunks将数据写入 UMEM 当中然后用户态kv程序将写好的 chunks 索引推送至 TX ring由于我们并没有在主机中加载 eBPF 程序所以对内核需要我们主动通知kick机制才能利用 TX 发送数据。最后用户态kv程序通过 Completion ring回收chunks。为此我设计了代表主机行为的xdp_broadcast()这个函数voidxdp_broadcast(kvs_token_t*tokens,intcount){uint64_taddr;uint32_tidx;char*pkt;intpacket_len;if(g_ebpf.mode!KVS_EBPF_MODE_TX||!tokens||count0)return;if(g_is_loading)return;recycle_cq(g_ebpf.fd);// 回收completion ringif(!g_ebpf.nfree)return;addrg_ebpf.frames[--g_ebpf.nfree];pktxsk_umem__get_data(g_ebpf.umem_buf,addr);packet_lenencapsulate_packet(pkt,(int)KVS_EBPF_FRAME_SIZE,tokens,count,g_ebpf.pkt,g_ebpf.mtu);if(packet_len0){if(g_ebpf.nfreeKVS_EBPF_NUM_FRAMES)g_ebpf.frames[g_ebpf.nfree]addr;return;}// tx ring满回退if(xsk_ring_prod__reserve(g_ebpf.tx,1,idx)!1){recycle_cq(g_ebpf.fd);if(xsk_ring_prod__reserve(g_ebpf.tx,1,idx)!1){if(g_ebpf.nfreeKVS_EBPF_NUM_FRAMES)g_ebpf.frames[g_ebpf.nfree]addr;return;}}xsk_ring_prod__tx_desc(g_ebpf.tx,idx)-addraddr;xsk_ring_prod__tx_desc(g_ebpf.tx,idx)-len(uint32_t)packet_len;xsk_ring_prod__submit(g_ebpf.tx,1);//提交tx给内核if(xsk_ring_prod__needs_wakeup(g_ebpf.tx)g_ebpf.fd0)sendto(g_ebpf.fd,NULL,0,MSG_DONTWAIT,NULL,0);}//空sendto实现kick机制2.2. 从机2.2.1. 内核态程序eBPF XDP内核态程序在从机上是作为独立进程启动的内核态程序用于驱动从机XDP进行主机数据包收包过滤与重定向至map中DMA直取进用户kv程序不经过协议栈// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause#includelinux/bpf.h#includelinux/if_ether.h#includelinux/in.h#includelinux/ip.h#includelinux/udp.h#includebpf/bpf_helpers.h#includebpf/bpf_endian.hconstvolatile__u16 kvs_udp_port8888;constvolatile__u32 kvs_queue_id0;struct{__uint(type,BPF_MAP_TYPE_XSKMAP);__uint(max_entries,64);__uint(key_size,sizeof(int));__uint(value_size,sizeof(int));}xsks_mapSEC(.maps);SEC(xdp)intebpf_sock_prog(structxdp_md*ctx){void*data_end(void*)(long)ctx-data_end;void*data(void*)(long)ctx-data;structethhdr*ethdata;structiphdr*ip;structudphdr*udp;__u32 qid;if((void*)(eth1)data_end)returnXDP_PASS;if(eth-h_proto!bpf_htons(ETH_P_IP))returnXDP_PASS;ip(void*)(eth1);if((void*)(ip1)data_end)returnXDP_PASS;if(ip-protocol!IPPROTO_UDP)returnXDP_PASS;if(ip-ihl!5)returnXDP_PASS;udp(void*)(ip1);if((void*)(udp1)data_end)returnXDP_PASS;if(udp-dest!bpf_htons(kvs_udp_port))returnXDP_PASS;qidctx-rx_queue_index;if(qid!kvs_queue_id)returnXDP_PASS;returnbpf_redirect_map(xsks_map,qid,0);}charLICENSE[]SEC(license)Dual BSD/GPL;2.2.2. 用户态程序初始化部分和之前大部分一样但是从机的用户kv程序启动时需要向 fill ring 中预先填充大量空闲的chunks索引。主机接收的数据已经进入从机的UMEM中用户态程序从RX ring 取出有数据的chunks索引找到数据包、并对这些包解封装完成解封装后向业务层抛入指令从fill ring回收空闲索引。代表从机行为的函数xdp_receive()intxdp_receive(intfd){uint32_trx_idx,fq_idx,n,filled,i;unsignedintwant;intprocessed;(void)fd;nxsk_ring_cons__peek(g_ebpf.rx,KVS_EBPF_RING_SIZE,rx_idx);//查看RX ring是否有新包if(!n)return0;for(i0;in;i){conststructxdp_desc*dxsk_ring_cons__rx_desc(g_ebpf.rx,rx_idxi);char*pktxsk_umem__get_data(g_ebpf.umem_buf,d-addr);char*payload;uint8_tsrc_mac[ETH_ALEN];uint32_tsrc_ip;intplen;if(pktdecapsulate_packet(pkt,d-len,payload,plen,src_ip,src_mac)0accept_from_master(src_ip,src_mac)){processed0;kvs_aof_replay(payload,plen,processed);//复用业务层AOF逻辑处理RESP命令}if(g_ebpf.nfreeKVS_EBPF_NUM_CHUNKS)g_ebpf.chunks[g_ebpf.nfree]d-addr;}xsk_ring_cons__release(g_ebpf.rx,n);// RX新包处理完毕标记消费wantn;if(g_ebpf.nfreewant)wantg_ebpf.nfree;filledxsk_ring_prod__reserve(g_ebpf.fq,want,fq_idx);for(i0;ifilled;i){if(!g_ebpf.nfree){g_ebpf.fq.cached_prod-(filled-i);filledi;break;}*xsk_ring_prod__fill_addr(g_ebpf.fq,fq_idxi)g_ebpf.chunks[--g_ebpf.nfree];//从fill ring中取chunks索引}if(filled)//提交空recvfrom kick唤醒xsk_ring_prod__submit(g_ebpf.fq,filled);if(xsk_ring_prod__needs_wakeup(g_ebpf.fq)g_ebpf.fd0)recvfrom(g_ebpf.fd,NULL,0,MSG_DONTWAIT,NULL,NULL);return0;}三、性能参考由于测试环境局限虚拟机只能使用veth且无法启用 XDP 程序运行的Native modeeBPF 测出来有一定劣势以下数据仅供参考测试模式SET QPSGET QPS什么都不开启13,64313,201传统 TCP 开启12,61513,700eBPF 开启12,68313,544开启数据同步无论是传统TCP还是eBPF后由于增加了广播复制的开销SET写操作的性能相比未开启时出现了轻微的合理衰减而GET读性能则基本保持稳定。