Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 44 additions & 23 deletions packages/rstack/src/fmt/runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ interface RunCache {
hashOptions: ReturnType<typeof createOptionsHasher>;
}

interface FmtWorkerPoolResult {
interface FmtFilesResult {
files: FmtFileResult[];
processedFileCount: number;
}
Expand Down Expand Up @@ -193,13 +193,38 @@ const runPriorityTasks = async (
return results;
};

/** Processes files in a worker pool while preserving input order. */
const runWithWorkers = async (
/** Collects per-file outcomes while preserving cache and processed-count semantics. */
const collectFmtResults = (
results: FmtFileRun[],
cache?: RunCache,
): FmtFilesResult => {
const processedFiles: FmtFileResult[] = [];
let processedFileCount = 0;

for (const { outcome, key, entry } of results) {
if (key !== undefined && entry) {
cache?.store.set(key, entry);
}
if (outcome === 'unsupported') {
continue;
}

processedFileCount++;
if (outcome !== 'unchanged') {
processedFiles.push(outcome);
}
}

return { files: processedFiles, processedFileCount };
};

/** Processes one pending file locally and multiple pending files in a worker pool. */
const runFmtTasks = async (
files: FmtFileRequest[],
shouldWrite: boolean,
maxWorkers?: number,
cache?: RunCache,
): Promise<FmtWorkerPoolResult> => {
): Promise<FmtFilesResult> => {
const tasks = files.map((file) => createRunTask(file, cache));
const pendingFileCount = tasks.reduce(
(count, task) => count + (isCachedUnsupported(task) ? 0 : 1),
Expand All @@ -209,6 +234,19 @@ const runWithWorkers = async (
return { files: [], processedFileCount: 0 };
}

// One pending file cannot benefit from parallelism, so avoid worker startup and IPC overhead.
if (pendingFileCount === 1) {
const { formatFile } = await import('./worker.ts');
const formatFileOnMainThread: FormatFile = (file, write, fileCache) =>
formatFile({ file, shouldWrite: write, cache: fileCache });
const results = await Promise.all(
tasks.map((task) =>
runFmtFile(task, shouldWrite, formatFileOnMainThread),
),
);
return collectFmtResults(results, cache);
}

const { createWorkerPool } = await import('./workerPool.ts');
const workerPool = await createWorkerPool(pendingFileCount, maxWorkers);

Expand All @@ -221,24 +259,7 @@ const runWithWorkers = async (
runFmtFile(task, shouldWrite, workerPool.formatFile),
),
);
const processedFiles: FmtFileResult[] = [];
let processedFileCount = 0;

for (const { outcome, key, entry } of results) {
if (key !== undefined && entry) {
cache?.store.set(key, entry);
}
if (outcome === 'unsupported') {
continue;
}

processedFileCount++;
if (outcome !== 'unchanged') {
processedFiles.push(outcome);
}
}

return { files: processedFiles, processedFileCount };
return collectFmtResults(results, cache);
} finally {
await workerPool.terminate();
}
Expand Down Expand Up @@ -284,7 +305,7 @@ const runFmtFiles = async ({
const result =
files.length === 0
? { files: [], processedFileCount: 0 }
: await runWithWorkers(files, shouldWrite, maxWorkers, runCache);
: await runFmtTasks(files, shouldWrite, maxWorkers, runCache);
await runCache?.store.save().catch(() => false);

return {
Expand Down
4 changes: 2 additions & 2 deletions packages/rstack/src/fmt/worker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,8 @@ const hashContent = (content: string | Uint8Array): string =>
hash('sha256', content, 'base64url').slice(0, 16);

/**
* Use synchronous direct I/O inside the dedicated worker to avoid libuv
* scheduling overhead. This prioritizes throughput over crash-safe replacement.
* Synchronous file I/O avoids libuv scheduling overhead in workers and single-file
* main-thread runs. This favors throughput over crash-safe file replacement.
*/
const formatFile = async ({
file,
Expand Down
11 changes: 8 additions & 3 deletions packages/rstack/tests/fmt/runnerWorkerPreflight.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -68,20 +68,25 @@ test('does not start the worker pool when every parser result is cached as unsup
});
});

test('starts the worker pool for a path-only unsupported entry without an extension', async () => {
test('rechecks a path-only unsupported entry on the main thread', async () => {
await withTempProject(async (rootPath) => {
const { cache, file } = await createCachedUnsupportedFile(
rootPath,
'script',
);
writeProjectFile(rootPath, 'script', '#!/usr/bin/env node\nconst value=1');

await expect(
runFmtFiles({
files: [file],
mode: 'check',
cache,
}),
).rejects.toThrow('worker startup failed');
expect(mocks.workerPoolCalls).toEqual([[1, undefined]]);
).resolves.toEqual({
exitCode: 1,
files: [{ path: file.path, status: 'different' }],
processedFileCount: 1,
});
expect(mocks.workerPoolCalls).toEqual([]);
});
});