From 2db7d1c9377efc410eb893f6059fa4a0b8894bf9 Mon Sep 17 00:00:00 2001 From: Soeren Schoenbrod Date: Tue, 17 Feb 2026 09:50:06 +0100 Subject: [PATCH] fix: return input channel directly when rechunk size matches When the input channel's chunk size already matches the requested output chunk size, return the input channel directly instead of creating an intermediate channel and spawning a rechunk task. This fixes a race condition in pipelines where the upstream producer uses a fixed-size buffer pool: with zero-copy passthrough, the same buffer references flow through both the input and intermediate output channels, extending the pipeline depth beyond the producer's pool size. Under slow-consumer conditions (e.g. JIT compilation on first run), the producer wraps around and overwrites buffers still queued downstream, causing data corruption. Co-Authored-By: Claude Opus 4.6 --- src/rechunk.jl | 11 +++++++++++ test/rechunk.jl | 11 +++++++++++ 2 files changed, 22 insertions(+) diff --git a/src/rechunk.jl b/src/rechunk.jl index 01d05c0..a105e79 100644 --- a/src/rechunk.jl +++ b/src/rechunk.jl @@ -293,6 +293,17 @@ output = rechunk(input, 1024) ``` """ function rechunk(in::SignalChannel{T,N}, chunk_size::Integer, channel_size=16) where {T<:Number,N} + # No-op passthrough: when input and output chunk sizes already match, return + # the input channel directly. This avoids creating an intermediate channel + # and task, which fixes a race condition: with zero-copy passthrough the same + # buffer references flow through both the input and output channels, extending + # the pipeline depth beyond the upstream producer's buffer pool size. Under + # slow-consumer conditions (e.g. JIT compilation on first run), the producer + # can wrap around and overwrite buffers still queued in the downstream channel. + if in.num_samples == chunk_size + return in + end + out = SignalChannel{T,N}(chunk_size, channel_size) # Estimate max outputs per input: input can complete partial + produce full chunks diff --git a/test/rechunk.jl b/test/rechunk.jl index 8bd9da4..8bda023 100644 --- a/test/rechunk.jl +++ b/test/rechunk.jl @@ -279,6 +279,17 @@ using FixedSizeArrays: FixedSizeMatrixDefault @test results[3] === inputs[3] end + @testset "rechunk channel - same size passthrough" begin + # When input and output chunk sizes match, rechunk should return + # the input channel directly (no intermediate channel or task). + # This prevents buffer-aliasing race conditions in pipelines where + # the upstream producer uses a fixed-size buffer pool. + input_chan = SignalChannel{ComplexF32,2}(1024) + output_chan = rechunk(input_chan, 1024) + + @test output_chan === input_chan + end + @testset "rechunk channel - upsampling" begin # Convert 100-sample chunks to 250-sample chunks input_chan = SignalChannel{ComplexF32,2}(100)