Merge branch 'sab' into Atomic-SAB

src/core/GameRunner.ts

@@ -37,6 +37,7 @@ export async function createGameRunner(
   clientID: ClientID,
   mapLoader: GameMapLoader,
   callBack: (gu: GameUpdateViewData | ErrorUpdate) => void,
+  tileUpdateSink?: (update: bigint) => void,
 ): Promise<GameRunner> {
   const config = await getConfig(gameStart.config, null);
   const gameMap = await loadGameMap(
@@ -86,6 +87,7 @@ export async function createGameRunner(
     game,
     new Executor(game, gameStart.gameID, clientID),
     callBack,
+    tileUpdateSink,
   );
   gr.init();
   return gr;
@@ -102,6 +104,7 @@ export class GameRunner {
     public game: Game,
     private execManager: Executor,
     private callBack: (gu: GameUpdateViewData | ErrorUpdate) => void,
+    private tileUpdateSink?: (update: bigint) => void,
   ) {}
 
   init() {
@@ -176,13 +179,24 @@ export class GameRunner {
       });
     }
 
-    // Many tiles are updated to pack it into an array
-    const packedTileUpdates = updates[GameUpdateType.Tile].map((u) => u.update);
+    // Many tiles are updated; either publish them via a shared sink or pack
+    // them into the view data.
+    let packedTileUpdates: BigUint64Array;
+    const tileUpdates = updates[GameUpdateType.Tile];
+    if (this.tileUpdateSink !== undefined) {
+      for (const u of tileUpdates) {
+        this.tileUpdateSink(u.update);
+      }
+      packedTileUpdates = new BigUint64Array();
+    } else {
+      const raw = tileUpdates.map((u) => u.update);
+      packedTileUpdates = new BigUint64Array(raw);
+    }
     updates[GameUpdateType.Tile] = [];
 
     this.callBack({
       tick: this.game.ticks(),
-      packedTileUpdates: new BigUint64Array(packedTileUpdates),
+      packedTileUpdates,
       updates: updates,
       playerNameViewData: this.playerViewData,
       tickExecutionDuration: tickExecutionDuration,
@@ -273,4 +287,8 @@ export class GameRunner {
     }
     return player.bestTransportShipSpawn(targetTile);
   }
+
+  public hasPendingTurns(): boolean {
+    return this.currTurn < this.turns.length;
+  }
 }

src/core/worker/SharedTileRing.ts

@@ -0,0 +1,79 @@
+export interface SharedTileRingBuffers {
+  header: SharedArrayBuffer;
+  data: SharedArrayBuffer;
+}
+
+export interface SharedTileRingViews {
+  header: Int32Array;
+  buffer: BigUint64Array;
+  capacity: number;
+}
+
+// Header indices
+export const TILE_RING_HEADER_WRITE_INDEX = 0;
+export const TILE_RING_HEADER_READ_INDEX = 1;
+export const TILE_RING_HEADER_OVERFLOW = 2;
+
+export function createSharedTileRingBuffers(
+  capacity: number,
+): SharedTileRingBuffers {
+  const header = new SharedArrayBuffer(3 * Int32Array.BYTES_PER_ELEMENT);
+  const data = new SharedArrayBuffer(
+    capacity * BigUint64Array.BYTES_PER_ELEMENT,
+  );
+  return { header, data };
+}
+
+export function createSharedTileRingViews(
+  buffers: SharedTileRingBuffers,
+): SharedTileRingViews {
+  const header = new Int32Array(buffers.header);
+  const buffer = new BigUint64Array(buffers.data);
+  return {
+    header,
+    buffer,
+    capacity: buffer.length,
+  };
+}
+
+export function pushTileUpdate(
+  views: SharedTileRingViews,
+  value: bigint,
+): void {
+  const { header, buffer, capacity } = views;
+
+  const write = Atomics.load(header, TILE_RING_HEADER_WRITE_INDEX);
+  const read = Atomics.load(header, TILE_RING_HEADER_READ_INDEX);
+  const nextWrite = (write + 1) % capacity;
+
+  // If the buffer is full, advance read (drop oldest) and mark overflow.
+  if (nextWrite === read) {
+    Atomics.store(header, TILE_RING_HEADER_OVERFLOW, 1);
+    const nextRead = (read + 1) % capacity;
+    Atomics.store(header, TILE_RING_HEADER_READ_INDEX, nextRead);
+  }
+
+  buffer[write] = value;
+  Atomics.store(header, TILE_RING_HEADER_WRITE_INDEX, nextWrite);
+}
+
+export function drainTileUpdates(
+  views: SharedTileRingViews,
+  maxItems: number,
+  out: bigint[],
+): void {
+  const { header, buffer, capacity } = views;
+
+  let read = Atomics.load(header, TILE_RING_HEADER_READ_INDEX);
+  const write = Atomics.load(header, TILE_RING_HEADER_WRITE_INDEX);
+
+  let count = 0;
+
+  while (read !== write && count < maxItems) {
+    out.push(buffer[read]);
+    read = (read + 1) % capacity;
+    count++;
+  }
+
+  Atomics.store(header, TILE_RING_HEADER_READ_INDEX, read);
+}

src/core/worker/Worker.worker.ts

@@ -2,6 +2,11 @@ import version from "../../../resources/version.txt";
 import { createGameRunner, GameRunner } from "../GameRunner";
 import { FetchGameMapLoader } from "../game/FetchGameMapLoader";
 import { ErrorUpdate, GameUpdateViewData } from "../game/GameUpdates";
+import {
+  createSharedTileRingViews,
+  pushTileUpdate,
+  SharedTileRingViews,
+} from "./SharedTileRing";
 import {
   AttackAveragePositionResultMessage,
   InitializedMessage,
@@ -16,6 +21,8 @@ import {
 const ctx: Worker = self as any;
 let gameRunner: Promise<GameRunner> | null = null;
 const mapLoader = new FetchGameMapLoader(`/maps`, version);
+let isProcessingTurns = false;
+let sharedTileRing: SharedTileRingViews | null = null;
 
 function gameUpdate(gu: GameUpdateViewData | ErrorUpdate) {
   // skip if ErrorUpdate
@@ -32,25 +39,58 @@ function sendMessage(message: WorkerMessage) {
   ctx.postMessage(message);
 }
 
+async function processPendingTurns() {
+  if (isProcessingTurns) {
+    return;
+  }
+  if (!gameRunner) {
+    return;
+  }
+
+  const gr = await gameRunner;
+  if (!gr || !gr.hasPendingTurns()) {
+    return;
+  }
+
+  isProcessingTurns = true;
+  try {
+    while (gr.hasPendingTurns()) {
+      gr.executeNextTick();
+    }
+  } finally {
+    isProcessingTurns = false;
+  }
+}
+
 ctx.addEventListener("message", async (e: MessageEvent<MainThreadMessage>) => {
   const message = e.data;
 
   switch (message.type) {
-    case "heartbeat":
-      (await gameRunner)?.executeNextTick();
-      break;
     case "init":
       try {
+        if (message.sharedTileRingHeader && message.sharedTileRingData) {
+          sharedTileRing = createSharedTileRingViews({
+            header: message.sharedTileRingHeader,
+            data: message.sharedTileRingData,
+          });
+        } else {
+          sharedTileRing = null;
+        }
+
         gameRunner = createGameRunner(
           message.gameStartInfo,
           message.clientID,
           mapLoader,
           gameUpdate,
+          sharedTileRing
+            ? (update: bigint) => pushTileUpdate(sharedTileRing!, update)
+            : undefined,
         ).then((gr) => {
           sendMessage({
             type: "initialized",
             id: message.id,
           } as InitializedMessage);
+          processPendingTurns();
           return gr;
         });
       } catch (error) {
@@ -67,6 +107,7 @@ ctx.addEventListener("message", async (e: MessageEvent<MainThreadMessage>) => {
       try {
         const gr = await gameRunner;
         await gr.addTurn(message.turn);
+        processPendingTurns();
       } catch (error) {
         console.error("Failed to process turn:", error);
         throw error;

src/core/worker/WorkerClient.ts

@@ -9,6 +9,7 @@ import { TileRef } from "../game/GameMap";
 import { ErrorUpdate, GameUpdateViewData } from "../game/GameUpdates";
 import { ClientID, GameStartInfo, Turn } from "../Schemas";
 import { generateID } from "../Util";
+import { SharedTileRingBuffers } from "./SharedTileRing";
 import { WorkerMessage } from "./WorkerMessages";
 
 export class WorkerClient {
@@ -22,6 +23,7 @@ export class WorkerClient {
   constructor(
     private gameStartInfo: GameStartInfo,
     private clientID: ClientID,
+    private sharedTileRingBuffers?: SharedTileRingBuffers,
   ) {
     this.worker = new Worker(new URL("./Worker.worker.ts", import.meta.url));
     this.messageHandlers = new Map();
@@ -70,6 +72,8 @@ export class WorkerClient {
         id: messageId,
         gameStartInfo: this.gameStartInfo,
         clientID: this.clientID,
+        sharedTileRingHeader: this.sharedTileRingBuffers?.header,
+        sharedTileRingData: this.sharedTileRingBuffers?.data,
       });
 
       // Add timeout for initialization
@@ -100,12 +104,6 @@ export class WorkerClient {
     });
   }
 
-  sendHeartbeat() {
-    this.worker.postMessage({
-      type: "heartbeat",
-    });
-  }
-
   playerProfile(playerID: number): Promise<PlayerProfile> {
     return new Promise((resolve, reject) => {
       if (!this.isInitialized) {

src/core/worker/WorkerMessages.ts

@@ -9,7 +9,6 @@ import { GameUpdateViewData } from "../game/GameUpdates";
 import { ClientID, GameStartInfo, Turn } from "../Schemas";
 
 export type WorkerMessageType =
-  | "heartbeat"
   | "init"
   | "initialized"
   | "turn"
@@ -31,15 +30,13 @@ interface BaseWorkerMessage {
   id?: string;
 }
 
-export interface HeartbeatMessage extends BaseWorkerMessage {
-  type: "heartbeat";
-}
-
 // Messages from main thread to worker
 export interface InitMessage extends BaseWorkerMessage {
   type: "init";
   gameStartInfo: GameStartInfo;
   clientID: ClientID;
+  sharedTileRingHeader?: SharedArrayBuffer;
+  sharedTileRingData?: SharedArrayBuffer;
 }
 
 export interface TurnMessage extends BaseWorkerMessage {
@@ -114,7 +111,6 @@ export interface TransportShipSpawnResultMessage extends BaseWorkerMessage {
 
 // Union types for type safety
 export type MainThreadMessage =
-  | HeartbeatMessage
   | InitMessage
   | TurnMessage
   | PlayerActionsMessage