feat(core): programs' threads, and FILIUS' echo server and generic client after Java, measured against Java

Programs run as generators that yield what they wait for (threads.ts); the engine resumes them after every event. TCP connect, send, receive, accept and name resolution exist as such steps; the blocking calls wrap them. EchoRun records eight stories in Java FILIUS, frames, tables and each program's window lines; echoJava.test.ts plays them card by card. Not taken from Java: a connection without a route hangs for ever; a server switched off leaves its connections' last FIN unanswered. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-01 16:27:44 +00:00
parent 0abce60aef
commit de3a1d4dfe
24 changed files with 13454 additions and 28 deletions
@@ -111,6 +111,8 @@ import { JavaCompatibleRandom } from './random.js';
 import { routeTableRows, type RouteRow } from './routeTable.js';
 import { EMPTY_TOPOLOGY_SNAPSHOT, withSnapshot, type NodeSnapshot, type TopologySnapshot } from './snapshot.js';
 import { RuntimeSpeed } from './speed.js';
+import { ApplicationThreads, waitUntil, type Steps, type Wait } from './threads.js';
+import { EchoServer, SimpleClient } from './apps/echo.js';
 import {
  PRODUCTION_TRACE_RETENTION,
  type HeaderField,
@@ -152,6 +154,8 @@ export type ScheduledEventKind =
  /** The lookups waiting for an address wake up, after what else happens at that time. */
  | { readonly kind: 'arpAnswered'; readonly nodeId: Id; readonly ipAddress: string; readonly macAddress: string }
  | { readonly kind: 'tcpTimeout'; readonly sessionId: Id }
+  /** A program's thread waited until this time (threads.ts). */
+  | { readonly kind: 'applicationWake'; readonly wakeId: number }
  /** A DNS query waited five round trips for its answer (DNSQueryAgent). */
  | { readonly kind: 'dnsQueryTimeout'; readonly socketId: Id }
  | {
@@ -333,10 +337,17 @@ export interface TcpSession {
  closingWaitsLeft: number | undefined;
  receiveAssembly: Uint8Array;
  receivedMessages: Uint8Array[];
+  /** Set when data went unacknowledged after the last attempt and the socket closed for it. */
+  dataTimedOut?: boolean;
 }

 export type TcpConnectionResult = 'connected' | 'invalidSocket' | 'unreachable' | 'timedOut';

+export type TcpSendResult = 'sent' | 'notConnected' | 'connectionLost' | 'timedOut';
+
+/** What a program's receive gets (receiveTcpSteps). */
+export type TcpReceived = { readonly type: 'message'; readonly payload: Uint8Array } | { readonly type: 'closed' } | { readonly type: 'notConnected' };
+
 /** Java's DNSQueryAgent waits five round trips for an answer. */
 export const DNS_QUERY_TIMEOUT_ROUND_TRIPS = 5;
 export type DnsResolution = { readonly type: 'answered'; readonly records: readonly DnsMessageRecord[] } | { readonly type: 'timedOut' };
@@ -850,6 +861,31 @@ export class NetworkRuntimeEngine {
  private readonly swiftArp: boolean;
  private readonly swiftSends: boolean;
  private readonly packetCapture: PacketCapture;
+  /** The programs' threads (threads.ts), and the wake-ups they wait for, by id. */
+  private readonly applicationThreads = new ApplicationThreads({
+    nowMilliseconds: () => this.stateValue.currentTimeMilliseconds,
+    scheduleWake: (deadlineMilliseconds, wake) => {
+      const wakeId = this.nextApplicationWakeId++;
+      this.applicationWakes.set(wakeId, wake);
+      this.schedule(deadlineMilliseconds, { kind: 'applicationWake', wakeId });
+    },
+    asEvent: (step) => {
+      const was = this.isProcessingScheduledEvents;
+      this.isProcessingScheduledEvents = true;
+      try {
+        step();
+      } finally {
+        this.isProcessingScheduledEvents = was;
+      }
+    },
+  });
+  private readonly applicationWakes = new Map<number, () => void>();
+  /** The programs, by node (initializePrograms). */
+  private readonly echoServersByNodeId = new Map<Id, EchoServer>();
+  private readonly simpleClientsByNodeId = new Map<Id, SimpleClient>();
+  /** Accepted connections a program has taken (acceptTcpSteps). */
+  private readonly handedOutTcpSocketIds = new Set<Id>();
+  private nextApplicationWakeId = 1;

  /**
   * `swiftCables`: a cable carries all frames at once, as in the Swift
@@ -938,11 +974,15 @@ export class NetworkRuntimeEngine {
        this.linkLocalRandom = new JavaCompatibleRandom(seed ^ 0x4c4c_4144n);
        this.dnsIdRandom = new JavaCompatibleRandom(seed ^ 0x444e_5349n);
        this.isProcessingScheduledEvents = false;
+        this.applicationThreads.clear();
+        this.applicationWakes.clear();
+        this.handedOutTcpSocketIds.clear();
        this.cableBusyUntilByPort.clear();
        this.initializeRipState(seed);
        this.initializeDhcpState();
        this.initializeDnsState();
        this.initializeNatState();
+        this.initializePrograms();
        return [this.output({ type: 'started' })];
      }

@@ -952,6 +992,11 @@ export class NetworkRuntimeEngine {
        state.phase = 'stopped';
        state.globalPacketLossEnabled = false;
        this.isProcessingScheduledEvents = false;
+        this.applicationThreads.clear();
+        this.applicationWakes.clear();
+        this.handedOutTcpSocketIds.clear();
+        this.echoServersByNodeId.clear();
+        this.simpleClientsByNodeId.clear();
        this.arpLookups = [];
        this.protocolThreads.clear();
        this.dnsQueriesBySocketId.clear();
@@ -996,6 +1041,61 @@ export class NetworkRuntimeEngine {
    }
  }

+  /** The node's echo server while the simulation runs, if it has one installed. */
+  echoServer(nodeId: Id): EchoServer | undefined {
+    return this.echoServersByNodeId.get(nodeId);
+  }
+
+  /** The node's generic client while the simulation runs, if it has one installed. */
+  simpleClient(nodeId: Id): SimpleClient | undefined {
+    return this.simpleClientsByNodeId.get(nodeId);
+  }
+
+  /**
+   * Each node's programs start with the simulation (Betriebssystem.starten),
+   * from the snapshot; what they show starts empty.
+   */
+  private initializePrograms(): void {
+    const snapshot = this.stateValue.topologySnapshot;
+    this.echoServersByNodeId.clear();
+    this.simpleClientsByNodeId.clear();
+    for (const node of snapshot.nodes) {
+      const echoServer = snapshot.echoServersByNodeId.get(node.id);
+      if (echoServer !== undefined) this.echoServersByNodeId.set(node.id, new EchoServer(this, node.id, echoServer));
+      const simpleClient = snapshot.simpleClientsByNodeId.get(node.id);
+      if (simpleClient !== undefined) this.simpleClientsByNodeId.set(node.id, new SimpleClient(this, node.id, simpleClient));
+    }
+  }
+
+  /** Begins a program's thread (threads.ts): it runs at once up to its first wait. */
+  startThread(steps: Steps<unknown>): void {
+    if (this.stateValue.phase !== 'running') return;
+    this.applicationThreads.start(steps);
+  }
+
+  /** Resumes the threads whose wait is over: after a call from outside that changed what they wait for. */
+  resumeThreads(): void {
+    this.applicationThreads.resume();
+  }
+
+  /** Runs steps to their end, waiting on the virtual clock as a caller of Java blocks. */
+  private runBlocking<T>(steps: Steps<T>): T {
+    for (let next = steps.next(); ; next = steps.next()) {
+      if (next.done === true) return next.value;
+      const wait: Wait = next.value;
+      const until = wait.until ?? (() => false);
+      if (wait.deadlineMilliseconds !== undefined) {
+        this.advanceUntil(until, wait.deadlineMilliseconds);
+        continue;
+      }
+      while (this.stateValue.phase === 'running' && !until()) {
+        const deadline = this.earliestDeadline(() => true);
+        if (deadline === undefined) throw new Error('blocking call waits for nothing that can happen');
+        this.advance(deadline);
+      }
+    }
+  }
+
  allocateFrameIdentity(): number {
    return this.stateValue.nextFrameIdentity++;
  }
@@ -1071,6 +1171,7 @@ export class NetworkRuntimeEngine {
        this.stateValue.currentTimeMilliseconds = event.deadlineMilliseconds;
        this.ageAllSwitchSatEntries();
        this.processScheduledRuntimeEvent(event.kind);
+        this.applicationThreads.resume();
        outputs.push(this.output({ type: 'fired', event }));
      }

@@ -1160,6 +1261,12 @@ export class NetworkRuntimeEngine {
      case 'tcpTimeout':
        this.handleTcpTimeout(kind.sessionId);
        return;
+      case 'applicationWake': {
+        const wake = this.applicationWakes.get(kind.wakeId);
+        this.applicationWakes.delete(kind.wakeId);
+        wake?.();
+        return;
+      }
      case 'remoteLinkDelivery':
      case 'lanRemoteLinkTransmission':
        throw new NotPortedError('remote links');
@@ -1667,6 +1774,32 @@ export class NetworkRuntimeEngine {
    return { type: 'resolved', address: findDnsRecords(domainName, 'A', resolution.records)[0]?.data };
  }

+  /** Resolver.holeIPAdresse in a program's thread. */
+  *resolveDnsAddressSteps(nodeId: Id, name: string): Steps<DnsAddressResolution> {
+    const domainName = name.endsWith('.') ? name : `${name}.`;
+    let lookup: DnsLookup | undefined;
+    const server = this.stateValue.topologySnapshot.deviceConfigurations.get(nodeId)?.dnsServer;
+    this.beginDnsResolution(nodeId, domainName, 'A', server, (result) => {
+      lookup = result;
+    });
+    yield* waitUntil(() => lookup !== undefined || this.stateValue.phase !== 'running');
+    const result = lookup as DnsLookup | undefined;
+    if (result === undefined || result.type === 'timedOut') return { type: 'timedOut' };
+    return { type: 'resolved', address: findDnsRecords(domainName, 'A', [...result.answers, ...result.authority])[0]?.data };
+  }
+
+  /**
+   * The address a socket of Java's goes to (Socket's constructor): the
+   * address itself when it is one (IP.ipCheck), else the name's A record;
+   * undefined when there is none or the server did not answer.
+   */
+  *resolveHostSteps(nodeId: Id, address: string): Steps<string | undefined> {
+    const ipAddress = javaIpCheck(address);
+    if (ipAddress !== null) return ipAddress;
+    const resolution = yield* this.resolveDnsAddressSteps(nodeId, address);
+    return resolution.type === 'resolved' ? resolution.address : undefined;
+  }
+
  /** Resolver.holeIPAdresseMailServer: the address of the domain's mail exchange. */
  resolveDnsMailServer(nodeId: Id, domain: string): DnsAddressResolution {
    const domainName = domain.endsWith('.') ? domain : `${domain}.`;
@@ -1818,6 +1951,11 @@ export class NetworkRuntimeEngine {

  /** The three-way handshake from a closed client socket, waiting on the virtual clock as Java blocks. */
  connectTcpWithResult(socketId: Id): TcpConnectionResult {
+    return this.runBlocking(this.connectTcpSteps(socketId));
+  }
+
+  /** The three-way handshake from a closed client socket, in a program's thread (TCPSocket.verbinden). */
+  *connectTcpSteps(socketId: Id): Steps<TcpConnectionResult> {
    const state = this.stateValue;
    const socket = state.socketsById.get(socketId);
    if (
@@ -1839,10 +1977,13 @@ export class NetworkRuntimeEngine {
    ) {
      return 'invalidSocket';
    }
+    // No route to the server: Java's socket thread dies of the
+    // RouteNotFoundException and verbinden waits for ever; the core fails at once.
+    if (this.routeForPacket(socket.nodeId, socket.remoteIPAddress).type === 'unavailable') return 'unreachable';
    this.updateTcpState(socketId, 'SYN_SENT');
    this.resetTcpSendAttempts(socketId);
    const initialDelivery = this.transmitTcpSegment(socketId, TCP_FLAG.synchronize, new Uint8Array(), 0, 'TCP SYN');
-    const waitedForTimeout = this.advanceTcpUntilSettled(socketId, { waitingStates: ['SYN_SENT'] });
+    const waitedForTimeout = yield* this.tcpSettledSteps(socketId, { waitingStates: ['SYN_SENT'] });
    if (state.socketsById.get(socketId)?.tcpState === 'ESTABLISHED') return 'connected';
    if (initialDelivery === undefined || initialDelivery.type !== 'delivered') return 'unreachable';
    return waitedForTimeout ? 'timedOut' : 'unreachable';
@@ -1850,8 +1991,19 @@ export class NetworkRuntimeEngine {

  /** Sends in segments of at most TCP_MAXIMUM_SEGMENT_SIZE bytes, each acknowledged before the next. */
  sendTcp(socketId: Id, payload: Uint8Array): boolean {
+    return this.runBlocking(this.sendTcpSteps(socketId, payload)) === 'sent';
+  }
+
+  /**
+   * sendTcp in a program's thread (TCPSocket.senden): `sent` when every
+   * segment was acknowledged; `notConnected` when the socket was not
+   * connected at the call, `connectionLost` when it stopped being so
+   * meanwhile, `timedOut` when a segment's last attempt went unanswered
+   * (Java's messages sw_tcpsocket_msg6, msg7, msg8).
+   */
+  *sendTcpSteps(socketId: Id, payload: Uint8Array): Steps<TcpSendResult> {
    const state = this.stateValue;
-    if (state.socketsById.get(socketId)?.tcpState !== 'ESTABLISHED') return false;
+    if (state.socketsById.get(socketId)?.tcpState !== 'ESTABLISHED') return 'notConnected';
    const chunks: Uint8Array[] = [];
    if (payload.length === 0) {
      chunks.push(new Uint8Array());
@@ -1861,15 +2013,17 @@ export class NetworkRuntimeEngine {
      }
    }
    for (const [index, chunk] of chunks.entries()) {
-      if (state.socketsById.get(socketId)?.tcpState !== 'ESTABLISHED') return false;
+      if (state.socketsById.get(socketId)?.tcpState !== 'ESTABLISHED') return 'connectionLost';
      this.resetTcpSendAttempts(socketId);
      const flags = index === chunks.length - 1 ? TCP_FLAG.push : 0;
      this.transmitTcpSegment(socketId, flags, chunk, 0, `TCP data segment ${index + 1}/${chunks.length}`);
-      this.advanceTcpUntilSettled(socketId, { waitingForAcknowledgement: true });
+      yield* this.tcpSettledSteps(socketId, { waitingForAcknowledgement: true });
      const session = state.tcpSessionsById.get(socketId);
-      if (session === undefined || session.state !== 'ESTABLISHED' || session.lastSentSegment !== undefined) return false;
+      if (session === undefined || session.state !== 'ESTABLISHED' || session.lastSentSegment !== undefined) {
+        return session?.dataTimedOut === true ? 'timedOut' : 'connectionLost';
+      }
    }
-    return true;
+    return 'sent';
  }

  /** The oldest whole message (up to a PSH) received on the socket. */
@@ -1879,6 +2033,41 @@ export class NetworkRuntimeEngine {
    return session.receivedMessages.shift();
  }

+  /**
+   * TCPSocket.empfangen in a program's thread: the next whole message, as
+   * soon as there is one; `closed` once the connection is no longer
+   * established (Java: null), `notConnected` when it was not at the call
+   * and nothing is left (Java: VerbindungsException). Better than Java: a
+   * message that came before the connection closed is still handed out
+   * (Java drops what its thread had not taken yet).
+   */
+  *receiveTcpSteps(socketId: Id): Steps<TcpReceived> {
+    const messages = () => this.stateValue.tcpSessionsById.get(socketId)?.receivedMessages ?? [];
+    const established = () => this.stateValue.socketsById.get(socketId)?.tcpState === 'ESTABLISHED';
+    if (messages().length === 0 && !established()) return { type: 'notConnected' };
+    yield* waitUntil(() => messages().length > 0 || !established() || this.stateValue.phase !== 'running');
+    const payload = messages().shift();
+    return payload === undefined ? { type: 'closed' } : { type: 'message', payload };
+  }
+
+  /**
+   * ServerSocket.oeffnen in a program's thread: the listener's next
+   * connection, once established; undefined when the listener closes.
+   * Java looks every 100 ms whether its socket is connected; the core at once.
+   */
+  *acceptTcpSteps(listenerId: Id): Steps<Id | undefined> {
+    const state = this.stateValue;
+    const next = () =>
+      (state.tcpAcceptedSocketIdsByListenerId.get(listenerId) ?? []).find(
+        (id) => !this.handedOutTcpSocketIds.has(id) && state.socketsById.get(id)?.tcpState === 'ESTABLISHED',
+      );
+    const listening = () => state.socketsById.get(listenerId)?.tcpState === 'LISTEN';
+    yield* waitUntil(() => next() !== undefined || !listening() || state.phase !== 'running');
+    const accepted = next();
+    if (accepted !== undefined) this.handedOutTcpSocketIds.add(accepted);
+    return accepted;
+  }
+
  closeTcpSocket(socketId: Id): void {
    const socket = this.stateValue.socketsById.get(socketId);
    if (socket === undefined || socket.protocol !== 'tcp' || socket.tcpState === undefined) return;
@@ -2302,6 +2491,7 @@ export class NetworkRuntimeEngine {
      // ends senden with schliessen, an orderly close with FIN. A connection
      // attempt that fails closes at once.
      if (socket.tcpState === 'ESTABLISHED') {
+        session.dataTimedOut = true;
        this.closeTcpSocket(socketId);
      } else {
        this.updateTcpState(socketId, 'CLOSED');
@@ -2345,10 +2535,15 @@ export class NetworkRuntimeEngine {
   * Advances the clock to the socket's timeouts while it waits in one of
   * the states or for an acknowledgement; whether it advanced at all.
   */
-  private advanceTcpUntilSettled(
+  /**
+   * Waits while the socket is in one of `waitingStates`, or while its data
+   * wait for their acknowledgement, as long as an attempt's time runs; true
+   * when it waited for one.
+   */
+  private *tcpSettledSteps(
    socketId: Id,
    options: { waitingStates?: readonly TcpSocketState[]; waitingForAcknowledgement?: boolean },
-  ): boolean {
+  ): Steps<boolean> {
    const waiting = () => {
      const socket = this.stateValue.socketsById.get(socketId);
      if (socket === undefined || socket.tcpState === undefined) return false;
@@ -2361,17 +2556,11 @@ export class NetworkRuntimeEngine {
        this.stateValue.tcpSessionsById.get(socketId)?.lastSentSegment !== undefined;
      return waitingOnState || waitingOnAck;
    };
-    let advancedToTimeout = false;
-    while (this.stateValue.phase === 'running' && waiting()) {
-      const deadlines = this.stateValue.pendingEvents
-        .filter((event) => event.kind.kind === 'tcpTimeout' && event.kind.sessionId === socketId)
-        .map((event) => event.deadlineMilliseconds);
-      if (deadlines.length === 0) return advancedToTimeout;
-      advancedToTimeout = true;
-      // Until the answer comes, or the attempt's time is up.
-      this.advanceUntil(() => !waiting(), Math.min(...deadlines));
-    }
-    return advancedToTimeout;
+    const attemptRuns = () => this.stateValue.pendingEvents.some((event) => event.kind.kind === 'tcpTimeout' && event.kind.sessionId === socketId);
+    if (this.stateValue.phase !== 'running' || !waiting() || !attemptRuns()) return false;
+    // Until the answer comes, or the last attempt's time is up.
+    yield* waitUntil(() => this.stateValue.phase !== 'running' || !waiting() || !attemptRuns());
+    return true;
  }

  private updateTcpState(socketId: Id, newState: TcpSocketState): void {