Skip to content

k4bench.plugin.runtime

k4bench.plugin.runtime

find_plugin_lib_dir 

find_plugin_lib_dir() -> Path

Return directory containing the k4Bench plugin libraries.

The returned directory must contain both the .so files AND the .components manifests that DDG4 uses to resolve factory names. Without the .components files, DDG4 can only load plugins whose library name matches the class name exactly (e.g. libFoo.so for class Foo), so bundled plugins like k4BenchRegionEventAction (which lives in libk4BenchRegionTimingAction.so) would be silently skipped.

Source code in k4bench/plugin/runtime.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def find_plugin_lib_dir() -> Path:
    """Return directory containing the k4Bench plugin libraries.

    The returned directory must contain both the .so files AND the
    .components manifests that DDG4 uses to resolve factory names.
    Without the .components files, DDG4 can only load plugins whose
    library name matches the class name exactly (e.g. libFoo.so for
    class Foo), so bundled plugins like k4BenchRegionEventAction
    (which lives in libk4BenchRegionTimingAction.so) would be silently
    skipped.
    """
    plugin_root = _find_plugin_root()
    library_patterns = [pattern for _, pattern, _ in _PLUGINS]
    # Derive the .components filename from the .so glob pattern.
    components_patterns = [p.split(".so")[0] + ".components" for p in library_patterns]

    search_dirs = [
        plugin_root / "install" / "lib",
        plugin_root / "install" / "lib64",
        plugin_root / "build",
    ]

    for libdir in search_dirs:
        if not libdir.exists():
            continue
        if (all(any(libdir.glob(p)) for p in library_patterns)
                and all(any(libdir.glob(p)) for p in components_patterns)):
            return libdir

    raise FileNotFoundError(
        "Could not locate k4Bench plugin libraries "
        f"({', '.join(library_patterns)})."
    )

ensure_plugin_built 

ensure_plugin_built() -> None

Build the k4Bench plugins if needed.

Source code in k4bench/plugin/runtime.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def ensure_plugin_built() -> None:
    """Build the k4Bench plugins if needed."""
    try:
        find_plugin_lib_dir()
        return
    except FileNotFoundError:
        pass

    plugin_root = _find_plugin_root()
    build_script = plugin_root / "build.sh"

    if not build_script.exists():
        raise FileNotFoundError(f"Missing plugin build script: {build_script}")

    result = subprocess.run(
        ["bash", str(build_script)],
        capture_output=True,
        text=True,
    )

    if result.returncode != 0:
        raise RuntimeError(
            "Failed to build k4Bench timing plugins:\n"
            f"{result.stdout}\n{result.stderr}"
        )

setup_plugin_environment 

setup_plugin_environment(*, env: dict[str, str], event_json_path: Path, region_json_path: Path | None = None) -> bool

Prepare environment variables for the k4Bench timing plugins.

Parameters:

Name Type Description Default
env dict[str, str]

Environment dictionary to mutate (typically a copy of os.environ).

 required
event_json_path Path

Output path for the per-event timing JSON.

 required
region_json_path Path | None

Output path for the per-region timing JSON. If None, the region plugin will still be loadable but will write to its default location (k4bench_regions.json in CWD) only if it ends up being activated by the steering script.

 None

Returns:

Type Description
bool

True if plugins are available and enabled. False if ddsim should run without per-event timing.
Source code in k4bench/plugin/runtime.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def setup_plugin_environment(
    *,
    env: dict[str, str],
    event_json_path: Path,
    region_json_path: Path | None = None,
) -> bool:
    """Prepare environment variables for the k4Bench timing plugins.

    Parameters
    ----------
    env
        Environment dictionary to mutate (typically a copy of os.environ).
    event_json_path
        Output path for the per-event timing JSON.
    region_json_path
        Output path for the per-region timing JSON. If None, the region
        plugin will still be loadable but will write to its default
        location (k4bench_regions.json in CWD) only if it ends up being
        activated by the steering script.

    Returns
    -------
    bool
        True if plugins are available and enabled.
        False if ddsim should run without per-event timing.
    """
    try:
        ensure_plugin_built()
        lib_dir = str(find_plugin_lib_dir())

        existing = env.get("LD_LIBRARY_PATH", "")
        env["LD_LIBRARY_PATH"] = f"{lib_dir}:{existing}" if existing else lib_dir

        env["K4BENCH_EVENT_JSON"] = str(event_json_path.resolve())

        if region_json_path is not None:
            env["K4BENCH_REGION_JSON"] = str(region_json_path.resolve())
        else:
            env.pop("K4BENCH_REGION_JSON", None)

        return True

    except (
        FileNotFoundError,
        RuntimeError,
        subprocess.SubprocessError,
    ) as exc:
        print(
            f"NOTE: k4Bench timing plugins unavailable "
            f"({exc}); continuing without per-event timing."
        )

        return False