Define explicit ownership for custom appenders and formatters
Add SharedLogState for synchronized value data and factories for independently owned appender state and captured formatter closures. Reject unsafe configuration state, update the test collector to value snapshots, and document the breaking APIs in version 3.0.0. Expand ORC ownership and sanitizer coverage and reject stale test executables.
This commit is contained in:
jdb committed
1 parent
851caef706
commit
b2d0e9dade
11 files changed
+808
-110
No files matched your search
+173
-9
@@ -1,9 +1,12 @@
|
||||
## Run each scenario in its own ThreadSanitizer-instrumented process.
|
||||
## Success means no race was reported; a crash is not required to expose UB.
|
||||
import std/[atomics, locks, os]
|
||||
import std/[atomics, json, locks, options, os, times]
|
||||
import ../src/namespaced_logging
|
||||
import ../src/namespaced_logging/testutil
|
||||
import ./runtime_allocator
|
||||
|
||||
const iterations = 1_000
|
||||
const collectorIterations = 100
|
||||
var start: Atomic[bool]
|
||||
var cloneLock: Lock
|
||||
var ready: Atomic[int]
|
||||
@@ -73,11 +76,11 @@ proc reconfigureWorker(service: LogService) {.thread.} =
|
||||
doAssert local.getLogger("worker").isEnabled(lvlInfo)
|
||||
|
||||
type
|
||||
CustomState = ref object
|
||||
CustomState = object
|
||||
value: int
|
||||
StatefulAppender = CustomLogAppender[CustomState]
|
||||
StatefulAppender = CustomLogAppender[SharedLogState[CustomState]]
|
||||
|
||||
proc ignoreMessage(state: CustomState, msg: LogMessage) {.gcsafe, noconv.} =
|
||||
proc ignoreMessage(state: var SharedLogState[CustomState], msg: LogMessage) {.gcsafe, noconv.} =
|
||||
discard
|
||||
|
||||
proc stateWorker(source: ptr StatefulAppender) {.thread.} =
|
||||
@@ -86,9 +89,9 @@ proc stateWorker(source: ptr StatefulAppender) {.thread.} =
|
||||
var copied: StatefulAppender
|
||||
withLock cloneLock:
|
||||
copied = StatefulAppender(clone(source[]))
|
||||
doAssert copied.state.value == 42
|
||||
doAssert copied.state.snapshot().value == 42
|
||||
|
||||
proc makeFormatter(prefix: string): LogMessageFormatter =
|
||||
proc makeFormatter(prefix: string): StatefulLogFormatterFunc =
|
||||
result = proc(msg: LogMessage): string {.gcsafe.} = prefix & msg.msg
|
||||
|
||||
proc formatterWorker(source: ptr ConsoleLogAppender) {.thread.} =
|
||||
@@ -100,7 +103,109 @@ proc formatterWorker(source: ptr ConsoleLogAppender) {.thread.} =
|
||||
var copied: ConsoleLogAppender
|
||||
withLock cloneLock:
|
||||
copied = ConsoleLogAppender(clone(source[]))
|
||||
doAssert copied.formatter(msg) == "prefix: message"
|
||||
doAssert copied.formatter.format(msg) == "prefix: message"
|
||||
|
||||
type
|
||||
SharedPayload = object
|
||||
entries: seq[string]
|
||||
destructions: ptr Atomic[int]
|
||||
finalCount: ptr Atomic[int]
|
||||
|
||||
proc `=destroy`(value: SharedPayload) =
|
||||
if value.destructions != nil:
|
||||
value.finalCount[].store(value.entries.len)
|
||||
value.destructions[].atomicInc()
|
||||
`=destroy`(value.entries)
|
||||
|
||||
proc sharedStateWorker(source: ptr SharedLogState[SharedPayload]) {.thread.} =
|
||||
let state = source[]
|
||||
ready.atomicInc()
|
||||
awaitStart()
|
||||
for i in 0 ..< iterations:
|
||||
var copied = state
|
||||
let moved = move(copied)
|
||||
moved.withState(proc(value: var SharedPayload) =
|
||||
value.entries.add("entry")
|
||||
doAssert value.entries[^1] == "entry")
|
||||
|
||||
type
|
||||
FactoryState = ref object
|
||||
count: int
|
||||
self: FactoryState
|
||||
output: SharedLogState[seq[int]]
|
||||
|
||||
proc makeState(output: SharedLogState[seq[int]]): FactoryState =
|
||||
result = FactoryState(output: output)
|
||||
result.self = result
|
||||
|
||||
proc logState(state: var FactoryState, msg: LogMessage) {.gcsafe, noconv.} =
|
||||
inc state.count
|
||||
doAssert state.self == state
|
||||
let count = state.count
|
||||
state.output.withState(proc(values: var seq[int]) = values.add(count))
|
||||
|
||||
proc factoryWorker(source: ptr LogService) {.thread.} =
|
||||
block:
|
||||
let local = source[].threadLocalRef()
|
||||
ready.atomicInc()
|
||||
awaitStart()
|
||||
let logger = local.getLogger("worker")
|
||||
let stamp = fromUnix(0).utc
|
||||
for i in 0 ..< iterations:
|
||||
logger.log(LogMessage(scope: "worker", level: lvlInfo, ts: stamp, msg: "message"))
|
||||
GC_fullCollect()
|
||||
|
||||
proc timestampWorker() {.thread.} =
|
||||
awaitStart()
|
||||
for i in 0 ..< iterations:
|
||||
let msg = initLogMessage("worker", lvlInfo, "message")
|
||||
doAssert msg.msg == "message"
|
||||
|
||||
proc collectorWorker(collector: LoggedMessages) {.thread.} =
|
||||
# Construct every managed field in this worker, including the timezone.
|
||||
let stamp = fromUnix(0).utc
|
||||
awaitStart()
|
||||
for i in 0 ..< collectorIterations:
|
||||
collector.add(LogMessage(scope: "worker", level: lvlInfo, ts: stamp,
|
||||
msg: "message", err: some[ref Exception](newException(ValueError, "error")),
|
||||
additionalData: %*{"number": i}))
|
||||
var snapshot = collector.get()
|
||||
doAssert snapshot.len > 0
|
||||
doAssert snapshot[^1].err.get.msg == "error"
|
||||
snapshot[0].msg = "private copy"
|
||||
doAssert collector.get()[0].msg == "message"
|
||||
|
||||
type
|
||||
FormatterSeed = object
|
||||
prefix: string
|
||||
destroyed: ptr Atomic[int]
|
||||
FormatterProbeObj = object
|
||||
count: int
|
||||
destroyed: ptr Atomic[int]
|
||||
FormatterProbe = ref FormatterProbeObj
|
||||
|
||||
proc `=destroy`(probe: FormatterProbeObj) =
|
||||
if probe.destroyed != nil: probe.destroyed[].atomicInc()
|
||||
|
||||
proc makeCountedFormatter(seed: FormatterSeed): StatefulLogFormatterFunc =
|
||||
let probe = FormatterProbe(destroyed: seed.destroyed)
|
||||
result = proc(msg: LogMessage): string {.gcsafe.} =
|
||||
inc probe.count
|
||||
seed.prefix & $probe.count & msg.msg
|
||||
|
||||
proc formatterLifetimeWorker(source: ptr LogFormatter) {.thread.} =
|
||||
block:
|
||||
var formatter: LogFormatter
|
||||
withLock cloneLock:
|
||||
formatter = clone(source[])
|
||||
ready.atomicInc()
|
||||
awaitStart()
|
||||
doAssert formatter.format(LogMessage(msg: "!")) == "count 1!"
|
||||
for i in 0 ..< iterations:
|
||||
let copied = clone(formatter)
|
||||
doAssert copied.format(LogMessage(msg: "!")) == "count 1!"
|
||||
doAssert copied.format(LogMessage(msg: "!")) == "count 2!"
|
||||
GC_fullCollect()
|
||||
|
||||
proc main() =
|
||||
doAssert paramCount() == 1
|
||||
@@ -125,7 +230,7 @@ proc main() =
|
||||
start.store(true)
|
||||
joinThreads(workers)
|
||||
of "custom-state":
|
||||
var source = initCustomLogAppender(CustomState(value: 42), ignoreMessage)
|
||||
var source = initCustomLogAppender(initSharedLogState(CustomState(value: 42)), ignoreMessage)
|
||||
var workers: array[4, Thread[ptr StatefulAppender]]
|
||||
for i in 0 ..< workers.len:
|
||||
createThread(workers[i], stateWorker, addr source)
|
||||
@@ -159,12 +264,71 @@ proc main() =
|
||||
joinThreads(workers)
|
||||
of "formatter":
|
||||
var source = initConsoleLogAppender()
|
||||
source.formatter = makeFormatter("prefix: ")
|
||||
source.formatter = initLogFormatter("prefix: ", makeFormatter)
|
||||
var workers: array[4, Thread[ptr ConsoleLogAppender]]
|
||||
for i in 0 ..< workers.len:
|
||||
createThread(workers[i], formatterWorker, addr source)
|
||||
start.store(true)
|
||||
joinThreads(workers)
|
||||
of "shared-state-lifetime":
|
||||
var destructions, finalCount: Atomic[int]
|
||||
var state = initSharedLogState(SharedPayload(
|
||||
destructions: addr destructions, finalCount: addr finalCount))
|
||||
var workers: array[4, Thread[ptr SharedLogState[SharedPayload]]]
|
||||
for i in 0 ..< workers.len:
|
||||
createThread(workers[i], sharedStateWorker, addr state)
|
||||
while ready.load() != workers.len: discard
|
||||
state = default(typeof(state))
|
||||
start.store(true)
|
||||
joinThreads(workers)
|
||||
doAssert destructions.load() == 1
|
||||
doAssert finalCount.load() == workers.len * iterations
|
||||
of "factory-state":
|
||||
let output = initSharedLogState(newSeq[int]())
|
||||
var service = initLogService()
|
||||
service.addAppender(initFactoryCustomLogAppender(output, makeState, logState))
|
||||
var workers: array[4, Thread[ptr LogService]]
|
||||
for i in 0 ..< workers.len:
|
||||
createThread(workers[i], factoryWorker, addr service)
|
||||
while ready.load() != workers.len: discard
|
||||
service = LogService()
|
||||
start.store(true)
|
||||
joinThreads(workers)
|
||||
let values = output.snapshot()
|
||||
doAssert values.len == workers.len * iterations
|
||||
var counts: array[iterations + 1, int]
|
||||
for value in values: inc counts[value]
|
||||
for i in 1 .. iterations: doAssert counts[i] == workers.len
|
||||
of "collector":
|
||||
let collector = initLoggedMessages()
|
||||
var workers: array[4, Thread[LoggedMessages]]
|
||||
for i in 0 ..< workers.len:
|
||||
createThread(workers[i], collectorWorker, collector)
|
||||
start.store(true)
|
||||
joinThreads(workers)
|
||||
doAssert collector.get().len == workers.len * collectorIterations
|
||||
of "formatter-lifetime":
|
||||
var destroyed: Atomic[int]
|
||||
var formatter = initLogFormatter(
|
||||
FormatterSeed(prefix: "count ", destroyed: addr destroyed), makeCountedFormatter)
|
||||
doAssert formatter.format(LogMessage(msg: "!")) == "count 1!"
|
||||
var workers: array[4, Thread[ptr LogFormatter]]
|
||||
for i in 0 ..< workers.len:
|
||||
createThread(workers[i], formatterLifetimeWorker, addr formatter)
|
||||
while ready.load() != workers.len: discard
|
||||
formatter = nil
|
||||
start.store(true)
|
||||
joinThreads(workers)
|
||||
GC_fullCollect()
|
||||
doAssert destroyed.load() == 1 + workers.len * (iterations + 1)
|
||||
of "timestamp":
|
||||
var workers: array[4, Thread[void]]
|
||||
for i in 0 ..< workers.len:
|
||||
createThread(workers[i], timestampWorker)
|
||||
start.store(true)
|
||||
joinThreads(workers)
|
||||
of "allocator":
|
||||
runAllocatorRegression()
|
||||
else:
|
||||
raise newException(ValueError, "unknown thread safety scenario")
|
||||
|
||||
|
||||
Reference in new issue
Block a user