[parallel] replace worktank

package.json

@@ -56,7 +56,6 @@
     "tiny-readdir": "^3.1.1",
     "tiny-readdir-glob": "^2.0.0",
     "tiny-spinner": "^2.0.5",
-    "worktank": "^3.0.2",
     "zeptomatch": "^2.1.0",
     "zeptomatch-escape": "^1.0.1",
     "zeptomatch-is-static": "^1.0.1"
@@ -78,4 +77,4 @@
     "typescript": "^6.0.3",
     "zeptoid": "^1.0.2"
   }
-}
+}

src/config_prettier.ts

@@ -139,8 +139,7 @@ const resolvePrettierConfigs = (configs: PrettierConfigWithOverrides[], filePath
     if (overrides) {
       for (let oi = 0, ol = overrides.length; oi < ol; oi++) {
         const override = overrides[oi];
-        const filePathRelative =
-          fastRelativeChildPath(override.folder, filePath) ?? normalizePathSeparatorsToPosix(path.relative(override.folder, filePath));
+        const filePathRelative = fastRelativeChildPath(override.folder, filePath) ?? normalizePathSeparatorsToPosix(path.relative(override.folder, filePath));
         if (!zeptomatch(override.filesPositive, filePathRelative)) continue;
         if (zeptomatch(override.filesNegative, filePathRelative)) continue;
         resolved = { ...resolved, ...override.options };

src/prettier_parallel.ts

@@ -1,21 +1,10 @@
 import os from "node:os";
-import process from "node:process";
-import WorkTank from "worktank";
 import { resolve } from "./utils.js";
 import type { Options, Prettier } from "./types.js";
+import WorkerPool from "./worker_pool.js";
 
 function makeParallel(options: Options): Prettier {
-  const pool = new WorkTank<Prettier>({
-    pool: {
-      name: "prettier",
-      size: options.parallelWorkers || Math.max(1, os.cpus().length - 1),
-    },
-    worker: {
-      autoInstantiate: true,
-      env: process.env,
-      methods: new URL("./prettier_serial.js", import.meta.url),
-    },
-  });
+  const pool = new WorkerPool(new URL("./prettier_worker.js", import.meta.url), options.parallelWorkers || Math.max(1, os.cpus().length - 1));
 
   return {
     async check(filePath, fileContent, formatOptions, contextOptions, pluginsDefaultOptions, pluginsCustomOptions) {

src/prettier_worker.ts

@@ -0,0 +1,26 @@
+import { parentPort } from "node:worker_threads";
+import type { Prettier, PrettierAsyncTask } from "./types.js";
+
+if (!parentPort) {
+  throw new Error("Parent port missing in parallel runner");
+}
+
+let prettierSerial: typeof import("./prettier_serial.js");
+parentPort.on("message", async (value: Pick<PrettierAsyncTask<keyof Prettier>, "method" | "args">) => {
+  try {
+    if (!prettierSerial) {
+      prettierSerial = await import(import.meta.resolve("./prettier_serial.js", import.meta.url));
+    }
+
+    const method = prettierSerial[value.method];
+    if (!method) {
+      throw new Error(`Unknown prettier method: [${value.method}]`);
+    }
+
+    const result = await (method as (...args: unknown[]) => unknown).apply(undefined, value.args as unknown[]);
+    parentPort!.postMessage({ result });
+  } catch (error) {
+    const { name, message, stack } = error instanceof Error ? error : new Error(typeof error === "string" ? error : "Unknown error");
+    parentPort!.postMessage({ error: { name, message, stack } });
+  }
+});

src/types.ts

@@ -95,6 +95,24 @@ type PrettierPlugin = import("prettier").Plugin;
 
 type PromiseMaybe<T> = T | Promise<T>;
 
+type SerializedError = {
+  name: string;
+  message: string;
+  stack?: string;
+};
+
+type PrettierAsyncTask<
+  T extends keyof Prettier = keyof Prettier,
+  U extends Parameters<Prettier[T]> = Parameters<Prettier[T]>,
+  R extends Awaited<ReturnType<Prettier[T]>> = Awaited<ReturnType<Prettier[T]>>,
+> = {
+  method: T;
+  args: U;
+  deferred?: Promise<R>;
+  resolve?: (value: R) => void;
+  reject?: (reason: unknown) => void;
+};
+
 export type {
   Bin,
   ContextOptions,
@@ -111,4 +129,6 @@ export type {
   PrettierConfigWithOverrides,
   PrettierPlugin,
   PromiseMaybe,
+  PrettierAsyncTask,
+  SerializedError,
 };

src/worker_pool.ts

@@ -0,0 +1,158 @@
+import { Worker, SHARE_ENV } from "node:worker_threads";
+import type { Prettier, PrettierAsyncTask, SerializedError } from "./types.js";
+
+type WorkerResult = {
+  result?: Awaited<ReturnType<Prettier[keyof Prettier]>>;
+  error?: SerializedError;
+};
+
+class WorkerPool {
+  private readonly queue = new Set<PrettierAsyncTask>();
+  private readonly workers = new Set<Worker>();
+  private readonly idleWorkers = new Set<Worker>();
+  private readonly workerState = new WeakMap<Worker, PrettierAsyncTask>();
+  private terminated = false;
+
+  constructor(
+    private readonly workerScript: URL,
+    private readonly maxPoolSize = 1,
+  ) {
+    for (let i = 0; i < this.maxPoolSize; i++) {
+      this.createWorker();
+    }
+  }
+
+  get isTerminated() {
+    return this.terminated;
+  }
+
+  get queueSize() {
+    return this.queue.size;
+  }
+
+  get poolSize() {
+    return this.workers.size;
+  }
+
+  destroy() {
+    this.terminated = true;
+
+    for (const task of this.queue) {
+      task.reject!(new Error("Worker pool terminated"));
+    }
+    this.queue.clear();
+
+    for (const worker of this.workers) {
+      const task = this.workerState.get(worker);
+      task?.reject!(new Error("Worker pool terminated"));
+      worker.terminate();
+    }
+    this.idleWorkers.clear();
+    this.workers.clear();
+  }
+
+  async exec<T extends keyof Prettier>(method: T, args: Parameters<Prettier[T]>): Promise<Awaited<ReturnType<Prettier[T]>>> {
+    if (this.isTerminated) {
+      throw new Error("Worker pool destroyed");
+    }
+    const task: PrettierAsyncTask<T> = { method, args };
+
+    const deferred = new Promise<Awaited<ReturnType<Prettier[T]>>>((resolve, reject) => {
+      task.resolve = resolve;
+      task.reject = reject;
+    });
+
+    this.assignTask(task as unknown as PrettierAsyncTask);
+
+    return deferred;
+  }
+
+  private assignTask(task: PrettierAsyncTask, worker?: Worker): void {
+    if (!worker) {
+      worker = this.idleWorkers.values().next().value;
+    }
+    if (!worker && this.poolSize < this.maxPoolSize) {
+      this.createWorker();
+      worker = this.idleWorkers.values().next().value;
+    }
+
+    if (!worker) {
+      this.queue.add(task);
+    } else {
+      this.idleWorkers.delete(worker);
+      this.queue.delete(task);
+      this.workerState.set(worker, task);
+      worker.postMessage({
+        method: task.method,
+        args: task.args,
+      });
+    }
+  }
+
+  private freeWorker(worker: Worker): void {
+    const task = this.queue.values().next().value;
+    if (task) {
+      this.assignTask(task, worker);
+    } else {
+      this.idleWorkers.add(worker);
+    }
+  }
+
+  private createWorker(): void {
+    if (this.isTerminated || this.poolSize >= this.maxPoolSize) {
+      return;
+    }
+
+    const worker = new Worker(this.workerScript, {
+      env: SHARE_ENV,
+      name: "prettier",
+    });
+    worker.on("message", (message: WorkerResult) => {
+      this.handleWorkerMessage(worker, message);
+    });
+    worker.on("error", (error: unknown) => {
+      this.handleWorkerDeath(worker, error instanceof Error ? error : new Error(String(error)));
+    });
+    worker.on("exit", (code) => {
+      this.handleWorkerDeath(worker, new Error(`Worker exited unexpectedly with exit code ${code}`));
+    });
+
+    this.workers.add(worker);
+    this.idleWorkers.add(worker);
+  }
+
+  private handleWorkerMessage(worker: Worker, message: WorkerResult): void {
+    const task = this.workerState.get(worker);
+    this.workerState.delete(worker);
+
+    if (task) {
+      if (message.error !== undefined) {
+        task.reject!(Object.assign(new Error(message.error.message), message.error));
+      } else {
+        task.resolve!(message.result as Awaited<ReturnType<Prettier[keyof Prettier]>>);
+      }
+    }
+
+    this.freeWorker(worker);
+  }
+
+  private handleWorkerDeath(worker: Worker, error: Error): void {
+    if (!this.workers.has(worker)) {
+      return;
+    }
+
+    const task = this.workerState.get(worker);
+    this.workerState.delete(worker);
+    task?.reject!(error);
+
+    this.idleWorkers.delete(worker);
+    this.workers.delete(worker);
+
+    const nextTask = this.queue.values().next().value;
+    if (nextTask) {
+      this.assignTask(nextTask);
+    }
+  }
+}
+
+export default WorkerPool;