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.
This commit is contained in:
jdb committed 2026-09-28 15:00:18 -05:00
1 parent b2d0e9dade
commit d5e3b9c301
7 files changed
+241 -125

No files matched your search

+7 -4
View File
@@ -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.
+5 -1
View File
@@ -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] = @[]
+122 -91
View File
@@ -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: ""))
+29 -13
View File
@@ -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
+24 -7
View File
@@ -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
+10
View File
@@ -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"
+44 -9
View File
@@ -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()