ehmry
/
syndicate-nim
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

CPS refactor

This commit is contained in:
Emery Hemingway 2024-02-19 22:40:40 +00:00
parent a8a17b4978
commit 9eae178723
9 changed files with 448 additions and 239 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
lock.json
View File

@ -38,11 +38,11 @@
"packages": [ "packages": [
"nimcrypto" "nimcrypto"
], ],
"path": "/nix/store/7b491gv9zlayilsh8k2gnyzw6znrh7xq-source", "path": "/nix/store/jwz8pqbv6rsm8w4fjzdb37r0wzjn5hv0-source",
"rev": "70151aa132f3a771996117c23c1fcaa8446a6f35", "rev": "d58da671799c69c0b3208b96c154e13c8b1a9e90",
"sha256": "1ldjz02p70wagqvk6vgcg16kjh7pkm1394qd1pcdmg8z39bm5ag3", "sha256": "12dm0gsy10ppga7zf7hpf4adaqjrd9b740n2w926xyazq1njf6k9",
"srcDir": "", "srcDir": "",
"url": "https://github.com/cheatfate/nimcrypto/archive/70151aa132f3a771996117c23c1fcaa8446a6f35.tar.gz" "url": "https://github.com/cheatfate/nimcrypto/archive/d58da671799c69c0b3208b96c154e13c8b1a9e90.tar.gz"
}, },
{ {
"method": "fetchzip", "method": "fetchzip",

424
src/sam/actors.nim
View File

@ -1,10 +1,10 @@
# SPDX-FileCopyrightText: ☭ Emery Hemingway # SPDX-FileCopyrightText: ☭ Emery Hemingway
# SPDX-License-Identifier: Unlicense # SPDX-License-Identifier: Unlicense
import std/[deques, hashes, options, times] import std/[deques, hashes, options, tables, times]
import pkg/cps import pkg/cps
import preserves import preserves
import ../syndicate/protocols/[protocol, sturdy] import ./protocols/[protocol, sturdy]
# const traceSyndicate {.booldefine.}: bool = true # const traceSyndicate {.booldefine.}: bool = true
const traceSyndicate* = true const traceSyndicate* = true
@ -18,9 +18,13 @@ export protocol.Handle
type type
Cont* = ref object of Continuation Cont* = ref object of Continuation
turn: Turn
facet: Facet facet: Facet
PublishProc* = proc (e: Entity; v: Value; h: Handle) {.cps: Cont.}
RetractProc* = proc (e: Entity; h: Handle) {.cps: Cont.}
MessageProc* = proc (e: Entity; v: Value) {.cps: Cont.}
SyncProc* = proc (e: Entity; peer: Cap) {.cps: Cont.}
Handler* = proc() {.closure.} Handler* = proc() {.closure.}
Work = Deque[Cont] Work = Deque[Cont]
@ -28,45 +32,53 @@ type
FacetState = enum fFresh, fRunning, fEnded FacetState = enum fFresh, fRunning, fEnded
Callback* = proc () {.nimcall.} Callback* = proc () {.closure.}
OutboundTable = Table[Handle, OutboundAssertion]
OutboundAssertion = ref object
handle: Handle
peer: Cap
established: bool
Facet* = ref object Facet* = ref object
## https://synit.org/book/glossary.html#facet ## https://synit.org/book/glossary.html#facet
actor: Actor actor: Actor
parent: Facet parent: Facet
children: seq[Facet] children: seq[Facet]
outbound: OutboundTable
stopHandlers: HandlerDeque stopHandlers: HandlerDeque
stopCallbacks: seq[Callback] stopCallbacks: seq[Callback]
state: FacetState state: FacetState
when traceSyndicate: id: FacetId
id: FacetId
Turn = ref object FacetProc* = proc (f: Facet) {.closure.}
## Type for callbacks to be called within a turn.
## The `Facet` parameter is the owning facet.
Turn {.byref.} = object
## https://synit.org/book/glossary.html#turn ## https://synit.org/book/glossary.html#turn
facet: Facet facet: Facet
entity: Entity entity: Entity
event: Option[protocol.Event]
work: Work work: Work
actions: seq[Cont]
event: Option[protocol.Event]
when traceSyndicate: when traceSyndicate:
desc: TurnDescription desc: TurnDescription
Entity* = ref object of RootObj Entity* = ref object of RootObj
## https://synit.org/book/glossary.html#entity ## https://synit.org/book/glossary.html#entity
facet*: Facet
oid*: sturdy.Oid # oid is how Entities are identified over the wire oid*: sturdy.Oid # oid is how Entities are identified over the wire
publishImpl*: PublishProc publishImpl*: PublishProc
retractImpl*: RetractProc retractImpl*: RetractProc
messageImpl*: MessageProc messageImpl*: MessageProc
syncImpl*: SyncProc syncImpl*: SyncProc
when traceSyndicate:
id: FacetId
Cap* {.final, preservesEmbedded.} = ref object of EmbeddedObj Cap* {.final, preservesEmbedded.} = ref object of EmbeddedObj
relay*: Facet relay*: Facet
target*: Entity target*: Entity
attenuation*: seq[sturdy.Caveat] attenuation*: seq[sturdy.Caveat]
Actor = ref object Actor* = ref object
## https://synit.org/book/glossary.html#actor ## https://synit.org/book/glossary.html#actor
# TODO: run on a seperate thread. # TODO: run on a seperate thread.
# crashHandlers: HandlerDeque # crashHandlers: HandlerDeque
@ -74,11 +86,12 @@ type
handleAllocator: Handle handleAllocator: Handle
facetIdAllocator: int facetIdAllocator: int
id: ActorId id: ActorId
turn: Turn
when traceSyndicate: when traceSyndicate:
traceStream: FileStream traceStream: FileStream
stopped: bool stopped: bool
template syndicate*(prc: typed): untyped = template turnWork*(prc: typed): untyped =
cps(Cont, prc) cps(Cont, prc)
proc activeFacet*(c: Cont): Facet {.cpsVoodoo.} = proc activeFacet*(c: Cont): Facet {.cpsVoodoo.} =
@ -93,6 +106,9 @@ using
cap: Cap cap: Cap
turn: Turn turn: Turn
proc `$`*(facet): string = $facet.id
proc `$`*(cap): string = "#:…"
proc hash*(facet): Hash = facet.unsafeAddr.hash proc hash*(facet): Hash = facet.unsafeAddr.hash
proc hash*(cap): Hash = cap.unsafeAddr.hash proc hash*(cap): Hash = cap.unsafeAddr.hash
@ -126,13 +142,19 @@ when traceSyndicate:
) )
actor.traceStream.writeLine($entry.toPreserves) actor.traceStream.writeLine($entry.toPreserves)
proc trace(actor; turn) = proc traceTurn(actor) =
if not actor.traceStream.isNil: if not actor.traceStream.isNil:
actor.trace(ActorActivation( actor.trace(ActorActivation(
orKind: ActorActivationKind.turn, orKind: ActorActivationKind.turn,
turn: turn.desc, turn: actor.turn.desc,
)) ))
proc traceTarget(facet): trace.Target =
Target(
actor: facet.actor.id,
facet: facet.id,
)
proc traceTarget(cap): trace.Target = proc traceTarget(cap): trace.Target =
let facet = cap.relay let facet = cap.relay
Target( Target(
@ -141,53 +163,80 @@ when traceSyndicate:
oid: cap.target.oid.toPreserves, oid: cap.target.oid.toPreserves,
) )
proc traceTarget(turn): trace.Target = proc traceEnqueue(actor; e: TargetedTurnEvent) =
let facet = turn.facet actor.turn.desc.actions.add ActionDescription(
Target( orKind: ActionDescriptionKind.enqueue,
actor: facet.actor.id, enqueue: ActionDescriptionEnqueue(event: e),
facet: facet.id,
) )
proc newExternalTurn(facet): Turn = proc traceDequeue(actor; e: TargetedTurnEvent) =
result = Turn(facet: facet) actor.turn.desc.actions.add ActionDescription(
when traceSyndicate: orKind: ActionDescriptionKind.dequeue,
result.desc = TurnDescription(cause: TurnCause(orKind: TurnCauseKind.external)) dequeue: ActionDescriptionDequeue(event: e),
)
proc pass*(a, b: Cont): Cont = proc pass*(a, b: Cont): Cont =
b.turn = a.turn assert not a.facet.isNil
if b.facet.isNil: b.facet = a.facet
b.facet = a.facet
# TODO: whelp a new continuation at facet boundaries?
b b
proc queue(t: Turn; c: Cont) = proc queueWork(facet; c: Cont) =
c.facet = t.facet c.facet = facet
t.work.addLast(c) facet.actor.turn.work.addLast(c)
proc queue(c: Cont): Cont {.cpsMagic.} = proc yieldWork(c: Cont): Cont {.cpsMagic.} =
queue(c.turn, c) ## Suspend and enqueue the caller until later in the turn.
assert not c.facet.isNil
c.facet.queueWork(c)
nil nil
proc complete(turn; c: Cont) = proc yieldToActions(c: Cont): Cont {.cpsMagic.} =
assert not c.facet.isNil
c.facet.actor.turn.actions.add(c)
nil
proc startExternalTurn(facet) =
let actor = facet.actor
assert actor.turn.work.len == 0
assert actor.turn.actions.len == 0
actor.turn.facet = facet
when traceSyndicate:
actor.turn.desc = TurnDescription(cause: TurnCause(orKind: TurnCauseKind.external))
proc terminate(actor; err: ref Exception) =
raise err
proc terminate(facet; err: ref Exception) =
terminate(facet.actor, err)
proc complete(c: Cont) =
var c = c var c = c
try: try:
while not c.isNil and not c.fn.isNil: while not c.isNil and not c.fn.isNil:
c.turn = turn
var y = c.fn var y = c.fn
var x = y(c) var x = y(c)
c = Cont(x) c = Cont(x)
except CatchableError as err: except CatchableError as err:
if not c.dismissed: if not c.dismissed:
writeStackFrames c writeStackFrames c
# terminate(c.facet, err) terminate(c.facet, err)
proc run(turn) = proc run(actor) =
let actor = turn.facet.actor
assert not actor.stopped assert not actor.stopped
while turn.work.len > 0: var n = 0
complete(turn, turn.work.popFirst()) while actor.turn.work.len > 0:
actor.turn.work.popFirst().complete()
inc n
echo n, " items completed from work queue"
when traceSyndicate:
actor.traceTurn()
var i: int
while i < actor.turn.actions.len:
complete(move actor.turn.actions[i])
inc i
echo i, " items completed from action queue"
turn.actions.setLen(0)
when traceSyndicate: when traceSyndicate:
actor.trace(turn)
if actor.stopped: if actor.stopped:
trace(actor, ActorActivation(orkind: ActorActivationKind.stop)) trace(actor, ActorActivation(orkind: ActorActivationKind.stop))
@ -196,11 +245,11 @@ proc start(actor; cont: Cont) =
var act = ActorActivation(orkind: ActorActivationKind.start) var act = ActorActivation(orkind: ActorActivationKind.start)
trace(actor, act) trace(actor, act)
actor.root.state = fRunning actor.root.state = fRunning
let turn = actor.root.newExternalTurn() actor.root.startExternalTurn()
turn.queue(cont) actor.root.queueWork(cont)
run(turn) run(actor)
proc stop(turn; actor) proc stop(actor)
proc collectPath(result: var seq[FacetId]; facet) = proc collectPath(result: var seq[FacetId]; facet) =
if not facet.parent.isNil: if not facet.parent.isNil:
@ -211,51 +260,47 @@ proc runNextStop(c: Cont; facet: Facet): Cont {.cpsMagic.} =
c.fn = facet.stopHandlers.pop() c.fn = facet.stopHandlers.pop()
result = c result = c
proc runNextFacetStop() {.syndicate.} = proc runNextFacetStop() {.cps: Cont.} =
activeFacet().runNextStop() activeFacet().runNextStop()
proc stop(turn; facet; reason: FacetStopReason) = proc stop(facet; reason: FacetStopReason) =
let actor = facet.actor
while facet.stopHandlers.len > 0: while facet.stopHandlers.len > 0:
var c = whelp runNextFacetStop() var c = whelp runNextFacetStop()
c.facet = facet c.facet = facet
complete(turn, c) complete(c)
while facet.stopCallbacks.len > 0: while facet.stopCallbacks.len > 0:
var cb = facet.stopCallbacks.pop() var cb = facet.stopCallbacks.pop()
cb() cb()
while facet.children.len > 0: while facet.children.len > 0:
stop(turn, facet.children.pop(), FacetStopReason.parentStopping) stop(facet.children.pop(), FacetStopReason.parentStopping)
when traceSyndicate: when traceSyndicate:
var act = ActionDescription(orKind: ActionDescriptionKind.facetstop) var act = ActionDescription(orKind: ActionDescriptionKind.facetstop)
collectPath(act.facetstop.path, facet) collectPath(act.facetstop.path, facet)
act.facetStop.reason = reason act.facetStop.reason = reason
turn.desc.actions.add act actor.turn.desc.actions.add act
if facet.parent.isNil: if facet.parent.isNil:
facet.actor.root = nil actor.root = nil
stop(turn, facet.actor) stop(actor)
proc stop(turn; actor) = proc stop(actor) =
if not actor.root.isNil: if not actor.root.isNil:
stop(turn, actor.root, FacetStopReason.actorStopping) stop(actor.root, FacetStopReason.actorStopping)
actor.stopped = true actor.stopped = true
proc bootActor*(name: string, c: Cont) = proc bootActor(name: string, c: Cont) =
start(newActor(name), c) start(newActor(name), c)
proc stopActor(c: Cont; a: Actor): Cont {.cpsMagic.} = # proc stopFacetAction(reason: FacetStopReason) {.syndicate.} =
stop(c.turn, a) # stop(c.facet, reason)
nil
proc stopFacet(c: Cont; f: Facet): Cont {.cpsMagic.} = proc stopActorAction() {.cps: Cont.} =
stop(c.turn, f, FacetStopReason.explicitAction) activeFacet().actor.stop()
nil
proc stopFacet*() {.syndicate.} = proc stopActor*(facet) =
queue() let c = whelp stopActorAction()
stop(activeTurn(), activeFacet(), FacetStopReason.explicitAction) c.facet = facet
facet.actor.turn.actions.add(c)
proc stopActor*() {.syndicate.} =
queue()
stop(activeTurn(), activeActor())
type type
AssertionRef* = ref object AssertionRef* = ref object
@ -270,67 +315,238 @@ proc nextHandle(facet: Facet): Handle =
inc(facet.actor.handleAllocator) inc(facet.actor.handleAllocator)
facet.actor.handleAllocator facet.actor.handleAllocator
proc publish*(cap: Cap; val: Value): Handle = proc actor(cap): Actor = cap.relay.actor
result = cap.relay.nextHandle()
type Bindings = Table[Value, Value]
proc attenuate(r: Cap; a: seq[Caveat]): Cap =
if a.len == 0: result = r
else: result = Cap(
relay: r.relay,
target: r.target,
attenuation: a & r.attenuation)
proc match(bindings: var Bindings; p: Pattern; v: Value): bool =
case p.orKind
of PatternKind.Pdiscard: result = true
of PatternKind.Patom:
result = case p.patom
of PAtom.Boolean: v.isBoolean
of PAtom.Double: v.isDouble
of PAtom.Signedinteger: v.isInteger
of PAtom.String: v.isString
of PAtom.Bytestring: v.isByteString
of PAtom.Symbol: v.isSymbol
of PatternKind.Pembedded:
result = v.isEmbedded
of PatternKind.Pbind:
if match(bindings, p.pbind.pattern, v):
bindings[p.pbind.pattern.toPreserves] = v
result = true
of PatternKind.Pand:
for pp in p.pand.patterns:
result = match(bindings, pp, v)
if not result: break
of PatternKind.Pnot:
var b: Bindings
result = not match(b, p.pnot.pattern, v)
of PatternKind.Lit:
result = p.lit.value == v
of PatternKind.PCompound:
case p.pcompound.orKind
of PCompoundKind.rec:
if v.isRecord and
p.pcompound.rec.label == v.label and
p.pcompound.rec.fields.len == v.arity:
result = true
for i, pp in p.pcompound.rec.fields:
if not match(bindings, pp, v[i]):
result = false
break
of PCompoundKind.arr:
if v.isSequence and p.pcompound.arr.items.len == v.sequence.len:
result = true
for i, pp in p.pcompound.arr.items:
if not match(bindings, pp, v[i]):
result = false
break
of PCompoundKind.dict:
if v.isDictionary:
result = true
for key, pp in p.pcompound.dict.entries:
let vv = step(v, key)
if vv.isNone or not match(bindings, pp, get vv):
result = true
break
proc match(p: Pattern; v: Value): Option[Bindings] =
var b: Bindings
if match(b, p, v):
result = some b
proc instantiate(t: Template; bindings: Bindings): Value =
case t.orKind
of TemplateKind.Tattenuate:
let v = instantiate(t.tattenuate.template, bindings)
let cap = v.unembed(Cap)
if cap.isNone:
raise newException(ValueError, "Attempt to attenuate non-capability")
result = attenuate(get cap, t.tattenuate.attenuation).embed
of TemplateKind.TRef:
let n = $t.tref.binding.int
try: result = bindings[n.toPreserves]
except KeyError:
raise newException(ValueError, "unbound reference: " & n)
of TemplateKind.Lit:
result = t.lit.value
of TemplateKind.Tcompound:
case t.tcompound.orKind
of TCompoundKind.rec:
result = initRecord(t.tcompound.rec.label, t.tcompound.rec.fields.len)
for i, tt in t.tcompound.rec.fields:
result[i] = instantiate(tt, bindings)
of TCompoundKind.arr:
result = initSequence(t.tcompound.arr.items.len)
for i, tt in t.tcompound.arr.items:
result[i] = instantiate(tt, bindings)
of TCompoundKind.dict:
result = initDictionary()
for key, tt in t.tcompound.dict.entries:
result[key] = instantiate(tt, bindings)
proc rewrite(r: Rewrite; v: Value): Value =
let bindings = match(r.pattern, v)
if bindings.isSome:
result = instantiate(r.template, get bindings)
proc examineAlternatives(cav: Caveat; v: Value): Value =
case cav.orKind
of CaveatKind.Rewrite:
result = rewrite(cav.rewrite, v)
of CaveatKind.Alts:
for r in cav.alts.alternatives:
result = rewrite(r, v)
if not result.isFalse: break
of CaveatKind.Reject: discard
of CaveatKind.unknown: discard
proc runRewrites(v: Value; a: openarray[Caveat]): Value =
result = v
for stage in a:
result = examineAlternatives(stage, result)
if result.isFalse: break
proc publish(c: Cont; e: Entity; v: Value; h: Handle): Cont {.cpsMagic.} =
pass c, e.publishImpl.call(e, v, h)
proc retract(c: Cont; e: Entity; h: Handle): Cont {.cpsMagic.} =
pass c, e.retractImpl.call(e, h)
proc message(c: Cont; e: Entity; v: Value): Cont {.cpsMagic.} =
pass c, e.messageImpl.call(e, v)
proc sync(c: Cont; e: Entity; p: Cap): Cont {.cpsMagic.} =
pass c, e.syncImpl.call(e, p)
proc turnPublish(cap: Cap; val: Value; h: Handle) {.turnWork.} =
when traceSyndicate: when traceSyndicate:
var act = ActionDescription(orKind: ActionDescriptionKind.enqueue) var traceEvent = TargetedTurnEvent(
act.enqueue.event = TargetedTurnEvent(
target: cap.traceTarget, target: cap.traceTarget,
detail: trace.TurnEvent(orKind: trace.TurnEventKind.assert) detail: trace.TurnEvent(orKind: trace.TurnEventKind.assert)
) )
act.enqueue.event.detail = trace.TurnEvent(orKind: TurnEventKind.assert) traceEvent.detail = trace.TurnEvent(orKind: TurnEventKind.assert)
act.enqueue.event.detail.assert = TurnEventAssert( traceEvent.detail.assert = TurnEventAssert(
assertion: AssertionDescription(orKind: AssertionDescriptionKind.value), assertion: AssertionDescription(orKind: AssertionDescriptionKind.value),
handle: result, handle: h,
) )
act.enqueue.event.detail.assert.assertion.value.value = val traceEvent.detail.assert.assertion.value.value = val
turn.desc.actions.add act cap.actor.traceEnqueue(traceEvent)
cap.target.publish(val, result) cap.relay.outbound[h] = OutboundAssertion(handle: h, peer: cap)
yieldToActions()
proc retract*(cap: Cap; h: Handle) =
when traceSyndicate: when traceSyndicate:
var act = ActionDescription(orKind: ActionDescriptionKind.enqueue) cap.actor.traceDequeue(traceEvent)
act.enqueue.event = TargetedTurnEvent( cap.target.publish(val, h)
target: turn.traceTarget, cap.relay.outbound[h].established = true
proc turnRetract(cap: Cap; h: Handle) {.turnWork.} =
when traceSyndicate:
var traceEvent = TargetedTurnEvent(
target: cap.traceTarget,
detail: trace.TurnEvent(orKind: trace.TurnEventKind.retract) detail: trace.TurnEvent(orKind: trace.TurnEventKind.retract)
) )
act.enqueue.event.detail.retract = TurnEventRetract(handle: h) traceEvent.detail.retract.handle = h
turn.desc.actions.add act cap.actor.traceEnqueue(traceEvent)
yieldToActions()
when traceSyndicate:
cap.actor.traceDequeue(traceEvent)
cap.target.retract(h) cap.target.retract(h)
proc message*(cap; val: Value) = proc turnMessage(cap: Cap; val: Value) {.turnWork.} =
var val = runRewrites(val, cap.attenuation)
when traceSyndicate: when traceSyndicate:
var act = ActionDescription(orKind: ActionDescriptionKind.enqueue) var traceEvent = TargetedTurnEvent(
act.enqueue.event = TargetedTurnEvent( target: cap.traceTarget,
target: turn.traceTarget,
detail: trace.TurnEvent(orKind: trace.TurnEventKind.message) detail: trace.TurnEvent(orKind: trace.TurnEventKind.message)
) )
act.enqueue.event.detail.message.body.value.value = val traceEvent.detail.message.body.value.value = val
turn.desc.actions.add act cap.actor.traceEnqueue(traceEvent)
cap.entity.message(val) yieldToActions()
proc sync*(peer: Cap) =
when traceSyndicate: when traceSyndicate:
var act = ActionDescription(orKind: ActionDescriptionKind.enqueue) cap.actor.traceDequeue(traceEvent)
act.enqueue.event = TargetedTurnEvent( cap.target.message(val)
target: turn.traceTarget,
proc turnSync(cap: Cap; peer: Cap) {.turnWork.} =
when traceSyndicate:
var traceEvent = TargetedTurnEvent(
target: cap.traceTarget,
detail: trace.TurnEvent(orKind: trace.TurnEventKind.sync) detail: trace.TurnEvent(orKind: trace.TurnEventKind.sync)
) )
act.enqueue.event.detail.sync.peer = peer.traceTarget traceEvent.detail.sync.peer = peer.traceTarget
turn.desc.actions.add act cap.actor.traceEnqueue(traceEvent)
peer.entity.sync() yieldToActions()
when traceSyndicate:
cap.actor.traceDequeue(traceEvent)
cap.target.sync(peer)
proc installStopHook*(c: Cont, facet: Facet): Cont {.cpsMagic.} = proc publish*(cap; val: Value): Handle =
var val = runRewrites(val, cap.attenuation)
# TODO: attenuation to nothing?
result = cap.relay.nextHandle()
cap.relay.queueWork(whelp turnPublish(cap, val, result))
proc publish*[T](cap; x: T): Handle =
publish(cap, x.toPreserves)
proc retract*(cap; h: Handle) =
cap.relay.queueWork(whelp turnRetract(cap, h))
proc message*(cap; val: Value) =
var val = runRewrites(val, cap.attenuation)
cap.relay.queueWork(whelp turnMessage(cap, val))
proc message*[T](cap; x: T) =
message(cap, x.toPreserves)
proc sync*(cap, peer: Cap) =
cap.relay.queueWork(whelp turnSync(cap, peer))
proc installStopHook(c: Cont, facet: Facet): Cont {.cpsMagic.} =
facet.stopHandlers.add(c.fn) facet.stopHandlers.add(c.fn)
return c return c
proc addOnStopHandler*(c: Cont; cb: Callback): Cont {.cpsMagic.} = proc addOnStopHandler(c: Cont; cb: Callback): Cont {.cpsMagic.} =
c.facet.stopCallbacks.add(cb) c.facet.stopCallbacks.add(cb)
result = c result = c
type FacetProc* = proc (f: Facet) {.nimcall.} proc onStop*(facet; cb: proc () {.closure.}) =
facet.stopCallbacks.add(cb)
proc bootActor*(name: string; bootProc: FacetProc): Actor = proc bootActor*(name: string; bootProc: FacetProc): Actor =
result = newActor(name) result = newActor(name)
# TODO: start a turn, trace reason is Eternal, result.root.startExternalTurn()
bootProc(result.root) bootProc(result.root)
proc runActor*(name: string; bootProc: FacetProc) =
let actor = bootActor(name, bootProc)
while not actor.stopped:
run(actor)

32
src/sam/actors/timers.nim
View File

@ -3,10 +3,8 @@
import std/[asyncdispatch, monotimes, times, posix, times, epoll] import std/[asyncdispatch, monotimes, times, posix, times, epoll]
import preserves import preserves
import syndicate import ../syndicate, ../protocols/timer
from ../protocols/dataspace import Observe
import ../protocols/timer
from ../syndicate/protocols/dataspace import Observe
export timer export timer
@ -24,12 +22,16 @@ proc eventfd(count: cuint, flags: cint): cint
proc now: float64 = getTime().toUnixFloat() proc now: float64 = getTime().toUnixFloat()
proc processTimers(ds: Cap) {.turnAction.} = proc spawnTimers*(ds: Cap): Actor {.discardable.} =
let pat = inject(grab Observe(pattern: dropType LaterThan), {0: grabLit()}) ## Spawn a timer actor.
during(ds, pat) do (seconds: float): bootActor("timers") do (root: Facet):
let period = seconds - now() let pat = inject(grab Observe(pattern: dropType LaterThan), {0: grabLit()})
if period < 0.001 or true: #[
let h = publish(ds, LaterThan(seconds: seconds).toPreserves) during(ds, pat) do (seconds: float):
let period = seconds - now()
if period < 0.001 or true:
let h = publish(ds, LaterThan(seconds: seconds).toPreserves)
]#
#[ #[
else: else:
@ -39,14 +41,12 @@ proc processTimers(ds: Cap) {.turnAction.} =
discard publish(turn, ds, LaterThan(seconds: seconds)) discard publish(turn, ds, LaterThan(seconds: seconds))
]# ]#
proc spawnTimers*(ds: Cap) = #[
## Spawn a timer actor. proc after*(ds: Cap; dur: Duration; cb: proc () {.closure.}) =
boot(newActor("timers"), whelp processTimers(ds))
proc after*(turn: Turn; ds: Cap; dur: Duration; act: TurnAction) =
## Execute `act` after some duration of time. ## Execute `act` after some duration of time.
let later = now() + dur.inMilliseconds.float64 * 1_000.0 let later = now() + dur.inMilliseconds.float64 * 1_000.0
onPublish(ds, grab LaterThan(seconds: later)): onPublish(ds, grab LaterThan(seconds: later)):
act(turn) cb()
]#
# TODO: periodic timer # TODO: periodic timer

41
src/sam/dataspaces.nim
View File

@ -4,7 +4,7 @@
import std/[hashes, options, tables] import std/[hashes, options, tables]
import pkg/cps import pkg/cps
import preserves import preserves
import ./actors, ./protocols/dataspace, ./skeletons import ./[actors, patterns, skeletons]
from ./protocols/protocol import Handle from ./protocols/protocol import Handle
from ./protocols/dataspace import Observe from ./protocols/dataspace import Observe
@ -14,28 +14,35 @@ type
index: Index index: Index
handleMap: Table[Handle, Value] handleMap: Table[Handle, Value]
method publish(ds: Dataspace; turn: Turn; a: AssertionRef; h: Handle) {.gcsafe.} = proc dsPublish(e: Entity; v: Value; h: Handle) {.cps: Cont.} =
if add(ds.index, turn, a.value): var ds = Dataspace(e)
var obs = a.value.preservesTo(Observe) if ds.index.add(v):
var obs = v.preservesTo(Observe)
if obs.isSome and obs.get.observer of Cap: if obs.isSome and obs.get.observer of Cap:
ds.index.add(turn, obs.get.pattern, Cap(obs.get.observer)) ds.index.add(obs.get.pattern, Cap(obs.get.observer))
ds.handleMap[h] = a.value ds.handleMap[h] = v
method retract(ds: Dataspace; turn: Turn; h: Handle) {.gcsafe.} = proc dsRetract(e: Entity; h: Handle) {.cps: Cont.} =
let v = ds.handleMap[h] var ds = Dataspace(e)
if remove(ds.index, turn, v): var v = ds.handleMap[h]
if ds.index.remove(v):
ds.handleMap.del h ds.handleMap.del h
var obs = v.preservesTo(Observe) var obs = v.preservesTo(Observe)
if obs.isSome and obs.get.observer of Cap: if obs.isSome and obs.get.observer of Cap:
ds.index.remove(turn, obs.get.pattern, Cap(obs.get.observer)) ds.index.remove(obs.get.pattern, Cap(obs.get.observer))
method message(ds: Dataspace; turn: Turn; a: AssertionRef) {.gcsafe.} = proc dsMessage(e: Entity; v: Value) {.cps: Cont.} =
ds.index.deliverMessage(turn, a.value) var ds = Dataspace(e)
ds.index.deliverMessage(v)
proc newDataspace*(f: Facet): Cap = proc newDataspace*(f: Facet): Cap =
newCap(f, Dataspace(index: initIndex())) var ds = Dataspace(
publishImpl: whelp dsPublish,
retractImpl: whelp dsRetract,
messageImpl: whelp dsMessage,
index: initIndex(),
)
newCap(f, ds)
type BootProc = proc (ds: Cap) proc observe*(cap: Cap; pat: Pattern; e: Entity): Handle =
publish(cap, Observe(pattern: pat, observer: newCap(cap.relay, e)))
proc newDataspace*(): Cap {.syndicate.} =
activeFacet().newDataspace()

45
src/sam/durings.nim
View File

@ -3,34 +3,32 @@
import std/[hashes, tables] import std/[hashes, tables]
import preserves import preserves
import ./actors, ./patterns, ./protocols/dataspace import ./[actors, patterns]
type type
DuringProc* = proc (a: Value; h: Handle) DuringProc* = proc (a: Value; h: Handle): FacetProc {.gcsafe.}
DuringActionKind = enum null, dead, act DuringActionKind = enum null, dead, act
DuringAction = object DuringAction = object
case kind: DuringActionKind case kind: DuringActionKind
of null, dead: discard of null, dead: discard
of act: of act:
action: Cont retractProc: FacetProc
DuringEntity {.final.}= ref object of Entity DuringEntity {.final.}= ref object of Entity
cb: DuringProc publishProc: DuringProc
assertionMap: Table[Handle, DuringAction] assertionMap: Table[Handle, DuringAction]
method publish(de: DuringEntity; turn: Turn; a: AssertionRef; h: Handle) = proc duringPublish(e: Entity; v: Value; h: Handle) {.cps: Cont.} =
let action = de.cb(turn, a.value, h) var de = DuringEntity(e)
# assert(not action.isNil "should have put in a no-op action") let handler = de.handler(de.facet, a.value, h)
let g = de.assertionMap.getOrDefault h let g = de.assertionMap.getOrDefault h
case g.kind case g.kind
of null: of null, dead:
de.assertionMap[h] = DuringAction(kind: act, action: action) de.assertionMap[h] = DuringAction(kind: act, action: handler)
of dead:
de.assertionMap.del h
freshen(turn, action)
of act: of act:
raiseAssert("during: duplicate handle in publish: " & $h) raiseAssert("during: duplicate handle in publish: " & $h)
method retract(de: DuringEntity; turn: Turn; h: Handle) = proc duringRetract(e: Entity; h: Handle) {.cps: Cont.} =
var de = DuringEntity(e)
let g = de.assertionMap.getOrDefault h let g = de.assertionMap.getOrDefault h
case g.kind case g.kind
of null: of null:
@ -40,25 +38,6 @@ method retract(de: DuringEntity; turn: Turn; h: Handle) =
of act: of act:
de.assertionMap.del h de.assertionMap.del h
if not g.action.isNil: if not g.action.isNil:
turn.queue(g.action) g.action(de.facet)
proc during*(cb: DuringProc): DuringEntity = DuringEntity(cb: cb) proc during*(cb: DuringProc): DuringEntity = DuringEntity(cb: cb)
proc observe*(turn: Turn; ds: Cap; pat: Pattern; e: Entity): Handle =
publish(turn, ds, Observe(pattern: pat, observer: newCap(turn, e)).toPreserves)
proc assertHandle(turn: Turn): Handle =
doAssert(
turn.event.isSome and turn.event.get.orKind == EventKind.Assert,
"operation not valid during this turn")
turn.event.get.assert.handle
proc awaitRetraction(cont: Cont): Cont {.cpsMagic.} =
{.error: "this cannot work".}
let h = cont.turn.assertHandle
while true:
let turn = cont.turn
if turn.retracts(h):
return c
else:
turn.facet.actor.qeueue(c)

60
src/sam/skeletons.nim
View File

@ -65,9 +65,9 @@ func isEmpty(cont: Continuation): bool =
cont.cache.len == 0 and cont.leafMap.len == 0 cont.cache.len == 0 and cont.leafMap.len == 0
type type
ContinuationProc = proc (c: Continuation; v: Value) {.gcsafe.} ContinuationProc = proc (c: Continuation; v: Value) {.closure.}
LeafProc = proc (l: Leaf; v: Value) {.gcsafe.} LeafProc = proc (l: Leaf; v: Value) {.closure.}
ObserverProc = proc (turn: Turn; group: ObserverGroup; vs: seq[Value]) {.gcsafe.} ObserverProc = proc (group: ObserverGroup; vs: seq[Value]) {.closure.}
proc getLeaves(cont: Continuation; constPaths: Paths): LeafMap = proc getLeaves(cont: Continuation; constPaths: Paths): LeafMap =
result = cont.leafMap.getOrDefault(constPaths) result = cont.leafMap.getOrDefault(constPaths)
@ -114,10 +114,10 @@ proc top(stack: TermStack): Value =
assert stack.len > 0 assert stack.len > 0
stack[stack.high] stack[stack.high]
proc modify(node: Node; turn: Turn; outerValue: Value; event: EventKind; proc modify(node: Node; outerValue: Value; event: EventKind;
modCont: ContinuationProc; modLeaf: LeafProc; modObs: ObserverProc) = modCont: ContinuationProc; modLeaf: LeafProc; modObs: ObserverProc) =
proc walk(cont: Continuation; turn: Turn) = proc walk(cont: Continuation) =
modCont(cont, outerValue) modCont(cont, outerValue)
for constPaths, constValMap in cont.leafMap.pairs: for constPaths, constValMap in cont.leafMap.pairs:
let constVals = projectPaths(outerValue, constPaths) let constVals = projectPaths(outerValue, constPaths)
@ -129,7 +129,7 @@ proc modify(node: Node; turn: Turn; outerValue: Value; event: EventKind;
for capturePaths, observerGroup in leaf.observerGroups.pairs: for capturePaths, observerGroup in leaf.observerGroups.pairs:
let captures = projectPaths(outerValue, capturePaths) let captures = projectPaths(outerValue, capturePaths)
if captures.isSome: if captures.isSome:
modObs(turn, observerGroup, get captures) modObs(observerGroup, get captures)
of removedEvent: of removedEvent:
let leaf = constValMap.getOrDefault(get constVals) let leaf = constValMap.getOrDefault(get constVals)
if not leaf.isNil: if not leaf.isNil:
@ -137,13 +137,13 @@ proc modify(node: Node; turn: Turn; outerValue: Value; event: EventKind;
for capturePaths, observerGroup in leaf.observerGroups.pairs: for capturePaths, observerGroup in leaf.observerGroups.pairs:
let captures = projectPaths(outerValue, capturePaths) let captures = projectPaths(outerValue, capturePaths)
if captures.isSome: if captures.isSome:
modObs(turn, observerGroup, get captures) modObs(observerGroup, get captures)
if leaf.isEmpty: if leaf.isEmpty:
constValMap.del(get constVals) constValMap.del(get constVals)
proc walk(node: Node; turn: Turn; termStack: TermStack) = proc walk(node: Node; termStack: TermStack) =
walk(node.continuation, turn) walk(node.continuation)
for selector, table in node.edges: for selector, table in node.edges:
let let
nextStack = pop(termStack, selector.popCount) nextStack = pop(termStack, selector.popCount)
@ -153,11 +153,11 @@ proc modify(node: Node; turn: Turn; outerValue: Value; event: EventKind;
if nextClass.kind != classNone: if nextClass.kind != classNone:
let nextNode = table.getOrDefault(nextClass) let nextNode = table.getOrDefault(nextClass)
if not nextNode.isNil: if not nextNode.isNil:
walk(nextNode, turn, push(nextStack, get nextValue)) walk(nextNode, push(nextStack, get nextValue))
if event == removedEvent and nextNode.isEmpty: if event == removedEvent and nextNode.isEmpty:
table.del(nextClass) table.del(nextClass)
walk(node, turn, @[@[outerValue].toPreserves]) walk(node, @[@[outerValue].toPreserves])
proc getOrNew[A, B, C](t: var Table[A, TableRef[B, C]], k: A): TableRef[B, C] = proc getOrNew[A, B, C](t: var Table[A, TableRef[B, C]], k: A): TableRef[B, C] =
result = t.getOrDefault(k) result = t.getOrDefault(k)
@ -227,7 +227,7 @@ proc getEndpoints(leaf: Leaf; capturePaths: Paths): ObserverGroup =
if captures.isSome: if captures.isSome:
discard result.cachedCaptures.change(get captures, +1) discard result.cachedCaptures.change(get captures, +1)
proc add*(index: var Index; turn: Turn; pattern: Pattern; observer: Cap) = proc add*(index: var Index; pattern: Pattern; observer: Cap) =
let let
cont = index.root.extend(pattern) cont = index.root.extend(pattern)
analysis = analyse pattern analysis = analyse pattern
@ -237,10 +237,10 @@ proc add*(index: var Index; turn: Turn; pattern: Pattern; observer: Cap) =
# TODO if endpoints.cachedCaptures.len > 0: # TODO if endpoints.cachedCaptures.len > 0:
var captureMap = newTable[seq[Value], Handle]() var captureMap = newTable[seq[Value], Handle]()
for capture in endpoints.cachedCaptures.items: for capture in endpoints.cachedCaptures.items:
captureMap[capture] = publish(turn, observer, capture.toPreserves) captureMap[capture] = publish(observer, capture.toPreserves)
endpoints.observers[observer] = captureMap endpoints.observers[observer] = captureMap
proc remove*(index: var Index; turn: Turn; pattern: Pattern; observer: Cap) = proc remove*(index: var Index; pattern: Pattern; observer: Cap) =
let let
cont = index.root.extend(pattern) cont = index.root.extend(pattern)
analysis = analyse pattern analysis = analyse pattern
@ -252,7 +252,7 @@ proc remove*(index: var Index; turn: Turn; pattern: Pattern; observer: Cap) =
if not endpoints.isNil: if not endpoints.isNil:
var captureMap: TableRef[seq[Value], Handle] var captureMap: TableRef[seq[Value], Handle]
if endpoints.observers.pop(observer, captureMap): if endpoints.observers.pop(observer, captureMap):
for handle in captureMap.values: retract(turn, handle) for handle in captureMap.values: retract(observer, handle)
if endpoints.observers.len == 0: if endpoints.observers.len == 0:
leaf.observerGroups.del(analysis.capturePaths) leaf.observerGroups.del(analysis.capturePaths)
if leaf.observerGroups.len == 0: if leaf.observerGroups.len == 0:
@ -260,7 +260,7 @@ proc remove*(index: var Index; turn: Turn; pattern: Pattern; observer: Cap) =
if constValMap.len == 0: if constValMap.len == 0:
cont.leafMap.del(analysis.constPaths) cont.leafMap.del(analysis.constPaths)
proc adjustAssertion(index: var Index; turn: Turn; outerValue: Value; delta: int): bool = proc adjustAssertion(index: var Index; outerValue: Value; delta: int): bool =
case index.allAssertions.change(outerValue, delta) case index.allAssertions.change(outerValue, delta)
of cdAbsentToPresent: of cdAbsentToPresent:
result = true result = true
@ -268,37 +268,37 @@ proc adjustAssertion(index: var Index; turn: Turn; outerValue: Value; delta: int
c.cache.incl(v) c.cache.incl(v)
proc modLeaf(l: Leaf; v: Value) = proc modLeaf(l: Leaf; v: Value) =
l.cache.incl(v) l.cache.incl(v)
proc modObserver(turn: Turn; group: ObserverGroup; vs: seq[Value]) = proc modObserver(group: ObserverGroup; vs: seq[Value]) =
let change = group.cachedCaptures.change(vs, +1) let change = group.cachedCaptures.change(vs, +1)
if change == cdAbsentToPresent: if change == cdAbsentToPresent:
for (observer, captureMap) in group.observers.pairs: for (observer, captureMap) in group.observers.pairs:
captureMap[vs] = publish(turn, observer, vs.toPreserves) captureMap[vs] = publish(observer, vs.toPreserves)
# TODO: this handle is coming from the facet? # TODO: this handle is coming from the facet?
modify(index.root, turn, outerValue, addedEvent, modContinuation, modLeaf, modObserver) modify(index.root, outerValue, addedEvent, modContinuation, modLeaf, modObserver)
of cdPresentToAbsent: of cdPresentToAbsent:
result = true result = true
proc modContinuation(c: Continuation; v: Value) = proc modContinuation(c: Continuation; v: Value) =
c.cache.excl(v) c.cache.excl(v)
proc modLeaf(l: Leaf; v: Value) = proc modLeaf(l: Leaf; v: Value) =
l.cache.excl(v) l.cache.excl(v)
proc modObserver(turn: Turn; group: ObserverGroup; vs: seq[Value]) = proc modObserver(group: ObserverGroup; vs: seq[Value]) =
if group.cachedCaptures.change(vs, -1) == cdPresentToAbsent: if group.cachedCaptures.change(vs, -1) == cdPresentToAbsent:
for (observer, captureMap) in group.observers.pairs: for (observer, captureMap) in group.observers.pairs:
var h: Handle var h: Handle
if captureMap.take(vs, h): if captureMap.take(vs, h):
retract(observer.target, turn, h) retract(observer, h)
modify(index.root, turn, outerValue, removedEvent, modContinuation, modLeaf, modObserver) modify(index.root, outerValue, removedEvent, modContinuation, modLeaf, modObserver)
else: discard else: discard
proc continuationNoop(c: Continuation; v: Value) = discard proc continuationNoop(c: Continuation; v: Value) = discard
proc leafNoop(l: Leaf; v: Value) = discard proc leafNoop(l: Leaf; v: Value) = discard
proc add*(index: var Index; turn: Turn; v: Value): bool = proc add*(index: var Index; v: Value): bool =
adjustAssertion(index, turn, v, +1) adjustAssertion(index, v, +1)
proc remove*(index: var Index; turn: Turn; v: Value): bool = proc remove*(index: var Index; v: Value): bool =
adjustAssertion(index, turn, v, -1) adjustAssertion(index, v, -1)
proc deliverMessage*(index: var Index; turn: Turn; v: Value) = proc deliverMessage*(index: var Index; v: Value) =
proc observersCb(turn: Turn; group: ObserverGroup; vs: seq[Value]) = proc observersCb(group: ObserverGroup; vs: seq[Value]) =
for observer in group.observers.keys: message(turn, observer, vs.toPreserves) for observer in group.observers.keys: message(observer, vs.toPreserves)
index.root.modify(turn, v, messageEvent, continuationNoop, leafNoop, observersCb) index.root.modify(v, messageEvent, continuationNoop, leafNoop, observersCb)

50
src/sam/syndicate.nim
View File

@ -37,22 +37,26 @@ proc `??`*(pat: Pattern; bindings: openArray[(int, Pattern)]): Pattern {.inline.
patterns.inject(pat, bindings) patterns.inject(pat, bindings)
type type
PublishProc = proc (turn: Turn; v: Value; h: Handle) {.closure.} PublishProc = proc (v: Value; h: Handle) {.closure.}
RetractProc = proc (turn: Turn; h: Handle) {.closure.} RetractProc = proc (h: Handle) {.closure.}
MessageProc = proc (turn: Turn; v: Value) {.closure.} MessageProc = proc (v: Value) {.closure.}
ClosureEntity = ref object of Entity ClosureEntity = ref object of Entity
publishImpl*: PublishProc publishCb*: PublishProc
retractImpl*: RetractProc retractCb*: RetractProc
messageImpl*: MessageProc messageCb*: MessageProc
method publish(e: ClosureEntity; turn: Turn; a: AssertionRef; h: Handle) = proc publishCont(e: Entity; v: Value; h: Handle) {.cps: Cont.} =
if not e.publishImpl.isNil: e.publishImpl(turn, a.value, h) var ce = ClosureEntity(e)
if not ce.publishCb.isNil: ce.publishCb(v, h)
method retract(e: ClosureEntity; turn: Turn; h: Handle) = proc retractCont(e: Entity; h: Handle) {.cps: Cont.} =
if not e.retractImpl.isNil: e.retractImpl(turn, h) var ce = ClosureEntity(e)
if not ce.retractCb.isNil: ce.retractCb(h)
method message(e: ClosureEntity; turn: Turn; a: AssertionRef) = proc messageCont(e: Entity; v: Value) {.cps: Cont.} =
if not e.messageImpl.isNil: e.messageImpl(turn, a.value) var ce = ClosureEntity(e)
if not ce.messageCb.isNil: ce.messageCb(v)
proc argumentCount(handler: NimNode): int = proc argumentCount(handler: NimNode): int =
handler.expectKind {nnkDo, nnkStmtList} handler.expectKind {nnkDo, nnkStmtList}
@ -108,7 +112,7 @@ proc wrapMessageHandler(handler: NimNode): NimNode =
handlerSym = genSym(nskProc, "message") handlerSym = genSym(nskProc, "message")
bindingsSym = ident"bindings" bindingsSym = ident"bindings"
quote do: quote do:
proc `handlerSym`(turn: Turn; `bindingsSym`: Value) = proc `handlerSym`(`bindingsSym`: Value) =
`varSection` `varSection`
if fromPreserves(`valuesSym`, bindings): if fromPreserves(`valuesSym`, bindings):
`body` `body`
@ -150,9 +154,8 @@ macro onPublish*(ds: Cap; pattern: Pattern; handler: untyped) =
discard observe(activeTurn(), `ds`, `pattern`, ClosureEntity(publishImpl: `handlerSym`)) discard observe(activeTurn(), `ds`, `pattern`, ClosureEntity(publishImpl: `handlerSym`))
]# ]#
#[ macro onMessage*(cap: Cap; pattern: Pattern; handler: untyped) =
macro onMessage*(ds: Cap; pattern: Pattern; handler: untyped) = ## Call `handler` when an message matching `pattern` is broadcasted at `cap`.
## Call `handler` when an message matching `pattern` is broadcasted at `ds`.
let let
argCount = argumentCount(handler) argCount = argumentCount(handler)
handlerProc = wrapMessageHandler(handler) handlerProc = wrapMessageHandler(handler)
@ -161,8 +164,10 @@ macro onMessage*(ds: Cap; pattern: Pattern; handler: untyped) =
if `argCount` != 0 and `pattern`.analyse.capturePaths.len != `argCount`: if `argCount` != 0 and `pattern`.analyse.capturePaths.len != `argCount`:
raiseAssert($`pattern`.analyse.capturePaths.len & " values captured but handler has " & $`argCount` & " arguments - " & $`pattern`) raiseAssert($`pattern`.analyse.capturePaths.len & " values captured but handler has " & $`argCount` & " arguments - " & $`pattern`)
`handlerProc` `handlerProc`
discard observe(activeTurn(), `ds`, `pattern`, ClosureEntity(messageImpl: `handlerSym`)) discard observe(`cap`, `pattern`, ClosureEntity(
]# messageImpl: whelp messageCont,
messageCb: `handlerSym`,
))
#[ #[
macro during*(ds: Cap; pattern: Pattern; publishBody, retractBody: untyped) = macro during*(ds: Cap; pattern: Pattern; publishBody, retractBody: untyped) =
@ -202,15 +207,15 @@ proc wrapHandler(body: NimNode; ident: string): NimNode =
proc `sym`() = proc `sym`() =
`body` `body`
macro onStop*(body: untyped) = #[
macro onStop*(facet: Facet; body: untyped) =
let let
handlerDef = wrapHandler(body, "onStop") handlerDef = wrapHandler(body, "onStop")
handlerSym = handlerDef[0] handlerSym = handlerDef[0]
result = quote do: result = quote do:
`handlerDef` `handlerDef`
addOnStopHandler(`handlerSym`) addOnStopHandler(facet, `handlerSym`)
#[
macro onStop*(body: untyped) = macro onStop*(body: untyped) =
quote do: quote do:
block: block:
@ -220,6 +225,3 @@ macro onStop*(body: untyped) =
`body` `body`
return return
]# ]#
proc runActor*(name: string; bootProc: FacetProc) =
let actor = spawnActor(name, bootProc)

2
syndicate.nimble
View File

@ -1,6 +1,6 @@
# Package # Package
version = "20240216" version = "20240219"
author = "Emery Hemingway" author = "Emery Hemingway"
description = "Syndicated actors for conversational concurrency" description = "Syndicated actors for conversational concurrency"
license = "Unlicense" license = "Unlicense"

25
tests/test_timers.nim
View File

@ -3,29 +3,34 @@
import std/times import std/times
import pkg/cps import pkg/cps
import syndicate import sam/syndicate
# import syndicate/actors/timers import sam/actors/timers
proc now: float64 = getTime().toUnixFloat() proc now: float64 = getTime().toUnixFloat()
runActor("timer-test") do (root: Facet): runActor("timer-test") do (root: Facet):
echo "actor calls boot proc with root facte ", root
let ds = facet.newDataspace() let ds = root.newDataspace()
let h = facet.publish(ds, "hello world!".toPreserves) echo "new dataspace ", ds
let h = publish(ds, "hello world!".toPreserves)
echo "published to handle ", h
facet.onMessage(ds, grab()) do (v: Value): onMessage(ds, grab()) do (v: Value):
stderr.writeLine "observed message ", v stderr.writeLine "observed message ", v
facet.message(ds, "hello world!".toPreserves) message(ds, "hello world!".toPreserves)
facet.retract(h) echo "sent a message"
facet.sync(ds) retract(ds, h)
echo "retracted handle ", h
# facet.sync(ds)
facet.onStop: root.onStop:
echo "anonymous stop handler was invoked" echo "anonymous stop handler was invoked"
echo "stopping actor" echo "stopping actor"
facet.stopActor() root.stopActor()
echo "actor stopped but still executing?" echo "actor stopped but still executing?"
#[ #[