eBPF例程—

XDP

传送门：XDP技术——Linux网络处理的高速公路

XDP（eXpress Data Path，快速数据面）是Linux内核中一种面向高性能网络处理的架构，提供了位于网络协议栈最底层的可编程数据包处理能力。XDP在Linux内核中提供了eBPF钩子，通过在网络接口驱动层挂载eBPF程序，数据包在进入网卡接收路径的最早阶段即可被即时处理，极大地降低了数据包在内核中的处理延迟，提升了整体网络吞吐效率。该机制支持多种灵活操作，包括数据包过滤、重定向、修改和丢弃，适用于防火墙、负载均衡、流量监控等多种场景。

XDP只位于接收路径上，在数据包从网卡进入Linux内核的路径上，数据包首先到达XDP，此时还没有创建完整的skb_buff，只使用一个轻量化的xdp_buff表示。XDP对网络数据包的处理，分为5种情况：

XDP_DROP：丢弃数据包，常用于防攻击，将攻击报文丢弃；
XDP_TX：通过当前网卡将数据包直接发送出去；
XDP_REDIRECT：将数据包重定向至另一个网卡发送出去，或重定向到XDP socket（用户态可以通过XDP socket接收这个数据包）；
XDP_PASS：不对数据包做特殊处理，直接交付到内核网络栈；
XDP_ABORT：数据包因出错、异常而丢包，与XDP_DROP不同的是，XDP_ABORT会做异常统计。

数据包大小记录

eBPF程序：xdppass.bpf.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


#include "vmlinux.h"
#include <bpf/bpf_helpers.h>

// 指示该eBPF程序的程序类型是XDP
// ctx是上下文指针，包含数据包的元信息
SEC("xdp")
int xdp_pass(struct xdp_md *ctx)
{
    // 获取数据包缓冲区的开始地址
    void *data = (void *)(long)ctx->data;
    // 获取数据包缓冲区的结束地址
    void *data_end = (void *)(long)ctx->data_end;
    // 通过计算两者差值，得到当前数据包以字节为单位的大小
    int pkt_sz = data_end - data;

    // 将数据包大小输出
    bpf_printk("packet size: %d", pkt_sz);
    // XDP程序的返回值，表示允许内核网络栈继续处理该数据包
    return XDP_PASS;
}

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

用户态程序：xdppass.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63


#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <signal.h>
#include <sys/resource.h>
#include <bpf/libbpf.h>
#include "xdppass.skel.h"

// 标记程序是否收到退出信号
static volatile sig_atomic_t stop;

// 信号处理的回调函数
static void handle_sigint(int sig)
{
    stop = 1;
}

int main(int argc, char **argv)
{
    struct xdppass_bpf *skel = NULL;
    int ifindex = 0;

    // 从命令行参数读取要挂载XDP程序的接口索引
    if (argc > 1)
        ifindex = atoi(argv[1]);

    // 打开并加载eBPF程序
    skel = xdppass_bpf__open_and_load();
    if (!skel) {
        fprintf(stderr, "Failed to open and load BPF skeleton\n");
        return 1;
    }

    // 将eBPF程序xdp_pass附加到指定接口索引的网卡上
    // 返回的bpf_link对象保存在skeleton的links.xdp_pass中，方便后续自动卸载
    // 若不进行保存，需调用bpf_link__destroy(link)函数来释放资源并断开关联
    skel->links.xdp_pass = bpf_program__attach_xdp(skel->progs.xdp_pass, ifindex);
    if (!skel->links.xdp_pass) {
        fprintf(stderr, "Failed to attach XDP program to ifindex %d\n", ifindex);
        goto cleanup;
    }

    // 注册SIGINT信号处理函数
    if (signal(SIGINT, handle_sigint) == SIG_ERR) {
        fprintf(stderr, "Failed to set signal handler\n");
        goto cleanup;
    }

    printf("Running... Press Ctrl+C to stop.\n");

    // 主循环，使程序持续运行
    while (!stop) {
        fprintf(stderr, ".");
        sleep(1);
    }

cleanup:
    // 释放、清理eBPF环境
    xdppass_bpf__destroy(skel);
    return 0;
}

运行eBPF程序

libbpf-bootstrap中的XDP示例程序采用Rust语言编写，本文将该代码移植为C语言版本，因此在example/c/文件夹下新建eBPF程序xdppass.bpf.c、用户态程序xdppass.c。同时需要在Makefile文件中添加该XDP程序xdppass。

此时即可编译该XDP程序，启动程序时需指定要挂载XDP程序的网络接口索引，网络接口索引可通过ip link show命令查看。

1
2
3


make xdppass
ip link show
sudo ./xdppass 1

eBPF将数据包信息输出到日志，可以通过以下命令实时查看。

1

sudo cat /sys/kernel/debug/tracing/trace_pipe

数据包解析以太网头

eBPF程序：xdppass.bpf.c

在eBPF/XDP程序里，解析数据包字段的本质就是从data转换指针、加偏移量，随后读取值。eBPF程序最终需要被LLVM/Clang编译成BPF 字节码，内核eBPF verifier只接受单个入口点的BPF程序（即不支持真正的多函数调用）。如果编写了非内联函数，编译器会生成额外的子函数，从而导致验证器拒绝加载。因此需要声明辅助函数parse_ethhdr()为内联，编译器就会把函数的代码直接替换到调用处，不生成真正的函数调用指令。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37


#include "vmlinux.h"
#include <bpf/bpf_helpers.h>

// 解析以太网头，返回值用于指示解析成功（0）或失败（-1）
static __always_inline int parse_ethhdr(void *data, void *data_end, struct ethhdr **ethhdr)
{
    // 边界检查：确保数据包长度足够包含一个完整的以太网头
    if (data + sizeof(struct ethhdr) > data_end)
        return -1;

    // 将解析出的以太网头指针返回给调用者
    *ethhdr = data;
    return 0;
}

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

    // 解析出以太网头
    if (parse_ethhdr(data, data_end, &eth) < 0)
        return XDP_PASS;

    // 输出源MAC地址与目的MAC地址
    bpf_printk("Src MAC: %02x:%02x:%02x:%02x:%02x:%02x, Dst MAC: %02x:%02x:%02x:%02x:%02x:%02x",
        eth->h_source[0], eth->h_source[1], eth->h_source[2],
        eth->h_source[3], eth->h_source[4], eth->h_source[5],
        eth->h_dest[0], eth->h_dest[1], eth->h_dest[2],
        eth->h_dest[3], eth->h_dest[4], eth->h_dest[5]);

    return XDP_PASS;
}

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

用户态程序：xdppass.c

与上一节的用户态进程完全一致。

运行eBPF程序

编译该XDP程序，启动程序时同样需指定要挂载XDP程序的网络接口索引。

1
2


make xdppass
sudo ./xdppass 2

eBPF将数据包信息输出到日志，可以通过以下命令实时查看。

1

sudo cat /sys/kernel/debug/tracing/trace_pipe

数据包重定向至其它端口

eBPF程序：xdpredirect.bpf.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27


#include "vmlinux.h"
#include <bpf/bpf_helpers.h>

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

// BPF_MAP_TYPE_DEVMAP类型的BPF映射专用于XDP的重定向操作
// 该哈希表使用键来查找对网络设备的引用
// 注意键可以是任意值，不局限于ifindex
struct {
    __uint(type, BPF_MAP_TYPE_DEVMAP_HASH);
    __uint(max_entries, 4);
    __type(key, __u32);
    __type(value, __u32);
} tx_port_map SEC(".maps");

SEC("xdp")
int xdp_redirect_prog(struct xdp_md *ctx)
{
    // 获取数据包进入的网络接口索引
    __u32 ingress_ifindex = ctx->ingress_ifindex;
    
    // 根据ingress_ifindex作为键，在tx_port_map中查找对应的出口接口索引
    // 查找到后,返回一个特殊的XDP动作值，重定向到对应的出口网口
    if (bpf_map_lookup_elem(&tx_port_map, &ingress_ifindex))
        return bpf_redirect_map(&tx_port_map, ingress_ifindex, 0);
    return XDP_PASS;
}

除了BPF_MAP_TYPE_DEVMAP_HASH，同样也支持使用BPF_MAP_TYPE_DEVMAP类型的BPF映射，该映射基于数组实现，因此传入的key值必须小于设置的max_entries，即不能越界。

用户态进程：xdpredirect.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76


#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <bpf/libbpf.h>
#include "xdpredirect.skel.h"

// 标记程序是否收到退出信号
static volatile sig_atomic_t stop;

// 信号处理的回调函数
static void handle_sigint(int sig)
{
    stop = 1;
}

int main(int argc, char **argv)
{
    struct xdpredirect_bpf *skel = NULL;
    int ingress_ifindex, egress_ifindex;
    int err;

    if (argc < 3) {
        fprintf(stderr, "Usage: %s <ingress_ifindex> <egress_ifindex>\n", argv[0]);
        return 1;
    }
    // 从命令行参数读取要挂载XDP程序的接口索引
    ingress_ifindex = atoi(argv[1]);
    // 从命令行参数读取重定向出端口的接口索引
    egress_ifindex = atoi(argv[2]);

    // 打开并加载eBPF程序
    skel = xdpredirect_bpf__open_and_load();
    if (!skel) {
        fprintf(stderr, "Failed to open and load BPF object\n");
        return 1;
    }

    // 设置BPF映射tx_port_map，即设置端口重定向信息
    __u32 key = ingress_ifindex;
    __u32 ifindex = egress_ifindex;
    err = bpf_map__update_elem(
        skel->maps.tx_port_map,
        &key, sizeof(key),
        &ifindex, sizeof(ifindex),
        0
    );
    if (err) {
        fprintf(stderr, "Failed to update devmap: %s\n", strerror(errno));
        goto cleanup;
    }

    // 将eBPF程序xdp_redirect_prog附加到指定接口索引的网卡上
    skel->links.xdp_redirect_prog = bpf_program__attach_xdp(skel->progs.xdp_redirect_prog, ingress_ifindex);
    if (!skel->links.xdp_redirect_prog) {
        fprintf(stderr, "Failed to attach XDP program to ifindex %d\n", ingress_ifindex);
        goto cleanup;
    }

    printf("Redirecting packets from ifindex %d to ifindex %d\n", ingress_ifindex, egress_ifindex);
    printf("Press Ctrl+C to stop\n");

    // 注册SIGINT信号处理函数
    signal(SIGINT, handle_sigint);

    // 主循环，使程序持续运行
    while (!stop) {
        sleep(1);
    }

cleanup:
    // 释放、清理eBPF环境
    xdpredirect_bpf__destroy(skel);
    return 0;
}

运行eBPF程序

测试拓扑如下：流量源向运行XDP的主机发送流量，XDP将数据包从网络端口2重定向至网络端口3，因此与之相连的监测主机通过Wireshark能够抓取到流量。

1

[流量源 <if=2>]  ──► [<if=2> 运行XDP的主机 <if=3>] ──► [<if=2> 监测主机]

在Makefile文件中添加该XDP程序xdpredirect。

编译该XDP程序，启动程序时输入参数指定附加的网络接口、重定向的目标网络接口。

1
2


make xdpredirect
sudo ./xdpredirect 2 3

在监测主机启动Wireshark，观测网络端口2接收到的流量。

1

sudo wireshark

数据包重定向至用户态

AF_XDP

通过XDP程序的XDP_REDIRECT动作，程序可以将入口帧重定向到用户态。在这个过程中，AF_XDP套接字使XDP程序可以将帧重定向到用户空间应用程序中的内存缓冲区。

传送门：AF_XDP - eBPF指南

UMEM（User Memory）是AF_XDP中用户态和内核态共享的连续内存区域，用于高性能数据包收发。用户态通过分配一块较大的物理连续内存，并将其注册到内核，形成UMEM区域。内核网络驱动收到的数据包可以直接DMA到UMEM中的buffer，用户态也可以直接在UMEM中构造要发送的数据包，实现真正的零拷贝收发，显著降低延迟和CPU开销。UMEM的运作需要四个环形队列，这些环形队列只传递buffer的索引，不传递数据本身，从而高效地协调用户态和内核态对UMEM的使用，注意这些环形队列都是都是单生产者单消费者模型。

Ring 名称	功能说明	生产者	消费者
Fill Ring	用户态把空闲的buffer地址交给内核，用于接收数据	用户态	内核
Rx Ring	内核收到数据后，把已填充数据的buffer地址交给用户态	内核	用户态
Tx Ring	用户态要发送数据时，把待发送的buffer地址交给内核	用户态	内核
Completion Ring	内核发送完成后，把已发送的buffer地址通知给用户态	内核	用户态

数据接收到流程如下（数据包发送的流程同理）：

用户态先把可用的buffer地址放进Fill Ring；
内核从Fill Ring取出可用的buffer地址，把数据包DMA到对应的buffer；
内核把填充数据的buffer地址放进Rx Ring；
用户态从Rx Ring取出buffer地址，处理数据包；
用户态处理完该buffer中的数据包后，把该buffer的地址重新放回Fill Ring。

eBPF程序：xdpredirect.bpf.c

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27


#include "vmlinux.h"
#include <bpf/bpf_helpers.h>

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

// BPF_MAP_TYPE_XSKMAP类型的BPF映射常用于关联RX队列和AF_XDP Socket
// 键通常是RX队列的编号、值为Socket的文件描述符
struct {
    __uint(type, BPF_MAP_TYPE_XSKMAP);
    __uint(max_entries, 1);
    __type(key, __u32);
    __type(value, __u32);
} xsks_map SEC(".maps");

SEC("xdp")
int xdp_redirect_prog(struct xdp_md *ctx) {
    // 获取数据包所在的RX队列索引
    int rx_queue_index = ctx->rx_queue_index;
    // 在xsks_map映射中查询该队列是否有绑定的AF_XDP Socket
    if (bpf_map_lookup_elem(&xsks_map, &rx_queue_index))
    {
        // 调用bpf_redirect_map将数据包重定向到对应的用户态Socket
        return bpf_redirect_map(&xsks_map, rx_queue_index, 0);
    }
    // 如果没有找到对应Socket，则放行数据包
    return XDP_PASS;
}

用户态程序：xdpredirect.c

传送门：XSK All in One

该用户态程序实现了一个基于AF_XDP（XDP Socket）的用户态数据包接收、XDP程序加载的示例，总体流程为：

创建内存区域：在用户态预分配一块连续的内存，用作AF_XDP Socket的零拷贝缓冲区；
创建UMEM对象：在分配的内存区域上创建UMEM内存区域；
加载并附加XDP eBPF程序：在指定网卡接口上安装XDP程序，进行数据包重定向；
创建XSK Socket：创建AF_XDP Socket，并绑定到指定接口和队列；
初始化Fill Ring：通过Fill Ring预先准备好充足的buffer，供内核随后接收数据；
设置BPF映射：使XDP程序能重定向到AF_XDP Socket；
主循环中轮询接收数据包：通过RX Ring读取数据包并处理，并归还缓冲区到Fill Ring；
退出处理：通过信号安全停止退出，释放资源。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300


#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <sys/mman.h>
#include <linux/if_link.h>
#include <xdp/xsk.h>
#include <bpf/libbpf.h>
#include "xdpredirect.skel.h"

#define NUM_FRAMES 4096
#define NUM_FRAMES_PER_RING (NUM_FRAMES / 2)
#define FRAME_SIZE 2048
#define UMEM_SIZE (NUM_FRAMES * FRAME_SIZE)
#define MAX_BATCH_SIZE 64 // 每次最多处理64个desc

// 标记程序是否收到退出信号
static volatile sig_atomic_t stop;

// Ctrl+C信号处理的回调函数
static void handle_sigint(int sig)
{
    // 设置标志位，通知主循环退出
    stop = 1;
}

// 为随后的UMEM内存创建，预先分配好所需的内存
void *create_umem_area()
{
    // UMEM是用户空间存放数据包的地方，是预申请好的连续的一整块内存
    // mmap则能够分配一块连续内存区域，大小为UMEM_SIZE
    // PROT_READ | PROT_WRITE表示区域可读可写
    void *area = mmap(NULL, UMEM_SIZE,
                      PROT_READ | PROT_WRITE,
                      MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE,
                      -1, 0);
    if (area == MAP_FAILED)
    {
        perror("ERROR: mmap for UMEM failed");
        return NULL;
    }
    return area;
}

// 真正创建UMEM内存区域
struct xsk_umem *create_xsk_umem(void *umem_area,
                                 struct xsk_ring_prod *fill,
                                 struct xsk_ring_cons *comp)
{
    // UMEM内存区域的配置
    struct xsk_umem_config cfg = {
        // Fill Ring的大小
        .fill_size = NUM_FRAMES_PER_RING,
        // Completion Ring的大小
        .comp_size = NUM_FRAMES_PER_RING,
        // 每个帧的大小
        .frame_size = FRAME_SIZE,
        // Headroom的大小，此处不设置表示无headroom
        .frame_headroom = 0,
        .flags = 0,
    };

    // 在分配好的内存区域umem_area上创建UMEM内存umem
    struct xsk_umem *umem = NULL;
    int ret = xsk_umem__create(&umem, umem_area, UMEM_SIZE, fill, comp, &cfg);
    if (ret)
    {
        fprintf(stderr, "xsk_umem__create failed: %s\n", strerror(-ret));
        return NULL;
    }
    printf("UMEM created successfully!\n");
    return umem;
}

// 创建AF_XDP Socket
struct xsk_socket *create_xsk_socket(const char *ifname,
                                     __u32 queue_id,
                                     struct xsk_umem *umem,
                                     struct xsk_ring_cons *rx,
                                     struct xsk_ring_prod *tx)
{
    // AF_XDP Socket的配置选项
    struct xsk_socket_config xsk_cfg = {
        // 配置RX Ring的大小
        .rx_size = NUM_FRAMES_PER_RING,
        // 配置TX Ring的大小
        .tx_size = NUM_FRAMES_PER_RING,
        // 创建Socket的过程会导致自动加载默认的XDP程序，本示例需要加载自己的XDP程序
        // XSK_LIBBPF_FLAGS__INHIBIT_PROG_LOAD可禁用自动加载XDP程序
        .libxdp_flags = XSK_LIBBPF_FLAGS__INHIBIT_PROG_LOAD,
        // XDP的模式
        // XDP_FLAGS_DRV_MODE（Naive模式，驱动程序内处理）
        // XDP_FLAGS_HW_MODE（性能模式，卸载到网卡内处理）
        // XDP_FLAGS_SKB_MODE（兼容性模式，在网络栈的早期阶段处理）
        .xdp_flags = XDP_FLAGS_DRV_MODE,
        .bind_flags = 0,
    };

    // 创建Socket，并绑定到网卡和队列上
    struct xsk_socket *xsk = NULL;
    int ret = xsk_socket__create(&xsk, ifname, queue_id, umem, rx, tx, &xsk_cfg);
    if (ret)
    {
        fprintf(stderr, "xsk_socket__create failed: %s\n", strerror(-ret));
        return NULL;
    }
    printf("XSK socket created successfully on interface %s queue %u\n", ifname, queue_id);
    return xsk;
}

// 在Fill Ring中填充空闲buffer的偏移地址
// 内核需要从Fill Ring中获取空闲buffer的偏移地址，才能将接收到的数据包存在相应位置
int populate_fill_ring(struct xsk_ring_prod *fill)
{
    // 尝试一次性在Fill Ring中申请NUM_FRAMES_PER_RING个槽位
    // 等价于在UMEM内存中获取NUM_FRAMES_PER_RING个空闲buffer
    // idx由函数填入，表示从Fill Ring的第idx个槽位开始，返回值为申请获得的槽位数量
    unsigned int idx;
    int ret = xsk_ring_prod__reserve(fill, NUM_FRAMES_PER_RING, &idx);
    if (ret != (int)NUM_FRAMES_PER_RING)
    {
        fprintf(stderr, "Failed to reserve fill ring slots: %d\n", ret);
        return -1;
    }

    // 在申请到的槽位中填写空闲buffer的偏移地址
    for (unsigned int i = 0; i < NUM_FRAMES_PER_RING; i++)
    {
        // UMEM由数个FRAME_SIZE大小的buffer组成，此时UMEM中的各buffer均为空闲
        // 将第0个buffer的偏移地址填入第idx个槽位中
        // 将第1个buffer的偏移地址填入第idx+1个槽位中
        // 将第i个buffer的偏移地址填入第idx+i个槽位中
        // 而第i个buffer的偏移地址是中UMEM区域内的偏移量，即i*FRAME_SIZE
        // 因此通过读取Fill Ring的各槽位，可以映射到UMEM获得空闲的buffer
        *xsk_ring_prod__fill_addr(fill, idx + i) = i * FRAME_SIZE;
    }

    // 用户态正式提交，让内核获知新加入了空闲buffer
    xsk_ring_prod__submit(fill, NUM_FRAMES_PER_RING);

    printf("Fill ring populated with %u frames.\n", NUM_FRAMES_PER_RING);
    return 0;
}

// 接收数据包并进行处理
void process_packets(struct xsk_ring_cons *rx, struct xsk_ring_prod *fill, void *umem_area)
{
    __u32 idxs[MAX_BATCH_SIZE];
    __u64 addrs[MAX_BATCH_SIZE];
    size_t i;
    int ret;

    // 从RX Ring中一次性取出MAX_BATCH_SIZE个元素，并将其槽位索引放置于idxs中
    // 每个元素指向的buffer存储有接收到的数据包
    size_t nb_descs = xsk_ring_cons__peek(rx, MAX_BATCH_SIZE, idxs);
    if (nb_descs == 0)
        return;

    // 遍历取出每一个元素
    for (i = 0; i < nb_descs; i++)
    {
        // 根据索引获取槽位中的信息，即buffer的偏移地址与数据包的长度
        const struct xdp_desc *desc = xsk_ring_cons__rx_desc(rx, idxs[i]);
        __u64 addr = desc->addr;
        __u32 len = desc->len;

        // 根据获得的buffer偏移地址，在UMEM内存中获得buffer的实际地址
        void *data = xsk_umem__get_data(umem_area, addr);
        // 输出打印数据包的长度以及第一个字节
        printf("Packet %zu: len=%u, first byte=0x%02x\n", i, len, ((unsigned char *)data)[0]);

        // 记录该buffer的偏移地址，以便随后放入Fill Ring
        addrs[i] = addr;
    }

    // 把使用过的nb_descs个buffer重新放回Fill Ring
    // 首先在Fill Ring中申请nb_descs个槽位，而idx由函数填入起始索引
    unsigned int idx;
    ret = xsk_ring_prod__reserve(fill, nb_descs, &idx);
    if (ret != nb_descs)
    {
        fprintf(stderr, "fill ring reserve failed: %d\n", ret);
    }

    // 遍历每一个申请到的槽位
    for (i = 0; i < nb_descs; i++)
    {
        // 将第0个buffer的偏移地址填入第idx个槽位中
        // 将第1个buffer的偏移地址填入第idx+1个槽位中
        // 将第i个buffer的偏移地址填入第idx+i个槽位中
        *xsk_ring_prod__fill_addr(fill, idx + i) = addrs[i];
    }

    // 用户态正式提交，让内核获知新加入了空闲buffer
    xsk_ring_prod__submit(fill, nb_descs);

    // 通知内核已经消费了Rx Ring中nb_descs个元素
    xsk_ring_cons__release(rx, nb_descs);
}

int main(int argc, char **argv)
{
    // 设置需要挂载的网络端口
    int ingress_ifindex = 2;
    // 该网络端口的名称
    char *ifname = "ens160";
    // 设置需要操作的队列
    __u32 queue_id = 0;

    int ret;

    // AF_Socket所需的4个Ring Buffer
    struct xsk_ring_prod fill;
    struct xsk_ring_cons comp;
    struct xsk_ring_cons rx;
    struct xsk_ring_prod tx;

    // 在用户态创建连续的内存区域
    void *umem_area = create_umem_area();
    if (!umem_area)
    {
        return 1;
    }

    // 在准备好的内存区域上创建UMEM区域
    struct xsk_umem *umem = create_xsk_umem(umem_area, &fill, &comp);
    if (!umem)
    {
        munmap(umem_area, UMEM_SIZE);
        return 1;
    }

    // 打开并加载eBPF程序
    struct xdpredirect_bpf *skel = xdpredirect_bpf__open_and_load();
    if (!skel)
    {
        fprintf(stderr, "Failed to open and load BPF object\n");
        goto cleanup;
    }

    // 将eBPF程序xdp_redirect_prog附加到指定接口索引的网络端口上
    skel->links.xdp_redirect_prog = bpf_program__attach_xdp(skel->progs.xdp_redirect_prog, ingress_ifindex);
    if (!skel->links.xdp_redirect_prog)
    {
        fprintf(stderr, "Failed to attach XDP program to ifindex %d\n", ingress_ifindex);
        goto cleanup;
    }

    // 创建AF_XDP Socket
    struct xsk_socket *xsk = create_xsk_socket(ifname, queue_id, umem, &rx, &tx);
    if (!xsk)
    {
        goto cleanup;
    }

    // 在Fill Ring中准备充足的槽位，以供内核进行数据包接收
    if (populate_fill_ring(&fill) != 0)
    {
        goto cleanup;
    }

    // 设置BPF映射xsks_map，即设置队列中接收到的数据重定向至用户态
    __u32 value_fd = xsk_socket__fd(xsk);
    ret = bpf_map__update_elem(
        skel->maps.xsks_map,
        &queue_id, sizeof(queue_id),
        &value_fd, sizeof(value_fd),
        0);
    if (ret)
    {
        fprintf(stderr, "Failed to update devmap: %s\n", strerror(errno));
        goto cleanup;
    }

    printf("Redirecting packets from ifindex %d to user space\n", ingress_ifindex);
    printf("Press Ctrl+C to stop\n");

    // 注册SIGINT信号处理函数
    signal(SIGINT, handle_sigint);

    // 主循环，使程序持续运行
    while (!stop)
    {
        // 不断从RX Ring中获取、接收数据包
        process_packets(&rx, &fill, umem_area);
        usleep(1000);
    }

cleanup:
    // 释放、清理eBPF环境
    xdpredirect_bpf__destroy(skel);
    // 删除AF_XDP Socket
    xsk_socket__delete(xsk);
    // 释放UMEM内存区域
    xsk_umem__delete(umem);
    // 释放注册的连续内存空间
    munmap(umem_area, UMEM_SIZE);
    return 0;
}

运行eBPF程序

为使用AF_XDP Socket等，需使用以下命令安装相关软件。

1

sudo apt install libxdp-dev xdp-tools

同时为引入头文件和库，需在Makefile中添加编译链接库lxdp。

1

xdpredirect: ALL_LDFLAGS += -lxdp

测试拓扑如下：流量源向运行XDP的主机发送流量，XDP将数据包从网络端口2重定向至用户态。

1

[流量源 <if=2>]  ──► [<if=2> 运行XDP的主机]

因此在运行XDP的主机启动eBPF程序，并在流量源构造、发送数据包（例如使用Scapy）。

1
2


make xdpredirect
sudo ./xdpredirect

eBPF例程——XDP

基于libbpf编写的eBPF/XDP程序，示例一记录每个数据包的大小，示例二进行数据包以太网头的解析，示例三将数据包重定向至其它网络端口，示例四将数据包重定向至用户态。

XDP

数据包大小记录

eBPF程序：xdppass.bpf.c

用户态程序：xdppass.c

运行eBPF程序

数据包解析以太网头

eBPF程序：xdppass.bpf.c

用户态程序：xdppass.c

运行eBPF程序

数据包重定向至其它端口

eBPF程序：xdpredirect.bpf.c

用户态进程：xdpredirect.c

运行eBPF程序

数据包重定向至用户态

AF_XDP

eBPF程序：xdpredirect.bpf.c

用户态程序：xdpredirect.c

运行eBPF程序