こんにちは、ティアフォーでパートタイムエンジニアをしている石川です。
本記事では、楽に「動的ライブラリ(及び実行バイナリ)の特定の関数をフックして何かしらの処理をする」手法について紹介していきます。
この記事は、同じくパートタイムエンジニアの西村さんによる作業の成果を元にして、石川が執筆したものです。ソースコードや図のいくつかも西村さんによる貢献です。
また、ティアフォーでは「自動運転の民主化」をともに実現していく、学生パートタイムエンジニアを常時募集しています。自動運転を実現するためには、Softwareに関してはOSからMiddlewareそしてApplicationに至るまで、Hardwareに関してはSensorからECUそして車両に至るまで異なるスキルを持つ様々な人々が不可欠です。もしご興味があれば以下のページからコンタクトいただければと思います。 https://tier4.jp/careers/tier4.jp
この記事の主題
この記事の主題は、楽に「動的ライブラリ(及び実行バイナリ)の特定の関数をフックして何かしらの処理をする」( ≒ ソースコードを改変したりビルドしなおしたりしない) ことです。
ティアフォーを中心として開発されているAutowareは、ROS (Robot Operating System) という通信ミドルウェアに依存しています。今回はAutowareの性能解析という業務を進める上で、ROS1内部の特定区間の処理時間(実時間)を計測したくなり、上記目的を達成する手段を用意することになりました。
手段としては、eBPF(extended Berkeley Packet Filter) を利用した方法と、LD_PRELOADを利用した方法の2種類を紹介します。
eBPFを利用した方法
まず、eBPFとは何かについて説明します。 eBPFとは、Linux kernel内部で実行される独自の命令セットを持つ仮想マシンにおいて、ユーザ空間から実行プログラムを送り込み、指定したevent (下図におけるkprobes, uprobes, tracepoints, perf_events に対応)にフックするなどして実行することができる機能のことです。
eBPFのユーザは、BPF bytecodeをbpf(2)によってkernelに送りこみ、安全なコードであるかのverifyとコンパイル/実行をkernelが行います。eBPFのユーザがBPF bytecodeを直接書くのは稀で、通常はC言語likeな高級言語で記述することができます。eBPFのユーザプログラムとのデータのやりとりは、“maps” と呼ばれるデータ構造を通して容易に行うようになっています。
今回は、「動的ライブラリ(及び実行バイナリ)の特定の関数をフックして何かしらの処理をする」ことを実現したかったので、フックイベントとしてuprobesを使用します。uprobesとは、eBPFとは独立した概念であり、ユーザアプリケーションにbreakpointを仕込んで、kernelでハンドリングするための仕組みです。
uprobesにフックしてeBPFのコードを実行するためのAPIなどについては、iovisor/bccリポジトリにおいてある bcc Reference Guide を見ればよいです。今回は、ユーザコードの関数の開始地点にフックを仕込む attach_uprobe() を利用します。
例として、ROS1内部の ros::Publication::publish(ros::SerializedMessage const&) をフックして文字列をprintするだけのコードを示します。
from __future__ import print_function from bcc import BPF libroscpp_path = "/path/to/libroscpp.so" bpf_text = """ #include <uapi/linux/ptrace.h> #include <linux/sched.h> int publish(struct pt_regs *ctx) { uint64_t pid = bpf_get_current_pid_tgid(); bpf_trace_printk("publish() called: msg addr - %p :pid/tid %d", (void *)ctx->di, pid); return 0; } """ b = BPF(text=bpf_text) b.attach_uprobe(name=libroscpp_path, sym="_ZN3ros11Publication7publishERNS_17SerializedMessageE", fn_name="publish") while 1: try: (task, pid, cpu, flags, ts, msg) = b.trace_fields() except ValueError: continue print("%s" % msg)
以上のスクリプトを実行することにより、libroscpp.so を動的リンクして実行されている全てのROS1プログラムにおける、指定した関数をフックすることができます。
ところで、libroscpp.soはC++によって書かれたライブラリなので、関数名がマングリングされており (上のコードにおけるattach_uprobe()の第二引数に渡している “_ZN3ros11Publication7publishERNS_17SerializedMessageE”)、面倒なことにビルド済みのバイナリからマングリングされた関数名を調べなくてはなりません。
# Get offset address of the specified function $ nm -D /path/to/binary_file | c++filt | grep <function_name> # Get the mangled function name $ nm -D /path/to/binary_file | grep <offset>
さて、このようにすることで実行バイナリや動的リンクされるsoファイルに変更を加えることなく、Pythonスクリプトを実行するのみでユーザ空間の関数をフックすることに成功しましたが、計測のオーバーヘッドの面でデメリットがあります。
関数にフックしてeBPFのコードを実行する際には、毎回ユーザ空間からカーネル空間へのコンテキストスイッチが生じるため、高い頻度で呼ばれる関数のフックに対して無視できないオーバーヘッドが生じてしまうという点です(eBPFはkprobesに対するフックやkernel eventなどの、kernel空間で発生するイベントに対するフックに適している選択肢と言えます)。
これを解決するため、次はユーザ空間のみで完結するLD_PRELOADを用いた方法を検討します。
LD_PRELOADを利用した方法
LD_PRELOADとは、/lib/ld-linux.so.2 の動作を制御する環境変数の一つであり、環境変数LD_PRELOADに共有オブジェクトを指定して実行することによって先にLD_PRELOADで指定した共有オブジェクトがリンクされます。
したがって、その共有オブジェクトに同名の関数を定義しておくことによって、元の共有ライブラリに定義されている関数の代わりにLD_PRELOADで指定した共有オブジェクトの関数を実行させることが可能になります。
さて、今回実現したいのは「特定の関数の開始時にフックして、ある処理を挟む」ことなので、LD_PRELOADで置き換えた関数の実行後に元の関数に処理を戻します。これを実現する方法はいくつかありますが、一番簡単な方法はRTLD_NEXTハンドルを引数に指定して、dlsym(3)を呼びだすことです (LD_PRELOADで置き換えた後の関数内でその関数を呼び出しても、置き換えた後の関数を再帰的に呼びだすだけなので無限ループしてしまいます)。
RTLD_NEXTハンドルとはGNU拡張による特殊なハンドルであり、現在の共有オブジェクトの次の共有オブジェクト以降で発見されるシンボルの値を取ってきます。つまり今回の場合では、LD_PRELOADで置き換える前の本来の共有ライブラリにて定義されている関数のシンボルを取ってくることができます。ちなみにRTLD_NEXTハンドルを利用するためには、GNU拡張を有効にするためにdlfcn.h をincludeする前に #define _GNU_SOURCE する必要があります。
さて、eBPFを使用したときと同様にros::Publication::publish(ros::SerializedMessage const&) が呼び出された際にフックして特定の処理を挟むことをしてみます。以下のコードを共有ライブラリとしてコンパイルし、LD_PRELOADに指定しつつROS1のプログラムを実行します。
#define _GNU_SOURCE #include<dlfcn.h> #include<iostream> using publish_type = bool (*)(void* ,void*); extern "C" bool _ZN3ros11Publication7publishERNS_17SerializedMessageE(void *p, void *q) { void *orig_pub = dlsym(RTLD_NEXT, "_ZN3ros11Publication7publishERNS_17SerializedMessageE"); std::cout << "call detect: publish 1st arg: " << p << " 2nd arg: " << q << std::endl; return ((publish_type)orig_pub)(p,q); }
マングリングされたそれぞれの関数を再定義し (マングリングされた関数名を調べる方法は前章のとおり)、再定義した関数の内部から dlsym(3)にRTLD_NEXTハンドルを渡して元の関数のシンボルを取得し、その元の関数を呼びだしています。
まとめ
本記事では、楽に「動的ライブラリ(及び実行バイナリ)の特定の関数をフックして何かしらの処理をする」ことを目的とし、eBPFによる方法とLD_PRELOADによる方法を紹介しました。本記事において紹介した手法は、動的ライブラリをリンクして実行するアプリケーション全般(やアプリケーションの実行バイナリ自体)に使用できるものですので、幅広く応用先があると思います。