ring_buffer_reserve()/ring_buffer_submit()でより効率的に処理する

ここまで基本的なデータのやり取りの仕方がわかりました。次に実際にもっと詳細なデータを取得できるようになりたいところですが，その前により効率的なリングバッファーの使い方を学んでおきましょう。

先ほどのBPFプログラムでは，データを準備したらbpf_ringbuf_output()を用いてリングバッファーにそのデータを保存していました。このとき，次のような問題が起きる可能性があります。

• バッファーが既に一杯のときEAGAINでエラーになる
• 内部でデータのmemcpy()が実施される

前者はEAGAINが返ってきたらもう一度実施すれば良いだけではあるのですが，どれくらい待てば良いかという問題があります。待つだけなら良いのですが，それだけ待ったあとのデータに意味があるのかどうかというのも検討すべきでしょう。後者については，メモリ効率や速度に影響する問題です。

そこで，より効率的なAPIとしてring_buffer_reserve()/ring_buffer_submit()が用意されています。

• ring_buffer_reserve()は指定したサイズの領域をリングバッファーに確保してくれるAPIです。戻り値として確保した領域のポインタが返されるため，そこにデータを直接書き込むことでメモリコピーの回数を減らせます。また，確保失敗時はエラーになるため，より早い段階で処理を継続するか諦めるかを判断できます。
• ring_buffer_submit()は渡されたリングバッファーのアドレスが確定したと判断し，必要に応じてユーザーランドにデータの受信を通知します。

他にも確保したものの，使用せずに解放したい場合はring_buffer_discard()が使えます。

さて，実際にBPFプログラムのコードを書き換えると次のようになります。

int syscalls__execve(struct trace_event_raw_sys_enter *ctx)
{
    struct event *event = NULL;

    /* リングバッファーの領域確保 */
    event = bpf_ringbuf_reserve(&events, sizeof(*event), 0);
    if (!event)
        return 0;

    event->pid = bpf_get_current_pid_tgid() >> 32;

    /* リングバッファーの中身を確定し，ユーザーランドに通知 */
    bpf_ringbuf_submit(event, 0);

    return 0;
}

今回の例だとあまり利点は感じられないのですが，使用方法のイメージはつくと思います。ring_buffer_reserve()の最後の引数は，現時点では使用しないため0にしておかなければなりません。0以外を指定するとエラーとなります。ring_buffer_submit()の最後の引数は，ring_buffer_output()と同じ通知タイミングを示すフラグです。

実際にビルドして実行した結果は，変更前と同じになることを確認しておきましょう。ここまでの変更点はリポジトリで，v1.2タグを付けています。

一点，注意しなければならないのは，ユーザープログラム側がリングバッファーにアクセスできるのは「予約した順」であり「コミットした順」ではないということです。たとえば複数のBPFプログラムが同じリングバッファーを共有していて，あるプログラムAがring_buffer_reserve()を呼んだあとに，別のプログラムBがring_buffer_reserve()を呼ぶとします。プログラムBはすぐにring_buffer_submit()でデータを送ったとしても，ユーザーランドプログラムにデータ受信が通知されるのは，プログラムAがring_buffer_submit()を呼び出したあとになります。

ちなみにring_buffer_output()は内部的に，ring_buffer_reserve()/ring_buffer_submit()と同等の処理を一度に行っています。

構造体データへの移植性の高いアクセス方法

BPFオブジェクトの再利用性を高めるために考えなくてはならないのが，「⁠構造体のメンバーへのアクセス方法」です。ユーザランドに公開されているデータ構造はともかくとして，カーネル内部で使われているデータ構造は比較的頻繁に更新されます。よってカーネルのバージョンごとのデータ構造を意識しなくてはなりません。

BPF CO-REではそのようなカーネルバージョンの差異を吸収するためのマクロを用意しています。マクロを使ってメンバーにアクセスすることで，BPFオブジェクトは異なるカーネルバージョンでも「それなりに動く」ようになります。

具体的な例として，現在のタスクの親プロセスのPID（PPID）を取得する方法を考えてみましょう。現在のプロセスのタスク構造体は「bpf_get_current_task()」で取得できます。タスク構造体（task）が取得できたら，あとは「task->real_parent->tgid」とたどるだけで親プロセスのPID（この例ではThread Group ID）を取得できます。しかしながらこれだとタスク構造体の中身が変わってしまうときちんとアクセスできない可能性があります。

そこでBPF CO-REではbpf_core_read.hで定義されている「BPF_CORE_READ()」という可変長引数を受け取れるマクロを使います。いくつかの例を示します。前段がC言語での一般的な表現で，後段がBPF_CORE_READ()マクロを利用した表現です。

例1：単純な構造体メンバーへのアロー演算子を利用したアクセス
src->a;
BPF_CORE_READ(src, a);

例2：多段のメンバーアクセス
src->a->b->c->d->e;
BPF_CORE_READ(src, a, b, c, d, e);

例3：アロー演算子とドット演算子の混在
src->a.b->c.d.e->f;
BPF_CORE_READ(src, a.b, c.d.e, f);

これだけ読めば，おおよそ使い方はイメージできるでしょう。なお，C言語のマクロ表現の制約上，アクセスできるメンバーの段数は9段までとなっています。またNULL終端された文字列を取得したいならBPF_CORE_READ_STR_INTO()が，ユーザー空間のメモリ用にBPF_CORE_READ_USER()とBPF_CORE_READ_USER_STR_INTO()が用意されていますので，用途に応じて使い分けると良いでしょう。

実際にPPIDを取得するようにした場合の変更点を取り上げます。まず，BPFプログラムは次のようになります。

    struct task_struct *task;

    /* リングバッファーの領域確保 */
    event = bpf_ringbuf_reserve(&events, sizeof(*event), 0);
    if (!event)
        return 0;

    event->pid = bpf_get_current_pid_tgid() >> 32;

    /* 現在のタスク構造体取得 */
    task = (struct task_struct *)bpf_get_current_task();
    /* task->real_parent->tgidへのアクセス */
    event->ppid = (pid_t)BPF_CORE_READ(task, real_parent, tgid);

ちなみに共通の構造体定義は「execsnoop.h」として共通のヘッダーファイルにまとめ，execsnoop.bpf.cとexecsnoop.cの双方からインクルードするようにしました。ユーザーランドプログラム側のexecsnoop.hはスケルトンヘッダーファイル（execsnoop.skel.h）よりあとにインクルードするようにしてください。ヘッダーファイルは次のようになっています。

/* SPDX-License-Identifier: CC0-1.0 */
#ifndef __EXECSNOOP_H__
#define __EXECSNOOP_H__

struct event {
    pid_t pid;
    pid_t ppid;
};
#endif /* __EXECSNOOP_H__ */

ppidメンバーが増えただけですね。またユーザーランドプログラム側も，そのPPIDも一緒に出力するようにしただけです。

int handle_event_cb(void *ctx, void *data, size_t size)
{
    struct event *event = data;
    fprintf(stdout, "% 8d  % 8d\n", event->pid, event->ppid);

    return 0;
}

int main(void)
（中略）
    fprintf(stdout, "%-8s  %-8s\n", "PID", "PPID");

実際にビルドして実行しみましょう。リポジトリではv1.3タグを付けています。

$ git checkout r695/3 v1.3
$ make
$ sudo ./execsnoop
PID       PPID
    6021      2276
    6022      6021
    6023      2276
    6024      6022

PIDだけでなく，PPIDも表示されるようになりました。