feat: use bytes for packet queue (#170)

This PR extracts the packet queue logic into a new class, to avoid
breaking its invariants

Author: MegaRedHand

src/packets/icmp.ts

@@ -13,9 +13,7 @@ export const ICMP_REPLY_TYPE_NUMBER = 0;
 //  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 //  |     Type      |     Code      |          Checksum             |
 //  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-//  |                             unused                            |
-//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-//  |      Internet Header + 64 bits of Original Data Datagram      |
+//  |                         variant data                          |
 //  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 abstract class IcmpPacket implements IpPayload {
   // 8 bits
@@ -33,6 +31,11 @@ abstract class IcmpPacket implements IpPayload {
     return ICMP_PROTOCOL_NUMBER;
   }
 
+  byteLength(): number {
+    const headerLength = 8;
+    return headerLength + this._dataToBytes().length;
+  }
+
   toBytes(): Uint8Array {
     const data = this._dataToBytes();
     const buffer = Uint8Array.from([
@@ -60,6 +63,15 @@ abstract class IcmpPacket implements IpPayload {
   abstract getDetails(layer: Layer): Record<string, string | number | object>;
 }
 
+//   0                   1                   2                   3
+//   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+//  |     Type      |     Code      |          Checksum             |
+//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+//  |           Identifier          |        Sequence Number        |
+//  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+//  |     Data ...
+//  +-+-+-+-+-
 class EchoMessage extends IcmpPacket {
   code = 0;
 

src/packets/ip.ts

@@ -7,6 +7,9 @@ export const TCP_PROTOCOL_NUMBER = 6;
 export const UDP_PROTOCOL_NUMBER = 17;
 
 export class EmptyPayload implements IpPayload {
+  byteLength() {
+    return 0;
+  }
   toBytes() {
     return new Uint8Array(0);
   }
@@ -118,6 +121,8 @@ export class IpAddressGenerator {
 }
 
 export interface IpPayload {
+  // Length of the payload in bytes
+  byteLength(): number;
   // The bytes equivalent of the payload
   toBytes(): Uint8Array;
   // The number of the protocol
@@ -162,7 +167,8 @@ export class IPv4Packet implements FramePayload {
   // Length of the datagram, in bytes
   // 16 bits
   get totalLength() {
-    return this.payload.toBytes().length + this.internetHeaderLength * 4;
+    const headerSize = this.internetHeaderLength * 4;
+    return headerSize + this.payload.byteLength();
   }
 
   // Identifying value assigned by the sender

src/packets/tcp.ts

@@ -162,10 +162,6 @@ export class TcpSegment implements IpPayload {
     this.data = data;
   }
 
-  getPacketType(): string {
-    return "TCP";
-  }
-
   computeChecksum(): number {
     const segmentBytes = this.toBytes({ withChecksum: false });
 
@@ -180,12 +176,15 @@ export class TcpSegment implements IpPayload {
     return computeIpChecksum(totalBytes);
   }
 
-  // Dummy Method for the moment
-  getDetails(layer: Layer) {
-    return { Layer: layer };
+  // ### IpPayload ###
+
+  byteLength(): number {
+    const headerSize = 5 /* number of rows */ * 4; /* bytes per row */
+    // TODO: include options
+    // const optionsSize = 0;
+    return headerSize + this.data.length;
   }
 
-  // ### IpPayload ###
   toBytes({
     withChecksum = true,
   }: { withChecksum?: boolean } = {}): Uint8Array {
@@ -207,6 +206,16 @@ export class TcpSegment implements IpPayload {
   protocol(): number {
     return TCP_PROTOCOL_NUMBER;
   }
+
+  getPacketType(): string {
+    return "TCP";
+  }
+
+  // Dummy Method for the moment
+  getDetails(layer: Layer) {
+    return { Layer: layer };
+  }
+
   // ### IpPayload ###
 }
 

src/types/devices/router.ts

@@ -14,14 +14,11 @@ import { sendRawPacket } from "../packet";
 export class Router extends NetworkDevice {
   static DEVICE_TEXTURE: Texture;
 
-  private processingPackets = false;
+  private packetQueue = new PacketQueue(1024);
+  // Time in ms to process a single byte
+  private timePerByte = 8;
+  // Number of bytes processed
   private processingProgress = 0;
-  // Time in ms to process a single packet
-  private timePerPacket = 250;
-
-  private packetQueue: IPv4Packet[] = [];
-  // TODO: we should count this in bytes
-  private maxQueueSize = 5;
 
   static getTexture() {
     if (!Router.DEVICE_TEXTURE) {
@@ -69,25 +66,21 @@ export class Router extends NetworkDevice {
   }
 
   addPacketToQueue(datagram: IPv4Packet) {
-    if (this.packetQueue.length >= this.maxQueueSize) {
+    const wasEmpty = this.packetQueue.isEmpty();
+    if (!this.packetQueue.enqueue(datagram)) {
       console.debug("Packet queue full, dropping packet");
       return;
     }
-    this.packetQueue.push(datagram);
-    // Start packet processor if not already running
-    if (!this.processingPackets) {
-      Ticker.shared.add(this.processPacket, this);
-      this.processingPackets = true;
+    if (wasEmpty) {
+      this.startPacketProcessor();
     }
   }
 
   processPacket(ticker: Ticker) {
-    this.processingProgress += ticker.deltaMS;
-    if (this.processingProgress < this.timePerPacket) {
+    const datagram = this.getPacketsToProcess(ticker.deltaMS);
+    if (!datagram) {
       return;
     }
-    this.processingProgress -= this.timePerPacket;
-    const datagram = this.packetQueue.pop();
     const devices = this.routePacket(datagram);
 
     if (!devices || devices.length === 0) {
@@ -104,13 +97,30 @@ export class Router extends NetworkDevice {
       sendRawPacket(this.viewgraph, this.id, newFrame);
     }
 
-    // Stop processing packets if queue is empty
-    if (this.packetQueue.length === 0) {
-      Ticker.shared.remove(this.processPacket, this);
-      this.processingPackets = false;
-      this.processingProgress = 0;
-      return;
+    if (this.packetQueue.isEmpty()) {
+      this.stopPacketProcessor();
+    }
+  }
+
+  startPacketProcessor() {
+    this.processingProgress = 0;
+    Ticker.shared.add(this.processPacket, this);
+  }
+
+  stopPacketProcessor() {
+    this.processingProgress = 0;
+    Ticker.shared.remove(this.processPacket, this);
+  }
+
+  getPacketsToProcess(timeMs: number): IPv4Packet | null {
+    this.processingProgress += timeMs;
+    const packetLength = this.packetQueue.getHead()?.totalLength;
+    const progressNeeded = this.timePerByte * packetLength;
+    if (this.processingProgress < progressNeeded) {
+      return null;
     }
+    this.processingProgress -= progressNeeded;
+    return this.packetQueue.dequeue();
   }
 
   routePacket(datagram: IPv4Packet): DeviceId[] {
@@ -147,3 +157,39 @@ export class Router extends NetworkDevice {
     return devices;
   }
 }
+
+class PacketQueue {
+  private queue: IPv4Packet[] = [];
+  private queueSizeBytes = 0;
+  private maxQueueSizeBytes: number;
+
+  constructor(maxQueueSizeBytes: number) {
+    this.maxQueueSizeBytes = maxQueueSizeBytes;
+  }
+
+  enqueue(packet: IPv4Packet) {
+    if (this.queueSizeBytes >= this.maxQueueSizeBytes) {
+      return false;
+    }
+    this.queue.push(packet);
+    this.queueSizeBytes += packet.totalLength;
+    return true;
+  }
+
+  dequeue(): IPv4Packet | undefined {
+    if (this.queue.length === 0) {
+      return;
+    }
+    const packet = this.queue.shift();
+    this.queueSizeBytes -= packet.totalLength;
+    return packet;
+  }
+
+  getHead(): IPv4Packet | undefined {
+    return this.queue[0];
+  }
+
+  isEmpty(): boolean {
+    return this.queue.length === 0;
+  }
+}