Fix: emit each sim device-log record atomically (#1697)

docs/logging.md

@@ -61,7 +61,7 @@ src/common/platform/                         ← shared device-side log
 ├── include/aicpu/device_log.h               low-level dev_vlog_* declarations
 ├── shared/aicpu/unified_log_device.cpp      C ABI → dev_vlog_* adapter
 ├── onboard/aicpu/device_log.cpp             onboard backend (CANN dlog)
-└── sim/aicpu/device_log.cpp                 sim backend (fprintf to stderr)
+└── sim/aicpu/device_log.cpp                 sim backend (one write(2) to stderr)
 ```
 
 Both architectures link these shared implementations into their platform
@@ -139,9 +139,12 @@ void dev_vlog_error (const char *func, const char *fmt, va_list);
 
 `unified_log_device.cpp` forwards the caller's `va_list` directly into
 `dev_vlog_*` — no intermediate `vsnprintf`-to-buffer round-trip in this
-layer. The sim backend is buffer-free (single `vfprintf(stderr, ...)`); the
-onboard backend still buffers internally because CANN's `dlog` is variadic
-only (no `va_list` variant).
+layer. Each backend then formats one whole record and emits it in a single
+call: the sim backend writes it with one `write(2)`, kept under `PIPE_BUF` so
+concurrent AICPU sim threads or forked chip workers sharing `stderr` never
+interleave partial records; the onboard backend buffers into a stack `char`
+array and issues one `dlog_*` because CANN's `dlog` is variadic only (no
+`va_list` variant).
 
 ## Multi-`.so` singleton
 

src/common/platform/include/aicpu/device_log.h

@@ -67,9 +67,11 @@ extern "C" void set_log_level(int level);
 // Platform-specific logging functions (low-level layer)
 //
 // va_list primitives used by the unified_log_* adapter to forward a caller's
-// variadic args without an intermediate vsnprintf-to-buffer round-trip. Sim
-// is buffer-free; onboard still buffers internally because CANN's dlog API
-// has no va_list variant. Caller owns va_start/va_end.
+// 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.
 // =============================================================================
 
 #include <cstdarg>

src/common/platform/sim/aicpu/device_log.cpp

@@ -18,8 +18,11 @@
 
 #include "aicpu/device_log.h"
 
+#include <cerrno>
 #include <cstdarg>
+#include <cstddef>
 #include <cstdio>
+#include <unistd.h>
 
 // =============================================================================
 // Level enable flags (mutated by the setter below)
@@ -55,35 +58,61 @@ void init_log_switch() {
 }
 
 // =============================================================================
-// Low-level dev_log_* / dev_vlog_* (sim: fprintf to stderr; no buffer needed)
+// 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.
 // =============================================================================
 
-void dev_vlog_debug(const char *func, const char *fmt, va_list args) {
-    fprintf(stderr, "[DEBUG] %s: ", func);
-    vfprintf(stderr, fmt, args);
-    fputc('\n', stderr);
-}
+namespace {
 
-void dev_vlog_info(const char *func, const char *fmt, va_list args) {
-    fprintf(stderr, "[INFO] %s: ", func);
-    vfprintf(stderr, fmt, args);
-    fputc('\n', stderr);
-}
+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;
 
-void dev_vlog_timing(const char *func, const char *fmt, va_list args) {
-    fprintf(stderr, "[TIMING] %s: ", func);
-    vfprintf(stderr, fmt, args);
-    fputc('\n', stderr);
-}
+    int prefix = snprintf(buffer, sizeof(buffer), "[%s] %s: ", level_tag, func);
+    size_t len = (prefix < 0) ? 0 : static_cast<size_t>(prefix);
+    if (len > kBodyLimit) {
+        len = kBodyLimit;  // prefix filled the buffer; reserve the newline slot
+    }
 
-void dev_vlog_warn(const char *func, const char *fmt, va_list args) {
-    fprintf(stderr, "[WARN] %s: ", func);
-    vfprintf(stderr, fmt, args);
-    fputc('\n', stderr);
-}
+    int body = vsnprintf(buffer + len, sizeof(buffer) - len, fmt, args);
+    if (body > 0) {
+        len += static_cast<size_t>(body);
+        if (len > kBodyLimit) {
+            len = kBodyLimit;  // body truncated; reserve the newline slot
+        }
+    }
 
-void dev_vlog_error(const char *func, const char *fmt, va_list args) {
-    fprintf(stderr, "[ERROR] %s: ", func);
-    vfprintf(stderr, fmt, args);
-    fputc('\n', stderr);
+    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<size_t>(written);
+    }
 }
+
+}  // 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_timing(const char *func, const char *fmt, va_list args) { emit_record("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_error(const char *func, const char *fmt, va_list args) { emit_record("ERROR", func, fmt, args); }

tests/ut/cpp/CMakeLists.txt

@@ -835,6 +835,25 @@ 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.
+add_executable(test_sim_device_log
+    common/test_sim_device_log.cpp
+    ${COMMON_PLATFORM_DIR}/sim/aicpu/device_log.cpp
+)
+target_include_directories(test_sim_device_log PRIVATE
+    ${GTEST_INCLUDE_DIRS}
+    ${COMMON_PLATFORM_DIR}/include
+)
+target_link_libraries(test_sim_device_log PRIVATE
+    ${GTEST_MAIN_LIB}
+    ${GTEST_LIB}
+    pthread
+)
+add_test(NAME test_sim_device_log COMMAND test_sim_device_log)
+set_tests_properties(test_sim_device_log PROPERTIES LABELS "no_hardware")
+
 # a2a3 host-side AICPU affinity selection (compute_allowed_cpus). Pure logic —
 # no CANN headers, no hardware: compile the probe .cpp alongside the test. The
 # test provides its own no-op logger stubs, so no logger sources are linked.

tests/ut/cpp/common/test_sim_device_log.cpp

@@ -0,0 +1,186 @@
+/*
+ * 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.
+ * -----------------------------------------------------------------------------------------------------------
+ */
+
+// 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. Each record is <= PIPE_BUF, so a single
+// write(2) to a pipe is atomic and cannot interleave with another writer.
+
+#include <cstdarg>
+#include <cstdio>
+#include <set>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include <sys/wait.h>
+#include <unistd.h>
+
+#include <gtest/gtest.h>
+
+#include "aicpu/device_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.
+void emit(int level_idx, const char *func, const char *fmt, ...) {
+    va_list ap;
+    va_start(ap, fmt);
+    switch (level_idx % 5) {
+    case 0:
+        dev_vlog_debug(func, fmt, ap);
+        break;
+    case 1:
+        dev_vlog_info(func, fmt, ap);
+        break;
+    case 2:
+        dev_vlog_timing(func, fmt, ap);
+        break;
+    case 3:
+        dev_vlog_warn(func, fmt, ap);
+        break;
+    default:
+        dev_vlog_error(func, fmt, ap);
+        break;
+    }
+    va_end(ap);
+}
+
+std::string record(int level_idx, const char *func, const std::string &body) {
+    return std::string("[") + kTags[level_idx % 5] + "] " + func + ": " + body;
+}
+
+// Redirect stderr onto a fresh pipe. All writers stay well under the pipe's
+// 64 KiB capacity, so they never block before the reader drains.
+struct Capture {
+    int read_fd = -1;
+    int saved_stderr = -1;
+};
+
+Capture begin_capture() {
+    int fds[2];
+    EXPECT_EQ(pipe(fds), 0);
+    fflush(stderr);
+    Capture cap;
+    cap.saved_stderr = dup(STDERR_FILENO);
+    EXPECT_GE(cap.saved_stderr, 0);
+    EXPECT_GE(dup2(fds[1], STDERR_FILENO), 0);
+    close(fds[1]);  // STDERR_FILENO is now the sole write handle
+    cap.read_fd = fds[0];
+    return cap;
+}
+
+// Restore stderr (closing the last write handle so read hits EOF) and slurp
+// everything the pipe holds.
+std::string end_capture(Capture &cap) {
+    fflush(stderr);
+    EXPECT_GE(dup2(cap.saved_stderr, STDERR_FILENO), 0);
+    close(cap.saved_stderr);
+
+    std::string out;
+    char buf[4096];
+    ssize_t n;
+    while ((n = read(cap.read_fd, buf, sizeof(buf))) > 0) {
+        out.append(buf, static_cast<size_t>(n));
+    }
+    close(cap.read_fd);
+    return out;
+}
+
+// Assert the captured stream is exactly `expected`, one intact record per
+// physical line — a torn (merged or split) record matches no expected string.
+void expect_intact(const std::string &captured, std::multiset<std::string> expected) {
+    size_t start = 0;
+    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);
+        ASSERT_NE(it, expected.end()) << "torn or unexpected record: '" << line << "'";
+        expected.erase(it);
+        start = nl + 1;
+    }
+    EXPECT_TRUE(expected.empty()) << expected.size() << " records never arrived intact";
+}
+
+}  // namespace
+
+TEST(SimDeviceLogTest, MultiThreadedRecordsStayIntact) {
+    constexpr int kThreads = 4;
+    constexpr int kPerThread = 200;
+
+    std::multiset<std::string> expected;
+    for (int t = 0; t < kThreads; ++t) {
+        for (int i = 0; i < kPerThread; ++i) {
+            char body[32];
+            snprintf(body, sizeof(body), "t%d-r%04d", t, i);
+            expected.insert(record(t, "worker", body));
+        }
+    }
+
+    Capture cap = begin_capture();
+    std::vector<std::thread> threads;
+    threads.reserve(kThreads);
+    for (int t = 0; t < kThreads; ++t) {
+        threads.emplace_back([t] {
+            for (int i = 0; i < kPerThread; ++i) {
+                emit(t, "worker", "t%d-r%04d", t, i);
+            }
+        });
+    }
+    for (auto &th : threads) {
+        th.join();
+    }
+    std::string captured = end_capture(cap);
+
+    ASSERT_NO_FATAL_FAILURE(expect_intact(captured, std::move(expected)));
+}
+
+TEST(SimDeviceLogTest, ForkedProcessesEmitWholeRecords) {
+    constexpr int kChildren = 8;
+    constexpr int kPerChild = 100;
+
+    std::multiset<std::string> expected;
+    for (int c = 0; c < kChildren; ++c) {
+        for (int i = 0; i < kPerChild; ++i) {
+            char body[32];
+            snprintf(body, sizeof(body), "c%d-r%03d", c, i);
+            expected.insert(record(c, "chip_worker", body));
+        }
+    }
+
+    Capture cap = begin_capture();
+    std::vector<pid_t> pids;
+    pids.reserve(kChildren);
+    for (int c = 0; c < kChildren; ++c) {
+        pid_t pid = fork();
+        ASSERT_GE(pid, 0);
+        if (pid == 0) {
+            for (int i = 0; i < kPerChild; ++i) {
+                emit(c, "chip_worker", "c%d-r%03d", c, i);
+            }
+            _exit(0);  // skip gtest/atexit teardown so nothing else hits the pipe
+        }
+        pids.push_back(pid);
+    }
+    for (pid_t pid : pids) {
+        int status = 0;
+        ASSERT_EQ(waitpid(pid, &status, 0), pid);
+        ASSERT_TRUE(WIFEXITED(status)) << "child terminated abnormally";
+        EXPECT_EQ(WEXITSTATUS(status), 0);
+    }
+    std::string captured = end_capture(cap);
+
+    ASSERT_NO_FATAL_FAILURE(expect_intact(captured, std::move(expected)));
+}