eBPF: 從 BPF to BPF Calls 到 Tail Calls

作者簡介：

李兆龍，西安郵電大學軟件工程專業大四學生，Linux 興趣小組成員。

本作品採用知識共享署名 - 非商業性使用 - 相同方式共享 4.0 國際許可協議進行許可。

本作品 (李兆龍博文, 由李兆龍創作)，由李兆龍確認，轉載請註明版權。

引言

這篇文章首先介紹尾調用的一般限制和用法，並與 BPF to BPF calls 做對比，最後給出一個我對內核源碼中 tail call sample 做的一個修改版本（應用 CO-RE）。（我在學習尾調用的時候苦於沒有一個能跑起來的簡單易懂的例子，所以最後自己擼了一個，這個版本我認爲是目前能找到的所有例子裏對初學者最友好，邏輯最清晰的一個）。

Tail Call

BPF 提供了一種在內核事件和用戶程序事件發生時安全注入代碼的能力，這就讓非內核開發人員也可以對內核進行控制，但是因爲 11 個 64 位寄存器和 32 位子寄存器、一個程序計數器和一個 512 字節的 BPF 堆棧空間以及 100 萬條指令（5.1+），遞歸深度 33 的固有限制，使得可以實現的邏輯是有限的（非圖靈完備）。

內核棧是很寶貴的，一般 BPF 到 BPF 的會使用額外的棧幀，尾調用最大的優勢就是其複用了當前的棧幀並跳轉至另外一個 eBPF 程序，可以在 [5] 中看到如下描述：

The important detail that it's not a normal call, but a tail call. The kernel stack is precious, so this helper reuses the current stack frame and jumps into another BPF program without adding extra call frame.

eBPF 程序都是獨立驗證的（調用者的堆棧和寄存器中的值被調用者不可訪問），所以狀態的傳遞一般可以使用 per-CPU map 傳遞，TC 還可以使用 skb_buff->cb 這樣的特殊數據項去傳遞數據 [8]；其次類型相同的 BPF 程序纔可以尾調用，而且還要與 JIT 編譯器相匹配，因此一個給定的 BPF 程序要麼是 JIT 編譯執行，要麼是解釋器執行（invoke interpreted programs）

尾調用的步驟需要用戶態和內核態配合，主要由兩個部分組成：

用戶態：BPF_MAP_TYPE_PROG_ARRAY類型的特殊 map，存儲自定義 index 到 bpf_program_fd 的到映射
內核態：bpf_tail_call輔助函數，其負責跳轉到另一個 eBPF 程序，其函數定義是這樣的static long (*bpf_tail_call)(void *ctx, void *prog_array_map, __u32 index)，ctx 是上下文，prog_array_map 是前面說的BPF_MAP_TYPE_PROG_ARRAY類型的 map，用於用戶態設置跳轉程序和用戶自定義 index 的映射，index 就是用戶自定義索引了。

bpf_tail_call如果運行成功，內核立即運行新 eBPF 程序的第一條指令（永遠不會返回到之前的程序）。如果跳轉的目標程序不存在（即 index 在 prog_array_map 中不存在），或者此程序鏈已達到最大尾調用數，則調用可能會失敗，如果調用失敗，調用者繼續執行後續指令。

[5] 中我們可以看到如下文字：

The chain of tail calls can form unpredictable dynamic loops therefore tail_call_cnt is used to limit the number of calls and currently is set to 32.

這個限制在內核中由宏 MAX_TAIL_CALL_CNT （用戶空間不可訪問）定義，當前設置爲 32（我並不知道這個所謂的unpredictable dynamic loops是什麼）。

上面提到了尾調用可以省內核棧空間，除了這一點以外我認爲其最大的優勢如下：

用於增加可執行的 eBPF 程序指令的最大執行數
eBPF 程序編排

上面兩個優勢不是在我在想當然的胡謅，舉個兩個例子分別解釋上面兩個觀點：

[10] 在 BMC 中有一個 eBPF 程序中有一個大循環，雖然 eBPF 程序只有 142 行，但是字節碼已經到了七十多萬行，如果不做邏輯拆分會在 verify 階段被拒絕。
[9] 中提出了一種通過配置文件任意組合 eBPF 程序的 eBPF 編排策略，給我的感覺是他們像做一個三方存儲，然後可以通過配置自動拉去需要的 eBPF 程序，然後自動編排，載入，執行，這裏編排的過程是用尾調用實現的，基本的流程如下：

BPF to BPF Calls

從 Linux 4.16 和 LLVM 6.0 開始，這個限制得到了解決，加載器、校驗器、解釋器和 JIT 中都開始支持函數調用。

最大的優勢是減小了生成的 BPF 代碼大小，因此對 CPU instruction cache 更友好。BPF 輔助函數的調用約定也適用於 BPF 函數間調用，即 r1 - r5 用於傳遞參數，返回結果放到 r0。r1 - r5 是 scratch registers，r6 - r9 像往常一樣是保留寄存器。最大嵌套調用深度是 8。調用方可以傳遞指針（例如，指向調用方的棧幀的指針）給被調用方。

尾調用的缺點是生成的程序鏡像大，但是省內存；BPF to BPF Calls 的優點是鏡像小，但是內存消耗大。內核 5.9 以前不允許 tail Call 和 BPF to BPF Call 調用協同工作，在 5.10 以後的 X86 架構上，允許同時使用這兩種調用類型。

在 [7] 中提到同時使用兩種類型有一定限制，否則會導致內核棧溢出：

我們以上圖的調用鏈舉例，這裏所說的限制就是每一個子程序的棧空間（ stack size）不能超過 256 字節（如果校驗器檢測到 bpf to bpf 調用，那主程序也會被當做子程序），這個限制使得 BPF 程序調用鏈最多能使用 8KB 的棧空間，計算方式：256 byte/stack 乘以尾調用數量上限 33。如果沒有這個限制，BPF 程序將使用 512 字節棧空間，最終消耗最多 16KB 的總棧空間，在某些架構上會導致棧溢出。

這裏需要提的一點是兩種類型在同時使用時到底是如何省內存的，舉個例子，subfunc1 執行 Tail Call 調用 func2，此時 subfunc 的棧幀已經被 func2 複用了，然後 func2 執行 BPF to BPF Calls 調用 subfunc2，此時第三個棧幀被創建，然後執行 Tail Call 調用 func3，五個邏輯過程使用了三個棧，這就節省了內存。

然後因爲開始時調用 subfunc1，所以最終的程序執行權仍然會回到 func1。

CO-RE Sample

我們可以在 11 中看到 kernel 中對於 Tail Call 的官方實例，我使用 libbpf CO-RE 特性修改了 [11]，使得 User 程序更容易理解 libbpf 使用 Tail Call。其次雖然掛載 ebpf 程序可以使用其他命令輔助掛載，比如 tc，prctl，但是示例程序我認爲可能老老實實調接口掛容易理解些，我們使用[11] 作爲例子。

例子中用到了seccomp filter，這個過濾器用於減少內核中系統調用暴露於應用程序的範圍，說人話就是限制每個進程使用的系統調用（參數修改是 task_struct 級別的），支持SECCOMP_SET_MODE_STRICT和SECCOMP_SET_MODE_FILTER兩種過濾模式，用於限定子集 (read，write， _exit，sigreturn) 的過濾和 eBPF 形式的過濾。[3]中提到可以避免 TOCTTOU，因爲 open 被限制了，自然也就沒有 TOCTTOU 錯誤的風險了[6]。

[3] 中對seccomp filter的解釋是這樣的：

System call filtering isn't a sandbox. It provides a clearly defined mechanism for minimizing the exposed kernel surface. It is meant to be a tool for sandbox developers to use.

[2] 中提到：

Designed to sandbox compute-bound programs that deal with untrusted byte code.

tracex5_kern.bpf.c

 #include "vmlinux.h"
 #include <bpf/bpf_helpers.h>
 #include <bpf/bpf_tracing.h>
 #include <bpf/bpf_core_read.h>
 #include "bpf_helpers.h"
 
 char LICENSE[] SEC("license") = "Dual BSD/GPL";
 
 struct {
 __uint(type, BPF_MAP_TYPE_PROG_ARRAY);
 __uint(max_entries, 1024);
 __type(key, u32);
 __type(value, u32);
 } progs SEC(".maps");
 
 
 SEC("kprobe/__seccomp_filter")
 int BPF_KPROBE(__seccomp_filter, int this_syscall, const struct seccomp_data *sd, const bool recheck_after_trace)
 {
 // 這裏注意ebpf程序棧空間只有512字節，太大這裏會報錯的，可以自己調大一點看看
 char comm_name[30];
 bpf_get_current_comm(comm_name, sizeof(comm_name));
 // 調用失敗以後會直接 fall through
     bpf_tail_call(ctx, &progs, this_syscall);
 
 char fmt[] = "syscall=%d common=%s\n";
 bpf_trace_printk(fmt, sizeof(fmt), this_syscall, comm_name);
     return 0;
 }
 
 /* we jump here when syscall number == __NR_write */
 SEC("kprobe/SYS__NR_write")
 int bpf_func_SYS__NR_write(struct pt_regs *ctx)
 {
 struct seccomp_data sd;
 bpf_probe_read(&sd, sizeof(sd), (void *)PT_REGS_PARM2(ctx));
 if (sd.args[2] > 0) {
 char fmt[] = "write(fd=%d, buf=%p, size=%d)\n";
 bpf_trace_printk(fmt, sizeof(fmt),
  sd.args[0], sd.args[1], sd.args[2]);
 }
 return 0;
 }
 
 SEC("kprobe/SYS__NR_read")
 int bpf_func_SYS__NR_read(struct pt_regs *ctx)
 {
 struct seccomp_data sd;
 bpf_probe_read(&sd, sizeof(sd), (void *)PT_REGS_PARM2(ctx));
 if (sd.args[2] > 0 && sd.args[2] <= 1024) {
 char fmt[] = "read(fd=%d, buf=%p, size=%d)\n";
 bpf_trace_printk(fmt, sizeof(fmt),
  sd.args[0], sd.args[1], sd.args[2]);
 }
 return 0;
 }
 
 SEC("kprobe/SYS__NR_open")
 int bpf_func_SYS__NR_open(struct pt_regs *ctx)
 {
 struct seccomp_data sd;
 bpf_probe_read(&sd, sizeof(sd), (void *)PT_REGS_PARM2(ctx));
 char fmt[] = "open(fd=%d, path=%p)\n";
 bpf_trace_printk(fmt, sizeof(fmt), sd.args[0], sd.args[1]);
 return 0;
 }

tracex5_user.c

 #include <stdlib.h>
 #include <unistd.h>
 #include <string.h>
 #include <signal.h>
 #include <time.h>
 #include <assert.h>
 #include <errno.h>
 #include <sys/resource.h>
 #include <linux/if_link.h>
 #include <linux/limits.h>
 
 #include <bpf/libbpf.h>
 #include "trace.skel.h"
 
 #define BPF_SYSFS_ROOT "/sys/fs/bpf"
 
 enum {
     SYS__NR_read = 3,
     SYS__NR_write = 4,
 SYS__NR_open = 5,
 };
 
 struct bpf_progs_desc {
 char name[256];
 enum bpf_prog_type type;
 int map_prog_idx;
 struct bpf_program *prog;
 };
 
 static struct bpf_progs_desc progs[] = {
 {"kprobe/__seccomp_filter", BPF_PROG_TYPE_KPROBE, -1, NULL},
 {"kprobe/SYS__NR_read", BPF_PROG_TYPE_KPROBE, SYS__NR_read, NULL},
 {"kprobe/SYS__NR_write", BPF_PROG_TYPE_KPROBE, SYS__NR_write, NULL},
 {"kprobe/SYS__NR_open", BPF_PROG_TYPE_KPROBE, SYS__NR_open, NULL},
 };
 
 static int libbpf_print_fn(enum libbpf_print_level level, const char *format, va_list args)
 {
 return vfprintf(stderr, format, args);
 }
 
 static volatile bool exiting = false;
 
 static void sig_handler(int sig)
 {
 exiting = true;
 }
 
 int main(int argc, char **argv)
 {
 struct trace_bpf *skel;
     int map_progs_fd, main_prog_fd, prog_count;
 int err;
 
 // 設置一些debug信息的回調
 libbpf_set_print(libbpf_print_fn);
 
 signal(SIGINT, sig_handler);
 signal(SIGTERM, sig_handler);
 
 // Load and verify BPF application
 skel = trace_bpf__open();
 if (!skel) {
 fprintf(stderr, "Failed to open and load BPF skeleton\n");
 return 1;
 }
 
 // Load and verify BPF programs
 err = trace_bpf__load(skel);
 if (err) {
 fprintf(stderr, "Failed to load and verify BPF skeleton\n");
 goto cleanup;
 }
 
     map_progs_fd = bpf_object__find_map_fd_by_name(skel->obj, "progs");
     prog_count = sizeof(progs) / sizeof(progs[0]);
     for (int i = 0; i < prog_count; i++) {
 progs[i].prog = bpf_object__find_program_by_title(skel->obj, progs[i].name);
 if (!progs[i].prog) {
 fprintf(stderr, "Error: bpf_object__find_program_by_title failed\n");
 return 1;
 }
 bpf_program__set_type(progs[i].prog, progs[i].type);
    }
 
     for (int i = 0; i < prog_count; i++) {
         int prog_fd = bpf_program__fd(progs[i].prog);
 if (prog_fd < 0) {
 fprintf(stderr, "Error: Couldn't get file descriptor for program %s\n", progs[i].name);
 return 1;
 }
         
         // -1指的是主程序
 if (progs[i].map_prog_idx != -1) {
 unsigned int map_prog_idx = progs[i].map_prog_idx;
 if (map_prog_idx < 0) {
 fprintf(stderr, "Error: Cannot get prog fd for bpf program %s\n", progs[i].name);
 return 1;
 }
             // 給 progs map 的 map_prog_idx 插入 prog_fd
 err = bpf_map_update_elem(map_progs_fd, &map_prog_idx, &prog_fd, 0);
 if (err) {
 fprintf(stderr, "Error: bpf_map_update_elem failed for prog array map\n");
 return 1;
 }
 }
    }
 
 // 只載入主程序，尾調用不載入，所以不可以調用trace_bpf__attach
 struct bpf_link* link = bpf_program__attach(skel->progs.__seccomp_filter);
 if (link == NULL) {
 fprintf(stderr, "Error: bpf_program__attach failed\n");
 return 1;
 }
 
 while(exiting){
 // 寫個裸循環會喫巨多CPU的
 sleep(1);
 }
 
 cleanup:
 // Clean up
 trace_bpf__destroy(skel);
 
 return err < 0 ? -err : 0;
 }

執行如下指令：

clang -g -O2 -target bpf -D__TARGET_ARCH_x86 -I/usr/src/kernels/5.4.119-19-0009.1（改成自己的）/include/ -idirafter /usr/local/include -idirafter /usr/include -c trace.bpf.c -o trace.bpf.o
gen skeleton trace.bpf.o > trace.skel.h
clang -g -O2 -Wall -I . -c trace.c -o trace.o
clang -Wall -O2 -g trace.o -lelf -lz -o trace
cat /sys/kernel/debug/tracing/trace_pipe 可以看到預期輸出。

tail call costs in eBPF

[13] 評估了爲緩解 Spectre 缺陷而引入的一些優化給 eBPF 的尾調用性能帶來了多少性能損耗（Spectre 是指大多數 CPU（英特爾、AMD、ARM）上存在的一系列利用硬件漏洞的漏洞）。

這篇文章還沒看，有精力了拜讀一下，先插個眼。

總結

eBPF 環境現在我還是很頭疼，開始在 VMware Fusion 上跑虛擬機，但是 m1 不支持 VMware Tools，導致使用極其不便，而且鏡像也需要特殊的修改才能用到 m1 上，用高版本的內核沒那麼容易；其次買的雲服務器版本只到 5.4.119，一些 eBPF 的特性也沒法用；至於 MAC m1 雙系統，我不太有精力去喫這個螃蟹了，一是資料少，其次 Tail Calls 和 BPF to BPF Calls 同時調用目前也只支持 X86，收益也不大。

好吧，我承認，歸根結底，還是不想花錢去買 Parallels Desktop。

參考：

linux 安全之 seccomp
Using seccomp to limit the kernel attack surface
SECure COMPuting with filters
seccomp(2) — Linux manual page
bpf: introduce bpf_tail_call() helper
Time-of-check to time-of-use wiki
cilium document
eBPF: Traffic Control Subsystem
Introducing Walmart’s L3AF Project: Control plane, chaining eBPF programs, and open-source plans
BMC: Accelerating Memcached using Safe In-kernel Caching and Pre-stack Processing
kernel tracex5_kern.c
kernel sockex3_kern.c
Evaluation of tail call costs in eBPF

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/oWze9RxIAsSbM82mlaKwWQ

引言

Tail Call

BPF to BPF Calls

CO-RE Sample

tail call costs in eBPF

總結

猜你喜歡