diff --git a/package-lock.json b/package-lock.json index 3ffb4d5..6c28c7f 100644 --- a/package-lock.json +++ b/package-lock.json @@ -29,7 +29,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" @@ -3788,11 +3787,6 @@ "url": "https://opencollective.com/express" } }, - "node_modules/immediato": { - "version": "1.1.0", - "resolved": "https://registry.npmjs.org/immediato/-/immediato-1.1.0.tgz", - "integrity": "sha512-6DTWQWiM3SyxAbNRDmMvFgZVwVP6wT8ciQv7GivxXejtXZFIcemC0Wlzfd/jEouJ2JroCIp4qZVloKW4BviUpQ==" - }, "node_modules/import-local": { "version": "3.2.0", "resolved": "https://registry.npmjs.org/import-local/-/import-local-3.2.0.tgz", @@ -4034,19 +4028,6 @@ "integrity": "sha512-RHxMLp9lnKHGHRng9QFhRCMbYAcVpn69smSGcq3f36xjgVVWThj4qqLbTLlq7Ssj8B+fIQ1EuCEGI2lKsyQeIw==", "dev": true }, - "node_modules/isoconcurrency": { - "version": "1.0.0", - "resolved": "https://registry.npmjs.org/isoconcurrency/-/isoconcurrency-1.0.0.tgz", - "integrity": "sha512-YhuPf5V6uOtQQHt9gIkOTbq75ceXqraDvxtZZeS/XbNsre6fmM+WpJgNTSkGX5jB3+gnbwoTVqW1c3qdfyVpOA==" - }, - "node_modules/isotimer": { - "version": "1.0.0", - "resolved": "https://registry.npmjs.org/isotimer/-/isotimer-1.0.0.tgz", - "integrity": "sha512-1p1wborMl9fFbulXx9YBpIqFnfUn/2tN8Ne9g3GLMaiQAPmN/wLlpNOKCNT822div3Sq7LKkApZJ+6JipDUusQ==", - "dependencies": { - "immediato": "^1.1.0" - } - }, "node_modules/issue-parser": { "version": "7.0.2", "resolved": "https://registry.npmjs.org/issue-parser/-/issue-parser-7.0.2.tgz", @@ -6847,11 +6828,6 @@ "url": "https://github.com/sponsors/sxzz" } }, - "node_modules/webworker-shim": { - "version": "1.1.4", - "resolved": "https://registry.npmjs.org/webworker-shim/-/webworker-shim-1.1.4.tgz", - "integrity": "sha512-W/40L5W6ZQyGhYr3hJ7N/2SjdK5OdFtnYm94j6xlRyjckegXnIGwz0EdxdkQx6VGTglJjK8mqBhMz3fd3AY4bg==" - }, "node_modules/when-exit": { "version": "2.1.5", "resolved": "https://registry.npmjs.org/when-exit/-/when-exit-2.1.5.tgz", @@ -6893,17 +6869,6 @@ "url": "https://github.com/sponsors/sindresorhus" } }, - "node_modules/worktank": { - "version": "3.0.2", - "resolved": "https://registry.npmjs.org/worktank/-/worktank-3.0.2.tgz", - "integrity": "sha512-ry5gPtWnakOnUBAAa2aiyWZwAFJuBtd/MwZH6o9DXnQHD4AZvidtl2uTLrb2d3Zjy9D04n84lHJNnIETQl7tuA==", - "dependencies": { - "isoconcurrency": "^1.0.0", - "isotimer": "^1.0.0", - "promise-make-naked": "^3.0.2", - "webworker-shim": "^1.1.4" - } - }, "node_modules/wrap-ansi": { "version": "8.1.0", "resolved": "https://registry.npmjs.org/wrap-ansi/-/wrap-ansi-8.1.0.tgz", @@ -9712,11 +9677,6 @@ "safer-buffer": ">= 2.1.2 < 3.0.0" } }, - "immediato": { - "version": "1.1.0", - "resolved": "https://registry.npmjs.org/immediato/-/immediato-1.1.0.tgz", - "integrity": "sha512-6DTWQWiM3SyxAbNRDmMvFgZVwVP6wT8ciQv7GivxXejtXZFIcemC0Wlzfd/jEouJ2JroCIp4qZVloKW4BviUpQ==" - }, "import-local": { "version": "3.2.0", "resolved": "https://registry.npmjs.org/import-local/-/import-local-3.2.0.tgz", @@ -9870,19 +9830,6 @@ "integrity": "sha512-RHxMLp9lnKHGHRng9QFhRCMbYAcVpn69smSGcq3f36xjgVVWThj4qqLbTLlq7Ssj8B+fIQ1EuCEGI2lKsyQeIw==", "dev": true }, - "isoconcurrency": { - "version": "1.0.0", - "resolved": "https://registry.npmjs.org/isoconcurrency/-/isoconcurrency-1.0.0.tgz", - "integrity": "sha512-YhuPf5V6uOtQQHt9gIkOTbq75ceXqraDvxtZZeS/XbNsre6fmM+WpJgNTSkGX5jB3+gnbwoTVqW1c3qdfyVpOA==" - }, - "isotimer": { - "version": "1.0.0", - "resolved": "https://registry.npmjs.org/isotimer/-/isotimer-1.0.0.tgz", - "integrity": "sha512-1p1wborMl9fFbulXx9YBpIqFnfUn/2tN8Ne9g3GLMaiQAPmN/wLlpNOKCNT822div3Sq7LKkApZJ+6JipDUusQ==", - "requires": { - "immediato": "^1.1.0" - } - }, "issue-parser": { "version": "7.0.2", "resolved": "https://registry.npmjs.org/issue-parser/-/issue-parser-7.0.2.tgz", @@ -11952,11 +11899,6 @@ "integrity": "sha512-u4J1hr6Y+C/hit34GwL3eIcUf3wngT8TlYUjK7tk0jxydKLAq3sYvifjdTDA6U3KCKRcQCQDGf/srjxfZF3l7g==", "dev": true }, - "webworker-shim": { - "version": "1.1.4", - "resolved": "https://registry.npmjs.org/webworker-shim/-/webworker-shim-1.1.4.tgz", - "integrity": "sha512-W/40L5W6ZQyGhYr3hJ7N/2SjdK5OdFtnYm94j6xlRyjckegXnIGwz0EdxdkQx6VGTglJjK8mqBhMz3fd3AY4bg==" - }, "when-exit": { "version": "2.1.5", "resolved": "https://registry.npmjs.org/when-exit/-/when-exit-2.1.5.tgz", @@ -11986,17 +11928,6 @@ "powershell-utils": "^0.2.0" } }, - "worktank": { - "version": "3.0.2", - "resolved": "https://registry.npmjs.org/worktank/-/worktank-3.0.2.tgz", - "integrity": "sha512-ry5gPtWnakOnUBAAa2aiyWZwAFJuBtd/MwZH6o9DXnQHD4AZvidtl2uTLrb2d3Zjy9D04n84lHJNnIETQl7tuA==", - "requires": { - "isoconcurrency": "^1.0.0", - "isotimer": "^1.0.0", - "promise-make-naked": "^3.0.2", - "webworker-shim": "^1.1.4" - } - }, "wrap-ansi": { "version": "8.1.0", "resolved": "https://registry.npmjs.org/wrap-ansi/-/wrap-ansi-8.1.0.tgz", diff --git a/package.json b/package.json index 2ca2eaf..8856797 100644 --- a/package.json +++ b/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" } -} +} \ No newline at end of file diff --git a/src/config_prettier.ts b/src/config_prettier.ts index b09e4d6..d6d0c7a 100644 --- a/src/config_prettier.ts +++ b/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 }; diff --git a/src/prettier_parallel.ts b/src/prettier_parallel.ts index 25c47d0..26b1527 100644 --- a/src/prettier_parallel.ts +++ b/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({ - 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) { diff --git a/src/prettier_worker.ts b/src/prettier_worker.ts new file mode 100644 index 0000000..c71de8b --- /dev/null +++ b/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, "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 } }); + } +}); diff --git a/src/types.ts b/src/types.ts index d809838..4903e25 100644 --- a/src/types.ts +++ b/src/types.ts @@ -95,6 +95,24 @@ type PrettierPlugin = import("prettier").Plugin; type PromiseMaybe = T | Promise; +type SerializedError = { + name: string; + message: string; + stack?: string; +}; + +type PrettierAsyncTask< + T extends keyof Prettier = keyof Prettier, + U extends Parameters = Parameters, + R extends Awaited> = Awaited>, +> = { + method: T; + args: U; + deferred?: Promise; + resolve?: (value: R) => void; + reject?: (reason: unknown) => void; +}; + export type { Bin, ContextOptions, @@ -111,4 +129,6 @@ export type { PrettierConfigWithOverrides, PrettierPlugin, PromiseMaybe, + PrettierAsyncTask, + SerializedError, }; diff --git a/src/worker_pool.ts b/src/worker_pool.ts new file mode 100644 index 0000000..25baf58 --- /dev/null +++ b/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>; + error?: SerializedError; +}; + +class WorkerPool { + private readonly queue = new Set(); + private readonly workers = new Set(); + private readonly idleWorkers = new Set(); + private readonly workerState = new WeakMap(); + 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(method: T, args: Parameters): Promise>> { + if (this.isTerminated) { + throw new Error("Worker pool destroyed"); + } + const task: PrettierAsyncTask = { method, args }; + + const deferred = new Promise>>((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>); + } + } + + 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; diff --git a/test/__tests__/__snapshots__/parallel.js.snap b/test/__tests__/__snapshots__/parallel.js.snap new file mode 100644 index 0000000..9b7bfd4 --- /dev/null +++ b/test/__tests__/__snapshots__/parallel.js.snap @@ -0,0 +1,6 @@ +// Jest Snapshot v1, https://jestjs.io/docs/snapshot-testing + +exports[`parallel mode keeps the error name, so ignore-unknown still works (stdout) 1`] = ` +"javascript.js +override.as-js-file" +`; diff --git a/test/__tests__/parallel.js b/test/__tests__/parallel.js index 6caf213..e7b7221 100644 --- a/test/__tests__/parallel.js +++ b/test/__tests__/parallel.js @@ -33,3 +33,32 @@ describe("files are processed when an explicit worker count is given", () => { write: [], }); }); + +describe("parallel mode works when an environment variable contains a replacement pattern", () => { + runCli("write", [ + "--list-different", + "formatted.js", + "unformatted.js", + "unformatted2.js", + ], { + parallel: true, + env: { PRETTIER_CLI_TEST_ENV: "abc$`def" }, + }).test({ + stdout: "unformatted.js\nunformatted2.js", + stderr: "", + status: 1, + write: [], + }); +}); + +describe("parallel mode keeps the error name, so ignore-unknown still works", () => { + runCli( + "ignore-unknown", + [".", "--ignore-unknown", "--list-different"], + { parallel: true }, + ).test({ + status: "non-zero", + stderr: "", + write: [], + }); +}); diff --git a/test/__tests__/worker-pool.js b/test/__tests__/worker-pool.js new file mode 100644 index 0000000..c880ebd --- /dev/null +++ b/test/__tests__/worker-pool.js @@ -0,0 +1,128 @@ +import path from "node:path"; +import { writeFileSync, mkdtempSync } from "node:fs"; +import os from "node:os"; +import { pathToFileURL } from "node:url"; +import WorkerPool from "../../dist/worker_pool.js"; + +// Settles to { value } | { error } | { hung: true }, so a hanging task fails the test instead of the whole run +const settle = (promise) => { + let timer; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve({ hung: true }), 5000); + }); + return Promise.race([promise.then((value) => ({ value }), (error) => ({ error })), timeout]).finally(() => clearTimeout(timer)); +}; + +const makeScript = (source) => { + const file = path.join(mkdtempSync(path.join(os.tmpdir(), "worker-pool-")), "worker.mjs"); + writeFileSync(file, source); + return pathToFileURL(file); +}; + +const MISSING_SCRIPT = pathToFileURL(path.join(os.tmpdir(), "worker-pool-missing", "worker.js")); + +const SCRIPTS = { + // check -> false, anything else -> an error with a custom name + results: ` + import { parentPort } from "node:worker_threads"; + parentPort.on("message", (m) => { + if (m.method === "check") parentPort.postMessage({ result: false }); + else parentPort.postMessage({ error: { name: "UndefinedParserError", message: "nope" } }); + }); + `, + // check -> the worker exits, anything else -> "ok" + crashOnCheck: ` + import { parentPort } from "node:worker_threads"; + parentPort.on("message", (m) => { + if (m.method === "check") process.exit(1); + parentPort.postMessage({ result: "ok" }); + }); + `, + // check -> throws something that is not an Error + throwsNonError: ` + import { parentPort } from "node:worker_threads"; + parentPort.on("message", () => { + throw "boom"; + }); + `, + // never replies + silent: ` + import { parentPort } from "node:worker_threads"; + parentPort.on("message", () => {}); + `, +}; + +describe("WorkerPool", () => { + const pools = []; + const makePool = (script, size = 1) => { + const pool = new WorkerPool(script, size); + pools.push(pool); + return pool; + }; + + // Live worker threads keep the test process from exiting + afterEach(() => { + while (pools.length) pools.pop().destroy(); + }); + + test("resolves falsy results and rejects with the original error name and message", async () => { + const pool = makePool(makeScript(SCRIPTS.results)); + expect(await settle(pool.exec("check", []))).toEqual({ value: false }); + const { error } = await settle(pool.exec("format", [])); + expect(error.name).toBe("UndefinedParserError"); + expect(error.message).toBe("nope"); + }); + + test("a crashing worker only fails its own task and queued tasks still run", async () => { + const pool = makePool(makeScript(SCRIPTS.crashOnCheck)); + const results = await Promise.all([settle(pool.exec("check", [])), settle(pool.exec("format", [])), settle(pool.exec("format", []))]); + expect(results[0].error?.message).toMatch(/exit code 1/); + expect(results[1]).toEqual({ value: "ok" }); + expect(results[2]).toEqual({ value: "ok" }); + }); + + test("a worker that throws a non-Error value still rejects the task with an Error", async () => { + const pool = makePool(makeScript(SCRIPTS.throwsNonError)); + const result = await settle(pool.exec("format", [])); + expect(result.hung).toBeUndefined(); + expect(result.error).toBeInstanceOf(Error); + expect(result.error.message).toMatch(/boom/); + }); + + test("a worker that fails to start rejects every task, with a message, without hanging", async () => { + const pool = makePool(MISSING_SCRIPT); + const results = await Promise.all([settle(pool.exec("format", [])), settle(pool.exec("format", []))]); + for (const result of results) { + expect(result.hung).toBeUndefined(); + expect(result.error.message).toMatch(/Cannot find module/); + } + const again = await settle(pool.exec("format", [])); + expect(again.hung).toBeUndefined(); + expect(again.error.message).toMatch(/Cannot find module/); + }); + + test("a worker that fails to start is not respawned forever", async () => { + const pool = makePool(MISSING_SCRIPT); + await settle(pool.exec("format", [])); + await new Promise((resolve) => setTimeout(resolve, 500)); + // Replacements are only created for queued work, so the pool is empty once nothing is pending + expect(pool.poolSize).toBe(0); + expect(pool.queueSize).toBe(0); + }); + + test("destroy rejects in-flight and queued tasks, and later tasks", async () => { + const pool = makePool(makeScript(SCRIPTS.silent)); + const pending = [settle(pool.exec("format", [])), settle(pool.exec("format", [])), settle(pool.exec("format", []))]; + await new Promise((resolve) => setTimeout(resolve, 200)); + pool.destroy(); + for (const result of await Promise.all(pending)) { + expect(result.hung).toBeUndefined(); + expect(result.error.message).toMatch(/terminated/); + } + const after = await settle(pool.exec("format", [])); + expect(after.hung).toBeUndefined(); + expect(after.error.message).toMatch(/destroyed/); + expect(pool.isTerminated).toBe(true); + expect(pool.poolSize).toBe(0); + }); +}); diff --git a/test/utils.js b/test/utils.js index 81116e8..9946e77 100644 --- a/test/utils.js +++ b/test/utils.js @@ -124,7 +124,11 @@ async function runCommand(dir, args, options) { const cwd = fixtures ? fixtures.path : TESTS_PATH; const argsForTesting = options.parallel ? [] : ["--no-parallel"]; const argsWithReplacements = args.map((arg) => arg.replaceAll("$CWD", cwd)); - const result = exec("node", [BIN_PATH, ...argsForTesting, ...argsWithReplacements], { cwd, stdio: "pipe" }); + const result = exec("node", [BIN_PATH, ...argsForTesting, ...argsWithReplacements], { + cwd, + stdio: "pipe", + env: options.env ? { ...process.env, ...options.env } : process.env, + }); if (options.input) { result.process.stdin.write(options.input);