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()