Skip to content

fix: close parallel section on mid-chain Parallel - #19

Merged
tr1v3r merged 1 commit into
masterfrom
fix/midchain-parallel-section
Sep 14, 2026
Merged

tr1v3r merged 1 commit into
masterfrom
fix/midchain-parallel-section

Conversation

@tr1v3r

@tr1v3r tr1v3r commented Sep 14, 2026

Copy link
Copy Markdown
Owner

Problem

A mid-chain Parallel(n) call did not actually close the current parallel section: stream.go's
Parallel() invoked s.ensureFlushed() but discarded its return value, so the section's
fused 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:

  • The two sections' fused stages merged into ONE pool sized by the second 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.
  • An Ordered() section lost its ordering guarantee at the next Parallel:
    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.
  • Advertised pipeline parallelism ("sections overlap") did not exist.

(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 a
cleared fused. Parallel() called it for its side effect only; since Go has no such side
effect 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:

s = *s.ensureFlushed()
s.parallelSize = n
s.ordered = false

The closed section's output becomes the new section's upstream. Each section now runs on its
own pool sized by its own Parallel call, and adjacent sections overlap naturally: the
downstream section's feeder pulls the upstream section's iter.Seq lazily, so both pools are
active 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 own
    2-worker pool even when Parallel(8) follows (fails on master: peak reaches 8).
  • TestParallelV2_MidChainParallelKeepsOrderedSection — ordered section + mid-chain
    Parallel + Limit(3) must select [0 1 2] (fails on master: [192 193 194]).
  • TestParallelV2_MidChainShortCircuitNoLeak — chained sections + infinite source +
    Limit short-circuit leaves no leaked goroutines.

Verified: go test ./... -race -count=1 green; new tests green under -race -count=3;
golangci-lint 0 issues.

Out of scope (follow-ups)

  • Ordered() called before Parallel() is still reset (separate PR).
  • Parallel(n <= 0) remains a no-op even with an open section (separate issue).

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
tr1v3r force-pushed the fix/midchain-parallel-section branch from bab6392 to ea94697 Compare September 14, 2026 02:31
@tr1v3r
tr1v3r merged commit 2e77ff7 into master Sep 14, 2026
2 checks passed
@tr1v3r
tr1v3r deleted the fix/midchain-parallel-section branch September 14, 2026 02:50
@tr1v3r
tr1v3r restored the fix/midchain-parallel-section branch September 14, 2026 02:52
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant