status-im
/
nim-dagger
mirror of https://github.com/status-im/nim-dagger.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

fix(asyncstatemachine): fixes not awaiting or asyncSpawning futures 

- adds a break in scheduler when CancelledError is caught
- tracks asyncSpawned state.run, so that it can be cancelled during stop
- removes usages of `then`
- ensures that no exceptions are leaked from async procs
This commit is contained in:
Eric 2024-12-09 18:18:53 +11:00
parent 19af79786e
commit dc244a36bd
No known key found for this signature in database
1 changed files with 27 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
codex/utils/asyncstatemachine.nim
View File

@ -1,12 +1,11 @@
import std/sugar import std/sugar
import pkg/questionable import pkg/questionable
import pkg/chronos import pkg/chronos
import pkg/upraises
import ../logutils import ../logutils
import ./then import ./then
import ./trackedfutures import ./trackedfutures
push: {.upraises:[].} {.push raises:[].}
type type
Machine* = ref object of RootObj Machine* = ref object of RootObj
@ -17,7 +16,7 @@ type
trackedFutures: TrackedFutures trackedFutures: TrackedFutures
State* = ref object of RootObj State* = ref object of RootObj
Query*[T] = proc(state: State): T Query*[T] = proc(state: State): T
Event* = proc(state: State): ?State {.gcsafe, upraises:[].} Event* = proc(state: State): ?State {.gcsafe, raises:[].}
logScope: logScope:
topics = "statemachine" topics = "statemachine"
@ -58,14 +57,20 @@ proc onError(machine: Machine, error: ref CatchableError): Event =
return proc (state: State): ?State = return proc (state: State): ?State =
state.onError(error) state.onError(error)
proc run(machine: Machine, state: State) {.async.} = proc run(machine: Machine, state: State) {.async: (raises:[]).} =
try:
if next =? await state.run(machine): if next =? await state.run(machine):
machine.schedule(Event.transition(state, next)) machine.schedule(Event.transition(state, next))
except CancelledError:
discard # do not propagate
except CatchableError as e:
machine.schedule(machine.onError(e))
proc scheduler(machine: Machine) {.async.} = proc scheduler(machine: Machine) {.async: (raises: []).} =
var running: Future[void] var running: Future[void].Raising([])
while machine.started: while machine.started:
let event = await machine.scheduled.get().track(machine) try:
let event = await machine.scheduled.get()
if next =? event(machine.state): if next =? event(machine.state):
if not running.isNil and not running.finished: if not running.isNil and not running.finished:
trace "cancelling current state", state = $machine.state trace "cancelling current state", state = $machine.state
@ -74,13 +79,9 @@ proc scheduler(machine: Machine) {.async.} =
machine.state = next machine.state = next
debug "enter state", state = fromState & " => " & $machine.state debug "enter state", state = fromState & " => " & $machine.state
running = machine.run(machine.state) running = machine.run(machine.state)
running asyncSpawn running.track(machine)
.track(machine) except CancelledError:
.cancelled(proc() = trace "state.run cancelled, swallowing", state = $machine.state) break # do not propagate bc it is asyncSpawned
.catch(proc(err: ref CatchableError) =
trace "error caught in state.run, calling state.onError", state = $machine.state
machine.schedule(machine.onError(err))
)
proc start*(machine: Machine, initialState: State) = proc start*(machine: Machine, initialState: State) =
if machine.started: if machine.started:
@ -90,13 +91,8 @@ proc start*(machine: Machine, initialState: State) =
machine.scheduled = newAsyncQueue[Event]() machine.scheduled = newAsyncQueue[Event]()
machine.started = true machine.started = true
try: asyncSpawn machine.scheduler().track(machine)
discard machine.scheduler().track(machine)
machine.schedule(Event.transition(machine.state, initialState)) machine.schedule(Event.transition(machine.state, initialState))
except CancelledError as e:
discard
except CatchableError as e:
error("Error in scheduler", error = e.msg)
proc stop*(machine: Machine) {.async.} = proc stop*(machine: Machine) {.async.} =
if not machine.started: if not machine.started: