diff --git a/.changeset/drain-results.md b/.changeset/drain-results.md new file mode 100644 index 000000000..951bd95e6 --- /dev/null +++ b/.changeset/drain-results.md @@ -0,0 +1,6 @@ +--- +'@doc-kit/core': minor +'@doc-kit/cli': patch +--- + +`runGenerators` with `discardResults: true` in its configuration keeps none of the requested generators' output: each still runs to its end, and once nothing else reads its output, its chunks go as they are produced. The CLI's `generate` and `preview` commands use it, as they don't read the results. diff --git a/docs/configuration.md b/docs/configuration.md index a0d750dd9..9d8f2cac9 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -137,6 +137,9 @@ Top-level, alongside `target` and `global`: CPU count. Each worker's heap is limited to 512MB; to raise the limit, run Node.js with `--max-old-space-size` (directly or through `NODE_OPTIONS`). - `chunkSize` {number} Items processed per worker thread. **Default:** `10`. +- `discardResults` {boolean} Keeps none of the requested generators' results, + so each can be let go as soon as it is produced. `doc-kit generate` and + `doc-kit preview` set it. **Default:** `false`. ## Generator options diff --git a/packages/cli/bin/commands/generate.mjs b/packages/cli/bin/commands/generate.mjs index 4122f6fb3..d58a4facf 100644 --- a/packages/cli/bin/commands/generate.mjs +++ b/packages/cli/bin/commands/generate.mjs @@ -64,6 +64,6 @@ export default generate.action( const config = await setConfig(opts); assertRunnableOptions(config); - await runGenerators(config); + await runGenerators({ ...config, discardResults: true }); }) ); diff --git a/packages/cli/bin/commands/preview.mjs b/packages/cli/bin/commands/preview.mjs index 84272fe67..04b4c51fc 100644 --- a/packages/cli/bin/commands/preview.mjs +++ b/packages/cli/bin/commands/preview.mjs @@ -62,7 +62,7 @@ export default preview.action( const config = await setConfig(opts); assertRunnableOptions(config); - await runGenerators(config); + await runGenerators({ ...config, discardResults: true }); // With the default coming from the configuration, this is always set output = config.global.output; diff --git a/packages/core/src/__tests__/caching.test.mjs b/packages/core/src/__tests__/caching.test.mjs index b6106f345..b8d15a3ba 100644 --- a/packages/core/src/__tests__/caching.test.mjs +++ b/packages/core/src/__tests__/caching.test.mjs @@ -1,4 +1,4 @@ -import { deepStrictEqual, strictEqual, throws } from 'node:assert'; +import { deepStrictEqual, rejects, strictEqual, throws } from 'node:assert'; import { describe, it } from 'node:test'; import { collectAsyncGenerator, createCache } from '../caching.mjs'; @@ -212,6 +212,61 @@ describe('caching', () => { deepStrictEqual(await Array.fromAsync(stream), [[1], [2], [3]]); }); + it('should drain a result, alongside its other readers', async () => { + const cache = createCache(); + const produced = []; + + cache.populateConsumerCounts(['a', 'b'], name => + name === 'b' ? 'a' : undefined + ); + + async function* gen() { + for (const value of [1, 2, 3]) { + produced.push(value); + + yield [value]; + } + } + + cache.store('a', gen()); + + const reader = await cache.stream('a'); + + await cache.drain('a'); + + strictEqual(cache.has('a'), false); + deepStrictEqual(produced, [1, 2, 3]); + + deepStrictEqual(await Array.fromAsync(reader), [[1], [2], [3]]); + }); + + it('should let go of the chunks of a drained result as they are produced', async () => { + const cache = createCache(); + const gate = Promise.withResolvers(); + const { released, chunkOf } = trackReleases(); + + cache.populateConsumerCounts(['a'], () => undefined); + + async function* gen() { + yield chunkOf(1); + yield chunkOf(2); + yield chunkOf(3); + await gate.promise; + } + + cache.store('a', gen()); + + const draining = cache.drain('a'); + + // The generator waits, past its chunks: the first is no longer held + await collectGarbage(() => released.has(1)); + + strictEqual(released.has(1), true); + + gate.resolve(); + await draining; + }); + it('should let go of a chunk once every streaming reader has read it, while it still runs', async () => { const cache = createCache(); const gate = Promise.withResolvers(); @@ -252,5 +307,33 @@ describe('caching', () => { await readValues(first); await readValues(second); }); + + it('should drain a non-streaming result', async () => { + const cache = createCache(); + + cache.populateConsumerCounts(['a'], () => undefined); + cache.store('a', Promise.resolve({ value: 1 })); + + const drained = await cache.drain('a'); + + strictEqual(drained, undefined); + strictEqual(cache.has('a'), false); + }); + + it('should reject draining a generator that fails', async () => { + const cache = createCache(); + + cache.populateConsumerCounts(['a'], () => undefined); + + async function* gen() { + yield [1]; + + throw new Error('drained failure'); + } + + cache.store('a', gen()); + + await rejects(cache.drain('a'), /drained failure/); + }); }); }); diff --git a/packages/core/src/__tests__/generators.test.mjs b/packages/core/src/__tests__/generators.test.mjs index 457393439..f4060c9ca 100644 --- a/packages/core/src/__tests__/generators.test.mjs +++ b/packages/core/src/__tests__/generators.test.mjs @@ -312,18 +312,23 @@ describe('createGenerator orchestration', () => { } ); - it('rejects with the failure of a dependency read as a stream', async () => { - const { runGenerators } = createGenerator(); + it('rejects with the failure of a dependency read as a stream, with or without results', async () => { const unhandled = []; const onUnhandled = error => unhandled.push(error); process.on('unhandledRejection', onUnhandled); try { - await assert.rejects( - runGenerators({ target: ['failing-reader'], threads: 2 }), - /source failed/ - ); + for (const discardResults of [false, true]) { + await assert.rejects( + createGenerator().runGenerators({ + target: ['failing-reader'], + threads: 2, + discardResults, + }), + /source failed/ + ); + } // A rejection left unhandled is reported once the microtasks have run await new Promise(resolve => setImmediate(resolve)); @@ -347,4 +352,14 @@ describe('createGenerator orchestration', () => { assert.deepStrictEqual(one, [{ array: true }]); assert.deepStrictEqual(two, [{ array: false }]); }); + + it('returns no results with discardResults', async () => { + const results = await createGenerator().runGenerators({ + target: ['gen-c', 'gen-c-all'], + threads: 1, + discardResults: true, + }); + + assert.equal(results, undefined); + }); }); diff --git a/packages/core/src/caching.mjs b/packages/core/src/caching.mjs index ce533e104..eabf4c3ea 100644 --- a/packages/core/src/caching.mjs +++ b/packages/core/src/caching.mjs @@ -141,8 +141,8 @@ export const createCache = () => { /** * Computes and records consumer counts across the dependency closure of the * requested generators. A generator is consumed once per dependent - * generator, plus once more when it is a requested target (read by the - * final collection). This drives eviction without coupling the cache to the + * generator, plus once more when it is a requested target (read when the run + * ends). This drives eviction without coupling the cache to the * generator registry: callers supply how to resolve a dependency. * * @param {string[]} targets - Requested generator cache keys @@ -179,7 +179,7 @@ export const createCache = () => { } } - // Each requested target is consumed once more by the final collection + // Each requested target is read once more when the run ends for (const name of targets) { remaining[name] = (remaining[name] ?? 0) + 1; } @@ -214,6 +214,30 @@ export const createCache = () => { return value; }, + /** + * Runs a generator to its end, keeping none of its result, which counts as + * one of its reads. + * + * @param {string} name - Cache key (resolved generator specifier) to drain + * @returns {Promise} + */ + drain: async name => { + const result = await results[name]; + + if (!isAsyncIterable(result)) { + release(name); + + return; + } + + const { ended } = teeOf(name, result); + + // Once nothing else reads it, its chunks go as they're produced + release(name); + + await ended; + }, + /** * Gives a reader of a generator's result as it is produced: chunk by chunk * if it streams, or as a single chunk once it resolves. As with `consume`, diff --git a/packages/core/src/generators.mjs b/packages/core/src/generators.mjs index 06b497d54..e374b923f 100644 --- a/packages/core/src/generators.mjs +++ b/packages/core/src/generators.mjs @@ -112,10 +112,12 @@ const createGenerator = () => { * Runs all requested generators with their dependencies. * * @param {import('./utils/configuration/types').Configuration} configuration - Runtime options - * @returns {Promise} Results of all requested generators + * @returns {Promise} Results of all requested + * generators, or `undefined` with `discardResults`, which lets each go as + * soon as it is produced */ const runGenerators = async configuration => { - const { target, threads } = configuration; + const { target, threads, discardResults } = configuration; // Resolve shorthand names and load the full dependency closure up front, // so scheduling below is fully synchronous. @@ -147,15 +149,14 @@ const createGenerator = () => { scheduleGenerator(specifier, generators, inputOf, configuration); } - // Start all collections in parallel (don't await sequentially). Consuming - // through the shared path lets the final read also trigger eviction. - const results = await Promise.all( - targets.map(specifier => cache.consume(specifier)) - ); + // Start all reads in parallel (don't await sequentially). Reading through + // the shared path lets the final read also trigger eviction. + const read = discardResults ? cache.drain : cache.consume; + const collected = await Promise.all(targets.map(read)); await pool.destroy(); - return results; + return discardResults ? undefined : collected; }; return { runGenerators }; diff --git a/packages/core/src/utils/configuration/types.d.ts b/packages/core/src/utils/configuration/types.d.ts index 2437302bc..d29b3b2a0 100644 --- a/packages/core/src/utils/configuration/types.d.ts +++ b/packages/core/src/utils/configuration/types.d.ts @@ -16,6 +16,10 @@ export type Configuration = { // Number of items to process per worker thread chunkSize: number; + + // Keeps none of the requested generators' results, for runs that only need + // what the generators write + discardResults?: boolean; } & { [K in keyof AllGenerators]: GlobalConfiguration & AllGenerators[K]['defaultConfiguration'];