From b02c466b946f85ea04d671ef69aebada1ee795c1 Mon Sep 17 00:00:00 2001 From: Jaremy Creechley Date: Tue, 19 Sep 2023 14:08:41 -0700 Subject: [PATCH] run a full GC and repreat the tests --- tests/datastore/testthreadproxyds.nim | 354 ++++++++++++++------------ 1 file changed, 185 insertions(+), 169 deletions(-) diff --git a/tests/datastore/testthreadproxyds.nim b/tests/datastore/testthreadproxyds.nim index 10197df..6bff32f 100644 --- a/tests/datastore/testthreadproxyds.nim +++ b/tests/datastore/testthreadproxyds.nim @@ -23,188 +23,204 @@ import ./querycommontests const NumThreads = 200 # IO threads aren't attached to CPU count -suite "Test Basic ThreadDatastore with SQLite": +proc testBasic() = - var - sqlStore: Datastore - ds: ThreadDatastore - taskPool: Taskpool - key = Key.init("/a/b").tryGet() - bytes = "some bytes".toBytes - otherBytes = "some other bytes".toBytes - - setupAll: - sqlStore = SQLiteDatastore.new(Memory).tryGet() - taskPool = Taskpool.new(NumThreads) - ds = ThreadDatastore.new(sqlStore, tp = taskPool).tryGet() - - teardownAll: - (await ds.close()).tryGet() - taskPool.shutdown() - - basicStoreTests(ds, key, bytes, otherBytes) - -suite "Test Query ThreadDatastore with SQLite": - - var - sqlStore: Datastore - ds: ThreadDatastore - taskPool: Taskpool - key = Key.init("/a/b").tryGet() - bytes = "some bytes".toBytes - otherBytes = "some other bytes".toBytes - - setup: - sqlStore = SQLiteDatastore.new(Memory).tryGet() - taskPool = Taskpool.new(NumThreads) - ds = ThreadDatastore.new(sqlStore, tp = taskPool).tryGet() - - teardown: - (await ds.close()).tryGet() - taskPool.shutdown() - - queryTests(ds, true) - -suite "Test Basic ThreadDatastore with fsds": - let - path = currentSourcePath() # get this file's name - basePath = "tests_data" - basePathAbs = path.parentDir / basePath - key = Key.init("/a/b").tryGet() - bytes = "some bytes".toBytes - otherBytes = "some other bytes".toBytes - - var - fsStore: FSDatastore - ds: ThreadDatastore - taskPool: Taskpool - - setupAll: - removeDir(basePathAbs) - require(not dirExists(basePathAbs)) - createDir(basePathAbs) - - fsStore = FSDatastore.new(root = basePathAbs, depth = 3).tryGet() - taskPool = Taskpool.new(NumThreads) - ds = ThreadDatastore.new(fsStore, withLocks = true, tp = taskPool).tryGet() - - teardownAll: - (await ds.close()).tryGet() - taskPool.shutdown() - - removeDir(basePathAbs) - require(not dirExists(basePathAbs)) - - basicStoreTests(fsStore, key, bytes, otherBytes) - -suite "Test Query ThreadDatastore with fsds": - let - path = currentSourcePath() # get this file's name - basePath = "tests_data" - basePathAbs = path.parentDir / basePath - - var - fsStore: FSDatastore - ds: ThreadDatastore - taskPool: Taskpool - - setup: - removeDir(basePathAbs) - require(not dirExists(basePathAbs)) - createDir(basePathAbs) - - fsStore = FSDatastore.new(root = basePathAbs, depth = 5).tryGet() - taskPool = Taskpool.new(NumThreads) - ds = ThreadDatastore.new(fsStore, withLocks = true, tp = taskPool).tryGet() - - teardown: - (await ds.close()).tryGet() - taskPool.shutdown() - - removeDir(basePathAbs) - require(not dirExists(basePathAbs)) - - queryTests(ds, false) - -suite "Test ThreadDatastore cancelations": - var - sqlStore: Datastore - ds: ThreadDatastore - taskPool: Taskpool - key = Key.init("/a/b").tryGet() - bytes = "some bytes".toBytes - otherBytes = "some other bytes".toBytes - - privateAccess(ThreadDatastore) # expose private fields - privateAccess(TaskCtx) # expose private fields - - setupAll: - sqlStore = SQLiteDatastore.new(Memory).tryGet() - taskPool = Taskpool.new(NumThreads) - ds = ThreadDatastore.new(sqlStore, tp = taskPool).tryGet() - - test "Should monitor signal and cancel": - var - signal = ThreadSignalPtr.new().tryGet() - res = ThreadResult[void]() - ctx = TaskCtx[void]( - ds: sqlStore, - res: addr res, - signal: signal) - fut = newFuture[void]("signalMonitor") - threadArgs = (addr ctx, addr fut) + suite "Test Basic ThreadDatastore with SQLite": var - thread: Thread[type threadArgs] + sqlStore: Datastore + ds: ThreadDatastore + taskPool: Taskpool + key = Key.init("/a/b").tryGet() + bytes = "some bytes".toBytes + otherBytes = "some other bytes".toBytes - proc threadTask(args: type threadArgs) = - var (ctx, fut) = args - proc asyncTask() {.async.} = - let - monitor = signalMonitor(ctx, fut[]) + setupAll: + sqlStore = SQLiteDatastore.new(Memory).tryGet() + taskPool = Taskpool.new(NumThreads) + ds = ThreadDatastore.new(sqlStore, tp = taskPool).tryGet() - await monitor + teardownAll: + (await ds.close()).tryGet() + taskPool.shutdown() - waitFor asyncTask() + basicStoreTests(ds, key, bytes, otherBytes) + GC_fullCollect() - createThread(thread, threadTask, threadArgs) - ctx.cancelled = true - check: ctx.signal.fireSync.tryGet +for i in 1..100: + testBasic() + GC_fullCollect() - joinThreads(thread) - - check: fut.cancelled - check: ctx.signal.close().isOk - - test "Should monitor and not cancel": - var - signal = ThreadSignalPtr.new().tryGet() - res = ThreadResult[void]() - ctx = TaskCtx[void]( - ds: sqlStore, - res: addr res, - signal: signal) - fut = newFuture[void]("signalMonitor") - threadArgs = (addr ctx, addr fut) +proc testQuery() = + suite "Test Query ThreadDatastore with SQLite": var - thread: Thread[type threadArgs] + sqlStore: Datastore + ds: ThreadDatastore + taskPool: Taskpool + key = Key.init("/a/b").tryGet() + bytes = "some bytes".toBytes + otherBytes = "some other bytes".toBytes - proc threadTask(args: type threadArgs) = - var (ctx, fut) = args - proc asyncTask() {.async.} = - let - monitor = signalMonitor(ctx, fut[]) + setup: + sqlStore = SQLiteDatastore.new(Memory).tryGet() + taskPool = Taskpool.new(NumThreads) + ds = ThreadDatastore.new(sqlStore, tp = taskPool).tryGet() - await monitor + teardown: + (await ds.close()).tryGet() + taskPool.shutdown() - waitFor asyncTask() + queryTests(ds, true) + GC_fullCollect() - createThread(thread, threadTask, threadArgs) - ctx.cancelled = false - check: ctx.signal.fireSync.tryGet +for i in 1..100: + testQuery() + GC_fullCollect() - joinThreads(thread) +proc testFsDs() = + suite "Test Basic ThreadDatastore with fsds": + let + path = currentSourcePath() # get this file's name + basePath = "tests_data" + basePathAbs = path.parentDir / basePath + key = Key.init("/a/b").tryGet() + bytes = "some bytes".toBytes + otherBytes = "some other bytes".toBytes - check: not fut.cancelled - check: ctx.signal.close().isOk + var + fsStore: FSDatastore + ds: ThreadDatastore + taskPool: Taskpool + + setupAll: + removeDir(basePathAbs) + require(not dirExists(basePathAbs)) + createDir(basePathAbs) + + fsStore = FSDatastore.new(root = basePathAbs, depth = 3).tryGet() + taskPool = Taskpool.new(NumThreads) + ds = ThreadDatastore.new(fsStore, withLocks = true, tp = taskPool).tryGet() + + teardownAll: + (await ds.close()).tryGet() + taskPool.shutdown() + + removeDir(basePathAbs) + require(not dirExists(basePathAbs)) + + basicStoreTests(fsStore, key, bytes, otherBytes) + +proc testFsDsQuery() = + suite "Test Query ThreadDatastore with fsds": + let + path = currentSourcePath() # get this file's name + basePath = "tests_data" + basePathAbs = path.parentDir / basePath + + var + fsStore: FSDatastore + ds: ThreadDatastore + taskPool: Taskpool + + setup: + removeDir(basePathAbs) + require(not dirExists(basePathAbs)) + createDir(basePathAbs) + + fsStore = FSDatastore.new(root = basePathAbs, depth = 5).tryGet() + taskPool = Taskpool.new(NumThreads) + ds = ThreadDatastore.new(fsStore, withLocks = true, tp = taskPool).tryGet() + + teardown: + (await ds.close()).tryGet() + taskPool.shutdown() + + removeDir(basePathAbs) + require(not dirExists(basePathAbs)) + + queryTests(ds, false) + +proc testCancels() = + suite "Test ThreadDatastore cancelations": + var + sqlStore: Datastore + ds: ThreadDatastore + taskPool: Taskpool + key = Key.init("/a/b").tryGet() + bytes = "some bytes".toBytes + otherBytes = "some other bytes".toBytes + + privateAccess(ThreadDatastore) # expose private fields + privateAccess(TaskCtx) # expose private fields + + setupAll: + sqlStore = SQLiteDatastore.new(Memory).tryGet() + taskPool = Taskpool.new(NumThreads) + ds = ThreadDatastore.new(sqlStore, tp = taskPool).tryGet() + + test "Should monitor signal and cancel": + var + signal = ThreadSignalPtr.new().tryGet() + res = ThreadResult[void]() + ctx = TaskCtx[void]( + ds: sqlStore, + res: addr res, + signal: signal) + fut = newFuture[void]("signalMonitor") + threadArgs = (addr ctx, addr fut) + + var + thread: Thread[type threadArgs] + + proc threadTask(args: type threadArgs) = + var (ctx, fut) = args + proc asyncTask() {.async.} = + let + monitor = signalMonitor(ctx, fut[]) + + await monitor + + waitFor asyncTask() + + createThread(thread, threadTask, threadArgs) + ctx.cancelled = true + check: ctx.signal.fireSync.tryGet + + joinThreads(thread) + + check: fut.cancelled + check: ctx.signal.close().isOk + + test "Should monitor and not cancel": + var + signal = ThreadSignalPtr.new().tryGet() + res = ThreadResult[void]() + ctx = TaskCtx[void]( + ds: sqlStore, + res: addr res, + signal: signal) + fut = newFuture[void]("signalMonitor") + threadArgs = (addr ctx, addr fut) + + var + thread: Thread[type threadArgs] + + proc threadTask(args: type threadArgs) = + var (ctx, fut) = args + proc asyncTask() {.async.} = + let + monitor = signalMonitor(ctx, fut[]) + + await monitor + + waitFor asyncTask() + + createThread(thread, threadTask, threadArgs) + ctx.cancelled = false + check: ctx.signal.fireSync.tryGet + + joinThreads(thread) + + check: not fut.cancelled + check: ctx.signal.close().isOk