BPF环形缓冲区

Posted rtoax

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了BPF环形缓冲区相关的知识，希望对你有一定的参考价值。

目录

BPF Ringbuf和BPF perfbuf

内存开销

活动订购

浪费工作和额外的数据复制

性能和适用性

给我看看代码！

BPF perfbuf：bpf_perf_event_output（）

BPF ringbuf：bpf_ringbuf_output（）

BPF ringbuf：保留/提交API

BPF ringbuf：数据通知控制

结论

 

现在有一个新的BPF数据结构可用：BPF环形缓冲区。它解决了BPF性能缓冲区（目前已成为从内核向用户空间发送数据的事实上的标准）的内存效率和事件重排序问题，同时达到或超过了其性能。它既提供了与perfbuf兼容的功能，可轻松迁移，又提供了具有更好可用性的新的reserve / commit API。同样，综合基准和实际基准都表明，在几乎所有情况下，都应考虑将其作为从BPF程序向用户空间发送数据的默认选择。

BPF Ringbuf和BPF perfbuf

如今，每当BPF程序需要将收集的数据发送到用户空间以进行后处理和记录时，通常会为此使用BPF性能缓冲区（perfbuf）。Perfbuf是每个CPU循环缓冲区的集合，它允许在内核和用户空间之间高效地交换数据。它在实践中效果很好，但是由于其基于CPU的设计，它在实践中存在两个主要缺点，事实证明很不方便：内存使用效率低和事件重新排序。

为了解决这些问题，从Linux 5.8开始，BPF提供了新的BPF数据结构（BPF映射）：BPF环形缓冲区（ringbuf）。它是一个多生产者单消费者（MPSC）队列，可以同时在多个CPU之间安全地共享。

BPF ringbuf支持BPF perfbuf熟悉的功能：

  • 可变长度数据记录；
  • 能够通过内存映射区域从用户空间有效读取数据，而无需向内核中进行额外的内存复制和/或系统调用；
  • epoll通知都支持，并且能够在绝对最小的等待时间内进行忙循环。

同时，BPF ringbuf解决了BPF perfbuf的以下问题：

  • 内存开销；
  • 数据排序；
  • 浪费工作和额外的数据复制。

内存开销

BPF perfbuf为每个CPU分配一个单独的缓冲区。这通常意味着BPF开发人员必须在分配足够大的每CPU缓冲区（以适应发射的数据的峰值）或提高内存效率（在稳定状态下不浪费大多数空缓冲区的不必要内存）之间做出权衡数据峰值期间的数据）。对于那些在大多数时间大部分时间都处于空闲状态但在短时间内周期性地涌入大量事件的应用程序而言，这尤其棘手。很难找到适当的平衡，因此BPF应用程序通常会出于安全考虑过度分配perfbuf内存，否则将不时出现不可避免的数据丢失。

通过在所有CPU之间共享，BPF ringbuf允许使用一个大的公共缓冲区来处理此问题。与BPF perfbuf相比，更大的缓冲区可以吸收更大的尖峰，但也可以允许整体使用更少的RAM。 随着CPU数量的增加，BPF ringbuf内存的使用也可以更好地扩展，因为从16个CPU迁移到32个CPU并不一定需要两倍的缓冲区来容纳更多的负载。不幸的是，使用BPF perfbuf，由于每个CPU缓冲区的原因，您几乎没有选择。

活动订购

如果BPF应用程序必须跟踪相关事件（例如，进程开始和退出，网络连接生命周期事件等），则事件的正确排序就变得至关重要。但是，这对于BPF perfbuf是有问题的。如果相关事件在不同的CPU上连续（数毫秒内）连续发生，则它们可能会无序交付。同样，这是由于BPF perfbuf的每CPU特性。

作为一个真实的例子，几年前我写的一个应用程序必须跟踪进程fork / exec / exit事件并收集每个进程的资源使用情况统计信息。我的应用程序必须为每个事件将事件发送到BPF perfbuf中，但是很多时候它们都是乱序到达的。这是因为由于内核调度程序将fork（），exec（）和exit（）从一个CPU迁移到另一个CPU，因此它们可以在非常短的连续时间内在不同的CPU上快速连续地发生。要解决这一问题，就需要大幅增加应用程序处理逻辑的复杂性，考虑到问题原本就非常简单的性质，这远远超过了人们的预期。

BPF ringbuf根本不是问题，它通过将事件发送到共享缓冲区中来解决此问题，并保证如果事件A在事件B之前提交，那么它也将在事件B之前被消耗。这通常会很明显简化处理逻辑。

浪费工作和额外的数据复制

使用BPF perfbuf，BPF程序必须准备数据样本，然后再将其复制到perf缓冲区中以发送到用户空间。这意味着相同的数据必须被复制两次：首先复制到局部变量或每个CPU数组中（对于不能容纳在小型BPF堆栈中的大样本），然后复制到perfbuf中。更糟糕的是，如果发现perfbuf剩余的空间不足，那么所有这些工作都将被浪费掉。

BPF ringbuf支持替代的保留/提交API来避免这种情况。可以首先保留数据空间。如果保留成功，则BPF程序可以直接使用该内存来准备数据样本。完成之后，将数据提交到用户空间是一种极其高效的操作，它不可能失败，也完全不执行任何额外的内存副本。如果由于缓冲区空间不足而导致保留失败，那么至少在完成所有工作以记录数据之前，您应该知道这一点，之后再放到地板上。ringbuf-reserve-commit下面的示例将显示实际情况。

性能和适用性

在所有实际应用中，BPF ringbuf的性能均优于BPF perfbuf（特别是在BCC / libbpf中使用perfbuf数据的默认设置有些欠佳）。如果您喜欢硬数字，则可以在此补丁中找到各种方案的广泛综合基准测试结果。

理论上，由于每个CPU缓冲区，BPF perfbuf可以支持更高的数据吞吐量，但这仅在我们每秒谈论数百万个事件时才有意义 。但是通过编写实际的高吞吐量应用程序的实验证实，如果将BPF ringbuf用作每个CPU缓冲区（类似于BPF perfbuf），它仍然是BPF perfbuf的更高性能替代品。特别是，如果采用手动数据可用性通知。您可以在内核自测之一中查看基本的multi-ringbuf示例（BPF端， 用户空间端）。稍后，我们将看一个手动控制数据可用性通知的示例。

您可能需要小心并首先进行实验的唯一情况是，当BPF程序必须从NMI（不可屏蔽中断）上下文运行时（例如，用于处理perf事件，例如cpu-cycles）。BPF ringbuf在内部使用了非常轻量级的自旋锁，这意味着，如果在NMI上下文中争用锁，则数据保留可能会失败。因此，在NMI上下文中，如果CPU争用很高，即使ringbuf本身仍有一些可用空间，也可能会丢失一些数据。

在所有其他情况下，选择新的BPF ringbuf是一个非常明显的选择。BPF ringbuf可提供更好的性能和内存效率，更好的顺序保证以及更好的API（在内核端和用户空间中）。

给我看看代码！

为了显示BPF ringbuf API，将它们与BPF perfbuf的API进行比较，并查看它们在实际中的典型用法，我编写了一个小的bpf-ringbuf-examples项目，在本文的其余部分中将继续介绍。

此仓库实现了同一BPF应用程序的三个变体，该变体将跟踪捕获新进程产生的所有进程执行者。对于每个exec()ID，进程ID（pid），进程名称（comm）和可执行文件路径（filename）被捕获到一个示例中，并发送到用户空间以进行后处理（在我们的演示中，只需将printf（）放入到标准输出）。这是所有三个示例的输出的样子（不要忘记使用来运行它sudo）：

$ sudo ./ringbuf-reserve-commit    # or ./ringbuf-output, or ./perfbuf-output
TIME     EVENT PID     COMM             FILENAME
19:17:39 EXEC  3232062 sh               /bin/sh
19:17:39 EXEC  3232062 timeout          /usr/bin/timeout
19:17:39 EXEC  3232063 ipmitool         /usr/bin/ipmitool
19:17:39 EXEC  3232065 env              /usr/bin/env
19:17:39 EXEC  3232066 env              /usr/bin/env
19:17:39 EXEC  3232065 timeout          /bin/timeout
19:17:39 EXEC  3232066 timeout          /bin/timeout
19:17:39 EXEC  3232067 sh               /bin/sh
19:17:39 EXEC  3232068 sh               /bin/sh
^C

这是 从BPF程序发送并在应用程序的用户空间部分使用的示例数据的C结构定义：

#define TASK_COMM_LEN 16
#define MAX_FILENAME_LEN 512

/* definition of a sample sent to user-space from BPF program */
struct event 
	int pid;
	char comm[TASK_COMM_LEN];
	char filename[MAX_FILENAME_LEN];
;

BPF perfbuf：bpf_perf_event_output（）

让我们从一个BPF perfbuf用例开始，这里可以找到BPF的一部分 。

首先，我们包括来自内核的一些基本BPF定义<linux/bpf.h>，来自libbpf的BPF帮助器定义以及来自的<bpf/bpf_helpers.h>应用程序类型"common.h"，它们在BPF和用户空间代码之间共享。我们还指定我们的程序受双重GPL-2.0 / BSD-3许可：

// SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
/* Copyright (c) 2020 Andrii Nakryiko */
#include <linux/bpf.h>
#include <bpf/bpf_helpers.h>
#include "common.h"

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

接下来，我们将BPF perfbuf本身定义为BPF_MAP_TYPE_PERF_EVENT_ARRAY映射。无需定义max_entries属性，因为libbpf会处理该属性，并将其自动调整为系统上可用CPU的数量。每个CPU缓冲区的大小是与用户空间分开指定的，我们将在稍后介绍。

/* BPF perfbuf map */
struct 
	__uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
	__uint(key_size, sizeof(int));
	__uint(value_size, sizeof(int));
 pb SEC(".maps");

因为样本（struct event在common.h中定义 ）非常大（> 512字节，因为我将文件名的最大捕获大小设置为512字节），所以我们无法在堆栈上准备数据。因此，我们将单元素的每个CPU阵列用作临时存储：

struct 
	__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
	__uint(max_entries, 1);
	__type(key, int);
	__type(value, struct event);
 heap SEC(".maps");

现在，我们定义BPF程序本身，并指定应将其附加到sched:sched_process_exec，这是在每次成功的exec() syscall上触发的。struct trace_event_raw_sched_process_exec也是在common.h中定义的， 只是从Linux来源复制/粘贴。它定义了该特定跟踪点的输入数据的布局。

SEC("tp/sched/sched_process_exec")
int handle_exec(struct trace_event_raw_sched_process_exec *ctx)

	unsigned fname_off = ctx->__data_loc_filename & 0xFFFF;
	struct event *e;
	int zero = 0;

	e = bpf_map_lookup_elem(&heap, &zero);
	if (!e) /* can't happen */
		return 0;

	e->pid = bpf_get_current_pid_tgid() >> 32;
	bpf_get_current_comm(&e->comm, sizeof(e->comm));
	bpf_probe_read_str(&e->filename, sizeof(e->filename), (void *)ctx + fname_off);

	bpf_perf_event_output(ctx, &pb, BPF_F_CURRENT_CPU, e, sizeof(*e));
	return 0;

BPF程序逻辑非常简单。它为我们的样本获取一个临时存储，并用来自跟踪点上下文的数据填充它。完成后，它将通过bpf_perf_event_output()调用将样本发送到BPF perfbuf 。该APIstruct event在当前CPU的性能缓冲区中保留空间，将sizeof(*e)数据字节从复制e到该保留空间，完成后它将向用户空间发出新数据可用的信号。届时，epoll子系统将唤醒用户空间处理程序，并将指针传递到该数据副本以进行处理。

这实际上就是BPF方面的全部。如果您曾经见过任何其他现代BPF应用程序，则非常简单。

现在让我们浏览用户空间方面 。它依靠BPF框架（您可以在此处了解更多信息 ），因此非常简短。完成最小的初始设置（设置libbpf日志记录处理程序，中断处理程序，RLIMIT_MEMLOCK BPF系统的缓冲限制）之后，它只是打开并加载BPF框架。如果一切成功，那么我们将使用libbpf的用户空间perf_buffer__new()API创建perf缓冲区使用者的实例：

	struct perf_buffer *pb = NULL;
	struct perf_buffer_opts pb_opts = ;
	struct perfbuf_output_bpf *skel;

	...

	/* Set up ring buffer polling */
	pb_opts.sample_cb = handle_event;
	pb = perf_buffer__new(bpf_map__fd(skel->maps.pb), 8 /* 32KB per CPU */, &pb_opts);
	if (libbpf_get_error(pb)) 
		err = -1;
		fprintf(stderr, "Failed to create perf buffer\\n");
		goto cleanup;

在这里，我们指定每个CPU缓冲区为32KB（8页x每页4096字节），对于每个提交的示例，libbpf将调用我们的handle_event() 回调，该回调仅输出数据printf()：

void handle_event(void *ctx, int cpu, void *data, unsigned int data_sz)

	const struct event *e = data;
	struct tm *tm;
	char ts[32];
	time_t t;

	time(&t);
	tm = localtime(&t);
	strftime(ts, sizeof(ts), "%H:%M:%S", tm);

	printf("%-8s %-5s %-7d %-16s %s\\n", ts, "EXEC", e->pid, e->comm, e->filename);

最后一步是只要有可用的数据就不断消耗它，直到需要退出为止（例如，如果用户按下Ctrl-C）：

	/* Process events */
	printf("%-8s %-5s %-7s %-16s %s\\n",
	       "TIME", "EVENT", "PID", "COMM", "FILENAME");
	while (!exiting) 
		err = perf_buffer__poll(pb, 100 /* timeout, ms */);
		/* Ctrl-C will cause -EINTR */
		if (err == -EINTR) 
			err = 0;
			break;
		
		if (err < 0) 
			printf("Error polling perf buffer: %d\\n", err);
			break;

BPF ringbuf：bpf_ringbuf_output（）

BPF ringbuf的bpf_ringbuf_output()API旨在遵循BPF perfbuf的语义，bpf_perf_event_output()从而使迁移变得轻而易举。为了说明可用性的相似度和接近度，我将逐字说明perfbuf-output和ringbuf-output示例之间的区别。您可以 在Github上查看完整的BPF端代码 和用户空间代码。

这是BPF的差异：

--- src/perfbuf-output.bpf.c	2020-10-25 18:52:22.247019800 -0700
+++ src/ringbuf-output.bpf.c	2020-10-25 18:44:14.510630322 -0700
@@ -6,12 +6,11 @@
 
 char LICENSE[] SEC("license") = "Dual BSD/GPL";
 
-/* BPF perfbuf map */
+/* BPF ringbuf map */
 struct 
-	__uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
-	__uint(key_size, sizeof(int));
-	__uint(value_size, sizeof(int));
- pb SEC(".maps");
+	__uint(type, BPF_MAP_TYPE_RINGBUF);
+	__uint(max_entries, 256 * 1024 /* 256 KB */);
+ rb SEC(".maps");
 
 struct 
 	__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
@@ -35,7 +34,7 @@
 	bpf_get_current_comm(&e->comm, sizeof(e->comm));
 	bpf_probe_read_str(&e->filename, sizeof(e->filename), (void *)ctx + fname_off);
 
-	bpf_perf_event_output(ctx, &pb, BPF_F_CURRENT_CPU, e, sizeof(*e));
+	bpf_ringbuf_output(&rb, e, sizeof(*e), 0);
 	return 0;

所以只有两个简单的更改：

  1. BPF ringbuf映射的定义略有不同。现在可以在BPF端定义它的大小（但现在是所有CPU共享的缓冲区的大小）。请记住，仍然可以在BPF端定义中忽略它，并在用户空间端使用bpf_map__set_max_entries()API指定（或重写（如果您在BPF端指定它，则覆盖）。另一个区别是size（max_entries属性）以字节数指定，唯一的限制是它应该是内核页面大小的倍数（几乎总是4096字节），并且是2的幂（类似于perfbuf的number of页，也必须是2的幂。BPF perfbuf大小是从用户空间一侧指定的，并且在多个内存页中。

  2. bpf_perf_event_output()被非常相似的替换， bpf_ringbuf_output()唯一的区别是ringbuf API不需要引用BPF程序的上下文。

这是BPF方面仅有的两个区别。

在用户空间方面，更改也很小。忽略perf_buffer <->ring_buffer重命名，可以归结为两个更改。首先，事件处理程序回调的定义现在可以返回错误（这将终止使用者循环），并且不占用产生事件的CPU的索引：

-void handle_event(void *ctx, int cpu, void *data, unsigned int data_sz)
+int handle_event(void *ctx, void *data, size_t data_sz)

	const struct event *e = data;
	struct tm *tm;

如果知道CPU索引很重要，则必须从BPF端显式地将其记录在样本中。此外，ring_bufferAPI不会为丢失的样本提供回调perf_buffer。如有必要，也需要从BPF方面对此进行明确处理。这样做是为了最小化共享（跨CPU）环形缓冲区中的锁争用，并且在不需要时不付出代价。另外，在实践中，除了报告这一点之外，几乎没有人可以做得到，这可以通过显式BPF代码更有效，更方便地完成。

第二个区别是ring_buffer__new()API稍微简单一些，它允许在不使用额外选项struct的情况下指定回调：

 	/* Set up ring buffer polling */
-	pb_opts.sample_cb = handle_event;
-	pb = perf_buffer__new(bpf_map__fd(skel->maps.pb), 8 /* 32KB per CPU */, &pb_opts);
-	if (libbpf_get_error(pb)) 
+	rb = ring_buffer__new(bpf_map__fd(skel->maps.rb), handle_event, NULL, NULL);
+	if (!rb) 
 		err = -1;
-		fprintf(stderr, "Failed to create perf buffer\\n");
+		fprintf(stderr, "Failed to create ring buffer\\n");
 		goto cleanup;

这样，简单地替换perf_buffer__poll()为即可ring_buffer__poll() 使您以完全相同的方式开始使用环形缓冲区数据：

 	printf("%-8s %-5s %-7s %-16s %s\\n",
 	       "TIME", "EVENT", "PID", "COMM", "FILENAME");
 	while (!exiting) 
-		err = perf_buffer__poll(pb, 100 /* timeout, ms */);
+		err = ring_buffer__poll(rb, 100 /* timeout, ms */);
 		/* Ctrl-C will cause -EINTR */
 		if (err == -EINTR) 
 			err = 0;
 			break;
 		
 		if (err < 0) 
-			printf("Error polling perf buffer: %d\\n", err);
+			printf("Error polling ring buffer: %d\\n", err);
 			break;

BPF ringbuf：保留/提交API

的目标 bpf_ringbuf_output()API允许从BPF perfbuf平滑过渡到BPF ringbuf，而无需对BPF代码进行任何实质性更改。但这也意味着它具有BPF perfbuf API的一些缺点：额外的内存复制和非常晚的数据保留。前者意味着您需要额外的空间来构造样本，然后再将其复制到缓冲区中。这不仅效率低下，而且对于单元素的每CPU阵列经常需要额外的复杂性。后者意味着，如果由于滞后的用户空间或由于传入事件的快速爆发使缓冲区溢出而导致缓冲区中没有剩余空间，则浪费样本构建所有工作的工作。但是，如果您知道无论如何都会删除数据，则可以首先跳过收集数据的过程，并节省一些资源以使用户方更快地进行追赶。但它' xxx_output()样式的API。

这就是bpf_ringbuf_reserve()/ bpf_ringbuf_commit()API派上用场的地方。Reserve（保留）功能使您可以做到这一点：尽早保留空间或确定不可能（NULL在这种情况下返回）。如果我们没有足够的数据来提交样本，则可以跳过花费所有资源来捕获数据的步骤。但是，如果保留成功，那么我们可以保证，一旦完成数据收集，将其发布到用户空间将永远不会失败。即，如果bpf_ringbuf_reserve()返回非NULL指针，则后续操作bpf_ringbuf_commit()将始终成功。

此外，环形缓冲区本身中的保留空间直到提交后才对用户空间可见，因此可以随意使用它来构造样本，无论它是复杂且多步骤的操作。而且，它不需要额外的内存复制和临时存储空间。唯一的限制是BPF验证者必须在验证时知道预留的大小，因此必须处理具有动态大小的样本bpf_ringbuf_output()并支付额外副本的费用。

但是在大多数情况下，保留/提交是您应该首选的方法。这是BPF程序代码的差异（完整的 BPF 和 用户空间 代码也位于Github上）：

--- src/ringbuf-output.bpf.c	2020-10-25 18:44:14.510630322 -0700
+++ src/ringbuf-reserve-submit.bpf.c	2020-10-25 18:36:53.409470270 -0700
@@ -12,29 +12,21 @@
 	__uint(max_entries, 256 * 1024 /* 256 KB */);
  rb SEC(".maps");
 
-struct 
-	__uint(type, BPF_MAP_TYPE_PERCPU_ARRAY);
-	__uint(max_entries, 1);
-	__type(key, int);
-	__type(value, struct event);
- heap SEC(".maps");
-
 SEC("tp/sched/sched_process_exec")
 int handle_exec(struct trace_event_raw_sched_process_exec *ctx)
 
 	unsigned fname_off = ctx->__data_loc_filename & 0xFFFF;
 	struct event *e;
-	int zero = 0;
 	
-	e = bpf_map_lookup_elem(&heap, &zero);
-	if (!e) /* can't happen */
+	e = bpf_ringbuf_reserve(&rb, sizeof(*e), 0);
+	if (!e)
 		return 0;
 
 	e->pid = bpf_get_current_pid_tgid() >> 32;
 	bpf_get_current_comm(&e->comm, sizeof(e->comm));
 	bpf_probe_read_str(&e->filename, sizeof(e->filename), (void *)ctx + fname_off);
 
-	bpf_ringbuf_output(&rb, e, sizeof(*e), 0);
+	bpf_ringbuf_submit(e, 0);
 	return 0;

每个CPU阵列已经一去不复返了，取而代之的是，我们使用的结果 bpf_ringbuf_reserve()将数据填充到样本中。

用户空间部分是完全相同的（以BPF骨架对象的名称为模），这是有道理的，因为最后您将消耗来自BPF环形缓冲区的完全相同的数据。

BPF ringbuf：数据通知控制

在处理高吞吐量的情况时，通常最大的开销来自提交样本时数据可用性的内核内信令（这使内核的poll / epoll系统唤醒在等待新数据时被阻塞的用户空间处理程序）。perfbuf和ringbuf都是如此。

Perfbuf可以设置采样通知来处理此问题，在这种情况下，只有每N个采样都会发送一个通知。您可以在用户空间中创建BPF perfbuf映射时执行此操作。而且，您需要确保它对您有用，直到第N个样本到来，您才会看到最后的N-1个样本。对于您的特定情况，这可能并不重要。

BPF ringbuf与此不同。bpf_ringbuf_output()并 bpf_ringbuf_commit()接受额外的flags参数，则可以指定BPF_RB_NO_WAKEUPorBPF_RB_FORCE_WAKEUP标志。指定 BPF_RB_NO_WAKEUP禁止发送内核数据可用性通知。虽然BPF_RB_FORCE_WAKEUP会强制发送通知。如果需要，这可以进行精确的手动控制。要查看如何完成此操作，请检查BPF ringbuf基准测试，该基准测试仅在环形缓冲区中放入可配置的数据量时才发送通知。

默认情况下，如果未指定标志，则BPF ringbuf代码将根据用户空间使用者是否落后来进行自适应通知，这将导致用户空间使用者从不丢失单个样本通知，但不会付出不必要的代价高架。没有一个标志是一个很好的安全默认值，但是如果您需要获得额外的性能，则根据您的自定义条件（例如，缓冲区中排队的数据量）手动控制数据通知可能会大大提高性能。

结论

这篇文章解释了BPF环形缓冲区正在解决的问题以及API选择背后的动机。希望通过代码示例以及指向内核自测和基准的额外链接，可以使您对BPF ringbuf API有所了解，并演示了API的简单和更高级用法，以满足您的应用程序需求。

以上是关于BPF环形缓冲区的主要内容，如果未能解决你的问题，请参考以下文章

怎么计算环形缓冲区长度

环形缓冲区大小和反写阈值

微控制器的环形缓冲区

环形缓冲区

关于环形缓冲区

数据结构之环形缓冲器