在BPF的可移植性和CO-RE一文的末尾提到了一个名为runqslower的工具，该工具用于展示在CPU run队列中停留的时间大于某一值的任务。现在以该工具来展示如何使用BPF CO-RE。
目录
环境
本地测试的话，建议采用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】：
- 用户空间可以且只能通过
BPF skeletob方式来访问和更新全局变量，更新后的变量会立即反应到BPF侧。需要注意的是，全局变量只是BPF侧的变量，用户空间其实是通过.rodata间接来操作这类变量，意味着如果用户侧也定义了一个相同的变量，则会被视为两个独立的变量。 - 用户空间操作全局变量的一般操作如下：
struct <name> *skel = <name>__open(); if (!skel) /* handle errors */ skel->rodata->my_cfg.feature_enabled = true; skel->rodata->my_cfg.pid_to_filter = 123; if (<name>__load(skel)) /* handle errors */
- 从下面解析的ELF文件的内容可以看到，使用
const volatile声明的全局变量min_us和targ_pid位于.rodata(read-only)段，用户空间的应用可以在加载BPF程序前读取或更新BPF侧的全局变量，runqslower通过这种方式设置了min_us的值。
# llvm-objdump -t runqslower.bpf.o runqslower.bpf.o: file format elf64-bpf SYMBOL TABLE: 0000000000000050 l .text 0000000000000000 LBB0_3 00000000000000a0 l .text 0000000000000000 LBB0_4 0000000000000100 l .text 0000000000000000 LBB1_3 0000000000000150 l .text 0000000000000000 LBB1_4 00000000000001f8 l .text 0000000000000000 LBB2_4 0000000000000248 l .text 0000000000000000 LBB2_5 00000000000002e0 l .text 0000000000000000 LBB2_8 0000000000000388 l .text 0000000000000000 LBB2_9 0000000000000000 l d .text 0000000000000000 .text 0000000000000000 g O license 0000000000000004 LICENSE 0000000000000020 g O .maps 0000000000000018 events #名为 events 的 maps 0000000000000160 g F .text 0000000000000238 handle__sched_switch #handle__sched_switch 代码段 0000000000000000 g F .text 00000000000000b0 handle__sched_wakeup #handle__sched_wakeup 代码段 00000000000000b0 g F .text 00000000000000b0 handle__sched_wakeup_new #handle__sched_wakeup_new 代码段 0000000000000000 g O .rodata 0000000000000008 min_us #全局变量 min_us 0000000000000000 g O .maps 0000000000000020 start #名为 start 的 maps 0000000000000008 g O .rodata 0000000000000004 targ_pid #全局变量 targ_pid
skel->rodata用于只读变量；skel->bss用于初始值为0的可变量；skel->data用于初始值非0的可变量。
【2】：
- 通常一个map具有如下属性：
- 类型
- 最大元素数目
- key的字节大小
- value的字节大小
- 可以使用如下接口对maps进行操作：
bpf_map_operation_elem(&some_map, some, args);
- 一般常见的接口如下，可以在内核/用户空间对maps中的元素进行增删改查操作：
bpf_map_lookup_elem bpf_map_update_elem bpf_map_delete_elem bpf_map_push_elem bpf_map_pop_elem bpf_map_peek_elem
【3】：
- 约定的SEC的命名方式如下，libbpf可以根据SEC字段自动检测BPF程序类型，然后关联特定的BPF程序类型，不同的程序类型决定了BPF程序的第一个入参关联的上下文。使用bpftool feature可以查看支持不同程序类型的BPF辅助函数。更多参见section_defs。
tp/<category>/<name>用于Tracepoints;kprobe/<func_name>用于kprobe ，kretprobe/<func_name>用于kretprobe;raw_tp/<name>用于原始Tracepoint;cgroup_skb/ingress,cgroup_skb/egress，以及整个cgroup/<subtype>程序族。
tp_btf/sched_wakeup、tp_btf/sched_wakeup_new、tp_btf/sched_switch跟踪了系统任务上下文切换相关的事件，可以在/sys/kernel/debug/tracing/events/sched下找到对应的事件定义。- 像
int handle__sched_wakeup(u64 *ctx)这样的用法仍然属于BCC的使用方式，BPF支持使用BPF_KPROBE/BPF_KRETPROBE来像内核函数一样给BPF程序传参，主要用于tp_btf/fentry/fexitBPF程序。用法如下(更多方式，参见这里)：
SEC("kprobe/xfs_file_open") int BPF_KPROBE(xfs_file_open, struct inode *inode, struct file *file) { ....... }
- 使用
BPF_KPROBE时需要保证，第一个参数必须是一个系统调用，由于tp_btf/sched_wakeup、tp_btf/sched_wakeup_new、tp_btf/sched_switch并不是系统调用，而是跟踪事件，因此不能使用BPF_KPROBE。
用户空间代码
用户侧代码通常包含如下头文件：
#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】
- 用户空间需要接收内核空间传递过来的信息，使用生成的skeleton头文件的如下函数操作内核程序：
<name>__open()– 创建并打开 BPF 应用(例如的runqslower的runqslower_bpf__open()函数);<name>__load()– 初始化，加载和校验BPF 应用部分;<name>__attach()– 附加所有可附加的BPF程序 (可选，可以直接使用libbpf API作更多控制);<name>__destroy()– 分离BPF 程序并使用其使用的所有资源。
obj = runqslower_bpf__open();，其中obj的结构体位于runqslower.skel.h，是根据BPF程序自动生成的，内容如下：
struct runqslower_bpf { struct bpf_object_skeleton *skeleton; struct bpf_object *obj; struct { struct bpf_map *start; struct bpf_map *events; struct bpf_map *rodata; } maps; /* 对应BPF程序中定义的两个.maps以及一个全局只读section .rodata */ struct { struct bpf_program *handle__sched_wakeup; struct bpf_program *handle__sched_wakeup_new; struct bpf_program *handle__sched_switch; } progs; /* 对应BPF程序使用SEC()定义的3个BPF程序 */ struct { struct bpf_link *handle__sched_wakeup; struct bpf_link *handle__sched_wakeup_new; struct bpf_link *handle__sched_switch; } links; /* 链接到BPF程序的link，可以使用bpftool link命令查看，可以显示链接的BPF程序，进程等信息 */ struct runqslower_bpf__rodata { __u64 min_us; pid_t targ_pid; } *rodata; /* 对应BPF程序的.rodata section */ };
- 其实整个处理过程简单归结为：创建
runqslower_bpf.skeleton对象，赋值runqslow的信息(maps，progs，links，rodata)，其中skeleton->data编码了BPF程序，后续会被解析为Efile对象；然后加载BPF程序，进行初始化和校验；然后attach之后，BPF程序开始正式运行。
【2】
- Skeleton 可以用于大部分场景，但有一个例外：perf events。这种情况下，不能使用
struct <name>__bpf中的links，而应该自定义一个struct bpf_link *links[]，原因是perf_event需要在每个CPU上进行操作。例如llcstat.c
static int open_and_attach_perf_event(__u64 config, int period, struct bpf_program *prog, struct bpf_link *links[]) { struct perf_event_attr attr = { .type = PERF_TYPE_HARDWARE, .freq = 0, .sample_period = period, .config = config, }; 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; }
TIPs
- 非内核5.3以上的版本中的循环都必须添加
#pragma unroll标志
#pragma unroll for (i = 0; i < 10; i++) { ... }
- bpf_printk 调试，仅适用于非生产环境
char comm[16]; u64 ts = bpf_ktime_get_ns(); u32 pid = bpf_get_current_pid_tgid(); bpf_get_current_comm(&comm, sizeof(comm)); bpf_printk("ts: %lu, comm: %s, pid: %d\n", ts, comm, pid);
- BPF涉及到的主要头文件有：
libbpf.h: 定义了通用的ebpf ELF对象的加载操作libbpf/include/uapi/linux/bpf.h: 定义了BPF的各种类型(prog_type，map_type，attach_type以及设计的结构体定义等)libbpf/src/bpf.h: 定义了通用的eBPF ELF操作bpf_core_read.h: 定义了读取内核结构的方法bpf_helpers.h: 定义了BPF程序用到的宏SEC()
总结
- 首先编写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程序。
