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
101 changes: 79 additions & 22 deletions packages/rstack/src/fmt/runner.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
import { cacheNamespace, createCacheKeyResolver, createOptionsHasher } from './cacheIdentity.ts';
import { loadFmtCacheStore } from './cacheStore.ts';
import type { FmtCacheEntry, FmtCacheStore } from './cacheStore.ts';
import type {
FmtFileCache,
FmtExitCode,
FmtFileRequest,
FmtFileResult,
Expand All @@ -11,6 +15,18 @@ import type { FmtWorkerPool } from './workerPool.ts';
type FormatFile = FmtWorkerPool['formatFile'];
type FmtFileOutcome = FmtFileResult | 'unchanged' | 'unsupported';

interface FmtFileRun {
outcome: FmtFileOutcome;
key?: string;
entry?: FmtCacheEntry;
}

interface RunCache {
store: FmtCacheStore;
resolveKey: ReturnType<typeof createCacheKeyResolver>;
hashOptions: ReturnType<typeof createOptionsHasher>;
}

interface FmtWorkerPoolResult {
files: FmtFileResult[];
processedFileCount: number;
Expand All @@ -32,22 +48,47 @@ const runFmtFile = async (
file: FmtFileRequest,
shouldWrite: boolean,
formatFile: FormatFile,
): Promise<FmtFileOutcome> => {
try {
const result = await formatFile(file, shouldWrite);
if (result === 'unchanged' || result === 'unsupported') {
return result;
cache?: RunCache,
): Promise<FmtFileRun> => {
let key: string | undefined;
let fileCache: FmtFileCache | undefined;

if (cache) {
key = cache.resolveKey(file.path);
if (key !== undefined) {
const optionsHash = cache.hashOptions(file.options);
if (optionsHash === undefined) {
key = undefined;
} else {
fileCache = {
entry: cache.store.get(key),
optionsHash,
};
}
}
}

return {
path: file.path,
status: shouldWrite ? 'written' : 'different',
};
try {
const result = await formatFile(file, shouldWrite, fileCache);
const outcome: FmtFileOutcome =
result.status === 'changed'
? {
path: file.path,
status: shouldWrite ? 'written' : 'different',
}
: result.status;

if (key !== undefined && result.cacheEntry) {
return { outcome, key, entry: result.cacheEntry };
}
return { outcome };
} catch (error) {
return {
path: file.path,
status: 'error',
error,
outcome: {
path: file.path,
status: 'error',
error,
},
};
}
};
Expand All @@ -57,7 +98,8 @@ const runPriorityFmtFiles = async (
files: FmtFileRequest[],
shouldWrite: boolean,
formatFile: FormatFile,
): Promise<FmtFileOutcome[]> => {
cache?: RunCache,
): Promise<FmtFileRun[]> => {
const priority: number[] = [];
const rest: number[] = [];

Expand All @@ -67,9 +109,9 @@ const runPriorityFmtFiles = async (

const order = priority.concat(rest);
const outcomes = await Promise.all(
order.map((index) => runFmtFile(files[index], shouldWrite, formatFile)),
order.map((index) => runFmtFile(files[index], shouldWrite, formatFile, cache)),
);
const results = new Array<FmtFileOutcome>(files.length);
const results = new Array<FmtFileRun>(files.length);
for (let index = 0; index < order.length; index++) {
results[order[index]] = outcomes[index];
}
Expand All @@ -81,28 +123,32 @@ const runFmtFilesInWorkerPool = async (
files: FmtFileRequest[],
shouldWrite: boolean,
maxWorkers?: number,
cache?: RunCache,
): Promise<FmtWorkerPoolResult> => {
const { createFmtWorkerPool } = await import('./workerPool.ts');
const workerPool = await createFmtWorkerPool(files.length, maxWorkers);

try {
const results =
workerPool.workerCount >= minPriorityWorkers
? await runPriorityFmtFiles(files, shouldWrite, workerPool.formatFile)
? await runPriorityFmtFiles(files, shouldWrite, workerPool.formatFile, cache)
: await Promise.all(
files.map((file) => runFmtFile(file, shouldWrite, workerPool.formatFile)),
files.map((file) => runFmtFile(file, shouldWrite, workerPool.formatFile, cache)),
);
const processedFiles: FmtFileResult[] = [];
let processedFileCount = 0;

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

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

Expand Down Expand Up @@ -133,12 +179,23 @@ const runFmtFiles = async ({
files,
mode,
maxWorkers,
cache,
}: RunFmtFilesOptions): Promise<FmtRunResult> => {
const shouldWrite = mode === 'write';
let runCache: RunCache | undefined;
if (files.length > 0 && cache && !shouldWrite) {
runCache = {
store: await loadFmtCacheStore(cache.filePath, cacheNamespace),
resolveKey: createCacheKeyResolver(cache.rootPath),
hashOptions: createOptionsHasher(),
};
}

const result =
files.length === 0
? { files: [], processedFileCount: 0 }
: await runFmtFilesInWorkerPool(files, shouldWrite, maxWorkers);
: await runFmtFilesInWorkerPool(files, shouldWrite, maxWorkers, runCache);
await runCache?.store.save().catch(() => false);

return {
...result,
Expand Down
23 changes: 23 additions & 0 deletions packages/rstack/src/fmt/types.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import type { Config as PrettierConfig, Options as PrettierOptions } from 'prettier';
import type { FmtCacheEntry } from './cacheStore.ts';

/** Plugin objects cannot cross worker boundaries and are not planned for support. */
type FmtPluginSpecifier = string | URL;
Expand Down Expand Up @@ -71,6 +72,23 @@ interface FmtFileRequest {
options: ResolvedFmtOptions;
}

interface FmtCacheContext {
/** Persistent cache file to load and update. */
filePath: string;
/** Root used to create portable per-file cache keys. */
rootPath: string;
}

interface FmtFileCache {
entry: FmtCacheEntry | undefined;
optionsHash: string;
}

interface FmtWorkerResult {
status: 'changed' | 'unchanged' | 'unsupported';
cacheEntry?: FmtCacheEntry;
}

type FmtMode = 'write' | 'check' | 'list-different';
type FmtExitCode = 0 | 1 | 2;

Expand All @@ -81,6 +99,8 @@ interface RunFmtFilesOptions {
mode: FmtMode;
/** Maximum number of formatting workers. */
maxWorkers?: number;
/** Internal persistent cache context. Currently used only by check and list modes. */
cache?: FmtCacheContext;
}

interface SuccessfulFmtFileResult {
Expand All @@ -106,14 +126,17 @@ interface FmtRunResult {

export type {
DiscoverFmtFilesOptions,
FmtCacheContext,
FmtConfig,
FmtConfigDefinition,
FmtExitCode,
FmtFileResult,
FmtFileRequest,
FmtFileCache,
FmtMode,
FmtPluginSpecifier,
FmtRunResult,
FmtWorkerResult,
ResolvedFmtConfig,
ResolvedFmtOptions,
RunFmtFilesOptions,
Expand Down
53 changes: 39 additions & 14 deletions packages/rstack/src/fmt/worker.ts
Original file line number Diff line number Diff line change
@@ -1,39 +1,64 @@
// Derived from @prettier/cli, see THIRD_PARTY_NOTICES.md

import { createHash } from 'node:crypto';
import { readFileSync, writeFileSync } from 'node:fs';
import { formatFmtSource } from './format.ts';
import type { FmtFileRequest } from './types.ts';

type FormatFileResult = 'changed' | 'unchanged' | 'unsupported';
import type { FmtCacheEntry } from './cacheStore.ts';
import type { FmtFileCache, FmtFileRequest, FmtWorkerResult } from './types.ts';

interface FormatFileTask {
file: FmtFileRequest;
shouldWrite: boolean;
cache?: FmtFileCache;
}

const hashContent = (content: Uint8Array): string =>
createHash('sha256').update(content).digest('hex');

/**
* Use synchronous direct I/O inside the dedicated worker to avoid libuv
* scheduling overhead. This prioritizes throughput over crash-safe replacement.
*/
const formatFile = async ({ file, shouldWrite }: FormatFileTask): Promise<FormatFileResult> => {
const result = await formatFmtSource(file, () => readFileSync(file.path, 'utf8'));
const formatFile = async ({
file,
shouldWrite,
cache,
}: FormatFileTask): Promise<FmtWorkerResult> => {
let source: string | undefined;
let contentHash: string | undefined;

if (cache && !shouldWrite) {
const content = readFileSync(file.path);
Comment thread
chenjiahan marked this conversation as resolved.
contentHash = hashContent(content);
source = content.toString('utf8');

const { entry, optionsHash } = cache;
if (entry?.[0] === contentHash && entry[1] === optionsHash) {
return { status: entry[2] === 'clean' ? 'unchanged' : 'changed' };
}
}

const { formatFmtSource } = await import('./format.ts');
const result = await formatFmtSource(file, () => (source ??= readFileSync(file.path, 'utf8')));
if (result.status === 'unsupported') {
return 'unsupported';
return { status: 'unsupported' };
}

const { source, formatted } = result;
if (source === formatted) {
return 'unchanged';
const unchanged = result.source === result.formatted;

if (!unchanged && shouldWrite) {
writeFileSync(file.path, result.formatted, 'utf8');
}

if (shouldWrite) {
writeFileSync(file.path, formatted, 'utf8');
const status = unchanged ? 'unchanged' : 'changed';
if (!cache || contentHash === undefined) {
return { status };
}

return 'changed';
const cacheEntry: FmtCacheEntry = [contentHash, cache.optionsHash, unchanged ? 'clean' : 'dirty'];
return { status, cacheEntry };
};

/** Confirms that the worker module and its runtime dependencies are ready. */
/** Confirms that the worker module is ready. Formatter dependencies load only on a cache miss. */
const initializeFmtWorker = (): true => true;

export { formatFile, initializeFmtWorker };
6 changes: 4 additions & 2 deletions packages/rstack/src/fmt/workerPool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import { availableParallelism } from 'node:os';
import Tinypool from 'tinypool';
import type { FmtFileRequest } from './types.ts';
import type { FmtFileCache, FmtFileRequest } from './types.ts';

type FmtWorkerMethods = typeof import('./worker.ts');

Expand All @@ -11,6 +11,7 @@ interface FmtWorkerPool {
formatFile: (
file: FmtFileRequest,
shouldWrite: boolean,
cache?: FmtFileCache,
) => ReturnType<FmtWorkerMethods['formatFile']>;
terminate: () => Promise<void>;
}
Expand Down Expand Up @@ -57,7 +58,8 @@ const createFmtWorkerPool = async (

return {
workerCount,
formatFile: (file, shouldWrite) => pool.run({ file, shouldWrite }, { name: 'formatFile' }),
formatFile: (file, shouldWrite, cache) =>
pool.run({ file, shouldWrite, cache }, { name: 'formatFile' }),
terminate: () => pool.destroy(),
};
};
Expand Down
Loading