[PATCH RFC v3 8/8] selftests/kcov_dataflow: add trigger-view.py

[Yunseong Kim]

Wire the kcov_dataflow test modules into a standard kselftest target and
add the viewer they are built on.

trigger-view.py loads a module with finit_module(), enables recording
(KCOV_DF_ENABLE, or KCOV_DF_REMOTE_ENABLE with "--remote" argument),
writes the module's debugfs trigger file, unloads it, and decodes
the buffer into a call tree or, with "--raw", a flat record list.
It symbolizes PCs against a vmlinux with addr2line or against
/proc/kallsyms, adding the running kernel's KASLR offset back to
the recorded PCs (runtime _text minus the link-time _text from
System.map / nm / a per-architecture default), and demangles
Rust v0 symbols.

test_modules.py is the KTAP runner: it drives run_capture() from
trigger-view.py for each module and compares the captured arguments,
struct fields and return values against the values the module uses
emitting the module's call tree as diagnostics. A test passes only
when the data came back intact, not merely when records appeared.
Modules that were not built are reported as SKIP.

The Makefile builds user_ioctl and binderfs as ordinary kselftest
programs and the .ko modules via Kbuild against KDIR (the Rust ones only
when the configured kernel has CONFIG_RUST=y); config lists the kernel
options the tests need, and settings raises the per-test timeout for the
in-VM runs. The whole target builds with

  make -C tools/testing/selftests TARGETS=kcov_dataflow \
       LLVM=1 CC=clang [RUSTC=... RUST_LIB_SRC=...]

and runs under run_kselftest.sh.

Assisted-by: Claude:claude-opus-4-6 [kiro-chat]
Signed-off-by: Yunseong Kim <yunseong.kim@xxxxxxxx>
---
 tools/testing/selftests/kcov_dataflow/.gitignore   |   4 +
 tools/testing/selftests/kcov_dataflow/Kbuild       |  10 +
 tools/testing/selftests/kcov_dataflow/Makefile     |  46 ++
 tools/testing/selftests/kcov_dataflow/README.rst   |  69 ++
 tools/testing/selftests/kcov_dataflow/config       |  11 +
 tools/testing/selftests/kcov_dataflow/settings     |   1 +
 .../selftests/kcov_dataflow/test_modules.py        | 249 +++++++
 .../selftests/kcov_dataflow/trigger-view.py        | 755 +++++++++++++++++++++
 8 files changed, 1145 insertions(+)

diff --git a/tools/testing/selftests/kcov_dataflow/.gitignore b/tools/testing/selftests/kcov_dataflow/.gitignore
new file mode 100644
index 0000000000000..4f2957a017957
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/.gitignore
@@ -0,0 +1,4 @@
+# SPDX-License-Identifier: GPL-2.0
+user_ioctl/user_ioctl
+binderfs/binderfs_test
+__pycache__/
diff --git a/tools/testing/selftests/kcov_dataflow/Kbuild b/tools/testing/selftests/kcov_dataflow/Kbuild
new file mode 100644
index 0000000000000..2e19e9008fdca
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/Kbuild
@@ -0,0 +1,10 @@
+# SPDX-License-Identifier: GPL-2.0
+#
+# Test modules, built as external modules against the configured kernel tree
+# by the selftest Makefile ("make -C $(KDIR) M=$(CURDIR) modules"). Every
+# directory opts its object into dataflow instrumentation with
+# KCOV_DATAFLOW_<object>.o := y, the same per-file switch in-tree code uses.
+obj-m				+= rust_ffi_contract/
+obj-m				+= eight_struct_args_c/
+obj-$(CONFIG_RUST)		+= eight_struct_args_rust/
+obj-$(CONFIG_RUST)		+= rust_kworker_remote/
diff --git a/tools/testing/selftests/kcov_dataflow/Makefile b/tools/testing/selftests/kcov_dataflow/Makefile
new file mode 100644
index 0000000000000..fc979e2d4ecc3
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/Makefile
@@ -0,0 +1,46 @@
+# SPDX-License-Identifier: GPL-2.0
+#
+# kcov_dataflow selftests
+#
+# user_ioctl and binderfs are ordinary kselftest programs. The test modules
+# (one per directory, listed in Kbuild) are built by kbuild against KDIR and
+# are loaded, triggered and checked by test_modules.py; trigger-view.py is
+# the interactive viewer the runner is built on.
+#
+# KDIR is the configured kernel build tree. It defaults to the source tree
+# this directory lives in; point it at the O= directory for out-of-tree
+# builds. Pass the same LLVM=1 CC=clang [RUSTC= RUST_LIB_SRC=] the kernel was
+# built with so that kbuild picks the toolchain that has the trace-args and
+# trace-ret passes.
+KDIR ?= $(abspath ../../../..)
+
+TEST_GEN_PROGS := user_ioctl/user_ioctl binderfs/binderfs_test
+TEST_PROGS := test_modules.py
+TEST_FILES := trigger-view.py
+
+CFLAGS += -Wall -O2 $(KHDR_INCLUDES)
+
+# The .ko files kbuild produces for KDIR's configuration, so that they are
+# built by "all" and copied by "install"; the Rust modules need CONFIG_RUST.
+KMODS := rust_ffi_contract eight_struct_args_c
+ifneq ($(shell grep -s ^CONFIG_RUST=y $(KDIR)/.config),)
+KMODS += eight_struct_args_rust rust_kworker_remote
+endif
+TEST_GEN_FILES := $(foreach m,$(KMODS),$(m)/$(m).ko)
+
+include ../lib.mk
+
+ifneq ($(wildcard $(KDIR)/.config),)
+$(TEST_GEN_FILES): modules
+modules:
+	$(Q)$(MAKE) -C $(KDIR) M=$(CURDIR) modules
+clean_modules:
+	$(Q)$(MAKE) -C $(KDIR) M=$(CURDIR) clean
+else
+$(TEST_GEN_FILES):
+	@echo "SKIP $(notdir $@): no configured kernel tree at $(KDIR), set KDIR="
+clean_modules:
+endif
+
+clean: clean_modules
+.PHONY: modules clean_modules
diff --git a/tools/testing/selftests/kcov_dataflow/README.rst b/tools/testing/selftests/kcov_dataflow/README.rst
new file mode 100644
index 0000000000000..1929a357aca47
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/README.rst
@@ -0,0 +1,69 @@
+.. SPDX-License-Identifier: GPL-2.0
+
+KCOV-Dataflow Selftests
+=======================
+
+Selftests for ``/sys/kernel/debug/kcov_dataflow`` (see
+Documentation/dev-tools/kcov-dataflow.rst).
+
+Layout
+------
+
+Makefile, Kbuild
+    kselftest build: the C programs are built by lib.mk, the test modules
+    (one directory each, listed in Kbuild) by kbuild against ``KDIR``.
+user_ioctl/
+    ioctl interface test (kselftest harness, TAP).
+binderfs/
+    binder ioctls under recording (TAP).
+rust_ffi_contract/, eight_struct_args_c/, eight_struct_args_rust/,
+rust_kworker_remote/
+    test modules; each README.rst says what the module exercises.
+test_modules.py
+    KTAP runner: loads every module, triggers it with recording active and
+    checks the captured arguments, struct fields and return values against
+    the values the module uses. Modules that are not built are SKIPped.
+trigger-view.py
+    Interactive viewer the runner is built on (call tree or ``--raw``
+    records, kallsyms/addr2line symbolization, ``--remote`` capture).
+
+Kernel
+------
+
+The kernel and the modules must be built with a clang that has the
+trace-args/trace-ret passes (and, for the Rust modules, a rustc built
+against that LLVM). The config fragment ``config`` lists what the tests
+need; with virtme-ng::
+
+    vng --build --config tools/testing/selftests/kcov_dataflow/config \
+        LLVM=1 CC=clang RUSTC=$RUSTC RUST_LIB_SRC=$RUST_LIB_SRC
+
+Build
+-----
+
+From the kernel tree, with the same toolchain variables::
+
+    make LLVM=1 headers
+    make -C tools/testing/selftests TARGETS=kcov_dataflow \
+        LLVM=1 CC=clang RUSTC=$RUSTC RUST_LIB_SRC=$RUST_LIB_SRC
+
+``KDIR`` defaults to the source tree; pass ``KDIR=<O dir>`` for out-of-tree
+builds. The Rust modules are built only when ``KDIR/.config`` has
+``CONFIG_RUST=y``. ``make ... install INSTALL_PATH=<dir>`` produces a
+self-contained tree with ``run_kselftest.sh``.
+
+Run
+---
+
+On the target (root, debugfs mounted)::
+
+    vng --user root --exec \
+        "tools/testing/selftests/kcov_dataflow/test_modules.py"
+    tools/testing/selftests/kcov_dataflow/user_ioctl/user_ioctl
+    tools/testing/selftests/kcov_dataflow/binderfs/binderfs_test
+
+or, from an installed tree, ``run_kselftest.sh -c kcov_dataflow``.
+``test_modules.py -t <module> -C 8`` runs one module and echoes eight
+records of context around each module record; ``trigger-view.py <module>
+[--raw] [-C N] [--remote] [--vmlinux vmlinux]`` shows the capture
+without checking it.
diff --git a/tools/testing/selftests/kcov_dataflow/config b/tools/testing/selftests/kcov_dataflow/config
new file mode 100644
index 0000000000000..7f3a2fda0641d
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/config
@@ -0,0 +1,11 @@
+CONFIG_KCOV=y
+CONFIG_KCOV_DATAFLOW_ARGS=y
+CONFIG_KCOV_DATAFLOW_RET=y
+CONFIG_KCOV_DATAFLOW_INSTRUMENT_ALL=y
+CONFIG_KCOV_DATAFLOW_NO_INLINE=y
+CONFIG_DEBUG_INFO_DWARF5=y
+CONFIG_DEBUG_FS=y
+CONFIG_MODULES=y
+CONFIG_ANDROID_BINDER_IPC=y
+CONFIG_ANDROID_BINDERFS=y
+CONFIG_RUST=y
diff --git a/tools/testing/selftests/kcov_dataflow/settings b/tools/testing/selftests/kcov_dataflow/settings
new file mode 100644
index 0000000000000..694d70710ff08
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/settings
@@ -0,0 +1 @@
+timeout=300
diff --git a/tools/testing/selftests/kcov_dataflow/test_modules.py b/tools/testing/selftests/kcov_dataflow/test_modules.py
new file mode 100755
index 0000000000000..13cb706a06ff6
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/test_modules.py
@@ -0,0 +1,249 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0
+"""
+test_modules.py - run the kcov_dataflow test modules, one KTAP test each.
+
+Every module is loaded, triggered with recording active and unloaded by
+trigger-view.py's run_capture(). The records that belong to the module are
+then compared with the values its trigger function passes and returns, so a
+test passes only when the instrumented arguments, struct field expansions
+and return values came back intact through the kcov_dataflow buffer. The
+module's call tree is echoed as KTAP diagnostics.
+
+    ./test_modules.py                 # all modules
+    ./test_modules.py -t rust_ffi_contract -C 8 --vmlinux vmlinux
+
+Modules that were not built (no CONFIG_RUST, no toolchain) are reported as
+SKIP; a kernel without /sys/kernel/debug/kcov_dataflow skips everything.
+"""
+import argparse
+import contextlib
+import importlib.util
+import io
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, "..", "kselftest"))
+import ksft  # noqa: E402
+
+
+def _load_trigger_view():
+    spec = importlib.util.spec_from_file_location(
+        "trigger_view", os.path.join(HERE, "trigger-view.py"))
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+tv = _load_trigger_view()
+
+
+class Check:
+    """Collects expectation failures for one module."""
+
+    def __init__(self):
+        self.failures = []
+
+    def eq(self, what, got, want):
+        if got != want:
+            self.failures.append(f"{what}: got {fmt(got)}, want {fmt(want)}")
+
+    def true(self, what, cond):
+        if not cond:
+            self.failures.append(what)
+
+
+def fmt(v):
+    if isinstance(v, list):
+        return "[" + ", ".join(fmt(x) for x in v) + "]"
+    if isinstance(v, int):
+        return f"0x{v:x}"
+    return str(v)
+
+
+def entries(cap, recs, func):
+    return [r for r in recs if r["type"] == tv.DF_TYPE_ENTRY and func in cap.funcs(r)]
+
+
+def rets(cap, recs, func):
+    return [r["val"] for r in recs if r["type"] == tv.DF_TYPE_RET and func in cap.funcs(r)]
+
+
+def flat_sum(n):
+    """sf_n() returns a->a + b->b + ... over s1..sn: 0x11 + 0x22 + ..."""
+    return sum(0x11 * k for k in range(1, n + 1))
+
+
+def nested_sum(n, _memo={}):
+    """
+    stf_n()/stpf_n() return field0 (0x11 * n) plus the recursive sums of the
+    embedded st1..st(n-1); the same values are used for the value-nested and
+    the pointer-linked towers.
+    """
+    if n not in _memo:
+        _memo[n] = 0x11 * n + sum(nested_sum(k) for k in range(1, n))
+    return _memo[n]
+
+
+def check_struct_family(cap, recs, c, p, flat_ns, stpf8_runs):
+    """
+    Shared expectations for eight_struct_args_c (p="") and
+    eight_struct_args_rust (p="r"): @flat_ns are the sf_N called by the
+    trigger, @stpf8_runs how often the pointer-linked tower is walked.
+    """
+    for n in flat_ns:
+        ents = entries(cap, recs, f"{p}sf_{n}")
+        c.true(f"{p}sf_{n}: ENTRY records", bool(ents))
+        for k in range(n):
+            # arg k is a struct s(k+1) * whose fields are 0x11, 0x22, ...
+            got = [r["vals"] for r in ents if r["arg_idx"] == k]
+            c.true(f"{p}sf_{n} arg[{k}]: ENTRY record", bool(got))
+            for vals in got:
+                c.eq(f"{p}sf_{n} arg[{k}] expanded fields", vals,
+                     [0x11 * (j + 1) for j in range(k + 1)])
+        # rustc may alias identical bodies (rsf_1 == rstf_1), so the RET
+        # list can carry the alias's calls too: check every value.
+        got = rets(cap, recs, f"{p}sf_{n}")
+        c.true(f"{p}sf_{n} RET values all {fmt(flat_sum(n))}: {fmt(got)}",
+               bool(got) and all(v == flat_sum(n) for v in got))
+
+    for fam, calls in ((f"{p}stf", 1), (f"{p}stpf", stpf8_runs)):
+        for n in range(1, 9):
+            got = rets(cap, recs, f"{fam}_{n}")
+            c.true(f"{fam}_{n}: RET records", bool(got))
+            c.true(f"{fam}_{n} RET values all {fmt(nested_sum(n))}: {fmt(got)}",
+                   all(v == nested_sum(n) for v in got))
+        c.eq(f"{fam}_8 RET count", len(rets(cap, recs, f"{fam}_8")), calls)
+
+    for f in (f"{p}sf_fwd", f"{p}sf_fwd_inner"):
+        c.eq(f"{f} RET", rets(cap, recs, f), [flat_sum(4)])
+
+    c.true(f"{p}sf_ret_struct: ENTRY records",
+           bool(entries(cap, recs, f"{p}sf_ret_struct")))
+    c.true(f"{p}sf_ret_struct: RET record",
+           bool(rets(cap, recs, f"{p}sf_ret_struct")))
+
+
+def check_eight_struct_args_c(cap, recs, c):
+    check_struct_family(cap, recs, c, "", range(1, 9), stpf8_runs=2)
+
+
+def check_eight_struct_args_rust(cap, recs, c):
+    check_struct_family(cap, recs, c, "r", (1, 2, 4, 8), stpf8_runs=1)
+
+
+def check_rust_ffi_contract(cap, recs, c):
+    """
+    ffi_alloc_buf(&alloc = {NULL, 0, 0, 0}, 256, 16, is_async=1) records
+    data_size + offsets_size and returns 0 without filling alloc->buffer;
+    ffi_check_result() then sees {NULL, 0x110, 0, 0}. The records must show
+    the violated contract at both boundaries.
+    """
+    ents = entries(cap, recs, "ffi_alloc_buf")
+    by_arg = {r["arg_idx"]: r for r in ents}
+    c.eq("ffi_alloc_buf ENTRY arg indexes", sorted(by_arg), [0, 1, 2, 3])
+    if 0 in by_arg:
+        c.eq("ffi_alloc_buf arg[0] struct ffi_alloc fields",
+             by_arg[0]["vals"], [0, 0, 0, 0])
+    if 1 in by_arg:
+        c.eq("ffi_alloc_buf arg[1] data_size", by_arg[1]["val"], 256)
+    if 2 in by_arg:
+        c.eq("ffi_alloc_buf arg[2] offsets_size", by_arg[2]["val"], 16)
+    if 3 in by_arg:
+        c.eq("ffi_alloc_buf arg[3] is_async", by_arg[3]["val"], 1)
+    c.eq("ffi_alloc_buf RET (claims success)", rets(cap, recs, "ffi_alloc_buf"), [0])
+
+    ents = entries(cap, recs, "ffi_check_result")
+    c.true("ffi_check_result: ENTRY record", bool(ents))
+    for r in ents:
+        c.eq("ffi_check_result arg[0] {buffer NULL: contract violated, "
+             "data_size, free_async, flags}", r["vals"], [0, 0x110, 0, 0])
+    got = rets(cap, recs, "ffi_check_result")
+    c.true(f"ffi_check_result RET -EFAULT: {fmt(got)}",
+           len(got) == 1 and got[0] & 0xffffffff == 0xfffffff2)
+
+
+def check_rust_kworker_remote(cap, recs, c):
+    """
+    The trigger only queues a work item and waits; the records come from the
+    kworker that called kcov_df_remote_start(REMOTE_HANDLE). All three phases
+    of CompositeStore must show up (v0-mangled names keep the method names).
+    """
+    c.true("records captured from the kworker", bool(recs))
+    names = set().union(*(cap.funcs(r) for r in recs)) if recs else set()
+    for phase in ("populate", "update", "drain"):
+        c.true(f"CompositeStore::{phase} recorded",
+               any("CompositeStore" in n and phase in n for n in names))
+
+
+TESTS = (
+    ("rust_ffi_contract", False, check_rust_ffi_contract),
+    ("eight_struct_args_c", False, check_eight_struct_args_c),
+    ("eight_struct_args_rust", False, check_eight_struct_args_rust),
+    ("rust_kworker_remote", True, check_rust_kworker_remote),
+)
+
+
+def diag_tree(cap, recs, vmlinux):
+    out = io.StringIO()
+    with contextlib.redirect_stdout(out):
+        tv.print_tree(recs, cap.syms, vmlinux, {}, cap.ko_path,
+                      cap.mod_text_start)
+    for line in out.getvalue().splitlines():
+        ksft.print_msg(line)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    parser.add_argument("-t", "--test", action="append",
+                        help="run only this module (repeatable)")
+    parser.add_argument("-C", "--context", type=int, default=0,
+                        help="echo N records before/after each module record")
+    parser.add_argument("--vmlinux", help="vmlinux for addr2line and KASLR")
+    args = parser.parse_args()
+
+    tests = [t for t in TESTS if not args.test or t[0] in args.test]
+    ksft.print_header()
+    ksft.set_plan(len(tests))
+
+    skip_all = None
+    if not os.path.exists(tv.KCOV_DF_PATH):
+        skip_all = f"{tv.KCOV_DF_PATH} not available (CONFIG_KCOV_DATAFLOW_ARGS/RET)"
+    elif os.geteuid() != 0:
+        skip_all = "must run as root"
+
+    vmlinux = tv.find_vmlinux(args.vmlinux)
+    for name, remote, check in tests:
+        if skip_all:
+            ksft.test_result_skip(f"{name}: {skip_all}")
+            continue
+        ko = tv.find_module(name)
+        if not ko:
+            ksft.test_result_skip(f"{name}: {name}.ko not built")
+            continue
+        try:
+            cap = tv.run_capture(ko, remote=remote, vmlinux=vmlinux,
+                                 log=ksft.print_msg)
+        except OSError as e:
+            ksft.test_result_fail(f"{name}: {e}")
+            continue
+
+        recs = cap.module_records()
+        ksft.print_msg(f"{name}: {cap.total_words} words, {len(cap.records)} "
+                       f"records, {len(recs)} from {name} "
+                       f"(kaslr_offset=0x{cap.kaslr_offset:x})")
+        diag_tree(cap, cap.context_records(args.context) if args.context
+                  else recs, vmlinux)
+
+        c = Check()
+        check(cap, recs, c)
+        for f in c.failures:
+            ksft.print_msg(f"FAIL {name}: {f}")
+        ksft.test_result(not c.failures, name)
+
+    ksft.finished()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tools/testing/selftests/kcov_dataflow/trigger-view.py b/tools/testing/selftests/kcov_dataflow/trigger-view.py
new file mode 100755
index 0000000000000..b17e49da402d7
--- /dev/null
+++ b/tools/testing/selftests/kcov_dataflow/trigger-view.py
@@ -0,0 +1,755 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0
+"""
+trigger-view.py - Load a test module, trigger it with kcov_dataflow
+recording active, then pretty-print the captured records.
+
+Usage:
+    python3 trigger-view.py eight_struct_args_c
+    python3 trigger-view.py rust_ffi_contract --raw -C 8
+    python3 trigger-view.py rust_kworker_remote --remote
+    python3 trigger-view.py <module> --vmlinux vmlinux --kaslr-offset 0x...
+
+run_capture() does the work and is also what test_modules.py drives:
+  1. Opens /sys/kernel/debug/kcov_dataflow, inits and mmaps the buffer
+  2. Loads the module via finit_module() (its init noise is not recorded)
+  3. Enables recording: KCOV_DF_ENABLE for this task, or with --remote
+     KCOV_DF_REMOTE_ENABLE with handle REMOTE_HANDLE, which the module's
+     kworker opens with kcov_df_remote_start(REMOTE_HANDLE)
+  4. Writes the trigger file(s) the module created under TRIGGER_DIR
+  5. Disables recording and unloads the module
+  6. Parses the records (layout: include/uapi/linux/kcov_dataflow.h)
+
+The CLI then prints them as a call tree, or flat with --raw, with kallsyms
+symbol resolution and addr2line source lines (vmlinux / module .ko).
+
+Recorded PCs have the KASLR offset removed (same as mainline kcov), so
+the runtime offset is derived from /proc/kallsyms and System.map / vmlinux
+(or a per-architecture default) and added back for symbolization; use
+--kaslr-offset to override. Records must contain at least one value word
+and one of the three record types, otherwise the parser resyncs word by
+word (e.g. after a userspace reset of area[0] mid-run).
+"""
+import os
+import sys
+import struct
+import ctypes
+import ctypes.util
+import argparse
+import fcntl
+import platform
+import subprocess
+import shutil
+
+# Constants -- must match include/uapi/linux/kcov_dataflow.h
+DF_TYPE_CMP = 0xC
+DF_TYPE_ENTRY = 0xE
+DF_TYPE_RET = 0xF
+MAGIC_BAD = 0xBADADD85
+BUF_SIZE = 1048576  # 1M words = 8MB
+
+# Record header word: bits 0-23 seq | 28-31 type | 32-47 nvals |
+# 48-55 arg/ret size | 56-63 arg index. Word 1 is the pc (KASLR offset
+# removed, like mainline kcov), word 2 the traced pointer (ENTRY/RET) or the
+# comparison type (CMP), then nvals value words.
+def hdr_seq(h):
+    return h & 0x00FFFFFF
+
+def hdr_type(h):
+    return (h >> 28) & 0xF
+
+def hdr_nvals(h):
+    return (h >> 32) & 0xFFFF
+
+def hdr_size(h):
+    return (h >> 48) & 0xFF
+
+def hdr_arg_idx(h):
+    return (h >> 56) & 0xFF
+
+RECORD_HDR_WORDS = 3
+
+# Runtime KASLR offset (see kaslr_offset()); added back to every recorded pc
+# so /proc/kallsyms lookups work, subtracted again for addr2line on vmlinux.
+KASLR_OFFSET = 0
+
+# Ioctl numbers
+def _IOR(t, nr, size):
+    return (2 << 30) | (ord(t) << 8) | nr | (size << 16)
+
+def _IOW(t, nr, size):
+    return (1 << 30) | (ord(t) << 8) | nr | (size << 16)
+
+def _IO(t, nr):
+    return (ord(t) << 8) | nr
+
+KCOV_DF_INIT_TRACK = _IOR('d', 1, 8)
+KCOV_DF_ENABLE = _IO('d', 100)
+KCOV_DF_DISABLE = _IO('d', 101)
+KCOV_DF_REMOTE_ENABLE = _IOW('d', 102, 8)  # arg: pointer to a __u64 handle
+KCOV_DF_REMOTE_DISABLE = _IO('d', 103)
+
+KCOV_DF_PATH = "/sys/kernel/debug/kcov_dataflow"
+
+# Every test module creates its trigger file(s) in this debugfs directory;
+# writing to them runs the instrumented test functions.
+TRIGGER_DIR = "/sys/kernel/debug/kcov_dataflow_test"
+
+# Remote handle registered with KCOV_DF_REMOTE_ENABLE; must match the
+# kcov_df_remote_start(1) call in the rust_kworker_remote test module
+# (KCOV_SUBSYSTEM_COMMON, instance 1).
+REMOTE_HANDLE = 1
+
+# syscall numbers
+_machine = platform.machine()
+if _machine == "aarch64":
+    SYS_FINIT_MODULE = 273
+    SYS_DELETE_MODULE = 106
+else:  # x86_64
+    SYS_FINIT_MODULE = 313
+    SYS_DELETE_MODULE = 176
+
+SELFTEST_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
+def load_kallsyms():
+    """Load kernel symbols for PC resolution."""
+    syms = []
+    try:
+        with open("/proc/kallsyms") as f:
+            for line in f:
+                parts = line.split()
+                if len(parts) >= 3:
+                    addr = int(parts[0], 16)
+                    name = parts[2]
+                    mod = parts[3].strip("[]") if len(parts) > 3 else ""
+                    syms.append((addr, name, mod))
+    except (PermissionError, FileNotFoundError):
+        pass
+    syms.sort()
+    return syms
+
+
+def runtime_text(syms):
+    """Runtime address of _text from kallsyms, 0 if hidden."""
+    return next((a for a, n, m in syms if n == "_text" and not m), 0)
+
+
+# Link-time address of _text per architecture, used only when neither
+# System.map nor vmlinux is available: x86_64 __START_KERNEL
+# (__START_KERNEL_map + CONFIG_PHYSICAL_START), arm64 KIMAGE_VADDR.
+LINKTIME_TEXT_DEFAULT = {
+    "x86_64": 0xffffffff81000000,
+    "aarch64": 0xffff800080000000,
+}
+
+
+def linktime_text(vmlinux=None):
+    """Return (link-time address of _text, source description) or (0, "")."""
+    rel = os.uname().release
+    candidates = []
+    if vmlinux:
+        candidates.append(os.path.join(os.path.dirname(vmlinux) or ".", "System.map"))
+    candidates += ["System.map", f"/boot/System.map-{rel}",
+                   f"/usr/lib/debug/boot/System.map-{rel}"]
+    for sm in candidates:
+        try:
+            with open(sm) as f:
+                for line in f:
+                    parts = line.split()
+                    if len(parts) == 3 and parts[2] == "_text":
+                        return int(parts[0], 16), sm
+        except (OSError, ValueError):
+            continue
+    if vmlinux and shutil.which("nm"):
+        try:
+            r = subprocess.run(["nm", "--defined-only", vmlinux],
+                               capture_output=True, text=True, timeout=300)
+            for line in r.stdout.splitlines():
+                parts = line.split()
+                if len(parts) == 3 and parts[2] == "_text":
+                    return int(parts[0], 16), f"nm {vmlinux}"
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+    link = LINKTIME_TEXT_DEFAULT.get(platform.machine(), 0)
+    return link, f"{platform.machine()} default" if link else ""
+
+
+def kaslr_offset(syms, vmlinux=None):
+    """
+    Runtime KASLR offset: recorded PCs have it removed (kcov's
+    canonicalize_ip()), /proc/kallsyms has it applied. Computed as the
+    runtime _text (kallsyms) minus the link-time _text (System.map, nm
+    vmlinux, or the architecture default). KASLR offsets are 2 MiB aligned
+    on x86_64 and arm64, which is used as a sanity check on the result.
+    """
+    runtime = runtime_text(syms)
+    if not runtime:
+        print("# warning: _text not in /proc/kallsyms (kptr_restrict?); "
+              "PCs will not symbolize", file=sys.stderr)
+        return 0
+    link, source = linktime_text(vmlinux)
+    if not link:
+        print(f"# warning: no System.map/vmlinux and no default _text for "
+              f"{platform.machine()}; pass --kaslr-offset", file=sys.stderr)
+        return 0
+    off = runtime - link
+    if off % (2 << 20):
+        print(f"# warning: kaslr offset 0x{off:x} from {source} is not 2 MiB "
+              f"aligned; check CONFIG_PHYSICAL_START/KIMAGE_VADDR or pass "
+              f"--kaslr-offset", file=sys.stderr)
+    return off
+
+
+# Rust symbol demangling via llvm-cxxfilt or rustfilt
+_demangler = None
+
+def _init_demangler():
+    global _demangler
+    for tool in ["llvm-cxxfilt", "rustfilt", "c++filt"]:
+        path = shutil.which(tool)
+        if path:
+            _demangler = path
+            return
+    _demangler = ""
+
+_demangled = {}
+
+def demangle(name):
+    """Demangle a Rust/C++ symbol name (memoized: one process per name)."""
+    global _demangler
+    if _demangler is None:
+        _init_demangler()
+    if not _demangler or not name.startswith("_R"):
+        return name
+    if name not in _demangled:
+        try:
+            r = subprocess.run([_demangler, name], capture_output=True,
+                               text=True, timeout=2)
+            _demangled[name] = r.stdout.strip() if r.returncode == 0 else name
+        except (OSError, subprocess.TimeoutExpired):
+            _demangled[name] = name
+    return _demangled[name]
+
+
+def find_vmlinux(vmlinux=None):
+    """Locate vmlinux for addr2line: explicit path, else the usual places."""
+    if vmlinux:
+        return vmlinux
+    for p in ["vmlinux", "/boot/vmlinux", "/usr/lib/debug/boot/vmlinux"]:
+        if os.path.exists(p):
+            return p
+    return None
+
+
+def _a2l_target(pc, vmlinux, ko_path, mod_text_base):
+    """(binary, address in it) to symbolize pc with, or None."""
+    if ko_path and mod_text_base and pc >= mod_text_base:
+        return ko_path, pc - mod_text_base
+    if vmlinux:
+        return vmlinux, pc - KASLR_OFFSET  # vmlinux holds link-time addresses
+    return None
+
+
+def resolve_lines(pcs, vmlinux, cache, ko_path=None, mod_text_base=0):
+    """
+    Resolve every pc in @pcs to file:line into @cache, one addr2line run
+    per binary: a DWARF5 vmlinux takes hundreds of ms to open, so one
+    process per record does not scale to thousands of records.
+    """
+    todo = {}
+    for pc in pcs:
+        if pc in cache:
+            continue
+        cache[pc] = ""
+        tgt = _a2l_target(pc, vmlinux, ko_path, mod_text_base)
+        if tgt:
+            todo.setdefault(tgt[0], []).append((pc, tgt[1]))
+    for binary, pairs in todo.items():
+        try:
+            r = subprocess.run(
+                ["addr2line", "-e", binary] + [f"0x{a:x}" for _, a in pairs],
+                capture_output=True, text=True, timeout=300)
+        except (subprocess.TimeoutExpired, FileNotFoundError):
+            continue
+        for (pc, _), loc in zip(pairs, r.stdout.splitlines()):
+            loc = loc.strip()
+            if loc and loc != "??:0" and loc != "??:?":
+                # Shorten path: keep only filename:line
+                cache[pc] = loc.rsplit("/", 1)[-1]
+
+
+def resolve_line(pc, vmlinux, cache, ko_path=None, mod_text_base=0):
+    """Resolve one PC to source file:line using addr2line (cached)."""
+    if pc not in cache:
+        resolve_lines([pc], vmlinux, cache, ko_path, mod_text_base)
+    return cache[pc]
+
+
+def get_kernel_meta():
+    """Collect kernel build metadata."""
+    meta = {"release": os.uname().release}
+    try:
+        with open("/proc/version") as f:
+            v = f.read().strip()
+        meta["version"] = v
+        # Extract compiler version
+        if "gcc" in v.lower():
+            meta["compiler"] = v.split("(")[1].split(")")[0] if "(" in v else ""
+        elif "clang" in v.lower():
+            idx = v.lower().find("clang")
+            meta["compiler"] = v[idx:idx+30].split(")")[0]
+    except OSError:
+        pass
+    return meta
+
+
+def print_kernel_meta(meta, ko_path=None):
+    """Print kernel metadata header/footer."""
+    print(f"# {'=' * 60}")
+    print(f"# Kernel: {meta.get('release', 'unknown')}")
+    print(f"# Build:  {meta.get('version', 'unknown')[:80]}")
+    if meta.get('compiler'):
+        print(f"# Compiler: {meta['compiler']}")
+    # Read rustc version from .ko .comment section
+    if ko_path:
+        try:
+            r = subprocess.run(
+                ["readelf", "-p", ".comment", ko_path],
+                capture_output=True, text=True, timeout=5)
+            for line in r.stdout.splitlines():
+                if "rustc" in line:
+                    ver = line.split("]", 1)[-1].strip()
+                    print(f"# Rustc: {ver}")
+                    break
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+    print(f"# {'=' * 60}")
+
+
+def lookup(pc, syms):
+    """Nearest kallsyms entry <= pc as (name, offset, module) or None."""
+    if not syms:
+        return None
+    lo, hi = 0, len(syms) - 1
+    while lo < hi:
+        mid = (lo + hi + 1) // 2
+        if syms[mid][0] <= pc:
+            lo = mid
+        else:
+            hi = mid - 1
+    addr, name, mod = syms[lo]
+    if addr > pc:
+        return None
+    return name, pc - addr, mod
+
+
+def symbolize(pc, syms):
+    """Find nearest symbol <= pc. Returns (display_name, module_tag)."""
+    hit = lookup(pc, syms)
+    if not hit:
+        return f"0x{pc:x}", ""
+    name, offset, mod = hit
+    dname = demangle(name)
+    display = f"{dname}+0x{offset:x}" if offset else dname
+    return display, f" [{mod}]" if mod else ""
+
+
+def format_val(v):
+    """Format a captured value."""
+    if v == MAGIC_BAD:
+        return "FAULT"
+    if v == 0:
+        return "0x0"
+    return f"0x{v:x}"
+
+
+def find_module(name):
+    """
+    Find the .ko for test @name: <name>/<name>.ko in the source tree, or
+    <name>.ko next to this script in an installed (make install) tree.
+    """
+    for ko_path in (os.path.join(SELFTEST_DIR, name, f"{name}.ko"),
+                    os.path.join(SELFTEST_DIR, f"{name}.ko")):
+        if os.path.exists(ko_path):
+            return ko_path
+    return None
+
+
+def finit_module(ko_path):
+    """Load a kernel module via finit_module syscall."""
+    libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True)
+    fd = os.open(ko_path, os.O_RDONLY)
+    ret = libc.syscall(SYS_FINIT_MODULE, fd, b"", 0)
+    os.close(fd)
+    if ret != 0:
+        errno = ctypes.get_errno()
+        raise OSError(errno, f"finit_module({ko_path}): {os.strerror(errno)}")
+
+
+def delete_module(name):
+    """Unload a kernel module."""
+    libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True)
+    ret = libc.syscall(SYS_DELETE_MODULE, name.encode(), 0)
+    if ret != 0:
+        errno = ctypes.get_errno()
+        raise OSError(errno, f"delete_module({name}): {os.strerror(errno)}")
+
+
+def trigger_module():
+    """
+    Write to every trigger file the loaded module created under TRIGGER_DIR.
+    Opened without O_CREAT: debugfs directories have no ->create, so a
+    "w"-mode open of a missing name fails with EOPNOTSUPP, not ENOENT.
+    """
+    try:
+        names = sorted(os.listdir(TRIGGER_DIR))
+    except OSError:
+        names = []
+    hits = []
+    for n in names:
+        path = os.path.join(TRIGGER_DIR, n)
+        try:
+            fd = os.open(path, os.O_WRONLY)
+        except OSError:
+            continue
+        try:
+            os.write(fd, b"1")
+        finally:
+            os.close(fd)
+        hits.append(path)
+    if not hits:
+        raise FileNotFoundError(f"no trigger file under {TRIGGER_DIR}")
+    return hits
+
+
+def parse_records(buf, total_words):
+    """Parse the ring buffer into a list of records."""
+    records = []
+    pos = 1
+    end = min(1 + total_words, BUF_SIZE)
+    while pos + RECORD_HDR_WORDS <= end:
+        hdr = buf[pos]
+        rtype = hdr_type(hdr)
+        num_vals = hdr_nvals(hdr)
+
+        # Every record the kernel writes has nvals >= 1 and a known type;
+        # anything else is garbage (e.g. a userspace reset mid-run): resync.
+        if rtype not in (DF_TYPE_ENTRY, DF_TYPE_RET, DF_TYPE_CMP) \
+                or num_vals == 0 or pos + RECORD_HDR_WORDS + num_vals > end:
+            pos += 1
+            continue
+
+        pc = int(buf[pos + 1]) + KASLR_OFFSET
+        ptr = int(buf[pos + 2])  # ENTRY/RET: traced pointer; CMP: cmp type
+        if rtype == DF_TYPE_CMP:
+            pos += RECORD_HDR_WORDS + num_vals
+            continue
+
+        # Valid records always have a non-zero PC (kernel text address)
+        if pc == 0:
+            pos += 1
+            continue
+
+        vals = [int(buf[pos + RECORD_HDR_WORDS + vi]) for vi in range(num_vals)]
+        records.append({
+            "type": rtype,
+            "seq": hdr_seq(hdr),
+            "pc": pc,
+            "ptr": ptr,
+            "arg_idx": hdr_arg_idx(hdr),
+            "size": hdr_size(hdr),
+            "val": vals[0],
+            "vals": vals,
+        })
+        pos += RECORD_HDR_WORDS + num_vals
+    return records
+
+
+class Capture:
+    """Everything run_capture() collected for one module run."""
+
+    def __init__(self, ko_path, mod_name, records, syms, total_words,
+                 mod_text_start, kaslr_off):
+        self.ko_path = ko_path
+        self.mod_name = mod_name
+        self.records = records
+        self.syms = syms
+        self.total_words = total_words
+        self.mod_text_start = mod_text_start
+        self.kaslr_offset = kaslr_off
+        self.runtime_text = runtime_text(syms)
+        self._mod_syms = any(m == mod_name for _, _, m in syms)
+        # Aliases: rustc's merge-functions makes identical bodies (e.g. the
+        # one-field rsf_1 and rstf_1) share one address, so a PC can carry
+        # several names.
+        self._names = {}
+        for addr, name, mod in syms:
+            self._names.setdefault((addr, mod), set()).add(name)
+
+    def is_module_pc(self, pc):
+        """True if pc lies in the test module (kallsyms, else .text start)."""
+        if self._mod_syms:
+            hit = lookup(pc, self.syms)
+            return bool(hit) and hit[2] == self.mod_name
+        # Fallback: if no module symbols (kptr_restrict), use .text start
+        return bool(self.mod_text_start) and pc >= self.mod_text_start
+
+    def funcs(self, rec):
+        """All raw kallsyms names of the function a record belongs to."""
+        hit = lookup(rec["pc"], self.syms)
+        if not hit:
+            return set()
+        name, offset, mod = hit
+        return self._names.get((rec["pc"] - offset, mod), {name})
+
+    def module_records(self):
+        return [r for r in self.records if self.is_module_pc(r["pc"])]
+
+    def context_records(self, n):
+        """Module records plus n records before/after each of them."""
+        keep = set()
+        for i, r in enumerate(self.records):
+            if self.is_module_pc(r["pc"]):
+                keep.update(range(max(0, i - n),
+                                  min(len(self.records), i + n + 1)))
+        return [self.records[i] for i in sorted(keep)]
+
+
+def run_capture(ko_path, remote=False, vmlinux=None, kaslr_override=None,
+                log=None):
+    """
+    Load @ko_path, record while its trigger file(s) are written, unload it
+    and return a Capture. @remote publishes the buffer for REMOTE_HANDLE
+    instead of enabling recording for this task. Raises OSError.
+    """
+    global KASLR_OFFSET
+    log = log or (lambda msg: print(f"# {msg}"))
+
+    # Ensure kallsyms shows real addresses
+    try:
+        with open("/proc/sys/kernel/kptr_restrict", "w") as f:
+            f.write("0")
+    except OSError:
+        pass
+
+    df_fd = os.open(KCOV_DF_PATH, os.O_RDWR)
+    try:
+        # Init + mmap
+        fcntl.ioctl(df_fd, KCOV_DF_INIT_TRACK, BUF_SIZE)
+        libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True)
+        libc.mmap.restype = ctypes.c_void_p
+        libc.mmap.argtypes = [
+            ctypes.c_void_p, ctypes.c_size_t, ctypes.c_int,
+            ctypes.c_int, ctypes.c_int, ctypes.c_long
+        ]
+        buf_ptr = libc.mmap(None, BUF_SIZE * 8, 0x3, 0x01, df_fd, 0)
+        if buf_ptr == ctypes.c_void_p(-1).value:
+            errno = ctypes.get_errno()
+            raise OSError(errno, f"mmap: {os.strerror(errno)}")
+        buf = (ctypes.c_uint64 * BUF_SIZE).from_address(buf_ptr)
+
+        # Load module first (its init generates noise with INSTRUMENT_ALL)
+        mod_name = os.path.basename(ko_path).replace(".ko", "")
+        finit_module(ko_path)
+        log(f"Loaded {mod_name}")
+        try:
+            # Module .text address, the PC filter fallback without kallsyms
+            mod_text_start = 0
+            try:
+                with open(f"/sys/module/{mod_name}/sections/.text") as f:
+                    mod_text_start = int(f.read().strip(), 16)
+            except (OSError, ValueError):
+                pass
+
+            # Enable recording AFTER load, BEFORE trigger (no loader noise).
+            # Remote: the handle is passed by pointer (a __u64 in a buffer),
+            # so the full 64-bit value survives 32-bit/compat callers.
+            if remote:
+                fcntl.ioctl(df_fd, KCOV_DF_REMOTE_ENABLE,
+                            struct.pack("Q", REMOTE_HANDLE))
+            else:
+                fcntl.ioctl(df_fd, KCOV_DF_ENABLE, 0)
+            buf[0] = 0
+            try:
+                for path in trigger_module():
+                    log(f"Triggered {path}")
+            finally:
+                fcntl.ioctl(df_fd, KCOV_DF_REMOTE_DISABLE if remote
+                            else KCOV_DF_DISABLE, 0)
+
+            # Read kallsyms while the module is still loaded
+            syms = load_kallsyms()
+        finally:
+            try:
+                delete_module(mod_name)
+            except OSError as e:
+                log(f"warning: {e}")
+
+        if kaslr_override is not None:
+            KASLR_OFFSET = kaslr_override
+        else:
+            KASLR_OFFSET = kaslr_offset(syms, find_vmlinux(vmlinux))
+
+        total = int(buf[0])
+        records = parse_records(buf, total)
+        return Capture(ko_path, mod_name, records, syms, total,
+                       mod_text_start, KASLR_OFFSET)
+    finally:
+        os.close(df_fd)
+
+
+def print_raw(records, syms, vmlinux=None, cache=None, ko_path=None, mod_text_base=0):
+    """Print records in raw format with source line on left."""
+    if cache is None:
+        cache = {}
+    # Pre-resolve all locations (one addr2line run) to find max width
+    resolve_lines([r["pc"] for r in records], vmlinux, cache, ko_path,
+                  mod_text_base)
+    locs = [cache[r["pc"]] for r in records]
+    max_w = max((len(l) for l in locs if l), default=0)
+    max_w = max(max_w, 10)  # minimum width
+
+    for i, r in enumerate(records):
+        name, mod = symbolize(r["pc"], syms)
+        sym = f"{name}{mod}"
+        t = "ENTRY" if r["type"] == DF_TYPE_ENTRY else "RET  "
+        arg_idx = r["arg_idx"]
+        size = r["size"]
+        left = f"{locs[i]:>{max_w}s}" if locs[i] else f"{'':>{max_w}s}"
+        vals = format_val(r["val"]) if len(r["vals"]) == 1 else \
+            "{" + ", ".join(format_val(v) for v in r["vals"]) + "}"
+        print(f"{left}   [{t}] seq={r['seq']:3d} {sym} "
+              f"arg[{arg_idx}]({size}) @0x{r['ptr']:x} = {vals}")
+
+
+def print_tree(records, syms, vmlinux=None, cache=None, ko_path=None, mod_text_base=0):
+    """Print records as indented call tree with source line on left."""
+    if cache is None:
+        cache = {}
+    # Pre-resolve all PCs (one addr2line run) for alignment
+    resolve_lines([r["pc"] for r in records], vmlinux, cache, ko_path,
+                  mod_text_base)
+    max_w = max((len(v) for v in cache.values() if v), default=10)
+    max_w = max(max_w, 10)
+
+    depth = 0
+    call_stack = []  # Stack of (name, mod, args_str, pc) for matching returns
+    i = 0
+    while i < len(records):
+        r = records[i]
+        name, mod = symbolize(r["pc"], syms)
+
+        if r["type"] == DF_TYPE_ENTRY:
+            # Collect all args for this call (same PC, consecutive entries);
+            # order by index, as the pass emits dead-arg traces last.
+            args = []
+            pc = r["pc"]
+            while i < len(records) and records[i]["type"] == DF_TYPE_ENTRY \
+                    and records[i]["pc"] == pc:
+                vals = records[i]["vals"]
+                if len(vals) > 1:
+                    fields = ", ".join(format_val(v) for v in vals)
+                    args.append((records[i]["arg_idx"], "{" + fields + "}"))
+                else:
+                    args.append((records[i]["arg_idx"],
+                                 format_val(records[i]["val"])))
+                i += 1
+            args_str = ", ".join(a for _, a in sorted(args, key=lambda x: x[0]))
+            call_stack.append((name, mod, args_str, pc))
+            depth += 1
+        else:
+            # Pop void calls (no return record) until we find matching PC
+            while call_stack and call_stack[-1][3] != r["pc"]:
+                depth = max(0, depth - 1)
+                indent = "  " * depth
+                vname, vmod, vargs, vpc = call_stack.pop()
+                loc = resolve_line(vpc, vmlinux, cache, ko_path, mod_text_base)
+                left = f"{loc:>{max_w}s}" if loc else f"{'':>{max_w}s}"
+                print(f"{left}   {indent}{vname}({vargs}){vmod}")
+            depth = max(0, depth - 1)
+            indent = "  " * depth
+            ret_size = r["size"]
+            loc = resolve_line(r["pc"], vmlinux, cache, ko_path, mod_text_base)
+            left = f"{loc:>{max_w}s}" if loc else f"{'':>{max_w}s}"
+            if call_stack:
+                cname, cmod, cargs, _ = call_stack.pop()
+                if ret_size == 0:
+                    print(f"{left}   {indent}{cname}({cargs}){cmod}")
+                else:
+                    print(f"{left}   {indent}{format_val(r['val'])} = {cname}({cargs}){cmod}")
+            else:
+                if ret_size == 0:
+                    print(f"{left}   {indent}{name}(){mod}")
+                else:
+                    print(f"{left}   {indent}{format_val(r['val'])} = {name}(){mod}")
+            i += 1
+
+    # Flush remaining void calls on the stack
+    while call_stack:
+        depth = max(0, depth - 1)
+        indent = "  " * depth
+        vname, vmod, vargs, vpc = call_stack.pop()
+        loc = resolve_line(vpc, vmlinux, cache, ko_path, mod_text_base)
+        left = f"{loc:>{max_w}s}" if loc else f"{'':>{max_w}s}"
+        print(f"{left}   {indent}{vname}({vargs}){vmod}")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Load a test module with kcov_dataflow and view records")
+    parser.add_argument("module", help="Test module name (e.g. eight_struct_args_c)")
+    parser.add_argument("--raw", action="store_true",
+                        help="Print raw records instead of tree")
+    parser.add_argument("--ko", help="Explicit path to .ko file")
+    parser.add_argument("--context", "-C", type=int, default=0,
+                        help="Show N records before/after each module record")
+    parser.add_argument("--vmlinux", help="Path to vmlinux for addr2line")
+    parser.add_argument("--remote", action="store_true",
+                        help="Use KCOV_DF_REMOTE_ENABLE for kworker capture")
+    parser.add_argument("--kaslr-offset", type=lambda x: int(x, 0),
+                        help="Override the runtime KASLR offset added to PCs")
+    args = parser.parse_args()
+
+    ko_path = args.ko or find_module(args.module)
+    if not ko_path or not os.path.exists(ko_path):
+        print(f"Cannot find module for '{args.module}'", file=sys.stderr)
+        print("Build it first: make -C tools/testing/selftests "
+              "TARGETS=kcov_dataflow LLVM=1 CC=clang", file=sys.stderr)
+        sys.exit(1)
+
+    try:
+        cap = run_capture(ko_path, remote=args.remote, vmlinux=args.vmlinux,
+                          kaslr_override=args.kaslr_offset)
+    except OSError as e:
+        print(f"{args.module}: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"# Captured {cap.total_words} words (kaslr_offset=0x{cap.kaslr_offset:x}, "
+          f"_text=0x{cap.runtime_text:x})")
+    print(f"# {len(cap.records)} records")
+
+    if cap.syms or cap.mod_text_start:
+        if args.context > 0:
+            records = cap.context_records(args.context)
+            print(f"# showing {len(records)} records with context={args.context} "
+                  f"around {cap.mod_name}\n")
+        else:
+            records = cap.module_records()
+            print(f"# {len(records)} from {cap.mod_name}\n")
+    else:
+        records = cap.records
+        print("")
+
+    meta = get_kernel_meta()
+    print_kernel_meta(meta, ko_path=ko_path)
+
+    vmlinux = find_vmlinux(args.vmlinux)
+    show = print_raw if args.raw else print_tree
+    show(records, cap.syms, vmlinux, {}, ko_path, cap.mod_text_start)
+
+    print_kernel_meta(meta, ko_path=ko_path)
+
+
+if __name__ == "__main__":
+    main()

-- 
2.47.3