Clarify shared logging state and owning handle types
This commit is contained in:
1 parent
761bd66793
commit
851caef706
2 files changed
+94
-95
No files matched your search
@@ -266,7 +266,7 @@ usability in multithreaded environments. The design centers on two key types:
|
||||
#### LogService (Value Type)
|
||||
```nim
|
||||
type LogService* = object
|
||||
global: GlobalLogService
|
||||
shared: ptr SharedLogServiceState
|
||||
```
|
||||
|
||||
The *LogService* object is intended to support uses cases such as:
|
||||
@@ -288,7 +288,7 @@ The *LogService* object is intended to support uses cases such as:
|
||||
#### ThreadLocalLogService (Reference Type)
|
||||
```nim
|
||||
type ThreadLocalLogService* = ref object
|
||||
shared: LogService
|
||||
serviceHandle: LogService
|
||||
configVersion: int
|
||||
appenders: seq[LogAppender]
|
||||
thresholds: TableRef[string, Level]
|
||||
@@ -321,7 +321,7 @@ during regular operation and support both the configuration functions of
|
||||
> *ThreadLocalLogService* objects should **never** be shared outside the
|
||||
> context of the thread in which they were initialized.
|
||||
|
||||
#### GlobalLogService (Internal)
|
||||
#### SharedLogServiceState (Internal)
|
||||
|
||||
Under the hood *LogService* holds a pointer to a manually allocated shared
|
||||
control block that serves as the single source of truth for logging
|
||||
@@ -376,12 +376,12 @@ changes:
|
||||
```nim
|
||||
proc ensureFreshness*(ls: ThreadLocalLogService) =
|
||||
# Cheap atomic check first
|
||||
if ls.configVersion == ls.global.configVersion.load():
|
||||
if ls.configVersion == ls.shared.configVersion.load():
|
||||
return # No changes, return immediately
|
||||
|
||||
# Only acquire lock and copy if versions differ
|
||||
withLock ls.global.lock:
|
||||
ls.configVersion = ls.global.configVersion.load
|
||||
withLock ls.shared.lock:
|
||||
ls.configVersion = ls.shared.configVersion.load
|
||||
# Sync state...
|
||||
```
|
||||
|
||||
|
||||
+88
-89
@@ -7,7 +7,7 @@ from std/logging import Level
|
||||
export Level
|
||||
|
||||
type
|
||||
GlobalLogServiceObj = object
|
||||
SharedLogServiceState = object
|
||||
owners: Atomic[int] # Lifetime of independently owned service handles
|
||||
configVersion: Atomic[int] # To signal the need for threads to update
|
||||
lock: Lock # To protect reads/writes across threads
|
||||
@@ -19,8 +19,6 @@ type
|
||||
errorHandler: ErrorHandlerFunc
|
||||
errorHandlerLock: Lock
|
||||
|
||||
GlobalLogService = ptr GlobalLogServiceObj
|
||||
|
||||
|
||||
ErrorHandlerFunc* =
|
||||
proc (err: ref Exception, msg: string) {.gcsafe,nimcall.}
|
||||
@@ -30,13 +28,13 @@ type
|
||||
## 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
|
||||
shared: ptr SharedLogServiceState
|
||||
|
||||
|
||||
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
|
||||
serviceHandle: LogService
|
||||
configVersion: int
|
||||
appenders: seq[LogAppender]
|
||||
thresholds: TableRef[string, Level]
|
||||
@@ -133,52 +131,52 @@ type
|
||||
|
||||
const UninitializedConfigVersion = low(int)
|
||||
|
||||
proc retainGlobal(gls: GlobalLogService) {.inline.} =
|
||||
if gls != nil:
|
||||
discard gls.owners.fetchAdd(1, moRelaxed)
|
||||
proc retainSharedState(state: ptr SharedLogServiceState) {.inline.} =
|
||||
if state != nil:
|
||||
discard state.owners.fetchAdd(1, moRelaxed)
|
||||
|
||||
|
||||
proc releaseGlobal(gls: GlobalLogService) =
|
||||
if gls != nil and gls.owners.fetchSub(1, moAcquireRelease) == 1:
|
||||
proc releaseSharedState(state: ptr SharedLogServiceState) =
|
||||
if state != nil and state.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)
|
||||
`=destroy`(state.appenders)
|
||||
`=destroy`(state.thresholds)
|
||||
deinitLock(state.errorHandlerLock)
|
||||
deinitLock(state.lock)
|
||||
deallocShared(state)
|
||||
|
||||
|
||||
proc `=destroy`*(ls: LogService) =
|
||||
releaseGlobal(ls.global)
|
||||
releaseSharedState(ls.shared)
|
||||
|
||||
|
||||
proc `=copy`*(dest: var LogService, source: LogService) =
|
||||
if dest.global == source.global: return
|
||||
retainGlobal(source.global)
|
||||
releaseGlobal(dest.global)
|
||||
dest.global = source.global
|
||||
if dest.shared == source.shared: return
|
||||
retainSharedState(source.shared)
|
||||
releaseSharedState(dest.shared)
|
||||
dest.shared = source.shared
|
||||
|
||||
|
||||
proc `=dup`*(source: LogService): LogService =
|
||||
retainGlobal(source.global)
|
||||
result.global = source.global
|
||||
retainSharedState(source.shared)
|
||||
result.shared = source.shared
|
||||
|
||||
|
||||
proc `=wasMoved`*(ls: var LogService) =
|
||||
ls.global = nil
|
||||
ls.shared = 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
|
||||
template shared(ls: ThreadLocalLogService): untyped =
|
||||
ls.serviceHandle.shared
|
||||
|
||||
|
||||
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
|
||||
ls.serviceHandle
|
||||
|
||||
|
||||
var consoleLogging {.global.}: LogWriterThreadState[ConsoleMessage]
|
||||
@@ -274,7 +272,7 @@ proc shutdownThreadedFileLogging() =
|
||||
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:
|
||||
if ls.shared == nil:
|
||||
raise newException(ValueError,
|
||||
"LogService not properly initialized. Use initLogService")
|
||||
|
||||
@@ -284,7 +282,7 @@ proc ensureFreshness*(ls: ThreadLocalLogService) =
|
||||
## 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.isNil or ls.global.isNil:
|
||||
if ls.isNil or ls.shared.isNil:
|
||||
raise newException(ValueError,
|
||||
"LogService not properly initialized. Use initLogService to obtain a" &
|
||||
"properly initialized LogService object")
|
||||
@@ -293,23 +291,24 @@ proc ensureFreshness*(ls: ThreadLocalLogService) =
|
||||
# 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
|
||||
let state = ls.serviceHandle.shared
|
||||
if ls.configVersion == state.configVersion.load(): return
|
||||
|
||||
withLock gls.lock:
|
||||
ls.configVersion = gls.configVersion.load
|
||||
withLock state.lock:
|
||||
ls.configVersion = state.configVersion.load
|
||||
|
||||
ls.appenders = @[]
|
||||
for a in gls.appenders: ls.appenders.add(clone(a))
|
||||
for a in state.appenders: ls.appenders.add(clone(a))
|
||||
|
||||
ls.thresholds = newTable[string, Level]()
|
||||
for ns, t in pairs(gls.thresholds): ls.thresholds[ns] = t
|
||||
for ns, t in pairs(state.thresholds): ls.thresholds[ns] = t
|
||||
|
||||
|
||||
proc initGlobalLogService(
|
||||
proc initSharedLogServiceState(
|
||||
rootLevel = lvlAll,
|
||||
errorHandler = defaultErrorHandlerFunc): GlobalLogService =
|
||||
result = cast[GlobalLogService](allocShared0(sizeof(GlobalLogServiceObj)))
|
||||
errorHandler = defaultErrorHandlerFunc): ptr SharedLogServiceState =
|
||||
result = cast[ptr SharedLogServiceState](
|
||||
allocShared0(sizeof(SharedLogServiceState)))
|
||||
result.owners.store(1)
|
||||
result.configVersion.store(0)
|
||||
initLock(result.lock)
|
||||
@@ -335,26 +334,26 @@ proc initLogService*(
|
||||
## Multiple calls to *initLogService* will result in completely independent
|
||||
## *LogService* objects with their own synchronized configuration and global
|
||||
## state.
|
||||
result.global = initGlobalLogService(rootLevel, errorHandler)
|
||||
result.shared = initSharedLogServiceState(rootLevel, errorHandler)
|
||||
|
||||
|
||||
proc threadLocalRef*(ls: LogService): ThreadLocalLogService =
|
||||
## 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,
|
||||
result = ThreadLocalLogService(serviceHandle: ls,
|
||||
configVersion: UninitializedConfigVersion)
|
||||
ensureFreshness(result)
|
||||
|
||||
|
||||
proc reportLoggingError(
|
||||
gls: GlobalLogService,
|
||||
state: ptr SharedLogServiceState,
|
||||
err: ref Exception,
|
||||
msg: string) =
|
||||
|
||||
var handler: ErrorHandlerFunc
|
||||
|
||||
withLock gls.errorHandlerLock:
|
||||
handler = gls.errorHandler
|
||||
withLock state.errorHandlerLock:
|
||||
handler = state.errorHandler
|
||||
|
||||
if not handler.isNil:
|
||||
try: handler(err, msg)
|
||||
@@ -398,23 +397,23 @@ proc setErrorHandler*(
|
||||
|
||||
if errHandler.isNil: return
|
||||
|
||||
withLock ls.global.errorHandlerLock:
|
||||
ls.global.errorHandler = errHandler
|
||||
withLock ls.shared.errorHandlerLock:
|
||||
ls.shared.errorHandler = errHandler
|
||||
|
||||
|
||||
proc setErrorHandler*(
|
||||
ls: ThreadLocalLogService,
|
||||
errHandler = defaultErrorHandlerFunc) {.gcsafe.} =
|
||||
setErrorHandler(ls.shared, errHandler)
|
||||
setErrorHandler(ls.serviceHandle, errHandler)
|
||||
|
||||
|
||||
proc setRootThreshold*(ls: var LogService, lvl: Level) {.gcsafe.} =
|
||||
ls.global.rootLevel.store(lvl)
|
||||
ls.global.configVersion.atomicInc
|
||||
ls.shared.rootLevel.store(lvl)
|
||||
ls.shared.configVersion.atomicInc
|
||||
|
||||
|
||||
proc setRootThreshold*(ls: ThreadLocalLogService, lvl: Level) {.gcsafe.} =
|
||||
setRootThreshold(ls.shared, lvl)
|
||||
setRootThreshold(ls.serviceHandle, lvl)
|
||||
|
||||
|
||||
func formatSimpleTextLog*(msg: LogMessage): string {.gcsafe.} = msg.msg
|
||||
@@ -424,24 +423,24 @@ func formatJsonStructuredLog*(msg: LogMessage): string {.gcsafe.} = $(%msg)
|
||||
|
||||
|
||||
proc setThreshold*(ls: var LogService, scope: string, lvl: Level) {.gcsafe.} =
|
||||
withLock ls.global.lock:
|
||||
ls.global.thresholds[scope] = lvl
|
||||
ls.global.configVersion.atomicInc
|
||||
withLock ls.shared.lock:
|
||||
ls.shared.thresholds[scope] = lvl
|
||||
ls.shared.configVersion.atomicInc
|
||||
|
||||
|
||||
proc setThreshold*(ls: ThreadLocalLogService, scope: string, lvl: Level) {.gcsafe.} =
|
||||
setThreshold(ls.shared, scope, lvl)
|
||||
setThreshold(ls.serviceHandle, scope, lvl)
|
||||
ensureFreshness(ls)
|
||||
|
||||
|
||||
proc setThresholds*(ls: var LogService, thresholds: TableRef[string, Level]) {.gcsafe.} =
|
||||
withLock ls.global.lock:
|
||||
for k,v in thresholds: ls.global.thresholds[k] = v
|
||||
ls.global.configVersion.atomicInc
|
||||
withLock ls.shared.lock:
|
||||
for k,v in thresholds: ls.shared.thresholds[k] = v
|
||||
ls.shared.configVersion.atomicInc
|
||||
|
||||
|
||||
proc setThresholds*(ls: ThreadLocalLogService, thresholds: TableRef[string, Level]) {.gcsafe.} =
|
||||
setThresholds(ls.shared, thresholds)
|
||||
setThresholds(ls.serviceHandle, thresholds)
|
||||
ensureFreshness(ls)
|
||||
|
||||
proc getLogger*(
|
||||
@@ -479,8 +478,8 @@ proc getLogger*(
|
||||
|
||||
|
||||
proc appenders*(ls: var LogService): seq[LogAppender] {.gcsafe.} =
|
||||
withLock ls.global.lock:
|
||||
for a in ls.global.appenders: result.add(clone(a))
|
||||
withLock ls.shared.lock:
|
||||
for a in ls.shared.appenders: result.add(clone(a))
|
||||
|
||||
proc appenders*(ls: ThreadLocalLogService): seq[LogAppender] {.gcsafe.} =
|
||||
for a in ls.appenders: result.add(clone(a))
|
||||
@@ -488,34 +487,34 @@ proc appenders*(ls: ThreadLocalLogService): seq[LogAppender] {.gcsafe.} =
|
||||
proc addAppender*(ls: var LogService, appender: LogAppender) {.gcsafe.} =
|
||||
## Add a log appender to the shared configuration. Local services refresh
|
||||
## their cached state on their next freshness check.
|
||||
withLock ls.global.lock:
|
||||
withLock ls.shared.lock:
|
||||
var copiedAppender = clone(appender)
|
||||
ls.global.appenders.add(copiedAppender)
|
||||
ls.global.configVersion.atomicInc
|
||||
ls.shared.appenders.add(copiedAppender)
|
||||
ls.shared.configVersion.atomicInc
|
||||
|
||||
|
||||
proc addAppender*(ls: ThreadLocalLogService, appender: LogAppender) {.gcsafe.} =
|
||||
addAppender(ls.shared, appender)
|
||||
addAppender(ls.serviceHandle, appender)
|
||||
ensureFreshness(ls)
|
||||
|
||||
|
||||
proc clearAppenders*(ls: var LogService) {.gcsafe.} =
|
||||
## 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
|
||||
withLock ls.shared.lock:
|
||||
ls.shared.appenders = @[]
|
||||
ls.shared.configVersion.atomicInc
|
||||
|
||||
|
||||
proc clearAppenders*(ls: ThreadLocalLogService) {.gcsafe.} =
|
||||
clearAppenders(ls.shared)
|
||||
clearAppenders(ls.serviceHandle)
|
||||
|
||||
|
||||
func getEffectiveThreshold(logger: Logger): Level {.gcsafe.} =
|
||||
## Get the effective logging level threshold for a logger. This is the most
|
||||
## specific level that is set for the logger or any of its parents. The root
|
||||
## logger is used as the default if no other level is set.
|
||||
result = logger.threadSvc.global.rootLevel.load
|
||||
result = logger.threadSvc.shared.rootLevel.load
|
||||
|
||||
# Check for exact match first
|
||||
if logger.threadSvc.thresholds.hasKey(logger.scope):
|
||||
@@ -738,7 +737,7 @@ method appendLogMessage[T](
|
||||
raise newException(ValueError, "CustomLogAppender.appendLogMessage: doLogMessage is nil")
|
||||
else: cla.doLogMessage(cla.state, msg)
|
||||
except Exception:
|
||||
ls.global.reportLoggingError(
|
||||
ls.shared.reportLoggingError(
|
||||
getCurrentException(),
|
||||
"unable to append to CustomLogAppender")
|
||||
|
||||
@@ -822,7 +821,7 @@ method appendLogMessage(
|
||||
else:
|
||||
cla.appendLogMessageSingleThreaded(ls, msg)
|
||||
except Exception:
|
||||
ls.global.reportLoggingError(
|
||||
ls.shared.reportLoggingError(
|
||||
getCurrentException(),
|
||||
"unable to append to ConsoleLogAppender")
|
||||
|
||||
@@ -892,7 +891,7 @@ method appendLogMessage(
|
||||
else:
|
||||
fla.appendLogMessageSingleThreaded(ls, msg)
|
||||
except Exception:
|
||||
ls.global.reportLoggingError(
|
||||
ls.shared.reportLoggingError(
|
||||
getCurrentException(),
|
||||
"unable to append to FileLogAppender")
|
||||
|
||||
@@ -958,22 +957,22 @@ when isMainModule:
|
||||
import std/[files, tempfiles, unittest]
|
||||
import ./namespaced_logging/testutil
|
||||
|
||||
suite "GlobalLogService Initialization":
|
||||
suite "LogService Initialization":
|
||||
|
||||
test "initLogService creates valid service":
|
||||
let ls = initLogService()
|
||||
let local = ls.threadLocalRef()
|
||||
check:
|
||||
ls.global != nil
|
||||
ls.global.owners.load() == 2
|
||||
ls.global.configVersion.load() == 0
|
||||
ls.global.rootLevel.load() == lvlAll
|
||||
ls.shared != nil
|
||||
ls.shared.owners.load() == 2
|
||||
ls.shared.configVersion.load() == 0
|
||||
ls.shared.rootLevel.load() == lvlAll
|
||||
local.configVersion >= 0
|
||||
local.configVersion == ls.global.configVersion.load
|
||||
local.configVersion == ls.shared.configVersion.load
|
||||
|
||||
test "initLogService with custom root level":
|
||||
let ls = initLogService(lvlInfo)
|
||||
check ls.global.rootLevel.load() == lvlInfo
|
||||
check ls.shared.rootLevel.load() == lvlInfo
|
||||
|
||||
test "threadLocalRef builds independent caches for an unchanged service":
|
||||
var service = initLogService()
|
||||
@@ -982,7 +981,7 @@ when isMainModule:
|
||||
let first = service.threadLocalRef()
|
||||
let second = service.threadLocalRef()
|
||||
check:
|
||||
first.global == second.global
|
||||
first.shared == second.shared
|
||||
first.configVersion == second.configVersion
|
||||
cast[pointer](first.thresholds) != cast[pointer](second.thresholds)
|
||||
first.thresholds["worker"] == second.thresholds["worker"]
|
||||
@@ -996,7 +995,7 @@ when isMainModule:
|
||||
|
||||
test "setRootThreshold updates level":
|
||||
ls.setRootThreshold(lvlError)
|
||||
check ls.global.rootLevel.load() == lvlError
|
||||
check ls.shared.rootLevel.load() == lvlError
|
||||
|
||||
test "setThreshold adds namespace threshold":
|
||||
ls.setThreshold("api", lvlWarn)
|
||||
@@ -1004,9 +1003,9 @@ when isMainModule:
|
||||
|
||||
test "ensureFreshness syncs configuration":
|
||||
# Modify global state directly
|
||||
withLock ls.global.lock:
|
||||
ls.global.thresholds["test"] = lvlError
|
||||
ls.global.configVersion.atomicInc()
|
||||
withLock ls.shared.lock:
|
||||
ls.shared.thresholds["test"] = lvlError
|
||||
ls.shared.configVersion.atomicInc()
|
||||
|
||||
# Local state should be out of sync
|
||||
check not ls.thresholds.hasKey("test")
|
||||
@@ -1021,18 +1020,18 @@ when isMainModule:
|
||||
ls.setThreshold("api", lvlWarn)
|
||||
|
||||
check:
|
||||
ls.configVersion == ls.global.configVersion.load
|
||||
ls.configVersion == ls.shared.configVersion.load
|
||||
ls.configVersion != ls2.configVersion
|
||||
ls2.configVersion != ls.global.configVersion.load
|
||||
ls2.configVersion != ls.shared.configVersion.load
|
||||
ls.thresholds["api"] == lvlWarn
|
||||
ls.global.thresholds["api"] == lvlWarn
|
||||
ls.shared.thresholds["api"] == lvlWarn
|
||||
not ls2.thresholds.contains("api")
|
||||
|
||||
ls2.ensureFreshness()
|
||||
|
||||
check:
|
||||
ls.configVersion == ls2.configVersion
|
||||
ls2.configVersion == ls.global.configVersion.load
|
||||
ls2.configVersion == ls.shared.configVersion.load
|
||||
ls2.thresholds["api"] == lvlWarn
|
||||
|
||||
test "ensureFreshness syncs configuration across threads":
|
||||
@@ -1083,13 +1082,13 @@ when isMainModule:
|
||||
joinThreads(tA, tB)
|
||||
|
||||
test "configuration version incrementing":
|
||||
let initialVersion = ls.global.configVersion.load()
|
||||
let initialVersion = ls.shared.configVersion.load()
|
||||
|
||||
ls.setRootThreshold(lvlWarn)
|
||||
check ls.global.configVersion.load() > initialVersion
|
||||
check ls.shared.configVersion.load() > initialVersion
|
||||
|
||||
ls.setThreshold("test", lvlError)
|
||||
check ls.global.configVersion.load() > initialVersion + 1
|
||||
check ls.shared.configVersion.load() > initialVersion + 1
|
||||
|
||||
|
||||
suite "Logger Creation and Usage":
|
||||
|
||||
Reference in new issue
Block a user