diff --git a/docs/dfx/host-trace.md b/docs/dfx/host-trace.md index 26e0df14bf..5232e51edd 100644 --- a/docs/dfx/host-trace.md +++ b/docs/dfx/host-trace.md @@ -40,7 +40,7 @@ writes follows it: `LOG_*` records, `[STRACE]` spans, `[CLOCK_ANCHOR]`, the specially, so there is no state in which part of a run's log is in one place and part in another. The first non-empty directory in a process wins, and a record falls back to stderr rather than being lost whenever the file cannot take it: a -path that does not fit, a failed open, a failed write, or a failed flush. +path that does not fit, a failed open, or a failed write. **Python's own log records are part of this stream too.** Seeding the native threshold installs a handler on the `simpler` logger that forwards each record @@ -52,15 +52,19 @@ the log rather than a second half of it. Records logged before a worker is initialized have no handler to forward through yet and stay on the root logger. See [logging.md](../logging.md). -Each process's file is fully buffered. It is flushed when a depth-zero invocation -record completes, which is the point at which the records so far describe a whole -invocation, and whenever a `WARN` or `ERROR` record is written, which is rare and -worth having on disk if the process dies. That flush runs on the thread that -emitted the record, so this reduces the observer effect by roughly the ratio of -records to flushes rather than removing it: expect a residual tail at each flush -boundary, not a clean floor. A system tracer would hand the bytes to a consumer -instead, and would bound its buffer and count what it drops; this does neither, -deliberately. +Each process has one bounded background writer. A producer formats a complete +record of at most 512 bytes and submits it through a 4096-slot lock-free queue; +after that writer is published, it never performs file/stderr I/O or waits for a +flush. The pre-writer hierarchical initialization window is the deliberate +exception: it writes synchronously so startup diagnostics survive the final +local fork. The writer appends each accepted record directly to the process file +(or stderr fallback). Queue-full, bounded-claim failure, and final output failure +increment an explicit process drop counter. Worker shutdown and `os._exit()` +call sites wait only boundedly for accepted records, so a stuck output cannot +turn task execution or teardown into an unbounded wait; a timeout reports both +pending and dropped counts. `WARN`, `ERROR`, and depth-zero spans use the same +async path as every other steady-state record. An overlong ordinary record is +truncated to 512 bytes and ends in `~\n`. ## Reading a run back diff --git a/docs/logging.md b/docs/logging.md index f72f3287d0..ab40b0b89d 100644 --- a/docs/logging.md +++ b/docs/logging.md @@ -29,9 +29,9 @@ Each host module: └─ simpler_host_log_bind_state(state*) Device logging: - AICPU keeps the device backend - ├─ sim: set_log_level(...) seeds its current level flags - └─ onboard: CANN level is sampled during device init + dev_vlog_* compatibility interface + ├─ sim: bound HostLogger → same state, envelope, and output + └─ onboard: separate CANN backend sampled during device init ``` One threshold controls `DEBUG / INFO / TIMING / WARN / ERROR`; `NUL` suppresses @@ -59,7 +59,7 @@ src/common/platform/ ├── include/aicpu/device_log.h device backend declarations ├── shared/aicpu/unified_log_device.cpp LOG_* ABI → dev_vlog_* adapter ├── onboard/aicpu/device_log.cpp onboard CANN backend -└── sim/aicpu/device_log.cpp sim AICPU stderr backend +└── sim/aicpu/device_log.cpp dev_vlog_* → bound HostLogger adapter ``` There is no standalone `libsimpler_log.so`. Host consumers compile the two @@ -118,24 +118,44 @@ forwards to its local backend. ### Layer 3 — backend primitives -`HostLogger::vlog` is the host-side authority for level gating. It formats a -complete record and performs one `write(2)` under its module-local mutex. +`HostLogger::vlog` is the host-side authority for level gating. It formats one +bounded record (at most the portable 512-byte `PIPE_BUF` floor) and submits it +to the process-owned queue. The queue has 4096 fixed slots and a lock-free +multi-producer claim path. Once the writer is published, producers never perform +file or stderr I/O, wait on a condition variable, acquire the drain mutex, or +grow memory. During hierarchical initialization only, before the final local +fork and writer startup, records use the synchronous destination so startup +failures are not silently discarded. A full queue or exhausted bounded claim is +a dropped record and increments +`dropped_record_count`. An immediately-unlinked named semaphore provides the +writer wakeup because unnamed semaphores are not available on every supported +host OS. + +Any ordinary human-readable record whose formatted envelope and body exceed +512 bytes is truncated to that fixed size and ends in `~\n`. This keeps one +record to one atomic pipe write when several forked processes share captured +stderr; callers that need a large payload must split it into separate records. + +One background writer drains the queue. When `CallConfig.output_prefix` is +present it appends every C++ host-log record to the process-private +`host..log`; otherwise it writes to stderr. A file open/write failure falls +back to stderr, and a failure of that final destination is counted as a drop. +Severity and span depth no longer choose synchronous producer-side flush paths. +Explicit lifecycle drains wait boundedly for records already accepted by the +process. + `HostLogger::log_host_span` additionally bounds and escapes machine-readable fields so a STRACE record fits the portable `PIPE_BUF` floor. Its early gate keeps direct ABI and legacy STRACE callers from paying those encoding costs when TIMING is disabled; `vlog` remains the final check if the threshold changes between the caller's query and emission. -The mutex serializes writers only within the DSO that owns it. Writers from -different DSOs, or from forked processes sharing a pipe, are indivisible only -when their single `write(2)` is no larger than that pipe's `PIPE_BUF`. -Machine-readable `[STRACE]` records satisfy that bound. Longer human-readable -records are best-effort and may interleave across module boundaries. Blocking -and drop accounting for those writes remain part of issue #1792 item 6. - -The AICPU `dev_vlog_*` functions remain separate. Sim formats a single stderr -record; onboard forwards through CANN dlog. Folding sim's device logger into -the host backend is tracked separately by issue #1792 item 5. +The AICPU `dev_vlog_*` interface remains source-compatible on both platforms. +Sim implements it as a thin `va_list` adapter into its bound `HostLogger`, so it +shares the live threshold, envelope, queue, destination, and fallback with +the other host-side modules in that process. Only real-silicon AICPU retains a +separate backend because its records go through CANN dlog rather than a host +process. ## Cross-DSO host state @@ -143,20 +163,35 @@ Each host DSO has a private `HostLogger` object, but every copy in one process reads the same `SimplerHostLogState`: ```c +struct SimplerHostLogState; +typedef int (*SimplerHostLogEnqueueFn)( + void *context, struct SimplerHostLogState *state, + const char *record, uint32_t size, int32_t anchor_pid); + typedef struct SimplerHostLogState { uint32_t abi_version; uint32_t struct_size; int32_t threshold; int32_t clock_anchor_pid; + int32_t log_directory_bound; + char log_directory[1024]; + int32_t sink_owner_pid; + int32_t sink_process_pid; + void *sink_context; + SimplerHostLogEnqueueFn sink_enqueue; + uint64_t dropped_record_count; + uint64_t pending_record_count; + uint64_t sink_producer_state; } SimplerHostLogState; int simpler_host_log_bind_state(SimplerHostLogState *state); ``` -The native `_task_interface` extension owns the state for the process. The -fields are plain fixed-width integers to keep the ABI compiler-independent; -`host_log.cpp` accesses mutable fields with atomic builtins. ABI version and -size are checked before a module accepts the pointer. +The native `_task_interface` extension owns the state for the process. Version, +size, thresholds, and counters use fixed-width integers; `sink_context` and +`sink_enqueue` are process-local native pointer values shared only by modules +from the same build. `host_log.cpp` accesses mutable fields with atomic +builtins. ABI version and size are checked before a module accepts the pointer. Only `simpler_host_log_bind_state` is exported from a host logging consumer. `HostLogger` and every `unified_log_*` definition are hidden. This avoids @@ -165,17 +200,40 @@ interposition while still giving loaders one stable binding entry point. `clock_anchor_pid` is also shared. Consequently the private logger copies coordinate one successful `[CLOCK_ANCHOR]` per process. A negative PID is a -temporary writer claim; a failed stderr write releases the claim so the next -record can retry. +temporary writer claim; a failed output releases the claim so the next record +can retry. The first non-empty `log_directory` binding wins, so every bound DSO +in the process chooses the same output without moving a file already in use. + +The owner publishes a C callback and opaque context in the same state. A private +logger in any bound DSO can therefore submit to the one process queue without +exporting a C++ object or relying on ELF interposition. The callback accepts a +record only after it owns a queue slot. Binding marks that private logger as a +consumer: it can recognize and use an existing sink, but it cannot create one. +Only the extension copy that owns the process state creates the writer, so a +transient DSO cannot leave its callback or thread behind after `dlclose()`. +`pending_record_count` tracks accepted +work for bounded drains; `dropped_record_count` tracks enqueue rejection and +final write failure. The high bit of `sink_producer_state` closes admission +before a fork boundary; its low bits keep `sink_context` alive until every DSO +caller that may have loaded it has returned. `_host_log_dropped_records()` +exposes the drop counter to diagnostics and tests, while +`_host_log_pending_records()` distinguishes accepted work still waiting for the +writer. Python and C++ flush defaults are both 1000 ms. Teardown and `os._exit()` +paths report a timeout with both counters instead of silently abandoning the +accepted backlog. ### Load and bind order Python seeds the extension-owned state before C++ loads consumers: ```python -_initialize_host_log(level) +_initialize_host_log(level, defer_writer=is_hierarchical) +# Hierarchical workers perform all local forks here. +_start_host_log_writer() self._impl.init(host_path, aicpu_path, aicore_path, dispatcher_path, device_id, prewarm_config, enable_sdma, sim_context_path) +# At submit, after CallConfig is available: +_set_host_log_directory(config.output_prefix) ``` `ChipWorker::init` then performs the module-specific work: @@ -229,28 +287,26 @@ trace/swimlane JSON while leaving event timestamps monotonic and relative. See ### AICPU sim ```text -[DEBUG] func: [file.cpp:line] message -[INFO] func: [file.cpp:line] message -[TIMING] func: [file.cpp:line] message -[WARN] func: [file.cpp:line] message -[ERROR] func: [file.cpp:line] message +[mono_ns=MONOTONIC_NS][T0xTID][LEVEL] func: [file.cpp:line] message ``` -This is still the sim device backend, so it has no host monotonic/tid prefix. -Onboard AICPU uses the CANN dlog format. Device TIMING uses CANN WARN and adds a -`[TIMING]` message tag. +Sim AICPU runs on a host CPU in the host process and uses the same envelope and +destination as other bound host modules. The `dev_vlog_*` names remain as the +compatibility boundary consumed by `unified_log_device.cpp`. Onboard AICPU uses +the CANN dlog format. Device TIMING uses CANN WARN and adds a `[TIMING]` message +tag. ## Configuration flow | Stage | Action | Source | | ----- | ------ | ------ | | Python import | Register `TIMING` / `NUL`; default the `simpler` logger to TIMING | `python/simpler/_log.py` | -| `Worker.init()` | Normalize the Python logger level, seed native state before the first fork, and point the `simpler` logger at the host logger | `python/simpler/worker.py` | +| `Worker.init()` | Normalize the Python level, attach it to the host logger, quiesce an old writer before local forks, then start a new writer after the final fork | `python/simpler/worker.py` | | `ChipWorker.init()` | Re-seed inherited native state in a chip child, then enter C++ | `python/simpler/task_interface.py` | | `_ChipWorker.init()` | Load sim context and host runtime, then bind each module's logger state | `src/common/worker/chip_worker.cpp` | | `simpler_init` | Onboard maps the bound threshold to CANN; attach and take executor binaries | `src/common/platform/{onboard,sim}/host/c_api_shared.cpp` | | Nested host load | Bind generated host orchestration/AICore logger state before entry | runtime maker / sim device runner | -| AICPU init | Snapshot the applicable device threshold | platform AICPU init | +| AICPU init | Sim binds the live host state; onboard snapshots CANN policy | platform AICPU init | The Python level is still sampled during worker initialization. Calling `logger.setLevel(...)` does not itself call the native setter; recreate or @@ -289,11 +345,14 @@ someone gave a custom number. ### Forked chip subprocesses -The hierarchical parent seeds native state before `fork()` and passes the -normalized level explicitly to `_chip_process_loop`. The child re-seeds its -inherited copy before loading runtime modules. This covers both chip-owning L3 -workers and higher-level processes that emit scheduler spans without loading a -chip runtime. +The hierarchical parent seeds native state and joins any prior writer before +`fork()`. It starts its new writer only after the final local child exists and +before remote activation creates other threads. A generic fork child starts a +writer after its fallible setup returns (setup may recursively fork a lower +subtree); a chip child starts one while initializing its `ChipWorker`. Normal +`os._exit()` paths and Worker/ChipWorker teardown perform a bounded drain. This +covers both chip-owning L3 workers and higher-level processes that emit +scheduler spans without loading a chip runtime. ### Onboard AICPU severity is CANN-owned @@ -321,6 +380,8 @@ There is no logger build step or logger field in `RuntimeBinaries`. Instead: - `_task_interface`, all host runtimes, sim-context, and sim AICore targets add `host_log.cpp` and `unified_log_host.cpp` to their source lists. +- Sim AICPU targets add `host_log.cpp` while retaining + `unified_log_device.cpp`; the existing device ABI delegates to HostLogger. - Host-compiled generated orchestration SOs receive the same sources through `KernelCompiler.get_orchestration_cache_inputs`; those sources therefore participate in the scene-test cache key. @@ -336,7 +397,7 @@ There is no logger build step or logger field in `RuntimeBinaries`. Instead: | Change the user-facing level model | `python/simpler/_log.py` and `docs/testing.md` | | Change host output or STRACE grammar | `src/common/log/host_log.cpp` | | Change the shared-state ABI | `src/common/log/include/common/host_log_state.h` | -| Change sim AICPU output | `src/common/platform/sim/aicpu/device_log.cpp` | +| Change sim AICPU adaptation | `src/common/platform/sim/aicpu/device_log.cpp` | | Change onboard CANN tagging | `src/common/platform/onboard/aicpu/device_log.cpp` | | Add a host logging consumer | compile both host logger sources, include `src/common/log/include`, and bind state during module init | | Add a level | `log_level.h`, `_log.py`, `simpler_setup/log_config.py`, and AICPU `set_log_level` | diff --git a/python/bindings/task_interface.cpp b/python/bindings/task_interface.cpp index 0da92d4996..9e37d1b715 100644 --- a/python/bindings/task_interface.cpp +++ b/python/bindings/task_interface.cpp @@ -1736,12 +1736,44 @@ NB_MODULE(_task_interface, m) { ); m.def( "_initialize_host_log", - [](int level) { + [](int level, bool defer_writer) { if (!simpler::log::is_valid_level(level)) return false; - HostLogger::get_instance().set_level(static_cast(level)); - return true; + HostLogger &logger = HostLogger::get_instance(); + if (defer_writer && !logger.prepare_to_fork()) return false; + logger.set_level(static_cast(level), /*defer_writer=*/true); + return defer_writer || logger.start_writer(); + }, + nb::arg("level"), nb::arg("defer_writer") = false, + "Seed the process-owned host-log state. A hierarchical worker defers its writer until after its last fork." + ); + m.def( + "_start_host_log_writer", + [] { + return HostLogger::get_instance().start_writer(); + }, + "Start this process's bounded host-log writer after its final local fork." + ); + m.def( + "_flush_host_log", + [](uint32_t timeout_ms) { + return HostLogger::get_instance().flush(timeout_ms); + }, + nb::arg("timeout_ms") = 1000, nb::call_guard(), + "Wait boundedly for all host-log records accepted by this process to be written." + ); + m.def( + "_host_log_dropped_records", + [] { + return HostLogger::get_instance().dropped_records(); + }, + "Return the number of host-log records rejected by or lost from this process sink." + ); + m.def( + "_host_log_pending_records", + [] { + return HostLogger::get_instance().pending_records(); }, - nb::arg("level"), "Seed the process-owned host-log state before workers fork or load runtime modules." + "Return the number of accepted host-log records not yet written by this process sink." ); m.def( "_set_host_log_directory", diff --git a/python/simpler/task_interface.py b/python/simpler/task_interface.py index 64b41df177..2b037affa7 100644 --- a/python/simpler/task_interface.py +++ b/python/simpler/task_interface.py @@ -26,7 +26,9 @@ from __future__ import annotations +import contextlib import ctypes +import sys import threading import uuid import weakref @@ -76,12 +78,24 @@ from _task_interface import ( _emit_host_log as _native_emit_host_log, ) +from _task_interface import ( + _flush_host_log as _native_flush_host_log, +) from _task_interface import ( _host_log_directory as _native_host_log_directory, ) +from _task_interface import ( + _host_log_dropped_records as _native_host_log_dropped_records, +) +from _task_interface import ( + _host_log_pending_records as _native_host_log_pending_records, +) from _task_interface import ( _initialize_host_log as _native_initialize_host_log, ) +from _task_interface import ( + _start_host_log_writer as _native_start_host_log_writer, +) from .buffer import Buffer, Tensor @@ -1240,8 +1254,8 @@ def committed(self) -> bool: return self._committed -def _initialize_host_log(log_level: int | None = None) -> None: - """Seed the extension-owned host-log state before runtime use or fork. +def _initialize_host_log(log_level: int | None = None, *, defer_writer: bool = False) -> None: + """Seed host-log state, optionally leaving its writer stopped for local forks. Also points the Python `simpler` logger at that same host logger, so the two stop being separate logging systems that agree only on a threshold. This is @@ -1253,11 +1267,63 @@ def _initialize_host_log(log_level: int | None = None) -> None: if log_level is None: log_level = _log.get_current_config() - if not _native_initialize_host_log(int(log_level)): + if int(log_level) not in (10, 20, 25, 30, 40, 60): raise ValueError(f"unsupported simpler log threshold: {log_level}") + if not _native_initialize_host_log(int(log_level), bool(defer_writer)): + raise RuntimeError(f"cannot initialize simpler host logging at threshold {log_level}") _log.attach_unified_log_handler(_native_emit_host_log, _native_host_log_directory) +def _start_host_log_writer() -> None: + """Start the process-owned writer after the process's final local fork.""" + if not _native_start_host_log_writer(): + raise RuntimeError("cannot start simpler host-log writer") + + +def _flush_host_log(timeout_ms: int = 1000) -> bool: + """Wait boundedly for this process's accepted host-log records.""" + return bool(_native_flush_host_log(int(timeout_ms))) + + +def _host_log_dropped_records() -> int: + """Return the process-owned sink's explicit loss counter.""" + return int(_native_host_log_dropped_records()) + + +def _host_log_pending_records() -> int: + """Return accepted records that the process writer has not completed.""" + return int(_native_host_log_pending_records()) + + +def _flush_host_log_or_warn(context: str, timeout_ms: int = 1000) -> bool: + """Flush boundedly and make a timeout or logger failure observable.""" + try: + flushed = _flush_host_log(timeout_ms) + except BaseException as flush_error: # noqa: BLE001 + # The native logger is the failed component, so stderr is the only + # non-recursive diagnostic path left during teardown. + with contextlib.suppress(BaseException): + sys.stderr.write(f"WARNING: host-log flush failed during {context}: {flush_error}\n") + return False + if flushed: + return True + + try: + pending: int | str = _host_log_pending_records() + except BaseException: # noqa: BLE001 + pending = "unknown" + try: + dropped: int | str = _host_log_dropped_records() + except BaseException: # noqa: BLE001 + dropped = "unknown" + with contextlib.suppress(BaseException): + sys.stderr.write( + f"WARNING: host-log flush timed out after {timeout_ms} ms during {context}; " + f"pending_records={pending}, dropped_records={dropped}; accepted records may be lost.\n" + ) + return False + + class ChipWorker: """Unified execution interface wrapping the host runtime C API. @@ -1377,6 +1443,7 @@ def finalize(self): try: self._impl.finalize() finally: + _flush_host_log_or_warn("ChipWorker.finalize()") with self._registry_lock: self._callable_registry.clear() self._identity_registry.clear() diff --git a/python/simpler/worker.py b/python/simpler/worker.py index 8a37be73c8..2b211b4d3b 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -83,7 +83,7 @@ def my_l4_orch(orch, args, config): from dataclasses import dataclass, field, replace from multiprocessing import resource_tracker from multiprocessing.shared_memory import SharedMemory -from typing import Any, cast +from typing import Any, NoReturn, cast import cloudpickle from _task_interface import ( # pyright: ignore[reportMissingImports] @@ -267,7 +267,9 @@ def my_l4_orch(orch, args, config): RemoteBufferExport, RemoteBufferHandle, TaskArgs, + _flush_host_log_or_warn, _initialize_host_log, + _start_host_log_writer, _Worker, ) from .worker_chip_orch_comm import ( @@ -4274,6 +4276,12 @@ def _wait_for_acceptance(self) -> None: raise +def _exit_after_host_log_flush(status: int) -> NoReturn: + """Boundedly preserve accepted records before a fork child uses os._exit().""" + _flush_host_log_or_warn("fork-child os._exit()") + os._exit(status) + + def _forked_child_main(buf: memoryview, label: str, setup, serve, make_group_leader: bool = False) -> None: """Run a forked child to completion, always terminating via ``os._exit``. @@ -4320,6 +4328,9 @@ def _on_cancel(_signum, _frame): try: try: ctx = setup() + # setup may recursively fork a complete lower-level subtree. + # Starting here keeps every C++ writer behind the final fork. + _start_host_log_writer() finally: setup_active = False except _StartupCancelled: @@ -4343,7 +4354,7 @@ def _on_cancel(_signum, _frame): else: exit_code = 0 finally: - os._exit(exit_code) + _exit_after_host_log_flush(exit_code) # --------------------------------------------------------------------------- @@ -7580,6 +7591,20 @@ def init( # noqa: PLR0912, PLR0915 try: self._cleanup_partial_init() finally: + # _start_hierarchical() quiesces the process-owned log writer + # before its first fork. If anything fails before the normal + # post-fork restart, restore that process-global service after + # rollback; otherwise this failed Worker silently disables logs + # from unrelated Workers and callers in the same parent. + if self.level >= 3: + try: + _start_host_log_writer() + except BaseException as log_restore_error: # noqa: BLE001 -- preserve the startup cause + with contextlib.suppress(BaseException): + sys.stderr.write( + f"[worker pid={os.getpid()}] WARN: failed to restore host-log writer after " + f"startup rollback: {log_restore_error}\n" + ) with self._hierarchical_start_cv: # Only an INITIALIZING epoch commits FAILED. FAILED is only # written by the init thread. CLOSED is absorbing. @@ -7782,7 +7807,7 @@ def _start_hierarchical(self) -> None: # noqa: PLR0912 -- three parallel fork l # and emits their spans. A chip child re-seeds its inherited state before # binding the logger copies embedded in the runtime modules it loads. chip_log_level = _simpler_log.get_current_config() - _initialize_host_log(chip_log_level) + _initialize_host_log(chip_log_level, defer_writer=True) # Bind the level word this process's host-scheduler spans lead with. The # C++ emit sites in Orchestrator / WorkerThread are level-agnostic — the @@ -7883,8 +7908,8 @@ def _setup(): if _mailbox_load_i32(_buffer_field_addr(buf, _OFF_STATE)) == _IDLE: _write_error(buf, 1, _format_exc(f"chip worker {idx} dev={dev_id} init", e)) _mailbox_store_i32(_buffer_field_addr(buf, _OFF_STATE), _INIT_FAILED) - os._exit(1) - os._exit(0) + _exit_after_host_log_flush(1) + _exit_after_host_log_flush(0) else: self._chip_pids.append(pid) if self._is_startup_root: @@ -7968,6 +7993,11 @@ def _setup(inner=inner_worker): # failure, exit, or hang aborts startup here. self._await_children_ready(self._next_level_shms, self._next_level_pids, "next_level", deadline) + # No local fork may follow this point. Only now is it safe to create the + # process-owned C++ writer thread; remote activation below creates its + # own health threads too. + _start_host_log_writer() + # Last local fork is done. Now — and only now — open and register remote # L3 sessions: opening starts the remote subtree and registering spawns # the RemoteL3Endpoint health thread, so both must follow every local @@ -11667,6 +11697,11 @@ def close(self) -> None: # noqa: PLR0912, PLR0915 -- lifecycle linearization: r if result is None: result = exc finally: + # CLOSED prevents new admissions and teardown has quiesced + # this Worker's producers. Preserve accepted records before + # publishing completion, but never turn a stuck output into + # an unbounded wait or a new close failure. + _flush_host_log_or_warn("Worker.close()") # The immutable outcome reference is the completion flag and # result. A reader can therefore never observe completion # without its error/incomplete payload. Publication precedes diff --git a/src/a2a3/platform/sim/aicpu/CMakeLists.txt b/src/a2a3/platform/sim/aicpu/CMakeLists.txt index ecb751f7b4..9917da9ab3 100644 --- a/src/a2a3/platform/sim/aicpu/CMakeLists.txt +++ b/src/a2a3/platform/sim/aicpu/CMakeLists.txt @@ -56,6 +56,7 @@ file(GLOB COMMON_SOURCES CONFIGURE_DEPENDS "${CMAKE_CURRENT_SOURCE_DIR}/../../../../common/platform/sim/aicpu/*.cpp" ) list(APPEND AICPU_SOURCES ${COMMON_SOURCES}) +list(APPEND AICPU_SOURCES "${CMAKE_CURRENT_SOURCE_DIR}/../../../../common/log/host_log.cpp") # Then, collect sources from CUSTOM_SOURCE_DIRS (runtime sources) if(DEFINED CUSTOM_SOURCE_DIRS) diff --git a/src/a2a3/platform/sim/host/device_runner.cpp b/src/a2a3/platform/sim/host/device_runner.cpp index a559d365ee..1db6504c5c 100644 --- a/src/a2a3/platform/sim/host/device_runner.cpp +++ b/src/a2a3/platform/sim/host/device_runner.cpp @@ -204,17 +204,20 @@ int DeviceRunner::ensure_binaries_loaded() { if (!load_sym("set_platform_scope_stats_base", reinterpret_cast(&set_platform_scope_stats_base_func_))) return PTO_RUNTIME_ERR_INTERNAL; - // The AICPU sim SO owns its level flags because it is RTLD_LOCAL. - // Forward the process-wide HostLogger threshold explicitly. + // The AICPU sim SO binds its private HostLogger before the compatibility + // level setter can emit a clock anchor. using SetLogLevelFunc = void (*)(int); SetLogLevelFunc set_log_level_func = nullptr; if (!load_sym("set_log_level", reinterpret_cast(&set_log_level_func))) return PTO_RUNTIME_ERR_INTERNAL; - set_log_level_func(HostLogger::get_instance().level()); - using SetHostLogStateFunc = void (*)(SimplerHostLogState *); + using SetHostLogStateFunc = int (*)(SimplerHostLogState *); SetHostLogStateFunc set_host_log_state_func = nullptr; if (!load_sym("set_host_log_state", reinterpret_cast(&set_host_log_state_func))) return PTO_RUNTIME_ERR_INTERNAL; - set_host_log_state_func(HostLogger::get_instance().state()); + if (set_host_log_state_func(HostLogger::get_instance().state()) != 0) { + LOG_ERROR("AICPU SO rejected the host-log state ABI"); + return PTO_RUNTIME_ERR_INTERNAL; + } + set_log_level_func(HostLogger::get_instance().level()); aicpu_so_loaded_ = true; LOG_INFO("DeviceRunner(sim): Loaded aicpu_execute from %s", aicpu_so_path_.c_str()); diff --git a/src/a5/platform/sim/aicpu/CMakeLists.txt b/src/a5/platform/sim/aicpu/CMakeLists.txt index 0c8ee16740..1137e74024 100644 --- a/src/a5/platform/sim/aicpu/CMakeLists.txt +++ b/src/a5/platform/sim/aicpu/CMakeLists.txt @@ -54,6 +54,7 @@ file(GLOB COMMON_SOURCES CONFIGURE_DEPENDS "${CMAKE_CURRENT_SOURCE_DIR}/../../../../common/platform/sim/aicpu/*.cpp" ) list(APPEND AICPU_SOURCES ${COMMON_SOURCES}) +list(APPEND AICPU_SOURCES "${CMAKE_CURRENT_SOURCE_DIR}/../../../../common/log/host_log.cpp") # Then, collect sources from CUSTOM_SOURCE_DIRS (runtime sources) if(DEFINED CUSTOM_SOURCE_DIRS) diff --git a/src/a5/platform/sim/host/device_runner.cpp b/src/a5/platform/sim/host/device_runner.cpp index 34c5fd901c..0ff773dbc9 100644 --- a/src/a5/platform/sim/host/device_runner.cpp +++ b/src/a5/platform/sim/host/device_runner.cpp @@ -190,17 +190,20 @@ int DeviceRunner::ensure_binaries_loaded() { if (!load_sym("set_platform_scope_stats_base", reinterpret_cast(&set_platform_scope_stats_base_func_))) return PTO_RUNTIME_ERR_INTERNAL; - // The AICPU sim SO owns its level flags because it is RTLD_LOCAL. - // Forward the process-wide HostLogger threshold explicitly. + // The AICPU sim SO binds its private HostLogger before the compatibility + // level setter can emit a clock anchor. using SetLogLevelFunc = void (*)(int); SetLogLevelFunc set_log_level_func = nullptr; if (!load_sym("set_log_level", reinterpret_cast(&set_log_level_func))) return PTO_RUNTIME_ERR_INTERNAL; - set_log_level_func(HostLogger::get_instance().level()); - using SetHostLogStateFunc = void (*)(SimplerHostLogState *); + using SetHostLogStateFunc = int (*)(SimplerHostLogState *); SetHostLogStateFunc set_host_log_state_func = nullptr; if (!load_sym("set_host_log_state", reinterpret_cast(&set_host_log_state_func))) return PTO_RUNTIME_ERR_INTERNAL; - set_host_log_state_func(HostLogger::get_instance().state()); + if (set_host_log_state_func(HostLogger::get_instance().state()) != 0) { + LOG_ERROR("AICPU SO rejected the host-log state ABI"); + return PTO_RUNTIME_ERR_INTERNAL; + } + set_log_level_func(HostLogger::get_instance().level()); aicpu_so_loaded_ = true; LOG_INFO("DeviceRunner(sim): Loaded aicpu_execute from %s", aicpu_so_path_.c_str()); diff --git a/src/common/log/host_log.cpp b/src/common/log/host_log.cpp index e1dd88c82a..a34e527816 100644 --- a/src/common/log/host_log.cpp +++ b/src/common/log/host_log.cpp @@ -15,16 +15,23 @@ #include "host_log.h" +#include +#include #include #include #include +#include #include #include #include +#include #include +#include +#include #include +#include #include -#include +#include #include @@ -39,10 +46,23 @@ using simpler::log::LogLevel; namespace { -// 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; +// Every queued record fits the portable atomic pipe-write floor. The writer +// therefore preserves one physical write per record when it falls back to a +// stderr pipe shared by several processes. +constexpr size_t kRecordCapacity = _POSIX_PIPE_BUF; +// About 2 MiB per logging process: large enough to absorb ordinary bursts, +// fixed enough that a permanently blocked destination cannot grow memory use. +constexpr size_t kQueueCapacity = 4096; +// Producers receive a fixed CPU budget for the lock-free MPSC position claim. +// Exhausting it is a counted drop, never an I/O wait or condition-variable sleep. +constexpr size_t kProducerClaimAttempts = 1024; +constexpr uint64_t kProducerStopFlag = UINT64_C(1) << 63; +static_assert(kQueueCapacity + 1 <= _POSIX_SEM_VALUE_MAX); + +#if defined(SIMPLER_HOST_LOG_TEST_HOOKS) +std::atomic g_after_queue_claim_hook{nullptr}; +std::atomic g_before_gap_wait_hook{nullptr}; +#endif // POSIX guarantees atomic pipe writes up to _POSIX_PIPE_BUF (512 bytes). A // conservative bound for the logger prefix, fixed-width STRACE fields, and @@ -50,6 +70,18 @@ constexpr size_t kRecordStackCapacity = 2048; constexpr size_t kHostSpanNameCapacity = 64; constexpr size_t kHostSpanAttributesCapacity = 192; static_assert(kHostSpanNameCapacity + kHostSpanAttributesCapacity <= _POSIX_PIPE_BUF - 256); +static_assert(kRecordCapacity >= 2); + +struct QueuedRecord { + uint32_t size; + int32_t anchor_pid; + char data[kRecordCapacity]; +}; + +struct QueueSlot { + std::atomic sequence; + QueuedRecord record; +}; std::string encode_host_span_field(const char *value, size_t capacity, bool attributes) { static constexpr char kHex[] = "0123456789ABCDEF"; @@ -89,8 +121,8 @@ std::string encode_host_span_field(const char *value, size_t capacity, bool attr // 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 -// truncated prefix and the caller must re-render into that many bytes. +// record. A return value of `capacity` or more means `buffer` holds a truncated +// record and the caller must replace its tail with the truncation marker. size_t format_record( char *buffer, size_t capacity, int64_t monotonic_ns, unsigned long tid, const char *level_tag, const char *func, const char *fmt, va_list args, bool append_newline @@ -130,10 +162,10 @@ size_t format_record( return length; } -bool write_stderr(const char *record, size_t size) { +bool write_fd(int fd, const char *record, size_t size) { size_t offset = 0; while (offset < size) { - const ssize_t written = ::write(STDERR_FILENO, record + offset, size - offset); + const ssize_t written = ::write(fd, record + offset, size - offset); if (written > 0) { offset += static_cast(written); } else if (written < 0 && errno == EINTR) { @@ -145,6 +177,24 @@ bool write_stderr(const char *record, size_t size) { return true; } +bool write_stderr(const char *record, size_t size) { return write_fd(STDERR_FILENO, record, size); } + +uint64_t atomic_load_u64(const uint64_t *value) { return __atomic_load_n(value, __ATOMIC_ACQUIRE); } + +void atomic_add_u64(uint64_t *value, uint64_t count) { __atomic_fetch_add(value, count, __ATOMIC_RELAXED); } + +void atomic_sub_u64(uint64_t *value, uint64_t count) { __atomic_fetch_sub(value, count, __ATOMIC_RELEASE); } + +void atomic_store_u64(uint64_t *value, uint64_t desired) { __atomic_store_n(value, desired, __ATOMIC_RELEASE); } + +void release_anchor_after_write_failure(SimplerHostLogState *state, int32_t pid) { + int32_t observed = __atomic_load_n(&state->clock_anchor_pid, __ATOMIC_ACQUIRE); + while ( + (observed == pid || observed == -pid) && + !__atomic_compare_exchange_n(&state->clock_anchor_pid, &observed, 0, false, __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE) + ) {} +} + long host_trace_tid() { #if defined(__linux__) && defined(SYS_gettid) return static_cast(syscall(SYS_gettid)); @@ -157,14 +207,16 @@ struct HostLogFileSink { HostLogFileSink(); std::mutex mutex; - FILE *stream = nullptr; + int fd = -1; pid_t pid = -1; std::string directory; }; HostLogFileSink &host_log_file_sink() { - static HostLogFileSink sink; - return sink; + // Process-lifetime allocation avoids static-destruction order racing the + // writer thread. The kernel closes the fd when the process exits. + static auto *sink = new HostLogFileSink; + return *sink; } void host_log_sink_before_fork() { host_log_file_sink().mutex.lock(); } @@ -178,65 +230,247 @@ HostLogFileSink::HostLogFileSink() { (void)pthread_atfork(host_log_sink_before_fork, host_log_sink_after_fork, host_log_sink_after_fork); } -// Append one already-formatted record. The <=`PIPE_BUF` single-write rule that -// makes a stderr record indivisible neither applies nor is needed here: this file -// has exactly one writer process and the sink mutex serializes the writers inside -// it, so a flush of up to the buffer's size cannot interleave with anything. -// -// Two properties a system tracer would have and this deliberately does not, so -// the difference is not mistaken for an oversight. The flush runs on the thread -// that emitted the record, where ftrace and Perfetto hand the bytes to a consumer -// and never let a producer touch the output; and a full buffer here blocks that -// thread rather than dropping and counting, where every comparable tracer bounds -// the buffer and exports a loss counter. What this buys is one write per root -// span instead of one per record — an order of magnitude, not the elimination of -// the observer effect. -bool write_log_file(const char *directory, const char *record, size_t size, bool flush) { +// Blocking I/O is confined to the process writer thread. A raw append fd makes +// completion and loss accounting exact: successful write(2) means the whole +// record reached the kernel, with no hidden stdio buffer left to fail later. +bool write_log_file(const char *directory, const char *record, size_t size) { HostLogFileSink &sink = host_log_file_sink(); std::scoped_lock lock(sink.mutex); const pid_t pid = getpid(); - const bool inherited = sink.stream != nullptr && sink.pid != pid; - const bool changed_directory = sink.stream != nullptr && sink.directory != directory; + const bool inherited = sink.fd >= 0 && sink.pid != pid; + const bool changed_directory = sink.fd >= 0 && sink.directory != directory; if (inherited) { - // Do not fclose(): its copied stdio buffer contains records already - // owned by the parent and must never be flushed again by the child. - // Dropping the FILE leaks it and its buffer in the child, one per - // process, which is the price of not duplicating the parent's records — - // C offers no portable way to discard a stream's buffer. - const int inherited_fd = fileno(sink.stream); - if (inherited_fd >= 0) (void)::close(inherited_fd); - sink.stream = nullptr; + (void)::close(sink.fd); + sink.fd = -1; sink.pid = -1; sink.directory.clear(); } else if (changed_directory) { - std::fclose(sink.stream); - sink.stream = nullptr; + (void)::close(sink.fd); + sink.fd = -1; sink.pid = -1; sink.directory.clear(); } - if (sink.stream == nullptr) { + if (sink.fd < 0) { char path[PATH_MAX]; const int length = std::snprintf(path, sizeof(path), "%s/host.%d.log", directory, pid); if (length <= 0 || static_cast(length) >= sizeof(path)) return false; - sink.stream = std::fopen(path, "a"); - if (sink.stream == nullptr) return false; - std::setvbuf(sink.stream, nullptr, _IOFBF, 1U << 20U); + int flags = O_WRONLY | O_CREAT | O_APPEND; +#ifdef O_CLOEXEC + flags |= O_CLOEXEC; +#endif + sink.fd = ::open(path, flags, 0666); + if (sink.fd < 0) return false; sink.pid = pid; sink.directory = directory; } - if (std::fwrite(record, 1, size, sink.stream) != size) return false; - return !flush || std::fflush(sink.stream) == 0; + if (write_fd(sink.fd, record, size)) return true; + (void)::close(sink.fd); + sink.fd = -1; + sink.pid = -1; + sink.directory.clear(); + return false; } -bool flush_log_file() { - HostLogFileSink &sink = host_log_file_sink(); - std::scoped_lock lock(sink.mutex); - return sink.stream == nullptr || std::fflush(sink.stream) == 0; +const char *bound_log_directory(const SimplerHostLogState *state) { + if (__atomic_load_n(&state->log_directory_bound, __ATOMIC_ACQUIRE) != 1) return nullptr; + return state->log_directory; } -} // namespace +bool write_record_now(const SimplerHostLogState *state, const char *record, size_t size) { + const char *directory = bound_log_directory(state); + if (directory != nullptr && write_log_file(directory, record, size)) return true; + return write_stderr(record, size); +} -namespace { +struct HostLogAsyncSink { + explicit HostLogAsyncSink(SimplerHostLogState *shared_state) : + state(shared_state), + pid(getpid()) { + for (size_t index = 0; index < kQueueCapacity; ++index) { + queue[index].sequence.store(index, std::memory_order_relaxed); + } + // macOS deliberately does not implement unnamed semaphores. A named + // semaphore works on both supported host OSes; unlink it immediately so + // the kernel object has this process's handles as its only lifetime. + char name[32]; + const int length = snprintf( + name, sizeof(name), "/sl-%x-%llx", static_cast(pid), + static_cast(reinterpret_cast(this)) + ); + if (length > 0 && static_cast(length) < sizeof(name)) { + ready = sem_open(name, O_CREAT | O_EXCL, 0600, 0); + if (ready != SEM_FAILED && sem_unlink(name) != 0) { + (void)sem_close(ready); + ready = SEM_FAILED; + } + } + } + + ~HostLogAsyncSink() { + if (ready != SEM_FAILED) (void)sem_close(ready); + } + + bool start() { + if (ready == SEM_FAILED) return false; + try { + writer = std::thread([this] { + run(); + }); + return true; + } catch (...) { + return false; + } + } + + bool stop_when_empty(uint32_t timeout_ms) { + if (!wait_until_empty(timeout_ms)) return false; + + stopping.store(true, std::memory_order_release); + (void)sem_post(ready); + if (writer.joinable()) writer.join(); + return true; + } + + bool wait_until_empty(uint32_t timeout_ms) { + std::unique_lock lock(completion_mutex); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); + for (;;) { + if ((atomic_load_u64(&state->sink_producer_state) & ~kProducerStopFlag) == 0 && + atomic_load_u64(&state->pending_record_count) == 0) { + return true; + } + const auto now = std::chrono::steady_clock::now(); + if (now >= deadline) return false; + completion_cv.wait_until(lock, std::min(deadline, now + std::chrono::milliseconds(1))); + } + } + + SimplerHostLogState *state; + pid_t pid; + + static int + enqueue(void *context, SimplerHostLogState *state, const char *record, uint32_t size, int32_t anchor_pid) { + auto *sink = static_cast(context); + if (sink == nullptr) { + atomic_sub_u64(&state->sink_producer_state, 1); + return 0; + } + if (sink->state != state || sink->pid != getpid() || size > kRecordCapacity) { + sink->producer_done(); + return 0; + } + + size_t position = sink->enqueue_position.load(std::memory_order_relaxed); + QueueSlot *slot = nullptr; + for (size_t attempt = 0; attempt < kProducerClaimAttempts; ++attempt) { + slot = &sink->queue[position % kQueueCapacity]; + const size_t sequence = slot->sequence.load(std::memory_order_acquire); + const auto difference = static_cast(sequence) - static_cast(position); + if (difference == 0) { + if (sink->enqueue_position.compare_exchange_weak( + position, position + 1, std::memory_order_relaxed, std::memory_order_relaxed + )) { + break; + } + } else if (difference < 0) { + sink->producer_done(); + return 0; // The consumer has not freed this lap: queue full. + } else { + position = sink->enqueue_position.load(std::memory_order_relaxed); + } + slot = nullptr; + } + if (slot == nullptr) { + sink->producer_done(); + return 0; + } + +#if defined(SIMPLER_HOST_LOG_TEST_HOOKS) + if (auto hook = g_after_queue_claim_hook.load(std::memory_order_acquire); hook != nullptr) hook(position); +#endif + + slot->record.size = size; + slot->record.anchor_pid = anchor_pid; + std::memcpy(slot->record.data, record, size); + atomic_add_u64(&state->pending_record_count, 1); + slot->sequence.store(position + 1, std::memory_order_release); + sink->queue_size.fetch_add(1, std::memory_order_release); + (void)sem_post(sink->ready); + sink->producer_done(); + return 1; + } + +private: + void producer_done() { + // Producers never take the drain waiter's mutex. A waiter polls the + // atomic producer count at a bounded interval, so it cannot miss this + // transition permanently even though no condition-variable signal is + // needed on this hot path. + atomic_sub_u64(&state->sink_producer_state, 1); + } + + bool pop(QueuedRecord *record) { + if (queue_size.load(std::memory_order_acquire) == 0) return false; + QueueSlot &slot = queue[dequeue_position % kQueueCapacity]; + if (slot.sequence.load(std::memory_order_acquire) != dequeue_position + 1) return false; + *record = slot.record; + slot.sequence.store(dequeue_position + kQueueCapacity, std::memory_order_release); + ++dequeue_position; + queue_size.fetch_sub(1, std::memory_order_release); + return true; + } + + void run() { + size_t ready_tokens = 0; + for (;;) { + if (ready_tokens == 0) { + int result; + do { + result = sem_wait(ready); + } while (result != 0 && errno == EINTR); + if (result != 0) return; + if (stopping.load(std::memory_order_acquire) && queue_size.load(std::memory_order_acquire) == 0) return; + ++ready_tokens; + } + + QueuedRecord record; + // Producers can publish adjacent MPSC positions out of order. A + // later token may wake us before the next position is visible, so + // retain every later token and sleep for the earlier producer's + // token instead of burning a CPU in an unbounded yield loop. + if (!pop(&record)) { +#if defined(SIMPLER_HOST_LOG_TEST_HOOKS) + if (auto hook = g_before_gap_wait_hook.load(std::memory_order_acquire); hook != nullptr) hook(); +#endif + int result; + do { + result = sem_wait(ready); + } while (result != 0 && errno == EINTR); + if (result != 0) return; + ++ready_tokens; + continue; + } + --ready_tokens; + + if (!write_record_now(state, record.data, record.size)) { + atomic_add_u64(&state->dropped_record_count, 1); + if (record.anchor_pid != 0) release_anchor_after_write_failure(state, record.anchor_pid); + } + atomic_sub_u64(&state->pending_record_count, 1); + completion_cv.notify_all(); + } + } + + std::array queue; + std::atomic queue_size{0}; + std::atomic enqueue_position{0}; + size_t dequeue_position = 0; + sem_t *ready = SEM_FAILED; + std::atomic stopping{false}; + std::mutex completion_mutex; + std::condition_variable completion_cv; + std::thread writer; +}; // A private logger stays silent until its owner seeds this state or its loader // binds the process-owned state. Missing binding is therefore observable as an @@ -246,6 +480,15 @@ SimplerHostLogState g_module_log_state{ sizeof(SimplerHostLogState), static_cast(LogLevel::NUL), 0, + 0, + {}, + 0, + 0, + nullptr, + nullptr, + 0, + 0, + 0, }; int32_t atomic_load_i32(const int32_t *value) { return __atomic_load_n(value, __ATOMIC_ACQUIRE); } @@ -256,6 +499,34 @@ bool atomic_compare_exchange_i32(int32_t *value, int32_t *expected, int32_t desi return __atomic_compare_exchange_n(value, expected, desired, false, __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE); } +void *atomic_load_pointer(void *const *value) { return __atomic_load_n(value, __ATOMIC_ACQUIRE); } + +void atomic_store_pointer(void **value, void *desired) { __atomic_store_n(value, desired, __ATOMIC_RELEASE); } + +SimplerHostLogEnqueueFn atomic_load_enqueue(SimplerHostLogEnqueueFn const *value) { + return __atomic_load_n(value, __ATOMIC_ACQUIRE); +} + +void atomic_store_enqueue(SimplerHostLogEnqueueFn *value, SimplerHostLogEnqueueFn desired) { + __atomic_store_n(value, desired, __ATOMIC_RELEASE); +} + +bool acquire_sink_producer(SimplerHostLogState *state) { + // Admission and stop are ordered by one atomic RMW. If stop won first, undo + // our transient producer reference and reject the record. If admission won + // first, prepare_to_fork() observes the reference and keeps sink_context + // alive until this producer returns. Unlike a CAS retry loop, this gives + // every producer a fixed admission cost under contention. + const uint64_t previous = __atomic_fetch_add(&state->sink_producer_state, 1, __ATOMIC_ACQ_REL); + if ((previous & kProducerStopFlag) != 0) { + atomic_sub_u64(&state->sink_producer_state, 1); + return false; + } + return true; +} + +void release_sink_producer(SimplerHostLogState *state) { atomic_sub_u64(&state->sink_producer_state, 1); } + } // namespace HostLogger &HostLogger::get_instance() { @@ -266,23 +537,140 @@ HostLogger &HostLogger::get_instance() { HostLogger::HostLogger() : state_(&g_module_log_state) {} +HostLogger::~HostLogger() { + // Normal executable/module teardown gets a bounded chance to preserve + // accepted records. os._exit() call sites flush explicitly because static + // destructors do not run there. + (void)prepare_to_fork(100); +} + int HostLogger::bind_state(SimplerHostLogState *state) { if (state == nullptr || state->abi_version != SIMPLER_HOST_LOG_STATE_ABI_VERSION || state->struct_size < sizeof(SimplerHostLogState) || !simpler::log::is_valid_level(atomic_load_i32(&state->threshold))) { return -1; } + state_owner_.store(false, std::memory_order_release); state_.store(state, std::memory_order_release); return 0; } +int HostLogger::adopt_state(SimplerHostLogState *state) { + if (state == nullptr || state->abi_version != SIMPLER_HOST_LOG_STATE_ABI_VERSION || + state->struct_size < sizeof(SimplerHostLogState) || + !simpler::log::is_valid_level(atomic_load_i32(&state->threshold))) { + return -1; + } + state_.store(state, std::memory_order_release); + state_owner_.store(true, std::memory_order_release); + return 0; +} + SimplerHostLogState *HostLogger::state() const { return state_.load(std::memory_order_acquire); } -void HostLogger::set_level(LogLevel level) { +void HostLogger::set_level(LogLevel level, bool defer_writer) { atomic_store_i32(&state()->threshold, static_cast(level)); + if (!defer_writer) (void)start_writer(); +} + +bool HostLogger::start_writer() { + SimplerHostLogState *shared = state(); + const int32_t pid = static_cast(getpid()); + int32_t owner = atomic_load_i32(&shared->sink_owner_pid); + const auto enqueue = atomic_load_enqueue(&shared->sink_enqueue); + void *context = atomic_load_pointer(&shared->sink_context); + if (owner == pid && enqueue != nullptr && context != nullptr) { + emit_clock_anchor_if_needed(); + return true; + } + + // A bound DSO may submit to an already-published process sink, but only the + // module that owns the process state may create that sink. Otherwise the + // callback and writer thread could outlive a transient dlopen() consumer. + if (!state_owner_.load(std::memory_order_acquire)) return false; + + // Another caller is already constructing this process's sink. Treat that + // as a failed duplicate start instead of letting -pid claim itself. + if (owner == -pid) return false; + if (!atomic_compare_exchange_i32(&shared->sink_owner_pid, &owner, -pid)) return false; + __atomic_fetch_or(&shared->sink_producer_state, kProducerStopFlag, __ATOMIC_ACQ_REL); + + const int32_t process_pid = atomic_load_i32(&shared->sink_process_pid); + if (process_pid != 0 && process_pid != pid) { + // A fork child inherits the parent's counters and callback addresses, + // but none of its threads or accepted records. Its new sink starts clean, + // even when the parent had already quiesced to owner=0 before fork. + atomic_store_u64(&shared->dropped_record_count, 0); + atomic_store_u64(&shared->pending_record_count, 0); + atomic_store_u64(&shared->sink_producer_state, kProducerStopFlag); + } + atomic_store_i32(&shared->sink_process_pid, pid); + + auto *candidate = new (std::nothrow) HostLogAsyncSink(shared); + if (candidate == nullptr || !candidate->start()) { + delete candidate; + atomic_store_enqueue(&shared->sink_enqueue, nullptr); + atomic_store_pointer(&shared->sink_context, nullptr); + int32_t claim = -pid; + (void)atomic_compare_exchange_i32(&shared->sink_owner_pid, &claim, 0); + __atomic_fetch_and(&shared->sink_producer_state, ~kProducerStopFlag, __ATOMIC_RELEASE); + return false; + } + + // The active sink normally has process lifetime. prepare_to_fork() is the + // explicit quiescent boundary that can join and reclaim it before a later + // hierarchical Worker forks in this process. + sink_.store(candidate, std::memory_order_release); + atomic_store_pointer(&shared->sink_context, candidate); + atomic_store_enqueue(&shared->sink_enqueue, &HostLogAsyncSink::enqueue); + atomic_store_i32(&shared->sink_owner_pid, pid); + __atomic_fetch_and(&shared->sink_producer_state, ~kProducerStopFlag, __ATOMIC_RELEASE); emit_clock_anchor_if_needed(); + return true; } +bool HostLogger::prepare_to_fork(uint32_t timeout_ms) { + SimplerHostLogState *shared = state(); + const int32_t pid = static_cast(getpid()); + int32_t owner = atomic_load_i32(&shared->sink_owner_pid); + if (owner == 0) return true; + if (owner == -pid) return false; + if (owner != pid) return true; + + auto *sink = static_cast(sink_.load(std::memory_order_acquire)); + if (sink == nullptr || sink->state != shared || sink->pid != getpid()) return false; + + if (!atomic_compare_exchange_i32(&shared->sink_owner_pid, &owner, -pid)) return false; + __atomic_fetch_or(&shared->sink_producer_state, kProducerStopFlag, __ATOMIC_ACQ_REL); + if (!sink->stop_when_empty(timeout_ms)) { + __atomic_fetch_and(&shared->sink_producer_state, ~kProducerStopFlag, __ATOMIC_RELEASE); + atomic_store_i32(&shared->sink_owner_pid, pid); + return false; + } + + atomic_store_enqueue(&shared->sink_enqueue, nullptr); + atomic_store_pointer(&shared->sink_context, nullptr); + sink_.store(nullptr, std::memory_order_release); + atomic_store_i32(&shared->sink_owner_pid, 0); + delete sink; + return true; +} + +bool HostLogger::flush(uint32_t timeout_ms) { + SimplerHostLogState *shared = state(); + if ((atomic_load_u64(&shared->sink_producer_state) & ~kProducerStopFlag) == 0 && + atomic_load_u64(&shared->pending_record_count) == 0) { + return true; + } + auto *sink = static_cast(sink_.load(std::memory_order_acquire)); + if (sink == nullptr || sink->state != shared || sink->pid != getpid()) return false; + return sink->wait_until_empty(timeout_ms); +} + +uint64_t HostLogger::dropped_records() const { return atomic_load_u64(&state()->dropped_record_count); } + +uint64_t HostLogger::pending_records() const { return atomic_load_u64(&state()->pending_record_count); } + int HostLogger::level() const { return atomic_load_i32(&state()->threshold); } int HostLogger::cann_level() const { return simpler::log::to_cann_log_level(static_cast(level())); } @@ -317,48 +705,65 @@ const char *HostLogger::level_name(LogLevel level) const { return "?"; } -bool HostLogger::emit(const char *level_tag, const char *func, const char *fmt, va_list args, bool flush) { +bool HostLogger::emit(const char *level_tag, const char *func, const char *fmt, va_list args, int32_t anchor_pid) { const int64_t monotonic_ns = simpler::log::monotonic_now_ns(); auto tid = static_cast(reinterpret_cast(pthread_self())); const bool append_newline = fmt[0] != '\0' && fmt[strlen(fmt) - 1] != '\n'; - // 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), monotonic_ns, tid, level_tag, func, fmt, args, append_newline - ); - if (length < sizeof(stack_buffer)) { - return write_record(stack_buffer, length, flush); + char record[kRecordCapacity]; + const size_t length = + format_record(record, sizeof(record), monotonic_ns, tid, level_tag, func, fmt, args, append_newline); + size_t size = length; + if (length >= sizeof(record)) { + // A bounded record is part of the producer non-blocking contract. Keep + // a visible truncation marker and a complete physical log line. + record[sizeof(record) - 2] = '~'; + record[sizeof(record) - 1] = '\n'; + size = sizeof(record); } - std::vector heap_buffer(length + 1); - const size_t heap_length = format_record( - heap_buffer.data(), heap_buffer.size(), monotonic_ns, tid, level_tag, func, fmt, args, append_newline - ); - return write_record(heap_buffer.data(), heap_length < heap_buffer.size() ? heap_length : length, flush); -} - -// The one place a destination is chosen. It is a property of this logger, so it -// holds for every record from every caller: nothing about a record's kind, its -// producer, or its level selects a sink here. -bool HostLogger::write_record(const char *record, size_t size, bool flush) { - const char *directory = log_directory(); - if (directory != nullptr && write_log_file(directory, record, size, flush)) return true; - std::scoped_lock lock(mutex_); - return write_stderr(record, size); + SimplerHostLogState *shared = state(); + const int32_t pid = static_cast(getpid()); + const int32_t owner = atomic_load_i32(&shared->sink_owner_pid); + const auto current_enqueue = atomic_load_enqueue(&shared->sink_enqueue); + void *current_context = atomic_load_pointer(&shared->sink_context); + const uint64_t producer_state = atomic_load_u64(&shared->sink_producer_state); + if (owner == 0 && current_enqueue == nullptr && current_context == nullptr && + (producer_state & kProducerStopFlag) == 0) { + // Hierarchical processes deliberately have no writer until their final + // local fork. Preserve initialization records with the synchronous + // path; steady-state producers below remain bounded and do no output I/O. + if (write_record_now(shared, record, size)) return true; + atomic_add_u64(&shared->dropped_record_count, 1); + if (anchor_pid != 0) release_anchor_after_write_failure(shared, anchor_pid); + return false; + } + if (!acquire_sink_producer(shared)) { + atomic_add_u64(&shared->dropped_record_count, 1); + return false; + } + if (atomic_load_i32(&shared->sink_owner_pid) != pid) { + release_sink_producer(shared); + atomic_add_u64(&shared->dropped_record_count, 1); + return false; + } + const auto enqueue = atomic_load_enqueue(&shared->sink_enqueue); + void *context = atomic_load_pointer(&shared->sink_context); + if (enqueue == nullptr || context == nullptr || enqueue(context, shared, record, size, anchor_pid) == 0) { + if (enqueue == nullptr || context == nullptr) release_sink_producer(shared); + atomic_add_u64(&shared->dropped_record_count, 1); + return false; + } + return true; } -bool HostLogger::emit_ungated(const char *level_tag, const char *func, const char *fmt, ...) { +bool HostLogger::emit_ungated(int32_t anchor_pid, const char *level_tag, const char *func, const char *fmt, ...) { va_list args; va_start(args, fmt); // Its one caller writes the clock anchor, which every reader of this stream // needs before it can place anything else in wall time. - const bool written = emit(level_tag, func, fmt, args, /*flush=*/true); + const bool written = emit(level_tag, func, fmt, args, anchor_pid); va_end(args); return written; } @@ -380,7 +785,7 @@ void HostLogger::emit_clock_anchor_if_needed() { std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()) .count(); const bool written = emit_ungated( - level_name(LogLevel::TIMING), "clock_anchor", "[CLOCK_ANCHOR] v=1 pid=%d mono_ns=%lld wall_ns=%lld", + pid_value, level_name(LogLevel::TIMING), "clock_anchor", "[CLOCK_ANCHOR] v=1 pid=%d mono_ns=%lld wall_ns=%lld", static_cast(pid), static_cast(monotonic_ns), static_cast(wall_ns) ); int32_t claim = -pid_value; @@ -392,9 +797,7 @@ void HostLogger::vlog(LogLevel level, const char *func, const char *fmt, va_list return; } emit_clock_anchor_if_needed(); - // A warning or an error is rare and is worth having on disk if the process - // dies; everything below that rides the buffer. - emit(level_name(level), func, fmt, args, /*flush=*/level >= LogLevel::WARN); + (void)emit(level_name(level), func, fmt, args); } void HostLogger::log(LogLevel level, const char *func, const char *fmt, ...) { @@ -437,7 +840,7 @@ void HostLogger::log_host_span(const SimplerHostSpan *span) { // One record grammar, in one place. Where it lands is the logger's business, // not this emitter's. - char record[kRecordStackCapacity]; + char record[kRecordCapacity]; (void)std::snprintf( record, sizeof(record), "[STRACE] v=1 pid=%d tid=%ld inv=%" PRIu64 " hid=%" PRIx64 " depth=%d name=%s ts=%" PRId64 " dur=%" PRId64 @@ -447,12 +850,15 @@ void HostLogger::log_host_span(const SimplerHostSpan *span) { ); log(LogLevel::TIMING, "emit_host_span", "%s", record); - // A closed root span is where the records so far describe a whole - // invocation, which is what makes an in-progress run readable. It is the one - // thing this emitter knows that the writer does not. - if (span->depth == 0) (void)flush_log_file(); } extern "C" __attribute__((visibility("default"))) int simpler_host_log_bind_state(SimplerHostLogState *state) { return HostLogger::get_instance().bind_state(state); } + +#if defined(SIMPLER_HOST_LOG_TEST_HOOKS) +extern "C" void simpler_host_log_set_queue_hooks_for_test(void (*after_claim)(size_t), void (*before_gap_wait)()) { + g_after_queue_claim_hook.store(after_claim, std::memory_order_release); + g_before_gap_wait_hook.store(before_gap_wait, std::memory_order_release); +} +#endif diff --git a/src/common/log/include/common/host_log_state.h b/src/common/log/include/common/host_log_state.h index 464665c9bf..05e4565541 100644 --- a/src/common/log/include/common/host_log_state.h +++ b/src/common/log/include/common/host_log_state.h @@ -13,7 +13,7 @@ #include -#define SIMPLER_HOST_LOG_STATE_ABI_VERSION 2U +#define SIMPLER_HOST_LOG_STATE_ABI_VERSION 3U /* Matches CallConfig::output_prefix, which is where the path comes from. */ #define SIMPLER_HOST_LOG_DIR_CAPACITY 1024 @@ -24,25 +24,41 @@ extern "C" { /* * Process-owned state shared by the private HostLogger copy compiled into - * every host-side DSO. The fields are plain integers and fixed char arrays so - * modules built by different host compiler versions share a C ABI; - * host_log.cpp performs all scalar accesses with compiler atomic builtins. + * every host-side DSO. Fixed-width fields plus a C callback keep the boundary + * independent of the C++ library ABI; host_log.cpp performs all mutable scalar + * accesses with compiler atomic builtins. * * clock_anchor_pid is positive after a successful anchor write and temporarily * negative while one writer owns the claim for that PID. Linux PIDs are * positive and bounded well below INT32_MAX. * - * log_directory is where this logger writes, one fully-buffered file per + * log_directory is where the process writer appends records, one file per * process. It is empty until a caller that knows the run's artifact directory - * supplies it, and every record goes to stderr while it is. The destination is - * a property of the logger, so it applies to every record from every caller — + * supplies it, and the writer uses stderr while it is. The destination is a + * property of the logger, so it applies to every record from every caller — * there is no per-record or per-call-site routing. * * The first non-empty path wins: log_directory_bound is release-stored after the * path is filled and acquire-loaded before it is read, so a reader sees either no * directory or the whole one, and a reader that has already opened the file never * has the path change under it. + * + * sink_owner_pid follows the anchor's claim convention while the process owner + * creates the bounded sink. sink_process_pid distinguishes a parent restarting + * after a quiescent fork boundary from a child that inherited owner=0; the child + * starts fresh counters. The high bit of sink_producer_state closes admission; + * its low bits count callers that may still hold sink_context. Bound private + * logger copies submit complete records through sink_enqueue without exporting + * or interposing another ELF symbol. */ +struct SimplerHostLogState; + +/* Return nonzero only after the complete record has been accepted by the + * process sink. The caller owns drop accounting for a zero return. */ +typedef int (*SimplerHostLogEnqueueFn)( + void *context, struct SimplerHostLogState *state, const char *record, uint32_t size, int32_t anchor_pid +); + typedef struct SimplerHostLogState { uint32_t abi_version; uint32_t struct_size; @@ -50,6 +66,13 @@ typedef struct SimplerHostLogState { int32_t clock_anchor_pid; int32_t log_directory_bound; char log_directory[SIMPLER_HOST_LOG_DIR_CAPACITY]; + int32_t sink_owner_pid; + int32_t sink_process_pid; + void *sink_context; + SimplerHostLogEnqueueFn sink_enqueue; + uint64_t dropped_record_count; + uint64_t pending_record_count; + uint64_t sink_producer_state; } SimplerHostLogState; typedef int (*SimplerHostLogBindStateFn)(SimplerHostLogState *state); diff --git a/src/common/log/include/host_log.h b/src/common/log/include/host_log.h index 8b1706381f..143950d335 100644 --- a/src/common/log/include/host_log.h +++ b/src/common/log/include/host_log.h @@ -21,8 +21,7 @@ #include #include -#include -#include +#include #include @@ -44,6 +43,11 @@ class SIMPLER_HOST_LOG_LOCAL HostLogger { // Bind this module-local logger implementation to process-owned state. // Must happen during module init, before the module starts worker threads. int bind_state(SimplerHostLogState *state); + // Internal owner-side counterpart used when an embedding executable + // supplies storage instead of this module's default state. Cross-DSO + // loaders must use bind_state(), so a transient consumer cannot own the + // process writer across dlclose(). + int adopt_state(SimplerHostLogState *state); SimplerHostLogState *state() const; void log(simpler::log::LogLevel level, const char *func, const char *fmt, ...); @@ -52,7 +56,22 @@ class SIMPLER_HOST_LOG_LOCAL HostLogger { // responsible for `va_start` / `va_end`. void vlog(simpler::log::LogLevel level, const char *func, const char *fmt, va_list args); - void set_level(simpler::log::LogLevel level); + // Owner initialization starts the bounded writer by default. Hierarchical + // workers defer it until their final local fork so no process forks with a + // C++ thread already running. + void set_level(simpler::log::LogLevel level, bool defer_writer = false); + bool start_writer(); + + // A later hierarchical Worker may fork in the same process. Quiesce and + // join an earlier writer before that fork; fail boundedly if its output is + // pinned instead of forking while a C++ thread is alive. + bool prepare_to_fork(uint32_t timeout_ms = 1000); + + // Drain records already accepted by this process owner. Producers must be + // quiescent if the caller needs a strict shutdown boundary. + bool flush(uint32_t timeout_ms = 1000); + uint64_t dropped_records() const; + uint64_t pending_records() const; // Write this process's records to `path`/host..log instead of stderr. // The caller is the one that knows where this run's artifacts go — @@ -83,7 +102,7 @@ class SIMPLER_HOST_LOG_LOCAL HostLogger { private: HostLogger(); - ~HostLogger() = default; + ~HostLogger(); HostLogger(const HostLogger &) = delete; HostLogger &operator=(const HostLogger &) = delete; @@ -91,16 +110,16 @@ class SIMPLER_HOST_LOG_LOCAL HostLogger { HostLogger &operator=(HostLogger &&) = delete; const char *level_name(simpler::log::LogLevel level) const; - bool emit(const char *level_tag, const char *func, const char *fmt, va_list args, bool flush); - // Writes one formatted record wherever this logger is configured to write. - // The only place a destination is chosen, and it is chosen per logger, not - // per record: no caller declares anything and no record kind is special. - bool write_record(const char *record, size_t size, bool flush); - bool emit_ungated(const char *level_tag, const char *func, const char *fmt, ...); + bool emit(const char *level_tag, const char *func, const char *fmt, va_list args, int32_t anchor_pid = 0); + bool emit_ungated(int32_t anchor_pid, const char *level_tag, const char *func, const char *fmt, ...); void emit_clock_anchor_if_needed(); std::atomic state_; - std::mutex mutex_; + std::atomic state_owner_{true}; + // The queue implementation is private to host_log.cpp. Keeping only an + // opaque pointer here prevents its C++ type and inline methods from becoming + // preemptible symbols in every DSO that compiles the host logger. + std::atomic sink_{nullptr}; }; #undef SIMPLER_HOST_LOG_LOCAL diff --git a/src/common/platform/include/aicpu/device_log.h b/src/common/platform/include/aicpu/device_log.h index 37595b88ef..698bf18d11 100644 --- a/src/common/platform/include/aicpu/device_log.h +++ b/src/common/platform/include/aicpu/device_log.h @@ -14,16 +14,16 @@ * * Layered design: * - Low-level dev_log_*() functions are platform-specific (CANN dlog on - * real hardware, fprintf(stderr,...) in simulation). + * real hardware, the process-owned HostLogger in simulation). * - Onboard fills DEBUG/INFO/WARN/ERROR from CheckLogLevel(AICPU,...); - * simulation fills all flags from the host-provided threshold. + * simulation queries the live threshold in its bound HostLogger state. * - TIMING is a simpler level between INFO and WARN. Both backends gate it * from the host threshold; onboard emits enabled messages through CANN * WARN because CANN has no intermediate level. * * Platform Support: * - a2a3 / a5 : Real hardware with CANN dlog API - * - a2a3sim / a5sim : Host-based simulation using fprintf(stderr,...) + * - a2a3sim / a5sim : Host-based simulation using HostLogger */ #pragma once @@ -44,14 +44,15 @@ #endif // ============================================================================= -// Severity enable flags (defined in platform-specific device_log.cpp) +// Platform-specific severity queries. Sim reads a live bound HostLogger state; +// onboard reads the CANN-derived flags cached by its platform backend. // ============================================================================= -extern bool g_is_log_enable_debug; -extern bool g_is_log_enable_info; -extern bool g_is_log_enable_timing; -extern bool g_is_log_enable_warn; -extern bool g_is_log_enable_error; +bool is_log_enable_debug(); +bool is_log_enable_info(); +bool is_log_enable_timing(); +bool is_log_enable_warn(); +bool is_log_enable_error(); // ============================================================================= // Configuration setters (called by AICPU kernel init from KernelArgs) @@ -59,7 +60,7 @@ extern bool g_is_log_enable_error; // Levels use Python-compatible thresholds: DEBUG=10, INFO=20, TIMING=25, // WARN=30, ERROR=40, NUL=60. Onboard applies the threshold to TIMING while -// CANN owns its native levels; simulation applies it to the full flag table. +// CANN owns its native levels; simulation updates its bound HostLogger state. extern "C" void set_log_level(int level); // Hand the process-owned host-log state to the simulation AICPU backend, which @@ -67,8 +68,9 @@ extern "C" void set_log_level(int level); // host-side loader resolves it by name from the AICPU SO handle. Declared here // so both sides agree on the signature at compile time — the struct is only // forward-declared, keeping and the state layout off device targets. +// Returns zero when the ABI is accepted and nonzero otherwise. struct SimplerHostLogState; -extern "C" void set_host_log_state(struct SimplerHostLogState *state); +extern "C" int set_host_log_state(struct SimplerHostLogState *state); // Apply the platform's logging policy to a newly loaded orchestration SO. // Simulation binds the process-owned host state; onboard requires no handoff. @@ -78,11 +80,9 @@ int bind_orchestration_host_log_state(void *handle, const char **error); // Platform-specific logging functions (low-level layer) // // va_list primitives used by the unified_log_* adapter to forward a caller's -// variadic args. Both backends format a whole record into one stack buffer and -// emit it in a single call: sim writes it with one write(2), kept under -// PIPE_BUF so concurrent threads / forked workers on a shared stderr never -// interleave partial records; onboard buffers because CANN's dlog API has no -// va_list variant. Caller owns va_start/va_end. +// variadic args. Sim delegates formatting and destination selection to +// HostLogger; onboard buffers because CANN's dlog API has no va_list variant. +// Caller owns va_start/va_end. // ============================================================================= #include @@ -94,14 +94,8 @@ void dev_vlog_warn(const char *func, const char *fmt, va_list args); void dev_vlog_error(const char *func, const char *fmt, va_list args); // ============================================================================= -// Helper Functions +// Initialization // ============================================================================= -inline bool is_log_enable_debug() { return g_is_log_enable_debug; } -inline bool is_log_enable_info() { return g_is_log_enable_info; } -inline bool is_log_enable_timing() { return g_is_log_enable_timing; } -inline bool is_log_enable_warn() { return g_is_log_enable_warn; } -inline bool is_log_enable_error() { return g_is_log_enable_error; } - // Initialize log switch (platform-specific implementation) void init_log_switch(); diff --git a/src/common/platform/onboard/aicpu/device_log.cpp b/src/common/platform/onboard/aicpu/device_log.cpp index 7b47cbac4f..44eb946693 100644 --- a/src/common/platform/onboard/aicpu/device_log.cpp +++ b/src/common/platform/onboard/aicpu/device_log.cpp @@ -24,11 +24,23 @@ #include #include +namespace { bool g_is_log_enable_debug = false; bool g_is_log_enable_info = false; bool g_is_log_enable_timing = false; bool g_is_log_enable_warn = false; bool g_is_log_enable_error = false; +} // namespace + +bool is_log_enable_debug() { return g_is_log_enable_debug; } + +bool is_log_enable_info() { return g_is_log_enable_info; } + +bool is_log_enable_timing() { return g_is_log_enable_timing; } + +bool is_log_enable_warn() { return g_is_log_enable_warn; } + +bool is_log_enable_error() { return g_is_log_enable_error; } void init_log_switch() { g_is_log_enable_debug = CheckLogLevel(AICPU, DLOG_DEBUG); diff --git a/src/common/platform/shared/aicpu/unified_log_device.cpp b/src/common/platform/shared/aicpu/unified_log_device.cpp index 34398f2c79..aa1c139c33 100644 --- a/src/common/platform/shared/aicpu/unified_log_device.cpp +++ b/src/common/platform/shared/aicpu/unified_log_device.cpp @@ -13,11 +13,12 @@ * @brief Unified logging - Device implementation. * * Forwards the unified C ABI to dev_vlog_* primitives via va_list — no - * intermediate vsnprintf-to-buffer round-trip in this layer. On sim, - * dev_vlog_* is a single vfprintf (buffer-free); on onboard, it still - * buffers internally because CANN's dlog has no va_list variant. + * intermediate vsnprintf-to-buffer round-trip in this layer. Sim dev_vlog_* + * delegates to the bound HostLogger; onboard buffers internally because + * CANN's dlog has no va_list variant. * - * Level flags come from device_log.cpp's globals (set at init time). + * Severity queries are platform-specific: sim reads the live bound threshold, + * while onboard reads the CANN-derived flags initialized by device_log.cpp. */ #include "common/unified_log.h" diff --git a/src/common/platform/sim/aicpu/device_log.cpp b/src/common/platform/sim/aicpu/device_log.cpp index 8d89483766..c2a71dff83 100644 --- a/src/common/platform/sim/aicpu/device_log.cpp +++ b/src/common/platform/sim/aicpu/device_log.cpp @@ -12,118 +12,84 @@ * @file device_log.cpp (sim) * @brief Simulation Platform Log Implementation * - * Level flags are populated by host via set_log_level() at AICPU kernel init - * (see kernel.cpp / aicpu_executor.cpp); this file does not read env vars. + * The process-owned HostLogger state is bound by the sim host before AICPU + * execution begins; this file does not read env vars. */ #include "aicpu/device_log.h" #include "common/host_log_binding.h" +#include "host_log.h" -#include #include -#include -#include -#include - -// ============================================================================= -// Level enable flags (mutated by the setter below) -// ============================================================================= - -bool g_is_log_enable_debug = false; -bool g_is_log_enable_info = false; -bool g_is_log_enable_timing = true; -bool g_is_log_enable_warn = true; -bool g_is_log_enable_error = true; namespace { SimplerHostLogState *g_host_log_state = nullptr; } +bool is_log_enable_debug() { return HostLogger::get_instance().is_enabled(simpler::log::LogLevel::DEBUG); } + +bool is_log_enable_info() { return HostLogger::get_instance().is_enabled(simpler::log::LogLevel::INFO); } + +bool is_log_enable_timing() { return HostLogger::get_instance().is_enabled(simpler::log::LogLevel::TIMING); } + +bool is_log_enable_warn() { return HostLogger::get_instance().is_enabled(simpler::log::LogLevel::WARN); } + +bool is_log_enable_error() { return HostLogger::get_instance().is_enabled(simpler::log::LogLevel::ERROR); } + // ============================================================================= // Setters (called by AICPU init from KernelArgs) // ============================================================================= extern "C" void set_log_level(int level) { - g_is_log_enable_debug = level <= 10; - g_is_log_enable_info = level <= 20; - g_is_log_enable_timing = level <= 25; - g_is_log_enable_warn = level <= 30; - g_is_log_enable_error = level <= 40; + if (g_host_log_state != nullptr && simpler::log::is_valid_level(level)) { + HostLogger::get_instance().set_level(static_cast(level)); + } } -extern "C" void set_host_log_state(SimplerHostLogState *state) { g_host_log_state = state; } +extern "C" int set_host_log_state(SimplerHostLogState *state) { + const int result = HostLogger::get_instance().bind_state(state); + if (result == 0) g_host_log_state = state; + return result; +} int bind_orchestration_host_log_state(void *handle, const char **error) { return simpler::log::bind_loaded_host_log_state(handle, g_host_log_state, error); } // ============================================================================= -// init_log_switch: sim respects host-pushed config. The no-op entry point is -// retained for ABI compatibility with onboard, where it queries CANN dlog. +// init_log_switch: the sim threshold lives in the bound host-log state. The +// no-op entry point shares the platform interface with the onboard CANN query. // ============================================================================= void init_log_switch() { - // Sim has no env / dlog to consult. Defaults already applied at static - // init; host overrides via set_log_level() before this - // is called. + // Sim has no CANN log switch to query. } // ============================================================================= // Low-level dev_log_* / dev_vlog_* // -// Each record "[TAG] func: body\n" is formatted into a single stack buffer and -// emitted with one write(). The buffer caps the record at 2048 bytes, below -// Linux PIPE_BUF (4096), so a record is delivered atomically when stderr is a -// pipe — concurrent AICPU sim threads and forked chip workers sharing stderr -// never interleave partial records. +// The shared AICPU adapter retains this va_list interface on both platforms. +// HostLogger owns the sim envelope, threshold and configured output; onboard +// supplies the corresponding CANN-backed definitions in its platform +// implementation. // ============================================================================= -namespace { - -void emit_record(const char *level_tag, const char *func, const char *fmt, va_list args) { - char buffer[2048]; - constexpr size_t kNewlineSlot = 1; - constexpr size_t kBodyLimit = sizeof(buffer) - kNewlineSlot; - - int prefix = snprintf(buffer, sizeof(buffer), "[%s] %s: ", level_tag, func); - size_t len = (prefix < 0) ? 0 : static_cast(prefix); - if (len > kBodyLimit) { - len = kBodyLimit; // prefix filled the buffer; reserve the newline slot - } - - int body = vsnprintf(buffer + len, sizeof(buffer) - len, fmt, args); - if (body > 0) { - len += static_cast(body); - if (len > kBodyLimit) { - len = kBodyLimit; // body truncated; reserve the newline slot - } - } - - buffer[len++] = '\n'; - // On a pipe this transfers the whole record (<= PIPE_BUF) in one atomic - // call; the loop only iterates for a non-pipe stderr (regular file, socket) - // where write(2) may be interrupted or short, and must not leave a record - // without its terminating newline. - for (size_t off = 0; off < len;) { - ssize_t written = write(STDERR_FILENO, buffer + off, len - off); - if (written < 0) { - if (errno == EINTR) { - continue; - } - break; // nothing the device-log backend can do on a hard failure - } - off += static_cast(written); - } +void dev_vlog_debug(const char *func, const char *fmt, va_list args) { + HostLogger::get_instance().vlog(simpler::log::LogLevel::DEBUG, func, fmt, args); } -} // namespace - -void dev_vlog_debug(const char *func, const char *fmt, va_list args) { emit_record("DEBUG", func, fmt, args); } - -void dev_vlog_info(const char *func, const char *fmt, va_list args) { emit_record("INFO", func, fmt, args); } +void dev_vlog_info(const char *func, const char *fmt, va_list args) { + HostLogger::get_instance().vlog(simpler::log::LogLevel::INFO, func, fmt, args); +} -void dev_vlog_timing(const char *func, const char *fmt, va_list args) { emit_record("TIMING", func, fmt, args); } +void dev_vlog_timing(const char *func, const char *fmt, va_list args) { + HostLogger::get_instance().vlog(simpler::log::LogLevel::TIMING, func, fmt, args); +} -void dev_vlog_warn(const char *func, const char *fmt, va_list args) { emit_record("WARN", func, fmt, args); } +void dev_vlog_warn(const char *func, const char *fmt, va_list args) { + HostLogger::get_instance().vlog(simpler::log::LogLevel::WARN, func, fmt, args); +} -void dev_vlog_error(const char *func, const char *fmt, va_list args) { emit_record("ERROR", func, fmt, args); } +void dev_vlog_error(const char *func, const char *fmt, va_list args) { + HostLogger::get_instance().vlog(simpler::log::LogLevel::ERROR, func, fmt, args); +} diff --git a/tests/st/host_build_graph_validation/test_host_build_graph_validation.py b/tests/st/host_build_graph_validation/test_host_build_graph_validation.py index f44d16f1a8..2a4896ce38 100644 --- a/tests/st/host_build_graph_validation/test_host_build_graph_validation.py +++ b/tests/st/host_build_graph_validation/test_host_build_graph_validation.py @@ -11,6 +11,7 @@ import functools import os +import time import pytest import torch @@ -22,6 +23,8 @@ DataType, TaskArgs, TensorArgType, + _flush_host_log, + _host_log_dropped_records, ) from simpler.worker import Worker @@ -42,6 +45,33 @@ } +def _wait_for_host_log(capfd, markers: tuple[str, ...], dropped_before: int, timeout_s: float = 5.0) -> str: + """Poll for required records; a single scheduler-dependent flush is not the verdict.""" + chunks: list[str] = [] + deadline = time.monotonic() + timeout_s + last_flush = False + while True: + last_flush = _flush_host_log(100) + captured = capfd.readouterr() + chunks.extend((captured.err, captured.out)) + log = "".join(chunks) + if all(marker in log for marker in markers): + dropped_after = _host_log_dropped_records() + assert dropped_after == dropped_before, ( + f"host-log drop counter changed while waiting for {markers}: " + f"before={dropped_before}, after={dropped_after}" + ) + return log + if time.monotonic() >= deadline: + dropped_after = _host_log_dropped_records() + missing = [marker for marker in markers if marker not in log] + raise AssertionError( + f"host-log records did not arrive within {timeout_s:.1f}s: missing={missing}, " + f"last_flush={last_flush}, dropped_delta={dropped_after - dropped_before}, tail={log[-2000:]!r}" + ) + time.sleep(0.01) + + @functools.cache def _build_callable(platform: str) -> ChipCallable: compiler = KernelCompiler(platform=platform) @@ -82,6 +112,7 @@ def _build_callable(platform: str) -> ChipCallable: def test_invalid_input_reports_code_five(st_platform, st_device_ids, case_name, capfd): worker = Worker(level=2, platform=st_platform, runtime=RUNTIME, device_id=int(st_device_ids[0])) buffer = None + dropped_before = _host_log_dropped_records() try: handle = worker.register(_build_callable(st_platform)) worker.init() @@ -99,10 +130,7 @@ def test_invalid_input_reports_code_five(st_platform, st_device_ids, case_name, with pytest.raises(RuntimeError, match=r"(run_runtime|run) failed with code -5\b"): worker.run(handle, args, config) - captured = capfd.readouterr() - log = captured.err + captured.out - assert "orch_error_code=5" in log - assert "INVALID_ARGS" in log + _wait_for_host_log(capfd, ("orch_error_code=5", "INVALID_ARGS"), dropped_before) finally: if buffer is not None: worker.free(buffer) diff --git a/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py b/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py index 376bc21df1..ea63310559 100644 --- a/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py +++ b/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py @@ -29,9 +29,17 @@ """ import os +import time import pytest -from simpler.task_interface import ArgDirection, CallConfig, ChipCallable, CoreCallable +from simpler.task_interface import ( + ArgDirection, + CallConfig, + ChipCallable, + CoreCallable, + _flush_host_log, + _host_log_dropped_records, +) from simpler.worker import Worker from simpler_setup.elf_parser import extract_text_section @@ -44,6 +52,34 @@ KERNELS = os.path.join(HERE, "kernels") ORCH_DIR = os.path.join(KERNELS, "orchestration") + +def _wait_for_host_log(capfd, markers: tuple[str, ...], dropped_before: int, timeout_s: float = 5.0) -> str: + """Poll for required records and report queue loss instead of trusting one flush deadline.""" + chunks: list[str] = [] + deadline = time.monotonic() + timeout_s + last_flush = False + while True: + last_flush = _flush_host_log(100) + captured = capfd.readouterr() + chunks.extend((captured.err, captured.out)) + log = "".join(chunks) + if all(marker in log for marker in markers): + dropped_after = _host_log_dropped_records() + assert dropped_after == dropped_before, ( + f"host-log drop counter changed while waiting for {markers}: " + f"before={dropped_before}, after={dropped_after}" + ) + return log + if time.monotonic() >= deadline: + dropped_after = _host_log_dropped_records() + missing = [marker for marker in markers if marker not in log] + raise AssertionError( + f"host-log records did not arrive within {timeout_s:.1f}s: missing={missing}, " + f"last_flush={last_flush}, dropped_delta={dropped_after - dropped_before}, tail={log[-2000:]!r}" + ) + time.sleep(0.01) + + # case -> dict(orch, code, runtime_env, kernel, marker, explain) # code : runtime status the host reports in sim (orch_error_code or sched_error_code) # runtime_env: CallConfig.runtime_env overrides that pin the offending resource small @@ -304,13 +340,16 @@ def test_fatal_code_surfaces_on_sim(st_platform, st_device_ids, case_name, monke case = CASES[case_name] if case.get("onboard_only"): pytest.skip("hang kernel would spin the simulator forever (no STARS watchdog on sim)") + dropped_before = _host_log_dropped_records() worker, handle, config = _make_worker(st_platform, int(st_device_ids[0]), case_name, monkeypatch) try: with pytest.raises(RuntimeError, match=rf"(run_runtime|run) failed with code -{case['code']}\b"): worker.run(handle, None, config) - captured = capfd.readouterr() - log = captured.err + captured.out - assert case["marker"] in log, f"missing '{case['marker']}' in host log" + log = _wait_for_host_log( + capfd, + (case["marker"], "error detail:", case["explain"], "error hint:"), + dropped_before, + ) _assert_annotated(log, case) finally: worker.close() @@ -325,6 +364,7 @@ def test_device_error_class_reaches_host_log(st_platform, st_device_ids, case_na """onboard: the watchdog may mask the code as 507xxx, but the device class still reaches the host log.""" configure_logging("error") case = CASES[case_name] + dropped_before = _host_log_dropped_records() worker, handle, config = _make_worker(st_platform, int(st_device_ids[0]), case_name, monkeypatch) try: # On hardware the op-execute / stream-sync watchdog can surface a generic @@ -332,9 +372,11 @@ def test_device_error_class_reaches_host_log(st_platform, st_device_ids, case_na # run fails. The point of the test is the device-classified host LOG. with pytest.raises(RuntimeError): worker.run(handle, None, config) - captured = capfd.readouterr() - log = captured.err + captured.out - assert case["marker"] in log, f"device error class '{case['marker']}' not in host log" + log = _wait_for_host_log( + capfd, + (case["marker"], "error detail:", case["explain"], "error hint:"), + dropped_before, + ) _assert_annotated(log, case) finally: worker.close() diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index 0d8b08c239..994afd2e46 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -1123,6 +1123,56 @@ target_link_libraries(test_host_log_unbound PRIVATE ) add_test(NAME test_host_log_unbound COMMAND test_host_log_unbound) +add_executable(test_host_log_nonblocking + common/test_host_log_nonblocking.cpp + ${HOST_LOG_TEST_SOURCES} +) +target_compile_definitions(test_host_log_nonblocking PRIVATE SIMPLER_HOST_LOG_TEST_HOOKS=1) +target_include_directories(test_host_log_nonblocking PRIVATE + ${GTEST_INCLUDE_DIRS} + ${SIMPLER_LOG_DIR} + ${SIMPLER_LOG_DIR}/include +) +target_link_libraries(test_host_log_nonblocking PRIVATE + ${GTEST_MAIN_LIB} + ${GTEST_LIB} + pthread +) +add_test(NAME test_host_log_nonblocking COMMAND test_host_log_nonblocking) +set_tests_properties(test_host_log_nonblocking PROPERTIES TIMEOUT 8 LABELS "no_hardware") + +add_library(test_host_log_consumer SHARED + common/test_host_log_consumer.cpp + ${HOST_LOG_TEST_SOURCES} +) +target_include_directories(test_host_log_consumer PRIVATE + ${SIMPLER_LOG_DIR} + ${SIMPLER_LOG_DIR}/include +) +target_link_libraries(test_host_log_consumer PRIVATE pthread) + +add_executable(test_host_log_cross_dso + common/test_host_log_cross_dso.cpp + ${HOST_LOG_TEST_SOURCES} +) +target_compile_definitions(test_host_log_cross_dso PRIVATE + TEST_HOST_LOG_CONSUMER_PATH="$" +) +target_include_directories(test_host_log_cross_dso PRIVATE + ${GTEST_INCLUDE_DIRS} + ${SIMPLER_LOG_DIR} + ${SIMPLER_LOG_DIR}/include +) +target_link_libraries(test_host_log_cross_dso PRIVATE + ${GTEST_MAIN_LIB} + ${GTEST_LIB} + ${CMAKE_DL_LIBS} + pthread +) +add_dependencies(test_host_log_cross_dso test_host_log_consumer) +add_test(NAME test_host_log_cross_dso COMMAND test_host_log_cross_dso) +set_tests_properties(test_host_log_cross_dso PROPERTIES LABELS "no_hardware") + set(COMMON_PLATFORM_DIR ${CMAKE_SOURCE_DIR}/../../../src/common/platform) function(add_device_phase_capture_test name host_strace device_strace expected) add_executable(${name} @@ -1178,12 +1228,13 @@ target_link_libraries(test_onboard_device_log_level PRIVATE add_test(NAME test_onboard_device_log_level COMMAND test_onboard_device_log_level) set_tests_properties(test_onboard_device_log_level PROPERTIES LABELS "no_hardware") -# Sim device-log record atomicity: concurrent threads and forked workers must -# not interleave partial records on the shared stderr. Pure host code — compile -# the sim backend alongside the test; no CANN, no hardware. +# Sim device logging uses the bound HostLogger threshold, envelope, and output. +# Pure host code — compile the sim adapter and HostLogger alongside the test; +# no CANN, no hardware. add_executable(test_sim_device_log common/test_sim_device_log.cpp ${COMMON_PLATFORM_DIR}/sim/aicpu/device_log.cpp + ${SIMPLER_LOG_DIR}/host_log.cpp ) target_include_directories(test_sim_device_log PRIVATE ${GTEST_INCLUDE_DIRS} diff --git a/tests/ut/cpp/a5/test_host_log_off.cpp b/tests/ut/cpp/a5/test_host_log_off.cpp index dd9e822637..f9282af109 100644 --- a/tests/ut/cpp/a5/test_host_log_off.cpp +++ b/tests/ut/cpp/a5/test_host_log_off.cpp @@ -56,6 +56,15 @@ SimplerHostLogState g_shared_log_state{ sizeof(SimplerHostLogState), static_cast(LogLevel::TIMING), 0, + 0, + {}, + 0, + 0, + nullptr, + nullptr, + 0, + 0, + 0, }; int capture_cann_log_level(int module_id, int level, int enable_event) { @@ -80,6 +89,7 @@ CapturedStdio run_with_config(LogLevel level, Fn &&fn) { HostLogger::get_instance().set_level(level); fn(); + EXPECT_TRUE(HostLogger::get_instance().flush()); fflush(stdout); fflush(stderr); @@ -121,6 +131,9 @@ TEST(HostLogTest, SharedStateBindingValidatesAbiAndOwnsThreshold) { g_shared_log_state.threshold = static_cast(LogLevel::ERROR); g_shared_log_state.clock_anchor_pid = 0; ASSERT_EQ(simpler_host_log_bind_state(&g_shared_log_state), 0); + // This executable supplies the process-owned storage; production consumers + // only take the exported bind path and therefore cannot create its writer. + ASSERT_EQ(HostLogger::get_instance().adopt_state(&g_shared_log_state), 0); EXPECT_EQ(HostLogger::get_instance().state(), &g_shared_log_state); EXPECT_EQ(HostLogger::get_instance().level(), static_cast(LogLevel::ERROR)); EXPECT_FALSE(HostLogger::get_instance().is_enabled(LogLevel::WARN)); @@ -154,6 +167,7 @@ TEST(HostLogTest, HostSpanEnabledFollowsTimingVisibility) { EXPECT_EQ(unified_log_host_span_enabled(), expected); } HostLogger::get_instance().set_level(LogLevel::TIMING); + EXPECT_TRUE(HostLogger::get_instance().flush()); } TEST(HostLogTest, ErrorLevelEmitsErrorOnly) { @@ -248,6 +262,7 @@ TEST(HostLogTest, EmitPrefixHasMonotonicNanosecondsAndTid) { } TEST(HostLogTest, TimingStartupEmitsOneClockAnchorPerProcess) { + ASSERT_TRUE(HostLogger::get_instance().prepare_to_fork()); int log_pipe[2]; ASSERT_EQ(pipe(log_pipe), 0); @@ -261,6 +276,7 @@ TEST(HostLogTest, TimingStartupEmitsOneClockAnchorPerProcess) { HostLogger::get_instance().set_level(LogLevel::TIMING); HostLogger::get_instance().log(LogLevel::TIMING, "child", "first-record"); HostLogger::get_instance().log(LogLevel::TIMING, "child", "second-record"); + if (!HostLogger::get_instance().flush()) _exit(3); _exit(0); } @@ -277,6 +293,7 @@ TEST(HostLogTest, TimingStartupEmitsOneClockAnchorPerProcess) { ASSERT_EQ(waitpid(child, &status, 0), child); ASSERT_TRUE(WIFEXITED(status)); ASSERT_EQ(WEXITSTATUS(status), 0); + ASSERT_TRUE(HostLogger::get_instance().start_writer()); const size_t anchor_pos = captured.find("[CLOCK_ANCHOR]"); ASSERT_NE(anchor_pos, std::string::npos); @@ -323,6 +340,19 @@ TEST(HostLogTest, AllOutputGoesToStderr) { EXPECT_NE(captured.err.find("debug-output-marker"), std::string::npos); } +TEST(HostLogTest, LongHumanRecordFitsPortableAtomicWriteBound) { + const std::string payload(4096, 'x'); + auto captured = run_with_config(LogLevel::ERROR, [&] { + HostLogger::get_instance().log(LogLevel::ERROR, "long_record", "%s", payload.c_str()); + }); + + EXPECT_EQ(captured.out, ""); + ASSERT_EQ(captured.err.size(), static_cast(_POSIX_PIPE_BUF)); + EXPECT_EQ(std::count(captured.err.begin(), captured.err.end(), '\n'), 1); + EXPECT_EQ(captured.err[captured.err.size() - 2], '~'); + EXPECT_EQ(captured.err.back(), '\n'); +} + 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'); @@ -384,7 +414,7 @@ std::string read_log_file(const char *directory, pid_t pid) { return std::string((std::istreambuf_iterator(input)), std::istreambuf_iterator()); } -TEST(HostLogTest, LogDirectorySendsEveryRecordToOneBufferedFilePerProcess) { +TEST(HostLogTest, LogDirectorySendsEveryRecordToOneAsyncFilePerProcess) { char directory_template[] = "/tmp/simpler-host-strace-XXXXXX"; char *directory = mkdtemp(directory_template); ASSERT_NE(directory, nullptr); @@ -434,7 +464,7 @@ TEST(HostLogTest, LogDirectorySendsEveryRecordToOneBufferedFilePerProcess) { EXPECT_EQ(rmdir(directory), 0); } -TEST(HostLogTest, LogDirectoryTakesOrdinaryRecordsAndFlushesTheSevereOnes) { +TEST(HostLogTest, ExplicitDrainMakesAllAcceptedFileRecordsVisible) { char directory_template[] = "/tmp/simpler-host-log-ordinary-XXXXXX"; char *directory = mkdtemp(directory_template); ASSERT_NE(directory, nullptr); @@ -443,22 +473,22 @@ TEST(HostLogTest, LogDirectoryTakesOrdinaryRecordsAndFlushesTheSevereOnes) { g_shared_log_state.clock_anchor_pid = 0; const auto captured = run_with_config(LogLevel::TIMING, [] { HostLogger::get_instance().log(LogLevel::ERROR, "fn", "disk-please"); - HostLogger::get_instance().log(LogLevel::INFO, "fn", "buffered-please"); + HostLogger::get_instance().log(LogLevel::TIMING, "fn", "queued-please"); }); EXPECT_EQ(captured.err, ""); - // Read before any root span closes: an error is rare and worth having on - // disk if the process dies, so it is flushed as it is written. The INFO - // record rides the buffer and need not be visible yet. + // run_with_config performs the explicit shutdown/test drain. Severity no + // longer selects a producer-side flush path: both records use one queue. const std::string contents = read_log_file(directory, getpid()); EXPECT_NE(contents.find("disk-please"), std::string::npos); + EXPECT_NE(contents.find("queued-please"), std::string::npos); const std::string path = std::string(directory) + "/host." + std::to_string(static_cast(getpid())) + ".log"; EXPECT_EQ(unlink(path.c_str()), 0); EXPECT_EQ(rmdir(directory), 0); } -TEST(HostLogTest, ForkedChildReopensItsOwnSpanFileWithoutFlushingParentBuffer) { +TEST(HostLogTest, ForkBoundaryDrainsParentAndChildOpensItsOwnFile) { char directory_template[] = "/tmp/simpler-host-strace-fork-XXXXXX"; char *directory = mkdtemp(directory_template); ASSERT_NE(directory, nullptr); @@ -472,6 +502,9 @@ TEST(HostLogTest, ForkedChildReopensItsOwnSpanFileWithoutFlushingParentBuffer) { SIMPLER_HOST_SPAN_ABI_VERSION, sizeof(SimplerHostSpan), 8, 0x1234, 1, 0, 100, 25, "parent.span", "" }; unified_log_host_span(&parent_span); + // Production uses this same quiescent boundary before a hierarchical + // Worker forks: accepted parent records are drained and the thread joined. + ASSERT_TRUE(HostLogger::get_instance().prepare_to_fork()); const pid_t child = fork(); ASSERT_GE(child, 0); @@ -479,9 +512,12 @@ TEST(HostLogTest, ForkedChildReopensItsOwnSpanFileWithoutFlushingParentBuffer) { const SimplerHostSpan child_span{ SIMPLER_HOST_SPAN_ABI_VERSION, sizeof(SimplerHostSpan), 9, 0x1234, 0, 0, 200, 25, "child.span", "" }; + HostLogger::get_instance().set_level(LogLevel::TIMING); unified_log_host_span(&child_span); + if (!HostLogger::get_instance().flush()) _exit(3); _exit(0); } + ASSERT_TRUE(HostLogger::get_instance().start_writer()); int status = 0; ASSERT_EQ(waitpid(child, &status, 0), child); ASSERT_TRUE(WIFEXITED(status)); @@ -495,14 +531,10 @@ TEST(HostLogTest, ForkedChildReopensItsOwnSpanFileWithoutFlushingParentBuffer) { const std::string child_contents((std::istreambuf_iterator(child_input)), std::istreambuf_iterator()); EXPECT_EQ(child_contents.find("name=parent.span"), std::string::npos); EXPECT_NE(child_contents.find("name=child.span"), std::string::npos); - // The parent's record is depth 1, so it is still unflushed in the parent's - // own buffer. It can only have reached the parent's file if the child - // flushed the copy it inherited — which is what this test is named for, and - // what the child's file alone cannot show. std::ifstream parent_input(parent_path); const std::string parent_contents((std::istreambuf_iterator(parent_input)), std::istreambuf_iterator()); - EXPECT_EQ(parent_contents.find("name=parent.span"), std::string::npos) - << "the child flushed the parent's copied stdio buffer"; + EXPECT_NE(parent_contents.find("name=parent.span"), std::string::npos); + EXPECT_EQ(parent_contents.find("name=child.span"), std::string::npos); EXPECT_EQ(unlink(parent_path.c_str()), 0); EXPECT_EQ(unlink(child_path.c_str()), 0); EXPECT_EQ(rmdir(directory), 0); @@ -510,7 +542,7 @@ TEST(HostLogTest, ForkedChildReopensItsOwnSpanFileWithoutFlushingParentBuffer) { TEST(HostLogTest, DisabledHostSpanProducesNoRecord) { const SimplerHostSpan span{ - SIMPLER_HOST_SPAN_ABI_VERSION, sizeof(SimplerHostSpan), 7, 0x1234, 0, 0, 100, 25, "host.dispatch", + SIMPLER_HOST_SPAN_ABI_VERSION, sizeof(SimplerHostSpan), 7, 0x1234, 0, 0, 100, 25, "node.dispatch", "run_id=7 role=scheduler" }; @@ -586,6 +618,7 @@ TEST(HostLogTest, HostSpanTruncationDropsAWholeEscapeRatherThanItsLastByte) { } TEST(HostLogTest, ForkedProcessesEmitWholePipeRecords) { + ASSERT_TRUE(HostLogger::get_instance().prepare_to_fork()); int log_pipe[2]; int start_pipe[2]; ASSERT_EQ(pipe(log_pipe), 0); @@ -593,7 +626,7 @@ TEST(HostLogTest, ForkedProcessesEmitWholePipeRecords) { const long pipe_buf = fpathconf(log_pipe[1], _PC_PIPE_BUF); ASSERT_GT(pipe_buf, 256); - const size_t payload_size = static_cast(std::min(pipe_buf - 256, 2048)); + constexpr size_t payload_size = 128; constexpr int child_count = 16; constexpr int records_per_child = 128; @@ -618,6 +651,7 @@ TEST(HostLogTest, ForkedProcessesEmitWholePipeRecords) { LogLevel::ERROR, "fork_writer", "child=%d seq=%d payload=%s", child, seq, payload.c_str() ); } + if (!HostLogger::get_instance().flush(5000)) _exit(3); _exit(0); } children.push_back(pid); @@ -657,6 +691,7 @@ TEST(HostLogTest, ForkedProcessesEmitWholePipeRecords) { } } reader.join(); + ASSERT_TRUE(HostLogger::get_instance().start_writer()); std::vector> seen(child_count, std::vector(records_per_child, false)); std::set anchor_pids; diff --git a/tests/ut/cpp/common/test_host_log_consumer.cpp b/tests/ut/cpp/common/test_host_log_consumer.cpp new file mode 100644 index 0000000000..188ecce7f4 --- /dev/null +++ b/tests/ut/cpp/common/test_host_log_consumer.cpp @@ -0,0 +1,20 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include "host_log.h" + +extern "C" __attribute__((visibility("default"))) void test_host_log_consumer_emit() { + HostLogger::get_instance().log(simpler::log::LogLevel::ERROR, "consumer", "cross-dso-record"); +} + +extern "C" __attribute__((visibility("default"))) int test_host_log_consumer_start_writer() { + return HostLogger::get_instance().start_writer() ? 1 : 0; +} diff --git a/tests/ut/cpp/common/test_host_log_cross_dso.cpp b/tests/ut/cpp/common/test_host_log_cross_dso.cpp new file mode 100644 index 0000000000..c39b7c9920 --- /dev/null +++ b/tests/ut/cpp/common/test_host_log_cross_dso.cpp @@ -0,0 +1,67 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include +#include + +#include +#include + +#include + +#include "common/host_log_state.h" +#include "host_log.h" + +using simpler::log::LogLevel; + +TEST(HostLogCrossDsoTest, BoundConsumerUsesProcessOwnedFileSink) { + char directory_template[] = "/tmp/simpler-host-log-cross-dso-XXXXXX"; + char *directory = mkdtemp(directory_template); + ASSERT_NE(directory, nullptr); + + HostLogger &owner = HostLogger::get_instance(); + owner.set_log_directory(directory); + owner.set_level(LogLevel::ERROR); + + void *handle = dlopen(TEST_HOST_LOG_CONSUMER_PATH, RTLD_NOW | RTLD_LOCAL); + ASSERT_NE(handle, nullptr) << dlerror(); + + dlerror(); + auto bind = reinterpret_cast(dlsym(handle, "simpler_host_log_bind_state")); + ASSERT_NE(bind, nullptr) << dlerror(); + ASSERT_EQ(bind(owner.state()), 0); + + dlerror(); + auto emit = reinterpret_cast(dlsym(handle, "test_host_log_consumer_emit")); + ASSERT_NE(emit, nullptr) << dlerror(); + dlerror(); + auto start_writer = reinterpret_cast(dlsym(handle, "test_host_log_consumer_start_writer")); + ASSERT_NE(start_writer, nullptr) << dlerror(); + + owner.log(LogLevel::ERROR, "owner", "owner-record"); + emit(); + ASSERT_TRUE(owner.flush()); + EXPECT_EQ(start_writer(), 1) << "a bound consumer should recognize the already-published owner sink"; + + const std::string path = std::string(directory) + "/host." + std::to_string(static_cast(getpid())) + ".log"; + std::ifstream input(path); + ASSERT_TRUE(input.good()); + const std::string captured((std::istreambuf_iterator(input)), std::istreambuf_iterator()); + input.close(); + + EXPECT_NE(captured.find("][ERROR] owner: owner-record\n"), std::string::npos); + EXPECT_NE(captured.find("][ERROR] consumer: cross-dso-record\n"), std::string::npos); + ASSERT_TRUE(owner.prepare_to_fork()); + EXPECT_EQ(start_writer(), 0) << "a transient bound DSO must not become the process sink owner"; + EXPECT_EQ(dlclose(handle), 0); + EXPECT_EQ(unlink(path.c_str()), 0); + EXPECT_EQ(rmdir(directory), 0); +} diff --git a/tests/ut/cpp/common/test_host_log_nonblocking.cpp b/tests/ut/cpp/common/test_host_log_nonblocking.cpp new file mode 100644 index 0000000000..9162039b75 --- /dev/null +++ b/tests/ut/cpp/common/test_host_log_nonblocking.cpp @@ -0,0 +1,238 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include + +#include "host_log.h" + +using simpler::log::LogLevel; + +extern "C" void simpler_host_log_set_queue_hooks_for_test(void (*after_claim)(size_t), void (*before_gap_wait)()); + +namespace { + +std::mutex g_gap_mutex; +std::condition_variable g_gap_cv; +bool g_first_claimed = false; +bool g_release_first = false; +bool g_writer_waiting_for_gap = false; + +void pause_first_queue_claim(size_t position) { + if (position != 0) return; + std::unique_lock lock(g_gap_mutex); + g_first_claimed = true; + g_gap_cv.notify_all(); + g_gap_cv.wait(lock, [] { + return g_release_first; + }); +} + +void observe_writer_gap_wait() { + std::scoped_lock lock(g_gap_mutex); + g_writer_waiting_for_gap = true; + g_gap_cv.notify_all(); +} + +} // namespace + +TEST(HostLogNonblockingTest, DeferredWriterPreservesInitializationRecordsAndDropCount) { + HostLogger &logger = HostLogger::get_instance(); + logger.set_level(LogLevel::ERROR, /*defer_writer=*/true); + + const uint64_t before = logger.dropped_records(); + testing::internal::CaptureStderr(); + logger.log(LogLevel::ERROR, "initialization", "before-writer"); + const std::string captured = testing::internal::GetCapturedStderr(); + EXPECT_NE(captured.find("][ERROR] initialization: before-writer\n"), std::string::npos); + EXPECT_EQ(logger.dropped_records(), before); + + const int saved_stderr = dup(STDERR_FILENO); + ASSERT_GE(saved_stderr, 0); + ASSERT_EQ(close(STDERR_FILENO), 0); + logger.log(LogLevel::ERROR, "initialization", "failed-before-writer"); + ASSERT_GE(dup2(saved_stderr, STDERR_FILENO), 0); + close(saved_stderr); + ASSERT_EQ(logger.dropped_records(), before + 1); + + // The first writer start records this process PID. It is not a fork-child + // transition and must not erase failures observed during initialization. + ASSERT_TRUE(logger.start_writer()); + EXPECT_EQ(logger.dropped_records(), before + 1); + EXPECT_TRUE(logger.prepare_to_fork()); +} + +TEST(HostLogNonblockingTest, WriterSleepsWhenAdjacentProducersPublishOutOfOrder) { + HostLogger &logger = HostLogger::get_instance(); + logger.set_level(LogLevel::ERROR); + { + std::scoped_lock lock(g_gap_mutex); + g_first_claimed = false; + g_release_first = false; + g_writer_waiting_for_gap = false; + } + simpler_host_log_set_queue_hooks_for_test(pause_first_queue_claim, observe_writer_gap_wait); + + std::thread first([&logger] { + logger.log(LogLevel::ERROR, "producer", "first-reserved"); + }); + { + std::unique_lock lock(g_gap_mutex); + EXPECT_TRUE(g_gap_cv.wait_for(lock, std::chrono::seconds(1), [] { + return g_first_claimed; + })); + } + + std::thread second([&logger] { + logger.log(LogLevel::ERROR, "producer", "second-published"); + }); + second.join(); + { + std::unique_lock lock(g_gap_mutex); + EXPECT_TRUE(g_gap_cv.wait_for(lock, std::chrono::seconds(1), [] { + return g_writer_waiting_for_gap; + })) << "writer did not return to sem_wait for the missing earlier slot"; + g_release_first = true; + } + g_gap_cv.notify_all(); + first.join(); + simpler_host_log_set_queue_hooks_for_test(nullptr, nullptr); + + EXPECT_TRUE(logger.flush()); + EXPECT_TRUE(logger.prepare_to_fork()); +} + +TEST(HostLogNonblockingTest, FullStderrPipeDoesNotStallProducer) { + int stderr_pipe[2]; + ASSERT_EQ(pipe(stderr_pipe), 0); +#ifdef F_SETPIPE_SZ + ASSERT_GE(fcntl(stderr_pipe[0], F_SETPIPE_SZ, 4096), 0); +#endif + + const pid_t pid = fork(); + ASSERT_GE(pid, 0); + if (pid == 0) { + close(stderr_pipe[0]); + if (dup2(stderr_pipe[1], STDERR_FILENO) < 0) _exit(10); + close(stderr_pipe[1]); + + alarm(2); + HostLogger::get_instance().set_level(LogLevel::ERROR); + const std::string payload(400, 'x'); + for (int i = 0; i < 10000; ++i) { + HostLogger::get_instance().log(LogLevel::ERROR, "producer", "record=%d %s", i, payload.c_str()); + } + if (HostLogger::get_instance().dropped_records() == 0) _exit(11); + // A blocked writer cannot be joined, so a later hierarchical startup + // must fail boundedly instead of proceeding to fork with it alive. + _exit(HostLogger::get_instance().prepare_to_fork(10) ? 12 : 0); + } + + close(stderr_pipe[1]); + int status = 0; + ASSERT_EQ(waitpid(pid, &status, 0), pid); + close(stderr_pipe[0]); + + ASSERT_TRUE(WIFEXITED(status)) << "producer stalled on stderr (wait status " << status << ")"; + EXPECT_EQ(WEXITSTATUS(status), 0); +} + +TEST(HostLogNonblockingTest, HardWriteFailureIncrementsDropCounter) { + int stderr_pipe[2]; + ASSERT_EQ(pipe(stderr_pipe), 0); + close(stderr_pipe[0]); + + const int saved_stderr = dup(STDERR_FILENO); + ASSERT_GE(saved_stderr, 0); + auto previous_sigpipe = std::signal(SIGPIPE, SIG_IGN); + ASSERT_NE(previous_sigpipe, SIG_ERR); + ASSERT_GE(dup2(stderr_pipe[1], STDERR_FILENO), 0); + close(stderr_pipe[1]); + + HostLogger::get_instance().set_level(LogLevel::ERROR); + const uint64_t before = HostLogger::get_instance().dropped_records(); + HostLogger::get_instance().log(LogLevel::ERROR, "producer", "write-must-fail"); + EXPECT_TRUE(HostLogger::get_instance().flush()); + + ASSERT_GE(dup2(saved_stderr, STDERR_FILENO), 0); + close(saved_stderr); + std::signal(SIGPIPE, previous_sigpipe); + EXPECT_EQ(HostLogger::get_instance().dropped_records(), before + 1); +} + +TEST(HostLogNonblockingTest, ExistingWriterIsJoinedBeforeALaterFork) { + HostLogger &logger = HostLogger::get_instance(); + logger.set_level(LogLevel::ERROR); + logger.log(LogLevel::ERROR, "producer", "before-second-worker"); + ASSERT_TRUE(logger.prepare_to_fork()); + + const pid_t pid = fork(); + ASSERT_GE(pid, 0); + if (pid == 0) { + logger.set_level(LogLevel::NUL); + _exit(logger.flush() ? 0 : 12); + } + + int status = 0; + ASSERT_EQ(waitpid(pid, &status, 0), pid); + ASSERT_TRUE(WIFEXITED(status)); + EXPECT_EQ(WEXITSTATUS(status), 0); + + // The parent can install a fresh writer after the fork boundary. + EXPECT_TRUE(logger.start_writer()); + EXPECT_TRUE(logger.prepare_to_fork()); +} + +TEST(HostLogNonblockingTest, ConcurrentProducersCanBeQuiescedSafely) { + const int saved_stderr = dup(STDERR_FILENO); + ASSERT_GE(saved_stderr, 0); + const int null_fd = open("/dev/null", O_WRONLY); + ASSERT_GE(null_fd, 0); + ASSERT_GE(dup2(null_fd, STDERR_FILENO), 0); + close(null_fd); + + HostLogger &logger = HostLogger::get_instance(); + logger.set_level(LogLevel::ERROR); + std::atomic running{true}; + std::vector producers; + for (int index = 0; index < 4; ++index) { + producers.emplace_back([&logger, &running, index] { + uint64_t sequence = 0; + while (running.load(std::memory_order_acquire)) { + logger.log(LogLevel::ERROR, "producer", "thread=%d sequence=%lu", index, sequence++); + } + }); + } + + for (int iteration = 0; iteration < 20; ++iteration) { + ASSERT_TRUE(logger.prepare_to_fork(1000)); + ASSERT_TRUE(logger.start_writer()); + } + running.store(false, std::memory_order_release); + for (std::thread &producer : producers) + producer.join(); + EXPECT_TRUE(logger.prepare_to_fork(1000)); + + ASSERT_GE(dup2(saved_stderr, STDERR_FILENO), 0); + close(saved_stderr); +} diff --git a/tests/ut/cpp/common/test_host_log_unbound.cpp b/tests/ut/cpp/common/test_host_log_unbound.cpp index 877779e452..69be276818 100644 --- a/tests/ut/cpp/common/test_host_log_unbound.cpp +++ b/tests/ut/cpp/common/test_host_log_unbound.cpp @@ -20,7 +20,7 @@ TEST(HostLogUnboundTest, PrivateModuleStateStartsSilent) { EXPECT_EQ(unified_log_host_span_enabled(), 0); const SimplerHostSpan span{ - SIMPLER_HOST_SPAN_ABI_VERSION, sizeof(SimplerHostSpan), 1, 0, 0, 0, 100, 25, "host.dispatch", "run_id=1" + SIMPLER_HOST_SPAN_ABI_VERSION, sizeof(SimplerHostSpan), 1, 0, 0, 0, 100, 25, "node.dispatch", "run_id=1" }; testing::internal::CaptureStderr(); HostLogger::get_instance().log(simpler::log::LogLevel::ERROR, "unbound", "must stay silent"); diff --git a/tests/ut/cpp/common/test_sim_device_log.cpp b/tests/ut/cpp/common/test_sim_device_log.cpp index da3fdc515c..4e4baf0b85 100644 --- a/tests/ut/cpp/common/test_sim_device_log.cpp +++ b/tests/ut/cpp/common/test_sim_device_log.cpp @@ -9,12 +9,9 @@ * ----------------------------------------------------------------------------------------------------------- */ -// Sim device-log atomicity: every dev_vlog_* call emits exactly one intact -// physical line, even when many AICPU sim threads or forked chip workers write -// the shared stderr concurrently. A record no larger than the pipe's PIPE_BUF -// reaches it in one indivisible write(2), so it cannot interleave with another -// writer. The records here are ~30 bytes, inside even the portable -// _POSIX_PIPE_BUF floor of 512 that macOS uses. +// Sim AICPU records use the bound HostLogger threshold and host envelope. The +// short records below also stay intact when many threads or forked chip workers +// share the stderr fallback. #include #include @@ -31,13 +28,31 @@ #include #include "aicpu/device_log.h" +#include "common/host_log_state.h" +#include "common/log_level.h" +#include "host_log.h" namespace { constexpr const char *kTags[] = {"DEBUG", "INFO", "TIMING", "WARN", "ERROR"}; -// dev_vlog_* gate on nothing (the unified_log_* adapter owns level filtering), -// so these thin wrappers always emit. level_idx selects the backend under test. +SimplerHostLogState g_log_state{ + SIMPLER_HOST_LOG_STATE_ABI_VERSION, + sizeof(SimplerHostLogState), + static_cast(simpler::log::LogLevel::ERROR), + 0, + 0, + {}, + 0, + 0, + nullptr, + nullptr, + 0, + 0, + 0, +}; + +// level_idx selects the compatibility entry point under test. void emit(int level_idx, const char *func, const char *fmt, ...) { va_list ap; va_start(ap, fmt); @@ -65,6 +80,37 @@ std::string record(int level_idx, const char *func, const std::string &body) { return std::string("[") + kTags[level_idx % 5] + "] " + func + ": " + body; } +void bind_level(simpler::log::LogLevel level) { + g_log_state.threshold = static_cast(level); + g_log_state.clock_anchor_pid = static_cast(getpid()); + g_log_state.log_directory_bound = 0; + g_log_state.log_directory[0] = '\0'; + ASSERT_EQ(set_host_log_state(&g_log_state), 0); + // This host-only executable supplies the process state. A production sim + // AICPU DSO remains a bound consumer and cannot create the owner writer. + ASSERT_EQ(HostLogger::get_instance().adopt_state(&g_log_state), 0); + set_log_level(static_cast(level)); +} + +// A failed assertion must not leave the process-global writer stopped for the +// next test. Normal paths call restart() after stderr capture is restored; +// early returns get a best-effort restart from the destructor. +class ScopedHostWriterRestart { +public: + ~ScopedHostWriterRestart() { + if (armed_) (void)HostLogger::get_instance().start_writer(); + } + + bool restart() { + if (!HostLogger::get_instance().start_writer()) return false; + armed_ = false; + return true; + } + +private: + bool armed_ = true; +}; + // Redirect stderr onto a fresh pipe, drained from the moment it is installed. // // A pipe holds a bounded amount of unread data — 16 KiB on macOS, 64 KiB on @@ -74,10 +120,8 @@ std::string record(int level_idx, const char *func, const std::string &body) { // runs concurrently so the writers never block, whatever they emit. // // The reader owns the buffer through a shared_ptr, so returning `Capture` by -// value cannot leave the thread writing into a moved-from string. `fork` is safe -// alongside it because the record path allocates nothing: it formats into a -// stack buffer and calls write(2), so a child cannot deadlock on a malloc lock -// this thread happened to hold. +// value cannot leave the thread writing into a moved-from string. No logger +// producer is active at the fork boundary in the process tests below. struct Capture { int read_fd = -1; int saved_stderr = -1; @@ -108,6 +152,7 @@ Capture begin_capture() { // Restore stderr, which drops this process's last write handle; with every child // already reaped the reader then sees EOF and finishes. std::string end_capture(Capture &cap) { + EXPECT_TRUE(HostLogger::get_instance().flush()); fflush(stderr); EXPECT_GE(dup2(cap.saved_stderr, STDERR_FILENO), 0); close(cap.saved_stderr); @@ -123,8 +168,13 @@ void expect_intact(const std::string &captured, std::multiset expec while (start < captured.size()) { size_t nl = captured.find('\n', start); ASSERT_NE(nl, std::string::npos) << "record missing terminating newline"; - std::string line = captured.substr(start, nl - start); - auto it = expected.find(line); + const std::string line = captured.substr(start, nl - start); + const size_t monotonic_end = line.find("]["); + ASSERT_NE(monotonic_end, std::string::npos) << "host envelope missing from: '" << line << "'"; + const size_t thread_end = line.find("][", monotonic_end + 2); + ASSERT_NE(thread_end, std::string::npos) << "host envelope missing from: '" << line << "'"; + const std::string payload = line.substr(thread_end + 1); + auto it = expected.find(payload); ASSERT_NE(it, expected.end()) << "torn or unexpected record: '" << line << "'"; expected.erase(it); start = nl + 1; @@ -134,10 +184,43 @@ void expect_intact(const std::string &captured, std::multiset expec } // namespace +TEST(SimDeviceLogTest, RejectsIncompatibleHostLogState) { + SimplerHostLogState incompatible = g_log_state; + incompatible.abi_version += 1; + + EXPECT_NE(set_host_log_state(&incompatible), 0); + EXPECT_EQ(set_host_log_state(&g_log_state), 0); +} + +TEST(SimDeviceLogTest, UsesHostEnvelopeAndLiveBoundThreshold) { + bind_level(simpler::log::LogLevel::ERROR); + + testing::internal::CaptureStderr(); + emit(3, "worker", "warn-hidden"); + emit(4, "worker", "error-visible"); + ASSERT_TRUE(HostLogger::get_instance().flush()); + std::string captured = testing::internal::GetCapturedStderr(); + + EXPECT_EQ(captured.find("warn-hidden"), std::string::npos); + EXPECT_NE(captured.find("][ERROR] worker: error-visible\n"), std::string::npos); + EXPECT_NE(captured.find("[mono_ns="), std::string::npos); + EXPECT_NE(captured.find("][T0x"), std::string::npos); + + // The adapter reads the shared state on every query; no second push into a + // private sim flag table is required. + g_log_state.threshold = static_cast(simpler::log::LogLevel::WARN); + testing::internal::CaptureStderr(); + emit(3, "worker", "warn-visible"); + ASSERT_TRUE(HostLogger::get_instance().flush()); + captured = testing::internal::GetCapturedStderr(); + EXPECT_NE(captured.find("][WARN] worker: warn-visible\n"), std::string::npos); +} + TEST(SimDeviceLogTest, MultiThreadedRecordsStayIntact) { constexpr int kThreads = 4; constexpr int kPerThread = 200; + bind_level(simpler::log::LogLevel::DEBUG); std::multiset expected; for (int t = 0; t < kThreads; ++t) { for (int i = 0; i < kPerThread; ++i) { @@ -169,6 +252,9 @@ TEST(SimDeviceLogTest, ForkedProcessesEmitWholeRecords) { constexpr int kChildren = 8; constexpr int kPerChild = 100; + bind_level(simpler::log::LogLevel::DEBUG); + ASSERT_TRUE(HostLogger::get_instance().prepare_to_fork()); + ScopedHostWriterRestart writer_restart; std::multiset expected; for (int c = 0; c < kChildren; ++c) { for (int i = 0; i < kPerChild; ++i) { @@ -185,9 +271,12 @@ TEST(SimDeviceLogTest, ForkedProcessesEmitWholeRecords) { pid_t pid = fork(); ASSERT_GE(pid, 0); if (pid == 0) { + g_log_state.clock_anchor_pid = static_cast(getpid()); + set_log_level(static_cast(simpler::log::LogLevel::DEBUG)); for (int i = 0; i < kPerChild; ++i) { emit(c, "chip_worker", "c%d-r%03d", c, i); } + if (!HostLogger::get_instance().flush()) _exit(3); _exit(0); // skip gtest/atexit teardown so nothing else hits the pipe } pids.push_back(pid); @@ -199,6 +288,7 @@ TEST(SimDeviceLogTest, ForkedProcessesEmitWholeRecords) { EXPECT_EQ(WEXITSTATUS(status), 0); } std::string captured = end_capture(cap); + ASSERT_TRUE(writer_restart.restart()); ASSERT_NO_FATAL_FAILURE(expect_intact(captured, std::move(expected))); } @@ -214,6 +304,9 @@ TEST(SimDeviceLogTest, WritersOutrunASmallPipeWithoutDeadlocking) { constexpr int kChildren = 4; constexpr int kPerChild = 400; + bind_level(simpler::log::LogLevel::DEBUG); + ASSERT_TRUE(HostLogger::get_instance().prepare_to_fork()); + ScopedHostWriterRestart writer_restart; std::multiset expected; for (int c = 0; c < kChildren; ++c) { for (int i = 0; i < kPerChild; ++i) { @@ -226,6 +319,7 @@ TEST(SimDeviceLogTest, WritersOutrunASmallPipeWithoutDeadlocking) { Capture cap = begin_capture(); if (fcntl(cap.read_fd, F_SETPIPE_SZ, 4096) < 0) { std::string discard = end_capture(cap); + ASSERT_TRUE(writer_restart.restart()); GTEST_SKIP() << "cannot shrink the pipe on this kernel"; } @@ -235,9 +329,12 @@ TEST(SimDeviceLogTest, WritersOutrunASmallPipeWithoutDeadlocking) { pid_t pid = fork(); ASSERT_GE(pid, 0); if (pid == 0) { + g_log_state.clock_anchor_pid = static_cast(getpid()); + set_log_level(static_cast(simpler::log::LogLevel::DEBUG)); for (int i = 0; i < kPerChild; ++i) { emit(c, "chip_worker", "c%d-r%03d", c, i); } + if (!HostLogger::get_instance().flush(5000)) _exit(3); _exit(0); } pids.push_back(pid); @@ -249,6 +346,7 @@ TEST(SimDeviceLogTest, WritersOutrunASmallPipeWithoutDeadlocking) { EXPECT_EQ(WEXITSTATUS(status), 0); } std::string captured = end_capture(cap); + ASSERT_TRUE(writer_restart.restart()); ASSERT_NO_FATAL_FAILURE(expect_intact(captured, std::move(expected))); } diff --git a/tests/ut/py/test_chip_worker.py b/tests/ut/py/test_chip_worker.py index 6cc70a498c..2e3fc50b82 100644 --- a/tests/ut/py/test_chip_worker.py +++ b/tests/ut/py/test_chip_worker.py @@ -343,6 +343,65 @@ def finalize(self): with pytest.raises(RuntimeError, match=r"while ChipWorker\.init\(\) is in progress"): worker.finalize() + def test_public_wrapper_flush_failure_does_not_skip_finalize_cleanup(self, monkeypatch, capsys): + import simpler.task_interface as task_interface_mod # noqa: PLC0415 + from _task_interface import ChipCallable # noqa: PLC0415 + from simpler.task_interface import ChipWorker # noqa: PLC0415 # pyright: ignore[reportAttributeAccessIssue] + + finalized = [] + + class FakeImpl: + initialized = True + device_id = 0 + + def finalize(self): + finalized.append(True) + + def fail_flush(_timeout_ms): + raise RuntimeError("injected host-log flush failure") + + worker = ChipWorker() + worker._impl = FakeImpl() + worker._callable_registry[0] = ChipCallable.build(signature=[], func_name="test", binary=b"\x00", children=[]) + worker._identity_registry[b"digest"] = object() + worker._live_handles[1] = b"digest" + monkeypatch.setattr(task_interface_mod, "_flush_host_log", fail_flush) + + worker.finalize() + + assert finalized == [True] + assert worker._callable_registry == {} + assert worker._identity_registry == {} + assert worker._live_handles == {} + expected_warning = ( + "WARNING: host-log flush failed during ChipWorker.finalize(): injected host-log flush failure" + ) + assert expected_warning in capsys.readouterr().err + + def test_public_wrapper_flush_timeout_is_reported_with_loss_counters(self, monkeypatch, capsys): + import simpler.task_interface as task_interface_mod # noqa: PLC0415 + from simpler.task_interface import ChipWorker # noqa: PLC0415 # pyright: ignore[reportAttributeAccessIssue] + + class FakeImpl: + initialized = True + device_id = 0 + + def finalize(self): + pass + + worker = ChipWorker() + worker._impl = FakeImpl() + monkeypatch.setattr(task_interface_mod, "_flush_host_log", lambda _timeout_ms: False) + monkeypatch.setattr(task_interface_mod, "_host_log_pending_records", lambda: 7) + monkeypatch.setattr(task_interface_mod, "_host_log_dropped_records", lambda: 3) + + worker.finalize() + + warning = capsys.readouterr().err + assert "host-log flush timed out after 1000 ms during ChipWorker.finalize()" in warning + assert "pending_records=7, dropped_records=3" in warning + assert "accepted records may be lost" in warning + # ============================================================================ # Mailbox CallConfig wire round-trip diff --git a/tests/ut/py/test_worker/test_host_worker.py b/tests/ut/py/test_worker/test_host_worker.py index ba03153522..a5e51aa17e 100644 --- a/tests/ut/py/test_worker/test_host_worker.py +++ b/tests/ut/py/test_worker/test_host_worker.py @@ -588,9 +588,10 @@ def fake_fork() -> int: monkeypatch.setattr( worker_mod, "_initialize_host_log", - lambda level: startup_events.append(("log", level)), + lambda level, *, defer_writer=False: startup_events.append(("log", level, defer_writer)), raising=False, ) + monkeypatch.setattr(worker_mod, "_start_host_log_writer", lambda: startup_events.append(("writer",))) monkeypatch.setattr(worker, "_await_children_ready", fake_await_children_ready) monkeypatch.setattr(worker_mod, "Orchestrator", lambda native, owner: (native, owner)) try: @@ -603,8 +604,8 @@ def fake_fork() -> int: 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", 60) - assert startup_events[1:] == [("fork",), ("fork",)] + assert startup_events[0] == ("log", 60, True) + assert startup_events[1:] == [("fork",), ("fork",), ("writer",)] def test_start_hierarchical_seeds_the_logger_when_the_process_owns_no_chips(monkeypatch): @@ -653,9 +654,10 @@ def fake_fork() -> int: monkeypatch.setattr( worker_mod, "_initialize_host_log", - lambda level: startup_events.append(("log", level)), + lambda level, *, defer_writer=False: startup_events.append(("log", level, defer_writer)), raising=False, ) + monkeypatch.setattr(worker_mod, "_start_host_log_writer", lambda: startup_events.append(("writer",))) monkeypatch.setattr(worker, "_await_children_ready", lambda *args, **kwargs: None) monkeypatch.setattr(worker_mod, "Orchestrator", lambda native, owner: (native, owner)) try: @@ -665,8 +667,8 @@ def fake_fork() -> int: shm.close() shm.unlink() - assert startup_events[0] == ("log", 60) - assert startup_events[1:] == [("fork",)] + assert startup_events[0] == ("log", 60, True) + assert startup_events[1:] == [("fork",), ("writer",)] def test_a_worker_above_l3_can_never_carry_device_ids(): diff --git a/tests/ut/py/test_worker/test_startup_readiness.py b/tests/ut/py/test_worker/test_startup_readiness.py index 73dc1e7a4d..d5af67fefe 100644 --- a/tests/ut/py/test_worker/test_startup_readiness.py +++ b/tests/ut/py/test_worker/test_startup_readiness.py @@ -2071,3 +2071,23 @@ def boom(self): assert w._lifecycle is worker_mod._Lifecycle.FAILED assert w._sub_pids == [] w.close() + + def test_failed_hierarchical_start_restores_process_log_writer(self, monkeypatch): + import simpler.worker as worker_mod # noqa: PLC0415 + + events: list[str] = [] + original = RuntimeError("injected failure after logger quiesce") + + monkeypatch.setattr(Worker, "_init_hierarchical", lambda self: None) + monkeypatch.setattr(Worker, "_start_hierarchical", _raiser(original)) + monkeypatch.setattr(Worker, "_cleanup_partial_init", lambda self: events.append("cleanup")) + monkeypatch.setattr(worker_mod, "_start_host_log_writer", lambda: events.append("writer")) + + w = Worker(level=3, num_sub_workers=1) + with pytest.raises(RuntimeError) as raised: + w.init() + + assert raised.value is original + assert events == ["cleanup", "writer"] + assert w._lifecycle is worker_mod._Lifecycle.FAILED + w.close()