feat(sim): F4.2 the sim protocol, the simulation room and LocalSimHost; M9 measured
packages/protocol holds the one list of /sim messages and checks every field, naming what is wrong. packages/sim has the simulation room for F and the web worker: the document's snapshot at start, the engine's clock on the wall clock with Java's speed as cable delay, subscriptions to cable activity, a card's new frames and a program's new window lines, capture and clock kept over a stop, the command line as first action, a core failure told to every client. LocalSimHost mirrors the document into the worker and back without echo. M9 in Chromium: 17 examples give the same capture as in Node, character for character; the room runs in a real worker. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,320 @@
|
||||
// A simulation room (plan §6): one engine for one project, the clients that
|
||||
// watch it, and what goes to them. The same room runs on F for everyone in a
|
||||
// project and in a web worker for one (LocalSimHost); it takes the
|
||||
// document's snapshot when the simulation starts, so that changes made while
|
||||
// it runs come in at the next start, as in Java.
|
||||
//
|
||||
// Time: the engine's clock follows the wall clock, a millisecond for a
|
||||
// millisecond, as Java's threads do; the speed is Java's slider, the delay of
|
||||
// a cable (RuntimeSpeed), not a time lapse — unlike Swift's tick (plan §6.3).
|
||||
// A tick catches up at most MAXIMUM_CATCH_UP_MILLISECONDS: after a tab slept,
|
||||
// the simulation goes on from where it was rather than running the missed
|
||||
// minutes in one go.
|
||||
//
|
||||
// The engine outlives a stop: the capture stays, as Java's Lauscher does, and
|
||||
// the next start goes on at the time it stopped, so that the capture's times
|
||||
// keep rising (plan §6.2).
|
||||
|
||||
import * as Y from 'yjs';
|
||||
import { DEFAULT_SPEED_PERCENT, NetworkRuntimeEngine, captureRows, type CapturedFrame, type CaptureRow, type Id, type LogLine } from '@mdg-filius/core';
|
||||
import { docSnapshot } from '@mdg-filius/doc';
|
||||
import {
|
||||
parseClientMessage,
|
||||
SIM_PROTOCOL_VERSION,
|
||||
type AppKind,
|
||||
type ClientMessage,
|
||||
type ErrorCode,
|
||||
type HostMessage,
|
||||
type SimMode,
|
||||
type Topic,
|
||||
type TraceFrame,
|
||||
} from '@mdg-filius/protocol';
|
||||
|
||||
export const TICK_MILLISECONDS = 200;
|
||||
export const MAXIMUM_CATCH_UP_MILLISECONDS = 2_000;
|
||||
|
||||
export interface RoomOptions {
|
||||
/** The wall clock, in milliseconds. */
|
||||
readonly now: () => number;
|
||||
/** A seed for each start. */
|
||||
readonly seed: () => number;
|
||||
}
|
||||
|
||||
interface Client {
|
||||
readonly send: (message: HostMessage) => void;
|
||||
readonly topics: Set<Topic>;
|
||||
/** Per card, the number of the last frame sent. */
|
||||
readonly traceSent: Map<Id, number>;
|
||||
/** Per program window, how many of its program's lines it has; none while the program does not run (a stop empties it). */
|
||||
readonly appSent: Map<string, number>;
|
||||
}
|
||||
|
||||
/** A program with a window: its lines. */
|
||||
interface WithLog {
|
||||
readonly log: { readonly all: readonly LogLine<unknown>[] };
|
||||
}
|
||||
|
||||
const message = <T extends Omit<HostMessage, 'v'>>(fields: T) => ({ v: SIM_PROTOCOL_VERSION, ...fields }) as unknown as HostMessage;
|
||||
|
||||
export class SimulationRoom {
|
||||
private readonly engine: NetworkRuntimeEngine;
|
||||
private mode: SimMode = 'design';
|
||||
private speed = DEFAULT_SPEED_PERCENT;
|
||||
private readonly clients = new Map<number, Client>();
|
||||
private nextClient = 1;
|
||||
private lastWall: number | undefined;
|
||||
/** The cable each card is plugged into, as the simulation started. */
|
||||
private linkOfPort = new Map<Id, Id>();
|
||||
/** Per card, how many frames it held at the last tick. */
|
||||
private framesSeen = new Map<Id, number>();
|
||||
private lastActivity: string[] = [];
|
||||
|
||||
constructor(
|
||||
private readonly doc: Y.Doc,
|
||||
private readonly options: RoomOptions,
|
||||
) {
|
||||
this.engine = new NetworkRuntimeEngine(options.seed(), undefined, { wallClock: options.now });
|
||||
}
|
||||
|
||||
/** A new client; it gets the room's state at once. */
|
||||
connect(send: (message: HostMessage) => void): number {
|
||||
const id = this.nextClient++;
|
||||
this.clients.set(id, { send, topics: new Set(), traceSent: new Map(), appSent: new Map() });
|
||||
send(this.state());
|
||||
return id;
|
||||
}
|
||||
|
||||
disconnect(client: number): void {
|
||||
this.clients.delete(client);
|
||||
}
|
||||
|
||||
/** How many clients the room has (F stops a room nobody watches). */
|
||||
get clientCount(): number {
|
||||
return this.clients.size;
|
||||
}
|
||||
|
||||
get currentMode(): SimMode {
|
||||
return this.mode;
|
||||
}
|
||||
|
||||
/** A client's message, as it came over the wire. */
|
||||
receive(clientId: number, raw: unknown): void {
|
||||
const client = this.clients.get(clientId);
|
||||
if (client === undefined) return;
|
||||
const parsed = parseClientMessage(raw);
|
||||
if ('error' in parsed) {
|
||||
const reqId = typeof (raw as { reqId?: unknown } | null)?.reqId === 'string' ? (raw as { reqId: string }).reqId : undefined;
|
||||
client.send(message({ kind: 'error', code: parsed.error.code, detail: parsed.error.detail, ...(reqId === undefined ? {} : { reqId }) }));
|
||||
return;
|
||||
}
|
||||
this.handle(client, parsed.message);
|
||||
}
|
||||
|
||||
private handle(client: Client, input: ClientMessage): void {
|
||||
switch (input.kind) {
|
||||
case 'sub':
|
||||
for (const topic of input.topics) {
|
||||
client.topics.add(topic);
|
||||
if (topic.startsWith('trace:')) client.traceSent.delete(topic.slice('trace:'.length));
|
||||
if (topic.startsWith('app:')) client.appSent.delete(topic);
|
||||
}
|
||||
this.flush([client], true);
|
||||
return;
|
||||
case 'unsub':
|
||||
for (const topic of input.topics) client.topics.delete(topic);
|
||||
return;
|
||||
case 'start':
|
||||
this.start();
|
||||
return;
|
||||
case 'stop':
|
||||
this.stop();
|
||||
return;
|
||||
case 'speed':
|
||||
this.speed = Math.min(100, Math.max(1, Math.trunc(input.value)));
|
||||
this.engine.setSimulationSpeed(this.speed);
|
||||
this.broadcast(this.state());
|
||||
return;
|
||||
case 'input': {
|
||||
let failure: ErrorCode | undefined = 'engineFailed';
|
||||
if (!this.guarded(() => (failure = this.act(input)))) return;
|
||||
client.send(failure === undefined ? message({ kind: 'ack', reqId: input.reqId }) : message({ kind: 'error', code: failure, reqId: input.reqId }));
|
||||
this.flush([...this.clients.values()], false);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private state(): HostMessage {
|
||||
return message({ kind: 'state', mode: this.mode, speed: this.speed, t: this.engine.state.currentTimeMilliseconds });
|
||||
}
|
||||
|
||||
private broadcast(output: HostMessage): void {
|
||||
for (const client of this.clients.values()) client.send(output);
|
||||
}
|
||||
|
||||
start(): void {
|
||||
if (this.mode === 'simulation') return;
|
||||
const snapshot = docSnapshot(this.doc);
|
||||
this.engine.setSimulationSpeed(this.speed);
|
||||
if (!this.guarded(() => this.engine.handle({ type: 'start', snapshot, seed: this.options.seed(), initialTimeMilliseconds: this.engine.state.currentTimeMilliseconds }))) return;
|
||||
this.linkOfPort = new Map(snapshot.links.flatMap((link) => [[link.sourcePortId, link.id] as const, [link.targetPortId, link.id] as const]));
|
||||
this.mode = 'simulation';
|
||||
this.lastWall = this.options.now();
|
||||
this.broadcast(this.state());
|
||||
this.flush([...this.clients.values()], false);
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
if (this.mode === 'design') return;
|
||||
this.engine.handle({ type: 'stop' });
|
||||
this.mode = 'design';
|
||||
this.lastWall = undefined;
|
||||
this.broadcast(this.state());
|
||||
this.flush([...this.clients.values()], false);
|
||||
}
|
||||
|
||||
/** Lets the simulation run to the wall clock's time, then sends what came of it. */
|
||||
tick(): void {
|
||||
if (this.mode !== 'simulation') return;
|
||||
const now = this.options.now();
|
||||
const elapsed = Math.min(MAXIMUM_CATCH_UP_MILLISECONDS, Math.max(0, now - (this.lastWall ?? now)));
|
||||
this.lastWall = now;
|
||||
if (!this.guarded(() => this.engine.handle({ type: 'advance', toMilliseconds: this.engine.state.currentTimeMilliseconds + elapsed }))) return;
|
||||
this.flush([...this.clients.values()], false);
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs a step of the engine. When the core throws (a part it has not got
|
||||
* yet), the room tells every client and goes back to design mode instead of
|
||||
* dying with its worker; false then.
|
||||
*/
|
||||
private guarded(step: () => void): boolean {
|
||||
try {
|
||||
step();
|
||||
return true;
|
||||
} catch (error) {
|
||||
this.broadcast(message({ kind: 'error', code: 'engineFailed', detail: error instanceof Error ? error.message : String(error) }));
|
||||
if (this.engine.state.phase === 'running') this.engine.handle({ type: 'stop' });
|
||||
this.mode = 'design';
|
||||
this.lastWall = undefined;
|
||||
this.broadcast(this.state());
|
||||
this.flush([...this.clients.values()], false);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/** An action in a program's window; the failure, if any. */
|
||||
private act(input: Extract<ClientMessage, { kind: 'input' }>): ErrorCode | undefined {
|
||||
if (this.mode !== 'simulation') return 'notRunning';
|
||||
if (!this.engine.state.topologySnapshot.nodes.some((node) => node.id === input.node)) return 'unknownNode';
|
||||
if (input.app !== 'terminal') return 'unknownAction';
|
||||
const terminal = this.engine.terminal(input.node);
|
||||
if (terminal === undefined) return 'unknownApp';
|
||||
switch (input.action) {
|
||||
case 'enter':
|
||||
if (typeof input.args.line !== 'string') return 'badArgs';
|
||||
terminal.enter(input.args.line);
|
||||
return undefined;
|
||||
case 'interrupt':
|
||||
terminal.interrupt();
|
||||
return undefined;
|
||||
default:
|
||||
return 'unknownAction';
|
||||
}
|
||||
}
|
||||
|
||||
private program(node: Id, app: AppKind): WithLog | undefined {
|
||||
switch (app) {
|
||||
case 'terminal':
|
||||
return this.engine.terminal(node);
|
||||
case 'webBrowser':
|
||||
return this.engine.webBrowser(node);
|
||||
case 'webServer':
|
||||
return this.engine.webServer(node);
|
||||
case 'emailClient':
|
||||
return this.engine.emailClient(node);
|
||||
case 'emailServer':
|
||||
return this.engine.emailServer(node);
|
||||
case 'echoServer':
|
||||
return this.engine.echoServer(node);
|
||||
case 'simpleClient':
|
||||
return this.engine.simpleClient(node);
|
||||
case 'gnutella':
|
||||
return this.engine.gnutella(node);
|
||||
}
|
||||
}
|
||||
|
||||
/** What the clients are owed: the cables' activity, new frames of the cards and new lines of the windows they watch. */
|
||||
private flush(clients: readonly Client[], subscribing: boolean): void {
|
||||
if (!subscribing) this.updateActivity();
|
||||
const rowsCache = new Map<Id, TraceFrame[]>();
|
||||
const traceFrames = (portId: Id, after: number): TraceFrame[] => {
|
||||
const frames = this.engine.capture.frames(portId).filter((frame) => frame.number > after);
|
||||
if (frames.length === 0) return [];
|
||||
if (after === 0 && rowsCache.has(portId)) return rowsCache.get(portId)!;
|
||||
const result = traceFramesOf(frames);
|
||||
if (after === 0) rowsCache.set(portId, result);
|
||||
return result;
|
||||
};
|
||||
for (const client of clients) {
|
||||
for (const topic of client.topics) {
|
||||
if (topic === 'activity') {
|
||||
if (subscribing) client.send(message({ kind: 'activity', links: this.lastActivity }));
|
||||
} else if (topic.startsWith('trace:')) {
|
||||
const portId = topic.slice('trace:'.length);
|
||||
const reset = !client.traceSent.has(portId);
|
||||
const frames = traceFrames(portId, client.traceSent.get(portId) ?? 0);
|
||||
if (frames.length === 0 && !reset) continue;
|
||||
client.traceSent.set(portId, frames.at(-1)?.number ?? client.traceSent.get(portId) ?? 0);
|
||||
client.send(message({ kind: 'trace', portId, frames, reset }));
|
||||
} else {
|
||||
const [, node, app] = /^app:([^/]+)\/(.+)$/.exec(topic)!;
|
||||
const program = this.program(node!, app as AppKind);
|
||||
const sent = client.appSent.get(topic);
|
||||
if (program === undefined) {
|
||||
// A window of a program that is not running: it starts empty with the next start.
|
||||
if (sent !== undefined || subscribing) client.send(message({ kind: 'app', node: node!, app: app as AppKind, from: 0, lines: [] }));
|
||||
client.appSent.delete(topic);
|
||||
continue;
|
||||
}
|
||||
// The window goes on from the lines it has; after a stop (none sent) it begins anew.
|
||||
const from = sent ?? 0;
|
||||
const lines = program.log.all.slice(from);
|
||||
if (sent !== undefined && lines.length === 0) continue;
|
||||
client.appSent.set(topic, program.log.all.length);
|
||||
client.send(message({ kind: 'app', node: node!, app: app as AppKind, from, lines }));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** The cables whose cards got frames since the last tick; sent when that changes. */
|
||||
private updateActivity(): void {
|
||||
const links = new Set<string>();
|
||||
for (const portId of this.engine.capture.portIds()) {
|
||||
const frames = this.engine.capture.frames(portId);
|
||||
const count = frames.at(-1)?.number ?? 0;
|
||||
if (count > (this.framesSeen.get(portId) ?? 0)) {
|
||||
const link = this.linkOfPort.get(portId);
|
||||
if (link !== undefined) links.add(link);
|
||||
}
|
||||
this.framesSeen.set(portId, count);
|
||||
}
|
||||
const activity = [...links].sort();
|
||||
if (activity.join() === this.lastActivity.join()) return;
|
||||
this.lastActivity = activity;
|
||||
for (const client of this.clients.values()) if (client.topics.has('activity')) client.send(message({ kind: 'activity', links: activity }));
|
||||
}
|
||||
}
|
||||
|
||||
/** A card's frames as the capture window shows them, both views. */
|
||||
export function traceFramesOf(frames: readonly CapturedFrame[]): TraceFrame[] {
|
||||
const byNumber = (rows: readonly CaptureRow[]) => {
|
||||
const grouped = new Map<number, CaptureRow[]>();
|
||||
for (const row of rows) grouped.set(row.number, [...(grouped.get(row.number) ?? []), row]);
|
||||
return grouped;
|
||||
};
|
||||
const rows = byNumber(captureRows(frames, { withAddresses: true }));
|
||||
const layered = byNumber(captureRows(frames, { withAddresses: false }));
|
||||
return frames.map((frame) => ({ number: frame.number, rows: rows.get(frame.number) ?? [], layeredRows: layered.get(frame.number) ?? [] }));
|
||||
}
|
||||
Reference in New Issue
Block a user