Fix: redirect host spans to per-process buffered files

docs/dfx/host-trace.md

@@ -24,6 +24,55 @@ constructing span attributes; `WARN`, `ERROR`, and `NUL` therefore disable the
 instrumentation work as well as the output. Device logging still uses the
 initialization-time policy described in [logging.md](../logging.md).
 
+## Where the records go
+
+The host logger writes to stderr until it is given a directory, and then it
+writes to `<directory>/host.<pid>.log` instead. The directory is
+`CallConfig.output_prefix` — the one every other diagnostic artifact already goes
+under — so a run that has one gets its host log beside its other artifacts, and a
+run that does not keeps its records on the console. There is no separate switch to
+configure, and the runtime never derives the path itself.
+
+**The destination belongs to the logger, not to a record.** Everything that logger
+writes follows it: `LOG_*` records, `[STRACE]` spans, `[CLOCK_ANCHOR]`, the
+`bind phase=` summaries, and the spans Python emits through
+`unified_log_host_span`. Nothing declares an intent and no record kind is treated
+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.
+
+**What stays on the console regardless:** messages Python logs through its own
+`logging` module. Those are a second logging system — same threshold, different
+envelope, no timestamp — so they are not part of this stream and cannot be
+ordered against it. Routing them through the host logger is a separate change.
+
+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.
+
+## Reading a run back
+
+Every record carries its own `pid`, so the tools take several inputs and
+concatenate them, and a directory expands to the `host.*.log` files inside it:
+
+```bash
+python -m simpler_setup.tools.strace_timing <output_prefix> --swimlane swimlane.json
+```
+
+A process writes its `[CLOCK_ANCHOR]` ahead of its first record and into the same
+stream, so each file is self-contained for wall-clock recovery. If an input is
+truncated or a process's file is missing, the pids in it stay monotonic-only and
+`clockAnchors` is absent from the output rather than wrong — the tool warns when a
+pid emitted spans and no anchor was found for it, which is the signal that this
+happened.
+
 ## Marker grammar
 
 Every host log record starts with a `CLOCK_MONOTONIC` nanosecond timestamp:

python/bindings/task_interface.cpp

@@ -1707,6 +1707,18 @@ NB_MODULE(_task_interface, m) {
         },
         nb::arg("level"), "Seed the process-owned host-log state before workers fork or load runtime modules."
     );
+    m.def(
+        "_set_host_log_directory",
+        [](const std::string &path) {
+            HostLogger::get_instance().set_log_directory(path.c_str());
+            const char *bound = HostLogger::get_instance().log_directory();
+            return bound == nullptr ? std::string() : std::string(bound);
+        },
+        nb::arg("path"),
+        "Write this process's host log to <path>/host.<pid>.log instead of stderr, and return the directory "
+        "actually in effect. Applies to every record this logger writes, including the host spans Python emits. "
+        "The first non-empty path in a process wins; an empty path leaves the logger on stderr."
+    );
     m.def(
         "_set_host_span_level_prefix",
         [](const std::string &word) {

python/simpler/worker.py

@@ -108,6 +108,9 @@ def my_l4_orch(orch, args, config):
 from _task_interface import (
     _host_spans_active as _native_host_spans_active,
 )
+from _task_interface import (
+    _set_host_log_directory as _native_set_host_log_directory,
+)
 
 from . import _log as _simpler_log
 from .buffer import (
@@ -3335,6 +3338,9 @@ def _read_config_from_mailbox(buf: memoryview) -> CallConfig:
     cfg.runtime_env.ring_dep_pool = ring_dep_pool
     # NUL-terminated C string in a 1024-byte field.
     cfg.output_prefix = prefix_bytes.split(b"\x00", 1)[0].decode("utf-8")
+    # A forked chip child owns its own log file under the same directory.
+    if cfg.output_prefix:
+        _native_set_host_log_directory(cfg.output_prefix)
     return cfg
 
 
@@ -10801,6 +10807,13 @@ def _chip_run_for(self, run_id: int) -> Any | None:
     def _submit_l3_locked(self, callable, args, cfg: CallConfig) -> RunHandle:
         assert self._orch is not None
         assert self._worker is not None
+        # This process's log belongs beside the run's other diagnostic artifacts,
+        # so the directory comes from the config that already names it. First one
+        # in a process wins; with no prefix the logger stays on stderr. Read
+        # defensively: wiring an output must never be what fails a submit.
+        log_directory = getattr(cfg, "output_prefix", "")
+        if log_directory:
+            _native_set_host_log_directory(log_directory)
         run_id = self._orch._begin_run()
         resources = _RunResources()
         handle = RunHandle(self, run_id, (callable, args, cfg), resources)

simpler_setup/tools/strace_timing.py

@@ -51,6 +51,7 @@
 
 import argparse
 import json
+import pathlib
 import re
 import sys
 from collections import defaultdict
@@ -95,6 +96,28 @@
 # The emitter percent-encodes any byte that would otherwise be record grammar —
 # see `encode_host_span_field` in src/common/log/host_log.cpp.
 _PERCENT_ESCAPE_RE = re.compile(r"%([0-9A-Fa-f]{2})")
+# One file per process, written under a run's `output_prefix` when the host logger
+# writes to files rather than stderr. It holds everything that logger emits, so a
+# process's spans and its `[CLOCK_ANCHOR]` are in the same file. See
+# docs/dfx/host-trace.md.
+_LOG_FILE_GLOB = "host.*.log"
+
+
+def _expand_log_source(source):
+    """Resolve one CLI input to the files to read.
+
+    A directory expands to its per-process log files, so a run's
+    ``output_prefix`` can be passed as-is instead of being globbed by the caller.
+    Sorted, because a reader comparing two runs should not have to care that the
+    shell and the filesystem disagree about order.
+    """
+    path = pathlib.Path(source)
+    if not path.is_dir():
+        return [path]
+    log_files = sorted(path.glob(_LOG_FILE_GLOB))
+    if not log_files:
+        raise SystemExit(f"{source} is a directory but holds no {_LOG_FILE_GLOB} files")
+    return log_files
 
 
 def decode_field(text):
@@ -1359,7 +1382,13 @@ def write_host_swimlane(args, spans, lines, anchors):
 
 def main(argv=None):
     ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
-    ap.add_argument("log", help="path to a host/CANN log containing [STRACE] lines (or '-' for stdin)")
+    ap.add_argument(
+        "log",
+        nargs="+",
+        help="one or more host/CANN logs containing [STRACE] lines, '-' for stdin, or a directory holding "
+        f"{_LOG_FILE_GLOB} files (a run's output_prefix). Several inputs are concatenated: records carry their "
+        "own pid, so a whole run's per-process logs, or logs from several runs, can be passed together.",
+    )
     ap.add_argument(
         "--trace-out", help="write a Chrome-trace/Perfetto JSON here (load in chrome://tracing or perfetto)"
     )
@@ -1401,11 +1430,14 @@ def main(argv=None):
     )
     args = ap.parse_args(argv)
 
-    if args.log == "-":
-        lines = sys.stdin.readlines()
-    else:
-        with open(args.log, encoding="utf-8", errors="replace") as f:
-            lines = f.readlines()
+    lines = []
+    for source in args.log:
+        if source == "-":
+            lines.extend(sys.stdin.readlines())
+            continue
+        for path in _expand_log_source(source):
+            with open(path, encoding="utf-8", errors="replace") as f:
+                lines.extend(f.readlines())
 
     spans = list(parse_spans(lines))
     anchors = list(parse_clock_anchors(lines))
@@ -1418,6 +1450,18 @@ def main(argv=None):
                 f"warning: multiple [CLOCK_ANCHOR] records found for pid {pid} ({count} records); using the last one",
                 file=sys.stderr,
             )
+    # Without an anchor a pid's records stay monotonic-only, and every renderer
+    # degrades to relative time without saying so. A process writes its anchor
+    # ahead of its first record, into whichever stream it is logging to, so a pid
+    # with spans and no anchor means that stream reached us incomplete.
+    unanchored = sorted({span.pid for span in spans} - set(anchor_counts))
+    if unanchored:
+        print(
+            f"warning: no [CLOCK_ANCHOR] record for pid(s) {', '.join(str(pid) for pid in unanchored)} that emitted "
+            "spans; their timestamps stay monotonic-only. Each process writes its anchor before its first record, so "
+            "check that every input is complete and that no process's stream is missing.",
+            file=sys.stderr,
+        )
     heads = count_record_heads(lines)
     if heads > len(spans):
         print(

src/common/log/host_log.cpp

@@ -153,6 +153,87 @@ long host_trace_tid() {
 #endif
 }
 
+struct HostLogFileSink {
+    HostLogFileSink();
+
+    std::mutex mutex;
+    FILE *stream = nullptr;
+    pid_t pid = -1;
+    std::string directory;
+};
+
+HostLogFileSink &host_log_file_sink() {
+    static HostLogFileSink sink;
+    return sink;
+}
+
+void host_log_sink_before_fork() { host_log_file_sink().mutex.lock(); }
+
+void host_log_sink_after_fork() { host_log_file_sink().mutex.unlock(); }
+
+HostLogFileSink::HostLogFileSink() {
+    // A child inherits only the thread that called fork(). Locking here makes
+    // that thread the mutex owner at the fork boundary, so both the parent and
+    // child handlers can release it without inheriting an owner that vanished.
+    (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) {
+    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;
+    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;
+        sink.pid = -1;
+        sink.directory.clear();
+    } else if (changed_directory) {
+        std::fclose(sink.stream);
+        sink.stream = nullptr;
+        sink.pid = -1;
+        sink.directory.clear();
+    }
+    if (sink.stream == nullptr) {
+        char path[PATH_MAX];
+        const int length = std::snprintf(path, sizeof(path), "%s/host.%d.log", directory, pid);
+        if (length <= 0 || static_cast<size_t>(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);
+        sink.pid = pid;
+        sink.directory = directory;
+    }
+    if (std::fwrite(record, 1, size, sink.stream) != size) return false;
+    return !flush || std::fflush(sink.stream) == 0;
+}
+
+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;
+}
+
 }  // namespace
 
 namespace {
@@ -236,7 +317,7 @@ 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 HostLogger::emit(const char *level_tag, const char *func, const char *fmt, va_list args, bool flush) {
     const int64_t monotonic_ns = simpler::log::monotonic_now_ns();
     auto tid = static_cast<unsigned long>(reinterpret_cast<uintptr_t>(pthread_self()));
 
@@ -252,22 +333,32 @@ bool HostLogger::emit(const char *level_tag, const char *func, const char *fmt,
         stack_buffer, sizeof(stack_buffer), monotonic_ns, tid, level_tag, func, fmt, args, append_newline
     );
     if (length < sizeof(stack_buffer)) {
-        std::scoped_lock lock(mutex_);
-        return write_stderr(stack_buffer, length);
+        return write_record(stack_buffer, length, flush);
     }
 
     std::vector<char> 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(heap_buffer.data(), heap_length < heap_buffer.size() ? heap_length : length);
+    return write_stderr(record, size);
 }
 
 bool HostLogger::emit_ungated(const char *level_tag, const char *func, const char *fmt, ...) {
     va_list args;
     va_start(args, fmt);
-    const bool written = emit(level_tag, func, fmt, args);
+    // 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);
     va_end(args);
     return written;
 }
@@ -301,7 +392,9 @@ void HostLogger::vlog(LogLevel level, const char *func, const char *fmt, va_list
         return;
     }
     emit_clock_anchor_if_needed();
-    emit(level_name(level), func, fmt, args);
+    // 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 HostLogger::log(LogLevel level, const char *func, const char *fmt, ...) {
@@ -311,6 +404,27 @@ void HostLogger::log(LogLevel level, const char *func, const char *fmt, ...) {
     va_end(args);
 }
 
+void HostLogger::set_log_directory(const char *path) {
+    if (path == nullptr || path[0] == '\0') return;
+    SimplerHostLogState *shared = state();
+    if (shared == nullptr) return;
+    // Claim 0 -> -1, fill, publish -1 -> 1, mirroring the anchor claim above. A
+    // reader accepts only 1, so it never observes a half-written path; and a
+    // later caller with a different path is refused rather than moving a file
+    // some thread may already hold open.
+    int32_t unclaimed = 0;
+    if (!atomic_compare_exchange_i32(&shared->log_directory_bound, &unclaimed, -1)) return;
+    (void)std::snprintf(shared->log_directory, sizeof(shared->log_directory), "%s", path);
+    int32_t claim = -1;
+    (void)atomic_compare_exchange_i32(&shared->log_directory_bound, &claim, 1);
+}
+
+const char *HostLogger::log_directory() const {
+    SimplerHostLogState *shared = state();
+    if (shared == nullptr || atomic_load_i32(&shared->log_directory_bound) != 1) return nullptr;
+    return shared->log_directory;
+}
+
 void HostLogger::log_host_span(const SimplerHostSpan *span) {
     if (!is_enabled(LogLevel::TIMING)) return;
     if (span == nullptr || span->abi_version != SIMPLER_HOST_SPAN_ABI_VERSION ||
@@ -320,11 +434,23 @@ void HostLogger::log_host_span(const SimplerHostSpan *span) {
     const std::string name = encode_host_span_field(span->name, kHostSpanNameCapacity, false);
     const std::string attributes =
         encode_host_span_field(span->attributes == nullptr ? "" : span->attributes, kHostSpanAttributesCapacity, true);
-    log(LogLevel::TIMING, "emit_host_span",
+
+    // One record grammar, in one place. Where it lands is the logger's business,
+    // not this emitter's.
+    char record[kRecordStackCapacity];
+    (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
         " %s",
         static_cast<int>(getpid()), host_trace_tid(), span->invocation_id, span->callable_hash, span->depth,
-        name.c_str(), span->timestamp_ns, span->duration_ns, attributes.c_str());
+        name.c_str(), span->timestamp_ns, span->duration_ns, attributes.c_str()
+    );
+
+    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) {