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

Cleanup actors module 

Use methods at Entity rather than proc pointers, but use proc
pointers within the DSL in the syndicate module.
This commit is contained in:
Emery Hemingway 2022-03-12 10:08:22 -06:00
parent 231928f243
commit aca382e178
4 changed files with 80 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/syndicate.nim
View File

@ -4,9 +4,25 @@
import std/macros import std/macros
import preserves import preserves
import syndicate/[actors, dataspaces, durings, patterns] import syndicate/[actors, dataspaces, durings, patterns]
export dataspaces, patterns export dataspaces, patterns, Handle
from syndicate/protocols/protocol import Handle type
PublishProc = proc (turn: var Turn; v: Assertion; h: Handle) {.closure.}
RetractProc = proc (turn: var Turn; h: Handle) {.closure.}
MessageProc = proc (turn: var Turn; v: Assertion) {.closure.}
ClosureEntity = ref object of Entity
publishImpl: PublishProc
retractImpl: RetractProc
messageImpl: MessageProc
method publish(e: ClosureEntity; turn: var Turn; v: Assertion; h: Handle) =
if not e.publishImpl.isNil: e.publishImpl(turn, v, h)
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; v: Assertion) =
if not e.messageImpl.isNil: e.messageImpl(turn, v)
proc wrapPublishHandler(handler: NimNode): NimNode = proc wrapPublishHandler(handler: NimNode): NimNode =
handler.expectKind nnkDo handler.expectKind nnkDo
@ -37,7 +53,7 @@ proc wrapPublishHandler(handler: NimNode): NimNode =
handleSym = ident"handle" handleSym = ident"handle"
handlerSym = genSym(nskProc, "publish") handlerSym = genSym(nskProc, "publish")
quote do: quote do:
proc `handlerSym`(entity: Entity; `turnSym`: var Turn; bindings: Assertion; `handleSym`: Handle) = proc `handlerSym`(`turnSym`: var Turn; bindings: Assertion; `handleSym`: Handle) =
`varSectionOuter` `varSectionOuter`
if fromPreserve(`valuesSym`, bindings): if fromPreserve(`valuesSym`, bindings):
`publishBody` `publishBody`
@ -70,7 +86,7 @@ proc wrapMessageHandler(handler: NimNode): NimNode =
turnSym = ident"turn" turnSym = ident"turn"
handlerSym = genSym(nskProc, "message") handlerSym = genSym(nskProc, "message")
quote do: quote do:
proc `handlerSym`(_: Entity; `turnSym`: var Turn; bindings: Assertion) = proc `handlerSym`(`turnSym`: var Turn; bindings: Assertion) =
`varSectionOuter` `varSectionOuter`
if fromPreserve(`valuesSym`, bindings): if fromPreserve(`valuesSym`, bindings):
`body` `body`
@ -81,7 +97,7 @@ macro onPublish*(turn: Turn; ds: Ref; pattern: Pattern; doHandler: untyped) =
handlerSym = handlerProc[0] handlerSym = handlerProc[0]
result = quote do: result = quote do:
`handlerProc` `handlerProc`
discard observe(`turn`, `ds`, `pattern`, newEntity(publish = `handlerSym`)) discard observe(`turn`, `ds`, `pattern`, ClosureEntity(publishImpl: `handlerSym`))
macro onMessage*(turn: Turn; ds: Ref; pattern: Pattern; doHandler: untyped) = macro onMessage*(turn: Turn; ds: Ref; pattern: Pattern; doHandler: untyped) =
let let
@ -89,5 +105,4 @@ macro onMessage*(turn: Turn; ds: Ref; pattern: Pattern; doHandler: untyped) =
handlerSym = handlerProc[0] handlerSym = handlerProc[0]
result = quote do: result = quote do:
`handlerProc` `handlerProc`
discard observe(`turn`, `ds`, `pattern`, newEntity(message = `handlerSym`)) discard observe(`turn`, `ds`, `pattern`, ClosureEntity(messageImpl: `handlerSym`))

122
src/syndicate/actors.nim
View File

@ -32,17 +32,8 @@ type
Caveat = sturdy.Caveat[Ref] Caveat = sturdy.Caveat[Ref]
Rewrite = sturdy.Rewrite[Ref] Rewrite = sturdy.Rewrite[Ref]
PublishProc* = proc (e: Entity; turn: var Turn; v: Assertion; h: Handle) {.gcsafe.}
RetractProc* = proc (e: Entity; turn: var Turn; h: Handle) {.gcsafe.}
MessageProc* = proc (e: Entity; turn: var Turn; v: Assertion) {.gcsafe.}
SyncProc* = proc (e: Entity; turn: var Turn; peer: Ref) {.gcsafe.}
Entity* = ref object of RootObj Entity* = ref object of RootObj
oid*: Oid # oid is how Entities are identified over the wire oid*: Oid # oid is how Entities are identified over the wire
publishImpl*: PublishProc
retractImpl*: RetractProc
messageImpl*: MessageProc
syncImpl*: SyncProc
Ref* {.unpreservable.} = ref object # TODO: rename Ref* {.unpreservable.} = ref object # TODO: rename
relay*: Facet relay*: Facet
@ -55,7 +46,7 @@ type
established: bool established: bool
OutboundTable = Table[Handle, OutboundAssertion] OutboundTable = Table[Handle, OutboundAssertion]
Actor = ref object Actor* = ref object
future: Future[void] future: Future[void]
name: string name: string
id: ActorId id: ActorId
@ -71,7 +62,7 @@ type
Turn* = object # an object that should remain on the stack Turn* = object # an object that should remain on the stack
id: TurnId id: TurnId
activeFacet*: Facet facet*: Facet
queues: Queues # a ref object that can outlive Turn queues: Queues # a ref object that can outlive Turn
ParentFacet = Option[Facet] ParentFacet = Option[Facet]
@ -93,36 +84,10 @@ using
turn: var Turn turn: var Turn
action: TurnAction action: TurnAction
proc setProcs*( method publish*(e: Entity; turn: var Turn; v: Assertion; h: Handle) {.base.} = discard
result: Entity, method retract*(e: Entity; turn: var Turn; h: Handle) {.base.} = discard
publish: PublishProc = nil, method message*(e: Entity; turn: var Turn; v: Assertion) {.base.} = discard
retract: RetractProc = nil, method sync*(e: Entity; turn: var Turn; peer: Ref) {.base.} = discard
message: MessageProc = nil,
sync: SyncProc = nil) {.inline.} =
result.publishImpl = publish
result.retractImpl = retract
result.messageImpl = message
result.syncImpl = sync
proc newEntity*(
publish: PublishProc = nil,
retract: RetractProc = nil,
message: MessageProc = nil,
sync: SyncProc = nil): Entity =
new result
result.setProcs(publish, retract, message, sync)
proc publish*(e: Entity; turn: var Turn; v: Assertion; h: Handle) =
if not e.publishImpl.isNil: e.publishImpl(e, turn, v, h)
proc retract*(e: Entity; turn: var Turn; h: Handle) =
if not e.retractImpl.isNil: e.retractImpl(e, turn, h)
proc message*(e: Entity; turn: var Turn; v: Assertion) =
if not e.messageImpl.isNil: e.messageImpl(e, turn, v)
proc sync*(e: Entity; turn: var Turn; peer: Ref) =
if not e.syncImpl.isNil: e.syncImpl(e, turn, peer)
proc labels(f: Facet): string = proc labels(f: Facet): string =
proc catLabels(f: Facet; labels: var string) = proc catLabels(f: Facet; labels: var string) =
@ -153,11 +118,7 @@ proc attenuate(r: Ref; a: Attenuation): Ref =
proc hash*(facet): Hash = proc hash*(facet): Hash =
facet.id.hash facet.id.hash
proc hash(r: Ref): Hash = proc hash*(r: Ref): Hash = !$(r.relay.hash !& r.target.unsafeAddr.hash)
!$(r.relay.hash !&
r.target.unsafeAddr.hash) # !&
# r.attenuation.toPreserve.hash)
# TODO: really convert for each hash?
proc nextHandle(facet: Facet): Handle = proc nextHandle(facet: Facet): Handle =
inc facet.actor.handleAllocator inc facet.actor.handleAllocator
@ -284,13 +245,13 @@ proc publish(turn: var Turn; r: Ref; v: Assertion; h: Handle) =
if not a.isFalse: if not a.isFalse:
let e = OutboundAssertion( let e = OutboundAssertion(
handle: h, peer: r, established: false) handle: h, peer: r, established: false)
turn.activeFacet.outbound[h] = e turn.facet.outbound[h] = e
enqueue(turn, r.relay) do (turn: var Turn): enqueue(turn, r.relay) do (turn: var Turn):
e.established = true e.established = true
publish(r.target, turn, a, e.handle) publish(r.target, turn, a, e.handle)
proc publish*(turn: var Turn; r: Ref; a: Assertion): Handle = proc publish*(turn: var Turn; r: Ref; a: Assertion): Handle =
result = turn.activeFacet.nextHandle() result = turn.facet.nextHandle()
publish(turn, r, a, result) publish(turn, r, a, result)
proc publish*[T](turn: var Turn; r: Ref; a: T): Handle = proc publish*[T](turn: var Turn; r: Ref; a: T): Handle =
@ -304,7 +265,7 @@ proc retract(turn: var Turn; e: OutboundAssertion) =
proc retract*(turn: var Turn; h: Handle) = proc retract*(turn: var Turn; h: Handle) =
var e: OutboundAssertion var e: OutboundAssertion
if turn.activeFacet.outbound.pop(h, e): if turn.facet.outbound.pop(h, e):
turn.retract(e) turn.retract(e)
proc message*(turn: var Turn; r: Ref; v: Assertion) = proc message*(turn: var Turn; r: Ref; v: Assertion) =
@ -345,13 +306,12 @@ proc newFacet(actor; parent: ParentFacet): Facet =
var initialAssertions: OutboundTable var initialAssertions: OutboundTable
newFacet(actor, parent, initialAssertions) newFacet(actor, parent, initialAssertions)
proc onStop(facet; action) =
facet.shutdownActions.add action
proc isInert(facet): bool = proc isInert(facet): bool =
facet.inertCheckPreventers == 0 and facet.children.len == 0 and facet.outbound.len == 0 result = facet.children.len == 0 and
(facet.outbound.len == 0 or facet.parent.isNone) and
facet.inertCheckPreventers == 0
proc preventInertCheck*(facet): (proc() {.gcsafe.}) = proc preventInertCheck*(facet): (proc() {.gcsafe.}) {.discardable.} =
var armed = true var armed = true
inc facet.inertCheckPreventers inc facet.inertCheckPreventers
proc disarm() = proc disarm() =
@ -363,7 +323,7 @@ proc preventInertCheck*(facet): (proc() {.gcsafe.}) =
proc inFacet(turn: var Turn; facet; act: TurnAction) = proc inFacet(turn: var Turn; facet; act: TurnAction) =
## Call an action with a facet using a temporary `Turn` ## Call an action with a facet using a temporary `Turn`
## that shares the `Queues` of the calling `Turn`. ## that shares the `Queues` of the calling `Turn`.
var t = Turn(activeFacet: facet, queues: turn.queues) var t = Turn(facet: facet, queues: turn.queues)
act(t) act(t)
proc terminate(actor; turn; reason: ref Exception) {.gcsafe.} proc terminate(actor; turn; reason: ref Exception) {.gcsafe.}
@ -372,8 +332,10 @@ proc terminate(facet; turn: var Turn; orderly: bool) {.gcsafe.} =
if facet.isAlive: if facet.isAlive:
facet.isAlive = false facet.isAlive = false
let parent = facet.parent let parent = facet.parent
if parent.isSome:
parent.get.children.excl facet
block: block:
var turn = Turn(activeFacet: facet, queues: turn.queues) var turn = Turn(facet: facet, queues: turn.queues)
for child in facet.children: for child in facet.children:
child.terminate(turn, orderly) child.terminate(turn, orderly)
if orderly: if orderly:
@ -383,24 +345,22 @@ proc terminate(facet; turn: var Turn; orderly: bool) {.gcsafe.} =
if orderly: if orderly:
if parent.isSome: if parent.isSome:
if parent.get.isInert: if parent.get.isInert:
run(parent.get) do (turn: var Turn): parent.get.terminate(turn, true)
parent.get.terminate(turn, true) # TODO: is this the right turn?
else: else:
run(facet.actor.root) do (turn: var Turn): terminate(facet.actor, turn, nil)
terminate(facet.actor, turn, nil) # TODO: is this the right turn?
proc stopIfInertAfter(action: TurnAction): TurnAction = proc stopIfInertAfter(action: TurnAction): TurnAction =
proc wrapper(turn: var Turn) = proc wrapper(turn: var Turn) =
action(turn) action(turn)
enqueue(turn, turn.activeFacet) do (turn: var Turn): enqueue(turn, turn.facet) do (turn: var Turn):
if (turn.activeFacet.parent.isSome and if (turn.facet.parent.isSome and
(not turn.activeFacet.parent.get.isAlive)) or (not turn.facet.parent.get.isAlive)) or
turn.activeFacet.isInert: turn.facet.isInert:
stop(turn) stop(turn)
wrapper wrapper
proc facet*(turn: var Turn; bootProc: TurnAction): Facet = proc facet*(turn: var Turn; bootProc: TurnAction): Facet =
result =newFacet(turn.activeFacet.actor, some turn.activeFacet) result =newFacet(turn.facet.actor, some turn.facet)
inFacet(turn, result, stopIfInertAfter(bootProc)) inFacet(turn, result, stopIfInertAfter(bootProc))
proc newActor(name: string; bootProc: TurnAction; initialAssertions: OutboundTable): Actor = proc newActor(name: string; bootProc: TurnAction; initialAssertions: OutboundTable): Actor =
@ -414,24 +374,22 @@ proc newActor(name: string; bootProc: TurnAction; initialAssertions: OutboundTab
result.future = newFuture[void]($result) result.future = newFuture[void]($result)
run( run(
newFacet(result, some result.root, initialAssertions), newFacet(result, some result.root, initialAssertions),
# stopIfInertAfter(bootProc) TODO stopIfInertAfter(bootProc))
bootProc)
proc newActor*(name: string; bootProc: TurnAction): Actor = proc bootActor*(name: string; bootProc: TurnAction): Actor =
var initialAssertions: OutboundTable var initialAssertions: OutboundTable
newActor(name, bootProc, initialAssertions) newActor(name, bootProc, initialAssertions)
proc spawn*(name: string; turn: var Turn; bootProc: TurnAction; initialAssertions = initHashSet[Handle]()) = proc spawn*(name: string; turn: var Turn; bootProc: TurnAction; initialAssertions = initHashSet[Handle]()) =
enqueue(turn, turn.activeFacet) do (turn: var Turn): enqueue(turn, turn.facet) do (turn: var Turn):
var newOutBound: Table[Handle, OutboundAssertion] var newOutBound: Table[Handle, OutboundAssertion]
for key in initialAssertions: for key in initialAssertions:
discard turn.activeFacet.outbound.pop(key, newOutbound[key]) discard turn.facet.outbound.pop(key, newOutbound[key])
callSoon: callSoon:
discard newActor(name, bootProc, newOutBound) discard newActor(name, bootProc, newOutBound)
proc newInertRef*(): Ref = proc newInertRef*(): Ref =
# TODO: really create a new actor? Novy does this only once. let a = bootActor("inert") do (turn: var Turn): turn.stop()
let a = newActor("") do (turn: var Turn): turn.stop()
Ref(relay: a.root) Ref(relay: a.root)
proc atExit*(actor; action) = actor.exitHooks.add action proc atExit*(actor; action) = actor.exitHooks.add action
@ -455,7 +413,7 @@ proc terminate(facet; e: ref Exception) =
facet.actor.terminate(turn, e) facet.actor.terminate(turn, e)
proc asyncCheck*(turn; fut: FutureBase) = proc asyncCheck*(turn; fut: FutureBase) =
let facet = turn.activeFacet let facet = turn.facet
fut.addCallback do (): fut.addCallback do ():
if fut.failed: terminate(facet, fut.error) if fut.failed: terminate(facet, fut.error)
@ -477,7 +435,7 @@ proc run*(facet; action: TurnAction; zombieTurn = false) =
# TODO: not Nim idiom # TODO: not Nim idiom
tryFacet(facet): tryFacet(facet):
var turn = Turn( var turn = Turn(
activeFacet: facet, facet: facet,
queues: newTable[Facet, seq[TurnAction]]()) queues: newTable[Facet, seq[TurnAction]]())
action(turn) action(turn)
run(turn.queues) run(turn.queues)
@ -487,29 +445,25 @@ proc stop*(turn: var Turn, facet: Facet) =
facet.terminate(turn, true) facet.terminate(turn, true)
proc stop*(turn: var Turn) = proc stop*(turn: var Turn) =
stop(turn, turn.activeFacet) stop(turn, turn.facet)
proc stopActor*(turn: var Turn) = proc stopActor*(turn: var Turn) =
let actor = turn.activeFacet.actor let actor = turn.facet.actor
enqueue(turn, turn.activeFacet.actor.root) do (turn: var Turn): enqueue(turn, turn.facet.actor.root) do (turn: var Turn):
terminate(actor, turn, nil) terminate(actor, turn, nil)
proc freshen*(turn: var Turn, act: TurnAction) = proc freshen*(turn: var Turn, act: TurnAction) =
assert(turn.queues.len == 0, "Attempt to freshen a non-stale Turn") assert(turn.queues.len == 0, "Attempt to freshen a non-stale Turn")
run(turn.activeFacet, act) run(turn.facet, act)
proc newRef*(relay: Facet; e: Entity): Ref = proc newRef*(relay: Facet; e: Entity): Ref =
Ref(relay: relay, target: e) Ref(relay: relay, target: e)
proc newRef*(turn; e: Entity): Ref = proc newRef*(turn; e: Entity): Ref =
Ref(relay: turn.activeFacet, target: e) Ref(relay: turn.facet, target: e)
proc sync*(turn, refer: Ref, cb: proc(t: Turn) {.gcsafe.}) = proc sync*(turn, refer: Ref, cb: proc(t: Turn) {.gcsafe.}) =
discard # TODO discard # TODO
proc log*(f: Facet, args: varargs[string, `$`]) =
echo f, args
proc runActor*(name: string; bootProc: TurnAction): Future[void] = proc runActor*(name: string; bootProc: TurnAction): Future[void] =
let actor = newActor(name, bootProc) bootActor(name, bootProc).future
result = actor.future

12
src/syndicate/durings.nim
View File

@ -1,7 +1,7 @@
# SPDX-FileCopyrightText: ☭ 2021 Emery Hemingway # SPDX-FileCopyrightText: ☭ 2021 Emery Hemingway
# SPDX-License-Identifier: Unlicense # SPDX-License-Identifier: Unlicense
import std/[hashes, macros, tables] import std/[hashes, tables]
import preserves import preserves
import ./actors, ./patterns, ./protocols/dataspace import ./actors, ./patterns, ./protocols/dataspace
@ -23,8 +23,7 @@ type
cb: DuringProc cb: DuringProc
assertionMap: Table[Handle, DuringAction] assertionMap: Table[Handle, DuringAction]
proc duringPublish(e: Entity; turn: var Turn; a: Assertion; h: Handle) = method publish(de: DuringEntity; turn: var Turn; a: Assertion; h: Handle) =
var de = DuringEntity(e)
let action = de.cb(turn, a) let action = de.cb(turn, a)
# assert(not action.isNil "should have put in a no-op action") # assert(not action.isNil "should have put in a no-op action")
let g = de.assertionMap.getOrDefault h let g = de.assertionMap.getOrDefault h
@ -37,8 +36,7 @@ proc duringPublish(e: Entity; turn: var Turn; a: Assertion; h: Handle) =
of act: of act:
raiseAssert("during: duplicate handle in publish: " & $h) raiseAssert("during: duplicate handle in publish: " & $h)
proc duringRetract(e: Entity; turn: var Turn; h: Handle) = method retract(de: DuringEntity; turn: var Turn; h: Handle) =
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:
@ -49,9 +47,7 @@ proc duringRetract(e: Entity; turn: var Turn; h: Handle) =
de.assertionMap.del h de.assertionMap.del h
g.action(turn) g.action(turn)
proc during*(cb: DuringProc): DuringEntity = proc during*(cb: DuringProc): DuringEntity = DuringEntity(cb: cb)
result = DuringEntity(cb: cb)
result.setProcs(publish = duringPublish, retract = duringRetract)
proc observe*(turn: var Turn; ds: Ref; pat: Pattern; e: Entity): Handle = proc observe*(turn: var Turn; ds: Ref; pat: Pattern; e: Entity): Handle =
publish(turn, ds, Observe(pattern: pat, observer: embed newRef(turn, e))) publish(turn, ds, Observe(pattern: pat, observer: embed newRef(turn, e)))

48
src/syndicate/relays.nim
View File

@ -51,33 +51,24 @@ type
proc releaseRefOut(r: Relay; e: WireSymbol) = proc releaseRefOut(r: Relay; e: WireSymbol) =
r.exported.drop e r.exported.drop e
proc syncPeerPublish(e: Entity; t: var Turn; v: Assertion; h: Handle) = method publish(spe: SyncPeerEntity; t: var Turn; v: Assertion; h: Handle) =
var se = SyncPeerEntity(e) spe.handleMap[h] = publish(t, spe.peer, v)
se.handleMap[h] = publish(t, se.peer, v)
proc syncPeerRetract(e: Entity; t: var Turn; h: Handle) = method retract(se: SyncPeerEntity; t: var Turn; h: Handle) =
var se = SyncPeerEntity(e)
var other: Handle var other: Handle
if se.handleMap.pop(h, other): if se.handleMap.pop(h, other):
retract(t, other) retract(t, other)
proc syncPeerMessage(e: Entity; t: var Turn; v: Assertion) = method message(se: SyncPeerEntity; t: var Turn; v: Assertion) =
var se = SyncPeerEntity(e)
if not se.e.isNil: if not se.e.isNil:
se.relay.releaseRefOut(se.e) se.relay.releaseRefOut(se.e)
message(t, se.peer, v) message(t, se.peer, v)
proc syncPeerSync(e: Entity; t: var Turn; peer: Ref) = method sync(se: SyncPeerEntity; t: var Turn; peer: Ref) =
var se = SyncPeerEntity(e)
sync(t, se.peer, peer) sync(t, se.peer, peer)
proc newSyncPeerEntity(r: Relay; p: Ref): SyncPeerEntity = proc newSyncPeerEntity(r: Relay; p: Ref): SyncPeerEntity =
result = SyncPeerEntity(relay: r, peer: p) SyncPeerEntity(relay: r, peer: p)
result.setProcs(
syncPeerPublish,
syncPeerRetract,
syncPeerMessage,
syncPeerSync)
proc rewriteRefOut(relay: Relay; `ref`: Ref; transient: bool; exported: var seq[WireSymbol]): WireRef = proc rewriteRefOut(relay: Relay; `ref`: Ref; transient: bool; exported: var seq[WireSymbol]): WireRef =
if `ref`.target of RelayEntity and `ref`.target.RelayEntity.relay == relay: if `ref`.target of RelayEntity and `ref`.target.RelayEntity.relay == relay:
@ -132,33 +123,28 @@ proc send(r: Relay; rOid: protocol.Oid; m: Event) =
proc send(re: RelayEntity; ev: Event) = proc send(re: RelayEntity; ev: Event) =
send(re.relay, protocol.Oid re.oid, ev) send(re.relay, protocol.Oid re.oid, ev)
proc relayPublish(e: Entity; t: var Turn; v: Assertion; h: Handle) = method publish(re: RelayEntity; t: var Turn; v: Assertion; h: Handle) =
var
re = RelayEntity(e)
re.send Event( re.send Event(
orKind: EventKind.Assert, orKind: EventKind.Assert,
`assert`: protocol.Assert[WireRef]( `assert`: protocol.Assert[WireRef](
assertion: re.relay.register(v, h), assertion: re.relay.register(v, h),
handle: h)) handle: h))
proc relayRetract(e: Entity; t: var Turn; h: Handle) = method retract(re: RelayEntity; t: var Turn; h: Handle) =
var re = RelayEntity(e)
re.relay.deregister h re.relay.deregister h
re.send Event( re.send Event(
orKind: EventKind.Retract, orKind: EventKind.Retract,
retract: Retract(handle: h)) retract: Retract(handle: h))
proc relayMessage(e: Entity; turn: var Turn; msg: Assertion) = method message(re: RelayEntity; turn: var Turn; msg: Assertion) =
var var
re = RelayEntity(e)
ev = Event(orKind: EventKind.Message) ev = Event(orKind: EventKind.Message)
(body, _) = rewriteOut(re.relay, msg, true) (body, _) = rewriteOut(re.relay, msg, true)
ev.message = Message[WireRef](body: body) ev.message = Message[WireRef](body: body)
re.send ev re.send ev
proc relaySync(e: Entity; turn: var Turn; peer: Ref) = method sync(re: RelayEntity; turn: var Turn; peer: Ref) =
var var
re = RelayEntity(e)
peerEntity = newSyncPeerEntity(re.relay, peer) peerEntity = newSyncPeerEntity(re.relay, peer)
exported: seq[WireSymbol] exported: seq[WireSymbol]
discard rewriteRefOut(re.relay, turn.newRef(peerEntity), false, exported) discard rewriteRefOut(re.relay, turn.newRef(peerEntity), false, exported)
@ -169,8 +155,7 @@ proc relaySync(e: Entity; turn: var Turn; peer: Ref) =
sync: Sync[WireRef](peer: embed toPreserve(false, WireRef))) # TODO: send the WireRef? sync: Sync[WireRef](peer: embed toPreserve(false, WireRef))) # TODO: send the WireRef?
proc newRelayEntity(label: string; r: Relay; o: Oid): RelayEntity = proc newRelayEntity(label: string; r: Relay; o: Oid): RelayEntity =
result = RelayEntity(label: label, relay: r, oid: o) RelayEntity(label: label, relay: r, oid: o)
result.setProcs(relayPublish, relayRetract, relayMessage, relaySync)
using using
relay: Relay relay: Relay
@ -214,9 +199,8 @@ proc close(r: Relay) = discard
proc dispatch(relay: Relay; turn: var Turn; `ref`: Ref; event: Event) = proc dispatch(relay: Relay; turn: var Turn; `ref`: Ref; event: Event) =
case event.orKind case event.orKind
of EventKind.Assert: of EventKind.Assert:
let (a, imported) = rewriteIn(relay, turn.activeFacet, event.assert.assertion) let (a, imported) = rewriteIn(relay, turn.facet, event.assert.assertion)
relay.inboundAssertions[event.assert.handle] = relay.inboundAssertions[event.assert.handle] = (publish(turn, `ref`, a), imported,)
(turn.publish(`ref`, a), imported,)
of EventKind.Retract: of EventKind.Retract:
let remoteHandle = event.retract.handle let remoteHandle = event.retract.handle
@ -226,7 +210,7 @@ proc dispatch(relay: Relay; turn: var Turn; `ref`: Ref; event: Event) =
turn.retract(outbound.localHandle) turn.retract(outbound.localHandle)
of EventKind.Message: of EventKind.Message:
let (a, imported) = rewriteIn(relay, turn.activeFacet, event.message.body) let (a, imported) = rewriteIn(relay, turn.facet, event.message.body)
assert imported.len == 0, "Cannot receive transient reference" assert imported.len == 0, "Cannot receive transient reference"
turn.message(`ref`, a) turn.message(`ref`, a)
@ -274,7 +258,7 @@ type
proc newRelay(turn: var Turn; opts: RelayOptions): Relay = proc newRelay(turn: var Turn; opts: RelayOptions): Relay =
result = Relay( result = Relay(
facet: turn.activeFacet, facet: turn.facet,
packetWriter: opts.packetWriter, packetWriter: opts.packetWriter,
untrusted: opts.untrusted) untrusted: opts.untrusted)
discard result.facet.preventInertCheck() discard result.facet.preventInertCheck()
@ -292,7 +276,7 @@ proc spawnRelay(name: string; turn: var Turn; opts: RelayActorOptions): Future[R
var wr = WireRef( var wr = WireRef(
orKind: WireRefKind.mine, orKind: WireRefKind.mine,
mine: WireRefMine(oid: opts.initialOid.get)) mine: WireRefMine(oid: opts.initialOid.get))
fut.complete rewriteRefIn(relay, turn.activeFacet, wr, imported) fut.complete rewriteRefIn(relay, turn.facet, wr, imported)
else: else:
fut.complete(nil) fut.complete(nil)
opts.nextLocalOid.map do (oid: Oid): opts.nextLocalOid.map do (oid: Oid):