diff --git a/.claude/rules/project-layout.md b/.claude/rules/project-layout.md index 8ae569a2f7..03035540e1 100644 --- a/.claude/rules/project-layout.md +++ b/.claude/rules/project-layout.md @@ -11,8 +11,13 @@ How this repo organizes Python packages, the build system, and example / test di | `_task_interface` | `python/bindings/` | nanobind `.so` at wheel root | Internal nanobind module | `simpler` exposes `Worker` and the `task_interface` submodule lazily (PEP 562 -`__getattr__`), so `from simpler import Worker` works while `import simpler` -still costs nothing and does not require the `_task_interface` extension. +`__getattr__`), so `from simpler import Worker` works while `import simpler` does +not require the `_task_interface` extension. Import time is not quite free: it +dlopens `build/lib/libsimpler_log.so` into the global symbol scope (~3 ms when +that library is built, ~25 µs when it is not), which must precede the first +worker fork. When `_task_interface` is available, `_log` passes the logger's +host-span entry point into the extension's nullable, extension-local sink slot. The dlopen and +binding are best-effort and raise nothing — see `python/simpler/_log_preload.py`. `simpler.task_interface.ChipTensor` is the GM-address-bearing device descriptor a `ChipWorker` consumes; `simpler_setup.TensorArg` is the address-free scene-test arg spec `NamedTuple`. They are separate types in separate diff --git a/docs/dfx/host-trace.md b/docs/dfx/host-trace.md index 896c645776..6904ef2a49 100644 --- a/docs/dfx/host-trace.md +++ b/docs/dfx/host-trace.md @@ -40,6 +40,12 @@ One line per span, emitted on scope exit | `ts` `dur` | start + duration in ns. Maps 1:1 onto a Chrome-trace `"X"` event. For host spans `ts` is `CLOCK_MONOTONIC` (`steady_clock`), same-host cross-process comparable. For `clk=dev` device spans (see below) `ts` is instead a **device-clock** start offset on a per-invocation origin — comparable to the other device spans (so the orch∪sched window is recoverable), not the host clock. | | `k=v ...` | optional per-span attributes (e.g. `ntensor=4`); a parser that doesn't recognize one ignores it. | +Span names and attributes percent-encode control bytes and record delimiters. +They are length-capped (with `~` marking truncation) so each marker remains a +single atomic pipe write even when forked workers share captured stderr. +`strace_timing.py` decodes both on the way back in, so a consumer reading its +output sees the original text; a consumer reading the raw log does not. + ## Span tree ```text @@ -82,22 +88,76 @@ including time the caller spends polling or doing other host work; blocking | 2 | `simpler_run.bind.args`, `simpler_run.bind.prebuilt`, `simpler_run.runner_run.device_wall` | | 3 | `simpler_run.runner_run.device_wall.{preamble,so_load,graph_build,config_validate,arena_wire,sm_reset,post_orch,orch,sched,task_slot_*}` | +## L3/L4 host scheduler spans + +Every hierarchical worker that drives next-level children emits these spans +through the same process-global `libsimpler_log.so` sink — an L3 with chips and +an L4 pod alike, since the orchestrator and scheduler code they run is the same: + +| Span | Host decision point | +| ---- | ------------------- | +| `l3.graph_build` | serialized Python graph callback | +| `l3.submit` | next-level task publication after slot allocation | +| `l3.dispatch` | scheduler handoff to a worker thread | +| `l3.frame_submit` | local child mailbox-frame publication | +| `l3.activate` | prepared-frame activation | +| `l3.complete` | terminal child progress handling | + +Their attributes carry the available `run_id`, `task_slot`, `group_index`, +`worker_id`, `dispatch_id`, and endpoint kind. + +Because the names do not encode which level emitted them, a pod run puts the L4 +process and each of its L3 processes on lanes that differ only by pid. The +per-level vocabulary that resolves this is tracked in +[#1793](https://github.com/hw-native-sys/simpler/issues/1793). + +One process contributes at most two host lanes, because the scheduler runs on +one thread: the facade thread emits `l3.graph_build` and `l3.submit`, and the +scheduler thread emits the other four. `role=worker` on `l3.frame_submit`, +`l3.activate` and `l3.complete` names the worker a dispatch targets, not the +thread that ran it. + +The spans reach the logger over a fixed POD C ABI, `SimplerHostSpan` in +`common/host_span.h`. `_task_interface` cannot link `libsimpler_log.so` — that +library is reached by `RTLD_GLOBAL` dlopen at runtime — so each extension/DSO +owns its own nullable sink function pointer instead of an undefined link symbol. +`simpler._log_preload` loads the library, and `simpler._log` passes the exported +entry-point address into `_task_interface` before Worker initialization. Every later +`fork()` therefore inherits the bound pointer and logger mapping, so parent and +child markers reach one sink. A process that never loads the logger leaves the +extension-local pointer null, which disables host spans without failing anything. A later +`ChipWorker.init` refreshes the binding after loading the required runtime +logger. `host_runtime.so` links the library directly and needs none of this. + ## Reading the markers — `strace_timing.py` ```bash # TPOT table (per-callable, decode = most-invoked hid bucket) python -m simpler_setup.tools.strace_timing path/to/host_or_device.log -# also emit a Chrome-trace / Perfetto JSON (lane = pid → host call tree) +# also emit the established per-invocation call-tree JSON python -m simpler_setup.tools.strace_timing path/to/log --trace-out strace.json + +# emit the L3/L4 host scheduler timeline on real OS pid/tid lanes +python -m simpler_setup.tools.strace_timing path/to/log --swimlane host_swimlane.json ``` The tool groups by `(pid, inv)`, rebuilds each invocation's tree from `depth`, buckets by `hid`, and prints each callable's mean `simpler_run` plus per-stage -means. With `--trace-out` it writes one `ph:"X"` event per span keyed by pid, so -the L3 parent and each L2 child render as separate lanes in +means. With `--trace-out` it writes one `ph:"X"` event per span on a synthetic +per-invocation lane, so each call renders as an isolated nested tree in [Perfetto](https://ui.perfetto.dev) / `chrome://tracing`. +`--swimlane` is a separate view. Host slices keep their real OS pid/tid, and +task submission-to-dispatch handoffs render as flow arrows. Chrome Trace JSON +has only one visible timestamp axis, so putting the raw per-invocation device +clock beside `CLOCK_MONOTONIC` would create a multi-day empty interval. The +converter therefore keeps `clk=dev` records, with their original ns timestamps, +in the top-level `unalignedDeviceSpans` array instead of `traceEvents`; it does +not guess a clock offset. Perfetto opens directly on the host activity, while +the existing tables, tree, and `--trace-out` still provide the device-phase +timing views. + ## Why markers, not a return value Android's atrace writes to the ftrace `trace_marker` sink and systrace renders diff --git a/python/bindings/CMakeLists.txt b/python/bindings/CMakeLists.txt index 68218717f1..7c03cb4001 100644 --- a/python/bindings/CMakeLists.txt +++ b/python/bindings/CMakeLists.txt @@ -65,6 +65,7 @@ target_include_directories(_task_interface PRIVATE ${CMAKE_SOURCE_DIR}/src/common/platform/include ${CMAKE_SOURCE_DIR}/src/common/platform/include/common ${CMAKE_SOURCE_DIR}/src/common/platform/include/host + ${CMAKE_SOURCE_DIR}/src/common/log/include ${CMAKE_CURRENT_SOURCE_DIR} ) diff --git a/python/bindings/task_interface.cpp b/python/bindings/task_interface.cpp index 10b4d7a2b6..ae45e3fff1 100644 --- a/python/bindings/task_interface.cpp +++ b/python/bindings/task_interface.cpp @@ -56,6 +56,7 @@ #include "callable_protocol.h" #include "chip_run_lane.h" #include "chip_worker.h" +#include "common/host_span_scope.h" #include "data_type.h" #include "dma_workspace.h" #include "worker_chip_orch_comm.h" @@ -993,6 +994,35 @@ NB_MODULE(_task_interface, m) { m.attr("MAX_TENSOR_DIMS") = MAX_TENSOR_DIMS; m.attr("MAX_REGISTERED_CALLABLE_IDS") = MAX_REGISTERED_CALLABLE_IDS; m.attr("RUNTIME_ENV_RING_COUNT") = RUNTIME_ENV_RING_COUNT; +#if SIMPLER_HOST_STRACE + m.attr("HOST_STRACE_ENABLED") = true; +#else + m.attr("HOST_STRACE_ENABLED") = false; +#endif + m.def( + "_bind_host_span_sink", + [](uintptr_t address) { + simpler::host_trace::bind_sink(reinterpret_cast(address)); + return simpler::host_trace::sink_available(); + }, + nb::arg("address"), "Bind this extension's host-span sink to the process-global logger, or zero to disable it." + ); + m.def( + "_host_span_sink_available", &simpler::host_trace::sink_available, + "Whether this extension's host-span sink is bound to the process-global logger." + ); + m.def( + "_emit_host_span", + [](const std::string &name, uint64_t invocation_id, uint64_t callable_hash, int32_t depth, int64_t timestamp_ns, + int64_t duration_ns, const std::string &attributes) { + simpler::host_trace::emit( + name.c_str(), invocation_id, callable_hash, depth, timestamp_ns, duration_ns, attributes.c_str() + ); + }, + nb::arg("name"), nb::arg("invocation_id"), nb::arg("callable_hash"), nb::arg("depth"), nb::arg("timestamp_ns"), + nb::arg("duration_ns"), nb::arg("attributes") = "", + "Emit one explicitly timed host span through this extension's bound logger sink." + ); // Byte size of a ChipTensor and the offset of its address_space field within it. // A task-args blob stores ChipTensors as a raw memcpy array, so a Python-side // blob walker locates tensor i's fields at i * CHIP_TENSOR_STRIDE_BYTES without diff --git a/python/simpler/__init__.py b/python/simpler/__init__.py index bf67a2b727..4a8aa637ed 100644 --- a/python/simpler/__init__.py +++ b/python/simpler/__init__.py @@ -8,13 +8,17 @@ # ----------------------------------------------------------------------------------------------------------- """Simpler runtime — public Python surface. -Host-side log filter setup happens in `ChipWorker.init` (see -`simpler.task_interface`): it `ctypes.CDLL`s libsimpler_log.so RTLD_GLOBAL, -calls its `simpler_log_init` C entry to seed the process-wide HostLogger, then -hands off to the C++ `_ChipWorker.init` which dlopens host_runtime.so (whose -`simpler_init` reads CANN dlog config off that same HostLogger, onboard only). -The level forwarded is a one-shot snapshot of the `simpler` Python logger. -Nothing log-related needs to happen at import time here. +Host-side log filter setup happens during worker initialization. A hierarchical +`Worker` seeds the parent before its first fork; `ChipWorker.init` (see +`simpler.task_interface`) repeats that initialization in each child before the +C++ `_ChipWorker.init` dlopens host_runtime.so. The level forwarded is a +one-shot snapshot of the `simpler` Python logger, and onboard `simpler_init` +maps it onto CANN's coarser dlog ladder. +Import time here only dlopens libsimpler_log.so into the global symbol scope +(`simpler._log_preload`, reached through `._log` below); no level is seeded and +no failure is raised if the library is absent. When `_task_interface` is +available, `_log` passes the logger entry point into the extension's nullable, +extension-local host-span sink slot before Worker initialization. `Worker` and the `task_interface` submodule resolve on first attribute access rather than at import time: both pull in the `_task_interface` extension, so diff --git a/python/simpler/_log.py b/python/simpler/_log.py index 5bc4b3ab6d..1ececd8a70 100644 --- a/python/simpler/_log.py +++ b/python/simpler/_log.py @@ -12,23 +12,40 @@ INFO and WARN and is the default so stable performance markers remain visible without enabling ordinary INFO traffic. NUL is a suppression sentinel. -`Worker.init()` snapshots the effective ``simpler`` logger threshold and -forwards it to `ChipWorker.init()` once. The Python wrapper seeds the -process-wide HostLogger before the C++ side loads host_runtime.so; onboard +`Worker.init()` snapshots the effective ``simpler`` logger threshold. A +hierarchical worker seeds the parent process before its first fork, and every +`ChipWorker.init()` seeds its child before loading host_runtime.so; onboard setup maps the same threshold onto CANN's coarser severity ladder. """ import logging +from ._log_preload import host_span_sink_address as _host_span_sink_address +from ._log_preload import preload as _preload_simpler_log + +# Load the logger mapping before probing the extension below. ``simpler`` +# imports this module eagerly; once the extension is available, its local sink +# slot is therefore bound before Worker initialization can fork. +_host_log_handle = _preload_simpler_log() + # DEFAULT_LOG_THRESHOLD is exposed by the _task_interface nanobind module so # Python and C++ share one constant. During a fresh `pip install -e .` the # pre-existing .so may be stale or absent, so fall back to the hardcoded # value (kept in sync manually with src/common/log/include/common/log_level.h). try: - from _task_interface import DEFAULT_LOG_THRESHOLD as _NATIVE_DEFAULT # pyright: ignore[reportMissingImports] + from _task_interface import ( # pyright: ignore[reportMissingImports] + DEFAULT_LOG_THRESHOLD as _NATIVE_DEFAULT, + ) except (ImportError, AttributeError): _NATIVE_DEFAULT = 25 +try: + from _task_interface import _bind_host_span_sink # pyright: ignore[reportMissingImports] +except (ImportError, AttributeError): + pass +else: + _bind_host_span_sink(_host_span_sink_address(_host_log_handle)) + # Public verbosity constants (Python integer levels). TIMING = 25 NUL = 60 diff --git a/python/simpler/_log_preload.py b/python/simpler/_log_preload.py new file mode 100644 index 0000000000..737386798a --- /dev/null +++ b/python/simpler/_log_preload.py @@ -0,0 +1,99 @@ +# Copyright (c) PyPTO Contributors. +# This program is free software, you can redistribute it and/or modify it under the terms and conditions of +# CANN Open Software License Agreement Version 2.0 (the "License"). +# Please refer to the License for details. You may not use this file except in compliance with the License. +# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, +# INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE in the root of the software repository for the full text of the License. +# ----------------------------------------------------------------------------------------------------------- + +"""Put ``libsimpler_log.so`` in the process-global symbol scope. + +The logger has to be loaded before host runtimes and before workers fork. The +package's earliest ``_task_interface`` import also passes its host-span entry +point into the extension, where a nullable function pointer provides the +cross-platform optional-sink contract. + +A missing library leaves that pointer null, which disables host spans and fails +nothing. The path is resolved with ``find_spec`` rather than by importing +``simpler_setup.environment``: locating the package does not execute it, so +``import simpler`` does not pull in the compiler-side surface for a diagnostic. +""" + +from __future__ import annotations + +import ctypes +import functools +import importlib.util +import os +from pathlib import Path + +_LIBRARY_NAME = "libsimpler_log.so" + +# RTLD_GLOBAL dlopen registry, keyed by path. host_runtime.so and the sim context +# resolve against these globals, so each must be loaded exactly once before any +# host_runtime.so dlopen; a second dlopen of a path already here is skipped +# rather than repeated. Never closed. Lives here rather than in task_interface so +# the import-time preload below and ChipWorker.init share one registry. +_preloaded_globals: dict[str, ctypes.CDLL] = {} + + +def preload_global(path: str) -> ctypes.CDLL: + """dlopen `path` with RTLD_NOW | RTLD_GLOBAL, idempotently (one CDLL per path). + + Eager resolution (RTLD_NOW) surfaces a missing-symbol problem at load time + rather than at first use. + """ + handle = _preloaded_globals.get(path) + if handle is None: + handle = ctypes.CDLL(path, mode=os.RTLD_NOW | os.RTLD_GLOBAL) + _preloaded_globals[path] = handle + return handle + + +def _resolve_library() -> Path | None: + """Locate libsimpler_log.so under either install layout, or None. + + Mirrors ``simpler_setup.environment._resolve_project_root``: a wheel keeps + the built libraries under ``simpler_setup/_assets``, a source tree or + editable install keeps them at the repo root. + """ + try: + spec = importlib.util.find_spec("simpler_setup") + except (ImportError, ValueError): + return None + if spec is None or spec.origin is None: + return None + + package = Path(spec.origin).resolve().parent + assets = package / "_assets" + project_root = assets if (assets / "src").is_dir() else package.parent + library = project_root / "build" / "lib" / _LIBRARY_NAME + return library if library.is_file() else None + + +@functools.cache +def preload() -> ctypes.CDLL | None: + """dlopen the process-global logger RTLD_GLOBAL, or return None quietly. + + Cached, so the repeated call from ``ChipWorker.init``'s own preload of this + library does not bump the loader refcount again. + """ + library = _resolve_library() + if library is None: + return None + try: + return preload_global(str(library)) + except OSError: + return None + + +def host_span_sink_address(handle: ctypes.CDLL | None) -> int: + """Return the logger's host-span entry-point address, or zero if absent.""" + if handle is None: + return 0 + try: + symbol = handle.simpler_log_emit_host_span + except AttributeError: + return 0 + return int(ctypes.cast(symbol, ctypes.c_void_p).value or 0) diff --git a/python/simpler/task_interface.py b/python/simpler/task_interface.py index d7c3fe20c4..56bd962c9f 100644 --- a/python/simpler/task_interface.py +++ b/python/simpler/task_interface.py @@ -27,7 +27,6 @@ from __future__ import annotations import ctypes -import os import threading import uuid import weakref @@ -37,6 +36,15 @@ from pathlib import Path from typing import TYPE_CHECKING, Any +# `preload_global` is the process-wide RTLD_GLOBAL dlopen registry. +# host_runtime.so resolves its undefined HostLogger / unified_log_* (and, on sim, +# sim_context_*) symbols against those globals, so each must be loaded — exactly +# once — before any host_runtime.so dlopen. The registry lives in _log_preload so +# the import-time logger preload and _initialize_simpler_log share one entry per path. +from ._log_preload import host_span_sink_address as _host_span_sink_address +from ._log_preload import preload as _preload_simpler_log +from ._log_preload import preload_global as _preload_global + if TYPE_CHECKING: # Annotation-only: `CallableHandle` is imported lazily at its use site, and # PEP 563 keeps these annotations as strings, so nothing is imported at @@ -1221,22 +1229,43 @@ def committed(self) -> bool: # Process-wide RTLD_GLOBAL preload registry. host_runtime.so resolves its # undefined HostLogger / unified_log_* (and, on sim, sim_context_*) symbols # against these globals, so they must be loaded — exactly once — before any -# host_runtime.so dlopen. Keyed by path; mirrors the C++ side's old -# std::once_flag semantics. Never closed. -_preloaded_globals: dict[str, ctypes.CDLL] = {} +# host_runtime.so dlopen. The registry lives in _log_preload so the import-time +# logger preload and ChipWorker.init below share one entry per path: a second +# dlopen of an already-registered path is skipped rather than repeated. -def _preload_global(path: str) -> ctypes.CDLL: - """dlopen `path` with RTLD_NOW | RTLD_GLOBAL, idempotently (one CDLL per path). +def _initialize_simpler_log(bins: Any | None, log_level: int | None = None) -> ctypes.CDLL | None: + """Seed this process's logger threshold, before any runtime use or fork. - Eager resolution (RTLD_NOW) mirrors the previous C++ dlopen flags and - surfaces any missing-symbol problem at load time rather than first use. + ``bins`` names the copy a chip child must load, which is the one its + host_runtime.so resolves against. A Worker process that owns no chips has no + ``bins`` and passes None: the library the package already preloaded at import + is seeded instead, and a process where that preload found nothing keeps a + null host-span sink and emits nothing. """ - handle = _preloaded_globals.get(path) - if handle is None: - handle = ctypes.CDLL(path, mode=os.RTLD_NOW | os.RTLD_GLOBAL) - _preloaded_globals[path] = handle - return handle + if log_level is None: + from . import _log # noqa: PLC0415 + + log_level = _log.get_current_config() + + if bins is None: + log_handle = _preload_simpler_log() + if log_handle is None: + return None + else: + if not bins.simpler_log_path: + raise ValueError("simpler log init: bins.simpler_log_path is required") + log_handle = _preload_global(str(bins.simpler_log_path)) + + bind_host_span_sink = getattr(_ti_module, "_bind_host_span_sink", None) + if bind_host_span_sink is not None: + bind_host_span_sink(_host_span_sink_address(log_handle)) + log_handle.simpler_log_init.argtypes = [ctypes.c_int] + log_handle.simpler_log_init.restype = ctypes.c_int + rc = log_handle.simpler_log_init(int(log_level)) + if rc != 0: + raise RuntimeError(f"simpler_log_init failed with code {rc}") + return log_handle class ChipWorker: @@ -1319,20 +1348,8 @@ def init( self._init_in_progress = True try: - if log_level is None: - from . import _log # noqa: PLC0415 - - log_level = _log.get_current_config() - # 1. libsimpler_log.so — RTLD_GLOBAL singleton, before host_runtime.so. - if not bins.simpler_log_path: - raise ValueError("ChipWorker.init: bins.simpler_log_path is required") - log_handle = _preload_global(str(bins.simpler_log_path)) - log_handle.simpler_log_init.argtypes = [ctypes.c_int] - log_handle.simpler_log_init.restype = ctypes.c_int - rc = log_handle.simpler_log_init(int(log_level)) - if rc != 0: - raise RuntimeError(f"simpler_log_init failed with code {rc}") + _initialize_simpler_log(bins, log_level) # 2. libcpu_sim_context.so — sim platforms only. if bins.sim_context_path: diff --git a/python/simpler/worker.py b/python/simpler/worker.py index 69acaa51d0..0c3c3d0a1b 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -83,10 +83,13 @@ def my_l4_orch(orch, args, config): import cloudpickle from _task_interface import ( # pyright: ignore[reportMissingImports] + HOST_STRACE_ENABLED, MAX_REGISTERED_CALLABLE_IDS, PTO_PIPELINE_MAX_DEPTH, RUNTIME_ENV_RING_COUNT, WorkerType, + _emit_host_span, + _host_span_sink_available, _l3_child_onboard_region_close, _l3_child_onboard_region_create, _mailbox_load_i32, @@ -217,6 +220,7 @@ def my_l4_orch(orch, args, config): RemoteBufferExport, RemoteBufferHandle, TaskArgs, + _initialize_simpler_log, _Worker, ) from .worker_chip_orch_comm import ( @@ -250,6 +254,11 @@ def my_l4_orch(orch, args, config): _WORKER_CHIP_ENDPOINT_ERROR_REGION_RE = re.compile(r"\bL3-L2 endpoint error\b[^\n]*\bregion=(\d+)\b") +def _host_spans_active() -> bool: + """Whether an emitted host span can currently reach the logger sink.""" + return HOST_STRACE_ENABLED and _host_span_sink_available() + + # --------------------------------------------------------------------------- # Unified mailbox layout (must match worker_manager.h MAILBOX_OFF_*) # --------------------------------------------------------------------------- @@ -7263,6 +7272,15 @@ def _start_hierarchical(self) -> None: # noqa: PLR0912 -- three parallel fork l for digest, state in self._identity_registry.items() ] + # Seed this process's logger before the first fork: the spans its own + # scheduler emits obey the Python logger level, and every child inherits + # the mapping. Every level needs this — `init()` rejects device_ids above + # L3, so a pod process owns no chips yet still drives next-level Workers + # and emits their spans. A chip-owning Worker names the copy its children + # load; any other process seeds the copy the package already preloaded. + chip_log_level = _simpler_log.get_current_config() + _initialize_simpler_log(self._l3_bins if device_ids else None, chip_log_level) + self._startup_reaped_pids = set() self._startup_ready_pids = set() self._startup_group_leader_pids = set() @@ -7301,7 +7319,6 @@ def _setup(): # Fork ChipWorker processes (L3 with device_ids). Always use the plain # task-loop variant; the base communicator is established lazily on first # ``orch.allocate_domain`` via CTRL_COMM_INIT. - chip_log_level = _simpler_log.get_current_config() if device_ids: for idx, dev_id in enumerate(device_ids): pid = os.fork() @@ -10244,7 +10261,23 @@ def _submit_l3_locked(self, callable, args, cfg: CallConfig) -> RunHandle: self._orch._scope_begin() scope_open = True with _callback_run(run_id, self): - callable(self._orch, args, cfg) + if _host_spans_active(): + graph_start_ns = time.monotonic_ns() + try: + callable(self._orch, args, cfg) + finally: + graph_end_ns = time.monotonic_ns() + _emit_host_span( + "l3.graph_build", + run_id, + 0, + 0, + graph_start_ns, + graph_end_ns - graph_start_ns, + f"run_id={run_id} role=facade", + ) + else: + callable(self._orch, args, cfg) scope_open = False self._orch._scope_end() self._orch._close_run_submission(run_id) diff --git a/simpler_setup/runtime_compiler.py b/simpler_setup/runtime_compiler.py index d031b5598e..00792b459e 100644 --- a/simpler_setup/runtime_compiler.py +++ b/simpler_setup/runtime_compiler.py @@ -13,7 +13,7 @@ import subprocess import tempfile from pathlib import Path -from typing import Optional, Union +from typing import Callable, Optional, Union from simpler import env_manager @@ -23,6 +23,42 @@ logger = logging.getLogger(__name__) +def place_binary( + src: Union[str, Path], dest: Union[str, Path], post_process: Optional[Callable[[Path], None]] = None +) -> Path: + """Copy `src` over `dest` atomically, preserving metadata. + + `shutil.copy2` alone truncates and rewrites the destination in place, which + corrupts the mapping of any process that already has `dest` dlopened — the + fault surfaces later, as a SIGSEGV inside the dynamic linker on an unrelated + dlopen or at interpreter exit, far from the rewrite. Writing a sibling + temporary and renaming leaves the old inode intact for those processes while + new dlopens pick up the replacement. + + `post_process` runs on the staged copy, before it becomes visible under + `dest`; a step that rewrites the artifact (`strip`) must go there rather than + run against `dest` afterwards, or it reintroduces the in-place rewrite. + + The temporary is created in `dest`'s directory so the rename stays within one + filesystem, which is what makes it atomic. + """ + src_path = Path(src) + dest_path = Path(dest) + fd, staged_name = tempfile.mkstemp(dir=str(dest_path.parent), prefix=f".{dest_path.name}.", suffix=".tmp") + os.close(fd) + staged = Path(staged_name) + try: + shutil.copy2(src_path, staged) + os.chmod(staged, os.stat(src_path).st_mode & 0o7777) + if post_process is not None: + post_process(staged) + os.replace(staged, dest_path) + except BaseException: + staged.unlink(missing_ok=True) + raise + return dest_path + + class BuildTarget: """CMake build target: composes a Toolchain with a source directory and output name. @@ -303,20 +339,24 @@ def _build(actual_build_dir: str) -> Union[bytes, Path]: dest_dir = Path(dispatcher_dest) dest_dir.mkdir(parents=True, exist_ok=True) dest_dispatcher = dest_dir / dispatcher_name - shutil.copy2(dispatcher_so, dest_dispatcher) # Cross-arch strip: aicpu .so is aarch64 even on x86 host; # GNU strip 2.38 on Ubuntu 22.04 cannot read it. Prefer # llvm-strip (multi-arch) when available. strip_bin = shutil.which("llvm-strip") or shutil.which("aarch64-linux-gnu-strip") or "strip" - subprocess.run([strip_bin, "-s", str(dest_dispatcher)], check=True) + + def _strip(staged: Path, strip_bin: str = strip_bin) -> None: + # strip_bin is selected only from the trusted host toolchain. + subprocess.run([strip_bin, "-s", str(staged)], check=True) # noqa: S603 + + place_binary(dispatcher_so, dest_dispatcher, post_process=_strip) if output_dir is not None: od = Path(output_dir) od.mkdir(parents=True, exist_ok=True) dest = od / binary_name - shutil.copy2(binary_path, dest) + place_binary(binary_path, dest) if target_platform == "host" and self.platform == "a5": topo_fallback = Path(cmake_source_dir) / "aicpu_cpu_topo_fallback.json" - shutil.copy2(topo_fallback, od / topo_fallback.name) + place_binary(topo_fallback, od / topo_fallback.name) return dest else: with open(binary_path, "rb") as f: @@ -496,7 +536,7 @@ def _build(actual_build_dir: str) -> Union[bytes, Path]: od = Path(output_dir) od.mkdir(parents=True, exist_ok=True) dest = od / binary_name - shutil.copy2(binary_path, dest) + place_binary(binary_path, dest) return dest else: with open(binary_path, "rb") as f: @@ -537,7 +577,7 @@ def _build(actual_build_dir: str) -> Union[bytes, Path]: od = Path(output_dir) od.mkdir(parents=True, exist_ok=True) dest = od / binary_name - shutil.copy2(binary_path, dest) + place_binary(binary_path, dest) return dest else: with open(binary_path, "rb") as f: diff --git a/simpler_setup/tools/README.md b/simpler_setup/tools/README.md index 66f314d476..d5a434431f 100644 --- a/simpler_setup/tools/README.md +++ b/simpler_setup/tools/README.md @@ -328,6 +328,9 @@ python -m simpler_setup.tools.strace_timing path/to/log --tree # Also emit a Chrome-trace / Perfetto JSON (one named lane per invocation, with # separate host and device(clk=dev) tracks; nested by span containment) python -m simpler_setup.tools.strace_timing path/to/log --trace-out strace.json + +# L3/L4 host scheduler timeline (real OS pid/tid lanes + cross-thread flows) +python -m simpler_setup.tools.strace_timing path/to/log --swimlane host_swimlane.json ``` Groups spans by `(pid, inv)`, rebuilds each invocation's tree from `depth`, @@ -352,6 +355,15 @@ to a file (`python test_*.py … --rounds N > run.log 2>&1`) and pass `run.log` here. Because grouping is per `(pid, inv)`, this captures **L3 multi-round** (every chip-child invocation), not just round 0. +`--swimlane` consumes both the `l3.*` scheduler markers and child +`simpler_run` markers. Host lanes retain their OS pid/tid. Because Chrome Trace +JSON has one visible timestamp axis, raw device-domain `clk=dev` slices are +stored in the top-level `unalignedDeviceSpans` array rather than placed beside +the unrelated host clock and stretching Perfetto into an empty-looking +multi-day viewport. Their ns timestamps remain unchanged; no clock offset is +invented. This does not alter the established per-invocation `--trace-out` +view. + --- ## deps_viewer diff --git a/simpler_setup/tools/strace_timing.py b/simpler_setup/tools/strace_timing.py index 014b88d4ba..d565185606 100644 --- a/simpler_setup/tools/strace_timing.py +++ b/simpler_setup/tools/strace_timing.py @@ -35,8 +35,10 @@ * a per-callable TPOT table (each invocation's simpler_run dur + the mean of each sub-stage across invocations), and * optionally a Chrome-trace / Perfetto JSON (``--trace-out``): one ``ph:"X"`` - event per span, lane = pid, so the host call tree renders as nested - slices (L3 parent and each L2 child get their own pid lane). + event per span on a synthetic per-invocation lane, so each host call tree + renders as nested slices, or + * a host scheduler swimlane (``--swimlane``) whose host lanes retain the + real OS pid/tid and whose cross-thread handoffs are Chrome flow events. """ from __future__ import annotations @@ -68,6 +70,18 @@ # A record start, matched independently of whether the rest of that record # survived the write that emitted it. _STRACE_HEAD_RE = re.compile(r"\[STRACE\]\s+v=\d+") +# The emitter percent-encodes any byte that would otherwise be record grammar — +# see `encode_host_span_field` in src/common/log/host_log.cpp. +_PERCENT_ESCAPE_RE = re.compile(r"%([0-9A-Fa-f]{2})") + + +def decode_field(text): + """Reverse the emitter's percent-encoding of a name or attribute value. + + A field the emitter truncated ends in ``~``, which is left in place: it is a + marker that the value is incomplete, not an encoded byte. + """ + return _PERCENT_ESCAPE_RE.sub(lambda m: chr(int(m.group(1), 16)), text) @dataclass @@ -106,7 +120,9 @@ def root(self): def by_name(self): m = {} for s in self.spans: - m.setdefault(s.name, s) + previous = m.get(s.name) + if previous is None or s.ts < previous.ts: + m[s.name] = s return m @@ -130,13 +146,25 @@ def parse_spans(lines): inv=int(m["inv"]), hid=m["hid"].lower(), depth=int(m["depth"]), - name=m["name"], + name=decode_field(m["name"]), ts=int(m["ts"]), dur=int(m["dur"]), attrs=m["attrs"].strip(), ) +def legacy_spans(spans): + """Return spans belonging to the established ``simpler_run`` views. + + L3/L4 host-scheduler markers share the STRACE grammar but answer a + different question. Keeping them out of invocation grouping preserves the + TPOT, rounds, tree, and ``--trace-out`` contracts when a log contains both + marker families. Filter only the newly introduced namespace so existing + marker families such as ``simpler_prewarm`` retain their old behavior. + """ + return [span for span in spans if not span.name.startswith("l3.")] + + def group_invocations(spans): """Group spans into Invocation objects keyed by (pid, inv).""" groups: dict = {} @@ -396,6 +424,206 @@ def to_chrome_trace(invocations, buckets=None): return {"traceEvents": events, "displayTimeUnit": "ms"} +def _parsed_attrs(span): + attrs = {} + for attribute in span.attrs.split(): + key, separator, value = attribute.partition("=") + if not separator: + continue + if re.fullmatch(r"-?\d+", value): + attrs[key] = int(value) + else: + attrs[key] = decode_field(value) + return attrs + + +# Highest-precedence match wins. One OS thread emits spans of several roles: the +# scheduler loop is the sole caller of both `dispatch_ready` and +# `manager->progress`, so it emits `l3.dispatch` (role=scheduler) alongside +# `l3.frame_submit` / `l3.activate` / `l3.complete`, whose `role=worker` names +# the worker a dispatch targets rather than the thread doing the work. +_HOST_THREAD_ROLES = ("facade", "scheduler", "worker") + + +def _host_thread_name(entries): + """Name one OS thread's lane from every span it emitted. + + `entries` are that thread's (span, parsed attributes) pairs. + """ + roles = set() + worker_ids = set() + for span, attrs in entries: + role = attrs.get("role") + if role == "facade" or span.name in {"l3.graph_build", "l3.submit"}: + roles.add("facade") + elif role in ("scheduler", "worker"): + roles.add(role) + elif span.name.startswith("l3."): + roles.add("worker") + if role == "worker": + worker_ids.add(attrs.get("worker_id")) + + for role in _HOST_THREAD_ROLES: + if role not in roles: + continue + if role == "facade": + return "orchestrator / facade" + if role == "scheduler": + return "scheduler" + worker_id = worker_ids.pop() if len(worker_ids) == 1 else None + return f"worker {worker_id}" if worker_id is not None else "worker" + + if any(span.name == "simpler_run" or span.name.startswith("simpler_run.") for span, _ in entries): + return "chip child" + return f"tid {entries[0][0].tid}" + + +def _flow_key(span, attrs): + run_id = attrs.get("run_id") + task_slot = attrs.get("task_slot", attrs.get("slot")) + if run_id is None or task_slot is None: + return None + return span.pid, run_id, task_slot + + +def to_host_swimlane(spans): + """Build a real-pid/tid host scheduling timeline for Perfetto. + + Host timestamps remain on their shared CLOCK_MONOTONIC axis. Chrome Trace + JSON has one timestamp axis, so raw ``clk=dev`` events cannot be rendered + alongside host events without either a false clock alignment or a huge + empty interval. Keep those raw events in ``unalignedDeviceSpans`` for + inspection, but do not add them to Perfetto's visible ``traceEvents``. + """ + # (span, parsed attributes) pairs, so the attributes travel with their span + # through every partition below. `Span` is an unhashable dataclass, so a + # side table would have to be keyed on identity. + entries = [(span, _parsed_attrs(span)) for span in spans] + events = [] + + host_entries = [entry for entry in entries if not entry[0].is_device] + device_entries = [entry for entry in entries if entry[0].is_device] + host_pids = sorted({span.pid for span, _ in host_entries}) + host_threads = sorted({(span.pid, span.tid) for span, _ in host_entries}) + + for pid in host_pids: + process_spans = [span for span, _ in host_entries if span.pid == pid] + role = "host" if any(span.name.startswith("l3.") for span in process_spans) else "chip child" + events.append( + { + "ph": "M", + "name": "process_name", + "pid": pid, + "tid": 0, + "args": {"name": f"simpler {role} (pid={pid})"}, + } + ) + for pid, tid in host_threads: + on_thread = [entry for entry in host_entries if entry[0].pid == pid and entry[0].tid == tid] + events.append( + { + "ph": "M", + "name": "thread_name", + "pid": pid, + "tid": tid, + "args": {"name": _host_thread_name(on_thread)}, + } + ) + for span, parsed in sorted(host_entries, key=lambda item: (item[0].ts, item[0].pid, item[0].tid, item[0].name)): + event_args = {"inv": span.inv, "hid": span.hid, "depth": span.depth, "attrs": span.attrs, **parsed} + events.append( + { + "name": span.name, + "ph": "X", + "ts": span.ts / 1000.0, + "dur": span.dur / 1000.0, + "pid": span.pid, + "tid": span.tid, + "args": event_args, + } + ) + + submits = defaultdict(list) + for span, attrs in host_entries: + if span.name != "l3.submit": + continue + key = _flow_key(span, attrs) + if key is not None: + submits[key].append(span) + for candidates in submits.values(): + candidates.sort(key=lambda item: item.ts) + + dispatches = [] + for span, attrs in host_entries: + if span.name != "l3.dispatch": + continue + key = _flow_key(span, attrs) + source = None + if key is not None: + for candidate in submits.get(key, []): + if candidate.ts > span.ts: + break + source = candidate + if source is None: + continue + dispatches.append((source, span, attrs)) + + for flow_id, (source, destination, attrs) in enumerate(sorted(dispatches, key=lambda item: item[1].ts), start=1): + dispatch_key = ( + f"dispatch:{source.pid}:{attrs['run_id']}:{attrs.get('task_slot', attrs.get('slot'))}:" + f"{attrs.get('group_index', -1)}:{attrs.get('worker_id', -1)}:{attrs.get('dispatch_id', 0)}" + ) + flow_args = {"dispatch_key": dispatch_key} + events.append( + { + "name": "task dispatch", + "cat": "host.scheduler", + "ph": "s", + "id": flow_id, + "ts": min(source.ts + source.dur, destination.ts) / 1000.0, + "pid": source.pid, + "tid": source.tid, + "args": flow_args, + } + ) + events.append( + { + "name": "task dispatch", + "cat": "host.scheduler", + "ph": "f", + "id": flow_id, + "ts": destination.ts / 1000.0, + "pid": destination.pid, + "tid": destination.tid, + "args": flow_args, + } + ) + + unaligned_device_spans = [] + for span, attrs in sorted( + device_entries, key=lambda item: (item[0].pid, item[0].inv, item[0].ts, item[0].tid, item[0].name) + ): + unaligned_device_spans.append( + { + "name": span.name, + "ts_ns": span.ts, + "dur_ns": span.dur, + "pid": span.pid, + "tid": span.tid, + "inv": span.inv, + "hid": span.hid, + "depth": span.depth, + "attrs": {"raw": span.attrs, **attrs}, + } + ) + + return { + "traceEvents": events, + "displayTimeUnit": "ms", + "unalignedDeviceSpans": unaligned_device_spans, + } + + def _print_agg_tree(invs, stream=sys.stdout): """Print a callable's spans as a nested tree built from the dotted span names (so e.g. ``simpler_run.bind.args`` nests under ``simpler_run.bind``), @@ -495,6 +723,10 @@ def main(argv=None): ap.add_argument( "--trace-out", help="write a Chrome-trace/Perfetto JSON here (load in chrome://tracing or perfetto)" ) + ap.add_argument( + "--swimlane", + help="write a real-pid/tid L3/L4 host swimlane JSON here (load in chrome://tracing or perfetto)", + ) ap.add_argument( "--rounds-table", action="store_true", @@ -523,7 +755,8 @@ def main(argv=None): "excluded from the timing below", file=sys.stderr, ) - invocations = group_invocations(spans) + legacy = legacy_spans(spans) + invocations = group_invocations(legacy) buckets = bucket_by_hid(invocations) if args.rounds_table: @@ -536,7 +769,16 @@ def main(argv=None): if args.trace_out: with open(args.trace_out, "w", encoding="utf-8") as f: json.dump(to_chrome_trace(invocations, buckets), f) - print(f"Wrote Chrome trace: {args.trace_out} ({len(spans)} spans)") + print(f"Wrote Chrome trace: {args.trace_out} ({len(legacy)} spans)") + + if args.swimlane: + with open(args.swimlane, "w", encoding="utf-8") as f: + json.dump(to_host_swimlane(spans), f) + host_count = sum(not span.is_device for span in spans) + print( + f"Wrote host swimlane: {args.swimlane} " + f"({host_count} host spans, {len(spans) - host_count} unaligned device spans)" + ) return 0 diff --git a/src/common/hierarchical/orchestrator.cpp b/src/common/hierarchical/orchestrator.cpp index 89d3d0f274..84737bb8f4 100644 --- a/src/common/hierarchical/orchestrator.cpp +++ b/src/common/hierarchical/orchestrator.cpp @@ -14,10 +14,14 @@ #include #include #include +#include #include +#include +#include #include #include +#include "common/host_span_scope.h" #include "worker_manager.h" void Orchestrator::init( @@ -711,6 +715,20 @@ SubmitResult Orchestrator::submit_impl( } TaskSlot slot = ar.slot; +#if SIMPLER_HOST_STRACE + std::optional submit_trace; + if (worker_type == WorkerType::NEXT_LEVEL) { + uint64_t trace_callable_hash = 0; + std::memcpy(&trace_callable_hash, callable.digest.data(), sizeof(trace_callable_hash)); + std::ostringstream trace_attrs; + trace_attrs << "run_id=" << run->id << " task_slot=" << slot + << " group_index=" << (args_list.size() == 1 ? 0 : -1) << " group_size=" << args_list.size(); + if (target_worker_ids.size() == 1) trace_attrs << " worker_id=" << target_worker_ids.front(); + trace_attrs << " role=facade"; + submit_trace.emplace("l3.submit", run->id, trace_callable_hash, 0, trace_attrs.str()); + } +#endif + TaskSlotState &s = slot_state(slot); s.reset(); s.run_id = run->id; diff --git a/src/common/hierarchical/worker_manager.cpp b/src/common/hierarchical/worker_manager.cpp index b77b3473f0..fb0f2e6d1d 100644 --- a/src/common/hierarchical/worker_manager.cpp +++ b/src/common/hierarchical/worker_manager.cpp @@ -22,12 +22,14 @@ #include #include #include +#include #include #include #include #include #include +#include "common/host_span_scope.h" #include "ring.h" namespace { @@ -55,6 +57,43 @@ std::string format_digest(const uint8_t *digest) { return out; } +#if SIMPLER_HOST_STRACE +const char *endpoint_kind_name(WorkerEndpointKind kind) { + switch (kind) { + case WorkerEndpointKind::LOCAL_MAILBOX: + return "local_mailbox"; + case WorkerEndpointKind::REMOTE_L3: + return "remote_l3"; + } + return "unknown"; +} + +RunId trace_run_id(Ring *ring, TaskSlot task_slot) { + if (ring == nullptr) return INVALID_RUN_ID; + TaskSlotState *state = ring->slot_state(task_slot); + return state == nullptr ? INVALID_RUN_ID : state->run_id; +} + +uint64_t trace_callable_hash(Ring *ring, TaskSlot task_slot) { + if (ring == nullptr) return 0; + TaskSlotState *state = ring->slot_state(task_slot); + if (state == nullptr) return 0; + uint64_t hash = 0; + std::memcpy(&hash, state->callable.digest.data(), sizeof(hash)); + return hash; +} + +std::string +trace_dispatch_attrs(RunId run_id, const WorkerDispatch &dispatch, const WorkerEndpointCaps &caps, const char *role) { + std::ostringstream attrs; + attrs << "run_id=" << run_id << " task_slot=" << dispatch.task_slot << " group_index=" << dispatch.group_index + << " worker_id=" << caps.worker_id << " dispatch_id=" << dispatch.dispatch_id + << " endpoint_kind=" << endpoint_kind_name(caps.kind) + << " prepare_only=" << static_cast(dispatch.prepare_only) << " role=" << role; + return attrs.str(); +} +#endif + // Wall-clock period between child liveness samples. Every mailbox wait spins, // so an iteration count would not map to a bounded wall time. constexpr std::chrono::milliseconds kChildLivenessPollPeriod{10}; @@ -390,7 +429,10 @@ void WorkerThread::dispatch_prepared(WorkerDispatch d) { WorkerThread::SubmitDispatchResult WorkerThread::submit_dispatch(WorkerDispatch d, LaneKind lane_kind, RunId expected_run_id) { - std::lock_guard admission_lk(admission_mu_); +#if SIMPLER_HOST_STRACE + const int64_t trace_start_ns = simpler::host_trace::now_ns(); +#endif + std::unique_lock admission_lk(admission_mu_); if (shutdown_.load(std::memory_order_acquire)) { return SubmitDispatchResult::STOPPING; } @@ -409,6 +451,10 @@ WorkerThread::submit_dispatch(WorkerDispatch d, LaneKind lane_kind, RunId expect } ++next_dispatch_id_; inflight_.fetch_add(1, std::memory_order_release); +#if SIMPLER_HOST_STRACE + const RunId trace_run = trace_run_id(ring_, d.task_slot); + const uint64_t trace_hash = trace_callable_hash(ring_, d.task_slot); +#endif try { endpoint_->submit_progress(ring_, d); } catch (const std::exception &e) { @@ -416,6 +462,14 @@ WorkerThread::submit_dispatch(WorkerDispatch d, LaneKind lane_kind, RunId expect } catch (...) { fail_submission(d, "submit_progress failed with unknown exception"); } + admission_lk.unlock(); +#if SIMPLER_HOST_STRACE + const int64_t trace_end_ns = simpler::host_trace::now_ns(); + const std::string trace_attrs = trace_dispatch_attrs(trace_run, d, endpoint_->caps(), "scheduler"); + simpler::host_trace::emit( + "l3.dispatch", trace_run, trace_hash, 0, trace_start_ns, trace_end_ns - trace_start_ns, trace_attrs.c_str() + ); +#endif return SubmitDispatchResult::SUBMITTED; } @@ -576,6 +630,11 @@ void WorkerThread::finish_progress_dispatch(const WorkerEndpointProgress &progre return; } +#if SIMPLER_HOST_STRACE + const RunId trace_run = trace_run_id(ring_, dispatch.task_slot); + const uint64_t trace_hash = trace_callable_hash(ring_, dispatch.task_slot); + std::string complete_attrs = trace_dispatch_attrs(trace_run, dispatch, endpoint_->caps(), "worker"); +#endif WorkerCompletion completion = progress.completion; if (accepted_dispatch_ids_.erase(dispatch.dispatch_id) == 0) { // This advances only the run-level waiter after a terminal endpoint @@ -598,6 +657,10 @@ void WorkerThread::finish_progress_dispatch(const WorkerEndpointProgress &progre accept_errors_.erase(accept_error); } +#if SIMPLER_HOST_STRACE + complete_attrs += " outcome=" + std::to_string(static_cast(completion.outcome)); + simpler::host_trace::SpanScope complete_trace("l3.complete", trace_run, trace_hash, 0, std::move(complete_attrs)); +#endif on_complete_(std::move(completion)); { std::lock_guard lane_lk(lane_mu_); @@ -641,6 +704,13 @@ void LocalMailboxEndpoint::submit_progress(Ring *ring, const WorkerDispatch &dis ); } +#if SIMPLER_HOST_STRACE + simpler::host_trace::SpanScope frame_submit_trace( + "l3.frame_submit", state.run_id, trace_callable_hash(ring, dispatch.task_slot), 0, + trace_dispatch_attrs(state.run_id, dispatch, caps_, "worker") + ); +#endif + // The lease slot id is a native pipeline slot bounded by the runtime's // PipelineContract, not a mailbox frame number. A two-frame endpoint maps // the two 1:1; a single-frame endpoint always publishes to frame 0. @@ -893,14 +963,24 @@ bool LocalMailboxEndpoint::poll_progress(WorkerEndpointProgress &progress) { } bool LocalMailboxEndpoint::activate_progress(RunId run_id) { - std::lock_guard lk(progress_mu_); + std::unique_lock lk(progress_mu_); if (endpoint_poisoned_) return false; for (size_t index = 0; index < task_frame_count_; ++index) { FrameRecord &record = frames_[index]; if (!record.occupied || !record.dispatch.prepare_only || record.run_id != run_id) continue; +#if SIMPLER_HOST_STRACE + std::ostringstream activate_attrs; + activate_attrs << "run_id=" << run_id << " task_slot=" << record.dispatch.task_slot + << " group_index=" << record.dispatch.group_index << " worker_id=" << caps_.worker_id + << " dispatch_id=" << record.dispatch.dispatch_id + << " endpoint_kind=" << endpoint_kind_name(caps_.kind) + << " prepare_only=" << static_cast(record.dispatch.prepare_only) << " role=worker"; + simpler::host_trace::SpanScope activate_trace("l3.activate", run_id, 0, 0, activate_attrs.str()); +#endif record.activation_requested = true; char *frame = task_frame(index); (void)try_publish_activation(record, frame); + lk.unlock(); return true; } return false; diff --git a/src/common/log/host_log.cpp b/src/common/log/host_log.cpp index f35e16fb27..aaf508cbaa 100644 --- a/src/common/log/host_log.cpp +++ b/src/common/log/host_log.cpp @@ -21,19 +21,71 @@ #include #include #include +#include #include +#include #include #include +#if defined(__linux__) +#include +#endif + +#include "common/host_span.h" + using simpler::log::LogLevel; namespace { -// Every STRACE marker renders well inside this, so the heap fallback below -// stays off the path a traced run pays per span. +// Every STRACE marker renders well inside this allocation bound, so the heap +// fallback below stays off the path a traced run pays per span. This capacity +// is not an atomic-write guarantee. constexpr size_t kRecordStackCapacity = 2048; +// POSIX guarantees atomic pipe writes up to _POSIX_PIPE_BUF (512 bytes). A +// conservative bound for the logger prefix, fixed-width STRACE fields, and +// newline is 256 bytes, leaving the other half for the encoded text fields. +constexpr size_t kHostSpanNameCapacity = 64; +constexpr size_t kHostSpanAttributesCapacity = 192; +static_assert(kHostSpanNameCapacity + kHostSpanAttributesCapacity <= _POSIX_PIPE_BUF - 256); + +std::string encode_host_span_field(const char *value, size_t capacity, bool attributes) { + static constexpr char kHex[] = "0123456789ABCDEF"; + std::string encoded; + encoded.reserve(capacity); + bool truncated = false; + size_t last_unit_size = 0; + for (const unsigned char *p = reinterpret_cast(value); *p != '\0'; ++p) { + const unsigned char c = *p; + const bool printable = c >= 0x20 && c <= 0x7E; + const bool grammar_character = attributes && (c == ' ' || c == '='); + const bool safe = + printable && c != '%' && c != '[' && c != ']' && (grammar_character || (c != ' ' && c != '=')); + const size_t required = safe ? 1 : 3; + if (encoded.size() + required > capacity) { + truncated = true; + break; + } + if (safe) { + encoded.push_back(static_cast(c)); + } else { + encoded.push_back('%'); + encoded.push_back(kHex[c >> 4]); + encoded.push_back(kHex[c & 0x0F]); + } + last_unit_size = required; + } + // A `%XX` escape is one indivisible unit, so a marker written over the tail + // of a full field drops that whole unit rather than its last byte — which + // would leave `%0A` as the undecodable `%0~`. + if (truncated && capacity != 0) { + if (encoded.size() == capacity) encoded.resize(encoded.size() - last_unit_size); + encoded.push_back('~'); + } + return encoded; +} + // Renders the timestamp/thread/level prefix, the caller's message, and an // optional trailing newline into `buffer`, and returns the length of the whole // record. A return value of `capacity` or more means `buffer` holds only a @@ -88,6 +140,14 @@ void write_stderr(const char *record, size_t size) { } } +long host_trace_tid() { +#if defined(__linux__) && defined(SYS_gettid) + return static_cast(syscall(SYS_gettid)); +#else + return static_cast(getpid()); +#endif +} + } // namespace HostLogger &HostLogger::get_instance() { @@ -151,10 +211,11 @@ void HostLogger::emit(const char *level_tag, const char *func, const char *fmt, const bool append_newline = fmt[0] != '\0' && fmt[strlen(fmt) - 1] != '\n'; - // One write per record: a record of at most PIPE_BUF bytes reaches a shared - // pipe indivisibly, so forked workers writing a captured stderr cannot - // interleave inside it. A longer record — a large LOG_ERROR dump, say — has - // no such guarantee; STRACE markers stay well below the limit. + // One write per record avoids thread interleaving under mutex_. On a shared + // pipe, only records no larger than that pipe's PIPE_BUF are indivisible + // across forked writers. Machine-readable host spans are separately + // budgeted to the portable _POSIX_PIPE_BUF floor (512 bytes); longer human + // log records use this same best-effort write path without that promise. char stack_buffer[kRecordStackCapacity]; const size_t length = format_record(stack_buffer, sizeof(stack_buffer), ts, tid, level_tag, func, fmt, args, append_newline); @@ -205,3 +266,20 @@ extern "C" int simpler_log_init(int log_level) { HostLogger::get_instance().set_level(static_cast(log_level)); return 0; } + +extern "C" void simpler_log_emit_host_span(const SimplerHostSpan *span) { + if (span == nullptr || span->abi_version != SIMPLER_HOST_SPAN_ABI_VERSION || + span->struct_size < sizeof(SimplerHostSpan) || span->name == nullptr) { + return; + } + const std::string name = encode_host_span_field(span->name, kHostSpanNameCapacity, false); + const std::string attributes = + encode_host_span_field(span->attributes == nullptr ? "" : span->attributes, kHostSpanAttributesCapacity, true); + HostLogger::get_instance().log( + LogLevel::TIMING, "emit_host_span", + "[STRACE] v=1 pid=%d tid=%ld inv=%llu hid=%llx depth=%d name=%s ts=%lld dur=%lld %s", + static_cast(getpid()), host_trace_tid(), static_cast(span->invocation_id), + static_cast(span->callable_hash), span->depth, name.c_str(), + static_cast(span->timestamp_ns), static_cast(span->duration_ns), attributes.c_str() + ); +} diff --git a/src/common/log/include/common/host_span.h b/src/common/log/include/common/host_span.h new file mode 100644 index 0000000000..bc7391092f --- /dev/null +++ b/src/common/log/include/common/host_span.h @@ -0,0 +1,43 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#pragma once + +#include + +#define SIMPLER_HOST_SPAN_ABI_VERSION 1U + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct SimplerHostSpan { + uint32_t abi_version; + uint32_t struct_size; + uint64_t invocation_id; + uint64_t callable_hash; + int32_t depth; + int32_t reserved; + int64_t timestamp_ns; + int64_t duration_ns; + const char *name; + const char *attributes; +} SimplerHostSpan; + +typedef void (*SimplerLogEmitHostSpanFn)(const SimplerHostSpan *span); + +/* The logger owns this C ABI. Consumers call it through a bound + * SimplerLogEmitHostSpanFn so an absent logger remains a supported state. */ +void simpler_log_emit_host_span(const SimplerHostSpan *span); + +#ifdef __cplusplus +} +#endif diff --git a/src/common/log/include/common/host_span_scope.h b/src/common/log/include/common/host_span_scope.h new file mode 100644 index 0000000000..ac4d439423 --- /dev/null +++ b/src/common/log/include/common/host_span_scope.h @@ -0,0 +1,119 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +/** + * @file host_span_scope.h + * @brief C++ emit helpers over the SimplerHostSpan C ABI. + * + * Timestamps are ns on CLOCK_MONOTONIC (steady_clock), so spans emitted here + * are comparable with the STRACE markers in common/strace.h, with Python's + * time.monotonic_ns(), and across a fork. + */ + +#pragma once + +#include "common/host_span.h" +#include "profiling_config.h" + +#if SIMPLER_HOST_STRACE + +#include +#include +#include +#include +#include + +namespace simpler::host_trace { + +inline SimplerLogEmitHostSpanFn &sink_slot() noexcept { + static SimplerLogEmitHostSpanFn sink = nullptr; + return sink; +} + +/** Bind this module's sink slot to the process logger before threads or forks. */ +inline void bind_sink(SimplerLogEmitHostSpanFn sink) noexcept { sink_slot() = sink; } + +/** False when this module has not been bound to libsimpler_log.so. */ +inline bool sink_available() noexcept { return sink_slot() != nullptr; } + +inline int64_t now_ns() noexcept { + return std::chrono::duration_cast(std::chrono::steady_clock::now().time_since_epoch()) + .count(); +} + +inline void emit( + const char *name, uint64_t invocation_id, uint64_t callable_hash, int32_t depth, int64_t timestamp_ns, + int64_t duration_ns, const char *attributes +) noexcept { + SimplerLogEmitHostSpanFn sink = sink_slot(); + if (sink == nullptr) return; + const SimplerHostSpan span{ + SIMPLER_HOST_SPAN_ABI_VERSION, + sizeof(SimplerHostSpan), + invocation_id, + callable_hash, + depth, + 0, + timestamp_ns, + duration_ns, + name, + attributes + }; + sink(&span); +} + +/** Times its own scope and emits on destruction. `name` must outlive the scope; + * every call site passes a literal. */ +class SpanScope { +public: + SpanScope(const char *name, uint64_t invocation_id, uint64_t callable_hash, int32_t depth, std::string attributes) : + name_(name), + invocation_id_(invocation_id), + callable_hash_(callable_hash), + depth_(depth), + timestamp_ns_(now_ns()), + attributes_(std::move(attributes)) {} + + ~SpanScope() { + const int64_t end_ns = now_ns(); + emit(name_, invocation_id_, callable_hash_, depth_, timestamp_ns_, end_ns - timestamp_ns_, attributes_.c_str()); + } + + SpanScope(const SpanScope &) = delete; + SpanScope &operator=(const SpanScope &) = delete; + SpanScope(SpanScope &&) = delete; + SpanScope &operator=(SpanScope &&) = delete; + +private: + const char *name_; + uint64_t invocation_id_; + uint64_t callable_hash_; + int32_t depth_; + int64_t timestamp_ns_; + std::string attributes_; +}; + +} // namespace simpler::host_trace + +#else + +#include + +namespace simpler::host_trace { + +inline void bind_sink(SimplerLogEmitHostSpanFn) noexcept {} +inline bool sink_available() noexcept { return false; } +inline int64_t now_ns() noexcept { return 0; } +inline void emit(const char *, uint64_t, uint64_t, int32_t, int64_t, int64_t, const char *) noexcept {} + +} // namespace simpler::host_trace + +#endif diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index bbe05a346e..03037dd498 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -224,6 +224,7 @@ add_library(hierarchical_objs OBJECT ) target_include_directories(hierarchical_objs PUBLIC ${HIERARCHICAL_SRC_DIR} + ${CMAKE_SOURCE_DIR}/../../../src/common/log/include ${CMAKE_SOURCE_DIR}/../../../src/common/platform/include ${CMAKE_SOURCE_DIR}/../../../src/common/task_interface ${WORKER_SRC_DIR} diff --git a/tests/ut/cpp/a5/test_host_log_off.cpp b/tests/ut/cpp/a5/test_host_log_off.cpp index 86564e07ed..a1f97eb987 100644 --- a/tests/ut/cpp/a5/test_host_log_off.cpp +++ b/tests/ut/cpp/a5/test_host_log_off.cpp @@ -15,6 +15,7 @@ #include #include +#include #include #include #include @@ -26,6 +27,7 @@ #include +#include "common/host_span.h" #include "host_log.h" using simpler::log::LogLevel; @@ -54,7 +56,8 @@ int capture_cann_log_level(int module_id, int level, int enable_event) { return 0; } -CapturedStdio run_with_config(LogLevel level, void (*fn)()) { +template +CapturedStdio run_with_config(LogLevel level, Fn &&fn) { fflush(stdout); fflush(stderr); FILE *out_tmp = tmpfile(); @@ -205,6 +208,66 @@ TEST(HostLogTest, AllOutputGoesToStderr) { EXPECT_NE(captured.err.find("debug-output-marker"), std::string::npos); } +TEST(HostLogTest, HostSpanEscapesDelimitersAndFitsAtomicPipeRecord) { + const std::string name = "bad name\n[STRACE]=x"; + const std::string attributes = "run_id=7 role=worker\n[STRACE] injected=1 " + std::string(4096, 'x'); + const SimplerHostSpan span{SIMPLER_HOST_SPAN_ABI_VERSION, + sizeof(SimplerHostSpan), + 7, + 0x1234, + 0, + 0, + 100, + 25, + name.c_str(), + attributes.c_str()}; + + auto captured = run_with_config(LogLevel::TIMING, [&] { + simpler_log_emit_host_span(&span); + }); + + const size_t marker = captured.err.find("[STRACE]"); + ASSERT_NE(marker, std::string::npos); + EXPECT_EQ(captured.err.find("[STRACE]", marker + 1), std::string::npos); + EXPECT_NE(captured.err.find("name=bad%20name%0A%5BSTRACE%5D%3Dx"), std::string::npos); + EXPECT_NE(captured.err.find("run_id=7 role=worker%0A%5BSTRACE%5D injected=1"), std::string::npos); + EXPECT_EQ(std::count(captured.err.begin(), captured.err.end(), '\n'), 1); + EXPECT_LE(captured.err.size(), static_cast(_POSIX_PIPE_BUF)); + ASSERT_GE(captured.err.size(), 2u); + EXPECT_EQ(captured.err[captured.err.size() - 2], '~'); +} + +// A `%XX` escape is three bytes that only mean anything together, so a field +// that fills its budget exactly on one must lose the whole escape to the +// truncation marker. Overwriting just the last byte would leave `%0A` as `%0~`, +// which no decoder can read back. +TEST(HostLogTest, HostSpanTruncationDropsAWholeEscapeRatherThanItsLastByte) { + // 3 (leading escape) + 186 + 3 (trailing escape) is exactly the 192-byte + // attribute budget, so the next byte truncates on an escape boundary. + const std::string attributes = "\n" + std::string(186, 'x') + "\ny"; + const SimplerHostSpan span{SIMPLER_HOST_SPAN_ABI_VERSION, + sizeof(SimplerHostSpan), + 7, + 0x1234, + 0, + 0, + 100, + 25, + "l3.dispatch", + attributes.c_str()}; + + auto captured = run_with_config(LogLevel::TIMING, [&] { + simpler_log_emit_host_span(&span); + }); + + EXPECT_EQ(captured.err.find("%0~"), std::string::npos) << "truncation marker landed inside an escape"; + // The leading escape survives whole; the trailing one is gone entirely. + EXPECT_NE(captured.err.find("dur=25 %0Axxx"), std::string::npos); + EXPECT_EQ(std::count(captured.err.begin(), captured.err.end(), '%'), 1); + ASSERT_GE(captured.err.size(), 2u); + EXPECT_EQ(captured.err[captured.err.size() - 2], '~'); +} + TEST(HostLogTest, ForkedProcessesEmitWholePipeRecords) { int log_pipe[2]; int start_pipe[2]; diff --git a/tests/ut/cpp/hierarchical/test_scheduler.cpp b/tests/ut/cpp/hierarchical/test_scheduler.cpp index a23430f483..93b768f90a 100644 --- a/tests/ut/cpp/hierarchical/test_scheduler.cpp +++ b/tests/ut/cpp/hierarchical/test_scheduler.cpp @@ -30,6 +30,7 @@ #include #include #include +#include #include #include #include @@ -37,6 +38,8 @@ #include #include "call_config.h" +#include "common/host_span.h" +#include "common/host_span_scope.h" #include "orchestrator.h" #include "ring.h" #include "scheduler.h" @@ -46,6 +49,51 @@ #include "worker_manager.h" #include "task_args.h" +namespace { + +std::mutex captured_host_spans_mu; +std::vector captured_host_span_names; +std::vector captured_host_span_attributes; + +class ScopedHostSpanCapture { +public: + ScopedHostSpanCapture() { + std::lock_guard lk(captured_host_spans_mu); + captured_host_span_names.clear(); + captured_host_span_attributes.clear(); + simpler::host_trace::bind_sink(&simpler_log_emit_host_span); + } + + ~ScopedHostSpanCapture() { simpler::host_trace::bind_sink(nullptr); } + + ScopedHostSpanCapture(const ScopedHostSpanCapture &) = delete; + ScopedHostSpanCapture &operator=(const ScopedHostSpanCapture &) = delete; +}; + +bool captured_host_span(const std::string &name) { + std::lock_guard lk(captured_host_spans_mu); + return std::find(captured_host_span_names.begin(), captured_host_span_names.end(), name) != + captured_host_span_names.end(); +} + +// The attributes of the first span emitted under `name`, or "" if none was. +std::string captured_host_span_attrs(const std::string &name) { + std::lock_guard lk(captured_host_spans_mu); + for (size_t i = 0; i < captured_host_span_names.size(); ++i) { + if (captured_host_span_names[i] == name) return captured_host_span_attributes[i]; + } + return ""; +} + +} // namespace + +extern "C" void simpler_log_emit_host_span(const SimplerHostSpan *span) { + if (span == nullptr || span->name == nullptr) return; + std::lock_guard lk(captured_host_spans_mu); + captured_host_span_names.emplace_back(span->name); + captured_host_span_attributes.emplace_back(span->attributes == nullptr ? "" : span->attributes); +} + // --------------------------------------------------------------------------- // MockMailboxWorker: in-process stand-in for the forked Python child loop. // @@ -1141,6 +1189,40 @@ TEST(WorkerManagerTest, WorkerThreadUsesOneProgressOwnerForActiveAndStagedLanes) allocator.shutdown(); } +TEST(WorkerManagerTest, DispatchAndCompletionEmitHostSpans) { + ScopedHostSpanCapture host_span_capture; + + Ring allocator; + allocator.init(/*heap_bytes=*/0); + TaskSlot slot = make_progress_slot(allocator, /*run_id=*/73, /*pipeline_slot=*/0, /*generation=*/1); + ASSERT_NE(slot, INVALID_SLOT); + + WorkerThread worker; + auto endpoint = std::make_unique(); + DeterministicProgressEndpoint *endpoint_ptr = endpoint.get(); + std::vector completed; + worker.start( + &allocator, + [&](WorkerCompletion completion) { + completed.push_back(std::move(completion)); + }, + [](WorkerDispatch) {}, std::move(endpoint) + ); + + worker.dispatch(WorkerDispatch{slot, 0}); + ASSERT_TRUE(endpoint_ptr->wait_submitted(1)); + EXPECT_TRUE(captured_host_span("l3.dispatch")); + + const WorkerDispatch submitted = endpoint_ptr->submitted().front(); + endpoint_ptr->emit(WorkerProgressKind::COMPLETED, submitted); + worker.progress(); + ASSERT_EQ(completed.size(), 1u); + EXPECT_TRUE(captured_host_span("l3.complete")); + + worker.stop(); + allocator.shutdown(); +} + TEST(WorkerManagerTest, AdmissionRejectionsCompleteClaimedDispatchesWithoutThrowing) { Ring allocator; allocator.init(/*heap_bytes=*/0); @@ -1350,6 +1432,8 @@ TEST(WorkerManagerTest, SubmitExceptionQuiescesEndpointOwnedPublication) { } TEST(WorkerManagerTest, TwoFrameLeaseSlotsDoNotDefineFifoOrAcceptance) { + ScopedHostSpanCapture host_span_capture; + alignas(8) std::array mailbox{}; Ring allocator; allocator.init(/*heap_bytes=*/0); @@ -1391,6 +1475,8 @@ TEST(WorkerManagerTest, TwoFrameLeaseSlotsDoNotDefineFifoOrAcceptance) { ASSERT_TRUE(endpoint.poll_progress(progress)); EXPECT_EQ(progress.kind, WorkerProgressKind::ACCEPTED); EXPECT_EQ(progress.dispatch.dispatch_id, 42u); + EXPECT_TRUE(captured_host_span("l3.frame_submit")); + EXPECT_TRUE(captured_host_span("l3.activate")); allocator.shutdown(); } @@ -1947,6 +2033,22 @@ struct CapacityOneProgressSchedulerFixture : public ProgressSchedulerFixture { uint32_t endpoint0_capacity() const override { return 1; } }; +TEST_F(ProgressSchedulerFixture, GroupSubmitReportsNoSingleWorkerAndNoSingleIndex) { + ScopedHostSpanCapture host_span_capture; + + RunId run = orchestrator.begin_run(); + SubmitResult group = orchestrator.submit_next_level_group( + C(9), {single_tensor_args(0x9000, TensorArgType::OUTPUT), single_tensor_args(0xA000, TensorArgType::OUTPUT)}, + config, {0, 1} + ); + orchestrator.close_run_submission(run); + + ASSERT_TRUE(captured_host_span("l3.submit")); + std::ostringstream expected; + expected << "run_id=" << run << " task_slot=" << group.task_slot << " group_index=-1 group_size=2 role=facade"; + EXPECT_EQ(captured_host_span_attrs("l3.submit"), expected.str()); +} + TEST_F(ProgressSchedulerFixture, SuccessorStagesButActivatesOnlyAfterFifoPromotion) { RunId first_run = orchestrator.begin_run(); SubmitResult first = @@ -2405,6 +2507,26 @@ TEST_F(SchedulerFixture, ACancellationThatWinsTheClaimStopsTheDispatch) { } } +// `l3.submit` is what a dispatch arrow is drawn back to, so the swimlane can +// only pair the two if these attributes carry the same run/slot the dispatch +// reports. A SUB submission is deliberately silent: it has no NEXT_LEVEL worker +// to hand off to. +TEST_F(SchedulerFixture, NextLevelSubmitEmitsAPairableHostSpan) { + ScopedHostSpanCapture host_span_capture; + + auto submitted = orch.submit_next_level(C(0x42), single_tensor_args(0xCAFE, TensorArgType::OUTPUT), cfg, 0); + + ASSERT_TRUE(captured_host_span("l3.submit")); + std::ostringstream expected; + expected << "run_id=" << run_id << " task_slot=" << submitted.task_slot + << " group_index=0 group_size=1 worker_id=0 role=facade"; + EXPECT_EQ(captured_host_span_attrs("l3.submit"), expected.str()); + + mock_worker.wait_running(); + mock_worker.complete(); + wait_consumed(submitted.task_slot); +} + TEST_F(SchedulerFixture, IndependentTaskDispatchedAndConsumed) { auto args_a = single_tensor_args(0xCAFE, TensorArgType::OUTPUT); auto res = orch.submit_next_level(C(42), args_a, cfg, 0); diff --git a/tests/ut/py/test_package_surface.py b/tests/ut/py/test_package_surface.py index d50ad47258..b7818d2c21 100644 --- a/tests/ut/py/test_package_surface.py +++ b/tests/ut/py/test_package_surface.py @@ -89,6 +89,21 @@ def test_importing_simpler_survives_without_the_extension(): assert out.stdout.strip() == "True", f"{out.stdout!r} {out.stderr!r}" +def test_importing_simpler_keeps_native_default_when_old_extension_lacks_sink_binder(): + """A stale extension may expose the threshold before it exposes the optional sink binder.""" + code = ( + "import sys, types; " + "stub = types.ModuleType('_task_interface'); " + "stub.DEFAULT_LOG_THRESHOLD = 30; " + "sys.modules['_task_interface'] = stub; " + "import simpler; print(simpler.DEFAULT_THRESHOLD)" + ) + out = subprocess.run( # noqa: S603 -- fixed argv, no shell + [sys.executable, "-c", code], capture_output=True, text=True, check=True + ) + assert out.stdout.strip() == "30", f"{out.stdout!r} {out.stderr!r}" + + def test_comm_endpoints_requires_the_extension_and_stays_lazy(): """W2 planning takes `BackendKind` from the extension, so it is not part of the no-extension surface. diff --git a/tests/ut/py/test_runtime_builder.py b/tests/ut/py/test_runtime_builder.py index e648bbb99d..0e41e1130b 100644 --- a/tests/ut/py/test_runtime_builder.py +++ b/tests/ut/py/test_runtime_builder.py @@ -615,6 +615,71 @@ def _raise_bad_pin(): builder._resolve_build_pto_isa_commit() +class TestPlaceBinary: + """place_binary must never rewrite the destination in place. + + A build that replaces a .so already dlopened by the running process — which + every `pip install -e .` does, and which `get_binaries(build=True)` does to + build/lib/libsimpler_log.so mid-test-session — corrupts that mapping if the + destination is truncated and rewritten. The fault surfaces far away, as a + SIGSEGV inside the dynamic linker on an unrelated dlopen or at interpreter + exit, so the invariant is asserted here rather than left to a crash. + """ + + def _library(self, path: Path) -> Path: + """Compile a trivial shared object at `path`.""" + import subprocess # noqa: PLC0415 + + source = path.with_suffix(".c") + source.write_text("int probe(void) { return 7; }\n", encoding="utf-8") + # gcc is a fixed test-only toolchain dependency, not user input. + subprocess.run( # noqa: S603 + ["gcc", "-shared", "-fPIC", "-o", str(path), str(source)], # noqa: S607 + check=True, + ) + return path + + def test_replaces_by_rename_not_in_place(self, tmp_path): + from simpler_setup.runtime_compiler import place_binary # noqa: PLC0415 + + dest = self._library(tmp_path / "libprobe.so") + src = self._library(tmp_path / "libnewer.so") + before = dest.stat().st_ino + + place_binary(src, dest) + + assert dest.stat().st_ino != before, "destination was rewritten in place, not renamed over" + + def test_leaves_an_open_mapping_of_the_old_inode_valid(self, tmp_path): + import ctypes # noqa: PLC0415 + + from simpler_setup.runtime_compiler import place_binary # noqa: PLC0415 + + dest = self._library(tmp_path / "libprobe.so") + handle = ctypes.CDLL(str(dest)) + assert handle.probe() == 7 + + place_binary(self._library(tmp_path / "libnewer.so"), dest) + + # The already-loaded copy keeps its own inode, so calling into it after + # the replacement is still defined behaviour. + assert handle.probe() == 7 + + def test_leaves_no_temporary_behind_when_post_process_fails(self, tmp_path): + from simpler_setup.runtime_compiler import place_binary # noqa: PLC0415 + + dest = self._library(tmp_path / "libprobe.so") + src = self._library(tmp_path / "libnewer.so") + + def _fail(_staged): + raise RuntimeError("strip failed") + + with pytest.raises(RuntimeError, match="strip failed"): + place_binary(src, dest, post_process=_fail) + + assert [p.name for p in tmp_path.iterdir() if p.name.endswith(".tmp")] == [] + + # --- Full integration tests (real compilation) --- diff --git a/tests/ut/py/test_strace_timing.py b/tests/ut/py/test_strace_timing.py index 38228db9f0..c7fa8ecb09 100644 --- a/tests/ut/py/test_strace_timing.py +++ b/tests/ut/py/test_strace_timing.py @@ -8,7 +8,18 @@ # See LICENSE in the root of the software repository for the full text of the License. # ----------------------------------------------------------------------------------------------------------- -from simpler_setup.tools.strace_timing import count_record_heads, parse_spans +import json + +from simpler_setup.tools.strace_timing import ( + bucket_by_hid, + count_record_heads, + group_invocations, + legacy_spans, + main, + parse_spans, + to_chrome_trace, + to_host_swimlane, +) def _record(pid, inv, name, attrs=""): @@ -60,3 +71,322 @@ def test_count_record_heads_sees_a_torn_record_that_parse_spans_drops(): assert count_record_heads(lines) == 2 assert len(list(parse_spans(lines))) == 1 + + +def _span_record( + *, + pid: int, + tid: int, + inv: int, + name: str, + ts: int, + dur: int, + attrs: str = "", + hid: str = "abc", + depth: int = 0, +) -> str: + return f"[STRACE] v=1 pid={pid} tid={tid} inv={inv} hid={hid} depth={depth} name={name} ts={ts} dur={dur} {attrs}\n" + + +def test_host_swimlane_keeps_real_host_lanes_and_builds_dispatch_flow(): + lines = [ + _span_record( + pid=41, + tid=410, + inv=7, + name="l3.graph_build", + ts=1_000, + dur=900, + attrs="run_id=7 role=facade", + ), + _span_record( + pid=41, + tid=410, + inv=7, + name="l3.submit", + ts=1_100, + dur=100, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 role=facade", + ), + _span_record( + pid=41, + tid=411, + inv=7, + name="l3.dispatch", + ts=1_400, + dur=80, + attrs=( + "run_id=7 task_slot=12 group_index=0 worker_id=3 dispatch_id=99 " + "endpoint_kind=local_mailbox role=scheduler" + ), + ), + ] + + trace = to_host_swimlane(list(parse_spans(lines))) + events = trace["traceEvents"] + slices = [event for event in events if event["ph"] == "X"] + flows = [event for event in events if event["ph"] in {"s", "f"}] + + assert {(event["pid"], event["tid"]) for event in slices} == {(41, 410), (41, 411)} + assert [(event["ph"], event["pid"], event["tid"]) for event in flows] == [ + ("s", 41, 410), + ("f", 41, 411), + ] + assert flows[0]["id"] == flows[1]["id"] + assert isinstance(flows[0]["id"], int) + assert slices[-1]["args"]["dispatch_id"] == 99 + thread_names = { + (event["pid"], event["tid"]): event["args"]["name"] + for event in events + if event["ph"] == "M" and event["name"] == "thread_name" + } + assert thread_names == {(41, 410): "orchestrator / facade", (41, 411): "scheduler"} + + +def test_host_swimlane_names_the_scheduler_lane_from_every_role_it_emits(): + """The scheduler thread emits `role=worker` spans before its first `role=scheduler` one. + + `scheduler.cpp`'s loop is the only caller of both `dispatch_ready()` and + `manager->progress()`, and inside `submit_dispatch` the `l3.frame_submit` + scope closes within `submit_progress` — before the `l3.dispatch` record + emitted after the admission lock is released. Naming the lane from the + first span it emitted therefore labels the scheduler `worker 3`. + """ + lines = [ + _span_record( + pid=41, + tid=411, + inv=7, + name="l3.frame_submit", + ts=1_410, + dur=40, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 dispatch_id=1 role=worker", + ), + _span_record( + pid=41, + tid=411, + inv=7, + name="l3.dispatch", + ts=1_400, + dur=80, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 dispatch_id=1 role=scheduler", + ), + _span_record( + pid=41, + tid=411, + inv=7, + name="l3.complete", + ts=2_000, + dur=30, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 dispatch_id=1 role=worker", + ), + ] + + trace = to_host_swimlane(list(parse_spans(lines))) + thread_names = { + (event["pid"], event["tid"]): event["args"]["name"] + for event in trace["traceEvents"] + if event["ph"] == "M" and event["name"] == "thread_name" + } + + assert thread_names == {(41, 411): "scheduler"} + + +def test_parse_spans_decodes_percent_escaped_name_and_attribute_values(): + """`encode_host_span_field` escapes whatever would otherwise be record grammar.""" + lines = [ + _span_record( + pid=41, + tid=410, + inv=7, + name="l3.odd%20name", + ts=100, + dur=10, + attrs="run_id=7 reason=submit%20failed%3A%20%5Bfatal%5D role=facade", + ) + ] + + span = next(iter(parse_spans(lines))) + assert span.name == "l3.odd name" + + trace = to_host_swimlane([span]) + args = next(event["args"] for event in trace["traceEvents"] if event["ph"] == "X") + assert args["reason"] == "submit failed: [fatal]" + # The raw field stays verbatim: it is the record as written. + assert "%5Bfatal%5D" in args["attrs"] + + +def test_host_swimlane_pairs_dispatch_with_latest_preceding_submit(): + lines = [ + _span_record( + pid=41, + tid=410, + inv=7, + name="l3.submit", + ts=100, + dur=20, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 role=facade", + ), + _span_record( + pid=41, + tid=411, + inv=7, + name="l3.dispatch", + ts=200, + dur=10, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 dispatch_id=1 role=scheduler", + ), + _span_record( + pid=41, + tid=410, + inv=7, + name="l3.submit", + ts=300, + dur=20, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 role=facade", + ), + _span_record( + pid=41, + tid=411, + inv=7, + name="l3.dispatch", + ts=400, + dur=10, + attrs="run_id=7 task_slot=12 group_index=0 worker_id=3 dispatch_id=2 role=scheduler", + ), + ] + + trace = to_host_swimlane(list(parse_spans(lines))) + flows = [event for event in trace["traceEvents"] if event["ph"] in {"s", "f"}] + + assert [event["id"] for event in flows] == [1, 1, 2, 2] + assert [event["ts"] for event in flows if event["ph"] == "s"] == [0.12, 0.32] + assert [event["args"]["dispatch_key"] for event in flows] == [ + "dispatch:41:7:12:0:3:1", + "dispatch:41:7:12:0:3:1", + "dispatch:41:7:12:0:3:2", + "dispatch:41:7:12:0:3:2", + ] + + +def test_host_swimlane_keeps_unaligned_device_clock_out_of_visible_timeline(): + spans = list( + parse_spans( + [ + _span_record( + pid=41, + tid=410, + inv=7, + name="l3.graph_build", + ts=1_000_000_000, + dur=900, + attrs="run_id=7 role=facade", + ), + _span_record( + pid=52, + tid=520, + inv=8, + name="simpler_run.runner_run.device_wall", + ts=300, + dur=40, + attrs="clk=dev rank=1", + ), + ] + ) + ) + + trace = to_host_swimlane(spans) + visible_slices = [event for event in trace["traceEvents"] if event.get("ph") == "X"] + + assert [(event["name"], event["ts"], event["dur"]) for event in visible_slices] == [ + ("l3.graph_build", 1_000_000.0, 0.9) + ] + assert trace["unalignedDeviceSpans"] == [ + { + "name": "simpler_run.runner_run.device_wall", + "ts_ns": 300, + "dur_ns": 40, + "pid": 52, + "tid": 520, + "inv": 8, + "hid": "abc", + "depth": 0, + "attrs": {"raw": "clk=dev rank=1", "clk": "dev", "rank": 1}, + } + ] + + +def test_legacy_trace_output_ignores_host_swimlane_markers(): + old = list( + parse_spans( + [ + _span_record(pid=61, tid=610, inv=3, name="simpler_run", ts=1_000, dur=500), + _span_record(pid=61, tid=610, inv=3, name="simpler_run.bind", ts=1_100, dur=50, depth=1), + _span_record(pid=61, tid=610, inv=4, name="simpler_prewarm.build", ts=1_200, dur=75), + ] + ) + ) + mixed = old + list( + parse_spans( + [ + _span_record( + pid=61, + tid=611, + inv=9, + name="l3.dispatch", + ts=900, + dur=20, + attrs="run_id=9 task_slot=4 group_index=0 worker_id=0 dispatch_id=1", + ) + ] + ) + ) + + old_invocations = group_invocations(legacy_spans(old)) + mixed_invocations = group_invocations(legacy_spans(mixed)) + + assert [span.name for span in legacy_spans(mixed)] == [ + "simpler_run", + "simpler_run.bind", + "simpler_prewarm.build", + ] + assert to_chrome_trace(old_invocations, bucket_by_hid(old_invocations)) == to_chrome_trace( + mixed_invocations, bucket_by_hid(mixed_invocations) + ) + + +def test_invocation_by_name_uses_earliest_timestamp_not_input_order(): + spans = list( + parse_spans( + [ + _span_record(pid=61, tid=610, inv=3, name="simpler_run.bind", ts=1_200, dur=20, depth=1), + _span_record(pid=61, tid=611, inv=3, name="simpler_run.bind", ts=1_100, dur=30, depth=1), + ] + ) + ) + + invocation = group_invocations(spans)[0] + + assert invocation.by_name()["simpler_run.bind"].ts == 1_100 + + +def test_swimlane_cli_writes_trace(tmp_path): + log_path = tmp_path / "run.log" + output_path = tmp_path / "host_swimlane.json" + log_path.write_text( + _span_record( + pid=71, + tid=710, + inv=2, + name="l3.graph_build", + ts=100, + dur=25, + attrs="run_id=2 role=facade", + ), + encoding="utf-8", + ) + + assert main([str(log_path), "--swimlane", str(output_path)]) == 0 + + trace = json.loads(output_path.read_text(encoding="utf-8")) + assert any(event.get("name") == "l3.graph_build" for event in trace["traceEvents"]) diff --git a/tests/ut/py/test_worker/test_host_worker.py b/tests/ut/py/test_worker/test_host_worker.py index e21a10285d..1bc52843f0 100644 --- a/tests/ut/py/test_worker/test_host_worker.py +++ b/tests/ut/py/test_worker/test_host_worker.py @@ -229,10 +229,12 @@ def get_orchestrator(self): runtime="host_build_graph", ) worker._chip_shms = [SharedMemory(create=True, size=MAILBOX_SIZE) for _ in range(2)] + worker._l3_bins = "bins" fake_parent = FakeParentWorker() worker._worker = cast(Any, fake_parent) worker._startup_deadline = time.monotonic() + 5.0 fork_pids = iter((12001, 12002)) + startup_events: list[tuple] = [] def fake_await_children_ready(shms, _pids, kind: str, _deadline: float) -> None: if kind != "chip": @@ -241,7 +243,18 @@ def fake_await_children_ready(shms, _pids, kind: str, _deadline: float) -> None: assert shm.buf is not None worker_mod._PIPELINE_LEASE_FMT.pack_into(shm.buf, worker_mod._OFF_PIPELINE_LEASE, depth, 0, 0) - monkeypatch.setattr(worker_mod.os, "fork", lambda: next(fork_pids)) + def fake_fork() -> int: + startup_events.append(("fork",)) + return next(fork_pids) + + monkeypatch.setattr(worker_mod.os, "fork", fake_fork) + monkeypatch.setattr(worker_mod._simpler_log, "get_current_config", lambda: 60) + monkeypatch.setattr( + worker_mod, + "_initialize_simpler_log", + lambda bins, level: startup_events.append(("log", bins, level)), + raising=False, + ) monkeypatch.setattr(worker, "_await_children_ready", fake_await_children_ready) monkeypatch.setattr(worker_mod, "Orchestrator", lambda native, owner: (native, owner)) try: @@ -254,6 +267,78 @@ def fake_await_children_ready(shms, _pids, kind: str, _deadline: float) -> None: assert fake_parent.configured_depths == [1] assert [call[1:] for call in fake_parent.next_level_calls] == [(12001, 2), (12002, 1)] assert fake_parent.initialized + assert startup_events[0] == ("log", "bins", 60) + assert startup_events[1:] == [("fork",), ("fork",)] + + +def test_start_hierarchical_seeds_the_logger_when_the_process_owns_no_chips(monkeypatch): + """A chipless hierarchical process still runs a scheduler that emits host spans. + + `init()` rejects `device_ids` above L3, so gating the seeding on them would + leave exactly the pod processes unseeded — and their `HostLogger` would keep + its constructor default however the `simpler` logger is configured. + """ + + class FakeParentWorker: + def __init__(self) -> None: + self.initialized = False + self.sub_workers: list[int] = [] + + def configure_pipeline_depth(self, depth: int) -> None: + pass + + def add_sub_worker(self, _mailbox_addr: int, pid: int) -> None: + self.sub_workers.append(pid) + + def init(self) -> None: + self.initialized = True + + def get_orchestrator(self): + return object() + + worker = Worker( + level=3, + device_ids=[], + num_sub_workers=1, + platform="a2a3", + runtime="host_build_graph", + ) + worker._sub_shms = [SharedMemory(create=True, size=MAILBOX_SIZE)] + worker._worker = cast(Any, FakeParentWorker()) + worker._startup_deadline = time.monotonic() + 5.0 + startup_events: list[tuple] = [] + + def fake_fork() -> int: + startup_events.append(("fork",)) + return 13001 + + monkeypatch.setattr(worker_mod.os, "fork", fake_fork) + monkeypatch.setattr(worker_mod._simpler_log, "get_current_config", lambda: 60) + monkeypatch.setattr( + worker_mod, + "_initialize_simpler_log", + lambda bins, level: startup_events.append(("log", bins, level)), + raising=False, + ) + monkeypatch.setattr(worker, "_await_children_ready", lambda *args, **kwargs: None) + monkeypatch.setattr(worker_mod, "Orchestrator", lambda native, owner: (native, owner)) + try: + worker._start_hierarchical() + finally: + for shm in worker._sub_shms: + shm.close() + shm.unlink() + + # bins is None: this process loads no chip binaries, so the copy the package + # already preloaded is the one to seed. + assert startup_events[0] == ("log", None, 60) + assert startup_events[1:] == [("fork",)] + + +def test_a_worker_above_l3_can_never_carry_device_ids(): + """The premise behind seeding unconditionally in `_start_hierarchical`.""" + with pytest.raises(RuntimeError, match="device_ids are only supported on L3 Workers"): + Worker(level=4, device_ids=[0], num_sub_workers=0, platform="a2a3", runtime="host_build_graph").init() class _FakeChipRun: @@ -2658,6 +2743,22 @@ def bad_graph(*_args): assert not worker._accepted_run_handles assert worker._ordered_cleanup_error is None + def test_graph_failure_still_emits_graph_build_span(self, monkeypatch): + worker, _events = self._submission_failure_worker(failures=0) + emitted = [] + timestamps = iter((100, 275)) + monkeypatch.setattr(worker_mod, "_host_spans_active", lambda: True) + monkeypatch.setattr(worker_mod.time, "monotonic_ns", lambda: next(timestamps)) + monkeypatch.setattr(worker_mod, "_emit_host_span", lambda *args: emitted.append(args)) + + def bad_graph(*_args): + raise ValueError("bad graph") + + with pytest.raises(ValueError, match="bad graph"): + worker._submit_l3_locked(bad_graph, None, cast(Any, object())) + + assert emitted == [("l3.graph_build", 1, 0, 0, 100, 175, "run_id=1 role=facade")] + def test_unsettled_graph_cancellation_abandons_the_handle_before_close(self): worker, events = self._submission_failure_worker(failures=2) graph_error = ValueError("bad graph")