From 9a8903930eb33705d8605509d774aa2cf72521b9 Mon Sep 17 00:00:00 2001 From: Claudio Wunder Date: Tue, 3 Nov 2026 19:01:37 +0100 Subject: [PATCH 1/3] perf(core): let worker.stream take chunks as they arrive worker.stream needed all of its input before it started any work. It now also takes an async iterable of batches, as a streamed input yields them, and starts each batch's work as it arrives. An input that arrives whole, as it does with one thread, works as before. Only a few chunks per thread are started at a time, and results are yielded in the order the chunks were submitted. That keeps generators that run at once from filling the pool with their own chunks. Piscina runs with stricterFIFO for it: by default it puts a task it can't hand out yet at the back of its queue, so early heavy tasks ran last. toBatches gives an input that is whole or streamed as batches, and createSignal is the wake-up the stream waits on. Assisted-by: Claude Opus 5.5 Assisted-by: Claude Sonnet 5.5 --- .changeset/stream-worker-chunks.md | 5 + .../src/threading/__tests__/index.test.mjs | 10 + .../src/threading/__tests__/parallel.test.mjs | 216 ++++++++++++++++++ packages/core/src/threading/constants.mjs | 7 + packages/core/src/threading/index.mjs | 4 + packages/core/src/threading/parallel.mjs | 174 +++++++++++--- .../src/utils/__tests__/generators.test.mjs | 14 ++ packages/core/src/utils/generators.mjs | 17 ++ packages/core/src/utils/misc.mjs | 26 +++ 9 files changed, 447 insertions(+), 26 deletions(-) create mode 100644 .changeset/stream-worker-chunks.md diff --git a/.changeset/stream-worker-chunks.md b/.changeset/stream-worker-chunks.md new file mode 100644 index 000000000..188f05c91 --- /dev/null +++ b/.changeset/stream-worker-chunks.md @@ -0,0 +1,5 @@ +--- +'@doc-kit/core': minor +--- + +`worker.stream` accepts an async iterable of batches, starting each one's work as it arrives, as well as an array. It keeps only a few chunks per thread in flight, so generators running at once share the worker pool, starts them in the order they were submitted and yields their results in that order. `toBatches`, from `@doc-kit/core/utils/generators.mjs`, gives an input that is either whole or streamed as batches. diff --git a/packages/core/src/threading/__tests__/index.test.mjs b/packages/core/src/threading/__tests__/index.test.mjs index 203278840..127c0edc7 100644 --- a/packages/core/src/threading/__tests__/index.test.mjs +++ b/packages/core/src/threading/__tests__/index.test.mjs @@ -128,4 +128,14 @@ describe('createWorkerPool', () => { await pool.destroy(); } }); + + it('should start tasks in the order they were submitted', async () => { + const pool = createWorkerPool(1); + + try { + strictEqual(pool.options.stricterFIFO, true); + } finally { + await pool.destroy(); + } + }); }); diff --git a/packages/core/src/threading/__tests__/parallel.test.mjs b/packages/core/src/threading/__tests__/parallel.test.mjs index 4a2c3653b..4c87e871a 100644 --- a/packages/core/src/threading/__tests__/parallel.test.mjs +++ b/packages/core/src/threading/__tests__/parallel.test.mjs @@ -2,6 +2,7 @@ import { deepStrictEqual, ok, rejects, strictEqual } from 'node:assert'; import { describe, it } from 'node:test'; import { loadGenerator } from '../../generators/loader.mjs'; +import { CHUNKS_PER_THREAD } from '../constants.mjs'; import createWorkerPool from '../index.mjs'; import createParallelWorker from '../parallel.mjs'; @@ -42,6 +43,52 @@ async function collectChunks(generator) { return chunks; } +/** + * Streams three times as many one-item chunks as two threads can run at once + * through a pool that holds each task until it is settled, newest first, and + * checks before each that as many run as there is room for: no more, and no + * fewer while any are left to start. Settling the newest first holds back the + * oldest chunk, so no result can be emitted until the last: the room must come + * from the settling alone. The results still come out in submission order. + * A stream that lost count of the held chunks would never end, so its tests + * need a timeout. + * + * @param {(items: number[]) => number[] | AsyncIterable} toInput + * @returns {Promise} + */ +async function checkChunksPerThread(toInput) { + const limit = 2 * CHUNKS_PER_THREAD; + const items = Array.from({ length: limit * 3 }, (_, index) => index); + const started = []; + const running = []; + const pool = { + run: task => + new Promise(resolve => { + started.push(...task.input); + running.push(() => resolve(task.input)); + }), + }; + const worker = createParallelWorker('ast-js', astJsGenerator, pool, { + threads: 2, + chunkSize: 1, + }); + + const result = collectStream(worker.stream(toInput(items), {})); + + for (let settled = 0; settled < items.length; settled++) { + await new Promise(resolve => setImmediate(resolve)); + + strictEqual(running.length, Math.min(limit, items.length - settled)); + + running.pop()(); + } + + const results = await result; + + deepStrictEqual(started, items); + deepStrictEqual(results, items); +} + describe('createParallelWorker', () => { it('should create a ParallelWorker with stream method', async () => { const pool = createWorkerPool(2); @@ -230,6 +277,12 @@ describe('createParallelWorker', () => { await rejects(result, /chunk failed/); }); + it( + 'should start a few chunks per thread at a time, and the next as each settles', + { timeout: 10_000 }, + () => checkChunksPerThread(items => items) + ); + it('should process one chunk at a time, in order, with a single thread', async () => { let running = 0; let mostRunning = 0; @@ -257,4 +310,167 @@ describe('createParallelWorker', () => { strictEqual(mostRunning, 1); deepStrictEqual(chunks, [[1, 2], [3, 4], [5]]); }); + + describe('with streamed batches', () => { + it( + 'starts a batch before the next one arrives', + // The second batch only arrives once the first result is read, so a + // stream that waited for it would never give one + { timeout: 10_000 }, + async () => { + const started = []; + const pool = { + run: async task => { + started.push(task.input); + + return task.input; + }, + }; + const worker = createParallelWorker('ast-js', astJsGenerator, pool, { + threads: 2, + chunkSize: 2, + }); + const gate = Promise.withResolvers(); + + async function* batches() { + yield [1, 2, 3]; + await gate.promise; + yield [4]; + } + + const reader = worker.stream(batches(), {}); + const first = await reader.next(); + + // The second batch hasn't arrived, but the first one's chunks ran + deepStrictEqual(first.value, [1, 2]); + deepStrictEqual(started, [[1, 2], [3]]); + + gate.resolve(); + + const rest = await collectChunks(reader); + + deepStrictEqual(rest, [[3], [4]]); + } + ); + + it( + 'yields in submission order across batches', + // The test waits for every chunk to start, so it would wait forever on a + // stream that held one back + { timeout: 10_000 }, + async () => { + const resolvers = []; + const pool = { + run: task => + new Promise(resolve => { + resolvers.push(() => resolve(task.input)); + }), + }; + const worker = createParallelWorker('ast-js', astJsGenerator, pool, { + threads: 2, + chunkSize: 1, + }); + + async function* batches() { + yield [0, 1]; + yield [2]; + } + + const result = collectStream(worker.stream(batches(), {})); + + while (resolvers.length < 3) { + await new Promise(resolve => setImmediate(resolve)); + } + + for (const resolve of resolvers.toReversed()) { + resolve(); + } + + const results = await result; + + deepStrictEqual(results, [0, 1, 2]); + } + ); + + it('throws a failure reading the batches after the chunks before it', async () => { + const pool = { run: async task => task.input }; + const worker = createParallelWorker('ast-js', astJsGenerator, pool, { + threads: 2, + chunkSize: 1, + }); + + async function* batches() { + yield [1, 2]; + throw new Error('source failed'); + } + + const chunks = []; + + await rejects(async () => { + for await (const chunk of worker.stream(batches(), {})) { + chunks.push(chunk); + } + }, /source failed/); + + deepStrictEqual(chunks, [[1], [2]]); + }); + + it( + 'starts a few chunks per thread at a time, across batches', + { timeout: 10_000 }, + () => + checkChunksPerThread(async function* (items) { + for (let index = 0; index < items.length; index += 5) { + yield items.slice(index, index + 5); + } + }) + ); + + it('runs each batch on this thread as it arrives with a single thread', async () => { + const events = []; + const generator = { + name: 'sequential', + async processChunk(items, indices) { + events.push(`chunk ${indices.map(index => items[index])}`); + + return indices.map(index => items[index]); + }, + }; + const worker = createParallelWorker('sequential', generator, null, { + threads: 1, + chunkSize: 2, + }); + + async function* batches() { + events.push('batch 1'); + yield [1, 2, 3]; + events.push('batch 2'); + yield [4]; + } + + const chunks = await collectChunks(worker.stream(batches(), {})); + + deepStrictEqual(chunks, [[1, 2], [3], [4]]); + deepStrictEqual(events, [ + 'batch 1', + 'chunk 1,2', + 'chunk 3', + 'batch 2', + 'chunk 4', + ]); + }); + + it('yields nothing for no batches', async () => { + const worker = createParallelWorker('ast-js', astJsGenerator, null, { + threads: 2, + chunkSize: 1, + }); + + async function* batches() {} + + const chunks = await collectChunks(worker.stream(batches(), {})); + + deepStrictEqual(chunks, []); + }); + }); }); diff --git a/packages/core/src/threading/constants.mjs b/packages/core/src/threading/constants.mjs index 9872c6686..1ea409ec2 100644 --- a/packages/core/src/threading/constants.mjs +++ b/packages/core/src/threading/constants.mjs @@ -22,3 +22,10 @@ export const WORKER_HEAP_SIZE = 512; * over half of a worker's heap. Node.js's `--max-semi-space-size` replaces it. */ export const WORKER_GEN_SIZE = 96; + +/** + * How many chunks of one generator can be started, running or queued, per + * thread at a time: enough to keep every thread busy while more are submitted, + * few enough that generators running at once share the pool. + */ +export const CHUNKS_PER_THREAD = 3; diff --git a/packages/core/src/threading/index.mjs b/packages/core/src/threading/index.mjs index 0a2ed5375..7cbf5b2b3 100644 --- a/packages/core/src/threading/index.mjs +++ b/packages/core/src/threading/index.mjs @@ -32,6 +32,10 @@ export default function createWorkerPool(threads) { maxYoungGenerationSizeMb: WORKER_GEN_SIZE, }, recordTiming: false, + // A task taken from the queue that no worker can run yet goes back to the + // front, not the back, so tasks start in the order they were submitted + // and results, yielded in that order, are not held up + stricterFIFO: true, workerData: { logLevel: logger.getLogLevel() }, }); } diff --git a/packages/core/src/threading/parallel.mjs b/packages/core/src/threading/parallel.mjs index 136bac69f..4b0874b37 100644 --- a/packages/core/src/threading/parallel.mjs +++ b/packages/core/src/threading/parallel.mjs @@ -1,6 +1,10 @@ 'use strict'; import logger from '#logger/index.mjs'; +import { toBatches } from '#utils/generators.mjs'; +import { createSignal, isAsyncIterable } from '#utils/misc.mjs'; + +import { CHUNKS_PER_THREAD } from './constants.mjs'; const parallelLogger = logger.child('parallel'); @@ -71,7 +75,8 @@ export default function createParallelWorker( pool, configuration ) { - const { threads, chunkSize } = configuration; + const { threads: defaultThreads = 1, chunkSize } = configuration; + const threads = Math.max(Number(defaultThreads), 1); const { name } = generator; @@ -79,54 +84,171 @@ export default function createParallelWorker( /** * Processes items in parallel, yielding results in submission order. * + * The items can also arrive in batches, as a dependency streams them: each + * batch's chunks start as soon as it arrives, while earlier ones still run. + * * @template T - * @param {T[]} items - Items to process + * @param {T[] | AsyncIterable} items - Items to process * @param {object} extra - Extra options * @yields {R[]} Chunk results in input order */ async *stream(items, extra) { - if (items.length === 0) { + const streamed = isAsyncIterable(items); + + if (!streamed && items.length === 0) { return; } - const chunks = createChunks(items.length, chunkSize); + const batches = toBatches(items); + + /** + * Splits a batch into chunks. + * + * @param {T[]} batch - Items to split + * @returns {number[][]} Each chunk's indices into the batch + */ + const distribute = batch => { + const chunks = createChunks(batch.length, chunkSize); - parallelLogger.debug( - `Distributing ${items.length} items across ${chunks.length} chunks`, - { generator: name, chunks: chunks.length, chunkSize, threads } - ); + parallelLogger.debug( + `Distributing ${batch.length} items across ${chunks.length} chunks`, + { generator: name, chunks: chunks.length, chunkSize, threads } + ); + + return chunks; + }; // On this thread, one chunk at a time, so their work isn't all held at once - if (threads <= 1 || items.length <= 2) { - for (const indices of chunks) { - yield await generator.processChunk(items, indices, extra); + if (threads <= 1 || (!streamed && items.length <= 2)) { + for await (const batch of batches) { + for (const indices of distribute(batch)) { + const result = await generator.processChunk(batch, indices, extra); + + yield result; + } } return; } - // Start every chunk immediately, but emit results in submission order. - // Handle errors as each task settles so a later failure cannot become an - // unhandled rejection while earlier chunks are still running. - const tasks = chunks.map(indices => { - const task = pool.run( - createTask(items, indices, extra, configuration, specifier, name) - ); + // Read every batch as it arrives, but start only a few chunks per thread + // at a time, so that generators running at once share the pool instead + // of one's chunks waiting behind all of another's. Emit results in + // submission order. Handle errors as each task settles so a later failure + // cannot become an unhandled rejection while earlier chunks are still + // running. + const tasks = []; + const waiting = []; + const limit = threads * CHUNKS_PER_THREAD; - return task.then( - result => ({ result }), - error => ({ error }) - ); - }); + let running = 0; + let started = 0; + let doneReading = false; + + // Notified as chunks start, and once every batch is read + const changed = createSignal(); + + /** + * Settles a chunk's task, making room for the next one. + * + * @param {{ result: unknown } | { error: unknown }} outcome - The task's + * result or error + * @returns {{ result: unknown } | { error: unknown }} The same outcome + */ + const settle = outcome => { + running--; + start(); + + return outcome; + }; + + /** + * Starts the waiting chunks there is room for. + */ + const start = () => { + while (running < limit && started < waiting.length) { + const { batch, indices } = waiting[started]; + + // A started chunk no longer needs to hold its batch here + waiting[started++] = undefined; + running++; + + const task = pool.run( + createTask(batch, indices, extra, configuration, specifier, name) + ); + + tasks.push( + task.then( + result => settle({ result }), + error => settle({ error }) + ) + ); + } + + changed.notify(); + }; + + /** + * Queues every chunk of a batch, and starts those there is room for. + * + * @param {T[]} batch - Items to queue + */ + const add = batch => { + for (const indices of distribute(batch)) { + waiting.push({ batch, indices }); + } + + start(); + }; + + /** + * Queues every batch as it arrives: an array's at once, so its first + * chunks start right away. + */ + const readBatches = async () => { + try { + if (!streamed) { + add(items); + + return; + } + + for await (const batch of items) { + add(batch); + } + } finally { + doneReading = true; + changed.notify(); + } + }; + + const reading = readBatches(); + + // A failure reading the batches is thrown below, once every chunk + // submitted before it has been emitted + reading.catch(() => {}); + + for (let index = 0; ; index++) { + while (index >= tasks.length) { + if (doneReading && started === waiting.length) { + await reading; + + return; + } + + await changed.wait(); + } + + const outcome = await tasks[index]; - for (const [index, task] of tasks.entries()) { - const outcome = await task; + // Emitted chunks are the consumer's to keep or let go + tasks[index] = undefined; if ('error' in outcome) { throw outcome.error; } - parallelLogger.debug(`Chunk ${index + 1}/${chunks.length} completed`, { + parallelLogger.debug(`Chunk ${index + 1}/${waiting.length} completed`, { generator: name, }); diff --git a/packages/core/src/utils/__tests__/generators.test.mjs b/packages/core/src/utils/__tests__/generators.test.mjs index 6fe290dba..a9b550373 100644 --- a/packages/core/src/utils/__tests__/generators.test.mjs +++ b/packages/core/src/utils/__tests__/generators.test.mjs @@ -7,6 +7,7 @@ import { getVersionFromSemVer, coerceSemVer, getCompatibleVersions, + toBatches, } from '../generators.mjs'; describe('getEntryDescription', () => { @@ -138,3 +139,16 @@ describe('getCompatibleVersions', () => { assert.equal(result.length, 2); }); }); + +describe('toBatches', () => { + it('gives an array as a single batch, and batches as they are', async () => { + async function* batches() { + yield [1]; + } + + const streamed = batches(); + + assert.deepEqual(toBatches([1, 2]), [[1, 2]]); + assert.equal(toBatches(streamed), streamed); + }); +}); diff --git a/packages/core/src/utils/generators.mjs b/packages/core/src/utils/generators.mjs index 516113bca..f8996d2b5 100644 --- a/packages/core/src/utils/generators.mjs +++ b/packages/core/src/utils/generators.mjs @@ -2,6 +2,7 @@ import { coerce, major } from 'semver'; +import { isAsyncIterable } from './misc.mjs'; import { transformNodeToString } from './unist.mjs'; /** @@ -42,6 +43,22 @@ export const groupNodesByModule = nodes => { return Map.groupBy(nodes, node => node.api); }; +/** + * Gives a generator's input as batches: as they arrive if it streams, or as a + * single batch if it arrives whole, as it does with one thread. + * + * @template T + * @param {Array | AsyncIterable>} input + * @returns {Array> | AsyncIterable>} + */ +export const toBatches = input => { + if (isAsyncIterable(input)) { + return input; + } + + return [input]; +}; + /** * Parses the SemVer string into a Node.js-alike version * diff --git a/packages/core/src/utils/misc.mjs b/packages/core/src/utils/misc.mjs index 92d91e01a..6b73c1960 100644 --- a/packages/core/src/utils/misc.mjs +++ b/packages/core/src/utils/misc.mjs @@ -70,3 +70,29 @@ export const deepMerge = (...objects) => { return result; }, {}); }; + +/** + * Creates a signal: waiting on it resolves the next time it is notified. + * + * @returns {{ wait: () => Promise, notify: () => void }} + */ +export const createSignal = () => { + let next = Promise.withResolvers(); + + return { + /** + * Waits for the next notification. + * + * @returns {Promise} + */ + wait: () => next.promise, + + /** + * Wakes what waits. + */ + notify: () => { + next.resolve(); + next = Promise.withResolvers(); + }, + }; +}; From afa24ca4736621fa173e11acc722556fb466e266 Mon Sep 17 00:00:00 2001 From: Claudio Wunder Date: Wed, 4 Nov 2026 19:06:31 +0100 Subject: [PATCH 2/3] refactor: simplify worker.stream's backpressure One enqueue now starts a batch's chunks, waiting for room while the in-flight limit is reached, and the consumer waits on an inline Promise.withResolvers() pair, in place of the waiting queue, settle, start and add, and the signal. An array's chunks still start right away, as before. The input is read as `streamed ? items : [items]`, so toBatches and createSignal move up the stack to their first callers. Assisted-by: Claude Opus 5.5 --- .changeset/stream-worker-chunks.md | 2 +- packages/core/src/threading/parallel.mjs | 87 ++++++++----------- .../src/utils/__tests__/generators.test.mjs | 14 --- packages/core/src/utils/generators.mjs | 17 ---- packages/core/src/utils/misc.mjs | 26 ------ 5 files changed, 37 insertions(+), 109 deletions(-) diff --git a/.changeset/stream-worker-chunks.md b/.changeset/stream-worker-chunks.md index 188f05c91..94d5d9761 100644 --- a/.changeset/stream-worker-chunks.md +++ b/.changeset/stream-worker-chunks.md @@ -2,4 +2,4 @@ '@doc-kit/core': minor --- -`worker.stream` accepts an async iterable of batches, starting each one's work as it arrives, as well as an array. It keeps only a few chunks per thread in flight, so generators running at once share the worker pool, starts them in the order they were submitted and yields their results in that order. `toBatches`, from `@doc-kit/core/utils/generators.mjs`, gives an input that is either whole or streamed as batches. +`worker.stream` accepts an async iterable of batches, starting each one's work as it arrives, as well as an array. It keeps only a few chunks per thread in flight, so generators running at once share the worker pool, starts them in the order they were submitted and yields their results in that order. diff --git a/packages/core/src/threading/parallel.mjs b/packages/core/src/threading/parallel.mjs index 4b0874b37..0e31b81e3 100644 --- a/packages/core/src/threading/parallel.mjs +++ b/packages/core/src/threading/parallel.mjs @@ -1,8 +1,7 @@ 'use strict'; import logger from '#logger/index.mjs'; -import { toBatches } from '#utils/generators.mjs'; -import { createSignal, isAsyncIterable } from '#utils/misc.mjs'; +import { isAsyncIterable } from '#utils/misc.mjs'; import { CHUNKS_PER_THREAD } from './constants.mjs'; @@ -99,7 +98,7 @@ export default function createParallelWorker( return; } - const batches = toBatches(items); + const batches = streamed ? items : [items]; /** * Splits a batch into chunks. @@ -138,39 +137,32 @@ export default function createParallelWorker( // cannot become an unhandled rejection while earlier chunks are still // running. const tasks = []; - const waiting = []; const limit = threads * CHUNKS_PER_THREAD; let running = 0; - let started = 0; let doneReading = false; - // Notified as chunks start, and once every batch is read - const changed = createSignal(); + // Resolved when a running chunk settles, while the reader waits for room + let room; + + // Resolved when a chunk starts or the reading ends, while the consumer + // waits for one + let added; /** - * Settles a chunk's task, making room for the next one. + * Starts a batch's chunks, waiting for room whenever the limit is + * reached: an array's first chunks start right away. * - * @param {{ result: unknown } | { error: unknown }} outcome - The task's - * result or error - * @returns {{ result: unknown } | { error: unknown }} The same outcome + * @param {T[]} batch - Items to start */ - const settle = outcome => { - running--; - start(); - - return outcome; - }; + const enqueue = async batch => { + for (const indices of distribute(batch)) { + while (running >= limit) { + room = Promise.withResolvers(); - /** - * Starts the waiting chunks there is room for. - */ - const start = () => { - while (running < limit && started < waiting.length) { - const { batch, indices } = waiting[started]; + await room.promise; + } - // A started chunk no longer needs to hold its batch here - waiting[started++] = undefined; running++; const task = pool.run( @@ -178,47 +170,38 @@ export default function createParallelWorker( ); tasks.push( - task.then( - result => settle({ result }), - error => settle({ error }) - ) + task + .then( + result => ({ result }), + error => ({ error }) + ) + .finally(() => { + running--; + room?.resolve(); + }) ); - } - changed.notify(); - }; - - /** - * Queues every chunk of a batch, and starts those there is room for. - * - * @param {T[]} batch - Items to queue - */ - const add = batch => { - for (const indices of distribute(batch)) { - waiting.push({ batch, indices }); + added?.resolve(); } - - start(); }; /** - * Queues every batch as it arrives: an array's at once, so its first - * chunks start right away. + * Starts every batch's chunks as it arrives. */ const readBatches = async () => { try { if (!streamed) { - add(items); + await enqueue(items); return; } for await (const batch of items) { - add(batch); + await enqueue(batch); } } finally { doneReading = true; - changed.notify(); + added?.resolve(); } }; @@ -230,13 +213,15 @@ export default function createParallelWorker( for (let index = 0; ; index++) { while (index >= tasks.length) { - if (doneReading && started === waiting.length) { + if (doneReading) { await reading; return; } - await changed.wait(); + added = Promise.withResolvers(); + + await added.promise; } const outcome = await tasks[index]; @@ -248,7 +233,7 @@ export default function createParallelWorker( throw outcome.error; } - parallelLogger.debug(`Chunk ${index + 1}/${waiting.length} completed`, { + parallelLogger.debug(`Chunk ${index + 1}/${tasks.length} completed`, { generator: name, }); diff --git a/packages/core/src/utils/__tests__/generators.test.mjs b/packages/core/src/utils/__tests__/generators.test.mjs index a9b550373..6fe290dba 100644 --- a/packages/core/src/utils/__tests__/generators.test.mjs +++ b/packages/core/src/utils/__tests__/generators.test.mjs @@ -7,7 +7,6 @@ import { getVersionFromSemVer, coerceSemVer, getCompatibleVersions, - toBatches, } from '../generators.mjs'; describe('getEntryDescription', () => { @@ -139,16 +138,3 @@ describe('getCompatibleVersions', () => { assert.equal(result.length, 2); }); }); - -describe('toBatches', () => { - it('gives an array as a single batch, and batches as they are', async () => { - async function* batches() { - yield [1]; - } - - const streamed = batches(); - - assert.deepEqual(toBatches([1, 2]), [[1, 2]]); - assert.equal(toBatches(streamed), streamed); - }); -}); diff --git a/packages/core/src/utils/generators.mjs b/packages/core/src/utils/generators.mjs index f8996d2b5..516113bca 100644 --- a/packages/core/src/utils/generators.mjs +++ b/packages/core/src/utils/generators.mjs @@ -2,7 +2,6 @@ import { coerce, major } from 'semver'; -import { isAsyncIterable } from './misc.mjs'; import { transformNodeToString } from './unist.mjs'; /** @@ -43,22 +42,6 @@ export const groupNodesByModule = nodes => { return Map.groupBy(nodes, node => node.api); }; -/** - * Gives a generator's input as batches: as they arrive if it streams, or as a - * single batch if it arrives whole, as it does with one thread. - * - * @template T - * @param {Array | AsyncIterable>} input - * @returns {Array> | AsyncIterable>} - */ -export const toBatches = input => { - if (isAsyncIterable(input)) { - return input; - } - - return [input]; -}; - /** * Parses the SemVer string into a Node.js-alike version * diff --git a/packages/core/src/utils/misc.mjs b/packages/core/src/utils/misc.mjs index 6b73c1960..92d91e01a 100644 --- a/packages/core/src/utils/misc.mjs +++ b/packages/core/src/utils/misc.mjs @@ -70,29 +70,3 @@ export const deepMerge = (...objects) => { return result; }, {}); }; - -/** - * Creates a signal: waiting on it resolves the next time it is notified. - * - * @returns {{ wait: () => Promise, notify: () => void }} - */ -export const createSignal = () => { - let next = Promise.withResolvers(); - - return { - /** - * Waits for the next notification. - * - * @returns {Promise} - */ - wait: () => next.promise, - - /** - * Wakes what waits. - */ - notify: () => { - next.resolve(); - next = Promise.withResolvers(); - }, - }; -}; From 51c7fb0ffdbbf4d02072ac6241739d23bde97df4 Mon Sep 17 00:00:00 2001 From: Claudio Wunder Date: Wed, 4 Nov 2026 19:22:24 +0100 Subject: [PATCH 3/3] docs: say worker.stream reads the next batch once there's room Since the reader waits for room, the comments and the changeset no longer said what it does: a batch's chunks start while earlier ones run, and the next batch is read once there's room for its chunks. The completed-chunk debug log also counts against the chunks distributed so far, rather than those started. Assisted-by: Claude Opus 5.5 --- .changeset/stream-worker-chunks.md | 2 +- packages/core/src/threading/parallel.mjs | 21 ++++++++++++++------- 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/.changeset/stream-worker-chunks.md b/.changeset/stream-worker-chunks.md index 94d5d9761..ce22833bf 100644 --- a/.changeset/stream-worker-chunks.md +++ b/.changeset/stream-worker-chunks.md @@ -2,4 +2,4 @@ '@doc-kit/core': minor --- -`worker.stream` accepts an async iterable of batches, starting each one's work as it arrives, as well as an array. It keeps only a few chunks per thread in flight, so generators running at once share the worker pool, starts them in the order they were submitted and yields their results in that order. +`worker.stream` accepts an async iterable of batches as well as an array, starting a batch's work while earlier ones still run and reading the next batch once there is room for it. It keeps only a few chunks per thread in flight, so generators running at once share the worker pool, starts them in the order they were submitted and yields their results in that order. diff --git a/packages/core/src/threading/parallel.mjs b/packages/core/src/threading/parallel.mjs index 0e31b81e3..3563cadc9 100644 --- a/packages/core/src/threading/parallel.mjs +++ b/packages/core/src/threading/parallel.mjs @@ -83,8 +83,9 @@ export default function createParallelWorker( /** * Processes items in parallel, yielding results in submission order. * - * The items can also arrive in batches, as a dependency streams them: each - * batch's chunks start as soon as it arrives, while earlier ones still run. + * The items can also arrive in batches, as a dependency streams them: a + * batch's chunks start while earlier ones still run, and the next batch is + * read once there is room for its chunks. * * @template T * @param {T[] | AsyncIterable} items - Items to process @@ -100,6 +101,9 @@ export default function createParallelWorker( const batches = streamed ? items : [items]; + // The chunks the batches read so far were split into + let distributed = 0; + /** * Splits a batch into chunks. * @@ -109,6 +113,8 @@ export default function createParallelWorker( const distribute = batch => { const chunks = createChunks(batch.length, chunkSize); + distributed += chunks.length; + parallelLogger.debug( `Distributing ${batch.length} items across ${chunks.length} chunks`, { generator: name, chunks: chunks.length, chunkSize, threads } @@ -130,9 +136,9 @@ export default function createParallelWorker( return; } - // Read every batch as it arrives, but start only a few chunks per thread - // at a time, so that generators running at once share the pool instead - // of one's chunks waiting behind all of another's. Emit results in + // Start only a few chunks per thread at a time, reading the next batch + // once there is room, so that generators running at once share the pool + // instead of one's chunks waiting behind all of another's. Emit results in // submission order. Handle errors as each task settles so a later failure // cannot become an unhandled rejection while earlier chunks are still // running. @@ -186,7 +192,8 @@ export default function createParallelWorker( }; /** - * Starts every batch's chunks as it arrives. + * Starts every batch's chunks, reading the next batch once there is room + * for them. */ const readBatches = async () => { try { @@ -233,7 +240,7 @@ export default function createParallelWorker( throw outcome.error; } - parallelLogger.debug(`Chunk ${index + 1}/${tasks.length} completed`, { + parallelLogger.debug(`Chunk ${index + 1}/${distributed} completed`, { generator: name, });