Replace actor and relay API

CPS doesn't give gcsafe assurances.

Tupfile

@@ -1,2 +1,3 @@
 include_rules
-: lock.json |> !nim_cfg |> | ./<lock>
+: |> !nim_lk |> {lockfile}
+: {lockfile} |> !nim_cfg |> | ./<lock>

lock.json

@@ -6,12 +6,22 @@
         "bigints"
       ],
       "path": "/nix/store/jvrm392g8adfsgf36prgwkbyd7vh5jsw-source",
-      "ref": "20231006",
       "rev": "86ea14d31eea9275e1408ca34e6bfe9c99989a96",
       "sha256": "15pcpmnk1bnw3k8769rjzcpg00nahyrypwbxs88jnwr4aczp99j4",
       "srcDir": "src",
       "url": "https://github.com/ehmry/nim-bigints/archive/86ea14d31eea9275e1408ca34e6bfe9c99989a96.tar.gz"
     },
+    {
+      "method": "fetchzip",
+      "packages": [
+        "cps"
+      ],
+      "path": "/nix/store/452hfhasrn3gl6vijfmzs69djl099j0j-source",
+      "rev": "b7c179f172e3a256a482a9daee3c0815ea423206",
+      "sha256": "1sn9s7iv83sw1jl5jgi2h7b0xpgsn13f9icp5124jvbp0qkxskx2",
+      "srcDir": "",
+      "url": "https://github.com/nim-works/cps/archive/b7c179f172e3a256a482a9daee3c0815ea423206.tar.gz"
+    },
     {
       "method": "fetchzip",
       "packages": [
@@ -28,12 +38,11 @@
       "packages": [
         "nimcrypto"
       ],
-      "path": "/nix/store/zyr8zwh7vaiycn1s4r8cxwc71f2k5l0h-source",
-      "ref": "traditional-api",
-      "rev": "602c5d20c69c76137201b5d41f788f72afb95aa8",
-      "sha256": "1dmdmgb6b9m5f8dyxk781nnd61dsk3hdxqks7idk9ncnpj9fng65",
+      "path": "/nix/store/jwz8pqbv6rsm8w4fjzdb37r0wzjn5hv0-source",
+      "rev": "d58da671799c69c0b3208b96c154e13c8b1a9e90",
+      "sha256": "12dm0gsy10ppga7zf7hpf4adaqjrd9b740n2w926xyazq1njf6k9",
       "srcDir": "",
-      "url": "https://github.com/cheatfate/nimcrypto/archive/602c5d20c69c76137201b5d41f788f72afb95aa8.tar.gz"
+      "url": "https://github.com/cheatfate/nimcrypto/archive/d58da671799c69c0b3208b96c154e13c8b1a9e90.tar.gz"
     },
     {
       "method": "fetchzip",
@@ -41,7 +50,6 @@
         "npeg"
       ],
       "path": "/nix/store/ffkxmjmigfs7zhhiiqm0iw2c34smyciy-source",
-      "ref": "1.2.1",
       "rev": "26d62fdc40feb84c6533956dc11d5ee9ea9b6c09",
       "sha256": "0xpzifjkfp49w76qmaylan8q181bs45anmp46l4bwr3lkrr7bpwh",
       "srcDir": "src",
@@ -53,11 +61,32 @@
         "preserves"
       ],
       "path": "/nix/store/6nnn5di5vip1vladlb7z56rbw18d1y7j-source",
-      "ref": "20240208",
       "rev": "2825bceecf33a15b9b7942db5331a32cbc39b281",
       "sha256": "145vf46fy3wc52j6vs509fm9bi5lx7c53gskbkpcfbkv82l86dgk",
       "srcDir": "src",
       "url": "https://git.syndicate-lang.org/ehmry/preserves-nim/archive/2825bceecf33a15b9b7942db5331a32cbc39b281.tar.gz"
+    },
+    {
+      "method": "fetchzip",
+      "packages": [
+        "stew"
+      ],
+      "path": "/nix/store/mqg8qzsbcc8xqabq2yzvlhvcyqypk72c-source",
+      "rev": "3c91b8694e15137a81ec7db37c6c58194ec94a6a",
+      "sha256": "17lfhfxp5nxvld78xa83p258y80ks5jb4n53152cdr57xk86y07w",
+      "srcDir": "",
+      "url": "https://github.com/status-im/nim-stew/archive/3c91b8694e15137a81ec7db37c6c58194ec94a6a.tar.gz"
+    },
+    {
+      "method": "fetchzip",
+      "packages": [
+        "sys"
+      ],
+      "path": "/nix/store/syhxsjlsdqfap0hk4qp3s6kayk8cqknd-source",
+      "rev": "4ef3b624db86e331ba334e705c1aa235d55b05e1",
+      "sha256": "1q4qgw4an4mmmcbx48l6xk1jig1vc8p9cq9dbx39kpnb0890j32q",
+      "srcDir": "src",
+      "url": "https://github.com/ehmry/nim-sys/archive/4ef3b624db86e331ba334e705c1aa235d55b05e1.tar.gz"
     }
   ]
 }

src/syndicate.nim

@@ -3,7 +3,8 @@
 
 ## This module implements the `Syndicate DSL <https://syndicate-lang.org/doc/syndicate/>`_.
 
-import std/[asyncdispatch, macros, tables, typetraits]
+import std/[macros, tables, typetraits]
+import pkg/sys/ioqueue
 
 import preserves
 export fromPreserves, toPreserves
@@ -34,21 +35,21 @@ proc `??`*(pat: Pattern; bindings: openArray[(int, Pattern)]): Pattern {.inline.
   patterns.inject(pat, bindings)
 
 type
-  PublishProc = proc (turn: var Turn; v: Value; h: Handle) {.closure, gcsafe.}
-  RetractProc = proc (turn: var Turn; h: Handle) {.closure, gcsafe.}
-  MessageProc = proc (turn: var Turn; v: Value) {.closure, gcsafe.}
+  PublishProc = proc (turn: var Turn; v: Value; h: Handle) {.closure.}
+  RetractProc = proc (turn: var Turn; h: Handle) {.closure.}
+  MessageProc = proc (turn: var Turn; v: Value) {.closure.}
   ClosureEntity = ref object of Entity
     publishImpl: PublishProc
     retractImpl: RetractProc
     messageImpl: MessageProc
 
-method publish(e: ClosureEntity; turn: var Turn; a: AssertionRef; h: Handle) {.gcsafe.} =
+method publish(e: ClosureEntity; turn: var Turn; a: AssertionRef; h: Handle) =
   if not e.publishImpl.isNil: e.publishImpl(turn, a.value, h)
 
-method retract(e: ClosureEntity; turn: var Turn; h: Handle) {.gcsafe.} =
+method retract(e: ClosureEntity; turn: var Turn; h: Handle) =
   if not e.retractImpl.isNil: e.retractImpl(turn, h)
 
-method message(e: ClosureEntity; turn: var Turn; a: AssertionRef) {.gcsafe.} =
+method message(e: ClosureEntity; turn: var Turn; a: AssertionRef) =
   if not e.messageImpl.isNil: e.messageImpl(turn, a.value)
 
 proc argumentCount(handler: NimNode): int =
@@ -187,16 +188,7 @@ macro during*(turn: untyped; ds: Cap; pattern: Pattern; publishBody: untyped) =
     `callbackProc`
     discard observe(`turn`, `ds`, `pattern`, during(`callbackSym`))
 
-type BootProc = proc (turn: var Turn; ds: Cap) {.gcsafe.}
-type DeprecatedBootProc = proc (ds: Cap; turn: var Turn) {.gcsafe.}
-
-proc runActor*(name: string; bootProc: BootProc) =
-  ## Run an `Actor` to completion.
-  let actor = bootDataspace(name, bootProc)
-  while actor.running:
-    waitFor sleepAsync(500)
-
-proc runActor*(name: string; bootProc: DeprecatedBootProc) {.deprecated.} =
-  ## Run an `Actor` to completion.
-  runActor(name) do (turn: var Turn, ds: Cap):
-    bootProc(ds, turn)
+proc runActor*(name: string; bootProc: TurnAction) =
+  ## Boot an actor `Actor` and churn ioqueue once.
+  let actor = bootActor(name, bootProc)
+  ioqueue.run()

src/syndicate/actors.nim

@@ -1,7 +1,7 @@
 # SPDX-FileCopyrightText: ☭ Emery Hemingway
 # SPDX-License-Identifier: Unlicense
 
-import std/[asyncfutures, hashes, monotimes, options, sets, tables, times]
+import std/[hashes, monotimes, options, sets, tables, times]
 import preserves
 import ../syndicate/protocols/[protocol, sturdy]
 
@@ -66,7 +66,7 @@ type
       turnIdAllocator: ref TurnId
       traceStream: FileStream
 
-  TurnAction* = proc (t: var Turn) {.gcsafe.}
+  TurnAction* = proc (t: var Turn) {.closure.}
 
   Queues = TableRef[Facet, seq[TurnAction]]
 
@@ -108,10 +108,10 @@ when tracing:
       result.add f.id.toPreserves
       f = f.parent
 
-method publish*(e: Entity; turn: var Turn; v: AssertionRef; h: Handle) {.base, gcsafe.} = discard
-method retract*(e: Entity; turn: var Turn; h: Handle) {.base, gcsafe.} = discard
-method message*(e: Entity; turn: var Turn; v: AssertionRef) {.base, gcsafe.} = discard
-method sync*(e: Entity; turn: var Turn; peer: Cap) {.base, gcsafe.} = discard
+method publish*(e: Entity; turn: var Turn; v: AssertionRef; h: Handle) {.base.} = discard
+method retract*(e: Entity; turn: var Turn; h: Handle) {.base.} = discard
+method message*(e: Entity; turn: var Turn; v: AssertionRef) {.base.} = discard
+method sync*(e: Entity; turn: var Turn; peer: Cap) {.base.} = discard
 
 using
   actor: Actor
@@ -171,7 +171,7 @@ proc match(bindings: var Bindings; p: Pattern; v: Value): bool =
   of PatternKind.Patom:
     result = case p.patom
       of PAtom.Boolean: v.isBoolean
-      of PAtom.Double: v.isDouble
+      of PAtom.Double: v.isFloat
       of PAtom.Signedinteger: v.isInteger
       of PAtom.String: v.isString
       of PAtom.Bytestring: v.isByteString
@@ -278,8 +278,7 @@ proc runRewrites*(a: Attenuation; v: Value): Value =
 proc publish(turn: var Turn; r: Cap; v: Value; h: Handle) =
   var a = runRewrites(r.attenuation, v)
   if not a.isFalse:
-    let e = OutboundAssertion(
-      handle: h, peer: r, established: false)
+    let e = OutboundAssertion(handle: h, peer: r)
     turn.facet.outbound[h] = e
     enqueue(turn, r.relay) do (turn: var Turn):
       e.established = true
@@ -341,9 +340,9 @@ proc replace*[T](turn: var Turn; cap: Cap; h: var Handle; v: T): Handle {.discar
     retract(turn, old)
   h
 
-proc stop*(turn: var Turn) {.gcsafe.}
+proc stop*(turn: var Turn)
 
-proc run*(facet; action: TurnAction; zombieTurn = false) {.gcsafe.}
+proc run*(facet; action: TurnAction; zombieTurn = false)
 
 proc newFacet(actor; parent: Facet; initialAssertions: OutboundTable): Facet =
   result = Facet(
@@ -363,7 +362,7 @@ proc isInert(facet): bool =
     (facet.outbound.len == 0 or facet.parent.isNil) and
     facet.inertCheckPreventers == 0
 
-proc preventInertCheck*(facet): (proc() {.gcsafe.}) {.discardable.} =
+proc preventInertCheck*(facet): (proc()) {.discardable.} =
   var armed = true
   inc facet.inertCheckPreventers
   proc disarm() =
@@ -378,9 +377,9 @@ proc inFacet(turn: var Turn; facet; act: TurnAction) =
   var t = Turn(facet: facet, queues: turn.queues)
   act(t)
 
-proc terminate(actor; turn; reason: ref Exception) {.gcsafe.}
+proc terminate(actor; turn; reason: ref Exception)
 
-proc terminate(facet; turn: var Turn; orderly: bool) {.gcsafe.} =
+proc terminate(facet; turn: var Turn; orderly: bool) =
   if facet.isAlive:
     facet.isAlive = false
     let parent = facet.parent
@@ -458,7 +457,7 @@ proc bootActor*(name: string; bootProc: TurnAction): Actor =
     else: result.traceStream = openFileStream(path, fmWrite)
   run(result, bootProc, initialAssertions)
 
-proc spawn*(name: string; turn: var Turn; bootProc: TurnAction; initialAssertions = initHashSet[Handle]()): Actor {.discardable.} =
+proc spawnActor*(name: string; turn: var Turn; bootProc: TurnAction; initialAssertions = initHashSet[Handle]()): Actor {.discardable.} =
   let actor = newActor(name, turn.facet.actor.handleAllocator)
   enqueue(turn, turn.facet) do (turn: var Turn):
     var newOutBound: Table[Handle, OutboundAssertion]
@@ -473,6 +472,9 @@ proc spawn*(name: string; turn: var Turn; bootProc: TurnAction; initialAssertion
     run(actor, bootProc, newOutBound)
   actor
 
+proc spawn*(name: string; turn: var Turn; bootProc: TurnAction; initialAssertions = initHashSet[Handle]()): Actor {.discardable.} =
+  spawnActor(name, turn, bootProc, initialAssertions)
+
 proc newInertCap*(): Cap =
   let a = bootActor("inert") do (turn: var Turn): turn.stop()
   Cap(relay: a.root)
@@ -493,13 +495,14 @@ proc terminate(actor; turn; reason: ref Exception) =
     proc finish(turn: var Turn) =
       actor.root.terminate(turn, reason.isNil)
       actor.exited = true
-    callSoon do ():
+    block:
       run(actor.root, finish, true)
 
 proc terminate*(facet; e: ref Exception) =
   run(facet.actor.root) do (turn: var Turn):
     facet.actor.terminate(turn, e)
 
+#[
 proc asyncCheck*(facet: Facet; fut: FutureBase) =
   ## Sets a callback on `fut` which propagates exceptions to `facet`.
   addCallback(fut) do ():
@@ -508,13 +511,14 @@ proc asyncCheck*(facet: Facet; fut: FutureBase) =
 proc asyncCheck*(turn; fut: FutureBase) =
   ## Sets a callback on `fut` which propagates exceptions to the facet of `turn`.
   asyncCheck(turn.facet, fut)
+]#
 
 template tryFacet(facet; body: untyped) =
   try: body
   except CatchableError as err: terminate(facet, err)
 
 proc run*(facet; action: TurnAction; zombieTurn = false) =
-  if zombieTurn or (facet.actor.exitReason.isNil and facet.isAlive):
+  if true and zombieTurn or (facet.actor.exitReason.isNil and facet.isAlive):
     tryFacet(facet):
       var queues = newTable[Facet, seq[TurnAction]]()
       block:
@@ -531,6 +535,7 @@ proc run*(cap: Cap; action: TurnAction) =
   ## Convenience proc to run a `TurnAction` in the scope of a `Cap`.
   run(cap.relay, action)
 
+#[
 proc addCallback*(fut: FutureBase; facet: Facet; act: TurnAction) =
   ## Add a callback to a `Future` that will be called at a later `Turn`
   ## within the context of `facet`.
@@ -555,7 +560,7 @@ proc addCallback*(fut: FutureBase; turn: var Turn; act: TurnAction) =
   else:
     addCallback(fut, turn.facet, act)
 
-proc addCallback*[T](fut: Future[T]; turn: var Turn; act: proc (t: var Turn, x: T) {.gcsafe.}) =
+proc addCallback*[T](fut: Future[T]; turn: var Turn; act: proc (t: var Turn, x: T) {.closure.}) =
   addCallback(fut, turn) do (turn: var Turn):
     if fut.failed: terminate(turn.facet, fut.error)
     else:
@@ -563,6 +568,7 @@ proc addCallback*[T](fut: Future[T]; turn: var Turn; act: proc (t: var Turn, x:
         turn.desc.cause = TurnCause(orKind: TurnCauseKind.external)
         turn.desc.cause.external.description = "Future".toPreserves
       act(turn, read fut)
+]#
 
 proc stop*(turn: var Turn, facet: Facet) =
   if facet.parent.isNil:
@@ -578,10 +584,17 @@ proc onStop*(facet: Facet; act: TurnAction) =
   ## Add a `proc (turn: var Turn)` action to `facet` to be called as it stops.
   add(facet.shutdownActions, act)
 
+proc stopActor*(facet: Facet) =
+  run(facet) do (turn: var Turn):
+    let actor = facet.actor
+    enqueue(turn, actor.root) do (turn: var Turn):
+      terminate(actor, turn, nil)
+
 proc stopActor*(turn: var Turn) =
-  let actor = turn.facet.actor
-  enqueue(turn, actor.root) do (turn: var Turn):
-    terminate(actor, turn, nil)
+  stopActor(turn.facet)
+
+proc stop*(actor: Actor) =
+  stopActor(actor.root)
 
 proc freshen*(turn: var Turn, act: TurnAction) =
   assert(turn.queues.len == 0, "Attempt to freshen a non-stale Turn")

src/syndicate/actors/timers.nim

@@ -1,35 +1,115 @@
 # SPDX-FileCopyrightText: ☭ Emery Hemingway
 # SPDX-License-Identifier: Unlicense
 
-import std/[asyncdispatch, monotimes, times]
+import std/[sets, times]
+import pkg/sys/[handles, ioqueue]
 import preserves
-import syndicate
-
-import ../protocols/timer
-from syndicate/protocols/dataspace import Observe
+import ../../syndicate, ../bags, ../protocols/[timer, dataspace]
 
 export timer
 
-type Observe = dataspace.Observe
+type
+  Observe = dataspace.Observe
 
-proc now: float64 = getTime().toUnixFloat()
+when defined(linux):
+  import std/[oserrors, posix]
+  type Time = posix.Time
 
-proc spawnTimers*(turn: var Turn; ds: Cap): Actor {.discardable.} =
-  ## Spawn a timer actor.
-  spawn("timer", turn) do (turn: var Turn):
+  {.pragma: timerfd, importc, header: "<sys/timerfd.h>".}
 
-    during(turn, ds, inject(grab Observe(pattern: dropType LaterThan), {0: grabLit()})) do (seconds: float):
-      let period = seconds - now()
-      if period < 0.001:
-        discard publish(turn, ds, LaterThan(seconds: seconds))
-      else:
-        addCallback(sleepAsync(period * 1_000), turn) do (turn: var Turn):
-          discard publish(turn, ds, LaterThan(seconds: seconds))
+  proc timerfd_create(clock_id: ClockId, flags: cint): cint {.timerfd.}
+  proc timerfd_settime(ufd: cint, flags: cint,
+                        utmr: var Itimerspec, otmr: var Itimerspec): cint {.timerfd.}
+  proc timerfd_gettime(ufd: cint, curr: var Itimerspec): cint {.timerfd.}
+
+  var
+    TFD_NONBLOCK {.timerfd.}: cint
+    TFD_CLOEXEC {.timerfd.}: cint
+    TFD_TIMER_ABSTIME {.timerfd.}: cint
+
+  proc `<`(a, b: Timespec): bool =
+    a.tv_sec.clong < b.tv_sec.clong or
+      (a.tv_sec.clong == b.tv_sec.clong and a.tv_nsec < b.tv_nsec)
+
+  proc `+`(a, b: Timespec): Timespec =
+    result.tv_sec = Time a.tv_sec.clong + b.tv_sec.clong
+    result.tv_nsec = a.tv_nsec + b.tv_nsec
+
+  func toFloat(ts: Timespec): float =
+    ts.tv_sec.float + ts.tv_nsec.float / 1_000_000_000
+
+  func toTimespec(f: float): Timespec =
+    result.tv_sec = Time(f)
+    result.tv_nsec = clong(uint64(f * 1_000_000_000) mod 1_000_000_000)
+
+  proc clock_realtime: Timespec =
+    if clock_gettime(CLOCK_REALTIME, result) < 0:
+      raiseOSError(osLastError(), "clock_gettime")
+
+  type
+    TimerDriver = ref object
+      facet: Facet
+        ## Owning facet of driver.
+      target: Cap
+        ## Destination for LaterThan assertions.
+      deadlines: Bag[float]
+        ## Deadlines that other actors are observing.
+      timers: HashSet[cint]
+        # TODO: use a single timer descriptor
+
+  proc spawnTimerDriver(facet: Facet; cap: Cap): TimerDriver =
+    let driver = TimerDriver(facet: facet, target: cap)
+    facet.onStop do (turn: var Turn):
+      for fd in driver.timers:
+        unregister(FD fd)
+        discard close(fd)
+    driver
+
+  proc earliestFloat(driver: TimerDriver): float =
+    assert driver.deadlines.len > 0
+    result = high float
+    for deadline in driver.deadlines:
+      if deadline < result:
+        result = deadline
+
+  proc await(driver: TimerDriver; deadline: float) {.asyncio.} =
+    ## Run timer driver concurrently with actor.
+    let fd = timerfd_create(CLOCK_REALTIME, TFD_NONBLOCK or TFD_CLOEXEC)
+    if fd < 0:
+      raiseOSError(osLastError(), "failed to acquire timer descriptor")
+    var
+      old: Itimerspec
+      its = Itimerspec(it_value: deadline.toTimespec)
+    if timerfd_settime(fd, TFD_TIMER_ABSTIME, its, old) < 0:
+      raiseOSError(osLastError(), "failed to set timeout")
+    driver.timers.incl(fd)
+    while clock_realtime() < its.it_value:
+      # Check if the timer is expired which
+      # could happen before waiting.
+      wait(FD fd, Read)
+    if deadline in driver.deadlines:
+      # Check if the deadline is still observed.
+      proc turnWork(turn: var Turn) =
+        discard publish(turn, driver.target, LaterThan(seconds: deadline))
+      run(driver.facet, turnWork)
+    discard close(fd)
+    driver.timers.excl(fd)
+
+  proc spawnTimerActor*(ds: Cap; turn: var Turn): Actor {.discardable.} =
+    ## Spawn a timer actor that responds to
+    ## dataspace observations of timeouts on `ds`.
+    spawnActor("timers", turn) do (turn: var Turn):
+      let driver = spawnTimerDriver(turn.facet, ds)
+      let pat = inject(grab Observe(pattern: dropType LaterThan), {0: grabLit()})
+      during(turn, ds, pat) do (deadline: float):
+        if change(driver.deadlines, deadline, +1) == cdAbsentToPresent:
+          discard trampoline(whelp await(driver, deadline))
+      do:
+        discard change(driver.deadlines, deadline, -1, clamp = true)
+          # TODO: retract assertions that are unobserved.
 
 proc after*(turn: var Turn; ds: Cap; dur: Duration; act: TurnAction) =
   ## Execute `act` after some duration of time.
-  let later = now() + dur.inMilliseconds.float64 * 1_000.0
+  var later = clock_realtime().toFloat() + dur.inMilliseconds.float / 1_000.0
   onPublish(turn, ds, grab LaterThan(seconds: later)):
     act(turn)
-
-# TODO: periodic timer

src/syndicate/bags.nim

@@ -46,3 +46,5 @@ proc `$`*(bag: Bag): string =
     if result.len > 1: result.add ' '
     result.add $x
   result.add '}'
+
+export tables.contains, tables.del, tables.len

src/syndicate/dataflow.nim

@@ -51,7 +51,7 @@ iterator observersOf[Sid, Oid](g: Graph[Sid, Oid]; oid: Oid): Sid =
   if g.edgesForward.hasKey(oid):
     for sid in g.edgesForward[oid]: yield sid
 
-proc repairDamage*[Sid, Oid](g: var Graph[Sid, Oid]; repairNode: proc (sid: Sid) {.gcsafe.}) =
+proc repairDamage*[Sid, Oid](g: var Graph[Sid, Oid]; repairNode: proc (sid: Sid) {.closure.}) =
   var repairedThisRound: Set[Oid]
   while true:
     var workSet = move g.damagedNodes

src/syndicate/dataspaces.nim

@@ -16,14 +16,14 @@ type
     index: Index
     handleMap: Table[Handle, Assertion]
 
-method publish(ds: Dataspace; turn: var Turn; a: AssertionRef; h: Handle) {.gcsafe.} =
+method publish(ds: Dataspace; turn: var Turn; a: AssertionRef; h: Handle) =
   if add(ds.index, turn, a.value):
     var obs = a.value.preservesTo(Observe)
     if obs.isSome and obs.get.observer of Cap:
       ds.index.add(turn, obs.get.pattern, Cap(obs.get.observer))
   ds.handleMap[h] = a.value
 
-method retract(ds: Dataspace; turn: var Turn; h: Handle) {.gcsafe.} =
+method retract(ds: Dataspace; turn: var Turn; h: Handle) =
   let v = ds.handleMap[h]
   if remove(ds.index, turn, v):
     ds.handleMap.del h
@@ -31,14 +31,14 @@ method retract(ds: Dataspace; turn: var Turn; h: Handle) {.gcsafe.} =
     if obs.isSome and obs.get.observer of Cap:
       ds.index.remove(turn, obs.get.pattern, Cap(obs.get.observer))
 
-method message(ds: Dataspace; turn: var Turn; a: AssertionRef) {.gcsafe.} =
+method message(ds: Dataspace; turn: var Turn; a: AssertionRef) =
   ds.index.deliverMessage(turn, a.value)
 
 proc newDataspace*(turn: var Turn): Cap =
   newCap(turn, Dataspace(index: initIndex()))
 
-type BootProc = proc (turn: var Turn; ds: Cap) {.gcsafe.}
-type DeprecatedBootProc = proc (ds: Cap; turn: var Turn) {.gcsafe.}
+type BootProc = proc (turn: var Turn; ds: Cap) {.closure.}
+type DeprecatedBootProc = proc (ds: Cap; turn: var Turn) {.closure.}
 
 proc bootDataspace*(name: string; bootProc: BootProc): Actor =
   bootActor(name) do (turn: var Turn):

src/syndicate/durings.nim

@@ -6,7 +6,7 @@ import preserves
 import ./actors, ./patterns, ./protocols/dataspace
 
 type
-  DuringProc* = proc (turn: var Turn; a: Value; h: Handle): TurnAction {.gcsafe.}
+  DuringProc* = proc (turn: var Turn; a: Value; h: Handle): TurnAction
   DuringActionKind = enum null, dead, act
   DuringAction = object
     case kind: DuringActionKind

src/syndicate/patterns.nim

@@ -313,7 +313,7 @@ proc grabDictionary*(bindings: sink openArray[(string, Pattern)]): Pattern =
   for (key, val) in bindings.items:
     result.dcompound.dict.entries[key.toSymbol] = val
 
-proc depattern(comp: DCompound; values: var seq[Value]; index: var int): Value {.gcsafe.}
+proc depattern(comp: DCompound; values: var seq[Value]; index: var int): Value
 
 proc depattern(pat: Pattern; values: var seq[Value]; index: var int): Value =
   case pat.orKind
@@ -328,7 +328,7 @@ proc depattern(pat: Pattern; values: var seq[Value]; index: var int): Value =
   of PatternKind.DCompound:
     result = depattern(pat.dcompound, values, index)
 
-proc depattern(comp: DCompound; values: var seq[Value]; index: var int): Value {.gcsafe.} =
+proc depattern(comp: DCompound; values: var seq[Value]; index: var int): Value =
   case comp.orKind
   of DCompoundKind.rec:
     result = initRecord(comp.rec.label, comp.rec.fields.len)

src/syndicate/relays.nim

@@ -1,8 +1,9 @@
 # SPDX-FileCopyrightText: ☭ Emery Hemingway
 # SPDX-License-Identifier: Unlicense
 
-import std/[asyncdispatch, options, tables]
+import std/[options, tables]
 from std/os import getEnv, `/`
+import pkg/sys/ioqueue
 import preserves
 import ../syndicate, /capabilities, ./durings, ./membranes, ./protocols/[gatekeeper, protocol, sturdy, transportAddress]
 
@@ -16,19 +17,18 @@ else:
 
 export `$`
 
-type
-  Oid = sturdy.Oid
-
 export Stdio, Tcp, WebSocket, Unix
 
 type
   Assertion = Value
-  WireRef = sturdy.WireRef
-  Turn = syndicate.Turn
+  Event = protocol.Event
   Handle = actors.Handle
+  Oid = sturdy.Oid
+  Turn = syndicate.Turn
+  WireRef = sturdy.WireRef
 
-  PacketWriter = proc (turn: var Turn; buf: seq[byte]) {.closure, gcsafe.}
-  RelaySetup = proc (turn: var Turn; relay: Relay) {.closure, gcsafe.}
+  PacketWriter = proc (turn: var Turn; buf: seq[byte]) {.closure.}
+  RelaySetup = proc (turn: var Turn; relay: Relay) {.closure.}
 
   Relay* = ref object
     facet: Facet
@@ -38,7 +38,6 @@ type
     exported: Membrane
     imported: Membrane
     nextLocalOid: Oid
-    pendingTurn: protocol.Turn
     wireBuf: BufferedDecoder
     packetWriter: PacketWriter
     peer: Cap
@@ -90,7 +89,7 @@ proc rewriteCapOut(relay: Relay; cap: Cap; exported: var seq[WireSymbol]): WireR
         mine: WireRefMine(oid: ws.oid))
 
 proc rewriteOut(relay: Relay; v: Assertion):
-    tuple[rewritten: Value, exported: seq[WireSymbol]] {.gcsafe.} =
+    tuple[rewritten: Value, exported: seq[WireSymbol]] =
   var exported: seq[WireSymbol]
   result.rewritten = mapEmbeds(v) do (pr: Value) -> Value:
     let o = pr.unembed(Cap); if o.isSome:
@@ -108,41 +107,39 @@ proc deregister(relay: Relay; h: Handle) =
     for e in outbound: releaseCapOut(relay, e)
 
 proc send(relay: Relay; turn: var Turn; rOid: protocol.Oid; m: Event) =
-  if relay.pendingTurn.len == 0:
-    # If the pending queue is empty then schedule a packet
-    # to be sent after pending I/O is processed.
-    callSoon do ():
-      relay.facet.run do (turn: var Turn):
-        var pkt = Packet(
-          orKind: PacketKind.Turn,
-          turn: move relay.pendingTurn)
-        trace "C: ", pkt
-        relay.packetWriter(turn, encode pkt)
-  relay.pendingTurn.add TurnEvent(oid: rOid, event: m)
+  # TODO: don't send right away.
+  var pendingTurn: protocol.Turn
+  pendingTurn.add TurnEvent(oid: rOid, event: m)
+  relay.facet.run do (turn: var Turn):
+    var pkt = Packet(
+      orKind: PacketKind.Turn,
+      turn: pendingTurn)
+    trace "C: ", pkt
+    relay.packetWriter(turn, encode pkt)
 
 proc send(re: RelayEntity; turn: var Turn; ev: Event) =
   send(re.relay, turn, protocol.Oid re.oid, ev)
 
-method publish(re: RelayEntity; t: var Turn; a: AssertionRef; h: Handle) {.gcsafe.} =
+method publish(re: RelayEntity; t: var Turn; a: AssertionRef; h: Handle) =
   re.send(t, Event(
       orKind: EventKind.Assert,
       `assert`: protocol.Assert(
         assertion: re.relay.register(a.value, h).rewritten,
         handle: h)))
 
-method retract(re: RelayEntity; t: var Turn; h: Handle) {.gcsafe.} =
+method retract(re: RelayEntity; t: var Turn; h: Handle) =
   re.relay.deregister h
   re.send(t, Event(
     orKind: EventKind.Retract,
     retract: Retract(handle: h)))
 
-method message(re: RelayEntity; turn: var Turn; msg: AssertionRef) {.gcsafe.} =
+method message(re: RelayEntity; turn: var Turn; msg: AssertionRef) =
   var (value, exported) = rewriteOut(re.relay, msg.value)
   assert(len(exported) == 0, "cannot send a reference in a message")
   if len(exported) == 0:
     re.send(turn, Event(orKind: EventKind.Message, message: Message(body: value)))
 
-method sync(re: RelayEntity; turn: var Turn; peer: Cap) {.gcsafe.} =
+method sync(re: RelayEntity; turn: var Turn; peer: Cap) =
   var
     peerEntity = newSyncPeerEntity(re.relay, peer)
     exported: seq[WireSymbol]
@@ -185,7 +182,7 @@ proc rewriteCapIn(relay; facet; n: WireRef, imported: var seq[WireSymbol]): Cap
     else: raiseAssert "attenuation not implemented"
 
 proc rewriteIn(relay; facet; v: Value):
-    tuple[rewritten: Assertion; imported: seq[WireSymbol]] {.gcsafe.} =
+    tuple[rewritten: Assertion; imported: seq[WireSymbol]] =
   var imported: seq[WireSymbol]
   result.rewritten = mapEmbeds(v) do (pr: Value) -> Value:
     let wr = pr.preservesTo WireRef; if wr.isSome:
@@ -196,7 +193,7 @@ proc rewriteIn(relay; facet; v: Value):
 
 proc close(r: Relay) = discard
 
-proc dispatch(relay: Relay; turn: var Turn; cap: Cap; event: Event) {.gcsafe.} =
+proc dispatch(relay: Relay; turn: var Turn; cap: Cap; event: Event) =
   case event.orKind
   of EventKind.Assert:
     let (a, imported) = rewriteIn(relay, turn.facet, event.assert.assertion)
@@ -224,7 +221,7 @@ proc dispatch(relay: Relay; turn: var Turn; cap: Cap; event: Event) {.gcsafe.} =
       for e in imported: relay.imported.del e
       ]#
 
-proc dispatch(relay: Relay; v: Value) {.gcsafe.} =
+proc dispatch(relay: Relay; v: Value) =
   trace "S: ", v
   run(relay.facet) do (t: var Turn):
     var pkt: Packet
@@ -250,8 +247,8 @@ proc dispatch(relay: Relay; v: Value) {.gcsafe.} =
       when defined(posix):
         stderr.writeLine("discarding undecoded packet ", v)
 
-proc recv(relay: Relay; buf: seq[byte]) =
-  feed(relay.wireBuf, buf)
+proc recv(relay: Relay; buf: openarray[byte]; slice: Slice[int]) =
+  feed(relay.wireBuf, buf, slice)
   var pr = decode(relay.wireBuf)
   if pr.isSome: dispatch(relay, pr.get)
 
@@ -265,7 +262,7 @@ type
     nextLocalOid*: Option[Oid]
 
 proc spawnRelay(name: string; turn: var Turn; opts: RelayActorOptions; setup: RelaySetup) =
-  spawn(name, turn) do (turn: var Turn):
+  spawnActor(name, turn) do (turn: var Turn):
     let relay = Relay(
         facet: turn.facet,
         packetWriter: opts.packetWriter,
@@ -300,16 +297,29 @@ proc accepted(cap: Cap): Resolved =
 
 when defined(posix):
 
-  import std/asyncfile
-  export Unix
+  import std/[oserrors, posix]
+  import pkg/sys/[files, handles, sockets]
+  export transportAddress.Unix
 
   type StdioControlEntity = ref object of Entity
+    buf: ref seq[byte]
+    relay: Relay
     stdin: AsyncFile
 
   method message(entity: StdioControlEntity; turn: var Turn; ass: AssertionRef) =
     if ass.value.preservesTo(ForceDisconnect).isSome:
       close(entity.stdin)
-      close(stdout)
+
+  proc loop(entity: StdioControlEntity) {.asyncio.} =
+    new entity.buf
+    entity.buf[].setLen(0x1000)
+    while true:
+      let n = read(entity.stdin, entity.buf)
+      if n == 0:
+        stderr.writeLine "empty read on stdin, stopping actor"
+        stopActor(entity.relay.facet)
+      else:
+        entity.relay.recv(entity.buf[], 0..<n)
 
   proc connectTransport(turn: var Turn; ds: Cap; ta: transportAddress.Stdio) =
     ## Connect to an external dataspace over stdio.
@@ -327,101 +337,111 @@ when defined(posix):
     spawnRelay("stdio", turn, opts) do (turn: var Turn; relay: Relay):
       let
         facet = turn.facet
-        asyncStdin = openAsync("/dev/stdin") # this is universal now?
+        fd = stdin.getOsFileHandle()
+        flags =  fcntl(fd.cint, F_GETFL, 0)
+      if flags < 0: raiseOSError(osLastError())
+      if fcntl(fd.cint, F_SETFL, flags or O_NONBLOCK) < 0:
+        raiseOSError(osLastError())
+      let entity = StdioControlEntity(
+          relay: relay, stdin: newAsyncFile(FD fd))
       publish(turn, ds, TransportConnection(
           `addr`: ta.toPreserves,
-          control: StdioControlEntity(stdin: asyncStdin).newCap(turn),
+          control: newCap(entity, turn),
           resolved: relay.peer.accepted,
         ))
-      const stdinReadSize = 0x2000
-      proc readCb(pktFut: Future[string]) {.gcsafe.} =
-        if not pktFut.failed:
-          var buf = pktFut.read
-          if buf.len == 0:
-            run(facet) do (turn: var Turn): stopActor(turn)
-          else:
-            relay.recv(cast[seq[byte]](buf))
-            asyncStdin.read(stdinReadSize).addCallback(readCb)
-      asyncStdin.read(stdinReadSize).addCallback(readCb)
+      discard trampoline:
+        whelp loop(entity)
 
   proc connectStdio*(turn: var Turn; ds: Cap) =
     ## Connect to an external dataspace over stdin and stdout.
     connectTransport(turn, ds, transportAddress.Stdio())
 
-  import std/asyncnet
-  from std/nativesockets import AF_INET, AF_UNIX, IPPROTO_TCP, SOCK_STREAM, Protocol
+  type
+    TcpEntity = ref object of Entity
+      relay: Relay
+      sock: AsyncConn[sockets.Protocol.TCP]
+      buf: ref seq[byte]
+      alive: bool
 
-  type SocketControlEntity = ref object of Entity
-    socket: AsyncSocket
+    UnixEntity = ref object of Entity
+      relay: Relay
+      sock: AsyncConn[sockets.Protocol.Unix]
+      buf: ref seq[byte]
+      alive: bool
 
-  method message(entity: SocketControlEntity; turn: var Turn; ass: AssertionRef) =
+    SocketEntity = TcpEntity | UnixEntity
+
+  method message(entity: SocketEntity; turn: var Turn; ass: AssertionRef) =
     if ass.value.preservesTo(ForceDisconnect).isSome:
-      close(entity.socket)
+      reset entity.alive
+      close(entity.sock)
 
-  type ShutdownEntity* = ref object of Entity
+  template socketLoop() {.dirty.}=
+    new entity.buf
+    entity.buf[].setLen(0x1000)
+    entity.alive = not entity.alive
+    while entity.alive:
+      let n = read(entity.sock, entity.buf)
+      if n < 0: raiseOSError(osLastError())
+      elif n == 0:
+        stderr.writeLine "empty read on socket, stopping actor"
+        stopActor(entity.relay.facet)
+      else:
+        entity.relay.recv(entity.buf[], 0..<n)
+    stderr.writeLine "breaking socketLoop"
+
+  proc loop(entity: TcpEntity) {.asyncio.} =
+    socketLoop()
+  proc loop(entity: UnixEntity) {.asyncio.} =
+    socketLoop()
+
+  type ShutdownEntity = ref object of Entity
   method retract(e: ShutdownEntity; turn: var Turn; h: Handle) =
     stopActor(turn)
 
-  proc connect(turn: var Turn; ds: Cap; transAddr: Value; socket: AsyncSocket) =
-    proc socketWriter(turn: var Turn; buf: seq[byte]) =
-      asyncCheck(turn, socket.send(cast[string](buf)))
+  template bootSocketEntity() {.dirty.} =
+    proc publish(turn: var Turn) =
+      publish(turn, ds, TransportConnection(
+          `addr`: ta.toPreserves,
+          control: newCap(entity, turn),
+          resolved: entity.relay.peer.accepted,
+        ))
+    run(entity.relay.facet, publish)
+    loop(entity)
+
+  proc boot(entity: TcpEntity; ta: transportAddress.Tcp; ds: Cap) {.asyncio.} =
+    entity.sock = connectTcpAsync(ta.host, Port ta.port)
+    bootSocketEntity()
+
+  proc boot(entity: UnixEntity; ta: transportAddress.Unix; ds: Cap) {.asyncio.} =
+    entity.sock = connectUnixAsync(ta.path)
+    bootSocketEntity()
+
+  template spawnSocketRelay() {.dirty.} =
+    proc writeConn(turn: var Turn; buf: seq[byte]) =
+      discard trampoline:
+        whelp write(entity.sock, buf)
     var ops = RelayActorOptions(
-        packetWriter: socketWriter,
+        packetWriter: writeConn,
         initialOid: 0.Oid.some,
       )
     spawnRelay("socket", turn, ops) do (turn: var Turn; relay: Relay):
-      let facet = turn.facet
-      facet.actor.atExit do (turn: var Turn): close(socket)
-      publish(turn, ds, TransportConnection(
-          `addr`: transAddr,
-          control: SocketControlEntity(socket: socket).newCap(turn),
-          resolved: relay.peer.accepted,
-        ))
-      const recvSize = 0x4000
-      proc recvCb(pktFut: Future[string]) {.gcsafe.} =
-        if pktFut.failed or pktFut.read.len == 0:
-          run(facet) do (turn: var Turn): stopActor(turn)
-        else:
-          relay.recv(cast[seq[byte]](pktFut.read))
-          if not socket.isClosed:
-            socket.recv(recvSize).addCallback(recvCb)
-      socket.recv(recvSize).addCallback(recvCb)
-
-  proc connect(turn: var Turn; ds: Cap; ta: Value; socket: AsyncSocket; fut: Future[void]) =
-    let facet = turn.facet
-    fut.addCallback do ():
-      run(facet) do (turn: var Turn):
-        if fut.failed:
-          var ass = TransportConnection(
-              `addr`: ta,
-              resolved: Resolved(orKind: ResolvedKind.Rejected),
-            )
-          ass.resolved.rejected.detail = embed fut.error
-          publish(turn, ds, ass)
-        else:
-          connect(turn, ds, ta, socket)
+      entity.relay = relay
+      atExit(turn.facet.actor) do (turn: var Turn):
+        entity.alive = false
+        close(entity.sock)
+      discard trampoline:
+        whelp boot(entity, ta, ds)
 
   proc connectTransport(turn: var Turn; ds: Cap; ta: transportAddress.Tcp) =
-    let
-      facet = turn.facet
-      socket = newAsyncSocket(
-          domain = AF_INET,
-          sockType = SOCK_STREAM,
-          protocol = IPPROTO_TCP,
-          buffered = false,
-        )
-    connect(turn, ds, ta.toPreserves, socket, connect(socket, ta.host, Port ta.port))
+    let entity =  TcpEntity()
+    spawnSocketRelay()
 
   proc connectTransport(turn: var Turn; ds: Cap; ta: transportAddress.Unix) =
-    ## Relay a dataspace over a UNIX socket.
-    let socket = newAsyncSocket(
-      domain = AF_UNIX,
-      sockType = SOCK_STREAM,
-      protocol = cast[Protocol](0),
-      buffered = false)
-    connect(turn, ds, ta.toPreserves, socket, connectUnix(socket, ta.path))
+    let entity =  UnixEntity()
+    spawnSocketRelay()
 
-proc walk(turn: var Turn; ds, origin: Cap; route: Route; transOff, stepOff: int) {.gcsafe.} =
+proc walk(turn: var Turn; ds, origin: Cap; route: Route; transOff, stepOff: int) =
   if stepOff < route.pathSteps.len:
     let
       step = route.pathSteps[stepOff]
@@ -462,10 +482,10 @@ proc connectRoute(turn: var Turn; ds: Cap; route: Route; transOff: int) =
   onPublish(turn, ds, acceptPat) do (origin: Cap):
     walk(turn, ds, origin, route, transOff, 0)
 
-type StepCallback = proc (turn: var Turn; step: Value; origin, next: Cap) {.gcsafe.}
+type StepCallback = proc (turn: var Turn; step: Value; origin, next: Cap) {.closure.}
 
 proc spawnStepResolver(turn: var Turn; ds: Cap; stepType: Value; cb: StepCallback) =
-  spawn($stepType & "-step", turn) do (turn: var Turn):
+  spawnActor($stepType & "-step", turn) do (turn: var Turn):
     let stepPat = grabRecord(stepType, grab())
     let pat = ?Observe(pattern: ResolvedPathStep?:{1: stepPat}) ?? {0: grabLit(), 1: grab()}
     during(turn, ds, pat) do (origin: Cap; stepDetail: Literal[Value]):
@@ -487,7 +507,7 @@ proc spawnStepResolver(turn: var Turn; ds: Cap; stepType: Value; cb: StepCallbac
 
 proc spawnRelays*(turn: var Turn; ds: Cap) =
   ## Spawn actors that manage routes and appeasing gatekeepers.
-  spawn("transport-connector", turn) do (turn: var Turn):
+  spawnActor("transport-connector", turn) do (turn: var Turn):
     let pat = ?Observe(pattern: !TransportConnection) ?? { 0: grab() }
       # Use a generic pattern and type matching
       # in the during handler because it is easy.
@@ -496,15 +516,26 @@ proc spawnRelays*(turn: var Turn; ds: Cap) =
     during(turn, ds, stdioPat) do:
       connectTransport(turn, ds, Stdio())
 
+
     # TODO: tcp pattern
     during(turn, ds, pat) do (ta: Literal[transportAddress.Tcp]):
-      connectTransport(turn, ds, ta.value)
+      try: connectTransport(turn, ds, ta.value)
+      except CatchableError as e:
+        publish(turn, ds, TransportConnection(
+            `addr`: ta.toPreserve,
+            resolved: rejected(embed e),
+          ))
 
     # TODO: unix pattern
     during(turn, ds, pat) do (ta: Literal[transportAddress.Unix]):
-      connectTransport(turn, ds, ta.value)
+      try: connectTransport(turn, ds, ta.value)
+      except CatchableError as e:
+        publish(turn, ds, TransportConnection(
+            `addr`: ta.toPreserve,
+            resolved: rejected(embed e),
+          ))
 
-  spawn("path-resolver", turn) do (turn: var Turn):
+  spawnActor("path-resolver", turn) do (turn: var Turn):
     let pat = ?Observe(pattern: !ResolvePath) ?? {0: grab()}
     during(turn, ds, pat) do (route: Literal[Route]):
       for i, transAddr in route.value.transports:
@@ -515,10 +546,16 @@ proc spawnRelays*(turn: var Turn; ds: Cap) =
     publish(turn, ds, ResolvedPathStep(
         origin: origin, pathStep: step, resolved: next.accepted))
 
-type BootProc* = proc (turn: var Turn; ds: Cap) {.gcsafe.}
+type BootProc* = proc (turn: var Turn; ds: Cap) {.closure.}
+
+const defaultRoute* = "<route [<stdio>]>"
 
 proc envRoute*: Route =
-  var text = getEnv("SYNDICATE_ROUTE")
+  ## Get an route to a Syndicate capability from the calling environment.
+  ## On UNIX this is the SYNDICATE_ROUTE environmental variable with a
+  ## fallack to a defaultRoute_.
+  ## See https://git.syndicate-lang.org/syndicate-lang/syndicate-protocols/raw/branch/main/schemas/gatekeeper.prs.
+  var text = getEnv("SYNDICATE_ROUTE", defaultRoute)
   if text == "":
     var tx = (getEnv("XDG_RUNTIME_DIR", "/run/user/1000") / "dataspace").toPreserves
     result.transports = @[initRecord("unix", tx)]
@@ -529,7 +566,18 @@ proc envRoute*: Route =
       raise newException(ValueError, "failed to parse $SYNDICATE_ROUTE " & $pr)
 
 proc resolve*(turn: var Turn; ds: Cap; route: Route; bootProc: BootProc) =
+  ## Resolve `route` within `ds` and call `bootProc` with resolved capabilities.
   during(turn, ds, ResolvePath ?: {0: ?route, 3: ?:ResolvedAccepted}) do (dst: Cap):
     bootProc(turn, dst)
 
+proc resolveEnvironment*(turn: var Turn; bootProc: BootProc) =
+  ## Resolve a capability from the calling environment
+  ## and call `bootProc`. See envRoute_.
+  let
+    ds = newDataspace(turn)
+    pat = ResolvePath ?: {0: ?envRoute(), 3: ?:ResolvedAccepted}
+  during(turn, ds, pat) do (dst: Cap):
+    bootProc(turn, dst)
+  spawnRelays(turn, ds)
+
 # TODO: define a runActor that comes preloaded with relaying

src/syndicate/skeletons.nim

@@ -65,9 +65,9 @@ func isEmpty(cont: Continuation): bool =
   cont.cache.len == 0 and cont.leafMap.len == 0
 
 type
-  ContinuationProc = proc (c: Continuation; v: Value) {.gcsafe.}
-  LeafProc = proc (l: Leaf; v: Value) {.gcsafe.}
-  ObserverProc = proc (turn: var Turn; group: ObserverGroup; vs: seq[Value]) {.gcsafe.}
+  ContinuationProc = proc (c: Continuation; v: Value) {.closure.}
+  LeafProc = proc (l: Leaf; v: Value) {.closure.}
+  ObserverProc = proc (turn: var Turn; group: ObserverGroup; vs: seq[Value]) {.closure.}
 
 proc getLeaves(cont: Continuation; constPaths: Paths): LeafMap =
   result = cont.leafMap.getOrDefault(constPaths)

syndicate.nimble

@@ -1,6 +1,6 @@
 # Package
 
-version = "20240208"
+version = "20240304"
 author        = "Emery Hemingway"
 description   = "Syndicated actors for conversational concurrency"
 license       = "Unlicense"
@@ -9,4 +9,4 @@ srcDir        = "src"
 
 # Dependencies
 
-requires "https://github.com/ehmry/hashlib.git#f9455d4be988e14e3dc7933eb7cc7d7c4820b7ac", "nim >= 2.0.0", "https://git.syndicate-lang.org/ehmry/preserves-nim.git >= 20240208"
+requires "https://github.com/ehmry/hashlib.git >= 20231130", "nim >= 2.0.0", "https://git.syndicate-lang.org/ehmry/preserves-nim.git >= 20240208", "https://github.com/ehmry/nim-sys.git#4ef3b624db86e331ba334e705c1aa235d55b05e1", "https://github.com/nim-works/cps"

tests/test_timers.nim

@@ -2,16 +2,25 @@
 # SPDX-License-Identifier: Unlicense
 
 import std/times
+import pkg/sys/ioqueue
 import syndicate, syndicate/actors/timers
 
-proc now: float64 = getTime().toUnixFloat()
+let actor = bootActor("timer-test") do (turn: var Turn):
+  let timers = newDataspace(turn)
+  spawnTimerActor(timers, turn)
 
-runActor("test_timers") do (ds: Cap; turn: var Turn):
-  onPublish(turn, ds, grab(LaterThan(seconds: now()+1.0))) do:
-    stderr.writeLine "slept one second once"
-    onPublish(turn, ds, grab(LaterThan(seconds: now()+1.0))) do:
-      stderr.writeLine "slept one second twice"
-      onPublish(turn, ds, grab(LaterThan(seconds: now()+1.0))) do:
-        stderr.writeLine "slept one second thrice"
-        quit()
-  spawnTimers(turn, ds)
+  onPublish(turn, timers, ?LaterThan(seconds: 1356100000)):
+    echo "now in 13th bʼakʼtun"
+
+  after(turn, timers, initDuration(seconds = 3)) do (turn: var Turn):
+    echo "third timer expired"
+    stopActor(turn)
+
+  after(turn, timers, initDuration(seconds = 1)) do (turn: var Turn):
+    echo "first timer expired"
+
+  after(turn, timers, initDuration(seconds = 2)) do (turn: var Turn):
+    echo "second timer expired"
+
+echo "single run of ioqueue"
+ioqueue.run()