fix: close parallel section on mid-chain Parallel - #19
Merged
Merged
Conversation
This was referenced Sep 14, 2026
Parallel() invoked ensureFlushed() but discarded the returned streamer, so a mid-chain call never closed the open section: fused stages from both sections merged into one pool sized by the second Parallel's n, and an Ordered() section silently lost its guarantee — Parallel(2).Ordered().Map(f).Parallel(4).Limit(3) selected the wrong elements ([192 193 194] instead of [0 1 2]). Adopt the flushed streamer so the section's output becomes the new upstream; adjacent sections overlap lazily as documented. Adds three H2 regressions (separate pools, ordered-section survival, chained-section short- circuit without goroutine leaks). Audit finding H2.
tr1v3r
force-pushed
the
fix/midchain-parallel-section
branch
from
September 14, 2026 02:31
bab6392 to
ea94697
Compare
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.
Problem
A mid-chain
Parallel(n)call did not actually close the current parallel section:stream.go'sParallel()invokeds.ensureFlushed()but discarded its return value, so the section'sfused stages and the pre-section upstream stayed on the streamer. The documented contract
("A mid-chain call closes the current section and opens a new one" — export.go / README
"sections overlap" pipeline-parallelism) was silently violated:
Parallel's n:Parallel(8).Filter(slow).Parallel(2).Map(...)ran Filter with peak concurrency 2, not 8 —the first section's worker count was silently dropped.
Ordered()section lost its ordering guarantee at the nextParallel:Parallel(2).Ordered().Map(f).Parallel(4).Limit(3)returned[f(192), f(193), f(194)]instead of
[f(0), f(1), f(2)]— Limit selected wrong elements.(Element values were unaffected — results stayed correct as multisets, which is why the
existing test suite, which only checks multisets, never caught it.)
Root cause
ensureFlushed()returns a new streamer carrying the flushed pipeline (next.seq) and acleared
fused.Parallel()called it for its side effect only; since Go has no such sideeffect here, the call was a no-op and the returned flushed pipeline was thrown away.
Fix
Adopt the flushed streamer before opening the new section:
The closed section's output becomes the new section's upstream. Each section now runs on its
own pool sized by its own
Parallelcall, and adjacent sections overlap naturally: thedownstream section's feeder pulls the upstream section's
iter.Seqlazily, so both pools areactive concurrently with bounded backpressure — no channel bridge or extra machinery needed,
and the README's "sections overlap" wording becomes true as-is (no doc rewording required).
Cancellation still propagates cleanly across the chain (downstream terminal exit → downstream
cancel+drain → its feeder observes ctx on the next element → upstream consumer unwinds →
upstream cancel+drain); guarded by
TestParallelV2_MidChainShortCircuitNoLeak(runtime.NumGoroutine polling, same pattern as the existing single-section leak test).
Tests
TestParallelV2_MidChainParallelSeparatePools— section A must run on exactly its own2-worker pool even when
Parallel(8)follows (fails on master: peak reaches 8).TestParallelV2_MidChainParallelKeepsOrderedSection— ordered section + mid-chainParallel+Limit(3)must select[0 1 2](fails on master:[192 193 194]).TestParallelV2_MidChainShortCircuitNoLeak— chained sections + infinite source +Limitshort-circuit leaves no leaked goroutines.Verified:
go test ./... -race -count=1green; new tests green under-race -count=3;golangci-lint0 issues.Out of scope (follow-ups)
Ordered()called beforeParallel()is still reset (separate PR).Parallel(n <= 0)remains a no-op even with an open section (separate issue).