Skip to content

Commit

Permalink
fix(asyncstatemachine): fixes not awaiting or asyncSpawning futures (#…
Browse files Browse the repository at this point in the history
…1033)

* fix(trackedfutures): removes usage of `then` from tracked futures

- removes usage of `then`, simplifying the logic, and allowing `then` to be removed completely
- updates annotations to reflect that all procs (sync and async) raise no exceptions

* 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
  • Loading branch information
emizzle authored Dec 13, 2024
1 parent 19af797 commit fb9f4be
Showing 1 changed file with 27 additions and 31 deletions.
58 changes: 27 additions & 31 deletions codex/utils/asyncstatemachine.nim
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
import std/sugar
import pkg/questionable
import pkg/chronos
import pkg/upraises
import ../logutils
import ./then
import ./trackedfutures

push: {.upraises:[].}
{.push raises:[].}

type
Machine* = ref object of RootObj
Expand All @@ -17,7 +16,7 @@ type
trackedFutures: TrackedFutures
State* = ref object of RootObj
Query*[T] = proc(state: State): T
Event* = proc(state: State): ?State {.gcsafe, upraises:[].}
Event* = proc(state: State): ?State {.gcsafe, raises:[].}

logScope:
topics = "statemachine"
Expand Down Expand Up @@ -58,29 +57,31 @@ proc onError(machine: Machine, error: ref CatchableError): Event =
return proc (state: State): ?State =
state.onError(error)

proc run(machine: Machine, state: State) {.async.} =
if next =? await state.run(machine):
machine.schedule(Event.transition(state, next))
proc run(machine: Machine, state: State) {.async: (raises:[]).} =
try:
if next =? await state.run(machine):
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.} =
var running: Future[void]
proc scheduler(machine: Machine) {.async: (raises: []).} =
var running: Future[void].Raising([])
while machine.started:
let event = await machine.scheduled.get().track(machine)
if next =? event(machine.state):
if not running.isNil and not running.finished:
trace "cancelling current state", state = $machine.state
await running.cancelAndWait()
let fromState = if machine.state.isNil: "<none>" else: $machine.state
machine.state = next
debug "enter state", state = fromState & " => " & $machine.state
running = machine.run(machine.state)
running
.track(machine)
.cancelled(proc() = trace "state.run cancelled, swallowing", state = $machine.state)
.catch(proc(err: ref CatchableError) =
trace "error caught in state.run, calling state.onError", state = $machine.state
machine.schedule(machine.onError(err))
)
try:
let event = await machine.scheduled.get()
if next =? event(machine.state):
if not running.isNil and not running.finished:
trace "cancelling current state", state = $machine.state
await running.cancelAndWait()
let fromState = if machine.state.isNil: "<none>" else: $machine.state
machine.state = next
debug "enter state", state = fromState & " => " & $machine.state
running = machine.run(machine.state)
asyncSpawn running.track(machine)
except CancelledError:
break # do not propagate bc it is asyncSpawned

proc start*(machine: Machine, initialState: State) =
if machine.started:
Expand All @@ -90,13 +91,8 @@ proc start*(machine: Machine, initialState: State) =
machine.scheduled = newAsyncQueue[Event]()

machine.started = true
try:
discard machine.scheduler().track(machine)
machine.schedule(Event.transition(machine.state, initialState))
except CancelledError as e:
discard
except CatchableError as e:
error("Error in scheduler", error = e.msg)
asyncSpawn machine.scheduler().track(machine)
machine.schedule(Event.transition(machine.state, initialState))

proc stop*(machine: Machine) {.async.} =
if not machine.started:
Expand Down

0 comments on commit fb9f4be

Please sign in to comment.