Compare commits

...

3 Commits
main ... mirror

7 changed files with 393 additions and 19 deletions

View File

@ -0,0 +1,36 @@
version 1 .
Reflect = <reflect @thing #:any> .
Type = <type @thing #:any @type TypeName> .
Facet = <facet @thing #:any @facet #:any> .
Attribute = <attribute @thing #:any @attribute any> .
TypeName =
/ =entity
/ =facet
/ =actor
/ =space
/ =external
/ @other symbol .
# Entities - user controlled properties, but here are some suggestions
EntityClass = <entity-class @value any> .
# Facet
FacetActor = <actor @actor #:any> .
FacetAlive = <alive @alive bool> .
FacetChild = <child @child #:any> .
FacetParent = =root / <parent @parent #:any> .
FacetAssertion = <assertion @handle int @target #:any>.
FacetInertPreventers = <inert-preventers @inertPreventers int> .
# Actor
ActorName = <name @name any> .
ActorRoot = <root @root #:any> .
ActorStatus = =running / =done / <crashed @reason string> .
# Space
SpaceTaskCount = <task-count @taskCount int> .
SpaceActor = <actor @actor #:any> .
SpaceStatus = =running / =paused / =terminated .

View File

@ -12,6 +12,7 @@ export * from './runtime/bag.js';
export * as Dataflow from './runtime/dataflow.js';
export { Field } from './runtime/dataflow.js';
export * from './runtime/dataspace.js';
export * from './runtime/mirror.js';
export * as Pattern from './runtime/pattern.js';
export * as QuasiValue from './runtime/quasivalue.js';
export * from './runtime/randomid.js';

View File

@ -9,6 +9,9 @@ import { ActionDescription, StructuredTask, TaskAction } from './task.js';
import { randomId } from './randomid.js';
import * as Q from '../gen/queuedTasks.js';
import { Mirror, Reflectable, _asRef } from './mirror';
import * as Refl from '../gen/mirror';
export type AnyValue = Value<Ref>;
//---------------------------------------------------------------------------
@ -34,12 +37,17 @@ export interface Entity {
retract(handle: Handle): void;
message(body: Assertion): void;
sync(peer: Ref): void;
data?: unknown;
readonly data?: unknown;
// Caller (the mirror) will ensure that we are not reflected more than once
// simultaneously, and will also check that it has jurisdiction over us
setMirror?(mirror: Mirror | null): void;
}
export type Cap = Ref;
export class Ref {
export class Ref implements Reflectable {
get [IsEmbedded](): true { return true; }
readonly relay: Facet;
@ -52,6 +60,10 @@ export class Ref {
this.attenuation = attenuation;
}
asRef(): Ref {
return this;
}
toString() {
let entityRepr = '' + this.target;
if (entityRepr === '[object Object]') {
@ -97,13 +109,19 @@ export const __setNextActorId = (v: number) => nextActorId = v;
export type DataflowGraph = Graph<DataflowBlock, Cell>;
export type DataflowBlock = () => void;
export class Actor {
name: AnyValue = Symbol.for('A-' + randomId(16));
export class Actor implements Reflectable {
private _name: AnyValue = Symbol.for('A-' + randomId(16));
readonly space: ActorSpace;
readonly root: Facet;
_dataflowGraph: DataflowGraph | null = null;
exitReason: ExitReason = null;
readonly exitHooks: Array<LocalAction> = [];
_reflectableRef?: Ref;
_reflection?: {
mirror: Mirror;
exitReasonHandle?: Handle;
nameHandle?: Handle;
};
static boot(
bootProc: LocalAction,
@ -131,6 +149,31 @@ export class Actor {
Turn.for(new Facet(this, this.root), stopIfInertAfter(bootProc));
}
asRef(): Ref {
return _asRef(this, this.root);
}
setMirror(mirror: Mirror | null): void {
if (mirror === null) {
delete this._reflection;
} else {
this._reflection = { mirror };
mirror.setFocusType(Refl.TypeName.actor());
mirror.constProp(Refl.ActorRoot(this.root.asRef()));
this._reflectName();
this._reflectExitReason();
}
}
get name(): AnyValue {
return this._name;
}
set name(n: AnyValue) {
this._name = n;
this._reflectName();
}
get dataflowGraph(): DataflowGraph {
if (this._dataflowGraph === null) {
this._dataflowGraph =
@ -152,6 +195,7 @@ export class Actor {
_terminateWith(reason: Exclude<ExitReason, null>) {
if (this.exitReason !== null) return;
this.exitReason = reason;
this._reflectExitReason();
if (!reason.ok) {
console.error(`${this} crashed:`, reason.err);
}
@ -160,6 +204,19 @@ export class Actor {
this.space.deregister(this);
}
_reflectExitReason() {
this._reflection?.mirror.setProp(
this._reflection,
'exitReasonHandle',
(this.exitReason === null ? Refl.ActorStatus.running() :
this.exitReason.ok ? Refl.ActorStatus.done() :
Refl.ActorStatus.crashed('' + this.exitReason.err)));
}
_reflectName() {
this._reflection?.mirror.setProp(this._reflection, 'nameHandle', Refl.ActorName(this.name));
}
repairDataflowGraph() {
if (this._dataflowGraph === null) return;
this._dataflowGraph.repairDamage(block => block());
@ -170,7 +227,7 @@ export class Actor {
}
}
export class Facet {
export class Facet implements Reflectable {
readonly id = nextActorId++;
readonly actor: Actor;
readonly parent: Facet | null;
@ -180,12 +237,59 @@ export class Facet {
// ^ shutdownActions are not exitHooks - those run even on error. These are for clean shutdown
isLive = true;
inertCheckPreventers = 0;
_reflectableRef?: Ref;
_reflection?: {
mirror: Mirror;
aliveHandle?: Handle;
childHandles: { [id: string]: Handle };
inertPreventersHandle?: Handle;
assertionHandles: { [id: string]: Handle };
};
constructor(actor: Actor, parent: Facet | null, initialAssertions: OutboundMap = new Map()) {
this.actor = actor;
this.parent = parent;
if (parent) parent.children.add(this);
this.outbound = initialAssertions;
if (parent) parent.addChild(this);
this.outbound = initialAssertions; // no mirror yet, no need to reflect
}
asRef(): Ref {
return _asRef(this, this);
}
setMirror(mirror: Mirror | null): void {
if (mirror === null) {
delete this._reflection;
} else {
this._reflection = { mirror, childHandles: {}, assertionHandles: {} };
mirror.setFocusType(Refl.TypeName.facet());
mirror.constProp(Refl.FacetActor(this.actor.asRef()));
mirror.setProp(this._reflection, 'aliveHandle', Refl.FacetAlive(this.isLive));
this.children.forEach(c => mirror.setProp(
this._reflection!.childHandles,
'' + c.id,
Refl.FacetChild(c.asRef())));
mirror.constProp(this.parent === null
? Refl.FacetParent.root()
: Refl.FacetParent.parent(this.parent.asRef()));
this.outbound.forEach(e => mirror.setProp(
this._reflection!.assertionHandles,
'' + e.handle,
Refl.FacetAssertion({ handle: e.handle, target: e.peer })));
this._reflectInertPreventers();
}
}
addChild(child: Facet) {
this.children.add(child);
this._reflection?.mirror.setProp(this._reflection.childHandles,
'' + child.id,
Refl.FacetChild(child.asRef()));
}
removeChild(child: Facet) {
this.children.delete(child);
this._reflection?.mirror.setProp(this._reflection.childHandles, '' + child.id, void 0);
}
wrap<T extends Tuple<any>>(f: (... args: T) => void): (... args: T) => void {
@ -214,13 +318,21 @@ export class Facet {
preventInertCheck(): () => void {
let armed = true;
this.inertCheckPreventers++;
this._reflectInertPreventers();
return () => {
if (!armed) return;
armed = false;
this.inertCheckPreventers--;
this._reflectInertPreventers();
};
}
_reflectInertPreventers() {
this._reflection?.mirror.setProp(this._reflection,
'inertPreventersHandle',
Refl.FacetInertPreventers(this.inertCheckPreventers));
}
_halfLink(other: Facet): void {
const h = nextHandle++;
const e = {
@ -230,14 +342,19 @@ export class Facet {
established: true,
};
this.outbound.set(h, e);
this._reflection?.mirror.setProp(
this._reflection.assertionHandles,
'' + h,
Refl.FacetAssertion({ handle: h, target: e.peer }));
}
_terminate(orderly: boolean): void {
if (!this.isLive) return;
this.isLive = false;
this._reflection?.mirror.setProp(this._reflection, 'aliveHandle', Refl.FacetAlive(this.isLive));
const parent = this.parent;
if (parent) parent.children.delete(this);
if (parent) parent.removeChild(this);
Turn.active._inFacet(this, () => {
this.children.forEach(child => child._terminate(orderly));
@ -285,7 +402,7 @@ export class StopOnRetract implements Partial<Entity> {
retract(_handle: Handle): void {
Turn.active.stop(Turn.activeFacet);
}
data = STOP_ON_RETRACT;
actor = STOP_ON_RETRACT;
}
export function _sync_impl(e: Partial<Entity>, peer: Ref): void {
@ -397,6 +514,13 @@ export class Turn {
this.enqueue(spawningFacet,
() => {
initialAssertions.forEach(key => spawningFacet.outbound.delete(key));
{
const r = spawningFacet._reflection;
r?.mirror.turn(() => {
initialAssertions.forEach(
h => r.mirror.setProp(r.assertionHandles, '' + h, void 0));
});
}
newActor.space.queueTask({
perform() { newActor._boot(bootProc); },
describe() { return { type: 'bootActor', detail }; },
@ -485,6 +609,10 @@ export class Turn {
const crossSpace = this.activeFacet.actor.space !== ref.relay.actor.space;
const e = { handle: h, peer: ref, crossSpace: crossSpace ? a : null, established: false };
this.activeFacet.outbound.set(h, e);
this.activeFacet._reflection?.mirror.setProp(
this.activeFacet._reflection.assertionHandles,
'' + h,
Refl.FacetAssertion({ handle: h, target: ref }));
this.enqueue(ref.relay,
() => {
e.established = true;
@ -520,6 +648,10 @@ export class Turn {
_retract(e: OutboundAssertion): void {
this.activeFacet.outbound.delete(e.handle);
this.activeFacet._reflection?.mirror.setProp(
this.activeFacet._reflection.assertionHandles,
'' + e.handle,
void 0);
this.enqueue(e.peer.relay,
() => {
if (e.established) {
@ -586,13 +718,14 @@ export class Turn {
}
enqueue(relay: Facet, a0: LocalAction, detail: () => ActionDescription): void {
if (this.queues === null) {
throw new Error("Attempt to reuse a committed Turn");
}
const a: StructuredTask<TaskAction> = {
this._enqueue(relay, {
perform() { Turn.active._inFacet(relay, a0); },
describe() { return { targetFacet: relay, action: detail() }; },
};
});
}
_enqueue(relay: Facet, a: StructuredTask<TaskAction>): void {
if (this.queues === null) throw new Error("Attempt to reuse a committed Turn");
this.queues.get(relay.actor)?.push(a) ?? this.queues.set(relay.actor, [a]);
}

View File

@ -1,11 +1,13 @@
/// SPDX-License-Identifier: GPL-3.0-or-later
/// SPDX-FileCopyrightText: Copyright © 2016-2024 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
import { IdentityMap, KeyedDictionary, stringify } from '@preserves/core';
import { Record, IdentityMap, KeyedDictionary, stringify } from '@preserves/core';
import { Index, IndexObserver } from './skeleton.js';
import { Actor, AnyValue, Assertion, DetailedAction, Entity, Facet, Handle, LocalAction, Ref, Turn } from './actor.js';
import { Actor, AnyValue, Assertable, Assertion, DetailedAction, Entity, Facet, Handle, LocalAction, Ref, Turn } from './actor.js';
import { Observe, toObserve } from '../gen/dataspace.js';
import * as P from '../gen/dataspacePatterns.js';
import * as Refl from '../gen/mirror.js';
import { Mirror } from './mirror.js';
export type DataspaceOptions = {
tracer?: (event: '+' | '-' | '!',
@ -83,10 +85,33 @@ export class Dataspace implements Partial<Entity> {
readonly observerMap = new IdentityMap<Ref, DataspaceObserver>();
readonly data = this;
private _reflection?: {
mirror: Mirror,
assertionHandles: { [key: string]: Handle },
};
constructor(options?: DataspaceOptions) {
this.options = options ?? {};
}
setMirror(mirror: Mirror | null): void {
if (mirror === null) {
delete this._reflection;
} else {
this._reflection = { mirror, assertionHandles: {} };
mirror.constProp(Refl.EntityClass<Ref>(Symbol.for('dataspace')));
this.handleMap.forEach((v, handle) =>
this._reflection?.mirror.setProp(
this._reflection.assertionHandles,
'' + handle,
this._assertionAttribute(handle, v)));
}
}
_assertionAttribute(handle: Handle, v: Assertion): Assertable {
return Record(Symbol.for('contents'), [handle, v]);
}
assert(v: Assertion, handle: Handle): void {
const is_new = this.index.addAssertion(v, Turn.active);
this.options.tracer?.('+', v, this, is_new);
@ -101,12 +126,20 @@ export class Dataspace implements Partial<Entity> {
if (this.options.dumpIndex ?? false) this.index.dump();
}
this.handleMap.set(handle, v);
this._reflection?.mirror.setProp(
this._reflection.assertionHandles,
'' + handle,
this._assertionAttribute(handle, v));
}
retract(handle: Handle): void {
const v = this.handleMap.get(handle);
if (v === void 0) return;
this.handleMap.delete(handle);
this._reflection?.mirror.setProp(
this._reflection.assertionHandles,
'' + handle,
void 0);
const is_last = this.index.removeAssertion(v, Turn.active);
this.options.tracer?.('-', v, this, is_last);
if (is_last) {

View File

@ -0,0 +1,116 @@
/// SPDX-License-Identifier: GPL-3.0-or-later
/// SPDX-FileCopyrightText: Copyright © 2023-2024 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
import { Actor, Ref, Facet, LocalAction, Turn, assertionFrom } from './actor';
import type { Handle, Assertable } from './actor';
import type { Field } from './dataflow';
import * as Refl from '../gen/mirror';
import { Observe } from '../gen/dataspace';
import * as P from './pattern';
import { ActorSpace } from './space';
import { assertionFacetObserver, Dataspace } from './dataspace';
import { stringify, isEmbedded, KeyedDictionary } from '@preserves/core';
export interface Reflectable {
asRef(): Ref;
}
/// A mirror reflects a "focus" object from a "source" space into an "image" in a *different*
/// "target" space.
///
/// From the point of view of the target space, the source space is a Plain Old JavaScript
/// program. In all other situations (DOM events, network activity etc.) explicit entry to
/// facet turns is required to operate within the target space. Mirrors should be the same.
///
export class Mirror {
readonly imageFacet!: Facet;
private focusType!: Field<Refl.TypeName>;
constructor (
public readonly focus: Ref,
private readonly image: Ref,
public readonly focusSpace: ActorSpace,
t: Turn,
) {
let installed = false;
t.facet(() => {
(this as any).imageFacet = t.activeFacet;
this.focusType = t.field(Refl.TypeName.entity(), 'focusType');
if ((this.focus.target.data !== this.focusSpace) &&
(this.focus.relay.actor.space !== this.focusSpace))
{
this.focusType.value = Refl.TypeName.external();
} else {
this.focus.target.setMirror?.(this);
installed = true;
}
t.assert(this.image,
Refl.Facet({ thing: this.focus, facet: this.focus.relay.asRef() }));
t.assertDataflow(() => ({
target: this.image,
assertion: Refl.Type({ thing: this.focus, type: this.focusType.value }),
}));
});
if (installed) this.imageFacet.onStop(() => this.focus.target.setMirror?.(null));
}
setFocusType(n: Refl.TypeName) {
this.turn(() => this.focusType.value = n);
}
turn(a: LocalAction) {
const t = Turn.active;
(t && t.activeFacet === this.imageFacet) ? a() : Turn.for(this.imageFacet, a);
}
constProp(prop: Assertable): void {
this.turn(() => {
const attribute = assertionFrom(prop);
Turn.active.assert(this.image, Refl.Attribute({ thing: this.focus, attribute }));
});
}
setProp<Name extends string>(
c: { [K in Name]?: Handle },
n: Name,
prop: Assertable | undefined,
): void {
this.turn(() => {
const a = prop
? Refl.Attribute({ thing: this.focus, attribute: assertionFrom(prop) })
: void 0;
const newHandle = Turn.active.replace(this.image, c[n], a);
(newHandle === void 0) ? delete c[n] : c[n] = newHandle;
});
}
}
export function spawnMirror(sourceSpace: ActorSpace, targetSpace = new ActorSpace()): Ref {
let image!: Ref;
Actor.boot(() => {
const t = Turn.active;
image = t.ref(new Dataspace({
tracer: (event, a, _ds, sig) => console.log('DS', event, stringify(a), sig),
}));
t.assert(image, Observe({
pattern: P.rec(Symbol.for('reflect'), P.bind()),
observer: t.ref(assertionFacetObserver(a => {
if (!Array.isArray(a)) return;
const [thing] = a;
if (!isEmbedded(thing)) return;
new Mirror(thing, image, sourceSpace, Turn.active);
})),
}));
}, void 0, targetSpace);
return image;
}
export function _asRef(
x: { _reflectableRef?: Ref, setMirror(mirror: Mirror | null): void },
facet: Facet,
): Ref {
if (x._reflectableRef === void 0) {
x._reflectableRef = new Ref(facet, { data: x, setMirror: m => x.setMirror(m) });
}
return x._reflectableRef;
}

View File

@ -1,8 +1,10 @@
/// SPDX-License-Identifier: GPL-3.0-or-later
/// SPDX-FileCopyrightText: Copyright © 2023-2024 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
import { IdentityMap, IdentitySet, forEachEmbedded } from '@preserves/core';
import { IdentityMap, IdentitySet, embeddedId, forEachEmbedded } from '@preserves/core';
import type { Actor, Assertion, ExitReason, Handle, Ref } from './actor.js';
import { Mirror } from './mirror.js';
import * as Refl from '../gen/mirror';
import type { StructuredTask, TaskDescription } from './task.js';
const LIMIT = 25000;
@ -24,14 +26,29 @@ export class ActorSpace {
delayedTasks: Array<StructuredTask<TaskDescription>> = [];
taskFlushHandle: ReturnType<typeof setTimeout> | null = null;
_reflection?: {
mirror: Mirror;
stateHandle?: Handle;
taskCountHandle?: Handle;
actorHandles: { [key: string]: Handle };
};
register(actor: Actor): boolean {
if (this.state === ActorSpaceState.TERMINATED) return false;
this.actors.add(actor);
this._reflection?.mirror.setProp(
this._reflection.actorHandles,
'' + embeddedId(actor),
Refl.SpaceActor(actor.asRef()));
return true;
}
deregister(actor: Actor) {
this.actors.delete(actor);
this._reflection?.mirror.setProp(
this._reflection.actorHandles,
'' + embeddedId(actor),
void 0);
if (this.actors.size === 0) {
this.shutdown({ ok: true });
}
@ -62,6 +79,7 @@ export class ActorSpace {
shutdown(reason: Exclude<ExitReason, null>) {
if (this.state === ActorSpaceState.TERMINATED) return;
this.state = ActorSpaceState.TERMINATED;
this._reflectState();
Array.from(this.actors.values()).forEach(a => a._terminateWith(reason));
}
@ -71,6 +89,8 @@ export class ActorSpace {
break;
case ActorSpaceState.PAUSED:
this.taskCounter++;
this._reflectTaskCount();
this.delayedTasks.push(t);
break;
@ -80,6 +100,7 @@ export class ActorSpace {
this.taskFlushHandle = setTimeout(() => this._scheduleDelayedTasks(), 0);
}
if (this.taskCounter >= LIMIT) {
this._reflectTaskCount();
this.delayedTasks.push(t);
} else {
queueMicrotask(() => t.perform());
@ -90,6 +111,7 @@ export class ActorSpace {
_scheduleDelayedTasks() {
this.taskCounter = 0;
this._reflectTaskCount();
this.delayedTasks.forEach(t => queueMicrotask(() => t.perform()));
this.delayedTasks = [];
}
@ -105,6 +127,7 @@ export class ActorSpace {
case ActorSpaceState.RUNNING:
this.state = ActorSpaceState.PAUSED;
this._reflectState();
if (this.taskFlushHandle !== null) {
clearTimeout(this.taskFlushHandle);
this.taskFlushHandle = null;
@ -121,6 +144,7 @@ export class ActorSpace {
case ActorSpaceState.PAUSED:
this.state = ActorSpaceState.RUNNING;
this._reflectState();
this._scheduleDelayedTasks();
return true;
@ -128,4 +152,34 @@ export class ActorSpace {
return true;
}
}
setMirror(mirror: Mirror | null): void {
if (mirror === null) {
delete this._reflection;
} else {
this._reflection = { mirror, actorHandles: {} };
mirror.setFocusType(Refl.TypeName.space());
this._reflectState();
this._reflectTaskCount();
this.actors.forEach(a => mirror.setProp(
this._reflection!.actorHandles,
'' + embeddedId(a),
Refl.SpaceActor(a.asRef())));
}
}
_reflectState() {
this._reflection?.mirror.setProp(
this._reflection,
'stateHandle',
(this.state === ActorSpaceState.RUNNING ? Refl.SpaceStatus.running() :
this.state === ActorSpaceState.PAUSED ? Refl.SpaceStatus.paused() :
this.state === ActorSpaceState.TERMINATED ? Refl.SpaceStatus.terminated() :
void 0));
}
_reflectTaskCount() {
this._reflection?.mirror.setProp(
this._reflection, 'taskCountHandle', Refl.SpaceTaskCount(this.taskCounter));
}
}

View File

@ -1,11 +1,11 @@
/// SPDX-License-Identifier: GPL-3.0-or-later
/// SPDX-FileCopyrightText: Copyright © 2021-2024 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
export * as dataspacePatterns from './gen/dataspacePatterns.js';
export * as dataspace from './gen/dataspace.js';
export * as dataspacePatterns from './gen/dataspacePatterns.js';
export * as gatekeeper from './gen/gatekeeper.js';
export * as protocol from './gen/protocol.js';
export * as noise from './gen/noise.js';
export * as protocol from './gen/protocol.js';
export * as service from './gen/service.js';
export * as stdenv from './gen/stdenv.js';
export * as stream from './gen/stream.js';
@ -16,4 +16,5 @@ export * as trace from './gen/trace.js';
export * as transportAddress from './gen/transportAddress.js';
export * as worker from './gen/worker.js';
export * as mirror from './gen/mirror.js';
export * as queuedTasks from './gen/queuedTasks.js';