Fix: keep forked host log records intact

simpler_setup/tools/strace_timing.py

@@ -48,11 +48,26 @@
 from collections import defaultdict
 from dataclasses import dataclass, field
 
+# The host-log record prefix `HostLogger::emit` writes ahead of every message.
+# Its func segment excludes ':' because `LOG_TIMING` passes `__FUNCTION__`, an
+# unqualified name; a qualified name containing '::' would stop this alternative
+# from matching, leaving the `[STRACE]` alternative to bound the record instead.
+_HOST_LOG_PREFIX = (
+    r"\[\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{6}\]"
+    r"\[T0x[0-9a-fA-F]+\]\[[A-Z]+\]\s+[^:\r\n]+:\s+"
+)
+# MULTILINE anchors the `$` alternative at every line end, so a caller passing a
+# multi-line blob rather than one line per item keeps every record but the last.
 _STRACE_RE = re.compile(
     r"\[STRACE\]\s+v=(?P<v>\d+)\s+pid=(?P<pid>\d+)\s+tid=(?P<tid>\d+)\s+"
     r"inv=(?P<inv>\d+)\s+hid=(?P<hid>[0-9a-fA-F]+)\s+depth=(?P<depth>\d+)\s+"
-    r"name=(?P<name>\S+)\s+ts=(?P<ts>\d+)\s+dur=(?P<dur>\d+)(?P<attrs>.*)"
+    r"name=(?P<name>\S+)\s+ts=(?P<ts>\d+)\s+dur=(?P<dur>\d+)(?P<attrs>.*?)"
+    rf"(?={_HOST_LOG_PREFIX}|\[STRACE\]|\r?$)",
+    re.MULTILINE,
 )
+# A record start, matched independently of whether the rest of that record
+# survived the write that emitted it.
+_STRACE_HEAD_RE = re.compile(r"\[STRACE\]\s+v=\d+")
 
 
 @dataclass
@@ -95,23 +110,31 @@ def by_name(self):
         return m
 
 
+def count_record_heads(lines):
+    """Count ``[STRACE]`` record starts, torn ones included.
+
+    Pairs with :func:`parse_spans`, which yields only records that survived
+    intact. A shortfall between the two counts is instrumentation loss, and
+    without it a torn record is indistinguishable from a real measurement.
+    """
+    return sum(len(_STRACE_HEAD_RE.findall(line)) for line in lines)
+
+
 def parse_spans(lines):
-    """Yield Span for every matching [STRACE] line."""
+    """Yield every complete span, including adjacent records on one line."""
     for line in lines:
-        m = _STRACE_RE.search(line)
-        if not m:
-            continue
-        yield Span(
-            pid=int(m["pid"]),
-            tid=int(m["tid"]),
-            inv=int(m["inv"]),
-            hid=m["hid"].lower(),
-            depth=int(m["depth"]),
-            name=m["name"],
-            ts=int(m["ts"]),
-            dur=int(m["dur"]),
-            attrs=m["attrs"].strip(),
-        )
+        for m in _STRACE_RE.finditer(line):
+            yield Span(
+                pid=int(m["pid"]),
+                tid=int(m["tid"]),
+                inv=int(m["inv"]),
+                hid=m["hid"].lower(),
+                depth=int(m["depth"]),
+                name=m["name"],
+                ts=int(m["ts"]),
+                dur=int(m["dur"]),
+                attrs=m["attrs"].strip(),
+            )
 
 
 def group_invocations(spans):
@@ -493,6 +516,13 @@ def main(argv=None):
             lines = f.readlines()
 
     spans = list(parse_spans(lines))
+    heads = count_record_heads(lines)
+    if heads > len(spans):
+        print(
+            f"warning: {heads - len(spans)} of {heads} [STRACE] records are incomplete and are "
+            "excluded from the timing below",
+            file=sys.stderr,
+        )
     invocations = group_invocations(spans)
     buckets = bucket_by_hid(invocations)
 

src/common/log/host_log.cpp

@@ -15,15 +15,81 @@
 
 #include "host_log.h"
 
+#include <cerrno>
 #include <chrono>
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
 #include <ctime>
 #include <pthread.h>
+#include <vector>
+
+#include <unistd.h>
 
 using simpler::log::LogLevel;
 
+namespace {
+
+// Every STRACE marker renders well inside this, so the heap fallback below
+// stays off the path a traced run pays per span.
+constexpr size_t kRecordStackCapacity = 2048;
+
+// 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.
+size_t format_record(
+    char *buffer, size_t capacity, const char *ts, unsigned long tid, const char *level_tag, const char *func,
+    const char *fmt, va_list args, bool append_newline
+) {
+    size_t length = 0;
+    auto tail = [&]() -> char * {
+        return length < capacity ? buffer + length : nullptr;
+    };
+    auto remaining = [&]() -> size_t {
+        return length < capacity ? capacity - length : 0;
+    };
+
+    const int prefix = snprintf(tail(), remaining(), "[%s][T0x%lx][%s] %s: ", ts, tid, level_tag, func);
+    if (prefix < 0) {
+        return 0;
+    }
+    length += static_cast<size_t>(prefix);
+
+    va_list formatting_args;
+    va_copy(formatting_args, args);
+    const int body = vsnprintf(tail(), remaining(), fmt, formatting_args);
+    va_end(formatting_args);
+    if (body > 0) {
+        length += static_cast<size_t>(body);
+    }
+
+    if (append_newline) {
+        if (length + 1 < capacity) {
+            buffer[length] = '\n';
+            buffer[length + 1] = '\0';
+        }
+        length += 1;
+    }
+    return length;
+}
+
+void write_stderr(const char *record, size_t size) {
+    size_t offset = 0;
+    while (offset < size) {
+        const ssize_t written = ::write(STDERR_FILENO, record + offset, size - offset);
+        if (written > 0) {
+            offset += static_cast<size_t>(written);
+        } else if (written < 0 && errno == EINTR) {
+            continue;
+        } else {
+            break;
+        }
+    }
+}
+
+}  // namespace
+
 HostLogger &HostLogger::get_instance() {
     static HostLogger instance;
     return instance;
@@ -83,13 +149,26 @@ void HostLogger::emit(const char *level_tag, const char *func, const char *fmt,
 
     auto tid = static_cast<unsigned long>(reinterpret_cast<uintptr_t>(pthread_self()));
 
-    std::scoped_lock lock(mutex_);
-    fprintf(stderr, "[%s][T0x%lx][%s] %s: ", ts, tid, level_tag, func);
-    vfprintf(stderr, fmt, args);
-    if (fmt[0] != '\0' && fmt[strlen(fmt) - 1] != '\n') {
-        fputc('\n', stderr);
+    const bool append_newline = fmt[0] != '\0' && fmt[strlen(fmt) - 1] != '\n';
+
+    // One write per record: a record of at most PIPE_BUF bytes reaches a shared
+    // pipe indivisibly, so forked workers writing a captured stderr cannot
+    // interleave inside it. A longer record — a large LOG_ERROR dump, say — has
+    // no such guarantee; STRACE markers stay well below the limit.
+    char stack_buffer[kRecordStackCapacity];
+    const size_t length =
+        format_record(stack_buffer, sizeof(stack_buffer), ts, tid, level_tag, func, fmt, args, append_newline);
+    if (length < sizeof(stack_buffer)) {
+        std::scoped_lock lock(mutex_);
+        write_stderr(stack_buffer, length);
+        return;
     }
-    fflush(stderr);
+
+    std::vector<char> heap_buffer(length + 1);
+    const size_t heap_length =
+        format_record(heap_buffer.data(), heap_buffer.size(), ts, tid, level_tag, func, fmt, args, append_newline);
+    std::scoped_lock lock(mutex_);
+    write_stderr(heap_buffer.data(), heap_length < heap_buffer.size() ? heap_length : length);
 }
 
 void HostLogger::vlog(LogLevel level, const char *func, const char *fmt, va_list args) {

src/common/log/include/host_log.h

@@ -17,8 +17,7 @@
  * both TIMING and WARN thresholds to CANN WARN.
  */
 
-#ifndef PLATFORM_HOST_LOG_H_
-#define PLATFORM_HOST_LOG_H_
+#pragma once
 
 #include <cstdarg>
 #include <cstdio>
@@ -32,9 +31,8 @@ class HostLogger {
 
     void log(simpler::log::LogLevel level, const char *func, const char *fmt, ...);
 
-    // va_list-taking primitives — used by unified_log_* adapters to forward
-    // a caller's variadic args without an intermediate vsnprintf-to-buffer
-    // round-trip. Caller is responsible for `va_start` / `va_end`.
+    // va_list-taking primitive used by unified_log_* adapters. Caller is
+    // 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);
@@ -66,5 +64,3 @@ class HostLogger {
     simpler::log::LogLevel current_level_;
     std::mutex mutex_;
 };
-
-#endif  // PLATFORM_HOST_LOG_H_

tests/ut/cpp/a5/test_host_log_off.cpp

@@ -15,7 +15,13 @@
 
 #include <cstdio>
 #include <cstdlib>
+#include <algorithm>
+#include <cerrno>
+#include <sstream>
 #include <string>
+#include <thread>
+#include <vector>
+#include <sys/wait.h>
 #include <unistd.h>
 
 #include <gtest/gtest.h>
@@ -198,3 +204,108 @@ TEST(HostLogTest, AllOutputGoesToStderr) {
     EXPECT_NE(captured.err.find("info-output-marker"), std::string::npos);
     EXPECT_NE(captured.err.find("debug-output-marker"), std::string::npos);
 }
+
+TEST(HostLogTest, ForkedProcessesEmitWholePipeRecords) {
+    int log_pipe[2];
+    int start_pipe[2];
+    ASSERT_EQ(pipe(log_pipe), 0);
+    ASSERT_EQ(pipe(start_pipe), 0);
+
+    const long pipe_buf = fpathconf(log_pipe[1], _PC_PIPE_BUF);
+    ASSERT_GT(pipe_buf, 256);
+    const size_t payload_size = static_cast<size_t>(std::min<long>(pipe_buf - 256, 2048));
+    constexpr int child_count = 16;
+    constexpr int records_per_child = 128;
+
+    std::vector<pid_t> children;
+    for (int child = 0; child < child_count; ++child) {
+        const pid_t pid = fork();
+        ASSERT_GE(pid, 0);
+        if (pid == 0) {
+            close(log_pipe[0]);
+            close(start_pipe[1]);
+            char start;
+            if (read(start_pipe[0], &start, 1) != 1 || dup2(log_pipe[1], STDERR_FILENO) < 0) {
+                _exit(2);
+            }
+            close(start_pipe[0]);
+            close(log_pipe[1]);
+
+            HostLogger::get_instance().set_level(LogLevel::DEBUG);
+            const std::string payload(payload_size, static_cast<char>('a' + child));
+            for (int seq = 0; seq < records_per_child; ++seq) {
+                HostLogger::get_instance().log(
+                    LogLevel::ERROR, "fork_writer", "child=%d seq=%d payload=%s", child, seq, payload.c_str()
+                );
+            }
+            _exit(0);
+        }
+        children.push_back(pid);
+    }
+
+    close(log_pipe[1]);
+    close(start_pipe[0]);
+    std::string captured;
+    std::thread reader([&] {
+        char buffer[8192];
+        while (true) {
+            const ssize_t count = read(log_pipe[0], buffer, sizeof(buffer));
+            if (count > 0) {
+                captured.append(buffer, static_cast<size_t>(count));
+            } else if (count < 0 && errno == EINTR) {
+                continue;
+            } else {
+                break;
+            }
+        }
+        close(log_pipe[0]);
+    });
+
+    const std::string starts(child_count, 'x');
+    EXPECT_EQ(write(start_pipe[1], starts.data(), starts.size()), static_cast<ssize_t>(starts.size()));
+    close(start_pipe[1]);
+
+    for (pid_t child : children) {
+        int status = 0;
+        const pid_t waited = waitpid(child, &status, 0);
+        EXPECT_EQ(waited, child);
+        if (waited == child) {
+            EXPECT_TRUE(WIFEXITED(status));
+            if (WIFEXITED(status)) {
+                EXPECT_EQ(WEXITSTATUS(status), 0);
+            }
+        }
+    }
+    reader.join();
+
+    std::vector<std::vector<bool>> seen(child_count, std::vector<bool>(records_per_child, false));
+    std::istringstream lines(captured);
+    std::string line;
+    int line_count = 0;
+    constexpr char payload_marker[] = " payload=";
+    while (std::getline(lines, line)) {
+        const size_t record_pos = line.find("child=");
+        const size_t payload_pos = line.find(payload_marker);
+        ASSERT_NE(record_pos, std::string::npos);
+        ASSERT_NE(payload_pos, std::string::npos);
+        ASSERT_EQ(line.find("child=", record_pos + 1), std::string::npos);
+
+        int child = -1;
+        int seq = -1;
+        ASSERT_EQ(sscanf(line.c_str() + record_pos, "child=%d seq=%d", &child, &seq), 2);
+        ASSERT_GE(child, 0);
+        ASSERT_LT(child, child_count);
+        ASSERT_GE(seq, 0);
+        ASSERT_LT(seq, records_per_child);
+        ASSERT_FALSE(seen[child][seq]);
+        seen[child][seq] = true;
+
+        const std::string payload = line.substr(payload_pos + sizeof(payload_marker) - 1);
+        ASSERT_EQ(payload.size(), payload_size);
+        EXPECT_TRUE(std::all_of(payload.begin(), payload.end(), [child](char value) {
+            return value == static_cast<char>('a' + child);
+        }));
+        ++line_count;
+    }
+    EXPECT_EQ(line_count, child_count * records_per_child);
+}