This commit is contained in:
Jaremy Creechley 2023-10-20 16:22:00 -07:00
parent 4844f7dad0
commit ab2e7b53ce
No known key found for this signature in database
GPG Key ID: 4E66FB67B21D3300
3 changed files with 94 additions and 91 deletions

View File

@ -280,6 +280,7 @@ else:
proc new*(
T: type FSDatastore,
root: string,
tp: Taskpool = nil,
depth = 2,
caseSensitive = true,
ignoreProtected = false): ?!FSDatastore =

View File

@ -274,6 +274,7 @@ else:
proc new*(
T: type SQLiteDatastore,
path: string,
tp: Taskpool = nil,
readOnly = false): ?!T =
let
@ -281,14 +282,14 @@ else:
if readOnly: SQLITE_OPEN_READONLY
else: SQLITE_OPEN_READWRITE or SQLITE_OPEN_CREATE
success T(
db: ? SQLiteDsDb.open(path, flags),
success SQLiteDatastore(
db: ? SQLiteDsDb[string, seq[byte]].open(path, flags),
readOnly: readOnly)
proc new*(
T: type SQLiteDatastore,
db: SQLiteDsDb): ?!T =
db: SQLiteDsDb): ?!SQLiteDatastore =
success T(
success SQLiteDatastore(
db: db,
readOnly: db.readOnly)

View File

@ -126,110 +126,111 @@ suite "Test Query ThreadDatastore with fsds":
queryTests(ds, false)
for i in 1..N:
suite "Test ThreadDatastore cancelations":
when datastoreUseAsync:
for i in 1..N:
suite "Test ThreadDatastore cancelations":
privateAccess(SQLiteDatastore) # expose private fields
privateAccess(ThreadProxy) # expose private fields
privateAccess(TaskCtx) # expose private fields
privateAccess(SQLiteDatastore) # expose private fields
privateAccess(ThreadProxy) # expose private fields
privateAccess(TaskCtx) # expose private fields
var sds: SQLiteDatastore
var sds: SQLiteDatastore
setupAll:
sds = SQLiteDatastore.new(Memory, tp = taskPool).tryGet()
setupAll:
sds = SQLiteDatastore.new(Memory, tp = taskPool).tryGet()
teardown:
GC_fullCollect() # run full collect after each test
teardown:
GC_fullCollect() # run full collect after each test
test "Should monitor signal and cancel":
var
signal = ThreadSignalPtr.new().tryGet()
test "Should monitor signal and cancel":
var
signal = ThreadSignalPtr.new().tryGet()
proc cancelTestTask(ctx: TaskCtx[bool]) {.gcsafe.} =
executeTask(ctx):
(?!bool).ok(true)
proc cancelTestTask(ctx: TaskCtx[bool]) {.gcsafe.} =
executeTask(ctx):
(?!bool).ok(true)
let ctx = newTaskCtx(bool, signal=signal)
ctx[].cancelled = true
dispatchTask(sds.db, signal):
sds.db.tp.spawn cancelTestTask(ctx)
let ctx = newTaskCtx(bool, signal=signal)
ctx[].cancelled = true
dispatchTask(sds.db, signal):
sds.db.tp.spawn cancelTestTask(ctx)
check:
ctx[].res.isErr == true
ctx[].cancelled == true
ctx[].running == false
check:
ctx[].res.isErr == true
ctx[].cancelled == true
ctx[].running == false
test "Should cancel future":
test "Should cancel future":
var
signal = ThreadSignalPtr.new().tryGet()
ms {.global.}: MutexSignal
flag {.global.}: Atomic[bool]
futFreed {.global.}: Atomic[bool]
ready {.global.}: Atomic[bool]
var
signal = ThreadSignalPtr.new().tryGet()
ms {.global.}: MutexSignal
flag {.global.}: Atomic[bool]
futFreed {.global.}: Atomic[bool]
ready {.global.}: Atomic[bool]
ms.init()
ms.init()
type
FutTestObj = object
val: int
TestValue = object
ThreadTestInt = (TestValue, )
type
FutTestObj = object
val: int
TestValue = object
ThreadTestInt = (TestValue, )
proc `=destroy`(obj: var TestValue) =
# echo "destroy TestObj!"
flag.store(true)
proc `=destroy`(obj: var TestValue) =
# echo "destroy TestObj!"
flag.store(true)
proc `=destroy`(obj: var FutTestObj) =
# echo "destroy FutTestObj!"
futFreed.store(true)
proc `=destroy`(obj: var FutTestObj) =
# echo "destroy FutTestObj!"
futFreed.store(true)
proc wait(flag: var Atomic[bool], name = "task") =
# echo "wait for " & name & " to be ready..."
# defer: echo ""
for i in 1..100:
# stdout.write(".")
if flag.load() == true:
return
os.sleep(10)
raise newException(Defect, "timeout")
proc wait(flag: var Atomic[bool], name = "task") =
# echo "wait for " & name & " to be ready..."
# defer: echo ""
for i in 1..100:
# stdout.write(".")
if flag.load() == true:
return
os.sleep(10)
raise newException(Defect, "timeout")
proc errorTestTask(ctx: TaskCtx[ThreadTestInt]) {.gcsafe, nimcall.} =
executeTask(ctx):
# echo "task:exec"
discard ctx[].signal.fireSync()
ready.store(true)
ms.wait()
echo "task context memory: ", ctx[]
(?!ThreadTestInt).ok(default(ThreadTestInt))
proc errorTestTask(ctx: TaskCtx[ThreadTestInt]) {.gcsafe, nimcall.} =
executeTask(ctx):
# echo "task:exec"
discard ctx[].signal.fireSync()
ready.store(true)
ms.wait()
echo "task context memory: ", ctx[]
(?!ThreadTestInt).ok(default(ThreadTestInt))
proc runTestTask() {.async.} =
let obj = FutTestObj(val: 42)
await sleepAsync(1.milliseconds)
proc runTestTask() {.async.} =
let obj = FutTestObj(val: 42)
await sleepAsync(1.milliseconds)
try:
let ctx = newTaskCtx(ThreadTestInt, signal=signal)
dispatchTask(sds.db, signal):
sds.db.tp.spawn errorTestTask(ctx)
ready.wait()
# echo "raise error"
raise newException(ValueError, "fake error")
finally:
# echo "fut FutTestObj: ", obj
assert obj.val == 42 # need to force future to keep ref here
try:
let ctx = newTaskCtx(ThreadTestInt, signal=signal)
dispatchTask(sds.db, signal):
sds.db.tp.spawn errorTestTask(ctx)
ready.wait()
# echo "raise error"
raise newException(ValueError, "fake error")
block:
await runTestTask()
except CatchableError as exc:
# echo "caught: ", $exc
discard
finally:
# echo "fut FutTestObj: ", obj
assert obj.val == 42 # need to force future to keep ref here
try:
block:
await runTestTask()
except CatchableError as exc:
# echo "caught: ", $exc
discard
finally:
# echo "finish"
check ready.load() == true
GC_fullCollect()
futFreed.wait("futFreed")
echo "future freed it's mem!"
check futFreed.load() == true
# echo "finish"
check ready.load() == true
GC_fullCollect()
futFreed.wait("futFreed")
echo "future freed it's mem!"
check futFreed.load() == true
ms.fire()
flag.wait("flag")
check flag.load() == true
ms.fire()
flag.wait("flag")
check flag.load() == true