mirror of
https://github.com/codex-storage/nim-codex.git
synced 2025-02-04 00:54:32 +00:00
f25c555d59
* Move to version 2.0.6 * Update nim-confutils submodule to latest version * Update dependencies * Update Nim version to 2.0.12 * Add gcsafe pragma * Add missing import * Update specific conf for Nim 2.x * Fix method signatures * Revert erasure coding attempt to fix bug * More gcsafe pragma * Duplicate code from libp2p because it is not exported anymore * Fix camelcase function names * Use alreadySeen because need is not a bool anymore * newLPStreamReadError does not exist anymore so use another error * Replace ValidIpAddress by IpAddress * Add gcsafe pragma * Restore maintenance parameter deleted by mistake when removing esasure coding fix attempt code * Update method signatures * Copy LPStreamReadError code from libp2p which was removed * Fix camel case * Fix enums in tests * Fix camel case * Extract node components to a variable to make Nim 2 happy * Update the tests using ValidIpAddress to IpAddress * Fix cast for value which is already an option * Set nim version to 2.0.x for CI * Set nim version to 2.0.x for CI * Move to miniupnp version 2.2.4 to avoid symlink error * Set core.symlinks to false for Windows for miniupnp >= 2.2.5 support * Update to Nim 2.0.14 * Update CI nim versions to 2.0.14 * Try with GCC 14 * Replace apt-fast by apt-get * Update ubuntu runner to latest * Use Ubuntu 20.04 for coverage * Disable CI cache for coverage * Add coverage property description * Remove commented test * Check the node value of seen instead of using alreadySeen * Fix the merge. The taskpool work was reverted. * Update nim-ethers submodule * Remove deprecated ValidIpAddress. Fix missing case and imports. * Fix a weird issue where nim-confutils cannot find NatAny * Fix tests and remove useless static keyword
108 lines
3.0 KiB
Nim
108 lines
3.0 KiB
Nim
import pkg/questionable
|
|
import pkg/chronos
|
|
import ../logutils
|
|
import ./trackedfutures
|
|
|
|
{.push raises:[].}
|
|
|
|
type
|
|
Machine* = ref object of RootObj
|
|
state: State
|
|
running: Future[void]
|
|
scheduled: AsyncQueue[Event]
|
|
started: bool
|
|
trackedFutures: TrackedFutures
|
|
State* = ref object of RootObj
|
|
Query*[T] = proc(state: State): T
|
|
Event* = proc(state: State): ?State {.gcsafe, raises:[].}
|
|
|
|
logScope:
|
|
topics = "statemachine"
|
|
|
|
proc new*[T: Machine](_: type T): T =
|
|
T(trackedFutures: TrackedFutures.new())
|
|
|
|
method `$`*(state: State): string {.base, gcsafe.} =
|
|
raiseAssert "not implemented"
|
|
|
|
proc transition(_: type Event, previous, next: State): Event =
|
|
return proc (state: State): ?State =
|
|
if state == previous:
|
|
return some next
|
|
|
|
proc query*[T](machine: Machine, query: Query[T]): ?T =
|
|
if machine.state.isNil:
|
|
none T
|
|
else:
|
|
some query(machine.state)
|
|
|
|
proc schedule*(machine: Machine, event: Event) =
|
|
if not machine.started:
|
|
return
|
|
|
|
try:
|
|
machine.scheduled.putNoWait(event)
|
|
except AsyncQueueFullError:
|
|
raiseAssert "unlimited queue is full?!"
|
|
|
|
method run*(state: State, machine: Machine): Future[?State] {.base, async.} =
|
|
discard
|
|
|
|
method onError*(state: State, error: ref CatchableError): ?State {.base.} =
|
|
raise (ref Defect)(msg: "error in state machine: " & error.msg, parent: error)
|
|
|
|
proc onError(machine: Machine, error: ref CatchableError): Event =
|
|
return proc (state: State): ?State =
|
|
state.onError(error)
|
|
|
|
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: (raises: []).} =
|
|
var running: Future[void].Raising([])
|
|
while machine.started:
|
|
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)
|
|
machine.trackedFutures.track(running)
|
|
asyncSpawn running
|
|
except CancelledError:
|
|
break # do not propagate bc it is asyncSpawned
|
|
|
|
proc start*(machine: Machine, initialState: State) =
|
|
if machine.started:
|
|
return
|
|
|
|
if machine.scheduled.isNil:
|
|
machine.scheduled = newAsyncQueue[Event]()
|
|
|
|
machine.started = true
|
|
let fut = machine.scheduler()
|
|
machine.trackedFutures.track(fut)
|
|
asyncSpawn fut
|
|
machine.schedule(Event.transition(machine.state, initialState))
|
|
|
|
proc stop*(machine: Machine) {.async.} =
|
|
if not machine.started:
|
|
return
|
|
|
|
trace "stopping state machine"
|
|
|
|
machine.started = false
|
|
await machine.trackedFutures.cancelTracked()
|
|
|
|
machine.state = nil
|