From d5e3b9c301149f1d2408aaf46fd1bdabbfaeaf47 Mon Sep 17 00:00:00 2001 From: Jonathan Bernard Date: Mon, 28 Sep 2026 15:00:18 -0500 Subject: [PATCH] Drain background writers before process exit Serialize writer submission and shutdown, enqueue explicit stop messages after accepted records, and finish the final file batch before joining. Keep drained process-wide channel storage until exit to avoid the Nim 2.2.10 Channel.close mutex teardown defect. Cover empty records, multiple destinations, joined producers, and concurrent shutdown. Include writer lifecycle sanitizer tests in nimble test and update usage documentation. Validation: all 32 unit tests, eight writer exit scenarios, and writer lifecycle ThreadSanitizer tests pass. The full suite retains the known configuration refresh, timestamp, and default allocator failures. --- README.md | 11 +- namespaced_logging.nimble | 6 +- src/namespaced_logging.nim | 213 +++++++++++++++++++++---------------- tests/README.md | 42 +++++--- tests/regressions.nim | 31 ++++-- tests/run_tsan.sh | 10 ++ tests/writer_child.nim | 53 +++++++-- 7 files changed, 241 insertions(+), 125 deletions(-) diff --git a/README.md b/README.md index 15bdeae..080ec60 100644 --- a/README.md +++ b/README.md @@ -303,6 +303,10 @@ remains available while any service copy, local service, or logger using that service remains alive. Join worker threads before exiting the application so they have an opportunity to finish logging. +When background writers are enabled with `-d:multithreaded`, normal application +exit drains the console and file queues and waits for the writers to finish. +No explicit flush or delay is needed after joining the threads that log. + You can change logging configuration through a service copy or a local service in any thread. Those changes are intended to take effect automatically on subsequent logging calls in every thread using that service, without recreating @@ -311,10 +315,9 @@ loggers or configuring each worker separately. > [!NOTE] > Thread-safety work is still in progress. Custom state and captured > formatters must follow the ownership rules above. Runtime configuration -> refresh, concurrent local-time timestamp construction, and delivery of queued -> background output at application exit still have known failures. Nim 2.2.10 -> also produces sanitizer reports in its default allocator during some -> cross-thread allocation patterns. See +> refresh and concurrent local-time timestamp construction still have known +> failures. Nim 2.2.10 also produces sanitizer reports in its default allocator +> during some cross-thread allocation patterns. See > [the regression tests](tests/README.md) for the remaining work. diff --git a/namespaced_logging.nimble b/namespaced_logging.nimble index 0329471..2abe850 100644 --- a/namespaced_logging.nimble +++ b/namespaced_logging.nimble @@ -34,10 +34,14 @@ proc runRegressionTests() = proc runThreadSafetyTests() = if fileExists("tests/bin/thread_safety"): rmFile("tests/bin/thread_safety") + if fileExists("tests/bin/writer_lifecycle"): rmFile("tests/bin/writer_lifecycle") exec "nimble c --mm:orc --threads:on --passC:-fsanitize=thread --passC:-g --passL:-fsanitize=thread --out:tests/bin/thread_safety tests/thread_safety.nim" if not fileExists("tests/bin/thread_safety"): raise newException(OSError, "Thread-safety test compilation produced no executable") - exec "bash tests/run_tsan.sh tests/bin/thread_safety" + exec "nimble c --mm:orc --threads:on --passC:-fsanitize=thread --passC:-g --passL:-fsanitize=thread --out:tests/bin/writer_lifecycle src/namespaced_logging.nim" + if not fileExists("tests/bin/writer_lifecycle"): + raise newException(OSError, "Writer lifecycle test compilation produced no executable") + exec "bash tests/run_tsan.sh tests/bin/thread_safety tests/bin/writer_lifecycle" task test, "Run all tests, including ThreadSanitizer, and report all failing suites.": var failedSuites: seq[string] = @[] diff --git a/src/namespaced_logging.nim b/src/namespaced_logging.nim index 28d0e4e..3a5ce36 100644 --- a/src/namespaced_logging.nim +++ b/src/namespaced_logging.nim @@ -1,4 +1,4 @@ -import std/[algorithm, atomics, exitprocs, json, locks, options, os, paths, +import std/[algorithm, atomics, exitprocs, json, locks, options, paths, sequtils, strutils, tables, times] import timeutils import std/logging as stdlog @@ -93,18 +93,21 @@ type local: StatefulLogFormatterFunc ConsoleMessage = object + stop: bool msg: string useStderr: bool FileMessage = object + stop: bool msg: string absPath: Path LogWriterThreadState[M] = object - initialized: Atomic[bool] - shutdown: Atomic[bool] + lock: Lock + initialized: bool + shutdown: bool chan: Channel[M] writerThread: Thread[void] @@ -204,9 +207,8 @@ proc service*(ls: ThreadLocalLogService): LogService = var consoleLogging {.global.}: LogWriterThreadState[ConsoleMessage] var fileLogging {.global.}: LogWriterThreadState[FileMessage] -var loggingThreadInitLock {.global.}: Lock - -initLock(loggingThreadInitLock) +initLock(consoleLogging.lock) +initLock(fileLogging.lock) proc initLogMessage*( @@ -263,33 +265,25 @@ proc defaultErrorHandlerFunc*( stderr.flushFile() except Exception: discard # we tried... +proc shutdownWriter[M](state: var LogWriterThreadState[M]) = + # Submission uses this same lock. The stop message follows all accepted + # records, and no producer can submit after shutdown begins. + withLock state.lock: + if state.shutdown: return + state.shutdown = true + if not state.initialized: return + state.chan.send(M(stop: true)) + joinThread(state.writerThread) + # These queues have process lifetime and cannot be restarted. Keep the + # drained channel storage until process exit: Nim 2.2.10's Channel.close + # destroys its internal mutex while holding it (a sanitizer failure). + state.initialized = false + proc shutdownThreadedConsoleLogging() = - if not consoleLogging.initialized.load(): return - - withLock loggingThreadInitLock: - consoleLogging.shutdown.store(true) # signal shutdown - - # Send sentinel values to wake up the writer thread - try: consoleLogging.chan.send(ConsoleMessage(msg: "", useStderr: false)) - except Exception: discard - - joinThread(consoleLogging.writerThread) - consoleLogging.chan.close() - consoleLogging.initialized.store(false) - + shutdownWriter(consoleLogging) proc shutdownThreadedFileLogging() = - if not fileLogging.initialized.load(): return - - fileLogging.shutdown.store(true) # signal shutdown - - withLock loggingThreadInitLock: - try: fileLogging.chan.send(FileMessage(msg: "", absPath: Path("/"))) - except Exception: discard - - joinThread(fileLogging.writerThread) - fileLogging.chan.close() - fileLogging.initialized.store(false) + shutdownWriter(fileLogging) proc ensureFreshness*(ls: var LogService) = @@ -630,41 +624,34 @@ template fatal*[L: Logger or Option[Logger], M](l: L, err: ref Exception, msg: M # ----------------------------------------------------------------------------- proc consoleWriterLoop() {.thread.} = - while not consoleLogging.shutdown.load(): - var didSomething = false - - let (hasData, cMsgObj) = consoleLogging.chan.tryRecv() - if hasData and cMsgObj.msg.len > 0: # Skip empty sentinel messages - try: - let output = - if cMsgObj.useStderr: stderr - else: stdout - output.write(cMsgObj.msg) - output.flushFile() - didSomething = true - except IOError: - discard - - # Small delay if no work to prevent busy waiting - if not didSomething: sleep(100) + while true: + let message = consoleLogging.chan.recv() + if message.stop: break + try: + let output = if message.useStderr: stderr else: stdout + output.write(message.msg) + output.flushFile() + except IOError: + discard proc fileWriterLoop() {.thread.} = const bufLen = 128 var msgsByPath = newTable[Path, seq[FileMessage]]() - while not fileLogging.shutdown.load(): - var didSomething = false - + while true: var msgBuf: array[bufLen, FileMessage] - var recvIdx = 0 + msgBuf[0] = fileLogging.chan.recv() + var recvIdx = 1 var writeIdx = 0 - var dataAvailable = true + var stopping = msgBuf[0].stop - while dataAvailable and recvIdx < bufLen: - # Fill our message buffer if we can - (dataAvailable, msgBuf[recvIdx]) = fileLogging.chan.tryRecv() - if dataAvailable: inc recvIdx + while not stopping and recvIdx < bufLen: + let (available, message) = fileLogging.chan.tryRecv() + if not available: break + msgBuf[recvIdx] = message + inc recvIdx + stopping = message.stop # Organize messages by destination file msgsByPath.clear() @@ -672,10 +659,9 @@ proc fileWriterLoop() {.thread.} = let fMsgObj = msgBuf[writeIdx] inc writeIdx - if fMsgObj.msg.len > 0: # skip empty sentinel messages + if not fMsgObj.stop: if not msgsByPath.contains(fMsgObj.absPath): msgsByPath[fMsgObj.absPath] = @[] msgsByPath[fMsgObj.absPath].add(fMsgObj) - didSomething = true # Write all messages in file order to optimize file open/flush/close for path, msgs in pairs(msgsByPath): @@ -691,34 +677,30 @@ proc fileWriterLoop() {.thread.} = flushFile(f) close(f) - # Wait a bit if we had no work to prevent busy waiting - if not didSomething: sleep(100) + # The final batch must be written before honoring its stop message. + if stopping: break -proc initThreadedConsoleLogging() = - if consoleLogging.initialized.load: return - - withLock loggingThreadInitLock: - if consoleLogging.initialized.load: return - consoleLogging.chan.open() - consoleLogging.shutdown.store(false) - - # Create writer thread with reference to the service - createThread(consoleLogging.writerThread, consoleWriterLoop) - consoleLogging.initialized.store(true) +proc sendConsoleMessage(message: ConsoleMessage) = + withLock consoleLogging.lock: + if consoleLogging.shutdown: + raise newException(IOError, "Console logging has shut down") + if not consoleLogging.initialized: + consoleLogging.chan.open() + createThread(consoleLogging.writerThread, consoleWriterLoop) + consoleLogging.initialized = true + consoleLogging.chan.send(message) -proc initThreadedFileLogging() = - if fileLogging.initialized.load: return - - withLock loggingThreadInitLock: - if fileLogging.initialized.load: return - fileLogging.chan.open() - fileLogging.shutdown.store(false) - - # Create writer thread with reference to the service - createThread(fileLogging.writerThread, fileWriterLoop) - fileLogging.initialized.store(true) +proc sendFileMessage(message: FileMessage) = + withLock fileLogging.lock: + if fileLogging.shutdown: + raise newException(IOError, "File logging has shut down") + if not fileLogging.initialized: + fileLogging.chan.open() + createThread(fileLogging.writerThread, fileWriterLoop) + fileLogging.initialized = true + fileLogging.chan.send(message) # ----------------------------------------------------------------------------- @@ -872,11 +854,8 @@ proc appendLogMessageMultiThreaded( ls: ThreadLocalLogService, msg: LogMessage) {.gcsafe.} = - if not consoleLogging.initialized.load(): - initThreadedConsoleLogging() - try: - consoleLogging.chan.send(ConsoleMessage( + sendConsoleMessage(ConsoleMessage( msg: cla.formatter.format(msg), useStderr: cla.useStderr)) except Exception: @@ -961,11 +940,8 @@ proc appendLogMessageMultiThreaded( ls: ThreadLocalLogService, msg: LogMessage) {.gcsafe.} = - if not fileLogging.initialized.load(): - initThreadedFileLogging() - try: - fileLogging.chan.send(FileMessage( + sendFileMessage(FileMessage( msg: fla.formatter.format(msg), absPath: fla.absPath)) except Exception: discard @@ -1060,7 +1036,7 @@ addExitProc(proc() = when isMainModule: - import std/[files, tempfiles, unittest] + import std/[files, os, tempfiles, unittest] import ./namespaced_logging/testutil suite "LogService Initialization": @@ -1560,3 +1536,58 @@ when isMainModule: lm[0].msg.contains("message at notice") lm[1].msg.contains("message at warn") lm[2].msg.contains("message at fatal") + + type WriterShutdownProbe = object + path: Path + ready: Atomic[int] + start: Atomic[bool] + accepted: array[4, int] + + proc submitDuringShutdown(args: tuple[probe: ptr WriterShutdownProbe, worker: int]) {.thread.} = + let probe = args.probe + let worker = args.worker + sendFileMessage(FileMessage(absPath: probe.path, msg: $worker & ":0")) + probe.accepted[worker] = 1 + probe.ready.atomicInc() + while not probe.start.load(): discard + for i in 1 ..< 10_000: + try: + sendFileMessage(FileMessage(absPath: probe.path, msg: $worker & ":" & $i)) + inc probe.accepted[worker] + except IOError: + break + + suite "Background writer lifecycle": + test "shutdown drains accepted records while excluding further sends": + let directory = createTempDir("namespaced_logging_lifecycle_", "") + defer: removeDir(directory) + var probe = WriterShutdownProbe(path: Path(directory / "messages.log")) + var workers: array[4, Thread[tuple[probe: ptr WriterShutdownProbe, worker: int]]] + for i in 0 ..< workers.len: + createThread(workers[i], submitDuringShutdown, (addr probe, i)) + while probe.ready.load() != workers.len: discard + probe.start.store(true) + shutdownThreadedFileLogging() + joinThreads(workers) + + # Shutdown is permanent and repeatable; it must never reopen the channel. + shutdownThreadedFileLogging() + expect IOError: + sendFileMessage(FileMessage(absPath: probe.path, msg: "too late")) + var received: array[4, int] + for line in readFile($probe.path).splitLines(): + if line.len == 0: continue + let parts = line.split(':') + require parts.len == 2 + let worker = parseInt(parts[0]) + require worker in 0 ..< workers.len + check parseInt(parts[1]) == received[worker] + inc received[worker] + check received == probe.accepted + + test "console shutdown wakes an empty writer and rejects further sends": + sendConsoleMessage(ConsoleMessage(msg: "")) + shutdownThreadedConsoleLogging() + shutdownThreadedConsoleLogging() + expect IOError: + sendConsoleMessage(ConsoleMessage(msg: "")) diff --git a/tests/README.md b/tests/README.md index edc3928..a293fca 100644 --- a/tests/README.md +++ b/tests/README.md @@ -8,9 +8,9 @@ Run from the repository root: | Command | Coverage | | --- | --- | | `nimble test` | All unit, regression, and ThreadSanitizer tests under ORC with threads enabled | -| `nimble testUnit` | Existing embedded unit tests | +| `nimble testUnit` | Embedded unit tests, including writer submission/shutdown coordination | | `nimble testRegression` | Custom-state/formatter ownership, handle lifetime, configuration freshness, refresh failure recovery, appender-state cleanup, and writer shutdown | -| `nimble testThreadSafety` | Separate ThreadSanitizer processes for service ownership, last-owner teardown, reconfiguration, shared state, factory state, formatter closures, collector snapshots, timestamps, and the Nim allocator, plus an independent-service control | +| `nimble testThreadSafety` | Separate ThreadSanitizer processes for service ownership, last-owner teardown, reconfiguration, shared state, factory state, formatter closures, collector snapshots, timestamps, the Nim allocator, and writer lifecycle, plus an independent-service control | `nimble test` attempts every suite even if an earlier suite fails to compile or run, summarizes the failing suites, and exits unsuccessfully if any suite fails. @@ -21,7 +21,9 @@ when a nested Nimble command fails to propagate its exit status. Executables are generated in ignored `tests/bin/`. The shutdown child is built with `-d:multithreaded`; `--threads:on` alone does not enable background writers. -The other tests do not need background I/O to exercise ownership. +The ownership tests do not need background I/O. The embedded writer lifecycle +tests call the internal writer functions directly; their sanitizer build is +`tests/bin/writer_lifecycle`. The full run and the sanitizer subset require Linux, Bash, GNU `timeout`, and a C compiler and runtime supporting `-fsanitize=thread`. The sanitizer instruments @@ -33,6 +35,7 @@ one scenario after building: ```sh TSAN_OPTIONS=halt_on_error=1:exitcode=66 timeout 30s tests/bin/thread_safety service +TSAN_OPTIONS=halt_on_error=1:exitcode=66 timeout 30s tests/bin/writer_lifecycle 'Background writer lifecycle::*' ``` The worker start barrier ensures argument setup finishes before the concurrent @@ -82,20 +85,33 @@ the unsafe forms are tested as compilation failures. The in-process regression tests are deterministic. The lifetime suite checks copying, self-assignment, replacement, moving, container copies, and exporting a handle from a local service. Shutdown tests launch -a fresh process for each destination and check all 10,000 messages after normal -exit, using isolated temporary paths and a 15-second deadline. They exercise the -real exit hook without sleeps to hide queued-message loss. Like the race tests, -the shutdown failure can depend on scheduling; a passing run is not proof of -thread safety or reliable draining. +a fresh process for each scenario and check all 10,000 numbered messages after +normal exit, using isolated temporary paths and a 15-second deadline. Scenarios +cover stdout, stderr, files, empty formatted messages, multiple file destinations, +and four joined producers. They verify message content and order within each +producer and destination, using the real exit hook without sleeps or explicit +flushes. File batches must preserve blank records and finish their final batch +before the writer stops. -This suite does not yet force an active producer/channel-close race or exhaust -ORC cycle-collector interleavings. It provides targeted regressions for ownership, -configuration, cleanup, timestamps, allocator behavior, and shutdown. +The embedded lifecycle tests start producers, release them to submit more +records while the file writer shuts down, and verify every accepted record. +They also check repeated shutdown, rejection of sends after shutdown, and +waking the console writer after an empty message. The sanitizer task runs these +tests as a separate process even when another scenario fails. + +Nim 2.2.10's `Channel.close()` destroys its internal mutex while holding it, +which the lifecycle sanitizer test exposed after the writers had joined. +Shutdown therefore drains and joins the writers, rejects further submissions, +and leaves the process-wide channels allocated until process exit. It does not +call the faulty runtime teardown or suppress its diagnostics. + +The suite provides targeted regressions; it does not exhaust thread scheduling +or ORC cycle-collector interleavings. The ownership scenarios, lifetime suite, and appender-state cleanup regression exercise the new ownership contracts. `isEnabled` freshness, failed-refresh -recovery, local-time timestamp construction, allocator reports, and writer -shutdown remain for subsequent work. +recovery, local-time timestamp construction, and allocator reports remain for +subsequent work. The standalone `ownership.nim` regression suite passed AddressSanitizer and LeakSanitizer with `--mm:orc -d:useMalloc`. An additional worker-thread build found diff --git a/tests/regressions.nim b/tests/regressions.nim index e093d70..92de9b0 100644 --- a/tests/regressions.nim +++ b/tests/regressions.nim @@ -134,7 +134,8 @@ suite "ORC ownership and configuration regressions": check destroyed == 1 suite "Background writer shutdown regressions": - for destination in ["file", "console"]: + for destination in ["file", "console", "stderr", "file-empty", "console-empty", + "files", "file-parallel", "console-parallel"]: test destination & " writer drains accepted messages at normal process exit": let directory = createTempDir("namespaced_logging_shutdown_", "") let outputPath = directory / "messages.log" @@ -150,9 +151,25 @@ suite "Background writer shutdown regressions": check exitCode == 0 require fileExists(outputPath) - let received = readFile(outputPath).splitLines() - # Check every record and its order, not just file existence. splitLines - # includes one final empty item because each record ends with a newline. - check received.len == 10_001 - for i in 0 ..< min(10_000, received.len - 1): - check received[i] == "message " & $i + if destination.endsWith("-parallel"): + let received = readFile(outputPath).splitLines() + require received.len == 10_001 + var next = [0, 2_500, 5_000, 7_500] + for line in received[0 ..< 10_000]: + require line.startsWith("message ") + let number = parseInt(line[8 .. ^1]) + require number in 0 ..< 10_000 + let worker = number div 2_500 + check number == next[worker] + inc next[worker] + check next == [2_500, 5_000, 7_500, 10_000] + else: + var expected: string + for i in 0 ..< 10_000: + if destination == "file-empty" and i mod 128 == 0: + expected.add("\n") + expected.add("message " & $i & "\n") + check readFile(outputPath) == expected + if destination == "files": + require fileExists(outputPath & ".second") + check readFile(outputPath & ".second") == expected diff --git a/tests/run_tsan.sh b/tests/run_tsan.sh index 146f723..0ab90ab 100644 --- a/tests/run_tsan.sh +++ b/tests/run_tsan.sh @@ -2,6 +2,7 @@ set -u binary=${1:?Expected the ThreadSanitizer test executable} +writer_binary=${2:?Expected the writer lifecycle test executable} result=0 for scenario in independent service service-lifetime service-reconfigure custom-state formatter shared-state-lifetime factory-state collector formatter-lifetime timestamp allocator; do echo "ThreadSanitizer scenario: $scenario" @@ -13,4 +14,13 @@ for scenario in independent service service-lifetime service-reconfigure custom- result=1 fi done + +echo "ThreadSanitizer scenario: writer-lifecycle" +if TSAN_OPTIONS=halt_on_error=1:exitcode=66 timeout 30s "$writer_binary" 'Background writer lifecycle::*'; then + echo "PASS: writer-lifecycle" +else + code=$? + echo "FAIL: writer-lifecycle (exit $code)" >&2 + result=1 +fi exit "$result" diff --git a/tests/writer_child.nim b/tests/writer_child.nim index c29210d..0b63ad6 100644 --- a/tests/writer_child.nim +++ b/tests/writer_child.nim @@ -1,4 +1,4 @@ -import std/os +import std/[os, strutils] import ../src/namespaced_logging when not defined(multithreaded): @@ -6,24 +6,59 @@ when not defined(multithreaded): proc formatLine(msg: LogMessage): string {.gcsafe.} = msg.msg & "\n" +proc formatOptionalLine(msg: LogMessage): string {.gcsafe.} = + if msg.msg.len > 0: msg.msg & "\n" else: "" + +proc writeMessages(args: tuple[service: LogService, worker: int]) {.thread.} = + let logger = args.service.threadLocalRef().getLogger("shutdown") + for i in 0 ..< 2_500: + # Keep the timestamp race out of this writer lifecycle test. + logger.log(LogMessage(scope: "shutdown", level: lvlInfo, + msg: "message " & $(args.worker * 2_500 + i))) + proc main() = doAssert paramCount() == 2 + let destination = paramStr(1) var service = initLogService() - case paramStr(1) - of "file": + case destination + of "file", "file-parallel", "file-empty": service.addAppender(initFileLogAppender(paramStr(2))) - of "console": + of "files": + service.addAppender(initFileLogAppender(paramStr(2), namespace = "first")) + service.addAppender(initFileLogAppender(paramStr(2) & ".second", namespace = "second")) + of "console", "console-parallel", "console-empty": # Keep stdout open through the library's exit hook. Redirecting it to a # file also prevents a full parent/child pipe from blocking the writer. doAssert open(stdout, paramStr(2), fmWrite) - service.addAppender(initConsoleLogAppender(formatter = formatLine)) + if destination == "console-empty": + service.addAppender(initConsoleLogAppender(formatter = formatOptionalLine)) + else: + service.addAppender(initConsoleLogAppender(formatter = formatLine)) + of "stderr": + doAssert open(stderr, paramStr(2), fmWrite) + service.addAppender(initConsoleLogAppender(useStderr = true, formatter = formatLine)) else: raise newException(ValueError, "unknown writer destination") - let local = service.threadLocalRef() - let logger = local.getLogger("shutdown") - for i in 0 ..< 10_000: - logger.info("message " & $i) + if destination.endsWith("-parallel"): + var workers: array[4, Thread[tuple[service: LogService, worker: int]]] + for i in 0 ..< workers.len: + createThread(workers[i], writeMessages, (service, i)) + joinThreads(workers) + else: + let local = service.threadLocalRef() + if destination == "files": + let first = local.getLogger("first") + let second = local.getLogger("second") + for i in 0 ..< 10_000: + first.info("message " & $i) + second.info("message " & $i) + else: + let logger = local.getLogger("shutdown") + for i in 0 ..< 10_000: + if destination.endsWith("-empty") and i mod 128 == 0: + logger.info("") + logger.info("message " & $i) # No sleep or explicit flush: normal exit must preserve accepted messages. main()