BPF CO-RE 示例代码解析
BPF CO-RE 示例代码解析
charlieroro
发布于 2021-02-04 10:29:14
发布于 2021-02-04 10:29:14
BPF CO-RE 示例代码解析
在BPF的可移植性和CO-RE一文的末尾提到了一个名为runqslower的工具，该工具用于展示在CPU run队列中停留的时间大于某一值的任务。现在以该工具来展示如何使用BPF CO-RE。
目录
- BPF CO-RE 示例代码解析
- 环境
- 编译
- 运行
- 代码解析
- 内核空间(BPF)代码
- 用户空间代码
- TIPs
- 总结
- 参考
环境
本地测试的话，建议采用Ubuntu，其内核本身已经开启了BTF选项，无需再对内核进行编译。我用的是Ubuntu 20.10，内核版本5.8.0。
# cat /boot//config-$(uname -r)|grep BTF
CONFIG_VIDEO_SONY_BTF_MPX=m
CONFIG_DEBUG_INFO_BTF=y编译
仅需要在runqslower目录下执行make即可。如果用的是自己生成的vmlinux，则需要在Makefile中增加对VMLINUX_BTF 的定义，值为本地编译的vmlinux的路径，如：
VMLINUX_BTF := /root/linux-5.10.5/vmlinux在BCC和libbpf的转换一文中可以了解到，BPF CO-RE的基本步骤如下，：
- 生成包含所有内核类型的头文件
vmlinux.h； - 使用Clang(版本10或更新版本)将BPF程序的源代码编译为
.o对象文件； - 从编译好的BPF对象文件中生成BPF skeleton 头文件(对应runqslower的BPF对象文件为
runqslower.bpf.o，也可以通过bpftool gen skeleton runqslower.bpf.o生成skeleton头文件) ; - 在用户空间代码中包含生成的BPF skeleton 头文件(BPF skeleton 头文件是给用户空间使用的)；
- 最后，编译用户空间代码，这样会嵌入BPF对象代码，后续就不用发布单独的文件。
其中第1、3步分别使用bpftool btf dump file和bpftool gen skeleton来生成vmliunx.h和skeleton 头文件。具体使用方式可以参见runqslower的Makefile文件。
运行
直接看下最终的效果，运行如下，可以看到该BPF应用其实就是一个普通的ELF可执行文件(无需独立发布BPF程序和用户侧程序)，大小仅为1M左右，如果要在另一台机器运行，直接拷贝过去即可(前提是目标内核开启了CONFIG_DEBUG_INFO_BTF选项)。
# ./runqslower 200
Tracing run queue latency higher than 200 us
TIME COMM PID LAT(us)
16:45:16 kworker/u256:1 6007 209
16:45:16 kworker/1:2 6045 1222
16:45:16 sshd 6045 331
16:45:16 swapper/0 6045 2120使用bpftool prog -p可以查看安装的bpf程序：
{
"id": 157,
"type": "tracing",
"name": "handle__sched_w",
"tag": "4eadb7a05d79f434",
"gpl_compatible": true,
"loaded_at": 1611822519,
"uid": 0,
"bytes_xlated": 176,
"jited": true,
"bytes_jited": 121,
"bytes_memlock": 4096,
"map_ids": [71,69
],
"btf_id": 65,
"pids": [{
"pid": 6012,
"comm": "runqslower"
}
]
},{
"id": 158,
"type": "tracing",
"name": "handle__sched_s",
"tag": "36ab461bac5b3a97",
"gpl_compatible": true,
"loaded_at": 1611822519,
"uid": 0,
"bytes_xlated": 584,
"jited": true,
"bytes_jited": 354,
"bytes_memlock": 4096,
"map_ids": [71,69,70
],
"btf_id": 65,
"pids": [{
"pid": 6012,
"comm": "runqslower"
}
]
}代码解析
按照上述编译中设计的顺序，首选应该编写BFP层的代码，然后再编写用户空间的代码。BPF CO-RE的处理逻辑基本与BCC保持一致。当触发相关事件时会运行内核空间代码，然后在用户空间接收内核代码传递的信息。
下面以代码注释的方式解析BPF CO-RE的一些使用规范，最后会做一个总结。
内核空间(BPF)代码
内核空间代码通常包含如下头文件：
#include "vmlinux.h" /* all kernel types */
#include <bpf/bpf_helpers.h> /* most used helpers: SEC, __always_inline, etc */
#include <bpf/bpf_core_read.h> /* for BPF CO-RE helpers */内核空间的BPF代码如下(假设生成的.o文件名为runqslower.bpf.o)：
// SPDX-License-Identifier: GPL-2.0
// Copyright (c) 2019 Facebook
/* BPF程序包含的头文件，可以看到内容想相当简洁 */
#include "vmlinux.h"
#include <bpf/bpf_helpers.h>
#include "runqslower.h"
#define TASK_RUNNING 0
#define BPF_F_CURRENT_CPU 0xffffffffULL
/* 在BPF代码侧，可以使用一个 const volatile 声明只读的全局变量，只读的全局变量，变量最后会存在于runqslower.bpf.o的.rodata只读段，用户侧可以在BPF程序加载前读取或修改该只读段的参数【1】 */
const volatile __u64 min_us = 0;
const volatile pid_t targ_pid = 0;
/* 定义名为 start 的map，类型为 BPF_MAP_TYPE_HASH。容量为10240，key类型为u32，value类型为u64。可以在【1】中查看BPF程序解析出来的.maps段【2】 */
struct {
__uint(type, BPF_MAP_TYPE_HASH);
__uint(max_entries, 10240);
__type(key, u32);
__type(value, u64);
} start SEC(".maps");
/* 由于 PERF_EVENT_ARRAY, STACK_TRACE 和其他特殊的maps(DEVMAP, CPUMAP, etc) 尚不支持key/value类型的BTF类型，因此需要直接指定 key_size/value_size */
struct {
__uint(type, BPF_MAP_TYPE_PERF_EVENT_ARRAY);
__uint(key_size, sizeof(u32));
__uint(value_size, sizeof(u32));
} events SEC(".maps");
/* record enqueue timestamp */
/* 自定义的辅助函数必须标记为 static __always_inline。该函数用于保存唤醒的任务事件，key为pid，value为唤醒的时间点 */
__always_inline
static int trace_enqueue(u32 tgid, u32 pid)
{
u64 ts;
if (!pid || (targ_pid && targ_pid != pid))
return 0;
ts = bpf_ktime_get_ns();
bpf_map_update_elem(&start, &pid, &ts, 0);
return 0;
}
/* 所有BPF程序提供的功能都需要通过 SEC() (来自 bpf_helpers.h )宏来自定义section名称【3】。可以在【1】中查看BPF程序解析出来的自定义函数 */
/* 唤醒一个任务，并保存当前时间 */
SEC("tp_btf/sched_wakeup")
int handle__sched_wakeup(u64 *ctx)
{
/* TP_PROTO(struct task_struct *p) */
struct task_struct *p = (void *)ctx[0];
return trace_enqueue(p->tgid, p->pid);
}
/* 唤醒一个新创建的任务，并保存当前时间。BPF的上下文为一个task_struct*结构体 */
SEC("tp_btf/sched_wakeup_new")
int handle__sched_wakeup_new(u64 *ctx)
{
/* TP_PROTO(struct task_struct *p) */
struct task_struct *p = (void *)ctx[0];
return trace_enqueue(p->tgid, p->pid);
}
/* 计算一个任务入run队列到出队列的时间 */
SEC("tp_btf/sched_switch")
int handle__sched_switch(u64 *ctx)
{
/* TP_PROTO(bool preempt, struct task_struct *prev,
* struct task_struct *next)
*/
struct task_struct *prev = (struct task_struct *)ctx[1];
struct task_struct *next = (struct task_struct *)ctx[2];
struct event event = {};
u64 *tsp, delta_us;
long state;
u32 pid;
/* ivcsw: treat like an enqueue event and store timestamp */
/* 如果被切换的任务的状态仍然是TASK_RUNNING，说明其又重新进入run队列，更新入队列的时间 */
if (prev->state == TASK_RUNNING)
trace_enqueue(prev->tgid, prev->pid);
/* 获取下一个任务的PID */
pid = next->pid;
/* fetch timestamp and calculate delta */
/* 如果该任务并没有被唤醒，则无法正常进行任务切换，返回0即可 */
tsp = bpf_map_lookup_elem(&start, &pid);
if (!tsp)
return 0; /* missed enqueue */
/* 当前切换时间减去该任务的入队列时间，计算进入run队列到真正调度的毫秒级时间 */
delta_us = (bpf_ktime_get_ns() - *tsp) / 1000;
if (min_us && delta_us <= min_us)
return 0;
/* 更新events section，以便用户侧读取 */
event.pid = pid;
event.delta_us = delta_us;
bpf_get_current_comm(&event.task, sizeof(event.task));
/* output */
bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU,
&event, sizeof(event));
/* 该任务已经出队列，删除map */
bpf_map_delete_elem(&start, &pid);
return 0;
}
char LICENSE[] SEC("license") = "GPL";【1】：
【2】：
【3】：
用户空间代码
用户侧代码通常包含如下头文件：
#include <bpf/bpf.h>
#include <bpf/libbpf.h>
#include "path/to/your/skeleton.skel.h"用户侧的主要代码如下：
int libbpf_print_fn(enum libbpf_print_level level,
const char *format, va_list args)
{
if (level == LIBBPF_DEBUG && !env.verbose)
return 0;
return vfprintf(stderr, format, args);
}
static int bump_memlock_rlimit(void)
{
struct rlimit rlim_new = {
.rlim_cur = RLIM_INFINITY,
.rlim_max = RLIM_INFINITY,
};
return setrlimit(RLIMIT_MEMLOCK, &rlim_new);
}
void handle_event(void *ctx, int cpu, void *data, __u32 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 %-16s %-6d %14llu\n", ts, e->task, e->pid, e->delta_us);
}
void handle_lost_events(void *ctx, int cpu, __u64 lost_cnt)
{
printf("Lost %llu events on CPU #%d!\n", lost_cnt, cpu);
}
int main(int argc, char **argv)
{
static const struct argp argp = {
.options = opts,
.parser = parse_arg,
.doc = argp_program_doc,
};
struct perf_buffer_opts pb_opts;
struct perf_buffer *pb = NULL;
struct runqslower_bpf *obj;
int err;
err = argp_parse(&argp, argc, argv, 0, NULL, NULL);
if (err)
return err;
/* 设置libbpf的日志打印 */
libbpf_set_print(libbpf_print_fn);
/* BPF的BPF maps以及其他内容使用了locked类型的内存, libbpf不会自动设置该值，因此必须手动指定 */
err = bump_memlock_rlimit();
if (err) {
fprintf(stderr, "failed to increase rlimit: %d", err);
return 1;
}
/* 获取BPF对象，程序被编码到了bpf_object_skeleton.data中【1】 */
obj = runqslower_bpf__open();
if (!obj) {
fprintf(stderr, "failed to open and/or load BPF object\n");
return 1;
}
/* initialize global data (filtering options) */
/* 通过.rodata段修改全局变量，注意此时并没有加载BPF程序 */
obj->rodata->targ_pid = env.pid;
obj->rodata->min_us = env.min_us;
/* 将BPF程序(使用mmap方式)加载到内存中 */
err = runqslower_bpf__load(obj);
if (err) {
fprintf(stderr, "failed to load BPF object: %d\n", err);
goto cleanup;
}
/* 附加BPF程序，此时runqslower_bpf.links生效【2】 */
err = runqslower_bpf__attach(obj);
if (err) {
fprintf(stderr, "failed to attach BPF programs\n");
goto cleanup;
}
printf("Tracing run queue latency higher than %llu us\n", env.min_us);
printf("%-8s %-16s %-6s %14s\n", "TIME", "COMM", "PID", "LAT(us)");
pb_opts.sample_cb = handle_event;
pb_opts.lost_cb = handle_lost_events;
pb = perf_buffer__new(bpf_map__fd(obj->maps.events), 64, &pb_opts);
err = libbpf_get_error(pb);
if (err) {
pb = NULL;
fprintf(stderr, "failed to open perf buffer: %d\n", err);
goto cleanup;
}
/* 轮询event事件，并通过挂载的perf钩子打印输出 */
while ((err = perf_buffer__poll(pb, 100)) >= 0)
;
printf("Error polling perf buffer: %d\n", err);
cleanup:
perf_buffer__free(pb);
runqslower_bpf__destroy(obj);
return err != 0;
}【1】
【2】
TIPs
总结
- 首先编写BPF程序，定义BPF的maps和sections；
- 编译BPF程序，然后根据编译出来的.o文件生成对应的skeleton头文件
- 用户空间的程序包含skeleton头文件，可以通过
const volatile定义的全局变量(在加载BPF程序前)给BPF程序传递参数。需要注意的是，全局变量在BPF程序加载后是不可变的，如果要在加载之后给BPF程序传递数据，可以使用map(全局变量就是为了节省在给BPF程序传递常量的情况下存在的，节省查找map的开销)； - 用户空间执行
open->load->attach->destroy来控制BPF程序的生命周期。
下一篇将使用BPF CO-RE方式重写一个XDP程序。
参考
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2021-02-02 ，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录