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)