logos-storage/nim-datastore
1
0
Fork 0
mirror of https://github.com/logos-storage/nim-datastore.git synced 2026-01-03 22:23:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

setup flags

Browse Source
This commit is contained in:
Jaremy Creechley 2023-09-18 13:23:05 -07:00
parent c96ae7c016
commit d2a8a72a7d
No known key found for this signature in database
GPG Key ID: 4E66FB67B21D3300
1 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
datastore/threads/threadproxyds.nim
View File

@ -38,7 +38,7 @@ type
ds: ptr Datastore ds: ptr Datastore
res: ThreadResult[T] res: ThreadResult[T]
cancelled: Atomic[bool] cancelled: Atomic[bool]
hasStarted: Atomic[bool] isActive: Atomic[bool]
semaphore: AsyncSemaphore semaphore: AsyncSemaphore
signal: ThreadSignalPtr signal: ThreadSignalPtr
@ -93,11 +93,12 @@ template dispatchTask(
result = failure(ctx.res.error()) # TODO: fix this, result shouldn't be accessed result = failure(ctx.res.error()) # TODO: fix this, result shouldn't be accessed
except CancelledError as exc: except CancelledError as exc:
trace "Cancelling thread future!", exc = exc.msg trace "Cancelling thread future!", exc = exc.msg
if ctx.hasStarted.load(moAcquireRelease): if ctx.isActive.load(moAcquireRelease):
# could do a spinlock here, but for now # could do a spinlock here until the other side cancels,
# it'd be better to leak than possibly corrupt memory # but for now it'd at least be better to leak than possibly
# since it's easier to detect and fix leaks # corrupt memory since it's easier to detect and fix leaks
GC_ref(ctx) GC_ref(ctx)
warn "request was cancelled while thread task is running", exc = exc.msg
ctx.cancelled.store(true, moAcquireRelease) ctx.cancelled.store(true, moAcquireRelease)
await ctx.signal.fire() await ctx.signal.fire()
raise exc raise exc
@ -110,7 +111,7 @@ proc signalMonitor[T](ctx: ptr TaskCtx, fut: Future[T]) {.async.} =
## ##
try: try:
ctx[].hasStarted.store(true, moAcquireRelease) ctx[].isActive.store(true, moAcquireRelease)
await ctx[].signal.wait() await ctx[].signal.wait()
trace "Received signal" trace "Received signal"
@ -124,6 +125,8 @@ proc signalMonitor[T](ctx: ptr TaskCtx, fut: Future[T]) {.async.} =
trace "Exception in thread signal monitor", exc = exc.msg trace "Exception in thread signal monitor", exc = exc.msg
ctx.res.err(exc) ctx.res.err(exc)
discard ctx[].signal.fireSync() discard ctx[].signal.fireSync()
finally:
ctx[].isActive.store(false, moAcquireRelease)
proc asyncHasTask( proc asyncHasTask(
ctx: ptr TaskCtx[bool], ctx: ptr TaskCtx[bool],