Refine references and syncing

This commit is contained in:
Tony Garnock-Jones 2021-02-23 15:38:57 +01:00
parent cd7efd615e
commit 4045d0d0ed
2 changed files with 44 additions and 28 deletions

View File

@ -9,24 +9,38 @@ export type ExitReason = null | { ok: true } | { ok: false, err: Error };
export const assert = Symbol('assert'); export const assert = Symbol('assert');
export const retract = Symbol('retract'); export const retract = Symbol('retract');
export const message = Symbol('message'); export const message = Symbol('message');
export const sync = Symbol('sync');
export interface Entity { export interface Entity {
[assert]?(turn: Turn, assertion: Assertion, handle: Handle): void; [assert]?(turn: Turn, assertion: Assertion, handle: Handle): void;
[retract]?(turn: Turn, handle: Handle): void; [retract]?(turn: Turn, handle: Handle): void;
[message]?(turn: Turn, message: Assertion): void; [message]?(turn: Turn, body: Assertion): void;
[sync]?(turn: Turn, peer: Entity): void;
} }
export class Ref { export class Ref implements Entity {
readonly actor: Actor; readonly relay: Actor;
readonly target: Entity; readonly target: Entity;
constructor(actor: Actor, target: Entity) { constructor(relay: Actor, target: Entity) {
this.actor = actor; this.relay = relay;
this.target = target; this.target = target;
} }
sync(turn: Turn, syncable: Ref) { [assert](turn: Turn, assertion: Assertion, handle: Handle): void {
turn.enqueue(syncable.actor, t => t.message(syncable, true)); this.target[assert]?.(turn, assertion, handle);
}
[retract](turn: Turn, handle: Handle): void {
this.target[retract]?.(turn, handle);
}
[message](turn: Turn, body: Assertion): void {
this.target[message]?.(turn, body);
}
[sync](turn: Turn, peer: Ref) {
this.target[sync] ? this.target[sync]!(turn, peer) : turn.message(peer, true);
} }
} }
@ -79,12 +93,12 @@ export class Turn {
this.actor = actor; this.actor = actor;
} }
ref(t: Entity): Ref { ref(e: Entity): Ref {
return new Ref(this.actor, t); return (e instanceof Ref) ? e : new Ref(this.actor, e);
} }
spawn(bootProc: LocalAction, initialAssertions = new IdentitySet<Handle>()): void { spawn(bootProc: LocalAction, initialAssertions = new IdentitySet<Handle>()): void {
this.tasks.push(() => { this.enqueue(void 0, () => {
const newOutbound: OutboundMap = new Map(); const newOutbound: OutboundMap = new Map();
initialAssertions.forEach(key => { initialAssertions.forEach(key => {
newOutbound.set(key, this.actor.outbound.get(key)!); // we trust initialAssertions newOutbound.set(key, this.actor.outbound.get(key)!); // we trust initialAssertions
@ -96,14 +110,14 @@ export class Turn {
} }
quit(): void { quit(): void {
this.tasks.push(t => this.actor.terminateWith(t, { ok: true })); this.enqueue(void 0, t => this.actor.terminateWith(t, { ok: true }));
} }
assert(location: Ref, assertion: Assertion): Handle { assert(ref: Ref, assertion: Assertion): Handle {
const h = nextHandle++; const h = nextHandle++;
this.enqueue(location.actor, t => { this.enqueue(ref.relay, t => {
this.actor.outbound.set(h, location); this.actor.outbound.set(h, ref);
location.target[assert]?.(t, assertion, h); ref[assert]?.(t, assertion, h);
}); });
return h; return h;
} }
@ -112,29 +126,30 @@ export class Turn {
this._retract(this.actor.outbound.get(h)!, h); this._retract(this.actor.outbound.get(h)!, h);
} }
replace(location: Ref, h: Handle | undefined, assertion: Assertion): Handle { replace(ref: Ref, h: Handle | undefined, assertion: Assertion): Handle {
const newHandle = this.assert(location, assertion); const newHandle = this.assert(ref, assertion);
if (h !== void 0) this.retract(h); if (h !== void 0) this.retract(h);
return newHandle; return newHandle;
} }
_retract(location: Ref, handle: Handle): void { _retract(ref: Ref, handle: Handle): void {
this.enqueue(location.actor, t => { this.enqueue(ref.relay, t => {
this.actor.outbound.delete(handle); this.actor.outbound.delete(handle);
location.target[retract]?.(t, handle); ref[retract]?.(t, handle);
}); });
} }
sync(loc: Ref): Promise<Turn> { sync(ref: Ref): Promise<Turn> {
return new Promise(resolve => this.enqueue(loc.actor, t => return new Promise(resolve =>
loc.sync(t, this.ref({ [message]: resolve })))); this.enqueue(ref.relay, t => ref[sync]?.(t, this.ref({ [message]: resolve }))));
} }
message(location: Ref, assertion: Assertion): void { message(ref: Ref, assertion: Assertion): void {
this.enqueue(location.actor, t => location.target[message]?.(t, assertion)); this.enqueue(ref.relay, t => ref[message]?.(t, assertion));
} }
enqueue(actor: Actor, a: LocalAction): void { enqueue(relay: Actor | undefined, a: LocalAction): void {
this.queues.get(actor)?.push(a) ?? this.queues.set(actor, [a]); (relay === void 0) ? this.tasks.push(a)
: this.queues.get(relay)?.push(a) ?? this.queues.set(relay, [a]);
} }
} }

View File

@ -59,7 +59,7 @@ let startTime = Date.now();
let prevValue = 0; let prevValue = 0;
const LIMIT = 500000; const LIMIT = 500000;
Turn.for(new Actor(), async (t: Turn) => { Turn.for(new Actor(), async (t: Turn) => {
const ds = new Ref(new Actor(), new Dataspace()); const ds = t.ref(new Dataspace());
// Box // Box
t.spawn(t => { t.spawn(t => {
@ -71,6 +71,7 @@ Turn.for(new Actor(), async (t: Turn) => {
setValue(t, 0); setValue(t, 0);
t.assert(ds, Observe(SetBox.constructorInfo.label, t.ref({ t.assert(ds, Observe(SetBox.constructorInfo.label, t.ref({
[message](t: Turn, [newValue]: [number]): void { [message](t: Turn, [newValue]: [number]): void {
// console.log(`Box: got ${newValue}`);
if (newValue % 25000 === 0) { if (newValue % 25000 === 0) {
const endTime = Date.now(); const endTime = Date.now();
const delta = (endTime - startTime) / 1000.0; const delta = (endTime - startTime) / 1000.0;