julian.kiedaisch/mdg-filius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): Java's IP and ICMP threads, measured with Java FILIUS

Browse Source 
Every packet that comes in goes through its node's IP thread (UDP, TCP) or
ICMP thread (ICMP), as Java's IPThread and ICMPThread: deliver, answer an
echo request or forward, one after the other; while one waits for ARP, the
next wait behind it. A packet a node sends to itself is delivered at once,
as Java's IP.senden does. The transport send threads and these are one
mechanism now (handToProtocolThread).

ArpRun has three stories through the router (routedUdpBehind,
routedPingBesideUdp, routedPingBehindPing); the oracle writes ICMP frames
with type, code, addresses and sequence number. All seven ArpRun stories
match Java FILIUS.

Measured and not taken: Java's ARP wakes one waiter with notify, not the
one whose answer came; the core wakes all that wait for the address.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
julian.kiedaisch
2026-09-30 18:32:48 +00:00
co-authored by Claude Opus 5.5
parent 7e0348636c
commit f5c61cdce7
13 changed files with 950 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/runtime/engine.ts
+96 -58
View File
@@ -24,8 +24,10 @@
// 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.
// Java's threads: per node a send thread for UDP and for TCP, the IP thread
// (delivers or forwards UDP and TCP) and the ICMP thread (answers, delivers
// or forwards ICMP); each takes one unit after the other and holds the rest
// while it waits for ARP. A socket's send blocks behind its thread.
// 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,
@@ -557,13 +559,17 @@ interface ArpLookup {
}
/**
* 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.
* Java's threads that each take one unit after the other and wait for ARP
* with it: per node the send thread of each transport protocol
* (TransportProtokoll: `udp`, `tcp`), the IP thread (IPThread: delivers or
* forwards every UDP and TCP packet that comes in, `ip`) and the ICMP thread
* (ICMPThread: answers, delivers or forwards every ICMP packet, `icmp`).
*/
interface TransportSender {
type ProtocolThreadKind = 'udp' | 'tcp' | 'ip' | 'icmp';
interface ProtocolThread {
busy: boolean;
/** How many segments it has taken, and how many are out. */
/** How many units it has taken, and how many it is done with. */
started: number;
finished: number;
readonly queue: ((finished: () => void) => void)[];
@@ -616,8 +622,8 @@ 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 node and kind (`nodeId/udp`, `nodeId/icmp`, …). */
private protocolThreads = new Map<string, ProtocolThread>();
/** Per cable end (node and port): when the last frame sent there arrives. */
private cableBusyUntilByPort = new Map<string, number>();
private readonly cablesCarryOneFrameAtATime: boolean;
@@ -703,7 +709,7 @@ export class NetworkRuntimeEngine {
state.globalPacketLossEnabled = false;
this.isProcessingScheduledEvents = false;
this.arpLookups = [];
this.transportSenders.clear();
this.protocolThreads.clear();
state.pendingEvents = [];
state.arpCachesByNodeId.clear();
state.switchForwardingTablesByNodeId.clear();
@@ -1093,7 +1099,7 @@ export class NetworkRuntimeEngine {
if (destinationIPAddress === LIMITED_BROADCAST_IP_ADDRESS) {
const broadcastInterfaces = boundInterface === undefined ? interfaces : [boundInterface];
for (const networkInterface of broadcastInterfaces) traceSent(networkInterface, packetIdentity);
return this.handToTransportSender(socket.nodeId, 'udp', handedOn, (finished) => {
return this.handToProtocolThread(socket.nodeId, 'udp', handedOn, (finished) => {
for (const networkInterface of broadcastInterfaces) {
const packet: IPv4Packet = {
identity: packetIdentity,
@@ -1127,55 +1133,51 @@ export class NetworkRuntimeEngine {
};
traceSent(networkInterface, packet.identity);
const portId = networkInterface.portId;
return this.handToTransportSender(socket.nodeId, 'udp', handedOn, (finished) =>
return this.handToProtocolThread(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.
* Hands a unit to one of the node's threads (ProtocolThreadKind). The
* thread holds its list while it waits for ARP. A socket's send then blocks
* until the unit in hand is done, as the caller does here
* (TransportProtokoll.senden); from within an event, the unit queues
* behind it. Those 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 unit's own when
* the thread takes it at once, else `handedOn`.
*/
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;
private handToProtocolThread<T>(nodeId: Id, kind: ProtocolThreadKind, handedOn: T, unit: (finished: () => void) => T): T {
const key = `${nodeId}/${kind}`;
const thread = getOrCreate(this.protocolThreads, key, () => ({ busy: false, started: 0, finished: 0, queue: [] }));
if (thread.busy && !this.isProcessingScheduledEvents) {
// The unit in hand is done once ARP has answered or given up.
const inHand = thread.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;
this.advanceUntil(() => thread.finished >= inHand, deadline);
if (this.stateValue.phase !== 'running' || this.protocolThreads.get(key) !== thread) return handedOn;
}
let result = handedOn;
sender.queue.push((finished) => {
result = send(finished);
thread.queue.push((finished) => {
result = unit(finished);
});
if (!sender.busy) this.startNextTransportSend(sender);
if (!thread.busy) this.startNextProtocolUnit(thread);
return result;
}
private startNextTransportSend(sender: TransportSender): void {
const next = sender.queue.shift();
private startNextProtocolUnit(thread: ProtocolThread): void {
const next = thread.queue.shift();
if (next === undefined) return;
sender.busy = true;
sender.started += 1;
let out = false;
thread.busy = true;
thread.started += 1;
let done = false;
next(() => {
if (out) return;
out = true;
sender.finished += 1;
sender.busy = false;
this.startNextTransportSend(sender);
if (done) return;
done = true;
thread.finished += 1;
thread.busy = false;
this.startNextProtocolUnit(thread);
});
}
@@ -1881,7 +1883,7 @@ export class NetworkRuntimeEngine {
});
const portId = networkInterface.portId;
const handedOn: IPv4DeliveryResult = { type: 'delivered', packetIdentity: packet.identity, nodeId: socket.nodeId };
return this.handToTransportSender(socket.nodeId, 'tcp', handedOn, (finished) => {
return this.handToProtocolThread(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;
@@ -3134,7 +3136,7 @@ export class NetworkRuntimeEngine {
this.receiveArpPacket(frame.payload.packet, frame.identity, nodeId, networkInterface);
return;
case 'ipv4':
this.receiveIPv4Packet(frame.payload.packet, frame.identity, nodeId, networkInterface);
this.receiveIPv4Packet(frame.payload.packet, frame.identity, nodeId, networkInterface, true);
return;
}
}
@@ -3581,7 +3583,7 @@ export class NetworkRuntimeEngine {
case 'local': {
const networkInterface = route.interface ?? this.networkInterfaces(fromNodeId)[0];
if (networkInterface === undefined) return { type: 'dropped', packetIdentity: packet.identity, nodeId: fromNodeId };
this.receiveIPv4Packet(packet, undefined, fromNodeId, networkInterface);
this.receiveIPv4Packet(packet, undefined, fromNodeId, networkInterface, false);
return { type: 'delivered', packetIdentity: packet.identity, nodeId: fromNodeId };
}
@@ -3636,11 +3638,17 @@ export class NetworkRuntimeEngine {
this.sendEthernetFrame(fromNodeId, networkInterface.portId, frame);
}
/**
* A packet that came in on a card goes to the node's IP or ICMP thread
* (`throughThread`); one the node sent to itself is delivered at once, as
* Java's IP.senden does.
*/
private receiveIPv4Packet(
packet: IPv4Packet,
frameIdentity: number | undefined,
nodeId: Id,
networkInterface: NetworkInterface,
throughThread: boolean,
): void {
this.recordTrace({
frameIdentity,
@@ -3661,6 +3669,29 @@ export class NetworkRuntimeEngine {
const decision = this.evaluateFirewall(packet, nodeId, networkInterface.portId, frameIdentity);
if (!decision.accepted) return;
}
if (throughThread) {
this.handToProtocolThread(nodeId, packet.protocol === 'icmp' ? 'icmp' : 'ip', undefined, (finished) =>
this.processIPv4Packet(packet, frameIdentity, nodeId, networkInterface, finished),
);
} else {
this.processIPv4Packet(packet, frameIdentity, nodeId, networkInterface, () => {});
}
}
/**
* What Java's IPThread and ICMPThread do with a packet: deliver it, answer
* an echo request, or forward it. `finished` when the thread is free again:
* at once, or when the answer or the forwarded packet has left (or ARP gave
* up on it).
*/
private processIPv4Packet(
packet: IPv4Packet,
frameIdentity: number | undefined,
nodeId: Id,
networkInterface: NetworkInterface,
finished: () => void,
): void {
const node = this.nodeSnapshot(nodeId)!;
const packetForProcessing = packet;
const localAddresses = new Set(this.networkInterfaces(nodeId).map((candidate) => candidate.ipAddress));
@@ -3673,9 +3704,12 @@ export class NetworkRuntimeEngine {
this.stateValue.topologySnapshot.firewallConfigurationsByNodeId.has(nodeId)
) {
const decision = this.evaluateFirewall(packetForProcessing, nodeId, networkInterface.portId, frameIdentity);
if (!decision.accepted) return;
if (!decision.accepted) {
finished();
return;
}
}
this.deliverIPv4Locally(packetForProcessing, nodeId, networkInterface);
if (!this.deliverIPv4Locally(packetForProcessing, nodeId, networkInterface, finished)) finished();
return;
}
if (node.kind !== 'router') {
@@ -3690,10 +3724,12 @@ export class NetworkRuntimeEngine {
beforeHeaders: ipv4Headers(packetForProcessing),
detail: 'IPv4 forwarding disabled',
});
finished();
return;
}
if (packetForProcessing.timeToLive <= 1) {
this.emitIcmpError(nodeId, networkInterface, 'timeExceeded', packet);
finished();
return;
}
@@ -3717,20 +3753,22 @@ export class NetworkRuntimeEngine {
onUnresolved: () => {
if (echoOrNotIcmp) this.emitIcmpError(nodeId, networkInterface, 'destinationHostUnreachable', packet);
},
onFinished: finished,
});
if (result.type === 'dropped' && this.routeForPacket(nodeId, packetForProcessing.receiverIPAddress).type === 'unavailable') {
this.emitIcmpError(nodeId, networkInterface, 'destinationNetworkUnreachable', packet);
}
}
private deliverIPv4Locally(packet: IPv4Packet, nodeId: Id, networkInterface: NetworkInterface): void {
/** Delivers the packet; true when an answer goes out, which calls `finished` once it has left. */
private deliverIPv4Locally(packet: IPv4Packet, nodeId: Id, networkInterface: NetworkInterface, finished: () => void): boolean {
const payload = packet.payload;
switch (payload.kind) {
case 'icmp': {
const message = payload.message;
switch (message.kind) {
case 'echoRequest': {
if (packet.receiverIPAddress === LIMITED_BROADCAST_IP_ADDRESS && packet.timeToLive !== 1) return;
if (packet.receiverIPAddress === LIMITED_BROADCAST_IP_ADDRESS && packet.timeToLive !== 1) return false;
const reply: IPv4Packet = {
identity: this.allocatePacketIdentity(),
senderIPAddress: networkInterface.ipAddress,
@@ -3756,8 +3794,8 @@ export class NetworkRuntimeEngine {
afterHeaders: ipv4Headers(reply),
detail: 'ICMP echo reply',
});
this.sendIPv4PacketWithOptions(nodeId, reply, undefined, false, 'sent');
return;
this.sendIPv4PacketWithOptions(nodeId, reply, undefined, false, 'sent', { onFinished: finished });
return true;
}
case 'echoReply':
case 'destinationNetworkUnreachable':
@@ -3779,9 +3817,9 @@ export class NetworkRuntimeEngine {
beforeHeaders: icmpHeaders(message),
detail: 'ICMP observation queued',
});
return;
return false;
}
return;
return false;
}
case 'udp':
getOrCreate(this.stateValue.deliveredIPv4PacketsByNodeId, nodeId, () => []).push({
@@ -3790,7 +3828,7 @@ export class NetworkRuntimeEngine {
packet,
});
this.deliverUdp(packet, payload.datagram, nodeId, networkInterface);
return;
return false;
case 'tcp':
getOrCreate(this.stateValue.deliveredIPv4PacketsByNodeId, nodeId, () => []).push({
nodeId,
@@ -3798,7 +3836,7 @@ export class NetworkRuntimeEngine {
packet,
});
this.deliverTcp(packet, payload.segment, nodeId, networkInterface);
return;
return false;
}
}