Đưa việc nặng ra khỏi main thread: Web Worker, transferable, SharedArrayBuffer và worker pool

// Đo chi phí gửi ngay trên main thread
const rows = await loadRows(); // ví dụ 200.000 object { id, name, price, tags }

const t0 = performance.now();
worker.postMessage({ type: "index", rows });
console.log(`postMessage chặn main thread ${(performance.now() - t0).toFixed(1)}ms`);

const samples = new Float32Array(8_000_000); // khoảng 32 MB
fillSamples(samples);

worker.postMessage({ type: "analyze", samples }, [samples.buffer]);

console.log(samples.byteLength); // 0 — buffer đã thuộc về worker

const canvas = document.querySelector<HTMLCanvasElement>("#chart")!;
const offscreen = canvas.transferControlToOffscreen();

const worker = new Worker(new URL("./chart.worker.ts", import.meta.url), { type: "module" });
worker.postMessage({ type: "init", canvas: offscreen, dpr: devicePixelRatio }, [offscreen]);

// Sau đó chỉ gửi dữ liệu và sự kiện nhỏ
new ResizeObserver(([entry]) => {
  const { width, height } = entry.contentRect;
  worker.postMessage({ type: "resize", width, height });
}).observe(canvas);

let ctx: OffscreenCanvasRenderingContext2D | null = null;
let canvas: OffscreenCanvas | null = null;
let dpr = 1;

self.onmessage = (event: MessageEvent) => {
  const msg = event.data;
  if (msg.type === "init") {
    canvas = msg.canvas as OffscreenCanvas;
    dpr = msg.dpr;
    ctx = canvas.getContext("2d");
  } else if (msg.type === "resize" && canvas) {
    canvas.width = Math.round(msg.width * dpr);
    canvas.height = Math.round(msg.height * dpr);
    draw();
  }
};

function draw() {
  if (!ctx || !canvas) return;
  ctx.clearRect(0, 0, canvas.width, canvas.height);
  // ... vẽ series
}

Cross-Origin-Opener-Policy: same-origin
Cross-Origin-Embedder-Policy: require-corp

import type { NextConfig } from "next";

const nextConfig: NextConfig = {
  async headers() {
    return [
      {
        // Chỉ trang editor cần SharedArrayBuffer
        source: "/editor/:path*",
        headers: [
          { key: "Cross-Origin-Opener-Policy", value: "same-origin" },
          { key: "Cross-Origin-Embedder-Policy", value: "credentialless" },
        ],
      },
    ];
  },
};

export default nextConfig;

// main.ts — cờ huỷ và tiến độ dùng chung
const control = new Int32Array(new SharedArrayBuffer(8)); // [0] = cờ huỷ, [1] = tiến độ (phần nghìn)
worker.postMessage({ type: "crunch", control, input });

cancelButton.onclick = () => Atomics.store(control, 0, 1);

function renderProgress() {
  progressBar.value = Atomics.load(control, 1) / 1000;
  if (!done) requestAnimationFrame(renderProgress);
}
requestAnimationFrame(renderProgress);

function crunch(control: Int32Array, input: Float64Array) {
  for (let i = 0; i < input.length; i++) {
    if ((i & 0xfff) === 0) {
      if (Atomics.load(control, 0) === 1) throw new DOMException("Cancelled", "AbortError");
      Atomics.store(control, 1, Math.floor((i / input.length) * 1000));
    }
    // ... tính toán nặng, đồng bộ
  }
}

export type Call = { kind: "call"; id: number; method: string; input: unknown };
export type Cancel = { kind: "cancel"; id: number };
export type Progress = { kind: "progress"; id: number; value: number };
export type SerializedError = { name: string; message: string; stack?: string };
export type Result =
  | { kind: "result"; id: number; ok: true; value: unknown }
  | { kind: "result"; id: number; ok: false; error: SerializedError };

export type TaskContext = { signal: AbortSignal; progress(value: number): void };
export type Api = Record<string, (input: any, ctx: TaskContext) => unknown>;

// Đánh dấu giá trị cần transfer thay vì clone
const marked = new WeakMap<object, Transferable[]>();

export function transfer<T extends object>(value: T, list: Transferable[]): T {
  marked.set(value, list);
  return value;
}

export function takeTransfer(value: unknown): Transferable[] {
  if (typeof value !== "object" || value === null) return [];
  const list = marked.get(value) ?? [];
  marked.delete(value);
  return list;
}

/// <reference lib="webworker" />
import { takeTransfer, type Api, type Call, type Cancel, type Result, type TaskContext } from "./protocol";

declare const self: DedicatedWorkerGlobalScope;

export function expose(api: Api) {
  const running = new Map<number, AbortController>();

  self.addEventListener("message", async (event: MessageEvent<Call | Cancel>) => {
    const msg = event.data;
    if (msg.kind === "cancel") {
      running.get(msg.id)?.abort(new DOMException("Cancelled", "AbortError"));
      return;
    }

    const controller = new AbortController();
    running.set(msg.id, controller);
    const ctx: TaskContext = {
      signal: controller.signal,
      progress: (value) => self.postMessage({ kind: "progress", id: msg.id, value }),
    };

    try {
      const fn = api[msg.method];
      if (!fn) throw new Error(`Unknown method "${msg.method}"`);
      const value = await fn(msg.input, ctx);
      self.postMessage({ kind: "result", id: msg.id, ok: true, value } satisfies Result, takeTransfer(value));
    } catch (err) {
      const e = err instanceof Error ? err : new Error(String(err));
      self.postMessage({
        kind: "result",
        id: msg.id,
        ok: false,
        error: { name: e.name, message: e.message, stack: e.stack },
      } satisfies Result);
    } finally {
      running.delete(msg.id);
    }
  });
}

import { takeTransfer, type Api, type Call, type Progress, type Result } from "./protocol";

export type CallOptions = { signal?: AbortSignal; onProgress?: (value: number) => void };
export type Remote<T extends Api> = {
  [K in keyof T]: (input: Parameters<T[K]>[0], opts?: CallOptions) => Promise<Awaited<ReturnType<T[K]>>>;
};

type Pending = { resolve(v: unknown): void; reject(e: unknown): void; onProgress?: (v: number) => void };

export function wrap<T extends Api>(worker: Worker) {
  let nextId = 1;
  const pending = new Map<number, Pending>();

  worker.addEventListener("message", (event: MessageEvent<Result | Progress>) => {
    const msg = event.data;
    const p = pending.get(msg.id);
    if (!p) return; // đã huỷ hoặc đã xong
    if (msg.kind === "progress") return p.onProgress?.(msg.value);
    pending.delete(msg.id);
    if (msg.ok) p.resolve(msg.value);
    else p.reject(Object.assign(new Error(msg.error.message), msg.error));
  });

  const failAll = (reason: unknown) => {
    for (const p of pending.values()) p.reject(reason);
    pending.clear();
  };
  worker.addEventListener("error", (e) => failAll(new Error(`Worker crashed: ${e.message}`)));
  worker.addEventListener("messageerror", () => failAll(new Error("Could not deserialize worker message")));

  const api = new Proxy({} as Remote<T>, {
    get(_target, method) {
      // Không có dòng này, `await api` sẽ coi proxy là thenable và treo mãi
      if (method === "then" || typeof method !== "string") return undefined;

      return (input: unknown, { signal, onProgress }: CallOptions = {}) =>
        new Promise((resolve, reject) => {
          if (signal?.aborted) return reject(signal.reason);
          const id = nextId++;
          const onAbort = () => {
            pending.delete(id);
            worker.postMessage({ kind: "cancel", id });
            reject(signal!.reason);
          };
          signal?.addEventListener("abort", onAbort, { once: true });
          const cleanup = () => signal?.removeEventListener("abort", onAbort);

          pending.set(id, {
            resolve: (v) => (cleanup(), resolve(v)),
            reject: (e) => (cleanup(), reject(e)),
            onProgress,
          });
          worker.postMessage({ kind: "call", id, method, input } satisfies Call, takeTransfer(input));
        });
    },
  });

  const dispose = (reason: unknown = new DOMException("Worker disposed", "AbortError")) => {
    worker.terminate();
    failAll(reason);
  };

  return { api, dispose };
}

import { expose } from "@/lib/rpc/expose";
import type { TaskContext } from "@/lib/rpc/protocol";

const api = {
  async primes(limit: number, { signal, progress }: TaskContext) {
    const found: number[] = [];
    for (let n = 2; n <= limit; n++) {
      if (n % 50_000 === 0) {
        progress(n / limit);
        await new Promise((r) => setTimeout(r)); // nhường để nhận được message huỷ
        signal.throwIfAborted();
      }
      if (isPrime(n)) found.push(n);
    }
    return found.length;
  },
};

export type MathApi = typeof api;
expose(api);

function isPrime(n: number) {
  for (let d = 2; d * d <= n; d++) if (n % d === 0) return false;
  return true;
}

import { wrap } from "@/lib/rpc/wrap";
import type { MathApi } from "@/workers/math.worker";

const { api, dispose } = wrap<MathApi>(
  new Worker(new URL("./workers/math.worker.ts", import.meta.url), { type: "module" }),
);

const controller = new AbortController();
const count = await api.primes(5_000_000, {
  signal: controller.signal,
  onProgress: (p) => (bar.value = p),
});

const worker = new Worker(new URL("./search.worker.ts", import.meta.url), { type: "module" });

import { defineConfig } from "vite";

export default defineConfig({
  worker: {
    // Mặc định là "iife"; "es" cần thiết khi worker dùng dynamic import / code splitting
    format: "es",
  },
});

"use client";

import { useEffect, useRef } from "react";
import { wrap, type Remote } from "@/lib/rpc/wrap";
import type { Api } from "@/lib/rpc/protocol";

export function useWorker<T extends Api>(create: () => Worker) {
  const ref = useRef<Remote<T> | null>(null);

  useEffect(() => {
    const { api, dispose } = wrap<T>(create());
    ref.current = api;
    // StrictMode chạy effect hai lần ở dev — cleanup đảm bảo không rò worker
    return () => {
      ref.current = null;
      dispose();
    };
    // eslint-disable-next-line react-hooks/exhaustive-deps
  }, []);

  return ref;
}

// Sử dụng: pattern new URL phải nằm nguyên văn ở đây để bundler nhận ra
// const search = useWorker<SearchApi>(() =>
//   new Worker(new URL("../workers/search.worker.ts", import.meta.url), { type: "module", name: "search" }),
// );

import { wrap, type Remote } from "./wrap";
import type { Api } from "./protocol";

type Slot<T extends Api> = {
  api: Remote<T>;
  dispose(reason?: unknown): void;
  dead: boolean;
  idleTimer?: ReturnType<typeof setTimeout>;
};

type Job<T extends Api> = {
  method: keyof T & string;
  input: unknown;
  signal?: AbortSignal;
  resolve(value: unknown): void;
  reject(reason: unknown): void;
};

const defaultSize = () => Math.max(1, Math.min(4, (navigator.hardwareConcurrency || 2) - 1));

export class WorkerPool<T extends Api> {
  private idle: Slot<T>[] = [];
  private queue: Job<T>[] = [];
  private live = 0;

  constructor(
    private spawn: () => Worker,
    private max = defaultSize(),
    private idleMs = 30_000,
  ) {}

  run<K extends keyof T & string>(
    method: K,
    input: Parameters<T[K]>[0],
    { signal }: { signal?: AbortSignal } = {},
  ): Promise<Awaited<ReturnType<T[K]>>> {
    return new Promise((resolve, reject) => {
      if (signal?.aborted) return reject(signal.reason);
      const job: Job<T> = { method, input, signal, resolve: resolve as Job<T>["resolve"], reject };
      // Việc còn trong hàng đợi: chỉ cần rút ra
      signal?.addEventListener(
        "abort",
        () => {
          const i = this.queue.indexOf(job);
          if (i !== -1) {
            this.queue.splice(i, 1);
            reject(signal.reason);
          }
        },
        { once: true },
      );
      this.queue.push(job);
      this.pump();
    });
  }

  private pump() {
    while (this.queue.length) {
      const slot = this.idle.pop() ?? (this.live < this.max ? this.create() : undefined);
      if (!slot) return;
      clearTimeout(slot.idleTimer);
      void this.execute(slot, this.queue.shift()!);
    }
  }

  private create(): Slot<T> {
    this.live++;
    const { api, dispose } = wrap<T>(this.spawn());
    return { api, dispose, dead: false };
  }

  private async execute(slot: Slot<T>, job: Job<T>) {
    // Việc đang chạy: cách huỷ chắc chắn duy nhất là giết worker
    const onAbort = () => {
      slot.dead = true;
      slot.dispose(job.signal!.reason);
    };
    job.signal?.addEventListener("abort", onAbort, { once: true });
    try {
      job.resolve(await slot.api[job.method](job.input as never));
    } catch (err) {
      job.reject(err);
    } finally {
      job.signal?.removeEventListener("abort", onAbort);
      if (slot.dead) this.live--;
      else this.release(slot);
      this.pump();
    }
  }

  private release(slot: Slot<T>) {
    slot.idleTimer = setTimeout(() => {
      this.idle = this.idle.filter((s) => s !== slot);
      this.live--;
      slot.dispose();
    }, this.idleMs);
    this.idle.push(slot);
  }

  destroy() {
    for (const slot of this.idle) slot.dispose();
    this.idle = [];
    for (const job of this.queue.splice(0)) job.reject(new DOMException("Pool destroyed", "AbortError"));
  }
}

let current: AbortController | null = null;

export async function search(q: string) {
  current?.abort(new DOMException("Superseded", "AbortError"));
  const controller = (current = new AbortController());
  try {
    return await searchApi.query({ q, limit: 20 }, { signal: controller.signal });
  } catch (err) {
    if (err instanceof DOMException && err.name === "AbortError") return null; // bị thay thế, không phải lỗi
    throw err;
  }
}

import { expose } from "@/lib/rpc/expose";
import { transfer, type TaskContext } from "@/lib/rpc/protocol";

type Summary = { rows: number; revenueByCountry: Map<string, number>; amounts: Float64Array };

const api = {
  async summarize(file: File, { signal, progress }: TaskContext): Promise<Summary> {
    const reader = file.stream().pipeThrough(new TextDecoderStream()).getReader();
    const revenueByCountry = new Map<string, number>();
    let amounts = new Float64Array(1 << 16);
    let rows = 0;
    let carry = "";
    let header: string[] | null = null;
    let readChars = 0;
    let lastReported = 0;

    for (;;) {
      const { value, done } = await reader.read();
      signal.throwIfAborted();
      if (done) break;

      readChars += value.length;
      const lines = (carry + value).split("\n");
      carry = lines.pop() ?? "";

      for (const line of lines) {
        // Ví dụ đơn giản: không xử lý dấu phẩy trong ngoặc kép.
        // Dữ liệu thật nên dùng một parser CSV đầy đủ, vẫn chạy trong worker.
        const cols = line.split(",");
        if (!header) {
          header = cols;
          continue;
        }
        const amount = Number(cols[3]);
        if (rows === amounts.length) {
          const bigger = new Float64Array(amounts.length * 2);
          bigger.set(amounts);
          amounts = bigger;
        }
        amounts[rows++] = amount;
        revenueByCountry.set(cols[2], (revenueByCountry.get(cols[2]) ?? 0) + amount);
      }

      const ratio = readChars / file.size; // xấp xỉ: ký tự so với byte
      if (ratio - lastReported > 0.01) {
        progress(Math.min(ratio, 1));
        lastReported = ratio;
      }
    }

    const result = { rows, revenueByCountry, amounts: amounts.slice(0, rows) };
    return transfer(result, [result.amounts.buffer]);
  },
};

export type CsvApi = typeof api;
expose(api);

import MiniSearch from "minisearch";
import { expose } from "@/lib/rpc/expose";

type Doc = { id: string; title: string; body: string; url: string };
let index: MiniSearch<Doc> | null = null;
const titles = new Map<string, { title: string; url: string }>();

const api = {
  async load(url: string) {
    // fetch và JSON.parse ngay trong worker: main thread không bao giờ chạm vào vài chục MB JSON
    const docs: Doc[] = await (await fetch(url)).json();
    index = new MiniSearch<Doc>({ fields: ["title", "body"], storeFields: [] });
    index.addAll(docs);
    for (const d of docs) titles.set(d.id, { title: d.title, url: d.url });
    return docs.length;
  },

  query({ q, limit }: { q: string; limit: number }) {
    if (!index) throw new Error("Index not loaded");
    return index
      .search(q, { prefix: true, fuzzy: 0.2 })
      .slice(0, limit)
      .map((hit) => ({ id: hit.id as string, score: hit.score, ...titles.get(hit.id as string)! }));
  },
};

export type SearchApi = typeof api;
expose(api);

import { expose } from "@/lib/rpc/expose";

const api = {
  async thumbnail({ file, maxSize }: { file: Blob; maxSize: number }) {
    const bitmap = await createImageBitmap(file);
    const scale = Math.min(1, maxSize / Math.max(bitmap.width, bitmap.height));
    const canvas = new OffscreenCanvas(Math.round(bitmap.width * scale), Math.round(bitmap.height * scale));
    const ctx = canvas.getContext("2d")!;
    ctx.imageSmoothingQuality = "high";
    ctx.drawImage(bitmap, 0, 0, canvas.width, canvas.height);
    bitmap.close();
    return canvas.convertToBlob({ type: "image/webp", quality: 0.85 });
  },
};

export type ImageApi = typeof api;
expose(api);

const pool = new WorkerPool<ImageApi>(
  () => new Worker(new URL("./workers/image.worker.ts", import.meta.url), { type: "module" }),
);

const thumbs = await Promise.all(files.map((file) => pool.run("thumbnail", { file, maxSize: 1600 })));

import { expose } from "@/lib/rpc/expose";
import { transfer } from "@/lib/rpc/protocol";

type Exports = {
  memory: WebAssembly.Memory;
  alloc(len: number): number;
  free(ptr: number, len: number): void;
  blur(ptr: number, width: number, height: number, radius: number): void;
};

// Compile một lần khi worker khởi động; server phải trả Content-Type: application/wasm
const ready = WebAssembly.instantiateStreaming(fetch(new URL("./image_ops.wasm", import.meta.url))).then(
  ({ instance }) => instance.exports as unknown as Exports,
);

const api = {
  async blur({ pixels, width, height, radius }: { pixels: Uint8ClampedArray; width: number; height: number; radius: number }) {
    const wasm = await ready;
    const len = pixels.byteLength;
    const ptr = wasm.alloc(len);
    // Đọc lại memory.buffer SAU khi alloc: memory có thể grow và buffer cũ bị detached
    new Uint8ClampedArray(wasm.memory.buffer, ptr, len).set(pixels);
    wasm.blur(ptr, width, height, radius);
    const out = new Uint8ClampedArray(wasm.memory.buffer.slice(ptr, ptr + len));
    wasm.free(ptr, len);
    return transfer(out, [out.buffer]);
  },
};

export type WasmApi = typeof api;
expose(api);

type SchedulerLike = {
  yield?: () => Promise<void>;
  postTask?: <T>(cb: () => T, opts?: { priority?: string; signal?: AbortSignal; delay?: number }) => Promise<T>;
};
const scheduler = (globalThis as { scheduler?: SchedulerLike }).scheduler;

export function yieldToMain(): Promise<void> {
  // scheduler.yield() nhường lượt nhưng phần tiếp theo được ưu tiên chạy trước các task khác
  if (scheduler?.yield) return scheduler.yield();
  return new Promise((resolve) => setTimeout(resolve, 0));
}

export async function processInChunks<T>(items: T[], fn: (item: T) => void, budgetMs = 8) {
  let deadline = performance.now() + budgetMs;
  for (const item of items) {
    fn(item);
    if (performance.now() > deadline) {
      await yieldToMain();
      deadline = performance.now() + budgetMs;
    }
  }
}

if (scheduler?.postTask) {
  const controller = new AbortController();
  scheduler.postTask(() => warmUpCaches(), { priority: "background", signal: controller.signal });
  // rời trang hoặc không còn cần: controller.abort()
} else {
  globalThis.requestIdleCallback?.(() => warmUpCaches());
}