Separate shared service ownership from thread-local logging caches

This commit is contained in:
jdb committed 2026-09-22 07:18:22 -05:00
1 parent c1c8a3c072
commit 761bd66793
6 files changed
+386 -157

No files matched your search

+150 -94
View File
@@ -7,19 +7,19 @@ from std/logging import Level
export Level
type
GlobalLogServiceObj {.acyclic.} = object
GlobalLogServiceObj = object
owners: Atomic[int] # Lifetime of independently owned service handles
configVersion: Atomic[int] # To signal the need for threads to update
initialized: Atomic[bool] # To protect deconstruction
lock: Lock # To protect reads/writes across threads
appenders: seq[LogAppender]
thresholds: TableRef[string, Level]
thresholds: Table[string, Level]
rootLevel: Atomic[Level]
errorHandler: ErrorHandlerFunc
errorHandlerLock: Lock
GlobalLogService = ref GlobalLogServiceObj
GlobalLogService = ptr GlobalLogServiceObj
ErrorHandlerFunc* =
@@ -27,27 +27,32 @@ type
LogService* = object
## Thread-local logging state intended for use by library consumers
configVersion: int # Version of the global state we have
global: GlobalLogService # Reference to the global state
## Copyable shared configuration handle with an atomic lifetime.
## Each thread must own its own handle; do not concurrently assign to or
## destroy the same handle variable while another thread reads it.
global: GlobalLogService
appenders: seq[LogAppender] # Thread-local copy of appenders
ThreadLocalLogService* = ref object
## Cache confined to the thread that creates it. Appender clones must
## themselves respect thread ownership of custom state and formatters.
shared: LogService
configVersion: int
appenders: seq[LogAppender]
thresholds: TableRef[string, Level]
ThreadLocalLogService* = ref LogService
Logger* = ref object
scope*: string
threadSvc: ThreadLocalLogService
LogAppender* = ref object of RootObj
## Base type for log appenders. Note that all LogAppenders must (and do)
## have a clone method that makes a thread-safe copy of the appender state
## that is used when adding appenders to the global service state as well
## as during individual thread state refresh.
## Base type for log appenders. Note that all LogAppenders must have a
## clone method which must return an independently owned appender for
## shared configuration and for each thread's cache. Nested state and
## formatter references also need safe ownership; allocating a new outer
## appender alone does not make shared ORC references safe.
namespace*: string
## Allows filtering of messages based on their scope. Messages are
@@ -128,6 +133,54 @@ type
const UninitializedConfigVersion = low(int)
proc retainGlobal(gls: GlobalLogService) {.inline.} =
if gls != nil:
discard gls.owners.fetchAdd(1, moRelaxed)
proc releaseGlobal(gls: GlobalLogService) =
if gls != nil and gls.owners.fetchSub(1, moAcquireRelease) == 1:
# The last owner has exclusive access to the control block. Its managed
# fields must be destroyed explicitly because it is manually allocated.
`=destroy`(gls.appenders)
`=destroy`(gls.thresholds)
deinitLock(gls.errorHandlerLock)
deinitLock(gls.lock)
deallocShared(gls)
proc `=destroy`*(ls: LogService) =
releaseGlobal(ls.global)
proc `=copy`*(dest: var LogService, source: LogService) =
if dest.global == source.global: return
retainGlobal(source.global)
releaseGlobal(dest.global)
dest.global = source.global
proc `=dup`*(source: LogService): LogService =
retainGlobal(source.global)
result.global = source.global
proc `=wasMoved`*(ls: var LogService) =
ls.global = nil
# The compiler-generated sink operation transfers the pointer after releasing
# the destination. It does not retain the transferred owner.
template global(ls: ThreadLocalLogService): untyped =
ls.shared.global
proc service*(ls: ThreadLocalLogService): LogService =
## Obtain an independently owned shared handle, e.g. to pass to a worker.
## This replaces copying ls[] when ThreadLocalLogService was ref LogService.
ls.shared
var consoleLogging {.global.}: LogWriterThreadState[ConsoleMessage]
var fileLogging {.global.}: LogWriterThreadState[FileMessage]
var loggingThreadInitLock {.global.}: Lock
@@ -218,88 +271,79 @@ proc shutdownThreadedFileLogging() =
fileLogging.initialized.store(false)
proc `=destroy`*(gls: var GlobalLogServiceObj) =
# only one thread should cleanup
if not gls.initialized.exchange(false): return
try: deinitLock(gls.lock)
except Exception: discard
try: deinitLock(gls.errorHandlerLock)
except Exception: discard
proc ensureFreshness*(ls: var LogService) =
## A shared handle has no cached configuration. Retained for callers that
## previously refreshed LogService values; only local services need refresh.
if ls.global == nil:
raise newException(ValueError,
"LogService not properly initialized. Use initLogService")
proc ensureFreshness*(ls: ThreadLocalLogService) =
## Check our thread-local version of the logging service configuration
## against the global log service definition. This uses an atomic version
## number to allow the check to be inexpensive, only incurring the more
## expensive copy operations if there are actual changes.
if ls.global.isNil or not ls.global.initialized.load():
if ls.isNil or ls.global.isNil:
raise newException(ValueError,
"LogService not properly initialized. Use initLogService to obtain a" &
"properly initialized LogService object")
if ls.configVersion == ls.global.configVersion.load(): return
# Keep a stable borrowed pointer while inspecting configuration. In
# particular, do not iterate a seq through a function-returned expression:
# that can create an owning temporary and register its shared appender refs
# with this thread's ORC cycle collector even while the lock is held.
let gls = ls.shared.global
if ls.configVersion == gls.configVersion.load(): return
withLock ls.global.lock:
ls.configVersion = ls.global.configVersion.load
withLock gls.lock:
ls.configVersion = gls.configVersion.load
ls.appenders = @[]
for a in ls.global.appenders: ls.appenders.add(clone(a))
for a in gls.appenders: ls.appenders.add(clone(a))
ls.thresholds = newTable[string, Level]()
for ns, t in pairs(ls.global.thresholds): ls.thresholds[ns] = t
proc ensureFreshness*(ls: ThreadLocalLogService) = ensureFreshness(ls[])
for ns, t in pairs(gls.thresholds): ls.thresholds[ns] = t
proc initGlobalLogService(
rootLevel = lvlAll,
errorHandler = defaultErrorHandlerFunc): GlobalLogService =
result = GlobalLogService()
result = cast[GlobalLogService](allocShared0(sizeof(GlobalLogServiceObj)))
result.owners.store(1)
result.configVersion.store(0)
initLock(result.lock)
initLock(result.errorHandlerLock)
result.appenders = @[]
result.thresholds = newTable[string, Level]()
result.thresholds = initTable[string, Level]()
result.rootLevel.store(rootLevel)
result.errorHandler = errorHandler
result.initialized.store(true)
proc initLogService(gls: GlobalLogService): LogService =
var lsRef: ThreadLocalLogService = ThreadLocalLogService(
configVersion: UninitializedConfigVersion, global: gls)
ensureFreshness(lsRef)
result = lsRef[]
proc initLogService*(
rootLevel = lvlAll,
errorHandler = defaultErrorHandlerFunc): LogService =
## Initialize a new LogService instance. The resulting LogService object can
## be safely copied between threads and provides thread-safe synchronization
## of logger configuration and shared state. It is important that the
## LogService object returned is retained by the main thread so that garbage
## collection does not harvest the global state while it is still in use on
## other threads.
## Initialize an independently owned shared configuration handle. Copies
## retain the shared control block atomically, including copies in worker
## arguments and local services. The original handle need not outlive them.
## The resulting *LogService* object can be safely passed between threads to
## initialize *ThreadLocalLogService* objects but note that thread-safe use
## still requires that appenders and formatters satisfy the ownership
## requirements documented on *LogAppender.*
##
## Multiple calls to *initLogService* will result in completely independent
## *LogService* objects with their own synchronized coniguration and global
## state. This is not expected to be a common usage, but it is supported.
## object used to actually configure the log service, attach appenders,
## configure thresholds, and create loggers. The ref returned by this
## procedure should also be retained by the main thread so that garbage
## collection does not harvest the global state while it is still in use.
let global = initGlobalLogService(rootLevel, errorHandler)
result = initLogService(global)
## *LogService* objects with their own synchronized configuration and global
## state.
result.global = initGlobalLogService(rootLevel, errorHandler)
proc threadLocalRef*(ls: LogService): ThreadLocalLogService =
new result
result[] = ls
## Build a fresh cache in the calling thread, retaining only the shared
## handle. Never copy another thread's cached appender or threshold refs.
result = ThreadLocalLogService(shared: ls,
configVersion: UninitializedConfigVersion)
ensureFreshness(result)
proc reportLoggingError(
@@ -361,7 +405,7 @@ proc setErrorHandler*(
proc setErrorHandler*(
ls: ThreadLocalLogService,
errHandler = defaultErrorHandlerFunc) {.gcsafe.} =
setErrorHandler(ls[], errHandler)
setErrorHandler(ls.shared, errHandler)
proc setRootThreshold*(ls: var LogService, lvl: Level) {.gcsafe.} =
@@ -370,7 +414,7 @@ proc setRootThreshold*(ls: var LogService, lvl: Level) {.gcsafe.} =
proc setRootThreshold*(ls: ThreadLocalLogService, lvl: Level) {.gcsafe.} =
setRootThreshold(ls[], lvl)
setRootThreshold(ls.shared, lvl)
func formatSimpleTextLog*(msg: LogMessage): string {.gcsafe.} = msg.msg
@@ -384,11 +428,10 @@ proc setThreshold*(ls: var LogService, scope: string, lvl: Level) {.gcsafe.} =
ls.global.thresholds[scope] = lvl
ls.global.configVersion.atomicInc
ensureFreshness(ls)
proc setThreshold*(ls: ThreadLocalLogService, scope: string, lvl: Level) {.gcsafe.} =
setThreshold(ls[], scope, lvl)
setThreshold(ls.shared, scope, lvl)
ensureFreshness(ls)
proc setThresholds*(ls: var LogService, thresholds: TableRef[string, Level]) {.gcsafe.} =
@@ -396,11 +439,10 @@ proc setThresholds*(ls: var LogService, thresholds: TableRef[string, Level]) {.g
for k,v in thresholds: ls.global.thresholds[k] = v
ls.global.configVersion.atomicInc
ensureFreshness(ls)
proc setThresholds*(ls: ThreadLocalLogService, thresholds: TableRef[string, Level]) {.gcsafe.} =
setThresholds(ls[], thresholds)
setThresholds(ls.shared, thresholds)
ensureFreshness(ls)
proc getLogger*(
ls: ThreadLocalLogService,
@@ -437,38 +479,36 @@ proc getLogger*(
proc appenders*(ls: var LogService): seq[LogAppender] {.gcsafe.} =
for a in ls.appenders: result.add(clone(a))
withLock ls.global.lock:
for a in ls.global.appenders: result.add(clone(a))
proc appenders*(ls: ThreadLocalLogService): seq[LogAppender] {.gcsafe.} =
ls[].appenders()
for a in ls.appenders: result.add(clone(a))
proc addAppender*(ls: var LogService, appender: LogAppender) {.gcsafe.} =
## Add a log appender to the global log service and refresh the local thread
## state. The updated global state will trigger other threads to refresh
## their state as well.
## Add a log appender to the shared configuration. Local services refresh
## their cached state on their next freshness check.
withLock ls.global.lock:
var copiedAppender = clone(appender)
ls.global.appenders.add(copiedAppender)
ls.global.configVersion.atomicInc
proc addAppender*(ls: ThreadLocalLogService, appender: LogAppender) {.gcsafe.} =
addAppender(ls.shared, appender)
ensureFreshness(ls)
proc addAppender*(ls: ThreadLocalLogService, appender: LogAppender) {.gcsafe.} =
addAppender(ls[], appender)
proc clearAppenders*(ls: var LogService) {.gcsafe.} =
## Remove all log appenders added to the global log service and refresh the
## local thread state. The updated global state will trigger other threads to
## refresh their state as well.
## Remove all shared appenders. Local services refresh their cached state
## on their next freshness check.
withLock ls.global.lock:
ls.global.appenders = @[]
ls.global.configVersion.atomicInc
proc clearAppenders*(ls: ThreadLocalLogService) {.gcsafe.} =
clearAppenders(ls[])
clearAppenders(ls.shared)
func getEffectiveThreshold(logger: Logger): Level {.gcsafe.} =
@@ -729,7 +769,7 @@ method clone*(cla: ConsoleLogAppender): LogAppender {.gcsafe.} =
proc appendLogMessageMultiThreaded(
cla: ConsoleLogAppender,
ls: ref LogService,
ls: ThreadLocalLogService,
msg: LogMessage) {.gcsafe.} =
if not consoleLogging.initialized.load():
@@ -758,7 +798,7 @@ proc appendLogMessageMultiThreaded(
proc appendLogMessageSingleThreaded(
cla: ConsoleLogAppender,
ls: ref LogService,
ls: ThreadLocalLogService,
msg: LogMessage) {.gcsafe.} =
try:
@@ -814,7 +854,7 @@ method clone*(fla: FileLogAppender): LogAppender {.gcsafe.} =
proc appendLogMessageMultiThreaded(
fla: FileLogAppender,
ls: ref LogService,
ls: ThreadLocalLogService,
msg: LogMessage) {.gcsafe.} =
if not fileLogging.initialized.load():
@@ -829,7 +869,7 @@ proc appendLogMessageMultiThreaded(
proc appendLogMessageSingleThreaded(
fla: FileLogAppender,
ls: ref LogService,
ls: ThreadLocalLogService,
msg: LogMessage) {.gcsafe.} =
try:
@@ -922,18 +962,34 @@ when isMainModule:
test "initLogService creates valid service":
let ls = initLogService()
let local = ls.threadLocalRef()
check:
ls.global != nil
ls.global.initialized.load() == true
ls.global.owners.load() == 2
ls.global.configVersion.load() == 0
ls.global.rootLevel.load() == lvlAll
ls.configVersion >= 0
ls.configVersion == ls.global.configVersion.load
local.configVersion >= 0
local.configVersion == ls.global.configVersion.load
test "initLogService with custom root level":
let ls = initLogService(lvlInfo)
check ls.global.rootLevel.load() == lvlInfo
test "threadLocalRef builds independent caches for an unchanged service":
var service = initLogService()
service.addAppender(initConsoleLogAppender())
service.setThreshold("worker", lvlInfo)
let first = service.threadLocalRef()
let second = service.threadLocalRef()
check:
first.global == second.global
first.configVersion == second.configVersion
cast[pointer](first.thresholds) != cast[pointer](second.thresholds)
first.thresholds["worker"] == second.thresholds["worker"]
first.appenders.len == 1
second.appenders.len == 1
first.appenders[0] != second.appenders[0]
suite "LogService Configuration":
setup:
let ls = threadLocalRef(initLogService())
@@ -961,7 +1017,7 @@ when isMainModule:
check ls.thresholds["test"] == lvlError
test "ensureFreshness syncs configuration between multiple loggers":
var ls2 = ls[] # use copy semantics
let ls2 = ls.service.threadLocalRef()
ls.setThreshold("api", lvlWarn)
check:
@@ -992,7 +1048,7 @@ when isMainModule:
initCond(cUpd)
proc runA(lsGiven: LogService) {.thread, gcsafe.} =
var ls = lsGiven
let ls = lsGiven.threadLocalRef()
withLock lock:
# 1a. Initialize both threads
cInitA.signal()
@@ -1003,7 +1059,7 @@ when isMainModule:
cUpd.signal()
proc runB(lsGiven: LogService) {.thread, gcsafe.} =
var ls = lsGiven
let ls = lsGiven.threadLocalRef()
withLock lock:
# 1b. Initialize both threads
cInitB.signal()
@@ -1016,10 +1072,10 @@ when isMainModule:
check ls.thresholds["api"] == lvlWarn
# 1. Initialize both threads
createThread(tA, runA, ls[])
createThread(tA, runA, ls.service)
withLock lock: wait(cInitA, lock)
createThread(tB, runB, ls[])
createThread(tB, runB, ls.service)
withLock lock: wait(cInitB, lock)
cInitAll.broadcast() # finished with initialization