Repository navigation
Conversation
|
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.13% 94.18% +0.04%
==========================================
Files 291 291
Lines 28900 29214 +314
Branches 2932 2969 +37
==========================================
+ Hits 27206 27515 +309
- Misses 1672 1677 +5
Partials 22 22 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
🚀 Deploying Preview to Cloudflare 🚀Preview Deployments by commit
|
| let doneReading = false; | ||
|
|
||
| // Notified as chunks start, and once every batch is read | ||
| const changed = createSignal(); |
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?
There was a problem hiding this comment.
Fair, it was heavy. There's a single enqueue now that waits while we're at the limit, and the consumer waits on an inline Promise.withResolvers(). No more waiting queue or signal. Arrays still kick off their first chunks right away, which the existing submission order tests rely on.
| } | ||
|
|
||
| const chunks = createChunks(items.length, chunkSize); | ||
| const batches = toBatches(items); |
There was a problem hiding this comment.
Nit: streamed ? items : [items] here would drop the need for toBatches.
There was a problem hiding this comment.
Yup, done. toBatches moved up to #1240, since groupWholeModules is the first thing that needs it (html and html-all use it too).
| '@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. |
There was a problem hiding this comment.
If toBatches goes away, drop the mention from the changeset too.
There was a problem hiding this comment.
Done, it's gone from here and #1240's changeset mentions it now.
e677946 to
51c7fb0
Compare
| * @param {(items: number[]) => number[] | AsyncIterable<number[]>} toInput | ||
| * @returns {Promise<void>} | ||
| */ | ||
| async function checkChunksPerThread(toInput) { |
There was a problem hiding this comment.
This seems extremely complex for a test...?
There was a problem hiding this comment.
Yeah, fair, it was overkill. It's just a counter now: each task takes a tick, and the test checks that the pool fills up to the limit, never past it, and that results still come out in order.
| for await (const batch of batches) { | ||
| for (const indices of distribute(batch)) { |
There was a problem hiding this comment.
Don't love the double for loop + yield await, that feels slow, is it?
There was a problem hiding this comment.
Not really. I measured it with a no-op processChunk: ~400ns per chunk vs ~370ns on main, and real chunks take milliseconds, so it's noise. With one thread it runs chunks one at a time on purpose, since nothing could run alongside and only one chunk's results are held at once.
51c7fb0 to
3acaa3d
Compare
| @@ -0,0 +1,5 @@ | |||
| --- | |||
| '@doc-kit/core': minor | |||
There was a problem hiding this comment.
| '@doc-kit/core': minor | |
| '@doc-kit/core': patch |
There was a problem hiding this comment.
Done, it's a patch now. Thanks for the cleanup commit too!
3db554e to
1f798eb
Compare
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>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
Each task now takes a turn of the event loop and counts how many run at once, instead of the test settling them by hand, newest first. It checks that the pool fills up to the limit and never past it, and that results come out in submission order; a stream that didn't refill the pool still hangs into the timeout. Assisted-by: Claude Opus 5.5 <noreply@anthropic.com>
As suggested in review. Assisted-by: Claude Opus 5.5 <noreply@anthropic.com>
1f798eb to
03fafb9
Compare
Description
This PR lets
worker.streamtake an async iterable of batches as well as an array, starting a batch's work while earlier ones still run. 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), whether its input is an array or streamed, so generators running at once share the worker pool instead of one filling it. It reads the next batch once there's room for its chunks, and 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.enqueuestarts a batch's chunks, waiting while the limit is reached, and the consumer waits on an inlinePromise.withResolvers()as chunks start. An array is read as[items], so its first chunks start right away.Nothing passes
worker.streama stream yet. With one thread nothing changes, and with more, an array's chunks now start a few per thread at a time instead of all at once, with the same output.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. 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.