Repository navigation
Conversation
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 <noreply@anthropic.com> Assisted-by: Claude Sonnet 5.5 <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
ovflowd
changed the title
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #1237 +/- ##
==========================================
+ Coverage 94.14% 94.21% +0.06%
==========================================
Files 291 291
Lines 28950 29366 +416
Branches 2944 2985 +41
==========================================
+ Hits 27256 27666 +410
- Misses 1672 1678 +6
Partials 22 22 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
4 tasks done
Contributor
|
🚀 Deploying Preview to Cloudflare 🚀Preview Deployments by commit
|
ovflowd
marked this pull request as ready for review
November 3, 2026 20:39
avivkeller
reviewed
Nov 3, 2026
| let doneReading = false; | ||
|
|
||
| // Notified as chunks start, and once every batch is read | ||
| const changed = createSignal(); |
Member
There was a problem hiding this comment.
The waiting queue + settle/start/add + signal feels heavy. Could this just backpressure in one enqueue (await while at the in-flight limit) and wake the consumer with an inline resolvers pair?
| } | ||
|
|
||
| const chunks = createChunks(items.length, chunkSize); | ||
| const batches = toBatches(items); |
Member
There was a problem hiding this comment.
Nit: streamed ? items : [items] here would drop the need for toBatches.
| '@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. |
Member
There was a problem hiding this comment.
If toBatches goes away, drop the mention from the changeset too.
This branch was successfully deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
This PR lets
worker.streamtake an async iterable of batches as well as an array, starting each batch's work as it arrives. It's the first part of a stack that splits #1205, which lets generators start on their dependency's first chunk instead of waiting for all of it.CHUNKS_PER_THREAD, 3), so generators running at once share the worker pool instead of one filling it, and it yields results in the order the chunks were submitted.stricterFIFO. By default it puts a task it can't hand out yet at the back of its queue, so early heavy tasks like module pages ran last.toBatchesgives an input that is whole or streamed as batches, andcreateSignalis the wake-up the stream waits on.Nothing passes
worker.streama stream yet, and an input that arrives whole, as it does with one thread, behaves as before.Validation
Added tests in
parallel.test.mjsfor the streamed input: the limit on chunks in flight and its refill, the order tasks start in, a failure reading the batches arriving after the chunks before it, a single thread running each batch as it arrives, and no batches.toBatcheshas its own. Built on 12 of Node.js's API docs at 1 and 4 threads, the output is byte-identical tomain's.Related Issues
Refs: #1205
Check List
node --run testand all tests passed.node --run format:check&node --run lint.