feat(core): sends return at once and go through Java's transport thread

A send hands its frame to the card and returns, as in Java; the frames travel while the clock advances. Swift's send ran the network until every frame had arrived; the option swiftSends keeps that for the Swift comparison. Waiting calls (TCP connect, acknowledgement) now run event by event until the answer instead of to the timeout. UDP and TCP send through a thread per node and protocol (Java's TransportProtokoll): one segment after the other; while IP waits for ARP, the next send blocks and a segment from an event queues behind it. Senders that wait take their turns in the order they came (a fair lock; Java's leaves the order to chance). All four ArpRun stories now match Java FILIUS, including when each call returns. Tests let frames arrive explicitly with letFramesArrive. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-30 17:49:05 +00:00
parent dbbc1fd70d
commit a88fa1512c
7 changed files with 343 additions and 71 deletions
@@ -20,6 +20,11 @@
 // ARP after Java's holeARPTabellenEintrag: a lookup asks twice, a round trip
 //   each, then host unreachable (Swift: all at once, or without end when the
 //   address exists anywhere; the option swiftArp keeps that for the comparison).
+// A send hands its frame to the card and returns, as in Java; the frames
+//   travel while the clock advances (Swift: the send returns when every frame
+//   in flight has arrived; the option swiftSends keeps that for the comparison).
+// Java's transport send thread: per node and protocol, one segment after the
+//   other; while IP waits for ARP, a socket's next send blocks.
 // DNS after Java (filius.software.dns: DNSServer, DNSServerMitarbeiter,
 //   Resolver, DNSQueryAgent), not Swift's typed format; recursion not yet.
 // Java counterparts: filius.hardware.Verbindung, filius.software.netzzugangsschicht,
@@ -542,6 +547,19 @@ interface ArpLookup {
  readonly done: (macAddress: string | undefined) => void;
 }

+/**
+ * Java's TransportProtokoll send thread, one per node and protocol: it takes
+ * the segments one after the other and hands each to IP, which waits there
+ * for ARP.
+ */
+interface TransportSender {
+  busy: boolean;
+  /** How many segments it has taken, and how many are out. */
+  started: number;
+  finished: number;
+  readonly queue: ((finished: () => void) => void)[];
+}
+
 /** What a sender learns when its packet has left, or when it could not. */
 interface IPv4SendCallbacks {
  /** The packet left (or failed) — at once, or when ARP has answered or given up. */
@@ -589,10 +607,13 @@ export class NetworkRuntimeEngine {
  /** ARP lookups waiting for their answer (Java: a thread in holeARPTabellenEintrag). */
  private arpLookups: ArpLookup[] = [];
  private nextArpLookupId = 1;
+  /** Per node and transport protocol (`nodeId/udp`, `nodeId/tcp`). */
+  private transportSenders = new Map<string, TransportSender>();
  /** Per cable end (node and port): when the last frame sent there arrives. */
  private cableBusyUntilByPort = new Map<string, number>();
  private readonly cablesCarryOneFrameAtATime: boolean;
  private readonly swiftArp: boolean;
+  private readonly swiftSends: boolean;

  /**
   * `swiftCables`: a cable carries all frames at once, as in the Swift
@@ -600,14 +621,18 @@ export class NetworkRuntimeEngine {
   * frame after the other (SimplexVerbindung sleeps the delay with the port's
   * buffer locked); a frame waits in the output buffer until the one before
   * has arrived, and shows in the packet log when it leaves.
+   *
+   * `swiftArp`, `swiftSends`: Swift's ARP and Swift's send that returns
+   * when the network is quiet, likewise only for the comparison.
   */
  constructor(
    seed: bigint | number = DEFAULT_SEED,
    packetTraceRetentionPolicy: TraceRetentionPolicy = PRODUCTION_TRACE_RETENTION,
-    options: { readonly swiftCables?: boolean; readonly swiftArp?: boolean } = {},
+    options: { readonly swiftCables?: boolean; readonly swiftArp?: boolean; readonly swiftSends?: boolean } = {},
  ) {
    this.cablesCarryOneFrameAtATime = options.swiftCables !== true;
    this.swiftArp = options.swiftArp === true;
+    this.swiftSends = options.swiftSends === true;
    const normalizedSeed = toSeed(seed);
    this.stateValue = initialState(normalizedSeed);
    this.random = new JavaCompatibleRandom(normalizedSeed);
@@ -669,6 +694,7 @@ export class NetworkRuntimeEngine {
        state.globalPacketLossEnabled = false;
        this.isProcessingScheduledEvents = false;
        this.arpLookups = [];
+        this.transportSenders.clear();
        state.pendingEvents = [];
        state.arpCachesByNodeId.clear();
        state.switchForwardingTablesByNodeId.clear();
@@ -1054,20 +1080,25 @@ export class NetworkRuntimeEngine {
        afterHeaders: udpHeaders(datagram),
      });

+    const handedOn: IPv4DeliveryResult = { type: 'delivered', packetIdentity, nodeId: socket.nodeId };
    if (destinationIPAddress === LIMITED_BROADCAST_IP_ADDRESS) {
-      for (const networkInterface of boundInterface === undefined ? interfaces : [boundInterface]) {
-        const packet: IPv4Packet = {
-          identity: packetIdentity,
-          senderIPAddress: networkInterface.ipAddress,
-          receiverIPAddress: destinationIPAddress,
-          timeToLive: 1,
-          protocol: 'udp',
-          payload: { kind: 'udp', datagram },
-        };
-        traceSent(networkInterface, packet.identity);
-        this.sendLimitedBroadcastPacket(socket.nodeId, networkInterface, packet);
-      }
-      return { type: 'delivered', packetIdentity, nodeId: socket.nodeId };
+      const broadcastInterfaces = boundInterface === undefined ? interfaces : [boundInterface];
+      for (const networkInterface of broadcastInterfaces) traceSent(networkInterface, packetIdentity);
+      return this.handToTransportSender(socket.nodeId, 'udp', handedOn, (finished) => {
+        for (const networkInterface of broadcastInterfaces) {
+          const packet: IPv4Packet = {
+            identity: packetIdentity,
+            senderIPAddress: networkInterface.ipAddress,
+            receiverIPAddress: destinationIPAddress,
+            timeToLive: 1,
+            protocol: 'udp',
+            payload: { kind: 'udp', datagram },
+          };
+          this.sendLimitedBroadcastPacket(socket.nodeId, networkInterface, packet);
+        }
+        finished();
+        return handedOn;
+      });
    }

    let networkInterface: NetworkInterface;
@@ -1086,7 +1117,57 @@ export class NetworkRuntimeEngine {
      payload: { kind: 'udp', datagram },
    };
    traceSent(networkInterface, packet.identity);
-    return this.sendIPv4Packet(socket.nodeId, packet, networkInterface.portId);
+    const portId = networkInterface.portId;
+    return this.handToTransportSender(socket.nodeId, 'udp', handedOn, (finished) =>
+      this.sendIPv4PacketWithOptions(socket.nodeId, packet, portId, true, 'sent', { onFinished: finished }),
+    );
+  }
+
+  /**
+   * Hands a segment to the node's send thread for the protocol (Java's
+   * TransportProtokoll.senden). The thread holds its list while IP waits for
+   * ARP: a socket's send then blocks until the segment in hand is out, as the
+   * caller does here; from within an event, the segment queues behind it.
+   * The senders that wait take their turns in the order they came, before
+   * the thread takes the next — as with a fair lock; Java's is not fair and
+   * leaves the order to chance. The result is the send's own when the thread
+   * takes it at once, else that it was handed on.
+   */
+  private handToTransportSender(
+    nodeId: Id,
+    protocol: 'udp' | 'tcp',
+    handedOn: IPv4DeliveryResult,
+    send: (finished: () => void) => IPv4DeliveryResult,
+  ): IPv4DeliveryResult {
+    const sender = getOrCreate(this.transportSenders, `${nodeId}/${protocol}`, () => ({ busy: false, started: 0, finished: 0, queue: [] }));
+    if (sender.busy && !this.isProcessingScheduledEvents) {
+      // The segment in hand is out once ARP has answered or given up.
+      const inHand = sender.started;
+      const deadline = this.stateValue.currentTimeMilliseconds + (ARP_ATTEMPTS + 1) * this.runtimeSpeed.roundTripTimeMilliseconds;
+      this.advanceUntil(() => sender.finished >= inHand, deadline);
+      if (this.stateValue.phase !== 'running' || this.transportSenders.get(`${nodeId}/${protocol}`) !== sender) return handedOn;
+    }
+    let result = handedOn;
+    sender.queue.push((finished) => {
+      result = send(finished);
+    });
+    if (!sender.busy) this.startNextTransportSend(sender);
+    return result;
+  }
+
+  private startNextTransportSend(sender: TransportSender): void {
+    const next = sender.queue.shift();
+    if (next === undefined) return;
+    sender.busy = true;
+    sender.started += 1;
+    let out = false;
+    next(() => {
+      if (out) return;
+      out = true;
+      sender.finished += 1;
+      sender.busy = false;
+      this.startNextTransportSend(sender);
+    });
  }

  private sendLimitedBroadcastPacket(fromNodeId: Id, networkInterface: NetworkInterface, packet: IPv4Packet): void {
@@ -1689,9 +1770,6 @@ export class NetworkRuntimeEngine {
    if (socket === undefined || socket.remoteIPAddress === undefined || socket.remotePort === undefined || session === undefined) {
      return undefined;
    }
-    // Java's socket starts waiting when it hands the segment down, before
-    // ARP has found the next hop; the timeout counts from here.
-    const handedDownAt = state.currentTimeMilliseconds;
    let segment: TcpSegment;
    if (retransmission && session.lastSentSegment !== undefined) {
      segment = session.lastSentSegment;
@@ -1738,9 +1816,16 @@ export class NetworkRuntimeEngine {
      afterHeaders: tcpHeaders(segment),
      detail: retransmission ? `${detail} retransmission attempt ${session.sendAttempts}` : detail,
    });
-    const delivery = this.sendIPv4Packet(socket.nodeId, packet, networkInterface.portId);
-    if (state.tcpSessionsById.get(socketId)?.lastSentSegment !== undefined) this.scheduleTcpTimeoutIfNeeded(socketId, handedDownAt);
-    return delivery;
+    const portId = networkInterface.portId;
+    const handedOn: IPv4DeliveryResult = { type: 'delivered', packetIdentity: packet.identity, nodeId: socket.nodeId };
+    return this.handToTransportSender(socket.nodeId, 'tcp', handedOn, (finished) => {
+      // Java's socket starts waiting when it has handed the segment down,
+      // before ARP has found the next hop; the timeout counts from here.
+      const handedDownAt = state.currentTimeMilliseconds;
+      const delivery = this.sendIPv4PacketWithOptions(socket.nodeId, packet, portId, true, 'sent', { onFinished: finished });
+      if (state.tcpSessionsById.get(socketId)?.lastSentSegment !== undefined) this.scheduleTcpTimeoutIfNeeded(socketId, handedDownAt);
+      return delivery;
+    });
  }

  private scheduleTcpTimeoutIfNeeded(socketId: Id, fromMilliseconds: number): void {
@@ -1826,10 +1911,9 @@ export class NetworkRuntimeEngine {
    socketId: Id,
    options: { waitingStates?: readonly TcpSocketState[]; waitingForAcknowledgement?: boolean },
  ): boolean {
-    let advancedToTimeout = false;
-    while (this.stateValue.phase === 'running') {
+    const waiting = () => {
      const socket = this.stateValue.socketsById.get(socketId);
-      if (socket === undefined || socket.tcpState === undefined) return advancedToTimeout;
+      if (socket === undefined || socket.tcpState === undefined) return false;
      const waitingOnState = options.waitingStates?.includes(socket.tcpState) === true;
      // Data waits for its acknowledgement only while connected: once the
      // socket closes (Java: senden gives up and calls schliessen), it returns.
@@ -1837,13 +1921,17 @@ export class NetworkRuntimeEngine {
        options.waitingForAcknowledgement === true &&
        socket.tcpState === 'ESTABLISHED' &&
        this.stateValue.tcpSessionsById.get(socketId)?.lastSentSegment !== undefined;
-      if (!waitingOnState && !waitingOnAck) return advancedToTimeout;
+      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;
-      this.advance(Math.min(...deadlines));
+      // Until the answer comes, or the attempt's time is up.
+      this.advanceUntil(() => !waiting(), Math.min(...deadlines));
    }
    return advancedToTimeout;
  }
@@ -2895,8 +2983,14 @@ export class NetworkRuntimeEngine {
    this.drainPendingNetworkDeliveryEventsIfNeeded();
  }

+  /**
+   * Swift's send, for the comparison: a frame sent from outside the events
+   * runs the network until every frame in flight has arrived. Java's send
+   * hands the frame to the card and returns; the frames travel while the
+   * clock advances.
+   */
  private drainPendingNetworkDeliveryEventsIfNeeded(): void {
-    if (this.isProcessingScheduledEvents) return;
+    if (this.isProcessingScheduledEvents || !this.swiftSends) return;
    const maximumEvents = 16_384;
    let processedEventCount = 0;
    while (processedEventCount < maximumEvents) {