《游戏引擎实现原理》15.3 与主循环、渲染线程的同步

// file: src/core/frame-sync.ts
// tier: L1
// verify: npx vitest run src/core/frame-sync.test.ts
//
// helio 内核：跨线程同步原语。
// Barrier 让若干线程在每个帧边界对齐；DoubleBuffer 让逻辑线程与渲染线程
// 各写各的、只在帧边界交换一次引用，避免一帧内读到写了一半的数据。

/** 屏障状态占两个字：word0 = 已到达计数，word1 = 代次。 */
export const BARRIER_WORDS = 2;

/** 分配一块可被多个 worker 共享的屏障缓冲区。 */
export function createBarrierBuffer(): SharedArrayBuffer {
  return new SharedArrayBuffer(BARRIER_WORDS * Int32Array.BYTES_PER_ELEMENT);
}

/**
 * 计数屏障（sense-reversing）。parties 个线程各自调用 await()；
 * 最后一个到达者把代次加一，其余线程在 word1 上等待代次变化后一起放行。
 */
export class Barrier {
  readonly #state: Int32Array;
  readonly #parties: number;

  constructor(buffer: SharedArrayBuffer, parties: number) {
    if (parties < 1) throw new RangeError(`parties 必须 >= 1，收到 ${parties}`);
    this.#state = new Int32Array(buffer);
    this.#parties = parties;
  }

  /** 当前代次：已完整通过的轮数。 */
  get generation(): number {
    return Atomics.load(this.#state, 1);
  }

  /** 登记一次到达；返回「我是不是最后一个到的」。不阻塞。 */
  arrive(): boolean {
    const arrived = Atomics.add(this.#state, 0, 1) + 1;
    if (arrived === this.#parties) {
      Atomics.store(this.#state, 0, 0); // 重置计数，供下一轮使用
      Atomics.add(this.#state, 1, 1); // 翻转代次，放行等待者
      Atomics.notify(this.#state, 1); // 唤醒所有等待者
      return true;
    }
    return false;
  }

  /** 阻塞直到本轮所有参与者都到达。 */
  await(): void {
    const gen = Atomics.load(this.#state, 1);
    if (this.arrive()) return;
    // 只在「代次仍等于我读到的值」时阻塞，被唤醒或已翻代则退出。
    while (Atomics.load(this.#state, 1) === gen) {
      Atomics.wait(this.#state, 1, gen);
    }
  }
}

/**
 * 双缓冲。逻辑线程用 write 写 back 槽，渲染线程读 front 槽；
 * 帧边界调用 swap 一次，把写完的 back 发布成 front。
 * 约定：write 必须整体替换槽值，不要就地改已经发布的对象。
 */
export class DoubleBuffer<T> {
  readonly #slots: [T, T];
  #front = 0;
  #back = 1;

  constructor(initial: T) {
    this.#slots = [initial, initial];
  }

  /** 渲染线程读到的、上一帧发布出来的快照。 */
  get front(): T {
    return this.#slots[this.#front];
  }

  /** 逻辑线程正在写的槽（尚未发布）。 */
  get back(): T {
    return this.#slots[this.#back];
  }

  write(value: T): void {
    this.#slots[this.#back] = value;
  }

  /** 帧边界发布：交换 front / back 的角色。 */
  swap(): void {
    const t = this.#front;
    this.#front = this.#back;
    this.#back = t;
  }
}

// file: src/apps/frame-sync-demo.ts
// tier: L1
// verify: node src/apps/frame-sync-demo.ts
//
// 用 worker_threads + Atomics 跑一个最小跨线程同步 demo：
// 主线程是「逻辑线程」，若干 worker 是「消费者」；每帧结束在屏障上对齐，
// 逻辑线程把该帧快照发布到共享槽，消费者在屏障之后读取，永远读到完整帧。
//
// 说明：本工程未装 @types/node，类型由 src/core/thread-shims.d.ts 提供；
// 该 shim 只声明了 Worker，其余运行时成员（isMainThread 等）在这里按需取用。

import { Worker } from 'node:worker_threads';
import * as nodeWorkerThreads from 'node:worker_threads';
import { Barrier, createBarrierBuffer } from '../core/frame-sync.ts';

interface RuntimeMembers {
  isMainThread: boolean;
  parentPort: { postMessage(value: unknown): void } | null;
  workerData: unknown;
}
const { isMainThread, parentPort, workerData } = nodeWorkerThreads as unknown as RuntimeMembers;

const FRAMES = 5;
const WORKERS = 2;
const PARTIES = WORKERS + 1; // 主线程 + 每个 worker 都在同一道屏障上

/** 共享帧槽布局：FRAMES 个槽，每槽两字 [frame, frame*7]。 */
const SLOT_WORDS = 2;

interface SyncPayload {
  barrierBuffer: SharedArrayBuffer;
  stateBuffer: SharedArrayBuffer;
  published: SharedArrayBuffer;
}

function runWorker(payload: SyncPayload): void {
  const barrier = new Barrier(payload.barrierBuffer, PARTIES);
  const state = new Int32Array(payload.stateBuffer);
  const published = new Int32Array(payload.published);
  let consistent = 0;
  for (let f = 0; f < FRAMES; f++) {
    barrier.await(); // 等到逻辑线程发布完本帧
    const pub = Atomics.load(published, 0); // 已发布帧数
    const slot = pub - 1; // 读最近一帧
    const a = Atomics.load(state, slot * SLOT_WORDS);
    const b = Atomics.load(state, slot * SLOT_WORDS + 1);
    if (b === a * 7) consistent++;
  }
  parentPort!.postMessage({ consistent });
}

async function main(): Promise<void> {
  const payload: SyncPayload = {
    barrierBuffer: createBarrierBuffer(),
    stateBuffer: new SharedArrayBuffer(FRAMES * SLOT_WORDS * Int32Array.BYTES_PER_ELEMENT),
    published: new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT),
  };
  const barrier = new Barrier(payload.barrierBuffer, PARTIES);
  const state = new Int32Array(payload.stateBuffer);
  const published = new Int32Array(payload.published);

  const results: Promise<number>[] = [];
  const workers: Worker[] = [];
  for (let i = 0; i < WORKERS; i++) {
    const w = new Worker(new URL(import.meta.url), { workerData: payload });
    workers.push(w);
    results.push(
      new Promise<number>((resolve) => {
        w.on('message', (m) => resolve((m as { consistent: number }).consistent));
      }),
    );
  }

  // 逻辑线程：写 back 槽 -> 发布（published+1）-> 屏障对齐
  for (let f = 0; f < FRAMES; f++) {
    Atomics.store(state, f * SLOT_WORDS, f);
    Atomics.store(state, f * SLOT_WORDS + 1, f * 7);
    Atomics.store(published, 0, f + 1);
    barrier.await();
  }

  const perWorker = await Promise.all(results);
  await Promise.all(workers.map((w) => w.terminate()));
  console.log(`frames=${FRAMES} workers=${WORKERS} parties=${PARTIES}`);
  console.log(`barrier generation=${barrier.generation}（应等于帧数 ${FRAMES}）`);
  console.log(`每个 worker 读到的一致帧数=${perWorker.join(', ')}（应为 ${FRAMES}）`);
  const ok = barrier.generation === FRAMES && perWorker.every((c) => c === FRAMES);
  console.log(ok ? 'SYNC-OK 所有消费者在每帧边界对齐且读到完整帧' : 'SYNC-FAIL');
}

if (isMainThread) {
  await main();
} else {
  runWorker(workerData as SyncPayload);
}

frames=5 workers=2 parties=3
barrier generation=5（应等于帧数 5）
每个 worker 读到的一致帧数=5, 5（应为 5）
SYNC-OK 所有消费者在每帧边界对齐且读到完整帧