使用 libbpf 编写 BPF 应用程序进阶技巧

本文地址：https://www.ebpf.top/post/top_and_tricks_for_bpf_libbpf

原文地址：https://www.pingcap.com/blog/tips-and-tricks-for-writing-linux-bpf-applications-with-libbpf/

2020 年初，当使用 BCC 工具分析我们数据库性能瓶颈并从 GitHub 上拉取代码时，我意外地发现 BCC 项目中额外多出了一个 libbpf-tools目录。我学习了 BPF 可移植性和 BCC 到 libbpf 转换文章，并且根据所学知识将之前提交的 bcc-tools 转换为了 libbpf-tools。最后，我完成了近 20 个工具的转换工作（参见 为什么我们将 BCC-Tools 转换为 libbpf-tools 用于 BPF 性能分析）。

在此过程中，我有幸得到了 Andrii Nakryiko（libbpf + BPF CO-RE 项目的负责人）的大量帮助。这是一段有趣的经历，我也学到了很多。在本文中，我将分享我在使用 libbpf 编写 BPF 程序方面的经验。我希望本文能对 libbpf 感兴趣的人有所帮助，帮助他们进一步开发和完善使用 libbpf 的 BPF 应用程序。

不过在继续阅读之前，建议先阅读这些文章以获取重要的背景信息：

• BPF Portability and CO-RE
• HOWTO: BCC to libbpf conversion
• Building BPF applications with libbpf-boostrap

本文假设你已经阅读了上述文章，因此这里不会有任何系统性的描述。相反，我会针对程序的某些细节部分提供对应的技巧。

程序框架（skeleton）

合并 open 和 loader 阶段

如编写的 BPF 代码不需要任何运行时调整（如调整 map 大小或设置额外配置），你可以调用 <name>__open_and_load() 将两个阶段合并，这会使我们的代码看起来更加简洁。例如：

obj = readahead_bpf__open_and_load();
if (!obj){fprintf(stderr, "failed to open and/or load BPF objectn");return 1;
}
err = readahead_bpf__attach(obj);

你可以在 readahead.c 中查看完整代码样例。【后续版本已经调整，原始提交参见 init readahead.c】

选择性附着 （attach）

默认情况下，<name>__attach() 会附加所有可自动 attach 的 BPF 程序。然而，有时你可能希望根据命令行参数选择性地 attach 对应的 BPF 程序。这种情况下，你可以选择主动调用 bpf_program__attach() 函数。例如：

err = biolatency_bpf__load(obj);
[...]
if (env.queued){obj->links.block_rq_insert =bpf_program__attach(obj->progs.block_rq_insert);err = libbpf_get_error(obj->links.block_rq_insert);[...]
}
obj->links.block_rq_issue =bpf_program__attach(obj->progs.block_rq_issue);
err = libbpf_get_error(obj->links.block_rq_issue);
[...]

你可以在 biolatency.c 看到完整的代码样例。【init biolatency.c 】

自定义 load 和 attach

框架适用于几乎所有情景，但有一种特殊情况：性能事件（perf events）。 这种情况下，你不需要使用 struct <name>__bpf 中的 link ，而是需要定义一个数组结构：struct bpf_link *links[]。这是因为 perf_event 需要在每个 CPU 上单独打开。

之后，你还需要自行 open 和 attach perf_event：

static int open_and_attach_perf_event(int freq, struct bpf_program *prog,struct bpf_link *links[])
{struct perf_event_attr attr = {.type = PERF_TYPE_SOFTWARE,.freq = 1,.sample_period = freq,.config = PERF_COUNT_SW_CPU_CLOCK,};int i, fd;for (i = 0; i < nr_cpus; i++){fd = syscall(__NR_perf_event_open, &attr, -1, i, -1, 0);if (fd < 0){fprintf(stderr, "failed to init perf sampling: %s\n",strerror(errno));return -1;}links[i] = bpf_program__attach_perf_event(prog, fd);if (libbpf_get_error(links[i])){fprintf(stderr, "failed to attach perf event on cpu: ""%d\n", i);links[i] = NULL;close(fd);return -1;}}return 0;
}

最后，在清理阶段，记得要销毁 links 中的每个 link，然后销毁 links 本身。

你可以在 runqlen.c 中看到完整的代码。

同一事件的多 BPF 处理程序

从 v0.2 开始，libbpf 支持在同一可执行文件和可链接格式（ELF）部分中有多个入口点 BPF 程序。因此，你可以将多个 BPF 程序附加到同一事件（例如 tracepoints 或 kprobes），而不必担心 ELF 部分名称冲突。有关详细信息，请参见 Add libbpf full support for BPF-to-BPF calls。现在，你可以自然地在类似下文事件中定义多个处理程序来处理：

SEC("tp_btf/irq_handler_entry")
int BPF_PROG(irq_handler_entry1, int irq, struct irqaction *action)
{[...]
}SEC("tp_btf/irq_handler_entry")
int BPF_PROG(irq_handler_entry2)
{[...]
}

你可以在 hardirqs.bpf.c 中看到完整的代码（代码基于 libbpf-bootstrap 构建）。【备注该文件已经不存在】

如果使用 libbpf 版本早于 v2.0，想要为一个事件定义多个处理程序，你必须使用多个程序类型，例如：

SEC("tracepoint/irq/irq_handler_entry")
int handle__irq_handler(struct trace_event_raw_irq_handler_entry *ctx)
{[...]
}SEC("tp_btf/irq_handler_entry")
int BPF_PROG(irq_handler_entry)
{[...]
}

你可以在 hardirqs.bpf.c 中看到完整的代码。

Map

减少预分配（pre-allocation）开销

【备注：https://github.com/iovisor/bcc/pull/4044 该参数会触发死锁，已经移除？

Using hash maps with BPF_F_NO_PREALLOC flag triggers a warning (0), and according
to kernel commit 94dacdbd5d2d,
this may cause deadlocks. Remove the flag from libbpf tools.】

从 Linux 4.6 开始，BPF hash maps 会默认执行内存预分配，并引入 BPF_F_NO_PREALLOC 标志。这样做的动机是为了避免 kprobe + bpf 死锁。社区尝试了其他解决方案，但最终，预分配所有 map 元素是最简单的解决方案，并且不影响用户空间的行为。

当完整的 map 预分配过于昂贵时，可使用 BPF_F_NO_PREALLOC 标志定义 map 以保持早期行为。详情请参阅 bpf: map pre-alloc。当 map 大小不大时（比如 MAX_ENTRIES = 256），这个标志是不必要的，因为 BPF_F_NO_PREALLOC 速度较慢。

以下是一个使用示例：

struct {__uint(type, BPF_MAP_TYPE_HASH);__uint(max_entries, MAX_ENTRIES);__type(key, u32);__type(value, u64);__uint(map_flags, BPF_F_NO_PREALLOC);
} start SEC(".maps");

你可以在 libbpf-tools 中看到更多的案例。

运行时确定 map 大小

libbpf-tools 的一个优点是可移植，因此 map 所需的最大空间可能因不同的机器而异。在这种情况下，你可以在加载之前定义 map 而不指定大小，然后运行时调整。例如：

在 <name>.bpf.c 中，定义 map ：

struct {__uint(type, BPF_MAP_TYPE_HASH);__type(key, u32);__type(value, u64);
} start SEC(".maps");

在 open 阶段之后，调用 bpf_map__resize() 进行动态调整。例如：

struct cpudist_bpf *obj;[...]
obj = cpudist_bpf__open();
bpf_map__resize(obj->maps.start, pid_max);

你可以在 cpudist.c 中查看完整的代码。【最新代码已经通过 bpf_map__set_max_entries 来调整？】

Per-CPU

在选择 map 类型时，如果与同一 CPU 相关联并发生多个事件，则可以使用 per-CPU 数组来跟踪时间戳，这比使用 hash map 更加简单和高效。然而，你必须确保内核在两次 BPF 程序调用之间不会将进程从一个 CPU 迁移到另一个 CPU。因此，你并非总是能使用这个技巧。下面的示例分析了软中断，并且满足了这两个条件：

struct {__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);__uint(max_entries, 1);__type(key, u32);__type(value, u64);
} start SEC(".maps");SEC("tp_btf/softirq_entry")
int BPF_PROG(softirq_entry, unsigned int vec_nr)
{u64 ts = bpf_ktime_get_ns();u32 key = 0;bpf_map_update_elem(&start, &key, &ts, 0);return 0;
}SEC("tp_btf/softirq_exit")
int BPF_PROG(softirq_exit, unsigned int vec_nr)
{u32 key = 0;u64 *tsp;[...]tsp = bpf_map_lookup_elem(&start, &key);[...]
}

你可以在 softirqs.bpf.c 看到完整的代码。

全局变量

不仅可以使用全局变量来自定义 BPF 程序逻辑，你还可以使用它们来替代 map，这使程序更加简单和高效。全局变量可以是任意大小。你可设定全局变量为一个固定的大小。

例如，因为 SOFTIRQ 类型的数量是固定的，你可以在 softirq.bpf.c 中定义全局数组来保存计数和直方图：

__u64 counts[NR_SOFTIRQS] = {};
struct hist hists[NR_SOFTIRQS] = {};

然后，你可以直接在用户空间遍历这个数组：

static int print_count(struct softirqs_bpf__bss *bss)
{const char *units = env.nanoseconds ? "nsecs" : "usecs";__u64 count;__u32 vec;printf("%-16s %6s%5sn", "SOFTIRQ", "TOTAL_", units);for (vec = 0; vec < NR_SOFTIRQS; vec++){count = __atomic_exchange_n(&bss->counts[vec], 0,__ATOMIC_RELAXED);if (count > 0)printf("%-16s %11llun", vec_names[vec], count);}return 0;
}

你可以在 softirqs.c 看到完整的代码。

注意直接通过指针访问字段

正如你在 BPF 可移植性和 CO-RE 文章中所了解的一样，libbpf + BPF_PROG_TYPE_TRACING 的方法为 BPF 验证器提供了依据。验证器能够原生地理解和追踪 BTF，并允许你直接（而且安全地）跟踪指针并读取内核内存。例如：

u64 inode = task->mm->exe_file->f_inode->i_ino;

这使用起来非常酷。然而，当你在条件语句中使用这样的表达式时，，会由于分支被优化掉在某些内核版本中引入 bug 。在这种情况下，直到 bpf: fix an incorrect branch elimination by verifier 被广泛引入之前，请使用 BPF_CORE_READ 以确保内核兼容性。你可以在 biolatency.bpf.c 中找到一个示例：

SEC("tp_btf/block_rq_issue")
int BPF_PROG(block_rq_issue, struct request_queue *q, struct request *rq)
{if (targ_queued && BPF_CORE_READ(q, elevator))return 0;return trace_rq_start(rq);
}

你可以看到，即使它是一个 tp_btf 程序且 q->elevator 速度更快，我还是使用了 BPF_CORE_READ(q, elevator)。

结论

本文介绍了使用 libbpf 编写 BPF 程序的一些技巧。你可以在 libbpf-tools 和 bpf中找到许多实际的示例。如果你有任何问题，欢迎加入 Slack 上的 TiDB 社区并向我们发送反馈。