Proper dataspaces
This commit is contained in:
parent
d8f6d82956
commit
1912574ed8
|
@ -3,8 +3,8 @@
|
||||||
|
|
||||||
import std/macros
|
import std/macros
|
||||||
import preserves
|
import preserves
|
||||||
import syndicate/[actors, dataspaces, patterns]
|
import syndicate/[actors, dataspaces, durings, patterns]
|
||||||
export patterns
|
export dataspaces, patterns
|
||||||
|
|
||||||
from syndicate/protocols/protocol import Handle
|
from syndicate/protocols/protocol import Handle
|
||||||
|
|
||||||
|
|
|
@ -36,3 +36,6 @@ proc change*[T](bag: var Bag[T]; key: T; delta: int; clamp = false): ChangeDescr
|
||||||
result = change(bag.mGetOrPut(key, 0), delta, clamp)
|
result = change(bag.mGetOrPut(key, 0), delta, clamp)
|
||||||
if result in {cdAbsentToAbsent, cdPresentToAbsent}:
|
if result in {cdAbsentToAbsent, cdPresentToAbsent}:
|
||||||
bag.del(key)
|
bag.del(key)
|
||||||
|
|
||||||
|
iterator items*[T](bag: Bag[T]): (int, T) =
|
||||||
|
for k, v in bag: yield(v, k)
|
||||||
|
|
|
@ -1,104 +1,46 @@
|
||||||
# SPDX-FileCopyrightText: ☭ 2021 Emery Hemingway
|
# SPDX-FileCopyrightText: ☭ 2022 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, ./bags, ./patterns, ./protocols/dataspace
|
import ./actors, ./protocols/dataspace, ./skeletons
|
||||||
|
|
||||||
from ./protocols/protocol import Handle
|
from ./protocols/protocol import Handle
|
||||||
|
|
||||||
|
template trace(args: varargs[untyped]): untyped = stderr.writeLine(args)
|
||||||
|
|
||||||
type
|
type
|
||||||
Observe = dataspace.Observe[Ref]
|
Observe = dataspace.Observe[Ref]
|
||||||
Turn = actors.Turn
|
Turn = actors.Turn
|
||||||
|
|
||||||
#[
|
Dataspace {.final.} = ref object of Entity
|
||||||
DataspaceEntity = ref object of Entity
|
index: Index
|
||||||
assertions: Bag[Assertion]
|
handleMap: Table[Handle, Assertion]
|
||||||
subscriptions: Table[Assertion, TableRef[Ref, TableRef[Assertion, Handle]]]
|
|
||||||
handleMap: Table[Handle, Assertion] # breaks toPreserve(Observe[Ref]())
|
|
||||||
|
|
||||||
method publish(ds: DataspaceEntity; turn: var Turn; rec: Assertion; h: Handle) =
|
method publish(ds: Dataspace; turn: var Turn; v: Assertion; h: Handle) =
|
||||||
if rec.isRecord:
|
if add(ds.index, turn, v):
|
||||||
ds.handleMap[h] = rec
|
|
||||||
if ds.assertions.change(rec, +1) == cdAbsentToPresent:
|
|
||||||
var obs: Observe
|
var obs: Observe
|
||||||
if fromPreserve(obs, rec):
|
if obs.fromPreserve v:
|
||||||
var seen = newTable[Assertion, Handle]()
|
ds.index.add(turn, obs.pattern, unembed obs.observer)
|
||||||
for prev, count in ds.assertions.pairs:
|
ds.handleMap[h] = v
|
||||||
if prev == rec.label:
|
|
||||||
seen[prev] = publish(turn, obs.observer.unembed, prev)
|
|
||||||
var patternSubs = ds.subscriptions.getOrDefault(rec.label)
|
|
||||||
if patternSubs.isNil:
|
|
||||||
patternSubs = newTable[Ref, TableRef[Value, Handle]]()
|
|
||||||
ds.subscriptions[rec.label] = patternSubs
|
|
||||||
patternSubs[obs.observer.unembed] = move seen
|
|
||||||
for peer, seen in ds.subscriptions[rec.label]:
|
|
||||||
if rec notin seen:
|
|
||||||
seen[rec] = publish(turn, peer, rec)
|
|
||||||
|
|
||||||
method retract(ds: DataspaceEntity; turn: var Turn; upstreamHandle: Handle) =
|
method retract(ds: Dataspace; turn: var Turn; h: Handle) =
|
||||||
let rec = ds.handleMap.getOrDefault(upstreamHandle)
|
try:
|
||||||
if rec.isRecord:
|
let v = ds.handleMap[h]
|
||||||
ds.handleMap.del upstreamHandle
|
if remove(ds.index, turn, v):
|
||||||
if ds.assertions.change(rec, -1) == cdPresentToAbsent:
|
ds.handleMap.del h
|
||||||
for peer, seen in ds.subscriptions[rec.label]:
|
var obs: Observe
|
||||||
var h: Handle
|
if obs.fromPreserve v:
|
||||||
if pop(seen, rec, h): retract(turn, h)
|
ds.index.remove(turn, obs.pattern, unembed obs.observer)
|
||||||
preserveTo(rec, Observe).map do (obs: Observe):
|
except KeyError: discard
|
||||||
let peerMap = ds.subscriptions[rec.label]
|
|
||||||
peerMap.del(obs.observer.unembed)
|
|
||||||
if peerMap.len == 0:
|
|
||||||
ds.subscriptions.del(rec.label)
|
|
||||||
|
|
||||||
method message(ds: DataspaceEntity; turn: var Turn; msg: Assertion) =
|
method message(ds: Dataspace; turn: var Turn; v: Assertion) =
|
||||||
if msg.isRecord:
|
ds.index.deliverMessage(turn, v)
|
||||||
for peer, seen in ds.subscriptions[msg.label].pairs:
|
|
||||||
message(turn, peer, msg)
|
|
||||||
]#
|
|
||||||
|
|
||||||
type DuringProc* = proc (turn: var Turn; a: Assertion): TurnAction {.gcsafe.}
|
type BootProc = proc (ds: Ref; turn: var Turn) {.gcsafe.}
|
||||||
|
|
||||||
type
|
proc bootDataspace*(name: string; bootProc: BootProc): Actor {.discardable.} =
|
||||||
DuringActionKind = enum null, dead, act
|
bootActor(name) do (turn: var Turn):
|
||||||
DuringAction = object
|
discard turn.facet.preventInertCheck()
|
||||||
case kind: DuringActionKind
|
let ds = newRef(turn, Dataspace(index: initIndex()))
|
||||||
of null, dead: discard
|
bootProc(ds, turn)
|
||||||
of act:
|
|
||||||
action: TurnAction
|
|
||||||
|
|
||||||
type DuringEntity = ref object of Entity
|
|
||||||
cb: DuringProc
|
|
||||||
assertionMap: Table[Handle, DuringAction]
|
|
||||||
|
|
||||||
proc duringPublish(e: Entity; turn: var Turn; a: Assertion; h: Handle) =
|
|
||||||
var de = DuringEntity(e)
|
|
||||||
let action = de.cb(turn, a)
|
|
||||||
# assert(not action.isNil "should have put in a no-op action")
|
|
||||||
let g = de.assertionMap.getOrDefault h
|
|
||||||
case g.kind
|
|
||||||
of null:
|
|
||||||
de.assertionMap[h] = DuringAction(kind: act, action: action)
|
|
||||||
of dead:
|
|
||||||
de.assertionMap.del h
|
|
||||||
freshen(turn, action)
|
|
||||||
of act:
|
|
||||||
raiseAssert("during: duplicate handle in publish: " & $h)
|
|
||||||
|
|
||||||
proc duringRetract(e: Entity; turn: var Turn; h: Handle) =
|
|
||||||
var de = DuringEntity(e)
|
|
||||||
let g = de.assertionMap.getOrDefault h
|
|
||||||
case g.kind
|
|
||||||
of null:
|
|
||||||
de.assertionMap[h] = DuringAction(kind: dead)
|
|
||||||
of dead:
|
|
||||||
raiseAssert("during: duplicate handle in retract: " & $h)
|
|
||||||
of act:
|
|
||||||
de.assertionMap.del h
|
|
||||||
g.action(turn)
|
|
||||||
|
|
||||||
proc during*(cb: DuringProc): DuringEntity =
|
|
||||||
result = DuringEntity(cb: cb)
|
|
||||||
result.setProcs(publish = duringPublish, retract = duringRetract)
|
|
||||||
|
|
||||||
proc observe*(turn: var Turn; ds: Ref; pat: Pattern; e: Entity): Handle =
|
|
||||||
publish(turn, ds, Observe(pattern: pat, observer: embed newRef(turn, e)))
|
|
||||||
|
|
|
@ -0,0 +1,57 @@
|
||||||
|
# SPDX-FileCopyrightText: ☭ 2021 Emery Hemingway
|
||||||
|
# SPDX-License-Identifier: Unlicense
|
||||||
|
|
||||||
|
import std/[hashes, macros, tables]
|
||||||
|
import preserves
|
||||||
|
import ./actors, ./patterns, ./protocols/dataspace
|
||||||
|
|
||||||
|
from ./protocols/protocol import Handle
|
||||||
|
|
||||||
|
type
|
||||||
|
Observe = dataspace.Observe[Ref]
|
||||||
|
Turn = actors.Turn
|
||||||
|
|
||||||
|
type
|
||||||
|
DuringProc* = proc (turn: var Turn; a: Assertion): TurnAction {.gcsafe.}
|
||||||
|
DuringActionKind = enum null, dead, act
|
||||||
|
DuringAction = object
|
||||||
|
case kind: DuringActionKind
|
||||||
|
of null, dead: discard
|
||||||
|
of act:
|
||||||
|
action: TurnAction
|
||||||
|
DuringEntity {.final.}= ref object of Entity
|
||||||
|
cb: DuringProc
|
||||||
|
assertionMap: Table[Handle, DuringAction]
|
||||||
|
|
||||||
|
proc duringPublish(e: Entity; turn: var Turn; a: Assertion; h: Handle) =
|
||||||
|
var de = DuringEntity(e)
|
||||||
|
let action = de.cb(turn, a)
|
||||||
|
# assert(not action.isNil "should have put in a no-op action")
|
||||||
|
let g = de.assertionMap.getOrDefault h
|
||||||
|
case g.kind
|
||||||
|
of null:
|
||||||
|
de.assertionMap[h] = DuringAction(kind: act, action: action)
|
||||||
|
of dead:
|
||||||
|
de.assertionMap.del h
|
||||||
|
freshen(turn, action)
|
||||||
|
of act:
|
||||||
|
raiseAssert("during: duplicate handle in publish: " & $h)
|
||||||
|
|
||||||
|
proc duringRetract(e: Entity; turn: var Turn; h: Handle) =
|
||||||
|
var de = DuringEntity(e)
|
||||||
|
let g = de.assertionMap.getOrDefault h
|
||||||
|
case g.kind
|
||||||
|
of null:
|
||||||
|
de.assertionMap[h] = DuringAction(kind: dead)
|
||||||
|
of dead:
|
||||||
|
raiseAssert("during: duplicate handle in retract: " & $h)
|
||||||
|
of act:
|
||||||
|
de.assertionMap.del h
|
||||||
|
g.action(turn)
|
||||||
|
|
||||||
|
proc during*(cb: DuringProc): DuringEntity =
|
||||||
|
result = DuringEntity(cb: cb)
|
||||||
|
result.setProcs(publish = duringPublish, retract = duringRetract)
|
||||||
|
|
||||||
|
proc observe*(turn: var Turn; ds: Ref; pat: Pattern; e: Entity): Handle =
|
||||||
|
publish(turn, ds, Observe(pattern: pat, observer: embed newRef(turn, e)))
|
|
@ -3,7 +3,7 @@
|
||||||
|
|
||||||
import std/[asyncdispatch, options, tables]
|
import std/[asyncdispatch, options, tables]
|
||||||
import preserves, preserves/parse
|
import preserves, preserves/parse
|
||||||
import ./actors, ./dataspaces, ./membranes, ./protocols/[protocol, sturdy]
|
import ./actors, ./durings, ./membranes, ./protocols/[protocol, sturdy]
|
||||||
|
|
||||||
when defined(traceSyndicate):
|
when defined(traceSyndicate):
|
||||||
template trace(args: varargs[untyped]): untyped = echo(args)
|
template trace(args: varargs[untyped]): untyped = echo(args)
|
||||||
|
|
|
@ -1,29 +1,15 @@
|
||||||
# SPDX-FileCopyrightText: ☭ 2021 Emery Hemingway
|
# SPDX-FileCopyrightText: ☭ 2021 Emery Hemingway
|
||||||
# SPDX-License-Identifier: Unlicense
|
# SPDX-License-Identifier: Unlicense
|
||||||
|
|
||||||
import ./assertions, ./bags, ./events
|
import std/[hashes, lists, options, sets, tables]
|
||||||
import preserves, preserves/records
|
import preserves
|
||||||
import lists, options, sets, tables
|
import ./actors, ./bags, ./patterns
|
||||||
|
|
||||||
|
template trace(args: varargs[untyped]) = stderr.writeLine(args)
|
||||||
|
|
||||||
type
|
type
|
||||||
NonEmptySkeleton*[Shape] = object
|
Value = Preserve[Ref]
|
||||||
shape: Shape
|
Path = seq[Value]
|
||||||
members: seq[Skeleton[Shape]]
|
|
||||||
Skeleton*[Shape] = Option[NonEmptySkeleton[Shape]]
|
|
||||||
|
|
||||||
Shape = string
|
|
||||||
|
|
||||||
Value = Preserve
|
|
||||||
|
|
||||||
HandlerCallback* = proc (event: EventKind; bindings: seq[Value]) {.gcsafe.}
|
|
||||||
|
|
||||||
Path = seq[Natural]
|
|
||||||
|
|
||||||
Analysis* = object
|
|
||||||
skeleton: Skeleton[Shape]
|
|
||||||
constPaths: seq[Path]
|
|
||||||
constVals: seq[Value]
|
|
||||||
capturePaths: seq[Path]
|
|
||||||
|
|
||||||
proc projectPath(v: Value; path: Path): Value =
|
proc projectPath(v: Value; path: Path): Value =
|
||||||
result = v
|
result = v
|
||||||
|
@ -33,50 +19,57 @@ proc projectPath(v: Value; path: Path): Value =
|
||||||
proc projectPaths(v: Value; paths: seq[Path]): seq[Value] =
|
proc projectPaths(v: Value; paths: seq[Path]): seq[Value] =
|
||||||
result.setLen(paths.len)
|
result.setLen(paths.len)
|
||||||
for i, path in paths: result[i] = projectPath(v, path)
|
for i, path in paths: result[i] = projectPath(v, path)
|
||||||
|
trace "projected ", v, " by paths ", paths, " to ", result
|
||||||
|
|
||||||
proc analyzeAssertion*(a: Value): Analysis =
|
type Class = distinct string
|
||||||
var path: Path
|
|
||||||
proc walk(analysis: var Analysis; a: Value): Skeleton[Shape] =
|
proc hash(cls: Class): Hash {.borrow.}
|
||||||
if a.preserveTo(Discard).isSome:
|
proc `==`(x, y: Class): bool {.borrow.}
|
||||||
discard
|
|
||||||
elif a.preserveTo(Capture).isSome:
|
proc classOf*(v: Value): Class =
|
||||||
analysis.capturePaths.add(path)
|
if v.isRecord:
|
||||||
result = walk(analysis, a.fields[0])
|
result = Class $v.label & "/" & $v.arity
|
||||||
else:
|
elif v.isSequence:
|
||||||
if a.kind == pkRecord:
|
result = Class $v.len
|
||||||
let class = classOf(a)
|
elif v.isDictionary:
|
||||||
result = some NonEmptySkeleton[Shape](shape: $class)
|
result = Class "{}"
|
||||||
path.add(0)
|
|
||||||
var i: int
|
proc classOf*(p: Pattern): Class =
|
||||||
for field in a.fields:
|
if p.orKind == PatternKind.DCompound:
|
||||||
path[path.high] = i
|
case p.dcompound.orKind
|
||||||
result.get.members.add(walk(analysis, field))
|
of DCompoundKind.rec:
|
||||||
inc(i)
|
result = Class $p.dcompound.rec.label & "/" & $p.dcompound.rec.fields.len
|
||||||
discard path.pop
|
of DCompoundKind.arr:
|
||||||
else:
|
result = Class $p.dcompound.arr.items.len
|
||||||
analysis.constPaths.add(path)
|
of DCompoundKind.dict:
|
||||||
analysis.constVals.add(a)
|
result = Class "{}"
|
||||||
result.skeleton = walk(result, a)
|
|
||||||
|
proc step(value, index: Value): Value =
|
||||||
|
try: result = value[index]
|
||||||
|
except KeyError, ValueError:
|
||||||
|
trace "step failed, ", index, " not in ", value
|
||||||
|
|
||||||
type
|
type
|
||||||
Handler = ref object
|
EventKind = enum addedEvent, removedEvent, messageEvent
|
||||||
cachedCaptures: Bag[seq[Value]]
|
|
||||||
callbacks: HashSet[HandlerCallback]
|
|
||||||
|
|
||||||
AssertionCache = HashSet[Value]
|
AssertionCache = HashSet[Value]
|
||||||
|
|
||||||
|
ObserverGroup = ref object
|
||||||
|
cachedCaptures: Bag[seq[Value]]
|
||||||
|
observers: Table[Ref, TableRef[seq[Value], Handle]]
|
||||||
|
|
||||||
Leaf = ref object
|
Leaf = ref object
|
||||||
cachedAssertions: AssertionCache
|
cachedAssertions: AssertionCache
|
||||||
handlerMap: Table[seq[Path], Handler]
|
observerGroups: Table[seq[Path], ObserverGroup]
|
||||||
|
|
||||||
Continuation = ref object
|
Continuation = ref object
|
||||||
cachedAssertions: AssertionCache
|
cachedAssertions: AssertionCache
|
||||||
leafMap: Table[seq[Path], TableRef[seq[Value], Leaf]] # TODO: not TableRef?
|
leafMap: Table[seq[Path], TableRef[seq[Value], Leaf]]
|
||||||
|
|
||||||
Selector = tuple[popCount: int; index: int]
|
Selector = tuple[popCount: int; index: Value]
|
||||||
|
|
||||||
Node = ref object
|
Node = ref object
|
||||||
edges: Table[Selector, TableRef[string, Node]]
|
edges: Table[Selector, TableRef[Class, Node]]
|
||||||
continuation: Continuation
|
continuation: Continuation
|
||||||
|
|
||||||
using
|
using
|
||||||
|
@ -85,87 +78,96 @@ using
|
||||||
node: Node
|
node: Node
|
||||||
|
|
||||||
proc isEmpty(leaf): bool =
|
proc isEmpty(leaf): bool =
|
||||||
leaf.cachedAssertions.len == 0 and leaf.handlerMap.len == 0
|
leaf.cachedAssertions.len == 0 and leaf.observerGroups.len == 0
|
||||||
|
|
||||||
type
|
type
|
||||||
ContinuationProc = proc (c: Continuation; v: Value) {.gcsafe.}
|
ContinuationProc = proc (c: Continuation; v: Value) {.gcsafe.}
|
||||||
LeafProc = proc (l: Leaf; v: Value) {.gcsafe.}
|
LeafProc = proc (l: Leaf; v: Value) {.gcsafe.}
|
||||||
HandlerProc = proc (h: Handler; vs: seq[Value]) {.gcsafe.}
|
ObserverProc = proc (turn: var Turn; h: ObserverGroup; vs: seq[Value]) {.gcsafe.}
|
||||||
|
|
||||||
proc modify(node; operation: EventKind; outerValue: Value;
|
proc modify(node; turn: var Turn; operation: EventKind; outerValue: Value;
|
||||||
mCont: ContinuationProc; mLeaf: LeafProc; mHandler: HandlerProc) =
|
mCont: ContinuationProc; mLeaf: LeafProc; mObserverGroup: ObserverProc) =
|
||||||
|
trace "modify node with outerValue ", outerValue
|
||||||
|
|
||||||
proc walkContinuation(continuation) {.gcsafe.}
|
proc walkNode(turn: var Turn; node; termStack: SinglyLinkedList[Value]) =
|
||||||
|
mCont(node.continuation, outerValue)
|
||||||
proc walkNode(node; termStack: SinglyLinkedList[seq[Value]]) =
|
if node.continuation.leafMap.len == 0:
|
||||||
# TODO: use a seq for the stack?
|
trace "node.continuation leafMap is empty"
|
||||||
walkContinuation(node.continuation)
|
for (constPaths, constValMap) in node.continuation.leafMap.pairs:
|
||||||
|
trace "got entry in node.continuation.leafMap for ", constPaths
|
||||||
|
let constValues = projectPaths(outerValue, constPaths)
|
||||||
|
var leaf = constValMap.getOrDefault(constValues)
|
||||||
|
if leaf.isNil and operation == addedEvent:
|
||||||
|
new leaf
|
||||||
|
constValMap[constValues] = leaf
|
||||||
|
if not leaf.isNil:
|
||||||
|
mLeaf(leaf, outerValue)
|
||||||
|
for (capturePaths, observerGroup) in leaf.observerGroups.pairs:
|
||||||
|
mObserverGroup(turn, observerGroup, projectPaths(outerValue, capturePaths))
|
||||||
|
if operation == removedEvent and leaf.isEmpty:
|
||||||
|
constValMap.del(constValues)
|
||||||
|
if constValues.len == 0:
|
||||||
|
node.continuation.leafMap.del(constPaths)
|
||||||
for (selector, table) in node.edges.pairs:
|
for (selector, table) in node.edges.pairs:
|
||||||
var nextStack = termStack
|
var nextStack = termStack
|
||||||
for _ in 1..selector.popCount:
|
for _ in 1..selector.popCount:
|
||||||
nextStack.head = nextStack.head.next
|
nextStack.head = nextStack.head.next
|
||||||
let nextValue = nextStack.head.value[selector.index]
|
trace "step ", nextStack.head.value, " with ", selector.index
|
||||||
if nextValue.isRecord:
|
let
|
||||||
let nextClass = classOf(nextValue)
|
nextValue = step(nextStack.head.value, selector.index)
|
||||||
let nextNode = table.getOrDefault($nextClass)
|
nextClass = classOf nextValue
|
||||||
|
if nextClass != Class"":
|
||||||
|
let nextNode = table.getOrDefault(nextClass)
|
||||||
if not nextNode.isNil:
|
if not nextNode.isNil:
|
||||||
nextStack.prepend(nextValue.record)
|
nextStack.prepend(nextValue)
|
||||||
walkNode(nextNode, nextStack)
|
walkNode(turn, nextNode, nextStack)
|
||||||
|
|
||||||
proc walkContinuation(continuation: Continuation) =
|
var stack: SinglyLinkedList[Value]
|
||||||
mCont(continuation, outerValue)
|
stack.prepend(outerValue)
|
||||||
for (constPaths, constValMap) in continuation.leafMap.pairs:
|
walkNode(turn, node, stack)
|
||||||
let constVals = projectPaths(outerValue, constPaths)
|
|
||||||
let leaf = constValMap.getOrDefault(constVals)
|
|
||||||
if leaf.isNil:
|
|
||||||
if operation == addedEvent:
|
|
||||||
constValMap[constVals] = Leaf()
|
|
||||||
else:
|
|
||||||
mLeaf(leaf, outerValue)
|
|
||||||
for (capturePaths, handler) in leaf.handlerMap.pairs:
|
|
||||||
mHandler(handler, projectPaths(outerValue, capturePaths))
|
|
||||||
if operation == removedEvent and leaf.isEmpty:
|
|
||||||
constValMap.del(constVals)
|
|
||||||
if constValMap.len == 0:
|
|
||||||
continuation.leafMap.del(constPaths)
|
|
||||||
var stack: SinglyLinkedList[seq[Value]]
|
|
||||||
stack.prepend(@[outerValue])
|
|
||||||
walkNode(node, stack)
|
|
||||||
|
|
||||||
proc extend[Shape](node; skeleton: Skeleton[Shape]): Continuation =
|
proc extend(node: Node; pat: Pattern): Continuation =
|
||||||
|
trace "extend node with ", pat
|
||||||
var path: Path
|
var path: Path
|
||||||
|
proc walkNode(node: Node; popCount: Natural; stepIndex: Value; pat: Pattern): tuple[popCount: Natural, nextNode: Node] =
|
||||||
proc walkNode(node; popCount, index: int; skeleton: Skeleton[Shape]): tuple[popCount: int, node: Node] =
|
trace "walkNode step ", stepIndex, " of ", pat
|
||||||
assert(not node.isNil)
|
case pat.orKind
|
||||||
if skeleton.isNone:
|
of PatternKind.DDiscard, PatternKind.DLit: result = (popCount, node)
|
||||||
return (popCount, node)
|
of PatternKind.DBind: result = walkNode(node, popCount, stepIndex, pat.dbind.pattern)
|
||||||
else:
|
of PatternKind.DCompound:
|
||||||
let selector: Selector = (popCount, index)
|
let selector = (popCount, stepIndex,)
|
||||||
var
|
var table = node.edges.getOrDefault(selector)
|
||||||
cls = skeleton.get.shape
|
|
||||||
var
|
|
||||||
table = node.edges.getOrDefault(selector)
|
|
||||||
if table.isNil:
|
if table.isNil:
|
||||||
table = newTable[string, Node]()
|
trace "allocate new table for selector ", selector, " for ", pat
|
||||||
|
table = newTable[Class, Node]()
|
||||||
node.edges[selector] = table
|
node.edges[selector] = table
|
||||||
var nextNode = table.getOrDefault(cls)
|
else:
|
||||||
if nextNode.isNil:
|
trace "got a table for ", pat, " with selector ", selector
|
||||||
nextNode = Node(continuation: Continuation())
|
let class = classOf pat
|
||||||
table[cls] = nextNode
|
result.nextNode = table.getOrDefault(class)
|
||||||
|
if result.nextNode.isNil:
|
||||||
|
trace "allocate result.nextNode for ", string class
|
||||||
|
new result.nextNode
|
||||||
|
table[class] = result.nextNode
|
||||||
|
new result.nextNode.continuation
|
||||||
for a in node.continuation.cachedAssertions:
|
for a in node.continuation.cachedAssertions:
|
||||||
if $classOf(projectPath(a, path)) == cls:
|
if class == classOf projectPath(a, path):
|
||||||
nextNode.continuation.cachedAssertions.incl(a)
|
result.nextNode.continuation.cachedAssertions.incl a
|
||||||
block:
|
result.popCount = 0
|
||||||
var popCount, index: int
|
template walkKey(pat: Pattern; stepIndex: Value) =
|
||||||
path.add(index)
|
trace "walkKey ", pat, " with step ", stepIndex
|
||||||
for member in skeleton.get.members:
|
path.add(stepIndex)
|
||||||
(popCount, nextNode) = walkNode(nextNode, result.popCount, index, member)
|
result = walkNode(result.nextNode, popCount, stepIndex, pat)
|
||||||
inc(index)
|
|
||||||
discard path.pop()
|
discard path.pop()
|
||||||
path.add(index)
|
case pat.dcompound.orKind
|
||||||
discard path.pop()
|
of DCompoundKind.rec:
|
||||||
result = (popCount.succ, nextNode)
|
for k, e in pat.dcompound.rec.fields: walkKey(e, k.toPreserve(Ref))
|
||||||
walkNode(node, 0, 0, skeleton).node.continuation
|
of DCompoundKind.arr:
|
||||||
|
for k, e in pat.dcompound.arr.items: walkKey(e, k.toPreserve(Ref))
|
||||||
|
of DCompoundKind.dict:
|
||||||
|
for k, e in pat.dcompound.dict.entries: walkKey(e, k)
|
||||||
|
result.popCount.inc
|
||||||
|
walkNode(node, 0, toPreserve(0, Ref), pat).nextNode.continuation
|
||||||
|
|
||||||
type
|
type
|
||||||
Index* = object
|
Index* = object
|
||||||
|
@ -175,93 +177,103 @@ type
|
||||||
proc initIndex*(): Index =
|
proc initIndex*(): Index =
|
||||||
Index(root: Node(continuation: Continuation()))
|
Index(root: Node(continuation: Continuation()))
|
||||||
|
|
||||||
using index: Index
|
proc add*(index: var Index; turn: var Turn; pattern: Pattern; observer: Ref) =
|
||||||
|
trace "add pattern ", pattern, " for ", observer
|
||||||
proc addHandler*(index; res: Analysis; callback: HandlerCallback) =
|
|
||||||
assert(not index.root.isNil)
|
|
||||||
let
|
let
|
||||||
constPaths = res.constPaths
|
analysis = analyse pattern
|
||||||
constVals = res.constVals
|
continuation = index.root.extend pattern
|
||||||
capturePaths = res.capturePaths
|
var constValMap = continuation.leafMap.getOrDefault(analysis.constPaths)
|
||||||
continuation = index.root.extend(res.skeleton)
|
|
||||||
var constValMap = continuation.leafMap.getOrDefault(constPaths)
|
|
||||||
if constValMap.isNil:
|
if constValMap.isNil:
|
||||||
constValMap = newTable[seq[Value], Leaf]()
|
trace "allocate constValMap in leafMap for ", analysis.constPaths
|
||||||
continuation.leafMap[constPaths] = constValMap
|
new constValMap
|
||||||
for a in continuation.cachedAssertions:
|
for a in continuation.cachedAssertions:
|
||||||
let key = projectPaths(a, constPaths)
|
let key = projectPaths(a, analysis.constPaths)
|
||||||
var leaf = constValMap.getOrDefault(key)
|
var leaf = constValMap.getOrDefault(key)
|
||||||
if leaf.isNil:
|
if leaf.isNil:
|
||||||
new leaf
|
new leaf
|
||||||
constValMap[key] = leaf
|
constValMap[key] = leaf
|
||||||
leaf.cachedAssertions.incl(a)
|
leaf.cachedAssertions.incl(a)
|
||||||
var leaf = constValMap.getOrDefault(constVals)
|
trace "update leafMap for ", analysis.constPaths
|
||||||
|
continuation.leafMap[analysis.constPaths] = constValMap
|
||||||
|
var leaf = constValMap.getOrDefault(analysis.constValues)
|
||||||
if leaf.isNil:
|
if leaf.isNil:
|
||||||
new leaf
|
new leaf
|
||||||
constValMap[constVals] = leaf
|
constValMap[analysis.constValues] = leaf
|
||||||
var handler = leaf.handlerMap.getOrDefault(capturePaths)
|
trace "get observerGroup for ", analysis.capturePaths
|
||||||
if handler.isNil:
|
var observerGroup = leaf.observerGroups.getOrDefault(analysis.capturePaths)
|
||||||
new handler
|
if observerGroup.isNil:
|
||||||
leaf.handlerMap[capturePaths] = handler
|
trace "allocate observerGroup for ", analysis.capturePaths
|
||||||
|
new observerGroup
|
||||||
for a in leaf.cachedAssertions:
|
for a in leaf.cachedAssertions:
|
||||||
let a = projectPaths(a, capturePaths)
|
discard observerGroup.cachedCaptures.change(projectPaths(a, analysis.capturePaths), +1)
|
||||||
if handler.cachedCaptures.contains(a):
|
leaf.observerGroups[analysis.capturePaths] = observerGroup
|
||||||
discard handler.cachedCaptures.change(a, +1)
|
var captureMap = newTable[seq[Value], Handle]()
|
||||||
handler.callbacks.incl(callback)
|
for (count, captures) in observerGroup.cachedCaptures:
|
||||||
for captures, count in handler.cachedCaptures.pairs:
|
captureMap[captures] = publish(turn, observer, captures)
|
||||||
callback(addedEvent, captures)
|
observerGroup.observers[observer] = captureMap
|
||||||
|
|
||||||
proc removeHandler*(index; res: Analysis; callback: HandlerCallback) =
|
proc remove*(index: var Index; turn: var Turn; pattern: Pattern; observer: Ref) =
|
||||||
let continuation = index.root.extend(res.skeleton)
|
var
|
||||||
try:
|
analysis = analyse pattern
|
||||||
let
|
continuation = index.root.extend pattern
|
||||||
constValMap = continuation.leafMap[res.constPaths]
|
let constValMap = continuation.leafMap.getOrDefault(analysis.constPaths)
|
||||||
leaf = constValMap[res.constVals]
|
if not constValMap.isNil:
|
||||||
handler = leaf.handlerMap[res.capturePaths]
|
let leaf = constValMap.getOrDefault(analysis.constValues)
|
||||||
handler.callbacks.excl(callback)
|
if not leaf.isNil:
|
||||||
if handler.callbacks.len == 0:
|
let observerGroup = leaf.observerGroups.getOrDefault(analysis.capturePaths)
|
||||||
leaf.handlerMap.del(res.capturePaths)
|
if not observerGroup.isNil:
|
||||||
|
let captureMap = observerGroup.observers.getOrDefault(observer)
|
||||||
|
if not captureMap.isNil:
|
||||||
|
for handle in captureMap.values: retract(observer.target, turn, handle)
|
||||||
|
observerGroup.observers.del(observer)
|
||||||
|
if observerGroup.observers.len == 0:
|
||||||
|
leaf.observerGroups.del(analysis.capturePaths)
|
||||||
if leaf.isEmpty:
|
if leaf.isEmpty:
|
||||||
constValMap.del(res.constVals)
|
constValMap.del(analysis.constValues)
|
||||||
if constValMap.len == 0:
|
if constValMap.len == 0:
|
||||||
continuation.leafMap.del(res.constPaths)
|
continuation.leafMap.del(analysis.constPaths)
|
||||||
except KeyError: discard
|
|
||||||
|
|
||||||
proc adjustAssertion*(index: var Index; outerValue: Value; delta: int): ChangeDescription =
|
proc adjustAssertion*(index: var Index; turn: var Turn; outerValue: Value; delta: int): bool =
|
||||||
result = index.allAssertions.change(outerValue, delta)
|
case index.allAssertions.change(outerValue, delta)
|
||||||
case result
|
|
||||||
of cdAbsentToPresent:
|
of cdAbsentToPresent:
|
||||||
|
result = true
|
||||||
index.root.modify(
|
index.root.modify(
|
||||||
|
turn,
|
||||||
addedEvent,
|
addedEvent,
|
||||||
outerValue,
|
outerValue,
|
||||||
(proc (c: Continuation; v: Value) = c.cachedAssertions.incl(v)),
|
(proc (c: Continuation; v: Value) = c.cachedAssertions.incl(v)),
|
||||||
(proc (l: Leaf; v: Value) = l.cachedAssertions.incl(v)),
|
(proc (l: Leaf; v: Value) = l.cachedAssertions.incl(v)),
|
||||||
(proc (h: Handler; vs: seq[Value]) =
|
(proc (turn: var Turn; group: ObserverGroup; vs: seq[Value]) =
|
||||||
if h.cachedCaptures.change(vs, +1) == cdAbsentToPresent:
|
if group.cachedCaptures.change(vs, +1) == cdAbsentToPresent:
|
||||||
#debugEcho " assertion of ", outerValue
|
for (observer, captureMap) in group.observers.pairs:
|
||||||
for cb in h.callbacks: cb(addedEvent, vs)))
|
let a = vs.toPreserve(Ref)
|
||||||
|
trace "publish to dataspace observer ", observer, " ", a
|
||||||
|
captureMap[vs] = publish(turn, observer, a)))
|
||||||
|
# TODO: this handle is coming from the facet?
|
||||||
of cdPresentToAbsent:
|
of cdPresentToAbsent:
|
||||||
|
result = true
|
||||||
index.root.modify(
|
index.root.modify(
|
||||||
|
turn,
|
||||||
removedEvent,
|
removedEvent,
|
||||||
outerValue,
|
outerValue,
|
||||||
(proc (c: Continuation; v: Value) = c.cachedAssertions.excl(v)),
|
(proc (c: Continuation; v: Value) = c.cachedAssertions.excl(v)),
|
||||||
(proc (l: Leaf; v: Value) = l.cachedAssertions.excl(v)),
|
(proc (l: Leaf; v: Value) = l.cachedAssertions.excl(v)),
|
||||||
(proc (h: Handler; vs: seq[Value]) =
|
(proc (turn: var Turn; group: ObserverGroup; vs: seq[Value]) =
|
||||||
if h.cachedCaptures.change(vs, -1) == cdPresentToAbsent:
|
if group.cachedCaptures.change(vs, -1) == cdPresentToAbsent:
|
||||||
#debugEcho "retraction of ", outerValue
|
for (observer, captureMap) in group.observers.pairs:
|
||||||
for cb in h.callbacks: cb(removedEvent, vs)))
|
retract(observer.target, turn, captureMap[vs])
|
||||||
else:
|
captureMap.del(vs)))
|
||||||
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 deliverMessage*(index; v: Value; leafCb: proc (l: Leaf; v: Value) {.gcsafe.}) =
|
proc add*(index: var Index; turn: var Turn; v: Assertion): bool =
|
||||||
proc handlerCb(h: Handler; vs: seq[Value]) =
|
adjustAssertion(index, turn, v, +1)
|
||||||
for cb in h.callbacks: cb(messageEvent, vs)
|
proc remove*(index: var Index; turn: var Turn; v: Assertion): bool =
|
||||||
index.root.modify(messageEvent, v, continuationNoop, leafCb, handlerCb)
|
adjustAssertion(index, turn, v, -1)
|
||||||
|
|
||||||
proc deliverMessage*(index; v: Value) =
|
proc deliverMessage*(index: Index; turn: var Turn; v: Value) =
|
||||||
proc handlerCb(h: Handler; vs: seq[Value]) =
|
proc observersCb(turn: var Turn; group: ObserverGroup; vs: seq[Value]) =
|
||||||
for cb in h.callbacks: cb(messageEvent, vs)
|
for observer in group.observers.keys: message(turn, observer, vs)
|
||||||
index.root.modify(messageEvent, v, continuationNoop, leafNoop, handlerCb)
|
index.root.modify(turn, messageEvent, v, continuationNoop, leafNoop, observersCb)
|
||||||
|
|
Loading…
Reference in New Issue