『聚合』 【Learning eBPF-2】eBPF 的“Hello world”

前一章讲了 eBPF 为什么这么吊，不理解没关系，现在开始，我们通过一个 “Hello world” 例子，来真正入门一下。

BCC Python 框架是上手 eBPF 的最友好方式。来看。

2.1 BCC 的 Hello World

下面的程序是一段 BCC 框架的 Hello World 程序。

#!/usr/bin/python3
from bcc import BPF

program = r"""
int hello(void *ctx) {
	bpf_trace_printk("Hello World!\n");
	return 0;
}
"""

b = BPF(text=program)
syscall = b.get_syscall_fnname("execve")
b.attach_kprobe(event=syscall, fn_name="hello")

b.trace_print()

这段程序包含了两部分：

• 运行在内核态的 eBPF 程序本身（hello()）；
• 运行在用户态的，用于加载 eBPF 程序到内核空间并读取它生成的 trace 控制程序（hello.py）。

下图显示了这段代码运行时的状态。

下面来逐行解释这段代码。

第一行告诉你，这是一个 Python 程序。实际上 #!/usr/bin/python3 是指定默认的 Python 解释器。

eBPF 程序本身是 C 语言编写的。这部分代码为：

int hello(void *ctx) {
	bpf_trace_printk("Hello World!");
	return 0;
}

其中，bpf_trace_printk() 是 eBPF 辅助函数，用于打印一条消息。有关辅助函数的更多讨论，见第 5 章。

这段 eBPF 程序是以静态字符串 program 的形式被定义在 Python 脚本中，并作为参数，传递给 BPF 对象：

b = BPF(text=program)

当然，C 程序最终会由 BCC 框架负责编译执行。

eBPF 程序需要绑定到一个事件上。在这个例子中，我们选择的事件为 execve 系统调用。当有任何应用程序运行时，都会调用 execve()，从而触发我们绑定的 eBPF。然而，execve 系统调用在不同架构的 Linux 上可能会有不同的实现方式。但是，eBPF 提供了一种非常方便的方式（通过名称）来寻找当前支持的系统调用，就像这样：

syscall = b.get_syscall_fnname("execve")

现在，变量 syscall 指代了系统调用。接下来，使用一个探针 kprobe（详见第 1 章）来将 hello() 函数绑定到 execve 事件上。

b.attach_kprobe(event=syscall, fn_name="hello")

此时，eBPF 程序已经被成功加载到内核，并完成了绑定。那么，当有一个进程被执行时，将触发这段 hello() 程序，完成一条消息的打印。剩下的工作，就是去读取 trace 的输出，并打印到标准输出中。

b.trace_print()

trace_print()函数将进入无限循环，直到你键入Ctrl+C终止这段 eBPF 程序。

下面这张图显示了这段 eBPF 程序的运行原理：

根据这张图回顾一下整个流程。

1）这段 Python 程序编译了 C 代码，载入内核，并与 execve() 完成绑定。

2）当有其他进程运行时，执行 execve() 系统调用，触发 eBPF 中的 hello() 程序段，打印一行输出（在 pipe 中，后文会再次提到）。

3）用户态的程序读取这些输出，并打印到屏幕上。

2.2 运行 Hello World

运行这段程序，其结果取决于你当前的运行环境正在或即将运行的进程。

如果这段代码啥也没输出，请再起一个终端，手动执行一个程序。eBPF 将打印一行行的 Hello world 消息。

书里没有提到，但是很重要，运行 BCC 框架的 eBPF 程序，需要先安装 bcc-python 库。译者使用 REHL8-x86 操作系统，因此通过 yum 包管理器来安装： yum install -y python3-bcc.x86_64。

这里书中再次强调，eBPF 程序是立即生效的。首先是不需要重启，其次是对应用程序无侵入（已经重复很多遍了）。这是因为，eBPF 所绑定的是 execve() 系统调用，因此和应用程序没关系。即使你写了一个脚本，手动调用这个系统调用，那么，这个 eBPF 也会触发。

打印输出除了 “Hello World” 字符串以外，还有其他信息。例如，执行 execve 的进程 ID 为 5412，并使用了 bash 命令等等。Python 程序从哪里读取这个输出信息的呢？实际上，bpf_trace_printk() 辅助函数会把打印写入 /sys/kernel/debug/tracing/trace_pipe 文件中。你可以通过 cat 指令来查看（需要 root 权限）。

eBPF 程序使用这种方式打印信息，虽然简单，但却有下面两点局限性：

• 仅支持字符串类型的输出。你想传结构体类型？没门。
• trace_pipe 文件只有这一个。也就是说，所有正在运行的 eBPF 都会把输出写入到这里。难受吧！

那么，有没有一种更好的方式传递数据呢？？答案就是：eBPF 映射（maps）。

2.3 eBPF 映射：maps

映射 maps 是 eBPF 的扩展功能，它是一类可以让 eBPF 程序和用户态程序访问的数据结构。

maps 支持内核态 eBPF 之间的通信，也支持 eBPF 到用户态程序之间的通信。主要的作用包括以下几种：

• 用户空间写入需要由 eBPF 程序检索的配置信息。
• 一个 eBPF 存储状态，以供另一个 eBPF 程序（或者同一 eBPF 的后续指令）使用。
• eBPF 程序将数据写入 maps ，以供用户空间应用程序读取，从而打印结果。

eBPF maps 有很多种类型，在 uapi/linux/bpf.h 文件中可以查看，内核文档中也有相关的介绍。

通常，eBPF maps 都是键值对类型结构，但具体 key 和 value 的指代和形式又有所区别。本章，将主要介绍 hash、perf 、 ring buffer 以及 eBPF 程序数组。

诚然，eBPF maps 不止这些。

有些 map，形似数组，但其 key 小得仅有 4 字节；【array】

有些 map，如哈希表，key的种类能够包罗万象；【hash】

有些 map，便利操作，或 FIFO 列队而伺，或 FILO 作栈而生；或 LRU 行冷热数据分离，或 LPM 做最长前缀匹配；【queue、stack、lru、lpm、Bloom filter】

有些 map，特殊对象专用，拓宽网络和尾调用的技术；【sockmaps、devmaps、program array、map-of-map】

有些 map，对应CPU核心，寻求并发操作的可能性。【cpu-*】

接下来的例子，我们来看一下使用哈希表类型的 map 基本用法。

2.3.1 哈希表 map

在上一个给出的例子中，我们的 eBPF 程序绑定了 execve() 系统调用。接下来，要用哈希表 HASH 做一下改编，key用来存储用户 ID，value 用来记录某个用户下的进程执行调用 execve() 的次数。这个程序统计了不同的用户分别运行了多少个程序。

来看这个 eBPF 程序的 C 代码。

BPF_HASH(counter_table); 				// A

int hello(void *ctx) {
    u64 uid;
    u64 counter = 0;
    u64 *p;
    
    uid = bpf_get_current_uid_gid() & 0xFFFFFFFF;	// B
    p = counter_table.lookup(&uid);			// C
    if (p != 0) {					// D
        counter = *p;
    }
    counter++;						// E
    counter_table.update(&uid, &counter);		// F
    return 0;
}

代码解释：

【A】BPF_HASH() 是一个 BCC 宏声明的哈希表。

【B】bpf_get_current_uid_gid() 是一个辅助函数，用来获取当前进程的用户 ID。这个辅助函数返回值是一个 64 位的值，其中，用户 ID 存储在低 32 位（高 32 位为用户组 ID）。

【C】通过 key 查找哈希表中的 value。这里是通过 uid 查找 p。返回一个指针。

【D】如果指定的 uid ，在哈希表中存在一个 p，将哈希表中的 p 值设置给 counter；若哈希表中不存在对应 uid 的 p，counter 的值将为默认值 0。

【E】无论 counter 值为多少，在这里都对其进行自增操作。

【F】使用新的 counter 值，更新对应 uid 的哈希表。

我们仔细看一下这两行代码。首先是查找哈希表 value：

p = counter_table.lookup(&uid);

然后是更新哈希表：

counter_table.update(&uid, &counter);

你可能会有点疑问了：C 语言能这么写？结构体可以直接调用成员函数？不对吧？实际上，你是对的，C 语言确实不支持在结构体中定义这样的函数。但是，BCC 框架中的 C，实际上是一种不严格的 C。BCC 在真正执行 C 代码的编译前，会重写这些不严格的语法（实际上是通过若干个 BCC 宏来实现的）。

接下来，和前面的例子一样，将这段 C 程序声明为一个 program 字符串，然后通过 BCC 将其编译载入内核，并绑定在 execve() 系统调用上。

b = BPF(text=program)
syscall = b.get_syscall_fnname("execve")
b.attach_kprobe(event=syscall, fn_name="hello")

但这次，还需要一些额外的工作，在用户态中读取哈希表的内容。

while true:						# A
    sleep(2)
    s = ""
    for k, v in b["counter_table"].items():		# B
        s += f"ID {k.value}: {v.value}\t"
    print(s)

代码解释：

【A】无限循环。每隔 2s 打印输出。

【B】BCC 框架会自动创建一个 Python 对象来指代哈希表。这个循环将会遍历 eBPF 定义的 counter_table 哈希表中的所有键值对，然后完成打印。

运行这段程序，你需要两个终端。终端 1 运行 eBPF 程序，终端 2 运行指令。

可以看到，每 2s 输出一行。我们关注最后一行的两个键值对：

• key = 501, value = 5
• key = 0, value = 2

在第二个终端里，作者的用户 ID 为 501。当运行 ls 命令时，值为 501 的 uid 计数器自增 1。而当运行 sudo ls 时，发生了两次 execve()。第一次是在 501 用户下的 sudo 命令，第二次是在 root 用户下的 ls 命令。

这个例子给出了使用哈希表 map 从内核态向用户态传递数据的方式。当然，你也可以使用数组类型的 map 来实现这个功能（因为 key 为整数）。

Linux 内核中存在一个 名为perf 的子系统，也可以传递内核态数据到用户空间，eBPF 刚好也支持这种方式。我们来看一下。

2.3.2 Perf 和 Ring buffer map

在这一小节中，我们再来看一种更复杂的 “Hello World” BCC 程序，它使用了 Perf 环形缓冲区，用来向用户态传递自定义的数据结构。

环形缓冲区：内核 5.8 版本才引入的结构，在这之前为普通的基于共享内存的缓冲区。实际上 perf 环形缓冲区更有优势，具体可以参考 Andrii Nakryiko 的这篇博客： https://nakryiko.com/posts/bpf-ringbuf/

那么，问题来了，什么是环形缓冲区？

环形缓冲区 是一种数据结构，它不是 eBPF 独有的。环形缓冲区实际上是一段内存空间，其空间中的地址在逻辑上首尾相连成环。环形缓冲区包括两个工作指针，一个负责读，一个负责写，二者同向移动。写指针指向的位置就是下个数据被写入的位置（数据可以任意长度，其长度信息包含在数据头中），同理，读指针指向的位置就是下一个需要读取的数据开头（根据数据头中的长度，控制读指针移动距离）。

下图直观的展示了环形缓冲区的样貌。

读指针和写指针始终朝着一个方向运动。若在某一时刻，读指针追上了写指针，则说明缓冲区没数据可读了。相反，若写指针追上了读指针，则说明缓冲区没空间可写了，那么此时，需要写入的数据就会被丢弃（丢弃计数器会增加）。如果你控制的好，读写指针以相同的速率运动，始终不会相遇，那么恭喜你，你便拥有了一个无限大的循环缓冲区可以使用。

了解了环形缓冲区的概念后，我们再来改进一下之前绑定到 execve() 的 eBPF 程序，来实时打印运行进程的简单信息。

BPF_PERF_OUTPUT(output);							// A

struct data_t {									// B
    int pid;
    int uid;
    char command[16];
    char message[12];
};

int hello(void *ctx) {
    struct data_t data = {};							// C
    char message[12] = "Hello World";

    dara.pid = bpf_get_current_pid_tgid() >> 32;				// D
    data.uid = bpf_get_current_uid_gid() & 0xFFFFFFFF;				// E

    bpf_get_current_comm(&data.command, sizeof(data.command));			// F
    bpf_probe_read_kernel(&data.message, sizeof(data.message), message);	// G

    output.perf_submit(ctx, &data, sizeof(data));				// H
    return 0;
}

代码解释：

【A】BCC 框架声明了一个宏定义 BPF_PERF_OUTPUT ，用来创建一个 perf 映射区域，以便内核态可以向用户态传递消息。这里定义为 output。

【B】每次 hello() 运行之时，都会填充一个结构体来存储关键字段。这是结构体定义，包括进程 ID、用户 ID、当前运行指令名称以及 message 信息。

【C】data 被定义为本地变量，message 被赋值为 "Hello world" 字符串。

【D】bpf_get_current_pid_tgid() ，辅助函数，用于获取触发当前 eBPF 程序的进程 ID。该函数返回一个 64 位的值，高 32 位是进程 ID（低 32 位为线程组 ID，对于单线程的进程，同为进程 ID）。

【E】bpf_get_current_uid_gid()，辅助函数，前文介绍过，用于获取用户 ID。

【F】bpf_get_current_comm()，辅助函数，用于获取当前执行的指令名称。

在 C 语言中，你不可以直接使用 "=" 赋值字符串，你需要传入一个待写入字符串的地址。

【G】这个例子中，message = "Hello World"，bpf_probe_read_kernel() 辅助函数会将它拷贝到 data 结构体的对应位置。

【H】此时，data 结构体中已经填充了 pid、uid、command[] 以及 message[]。这里调用 output.perf_submit() 将 data 结构体提交到 map 中。

接下来，与第一个 “Hello World” 程序类似，这一段 C 程序将被定义为一段字符串 program，下面是 Python 代码。

b = BPF(text=program)							# A
syscall = b.get_syscall_fnname("execve")
b.attach_kprobe(event=syscall, fn_name="hello")

def print_event(cpu, data, size):					# B
    data = b["output"].event(data)
    print(f"{data.pid} {data.uid} {data.command.decode()} {data.message.decode()}")

b["output"].open_perf_buffer(print_event)				# C
while True:								![image](uploading...)# D
    b.perf_buffer_poll()

代码解释：

【A】编译、加载、绑定 eBPF C程序。不再赘述。

【B】print_event()是一个回调函数，用于将 data 的内容打印到屏幕上。BCC 已经做了很多复繁重的工作，因此你只需要简单的 b["output"].event() 来从内核态 map 中获取数据。

【C】b["output"].open_perf_buffer() 用于打开 perf ring buffer。该函数接收 print_event 参数，是将其声明为一个回调。即，当 perf ring buffer 中有数据时，触发回调，打印这个数据。

【D】无限循环，调用 perf_buffer_poll() 拉取 perf ring buffer 内容。

运行这段程序，你能得到以下输出：

和以前一样，你可能需要另起一个终端，执行命令，来验证你的程序。

这个例子和最初的 “Hello World” 程序最大的不同就是，我们不再使用有限的 trace pipe 传递数据，而使用了 perf ring buffer。执行原理通先前也有了些许区别，如下图所示。

通过环形缓冲区传递数据会不会仍然使用了 trace pipe 呢？你可以运行一下命令检验一下：

cat /sys/kernel/debug/tracing/trace_pipe

这个例子还给出了一些辅助函数的使用示例，第 7 章我们会更加详细讨论。

这些辅助函数主要辅助于检索事件触发时的上下文信息，辅助函数的合理使用，能够极大提高性能。因为这些上下文信息产生于内核、收集于内核、最后仍然应用于内核。这减少了很多不必要的内核态和用户态的切换。

2.3.3 函数调用

能否在 eBPF 程序的 C 代码中将重复代码块抽象成函数，并执行函数调用？这个看似简单的动作，在早先的 eBPF版本中并不支持（仅支持调用辅助函数）。如果你非要调用自定义函数，有没有方法呢？当然有，你可以将其声明为内联函数。就像下面这样。

static __always_inline void my_function(void *ctx, int val)

__always_inline 修饰符会在编译期间，对当前函数进行优化。

那么，普通函数和内联函数有什么区别呢？我们可以用一张图来加以说明：

对于普通函数（上图左侧），当函数 F 被调用时，顺序执行的指令会跳转到函数 F 的起始地址（函数调用实际上就是地址切换），执行 F 的指令序列。当函数 F 执行完毕，return 语句会再次跳转回函数 F 调用前的位置，接续进行。

对于内联函数（上图右侧），并没有地址跳转，因为编译时这个函数会完全编译到顺序执行的指令序列中。

但是，内联函数是有局限性的。如果你在多个位置调用了同一个内联函数，那么在最终的可执行文件中，必然会产生该函数的多个指令副本。（这也是为啥通过 kprobe 探针无法绑定到内核内联函数的原因，我们第 7 章再来看这个问题）

直到 4.16 版本的内核以及 6.0 版本的 LLVM，eBPF 中内联函数的限制才被取消。因此，在这之后，你可以放心地定义函数调用（但必须是 static 的）。

2.3.4 尾调用

尾调用是什么？引用 ebpf.io 网站的一句介绍：“尾调用允许 eBPF 调用和执行另一个 eBPF 并替换执行上下文，类似于一个进程执行 execve() 系统调用的方式。”

换句话说，尾调用之后，函数不会再返回给调用者了。

Tail calls can call and execute another eBPF program and replace the execution context, similar to how the execve() system call operates for regular processes.

尾调用也不是 eBPF 独有的思想。eBPF 为什么要使用尾调用呢？这是因为，eBPF 的运行栈太有限了（仅有 512 字节），在递归调用函数时（实际上是向运行栈中一节一节地添加栈帧），很容易导致栈溢出。而尾调用恰恰允许在不增加堆栈的情况下，调用一系列函数。这是非常有效且实用的。

你可以使用下面的辅助函数来增加一个尾调用：

long bpf_tail_call(void *ctx, struct bpf_map *prog_array_map, u32 index)

其三个参数的含义分别是：

• ctx 向被调用者传递当前 eBPF 程序的上下文信息。
• prog_array_map 是一个程序数组（BPF_MAP_TYPE_PROG_ARRAY）类型的 eBPF map，用于记录一组 eBPF 程序的文件描述符。
• index 为程序数组中需要调用的 eBPF 程序索引。

这个辅助函数一旦运行成功，就不会返回了。因为调用者的运行栈已经被下一个 eBPF 程序的运行栈替换了。当然，如果指定 index 的 eBPF 程序不存在，该辅助函数也会执行失败，此时调用者继续执行。无事发生。

需要注意的是，若使用尾调用，所有需要执行的 eBPF 程序需要同时加载到内核中。而且还需要设置好程序数组 map。

使用 BCC 框架如何进行尾调用呢？可以使用下面简单的方式：

prog_array_map.call(ctx, index)

在编译它之前，BCC 框架会自动将其转换为标准的尾调用辅助函数：

bpf_tail_call(ctx, prog_array_map, index)

下面来看一个使用尾调用的 BCC 框架的具体例子。

BPF_PROG_ARRAY(syscall, 300);						// A

int hello(struct bpf_raw_tracepoint_args *ctx) {			// B
    int opcode = ctx->args[1];						// C
    syscall.call(ctx, opcode);						// D
    bpf_trace_printk("Another syscall: %d", opcode);			// E
    return 0;
}

int hello_execve(void *ctx) {						// F
    bpf_trace_printk("Executing a program");
    return 0;
}

int hello_timer(struct bpf_raw_tracepoint_args *ctx) {			// G
    if (ctx->args[1] == 222) {
        bpf_trace_printk("Creating a timer");
    } else if (ctx->args[1] == 226) {
        bpf_trace_printk("Deleting a timer");
    } else {
        bpf_trace_printk("Some other timer operation");
    }
    return 0;
}

int ignore_opcode(void *ctx) { 						// H
    return 0;
}

代码解释：

【A】BPF_PROG_ARRAY 宏定义，对应映射类型 BPF_MAP_TYPE_PROG_ARRAY。在这里，命名为 syscall，容量为 300。

【B】即将被用户态代码绑定在 sys_enter 类别的 Tracepoint 上，即当有任何系统调用被执行时，都会触发这个函数。bpf_raw_tracepoint_args 类型的结构体 ctx 存放上下文信息。

译者注：sys_enter 是 raw_syscalls 类型的 Tracepoint；同族还有 sys_exit。

详细信息可查看文件：/sys/kernel/debug/tracing/events/raw_syscalls/sys_enter/format

【C】对于 sys_enter 类型的追踪点，其参数第 2 项为操作码，即指代即将执行的系统调用号。这里赋值给变量 opcode。

【D】这一步，我们把 opcode 作为索引，进行尾调用，执行下一个 eBPF 程序。

再次提醒，这里的写法是 BCC 优化，在真正编译前，BCC 最终会将其重写为 bpf_tail_call 辅助函数。

【E】如果尾调用成功，这一行将永远不会被执行。添加这一行的原因是保底输出，防止程序数组 map 没有命中。

【F】hello_execve()，程序数组的一项，对应 execve()系统调用。经由尾调用触发。

【G】hello_timer()，程序数组的一项，对应计时器相关的系统调用。经由尾调用触发。

【H】ignore_opcode()，程序数组的一项，用于忽略我们不关心的系统调用。经由尾调用触发。

现在，我们来看一下用户态的程序（重点，如何加载和设置尾调用）。

b = BPF(text=program)
b.attach_raw_tracepoint(tp="sys_enter", fn_name="hello")		# A

ignore_fn = b.load_func("ignore_opcode", BPF.RAW_TRACEPOINT)		# B
exec_fn = b.load_func("hello_exec", BPF.RAW_TRACEPOINT)
timer_fn = b.load_func("hello_timer", BPF.RAW_TRACEPOINT)

prog_array = b.get_table("syscall")					# C
prog_array[ct.c_int(59)] = ct.c_int(exec_fn.fd)
prog_array[ct.c_int(222)] = ct.c_int(timer_fn.fd)
prog_array[ct.c_int(223)] = ct.c_int(timer_fn.fd)
prog_array[ct.c_int(224)] = ct.c_int(timer_fn.fd)
prog_array[ct.c_int(225)] = ct.c_int(timer_fn.fd)
prog_array[ct.c_int(226)] = ct.c_int(timer_fn.fd)

# Ignore same syscalls that come up a lot				# D
prog_array[ct.c_int(21)] = ct.c_int(ignore_fn.fd)
prog_array[ct.c_int(22)] = ct.c_int(ignore_fn.fd)
prog_array[ct.c_int(25)] = ct.c_int(ignore_fn.fd)
...

b.trace_print()								# E

代码解释：

【A】与前文绑定到 kprobe 不同，这次用户态将 hello() 主 eBPF 程序绑定到 sys_enter 追踪点（Tracepoint）上.

【B】这些 load_func() 方法用来将每个尾调用函数载入内核，并返回尾调用函数的文件描述符。尾调用需要和父调用保持相同的程序类型（这里是 BPF.RAW_TRACEPOINT）。

一定不要混淆，每个尾调用程序本身就是一个 eBPF 程序。

【C】接下来，向我们创建好的 syscall 程序数组中添充条目。大可不必全部填满，如果执行时遇到空的，那也没啥影响。同样的，将多个 index 指向同一个尾调用也是可以的（事实上这段程序就是这样做的，将计时器相关的系统调用指向同一个 eBPF 尾调用）。

译者注：这里的 ct.c_int() 来自 Python 的 ctypes 库，用于 Python 到 C 的类型转换。

【D】由于一些系统调用会频繁地被执行，所以使用 ignore_opcode() 尾调用将他们忽略掉。

【E】不断打印输出，直到用户终止程序。

运行这段程序，获得下面的输出：

当遇到尾调用没匹配上的系统调用时，会输出 “Another syscall”。

内核 4.2 版本才开始支持尾调用，然而在很长的一段时间内，尾调用和 BPF 的编译过程不太兼容（尾调用需要 JIT 编译器的支持）。直到 5.10 版本才解決了这个问题。

你可以最多链接 33 个尾调用（而每个 eBPF 程序的指令复杂度最大支持 100w）。这样一来，eBPF 才能真正发挥出巨大潜力来了。

2.4 小结

本章给出了 eBPF BCC 框架实现的 “Hello World” 程序，以及它的一些变体。同时，也介绍了 eBPF maps 在内核和用户态交互之间的应用。

BCC 框架为我们提供了很好的封装，我们不需要了解程序具体要如何编译、如何载入内核以及如何绑定事件，即可成功运行我们的自定义逻辑。

但作为学习者，仅了解这些是不够的。eBPF 程序到底怎么执行？看来要深入地剖析了。且听下回分解。