From 8540d98b4df3a5a4a6d2faa55d2a793715dace87 Mon Sep 17 00:00:00 2001 From: Soeren Schoenbrod Date: Mon, 13 Jul 2026 13:25:08 +0000 Subject: [PATCH 1/4] feat!: default to Matrix buffers with optional backing array type SignalChannel, RechunkState, and the transforms (rechunk, mux, add) gain a backing matrix type parameter M that defaults to Matrix{T}. Plain matrices can now be put! directly; FixedSizeArrays is no longer required and becomes an opt-in backing type, e.g. SignalChannel{T,N,FixedSizeMatrixDefault{T}}. Benchmarks showed FixedSizeArrays gave no throughput benefit here: the channel passes references and the bulk copies compile to memmove regardless of the array type. FixedSizeArrays is therefore dropped from the package dependencies and is now a test-only dependency. BREAKING CHANGE: SignalChannel is now SignalChannel{T,N,M}; the default eltype is Matrix{T} instead of FixedSizeMatrixDefault{T}. Code relying on take! returning a FixedSizeMatrixDefault must construct channels with the explicit backing type SignalChannel{T,N,FixedSizeMatrixDefault{T}} or adapt to Matrix{T}. Co-Authored-By: Claude Opus 4.8 (1M context) --- Project.toml | 4 +- README.md | 20 ++++ benchmark/channel_benchmarks.jl | 5 +- benchmark/rechunk_benchmarks.jl | 3 +- benchmark/rechunk_state_benchmarks.jl | 7 +- benchmark/soapysdr_benchmarks.jl | 11 +-- ext/SignalChannelsSoapySDRExt.jl | 13 ++- src/channel_combine.jl | 18 ++-- src/channel_utilities.jl | 2 +- src/rechunk.jl | 60 +++++++----- src/signal_channel.jl | 134 ++++++++++++++++---------- test/backing_type.jl | 67 +++++++++++++ test/channel_combine.jl | 39 ++++---- test/channel_utilities.jl | 17 ++-- test/periodogram.jl | 7 +- test/rechunk.jl | 43 ++++----- test/runtests.jl | 1 + test/signal_channel.jl | 29 +++--- test/soapysdr_ext.jl | 7 +- test/stream_utilities.jl | 11 +-- 20 files changed, 306 insertions(+), 192 deletions(-) create mode 100644 test/backing_type.jl diff --git a/Project.toml b/Project.toml index f01a923..ee08c1f 100644 --- a/Project.toml +++ b/Project.toml @@ -5,7 +5,6 @@ authors = ["Soeren Schoenbrod "] [deps] DSP = "717857b8-e6f2-59f4-9121-6e50c889abd2" -FixedSizeArrays = "3821ddf9-e5b5-40d5-8e25-6813ab96b5e2" LiveLayoutUnicodePlots = "8b8f1234-5678-9012-3456-789012345678" PipeChannels = "316ea068-b8d4-4418-b34f-158552ec34f5" UnicodePlots = "b8865327-cd53-5732-bb35-84acbb429228" @@ -29,9 +28,10 @@ Unitful = "1" julia = "1.12" [extras] +FixedSizeArrays = "3821ddf9-e5b5-40d5-8e25-6813ab96b5e2" SoapyLoopback_jll = "8d250ce8-a086-56bd-8a36-e4e8f4202680" SoapySDR = "4fa426e6-03fd-45b2-b2dd-25f726820c03" Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40" [targets] -test = ["Test", "SoapySDR", "SoapyLoopback_jll"] +test = ["Test", "SoapySDR", "SoapyLoopback_jll", "FixedSizeArrays"] diff --git a/README.md b/README.md index 32c9c38..a1e72d5 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,26 @@ for data in chan end ``` +### Backing array type + +By default a `SignalChannel` stores buffers as plain `Matrix{T}` (Julia's built-in +dense array), so you can `put!` ordinary matrices directly. The backing matrix type +is a type parameter, so you can opt into another `AbstractMatrix{T}` when you want its +semantics — for example `FixedSizeArrays.FixedSizeMatrixDefault{T}`, which guarantees +buffer dimensions cannot change after creation: + +```julia +using FixedSizeArrays: FixedSizeMatrixDefault + +# 4 antenna channels, 1024 samples, backed by FixedSizeMatrixDefault +chan = SignalChannel{ComplexF32,4,FixedSizeMatrixDefault{ComplexF32}}(1024) +``` + +The channel only ever passes references to buffers, so the backing type has no +measurable effect on channel throughput — choose it for the semantics you want, not +for speed. Transforms like `rechunk`, `mux`, and `add` preserve the backing type of +their input channel. + ### Rechunking ```julia diff --git a/benchmark/channel_benchmarks.jl b/benchmark/channel_benchmarks.jl index 685cca2..261bdd5 100644 --- a/benchmark/channel_benchmarks.jl +++ b/benchmark/channel_benchmarks.jl @@ -1,6 +1,5 @@ using BenchmarkTools using SignalChannels -using FixedSizeArrays: FixedSizeMatrixDefault # Detect API version: new API has num_antenna_channels as type parameter (SignalChannel{T,N}) # Old API has it as constructor argument (SignalChannel{T}(num_samples, num_channels, buffer_size)) @@ -20,12 +19,12 @@ function setup_channel_benchmark(num_samples::Int, buffer_size::Int) SignalChannel{ComplexF32}(num_samples, 1, buffer_size) end - data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, num_samples, 1)) + data = Matrix{ComplexF32}(rand(ComplexF32, num_samples, 1)) return (ch, data) end # Benchmark: push data through the channel and drain output -function run_channel_benchmark!(ch, data::FixedSizeMatrixDefault{ComplexF32}, num_items::Int) +function run_channel_benchmark!(ch, data::Matrix{ComplexF32}, num_items::Int) # Producer task producer = Threads.@spawn begin for _ in 1:num_items diff --git a/benchmark/rechunk_benchmarks.jl b/benchmark/rechunk_benchmarks.jl index 62702d4..0d814fa 100644 --- a/benchmark/rechunk_benchmarks.jl +++ b/benchmark/rechunk_benchmarks.jl @@ -1,6 +1,5 @@ using BenchmarkTools using SignalChannels -using FixedSizeArrays: FixedSizeMatrixDefault # Detect API version: new API has num_antenna_channels as type parameter (SignalChannel{T,N}) # Old API has it as constructor argument (SignalChannel{T}(num_samples, num_channels, buffer_size)) @@ -25,7 +24,7 @@ function setup_pipeline(input_size::Int, output_size::Int, output_channel_size:: # Create the rechunk pipeline - this spawns the task but it blocks waiting for input output = rechunk(input, output_size, output_channel_size) - buffers = [FixedSizeMatrixDefault{ComplexF32}(zeros(ComplexF32, input_size, 1)) for _ in 1:NUM_BUFFERS] + buffers = [Matrix{ComplexF32}(zeros(ComplexF32, input_size, 1)) for _ in 1:NUM_BUFFERS] return (input, output, buffers) end diff --git a/benchmark/rechunk_state_benchmarks.jl b/benchmark/rechunk_state_benchmarks.jl index 0bb7989..87f2065 100644 --- a/benchmark/rechunk_state_benchmarks.jl +++ b/benchmark/rechunk_state_benchmarks.jl @@ -6,7 +6,6 @@ const RECHUNK_STATE_SUPPORTED = isdefined(SignalChannels, :RechunkState) if RECHUNK_STATE_SUPPORTED using SignalChannels: RechunkState, rechunk!, reset! - using FixedSizeArrays: FixedSizeMatrixDefault # ============================================================================ # RechunkState / rechunk! Benchmarks @@ -37,7 +36,7 @@ if RECHUNK_STATE_SUPPORTED max_output_buffers = cld(total_input_samples, output_size) + 2 state = RechunkState{T,N}(output_size, max_output_buffers, max_output_buffers) - inputs = [FixedSizeMatrixDefault{T}(rand(T, input_size, N)) for _ in 1:num_buffers] + inputs = [Matrix{T}(rand(T, input_size, N)) for _ in 1:num_buffers] return (state, inputs) end @@ -193,7 +192,7 @@ if RECHUNK_STATE_SUPPORTED # Inner function with Val{N} for compile-time specialization function _setup_passthrough(::Type{T}, size::Int, ::Val{N}, num_buffers::Int) where {T,N} state = RechunkState{T,N}(size, num_buffers + 2, num_buffers + 2) - inputs = [FixedSizeMatrixDefault{T}(rand(T, size, N)) for _ in 1:num_buffers] + inputs = [Matrix{T}(rand(T, size, N)) for _ in 1:num_buffers] return (state, inputs) end @@ -206,7 +205,7 @@ if RECHUNK_STATE_SUPPORTED function _setup_near_passthrough(::Type{T}, size::Int, ::Val{N}, num_buffers::Int) where {T,N} # Output is 1 sample smaller, so copies are required state = RechunkState{T,N}(size - 1, num_buffers + 100, num_buffers + 100) - inputs = [FixedSizeMatrixDefault{T}(rand(T, size, N)) for _ in 1:num_buffers] + inputs = [Matrix{T}(rand(T, size, N)) for _ in 1:num_buffers] return (state, inputs) end diff --git a/benchmark/soapysdr_benchmarks.jl b/benchmark/soapysdr_benchmarks.jl index 862fe9c..0313684 100644 --- a/benchmark/soapysdr_benchmarks.jl +++ b/benchmark/soapysdr_benchmarks.jl @@ -1,6 +1,5 @@ using BenchmarkTools using SignalChannels -using FixedSizeArrays: FixedSizeMatrixDefault using Test: @testset, @test # Detect API version: new API has num_antenna_channels as type parameter (SignalChannel{T,N}) @@ -64,7 +63,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER end # Pre-allocate test data - data = FixedSizeMatrixDefault{ComplexF32}(zeros(ComplexF32, chunk_size, 1)) + data = Matrix{ComplexF32}(zeros(ComplexF32, chunk_size, 1)) for i in 1:chunk_size data[i, 1] = ComplexF32(i / chunk_size, 0.5f0) end @@ -111,7 +110,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER # Benchmark: Single put! operation on SignalChannel # Should show zero or constant allocations per call - function benchmark_single_put!(channel, data::FixedSizeMatrixDefault{ComplexF32}) + function benchmark_single_put!(channel, data::Matrix{ComplexF32}) put!(channel, data) return nothing end @@ -156,7 +155,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER tx_stats_channel, tx_warning_channel = stream_data(device_args, config, tx_channel) # Create test data buffer - test_data = FixedSizeMatrixDefault{ComplexF32}(zeros(ComplexF32, chunk_size, 1)) + test_data = Matrix{ComplexF32}(zeros(ComplexF32, chunk_size, 1)) for i in 1:chunk_size test_data[i, 1] = ComplexF32(i / chunk_size, (chunk_size - i) / chunk_size) end @@ -189,7 +188,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER function benchmark_loopback!( tx_channel, rx_channel, - data::FixedSizeMatrixDefault{ComplexF32}, + data::Matrix{ComplexF32}, num_buffers::Int, ) # Start producer @@ -219,7 +218,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER function benchmark_loopback_single_take!( tx_channel, rx_channel, - data::FixedSizeMatrixDefault{ComplexF32}, + data::Matrix{ComplexF32}, ) # Send one buffer via TX put!(tx_channel, data) diff --git a/ext/SignalChannelsSoapySDRExt.jl b/ext/SignalChannelsSoapySDRExt.jl index 1a0c099..a9ab528 100644 --- a/ext/SignalChannelsSoapySDRExt.jl +++ b/ext/SignalChannelsSoapySDRExt.jl @@ -3,7 +3,6 @@ module SignalChannelsSoapySDRExt using SignalChannels using SoapySDR using Unitful -using FixedSizeArrays: FixedSizeMatrixDefault using DSP: hamming # Helper to check if a value is within any of the given ranges @@ -271,7 +270,7 @@ function SignalChannels.stream_data( end end - setup_channel = Channel{SignalChannel{T,N}}(1) + setup_channel = Channel{SignalChannel{T,N,Matrix{T}}}(1) warning_channel = Channel{SignalChannels.StreamWarning}(warning_buffer_size) task = Threads.@spawn begin @@ -312,7 +311,7 @@ function SignalChannels.stream_data( # When not in passthrough mode, rechunk! copies to its internal buffer pool, # so fewer input buffers would suffice, but we use the same count for simplicity. num_input_buffers = buffers_in_flight + 2 - input_buffer_pool = [FixedSizeMatrixDefault{T}(fill(zero(T), mtu, nchannels)) for _ in 1:num_input_buffers] + input_buffer_pool = [zeros(T, mtu, nchannels) for _ in 1:num_input_buffers] input_buffer_idx = 1 SoapySDR.activate!(stream) do @@ -453,10 +452,10 @@ end function SignalChannels.stream_data( dev_args, configs::NTuple{N,SignalChannels.SDRChannelConfig}, - in::SignalChannel{T,N}; + in::SignalChannel{T,N,M}; warning_buffer_size::Integer=16, stats_buffer_size::Integer=1000, -) where {T<:Number,N} +) where {T<:Number,N,M} if Threads.nthreads() < 2 error("stream_data requires Julia to be started with multiple threads. " * "Start Julia with `julia --threads=auto` or set JULIA_NUM_THREADS environment variable.") @@ -506,10 +505,10 @@ function SignalChannels.stream_data( max_outputs_per_batch = cld(batch_size * input_chunk_size, Int(mtu)) + 1 # For TX we don't need as many buffers since we write synchronously num_buffers = max_outputs_per_batch + 2 - rechunk_state = SignalChannels.RechunkState{T,N}(Int(mtu), num_buffers, max_outputs_per_batch) + rechunk_state = SignalChannels.RechunkState{T,N,M}(Int(mtu), num_buffers, max_outputs_per_batch) # Pre-allocate batch buffer for batch takes - input_batch = Vector{FixedSizeMatrixDefault{T}}(undef, batch_size) + input_batch = Vector{M}(undef, batch_size) SoapySDR.activate!(stream) do timeout_us = Int(uconvert(u"μs", 0.9u"s").val) diff --git a/src/channel_combine.jl b/src/channel_combine.jl index 22023ec..942af6c 100644 --- a/src/channel_combine.jl +++ b/src/channel_combine.jl @@ -1,5 +1,3 @@ -using FixedSizeArrays: FixedSizeMatrixDefault - """ mux(ch1::SignalChannel{T,N}, ch2::SignalChannel{T,N}; channel_size::Integer=10, sync::Bool=true) where {T,N} @@ -40,24 +38,24 @@ out = mux(ch1, ch2; sync=false) ``` """ function mux( - ch1::SignalChannel{T,N}, - ch2::SignalChannel{T,N}; + ch1::SignalChannel{T,N,M}, + ch2::SignalChannel{T,N,M}; channel_size::Integer = 10, sync::Bool = true, -) where {T,N} +) where {T,N,M} if ch1.num_samples != ch2.num_samples error( "mux requires both channels to have the same num_samples. Got $(ch1.num_samples) and $(ch2.num_samples)", ) end num_samples = ch1.num_samples - out = SignalChannel{T,N}(num_samples, channel_size) + out = SignalChannel{T,N,M}(num_samples, channel_size) # Pre-allocate buffer slots to avoid allocation in the hot loop. # channel_size + 2 ensures we never overwrite a buffer still in flight # in the output channel. num_slots = channel_size + 2 - buffer_slots = [FixedSizeMatrixDefault{T}(undef, num_samples, N) for _ = 1:num_slots] + buffer_slots = [M(undef, num_samples, N) for _ = 1:num_slots] task = let ch1 = ch1, @@ -131,7 +129,7 @@ ch3 = SignalChannel{ComplexF32,1}(1024, 10) out = add(ch1, ch2, ch3) ``` """ -function add(channels::SignalChannel{T,N}...; channel_size::Integer = 10) where {T,N} +function add(channels::SignalChannel{T,N,M}...; channel_size::Integer = 10) where {T,N,M} length(channels) >= 2 || throw(ArgumentError("add requires at least 2 channels, got $(length(channels))")) @@ -145,14 +143,14 @@ function add(channels::SignalChannel{T,N}...; channel_size::Integer = 10) where end end - out = SignalChannel{T,N}(num_samples, channel_size) + out = SignalChannel{T,N,M}(num_samples, channel_size) rest = channels[2:end] # Pre-allocate buffer slots to avoid allocation in the hot loop. # channel_size + 2 ensures we never overwrite a buffer still in flight # in the output channel. num_slots = channel_size + 2 - buffer_slots = [FixedSizeMatrixDefault{T}(undef, num_samples, N) for _ = 1:num_slots] + buffer_slots = [M(undef, num_samples, N) for _ = 1:num_slots] task = let channels = channels, diff --git a/src/channel_utilities.jl b/src/channel_utilities.jl index cbae494..b2bf8f3 100644 --- a/src/channel_utilities.jl +++ b/src/channel_utilities.jl @@ -174,7 +174,7 @@ function read_from_file(file_path::String, num_samples::Integer, num_antenna_cha try while !any(eof, streams) # Create buffer for this chunk - buff = FixedSizeMatrixDefault{T}(undef, num_samples, num_antenna_channels) + buff = Matrix{T}(undef, num_samples, num_antenna_channels) # Read data for each channel all_complete = true diff --git a/src/rechunk.jl b/src/rechunk.jl index a105e79..96d2b2c 100644 --- a/src/rechunk.jl +++ b/src/rechunk.jl @@ -1,5 +1,5 @@ """ - RechunkState{T,N} + RechunkState{T,N,M} Mutable state for rechunking operations. Holds pre-allocated buffers and tracks the current position in the output chunk being filled. @@ -11,13 +11,14 @@ enabling the compiler to unroll the per-channel copy loop for zero allocations. # Type Parameters - `T`: Element type of the data being rechunked - `N`: Number of antenna channels (compile-time constant for loop unrolling) +- `M`: Backing matrix type, `M <: AbstractMatrix{T}` (default: `Matrix{T}`) # Fields - `output_chunk_size::Int`: Number of samples per output chunk -- `buffer_pool::Vector{FixedSizeMatrixDefault{T}}`: Pre-allocated output buffers +- `buffer_pool::Vector{M}`: Pre-allocated output buffers - `buffer_idx::Int`: Current index into buffer_pool (cycles through) - `chunk_filled::Int`: Number of samples currently in the buffer being filled -- `output_vector::Vector{FixedSizeMatrixDefault{T}}`: Pre-allocated vector for rechunk! results +- `output_vector::Vector{M}`: Pre-allocated vector for rechunk! results # Examples ```julia @@ -36,33 +37,38 @@ end partial = get_partial_buffer(state) ``` """ -mutable struct RechunkState{T,N} +mutable struct RechunkState{T,N,M<:AbstractMatrix{T}} output_chunk_size::Int - buffer_pool::Vector{FixedSizeMatrixDefault{T}} + buffer_pool::Vector{M} buffer_idx::Int chunk_filled::Int - output_vector::Vector{FixedSizeMatrixDefault{T}} + output_vector::Vector{M} output_count::Int - function RechunkState{T,N}(output_chunk_size::Integer, num_buffers::Integer, max_outputs_per_input::Integer) where {T,N} - buffer_pool = [FixedSizeMatrixDefault{T}(undef, output_chunk_size, N) for _ in 1:num_buffers] - output_vector = Vector{FixedSizeMatrixDefault{T}}(undef, max_outputs_per_input) - return new{T,N}(output_chunk_size, buffer_pool, 1, 0, output_vector, 0) + function RechunkState{T,N,M}(output_chunk_size::Integer, num_buffers::Integer, max_outputs_per_input::Integer) where {T,N,M} + buffer_pool = [M(undef, output_chunk_size, N) for _ in 1:num_buffers] + output_vector = Vector{M}(undef, max_outputs_per_input) + return new{T,N,M}(output_chunk_size, buffer_pool, 1, 0, output_vector, 0) end end +# Default backing type to Matrix{T} when M is not specified +function RechunkState{T,N}(output_chunk_size::Integer, num_buffers::Integer, max_outputs_per_input::Integer) where {T,N} + return RechunkState{T,N,Matrix{T}}(output_chunk_size, num_buffers, max_outputs_per_input) +end + # Convenience constructor that takes nchannels as runtime value # max_outputs_per_input defaults to num_buffers (conservative estimate) # Note: This constructor may not fully specialize the inner loop. For best performance, # use the Val-based constructor below. function RechunkState{T}(output_chunk_size::Integer, nchannels::Integer, num_buffers::Integer; max_outputs_per_input::Integer=num_buffers) where {T} - return RechunkState{T,nchannels}(output_chunk_size, num_buffers, max_outputs_per_input) + return RechunkState{T,nchannels,Matrix{T}}(output_chunk_size, num_buffers, max_outputs_per_input) end # Val-based constructor for compile-time specialization of channel count # This ensures the per-channel copy loop is fully unrolled for zero allocations function RechunkState{T}(output_chunk_size::Integer, ::Val{N}, num_buffers::Integer; max_outputs_per_input::Integer=num_buffers) where {T,N} - return RechunkState{T,N}(output_chunk_size, num_buffers, max_outputs_per_input) + return RechunkState{T,N,Matrix{T}}(output_chunk_size, num_buffers, max_outputs_per_input) end # Accessor for number of channels (from type parameter) @@ -146,7 +152,7 @@ end end """ - rechunk!(state::RechunkState{T,N}, input::FixedSizeMatrixDefault{T}) where {T,N} + rechunk!(state::RechunkState{T,N}, input::AbstractMatrix{T}) where {T,N} Process an input buffer through the rechunk state, returning a view of completed output buffers. @@ -157,11 +163,13 @@ performance (~600 M samples/s for ComplexF32). **Zero-copy passthrough**: When `input` exactly matches the output chunk size and no partial data is buffered, the input is included directly without copying. This means the caller must not modify or reuse the input buffer after passing it -to `rechunk!` if they need the output to remain valid. +to `rechunk!` if they need the output to remain valid. For zero-copy passthrough +to store the input by reference, `input` should match the state's backing type +`M`; otherwise it is converted into the pre-allocated `M` buffers. # Arguments - `state`: RechunkState holding pre-allocated buffers and current position -- `input`: Input matrix to rechunk (samples × channels), must be `FixedSizeMatrixDefault{T}` +- `input`: Input matrix to rechunk (samples × channels) # Returns A `SubArray` view into the state's output vector containing the completed buffers. @@ -178,14 +186,14 @@ outputs = rechunk!(state, input) put!(channel, outputs) # Batch put all outputs at once ``` """ -@inline function rechunk!(state::RechunkState{T,N}, input::FixedSizeMatrixDefault{T}) where {T,N} +@inline function rechunk!(state::RechunkState{T,N}, input::AbstractMatrix{T}) where {T,N} output_count = rechunk_one!(state, input, 0) state.output_count = output_count return view(state.output_vector, 1:output_count) end """ - rechunk!(state::RechunkState{T,N}, inputs::AbstractVector{<:FixedSizeMatrixDefault{T}}) where {T,N} + rechunk!(state::RechunkState{T,N}, inputs::AbstractVector{<:AbstractMatrix{T}}) where {T,N} Process multiple input buffers through the rechunk state in a single call, returning a view of all completed output buffers. @@ -197,7 +205,7 @@ This batch version is more efficient than calling `rechunk!` repeatedly because: # Arguments - `state`: RechunkState holding pre-allocated buffers and current position -- `inputs`: Vector (or view) of `FixedSizeMatrixDefault{T}` input matrices to rechunk +- `inputs`: Vector (or view) of `AbstractMatrix{T}` input matrices to rechunk # Returns A `SubArray` view into the state's output vector containing all completed buffers @@ -206,7 +214,7 @@ from processing all inputs. The view is valid until the next call to `rechunk!`. # Examples ```julia state = RechunkState{ComplexF32}(1024, 4, 20; max_outputs_per_input=10) -input_batch = Vector{FixedSizeMatrixDefault{ComplexF32}}(undef, 4) +input_batch = Vector{Matrix{ComplexF32}}(undef, 4) # ... fill input_batch ... num_taken = take!(channel, input_batch) @@ -215,7 +223,7 @@ outputs = rechunk!(state, @view input_batch[1:num_taken]) put!(output_channel, outputs) ``` """ -@inline function rechunk!(state::RechunkState{T,N}, inputs::AbstractVector{<:FixedSizeMatrixDefault{T}}) where {T,N} +@inline function rechunk!(state::RechunkState{T,N}, inputs::AbstractVector{<:AbstractMatrix{T}}) where {T,N} output_count = 0 for input in inputs output_count = rechunk_one!(state, input, output_count) @@ -292,7 +300,7 @@ input = SignalChannel{ComplexF32,4}(512) output = rechunk(input, 1024) ``` """ -function rechunk(in::SignalChannel{T,N}, chunk_size::Integer, channel_size=16) where {T<:Number,N} +function rechunk(in::SignalChannel{T,N,M}, chunk_size::Integer, channel_size=16) where {T<:Number,N,M} # 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 @@ -304,7 +312,7 @@ function rechunk(in::SignalChannel{T,N}, chunk_size::Integer, channel_size=16) w return in end - out = SignalChannel{T,N}(chunk_size, channel_size) + out = SignalChannel{T,N,M}(chunk_size, channel_size) # Estimate max outputs per input: input can complete partial + produce full chunks # For downsampling (large input -> small output), this could be large @@ -318,15 +326,15 @@ function rechunk(in::SignalChannel{T,N}, chunk_size::Integer, channel_size=16) w num_buffers = channel_size + max_outputs + 2 # N is now a compile-time constant from the type parameter - task = Threads.@spawn _rechunk_task(T, Val(N), in, out, chunk_size, num_buffers, max_outputs) + task = Threads.@spawn _rechunk_task(T, Val(N), M, in, out, chunk_size, num_buffers, max_outputs) bind(out, task) bind(in, task) # Propagate errors upstream return out end -# Inner task function with compile-time N for zero-allocation rechunking -function _rechunk_task(::Type{T}, ::Val{N}, in, out, chunk_size, num_buffers, max_outputs) where {T,N} - state = RechunkState{T,N}(chunk_size, num_buffers, max_outputs) +# Inner task function with compile-time N and backing type M for zero-allocation rechunking +function _rechunk_task(::Type{T}, ::Val{N}, ::Type{M}, in, out, chunk_size, num_buffers, max_outputs) where {T,N,M} + state = RechunkState{T,N,M}(chunk_size, num_buffers, max_outputs) for data in in outputs = rechunk!(state, data) diff --git a/src/signal_channel.jl b/src/signal_channel.jl index 8cea7f1..51331ca 100644 --- a/src/signal_channel.jl +++ b/src/signal_channel.jl @@ -1,5 +1,4 @@ import Base.close, Base.put!, Base.close, Base.isempty -using FixedSizeArrays: FixedSizeMatrixDefault using PipeChannels: PipeChannel """ @@ -47,7 +46,7 @@ struct TxStats end """ - SignalChannel{T,N} <: AbstractChannel{T} + SignalChannel{T,N,M} <: AbstractChannel{T} A specialized channel type that enforces matrix dimensions for multi-channel signal data. This ensures type safety when working with multi-antenna or multi-channel signal processing @@ -56,12 +55,16 @@ applications. The number of antenna channels `N` is a type parameter, enabling compile-time specialization for zero-allocation performance in tight loops. -Data is always stored as a fixed-size matrix with dimensions `(num_samples, N)`. +Data is stored as a matrix with dimensions `(num_samples, N)`. For single-channel signals (`N = 1`), this results in a column vector represented as a matrix with shape `(num_samples, 1)`. -The use of FixedSizeMatrixDefault ensures that buffer dimensions cannot be changed after creation, -providing additional safety guarantees. +The backing matrix type `M` is also a type parameter and defaults to `Matrix{T}` +(Julia's built-in dense array). Any `AbstractMatrix{T}` may be used instead — for +example `FixedSizeArrays.FixedSizeMatrixDefault{T}` (which guarantees the buffer +dimensions cannot change after creation) or a `StaticArrays` type. Because the +channel only ever passes *references* to buffers, the choice of `M` has no +measurable effect on channel throughput; pick it for the semantics you want. Uses a lock-free PipeChannel internally for zero-allocation performance in real-time applications. @@ -71,42 +74,52 @@ may call `take!`. Multiple producers or consumers will cause data races. # Type Parameters - `T`: Element type (e.g., `ComplexF32`, `Float64`) - `N`: Number of antenna channels (compile-time constant) +- `M`: Backing matrix type, `M <: AbstractMatrix{T}` (default: `Matrix{T}`) # Fields - `num_samples::Int`: Number of samples per buffer (rows) -- `channel::PipeChannel{FixedSizeMatrixDefault{T}}`: Underlying lock-free channel with fixed-size matrices +- `channel::PipeChannel{M}`: Underlying lock-free channel of buffers # Examples ```julia -# Create a single-channel for 1024 samples (shape: 1024×1) +# Create a single-channel for 1024 samples (shape: 1024×1), backed by Matrix{ComplexF32} chan = SignalChannel{ComplexF32}(1024) # or explicitly: SignalChannel{ComplexF32,1}(1024) # Create a channel for 1024 samples across 4 antenna channels (shape: 1024×4) chan = SignalChannel{ComplexF32,4}(1024) +# Opt into a different backing matrix type (e.g. FixedSizeArrays) +using FixedSizeArrays: FixedSizeMatrixDefault +chan = SignalChannel{ComplexF32,4,FixedSizeMatrixDefault{ComplexF32}}(1024) + # Put data (must match dimensions) data = rand(ComplexF32, 1024, 1) # Single channel put!(chan, data) # Take data -received = take!(chan) # Returns FixedSizeMatrixDefault{ComplexF32} with size (1024, 1) or (1024, 4) +received = take!(chan) # Returns an M with size (1024, 1) or (1024, 4) ``` """ -struct SignalChannel{T,N} <: AbstractChannel{T} +struct SignalChannel{T,N,M<:AbstractMatrix{T}} <: AbstractChannel{T} num_samples::Int - channel::PipeChannel{FixedSizeMatrixDefault{T}} - function SignalChannel{T,N}( + channel::PipeChannel{M} + function SignalChannel{T,N,M}( num_samples::Integer, sz::Integer=16, - ) where {T,N} - return new{T,N}(num_samples, PipeChannel{FixedSizeMatrixDefault{T}}(sz)) + ) where {T,N,M} + return new{T,N,M}(num_samples, PipeChannel{M}(sz)) end end -# Convenience constructor: SignalChannel{T}(num_samples) defaults to N=1 +# Convenience constructor: SignalChannel{T,N}(num_samples) defaults the backing type to Matrix{T} +function SignalChannel{T,N}(num_samples::Integer, sz::Integer=16) where {T,N} + return SignalChannel{T,N,Matrix{T}}(num_samples, sz) +end + +# Convenience constructor: SignalChannel{T}(num_samples) defaults to N=1, M=Matrix{T} function SignalChannel{T}(num_samples::Integer, sz::Integer=16) where {T} - return SignalChannel{T,1}(num_samples, sz) + return SignalChannel{T,1,Matrix{T}}(num_samples, sz) end # Accessor for number of antenna channels (from type parameter) @@ -143,14 +156,14 @@ chan = SignalChannel{ComplexF32,4}(1024, 10) do c end ``` """ -function SignalChannel{T,N}( +function SignalChannel{T,N,M}( func::Function, num_samples::Integer, size=16; taskref=nothing, spawn=false, -) where {T,N} - chnl = SignalChannel{T,N}(num_samples, size) +) where {T,N,M} + chnl = SignalChannel{T,N,M}(num_samples, size) task = Task(() -> func(chnl)) task.sticky = !spawn bind(chnl, task) @@ -163,7 +176,18 @@ function SignalChannel{T,N}( return chnl end -# Convenience: SignalChannel{T}(func, num_samples, size) defaults to N=1 +# Convenience: SignalChannel{T,N}(func, ...) defaults the backing type to Matrix{T} +function SignalChannel{T,N}( + func::Function, + num_samples::Integer, + size=16; + taskref=nothing, + spawn=false, +) where {T,N} + return SignalChannel{T,N,Matrix{T}}(func, num_samples, size; taskref=taskref, spawn=spawn) +end + +# Convenience: SignalChannel{T}(func, num_samples, size) defaults to N=1, M=Matrix{T} function SignalChannel{T}( func::Function, num_samples::Integer, @@ -171,19 +195,10 @@ function SignalChannel{T}( taskref=nothing, spawn=false, ) where {T} - return SignalChannel{T,1}(func, num_samples, size; taskref=taskref, spawn=spawn) + return SignalChannel{T,1,Matrix{T}}(func, num_samples, size; taskref=taskref, spawn=spawn) end -""" - put!(c::SignalChannel{T,N}, v::FixedSizeMatrixDefault) - -Put a fixed-size matrix into the channel. Validates that the matrix dimensions match the channel's -`num_samples` and `N` (number of antenna channels). - -# Throws -- `ArgumentError`: If matrix dimensions don't match the channel configuration -""" -function Base.put!(c::SignalChannel{T,N}, v::FixedSizeMatrixDefault{T}) where {T,N} +@inline function _check_put_dims(c::SignalChannel{T,N}, v) where {T,N} if size(v, 1) != c.num_samples || size(v, 2) != N throw( ArgumentError( @@ -191,17 +206,32 @@ function Base.put!(c::SignalChannel{T,N}, v::FixedSizeMatrixDefault{T}) where {T ), ) end + return nothing +end + +""" + put!(c::SignalChannel{T,N,M}, v::AbstractMatrix{T}) + +Put a matrix into the channel. Validates that the matrix dimensions match the channel's +`num_samples` and `N` (number of antenna channels). + +If `v` is already of the channel's backing type `M` it is stored by reference +(zero-copy). This is the common case for the default `M = Matrix{T}`. Otherwise `v` +is converted into an `M` via the `M(v)` constructor, which allocates a new buffer. + +# Throws +- `ArgumentError`: If matrix dimensions don't match the channel configuration +""" +# Fast path: value is already the backing type — store it by reference (zero-copy). +function Base.put!(c::SignalChannel{T,N,M}, v::M) where {T,N,M} + _check_put_dims(c, v) Base.put!(c.channel, v) end -# Prevent accidental use of regular matrices - only FixedSizeMatrixDefault is allowed for performance -function Base.put!(c::SignalChannel{T,N}, v::AbstractMatrix{T}) where {T,N} - throw( - ArgumentError( - "SignalChannel only accepts FixedSizeMatrixDefault for performance. " * - "Got $(typeof(v)). Convert with: FixedSizeMatrixDefault{$T}(your_matrix)", - ), - ) +# Fallback: convert any other matrix into the backing type via its constructor. +function Base.put!(c::SignalChannel{T,N,M}, v::AbstractMatrix{T}) where {T,N,M} + _check_put_dims(c, v) + Base.put!(c.channel, M(v)::M) end # Delegate Base methods to the underlying channel @@ -215,15 +245,17 @@ Base.close(c::SignalChannel, excp::Exception=Base.closed_exception()) = # ============================================================================ """ - put!(c::SignalChannel{T,N}, values::AbstractVector{<:FixedSizeMatrixDefault{T}}) where {T,N} + put!(c::SignalChannel{T,N,M}, values::AbstractVector{<:AbstractMatrix{T}}) where {T,N,M} -Add multiple fixed-size matrices to the channel in a single batch operation. +Add multiple matrices to the channel in a single batch operation. Blocks until all items are written. Returns the input vector. This is more efficient than calling `put!` repeatedly because it uses the underlying PipeChannel's batch operation, reducing atomic overhead. -All matrices must match the channel's `num_samples` and `N` (number of antenna channels). +The element type of `values` must be the channel's backing type `M` (e.g. a +`Vector{M}`). All matrices must match the channel's `num_samples` and `N` +(number of antenna channels). # Throws - `ArgumentError`: If any matrix dimensions don't match the channel configuration @@ -232,11 +264,11 @@ All matrices must match the channel's `num_samples` and `N` (number of antenna c # Examples ```julia chan = SignalChannel{ComplexF32,4}(1024) -buffers = [FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 1024, 4)) for _ in 1:8] +buffers = [rand(ComplexF32, 1024, 4) for _ in 1:8] put!(chan, buffers) # Batch put all 8 buffers ``` """ -function Base.put!(c::SignalChannel{T,N}, values::AbstractVector{<:FixedSizeMatrixDefault{T}}) where {T,N} +function Base.put!(c::SignalChannel{T,N,M}, values::AbstractVector{<:AbstractMatrix{T}}) where {T,N,M} # Validate all matrices have correct dimensions for (i, v) in enumerate(values) if size(v, 1) != c.num_samples || size(v, 2) != N @@ -257,7 +289,7 @@ Remove and return exactly `n` matrices from the channel in a single batch operat Blocks until all `n` items are available. # Returns -- `Vector{FixedSizeMatrixDefault{T}}`: Vector of exactly `n` matrices +- `Vector{M}`: Vector of exactly `n` matrices (where `M` is the channel's backing type) # Throws - `InvalidStateException`: If the channel is closed before `n` items can be read @@ -274,7 +306,7 @@ function Base.take!(c::SignalChannel{T,N}, n::Integer) where {T,N} end """ - take!(c::SignalChannel{T,N}, output::AbstractVector{<:FixedSizeMatrixDefault{T}}) where {T,N} + take!(c::SignalChannel{T,N,M}, output::AbstractVector{<:AbstractMatrix{T}}) where {T,N,M} Remove matrices from the channel into a pre-allocated output vector. Blocks until the entire output buffer is filled. Returns `length(output)`. @@ -290,11 +322,11 @@ This variant avoids allocation by writing into a provided buffer. # Examples ```julia chan = SignalChannel{ComplexF32,4}(1024) -buffer = Vector{FixedSizeMatrixDefault{ComplexF32}}(undef, 8) +buffer = Vector{Matrix{ComplexF32}}(undef, 8) take!(chan, buffer) # Fills buffer with 8 matrices ``` """ -function Base.take!(c::SignalChannel{T,N}, output::AbstractVector{<:FixedSizeMatrixDefault{T}}) where {T,N} +function Base.take!(c::SignalChannel{T,N}, output::AbstractVector{<:AbstractMatrix{T}}) where {T,N} Base.take!(c.channel, output) end @@ -308,11 +340,11 @@ Base.isempty(c::SignalChannel) = Base.isempty(c.channel) Base.n_avail(c::SignalChannel) = Base.n_avail(c.channel) Base.isfull(c::SignalChannel) = Base.isfull(c.channel) Base.wait(c::SignalChannel) = Base.wait(c.channel) -Base.eltype(::Type{SignalChannel{T,N}}) where {T,N} = FixedSizeMatrixDefault{T} +Base.eltype(::Type{SignalChannel{T,N,M}}) where {T,N,M} = M # Iterator support: allows `for buffer in channel` syntax. # The @inline annotation is critical to avoid heap allocation of the (value, state) -# tuple for non-isbits types like FixedSizeMatrixDefault. +# tuple for non-isbits types like Matrix (the buffer element type). @inline Base.iterate(c::SignalChannel, state=nothing) = Base.iterate(c.channel, state) Base.IteratorSize(::Type{<:SignalChannel}) = Base.SizeUnknown() @@ -332,8 +364,8 @@ output = similar(input) # Same dimensions, buffer size 16 buffered = similar(input, 32) # Same dimensions, buffer size 32 ``` """ -Base.similar(c::SignalChannel{T,N}, size::Int=16) where {T,N} = - SignalChannel{T,N}(c.num_samples, size) +Base.similar(c::SignalChannel{T,N,M}, size::Int=16) where {T,N,M} = + SignalChannel{T,N,M}(c.num_samples, size) """ diff --git a/test/backing_type.jl b/test/backing_type.jl new file mode 100644 index 0000000..2d6f483 --- /dev/null +++ b/test/backing_type.jl @@ -0,0 +1,67 @@ +module BackingTypeTest + +using Test: @test, @testset +using SignalChannels: SignalChannel, rechunk, mux, add, num_antenna_channels +using FixedSizeArrays: FixedSizeMatrixDefault + +const FSM = FixedSizeMatrixDefault + +@testset "Backing matrix type parameter" begin + @testset "Default backing type is Matrix" begin + chan = SignalChannel{ComplexF32,4}(1024) + @test eltype(chan) == Matrix{ComplexF32} + + single = SignalChannel{ComplexF32}(1024) + @test eltype(single) == Matrix{ComplexF32} + end + + @testset "Opt-in FixedSizeArrays backing" begin + chan = SignalChannel{ComplexF32,4,FSM{ComplexF32}}(1024) + @test eltype(chan) == FSM{ComplexF32} + @test num_antenna_channels(chan) == 4 + + data = FSM{ComplexF32}(rand(ComplexF32, 1024, 4)) + @async put!(chan, data) + received = take!(chan) + @test received isa FSM{ComplexF32} + @test received == data + end + + @testset "put! converts a plain matrix into the backing type" begin + # A FixedSizeArrays-backed channel accepts a plain Matrix and stores it + # as the backing type via convert. + chan = SignalChannel{Float64,1,FSM{Float64}}(50, 4) + @async put!(chan, fill(1.5, 50, 1)) # plain Matrix + received = take!(chan) + @test received isa FSM{Float64} + @test all(received .== 1.5) + end + + @testset "similar preserves backing type" begin + chan = SignalChannel{ComplexF32,2,FSM{ComplexF32}}(256, 8) + s = similar(chan) + @test eltype(s) == FSM{ComplexF32} + @test s.num_samples == 256 + end + + @testset "rechunk preserves backing type" begin + src = SignalChannel{ComplexF32,1,FSM{ComplexF32}}(512, 100) do ch + for i in 1:4 + put!(ch, FSM{ComplexF32}(fill(ComplexF32(i, 0), 512, 1))) + end + end + out = rechunk(src, 1024) + @test eltype(out) == FSM{ComplexF32} + buf = take!(out) + @test buf isa FSM{ComplexF32} + @test size(buf) == (1024, 1) + end + + @testset "add/mux preserve backing type" begin + mk() = SignalChannel{Float64,1,FSM{Float64}}(10, 5) + @test eltype(add(mk(), mk())) == FSM{Float64} + @test eltype(mux(mk(), mk())) == FSM{Float64} + end +end + +end # module BackingTypeTest diff --git a/test/channel_combine.jl b/test/channel_combine.jl index bac233f..84e5c7c 100644 --- a/test/channel_combine.jl +++ b/test/channel_combine.jl @@ -2,7 +2,6 @@ module ChannelCombineTest using Test: @test, @testset, @test_throws using SignalChannels: SignalChannel, mux, add -using FixedSizeArrays: FixedSizeMatrixDefault @testset "Channel Combine" begin @testset "mux" begin @@ -13,7 +12,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task1 = Threads.@spawn begin for _ = 1:3 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(1.0), chunk_size, 1), ) put!(ch1, chunk) @@ -23,7 +22,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task2 = Threads.@spawn begin for i = 1:5 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(Float32(i)), chunk_size, 1), ) put!(ch2, chunk) @@ -54,7 +53,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task2 = Threads.@spawn begin for _ = 1:3 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(2.0), chunk_size, 1), ) put!(ch2, chunk) @@ -79,7 +78,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task1 = Threads.@spawn begin for _ = 1:3 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(1.0), chunk_size, 1), ) put!(ch1, chunk) @@ -106,7 +105,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task1 = Threads.@spawn begin for i = 1:3 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(Float32(i)), chunk_size, 1), ) put!(ch1, chunk) @@ -116,7 +115,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task2 = Threads.@spawn begin for i = 1:5 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(Float32(10 + i)), chunk_size, 1), ) put!(ch2, chunk) @@ -146,7 +145,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault ch2 = SignalChannel{Complex{Int16},1}(chunk_size, 10) task1 = Threads.@spawn begin - chunk = FixedSizeMatrixDefault{Complex{Int16}}( + chunk = Matrix{Complex{Int16}}( fill(Complex{Int16}(100, 0), chunk_size, 1), ) put!(ch1, chunk) @@ -155,7 +154,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task2 = Threads.@spawn begin for i = 1:2 - chunk = FixedSizeMatrixDefault{Complex{Int16}}( + chunk = Matrix{Complex{Int16}}( fill(Complex{Int16}(100 * (i + 1), 0), chunk_size, 1), ) put!(ch2, chunk) @@ -191,7 +190,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task1 = Threads.@spawn begin for _ = 1:3 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(1.0), chunk_size, 1), ) put!(ch1, chunk) @@ -201,7 +200,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task2 = Threads.@spawn begin for i = 1:5 - chunk = FixedSizeMatrixDefault{ComplexF32}( + chunk = Matrix{ComplexF32}( fill(ComplexF32(Float32(i)), chunk_size, 1), ) put!(ch2, chunk) @@ -238,7 +237,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault for _ = 1:3 put!( ch1, - FixedSizeMatrixDefault{ComplexF32}( + Matrix{ComplexF32}( fill(ComplexF32(1.0), chunk_size, 1), ), ) @@ -250,7 +249,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault for _ = 1:3 put!( ch2, - FixedSizeMatrixDefault{ComplexF32}( + Matrix{ComplexF32}( fill(ComplexF32(2.0), chunk_size, 1), ), ) @@ -278,7 +277,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault for _ = 1:2 put!( ch1, - FixedSizeMatrixDefault{ComplexF32}( + Matrix{ComplexF32}( fill(ComplexF32(1.0), chunk_size, 1), ), ) @@ -290,7 +289,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault for _ = 1:2 put!( ch2, - FixedSizeMatrixDefault{ComplexF32}( + Matrix{ComplexF32}( fill(ComplexF32(2.0), chunk_size, 1), ), ) @@ -302,7 +301,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault for _ = 1:2 put!( ch3, - FixedSizeMatrixDefault{ComplexF32}( + Matrix{ComplexF32}( fill(ComplexF32(3.0), chunk_size, 1), ), ) @@ -329,7 +328,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault for _ = 1:2 put!( ch1, - FixedSizeMatrixDefault{ComplexF32}( + Matrix{ComplexF32}( fill(ComplexF32(1.0, 0.0), chunk_size, 2), ), ) @@ -341,7 +340,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault for _ = 1:2 put!( ch2, - FixedSizeMatrixDefault{ComplexF32}( + Matrix{ComplexF32}( fill(ComplexF32(0.0, 1.0), chunk_size, 2), ), ) @@ -370,7 +369,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task1 = Threads.@spawn begin put!( ch1, - FixedSizeMatrixDefault{Complex{Int16}}( + Matrix{Complex{Int16}}( fill(Complex{Int16}(100, 0), chunk_size, 1), ), ) @@ -380,7 +379,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task2 = Threads.@spawn begin put!( ch2, - FixedSizeMatrixDefault{Complex{Int16}}( + Matrix{Complex{Int16}}( fill(Complex{Int16}(200, 50), chunk_size, 1), ), ) diff --git a/test/channel_utilities.jl b/test/channel_utilities.jl index 7670943..76e9780 100644 --- a/test/channel_utilities.jl +++ b/test/channel_utilities.jl @@ -2,7 +2,6 @@ module ChannelUtilitiesTest using Test: @test, @testset, @test_throws using SignalChannels: SignalChannel, PipeChannel, consume_channel, tee, write_to_file, read_from_file -using FixedSizeArrays: FixedSizeMatrixDefault @testset "Channel Utilities" begin @testset "consume_channel" begin @@ -30,7 +29,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) put!(chan, data) end close(chan) @@ -50,7 +49,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) put!(input_chan, data) end close(input_chan) @@ -164,7 +163,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) put!(input_chan, data) end close(input_chan) @@ -199,7 +198,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:5 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, i), 100, 3)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, i), 100, 3)) put!(input_chan, data) end close(input_chan) @@ -233,7 +232,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:3 - data = FixedSizeMatrixDefault{Float64}(fill(Float64(i * 10), 50, 2)) + data = Matrix{Float64}(fill(Float64(i * 10), 50, 2)) put!(input_chan, data) end close(input_chan) @@ -262,7 +261,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:5 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, i), 100, 3)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, i), 100, 3)) put!(input_chan, data) end close(input_chan) @@ -292,7 +291,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:5 - data = FixedSizeMatrixDefault{Float64}(fill(Float64(i), 100, 2)) + data = Matrix{Float64}(fill(Float64(i), 100, 2)) put!(input_chan, data) end close(input_chan) @@ -344,7 +343,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault # Producer that tries to put data producer_task = Threads.@spawn begin for i in 1:100 - put!(input_chan, FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 100, 2))) + put!(input_chan, Matrix{ComplexF32}(fill(ComplexF32(i, 0), 100, 2))) end close(input_chan) end diff --git a/test/periodogram.jl b/test/periodogram.jl index 47aaa1f..680e6b4 100644 --- a/test/periodogram.jl +++ b/test/periodogram.jl @@ -2,7 +2,6 @@ module PeriodogramTest using Test: @test, @testset, @test_throws using SignalChannels: SignalChannel, PeriodogramData, calculate_periodogram, periodogram_liveplot -using FixedSizeArrays: FixedSizeMatrixDefault using Unitful using DSP @@ -28,7 +27,7 @@ end task = @async begin for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 1024, 1)) + data = Matrix{ComplexF32}(rand(ComplexF32, 1024, 1)) put!(data_chan, data) end close(data_chan) @@ -49,7 +48,7 @@ end task = @async begin for i in 1:2 - data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 512, 4)) + data = Matrix{ComplexF32}(rand(ComplexF32, 512, 4)) put!(data_chan, data) end close(data_chan) @@ -67,7 +66,7 @@ end task = @async begin for i in 1:2 - data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 512, 1)) + data = Matrix{ComplexF32}(rand(ComplexF32, 512, 1)) put!(data_chan, data) end close(data_chan) diff --git a/test/rechunk.jl b/test/rechunk.jl index 8bda023..62650e0 100644 --- a/test/rechunk.jl +++ b/test/rechunk.jl @@ -2,7 +2,6 @@ module RechunkTest using Test: @test, @testset using SignalChannels: SignalChannel, rechunk, RechunkState, rechunk!, get_partial_buffer, reset!, get_num_antenna_channels -using FixedSizeArrays: FixedSizeMatrixDefault @testset "Rechunk" begin @testset "RechunkState - basic construction" begin @@ -20,13 +19,13 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{Float64}(1024, 1, 5) # First input: 512 samples - should yield nothing (partial fill) - input1 = FixedSizeMatrixDefault{Float64}(fill(1.0, 512, 1)) + input1 = Matrix{Float64}(fill(1.0, 512, 1)) results1 = rechunk!(state, input1) @test length(results1) == 0 @test state.chunk_filled == 512 # Second input: 512 more samples - should complete one chunk - input2 = FixedSizeMatrixDefault{Float64}(fill(2.0, 512, 1)) + input2 = Matrix{Float64}(fill(2.0, 512, 1)) results2 = rechunk!(state, input2) @test length(results2) == 1 @test size(results2[1]) == (1024, 1) @@ -40,7 +39,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{ComplexF32}(100, 4, 5) # Input with 250 samples × 4 channels - input = FixedSizeMatrixDefault{ComplexF32}(undef, 250, 4) + input = Matrix{ComplexF32}(undef, 250, 4) for ch in 1:4 input[:, ch] .= ComplexF32(ch, 0) end @@ -62,7 +61,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault # 10000 samples input → 1024 samples output state = RechunkState{Float32}(1024, 2, 12) - input = FixedSizeMatrixDefault{Float32}(fill(3.14f0, 10000, 2)) + input = Matrix{Float32}(fill(3.14f0, 10000, 2)) results = rechunk!(state, input) # 10000 / 1024 = 9.765 → 9 complete chunks @@ -77,7 +76,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault # When input is exact multiple of output, no remainder state = RechunkState{Int}(100, 1, 5) - input = FixedSizeMatrixDefault{Int}(fill(42, 500, 1)) + input = Matrix{Int}(fill(42, 500, 1)) results = rechunk!(state, input) @test length(results) == 5 @@ -91,7 +90,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @test get_partial_buffer(state) === nothing # Add partial data - input = FixedSizeMatrixDefault{Float64}(fill(1.0, 50, 2)) + input = Matrix{Float64}(fill(1.0, 50, 2)) rechunk!(state, input) # Should not yield any complete chunks partial = get_partial_buffer(state) @@ -106,7 +105,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{Float64}(100, 2, 5) # Add some partial data - input = FixedSizeMatrixDefault{Float64}(fill(1.0, 75, 2)) + input = Matrix{Float64}(fill(1.0, 75, 2)) rechunk!(state, input) @test state.chunk_filled == 75 @test state.buffer_idx == 1 @@ -128,7 +127,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault # Copy data immediately since buffers are pooled and will be reused all_values = Vector{Int}[] for i in 1:20 - input = FixedSizeMatrixDefault{Int}(fill(i, 10, 1)) + input = Matrix{Int}(fill(i, 10, 1)) for chunk in rechunk!(state, input) push!(all_values, copy(vec(chunk[:, 1]))) end @@ -146,9 +145,9 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{Float64}(100, 2, 10) # Create inputs with sequential values - inputs = [FixedSizeMatrixDefault{Float64}(fill(Float64(i), 37, 2)) for i in 1:10] + inputs = [Matrix{Float64}(fill(Float64(i), 37, 2)) for i in 1:10] - all_outputs = FixedSizeMatrixDefault{Float64}[] + all_outputs = Matrix{Float64}[] for input in inputs for output in rechunk!(state, input) push!(all_outputs, copy(output)) @@ -172,7 +171,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "RechunkState - view interface" begin state = RechunkState{Float32}(50, 1, 5; max_outputs_per_input=3) - input = FixedSizeMatrixDefault{Float32}(fill(1.0f0, 125, 1)) + input = Matrix{Float32}(fill(1.0f0, 125, 1)) # rechunk! returns a view outputs = rechunk!(state, input) @@ -194,7 +193,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault # the input should be returned directly without copying state = RechunkState{Float64}(100, 2, 5) - input = FixedSizeMatrixDefault{Float64}(fill(1.0, 100, 2)) + input = Matrix{Float64}(fill(1.0, 100, 2)) results = rechunk!(state, input) @test length(results) == 1 @@ -207,13 +206,13 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{Float64}(100, 2, 5) # First, add some partial data - partial_input = FixedSizeMatrixDefault{Float64}(fill(1.0, 50, 2)) + partial_input = Matrix{Float64}(fill(1.0, 50, 2)) rechunk!(state, partial_input) @test state.chunk_filled == 50 # Now add input that matches output size - should NOT passthrough # because there's partial data buffered - input = FixedSizeMatrixDefault{Float64}(fill(2.0, 100, 2)) + input = Matrix{Float64}(fill(2.0, 100, 2)) results = rechunk!(state, input) @test length(results) == 1 @@ -230,7 +229,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{Float64}(100, 2, 10; max_outputs_per_input=5) # Create batch of 4 inputs, each with 37 samples - inputs = [FixedSizeMatrixDefault{Float64}(fill(Float64(i), 37, 2)) for i in 1:4] + inputs = [Matrix{Float64}(fill(Float64(i), 37, 2)) for i in 1:4] # Process all 4 inputs at once results = rechunk!(state, inputs) @@ -250,7 +249,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{Float64}(50, 1, 10; max_outputs_per_input=5) # Create batch of 4 inputs, but only process first 2 via view - inputs = [FixedSizeMatrixDefault{Float64}(fill(Float64(i), 30, 1)) for i in 1:4] + inputs = [Matrix{Float64}(fill(Float64(i), 30, 1)) for i in 1:4] results = rechunk!(state, @view inputs[1:2]) @@ -266,7 +265,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault state = RechunkState{Float64}(100, 2, 10; max_outputs_per_input=5) # Create inputs that exactly match output size - inputs = [FixedSizeMatrixDefault{Float64}(fill(Float64(i), 100, 2)) for i in 1:3] + inputs = [Matrix{Float64}(fill(Float64(i), 100, 2)) for i in 1:3] results = rechunk!(state, inputs) @@ -297,7 +296,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:5 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) put!(input_chan, data) end close(input_chan) @@ -327,7 +326,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:2 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 1000, 2)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 1000, 2)) put!(input_chan, data) end close(input_chan) @@ -352,7 +351,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault # Send exactly 50 samples (should produce 2 chunks of 25) task = @async begin for i in 1:5 - data = FixedSizeMatrixDefault{Float64}(fill(Float64(i), 10, 1)) + data = Matrix{Float64}(fill(Float64(i), 10, 1)) put!(input_chan, data) end close(input_chan) @@ -386,7 +385,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault channel_size = 16 # Create data with unique values so we can detect corruption - all_original = [FixedSizeMatrixDefault{ComplexF32}(ComplexF32.(i .+ (1:num_samples) ./ num_samples, 0) |> x -> reshape(x, :, 1)) for i in 1:num_chunks] + all_original = [Matrix{ComplexF32}(ComplexF32.(i .+ (1:num_samples) ./ num_samples, 0) |> x -> reshape(x, :, 1)) for i in 1:num_chunks] input = SignalChannel{ComplexF32,1}(num_samples, channel_size) intermediate = rechunk(input, 2048, channel_size) diff --git a/test/runtests.jl b/test/runtests.jl index 9f7e149..74ee5bc 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -1,4 +1,5 @@ include("signal_channel.jl") +include("backing_type.jl") include("rechunk.jl") include("channel_utilities.jl") include("channel_combine.jl") diff --git a/test/signal_channel.jl b/test/signal_channel.jl index 2c64346..ed4aa78 100644 --- a/test/signal_channel.jl +++ b/test/signal_channel.jl @@ -2,7 +2,6 @@ module MatrixChannelTest using Test: @test, @testset, @test_throws using SignalChannels: SignalChannel, PipeChannel, num_antenna_channels -using FixedSizeArrays: FixedSizeMatrixDefault @testset "SignalChannel" begin @testset "Construction" begin @@ -21,7 +20,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "Put and take with correct dimensions" begin chan = SignalChannel{ComplexF32,4}(1024) - data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 1024, 4)) + data = Matrix{ComplexF32}(rand(ComplexF32, 1024, 4)) @async put!(chan, data) received = take!(chan) @@ -32,11 +31,11 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "Put with incorrect dimensions throws error" begin chan = SignalChannel{ComplexF32,4}(1024) - wrong_data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 512, 4)) # Wrong number of samples + wrong_data = Matrix{ComplexF32}(rand(ComplexF32, 512, 4)) # Wrong number of samples @test_throws ArgumentError put!(chan, wrong_data) - wrong_data2 = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 1024, 2)) # Wrong number of channels + wrong_data2 = Matrix{ComplexF32}(rand(ComplexF32, 1024, 2)) # Wrong number of channels @test_throws ArgumentError put!(chan, wrong_data2) end @@ -46,7 +45,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @test isempty(chan) @test !isready(chan) - data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 1024, 4)) + data = Matrix{ComplexF32}(rand(ComplexF32, 1024, 4)) put!(chan, data) @test isready(chan) @@ -59,7 +58,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "Construction with function" begin chan = SignalChannel{ComplexF32,4}(1024, 5) do c for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 1024, 4)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 1024, 4)) put!(c, data) end end @@ -77,7 +76,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "Iteration" begin chan = SignalChannel{ComplexF32,4}(1024, 5) do c for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 1024, 4)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 1024, 4)) put!(c, data) end end @@ -95,7 +94,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:5 - data = FixedSizeMatrixDefault{Float64}(fill(Float64(i), 100, 2)) + data = Matrix{Float64}(fill(Float64(i), 100, 2)) put!(chan, data) end close(chan) @@ -112,14 +111,14 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "eltype" begin chan_f32 = SignalChannel{ComplexF32,4}(1024) - @test eltype(chan_f32) == FixedSizeMatrixDefault{ComplexF32} + @test eltype(chan_f32) == Matrix{ComplexF32} chan_f64 = SignalChannel{Float64,2}(512) - @test eltype(chan_f64) == FixedSizeMatrixDefault{Float64} + @test eltype(chan_f64) == Matrix{Float64} # Test single channel chan_single = SignalChannel{ComplexF32}(1024) - @test eltype(chan_single) == FixedSizeMatrixDefault{ComplexF32} + @test eltype(chan_single) == Matrix{ComplexF32} end @testset "Single channel with matrices" begin @@ -129,11 +128,11 @@ using FixedSizeArrays: FixedSizeMatrixDefault @test isopen(chan) # Put and take matrix with shape (1024, 1) - data = FixedSizeMatrixDefault{ComplexF32}(rand(ComplexF32, 1024, 1)) + data = Matrix{ComplexF32}(rand(ComplexF32, 1024, 1)) @async put!(chan, data) received = take!(chan) - @test received isa FixedSizeMatrixDefault{ComplexF32} + @test received isa Matrix{ComplexF32} @test size(received) == (1024, 1) @test received == data end @@ -141,13 +140,13 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "Single channel with function constructor" begin chan = SignalChannel{ComplexF32}(1024) do c for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 1024, 1)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 1024, 1)) put!(c, data) end end result1 = take!(chan) - @test result1 isa FixedSizeMatrixDefault{ComplexF32} + @test result1 isa Matrix{ComplexF32} @test size(result1) == (1024, 1) @test all(result1 .== ComplexF32(1, 0)) diff --git a/test/soapysdr_ext.jl b/test/soapysdr_ext.jl index 6bfb341..cf33f7c 100644 --- a/test/soapysdr_ext.jl +++ b/test/soapysdr_ext.jl @@ -8,7 +8,6 @@ if Base.find_package("SoapySDR") !== nothing using SoapySDR using SoapySDR: Device, SoapySDRDeviceError using Unitful - using FixedSizeArrays: FixedSizeMatrixDefault using SignalChannels: stream_data, SDRChannelConfig @@ -75,7 +74,7 @@ if Base.find_package("SoapySDR") !== nothing tx_stats_channel, tx_warning_channel = stream_data(device_args, config, tx_channel) # Create test pattern with recognizable data - test_pattern = FixedSizeMatrixDefault{ComplexF32}(zeros(ComplexF32, mtu, 1)) + test_pattern = Matrix{ComplexF32}(zeros(ComplexF32, mtu, 1)) for i in 1:mtu test_pattern[i, 1] = ComplexF32(i / mtu, (mtu - i) / mtu) end @@ -167,7 +166,7 @@ if Base.find_package("SoapySDR") !== nothing stats_channel, warning_channel = stream_data(device_args, config, tx_channel) # Send some data - buffer = FixedSizeMatrixDefault{ComplexF32}(zeros(ComplexF32, mtu, 1)) + buffer = Matrix{ComplexF32}(zeros(ComplexF32, mtu, 1)) buffer[1:10, 1] .= ComplexF32(1.0 + 1.0im) put!(tx_channel, buffer) @@ -250,7 +249,7 @@ if Base.find_package("SoapySDR") !== nothing # Each buffer has a unique pattern based on buffer index sent_data = ComplexF32[] for buf_idx in 1:num_tx_buffers - buffer = FixedSizeMatrixDefault{ComplexF32}(zeros(ComplexF32, tx_chunk, 1)) + buffer = Matrix{ComplexF32}(zeros(ComplexF32, tx_chunk, 1)) for i in 1:tx_chunk # Create pattern: real = buffer_idx, imag = sample position buffer[i, 1] = ComplexF32(buf_idx, i) diff --git a/test/stream_utilities.jl b/test/stream_utilities.jl index 2e1fb12..798d13f 100644 --- a/test/stream_utilities.jl +++ b/test/stream_utilities.jl @@ -2,13 +2,12 @@ module StreamUtilitiesTest using Test: @test, @testset using SignalChannels: SignalChannel, PipeChannel, spawn_signal_channel_thread, membuffer, num_antenna_channels -using FixedSizeArrays: FixedSizeMatrixDefault @testset "Stream Utilities" begin @testset "spawn_signal_channel_thread with SignalChannel" begin chan = spawn_signal_channel_thread(T=ComplexF32, num_samples=100, num_antenna_channels=2) do out for i in 1:3 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) put!(out, data) end end @@ -27,7 +26,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "spawn_signal_channel_thread closes on completion" begin chan = spawn_signal_channel_thread(T=Float64, num_samples=50, num_antenna_channels=1) do out - data = FixedSizeMatrixDefault{Float64}(fill(1.0, 50, 1)) + data = Matrix{Float64}(fill(1.0, 50, 1)) put!(out, data) end @@ -41,7 +40,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault @testset "spawn_signal_channel_thread closes on error" begin chan = spawn_signal_channel_thread(T=Float64, num_samples=50, num_antenna_channels=1) do out - data = FixedSizeMatrixDefault{Float64}(fill(1.0, 50, 1)) + data = Matrix{Float64}(fill(1.0, 50, 1)) put!(out, data) error("Intentional error") end @@ -64,7 +63,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault task = @async begin for i in 1:5 - data = FixedSizeMatrixDefault{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) + data = Matrix{ComplexF32}(fill(ComplexF32(i, 0), 100, 2)) put!(input_chan, data) end close(input_chan) @@ -86,7 +85,7 @@ using FixedSizeArrays: FixedSizeMatrixDefault # Quickly put many items task = @async begin for i in 1:50 - data = FixedSizeMatrixDefault{Float64}(fill(Float64(i), 50, 1)) + data = Matrix{Float64}(fill(Float64(i), 50, 1)) put!(input_chan, data) end close(input_chan) From a53733405145315b1d6490b9531b07db119dcced Mon Sep 17 00:00:00 2001 From: Soeren Schoenbrod Date: Mon, 13 Jul 2026 13:36:13 +0000 Subject: [PATCH 2/4] chore(benchmark): run PR benchmarks against both old and new SignalChannel APIs AirspeedVelocity runs the PR head's benchmark scripts against both the PR and the baseline (main). Since the backing-array-type change is breaking, the head scripts must stay runnable on the old FixedSizeMatrixDefault-only API. Add a shared HAS_BACKING_TYPE_PARAM detector and an as_buffer helper in benchmarks.jl that feeds each package version its native zero-copy buffer type (Matrix on the new API, FixedSizeMatrixDefault on the old one), and relax buffer parameter annotations to AbstractMatrix. Verified the head scripts run against both the new working tree and the registered v6.2.2. Co-Authored-By: Claude Opus 4.8 (1M context) --- benchmark/benchmarks.jl | 19 +++++++++++++++++++ benchmark/channel_benchmarks.jl | 4 ++-- benchmark/rechunk_benchmarks.jl | 2 +- benchmark/rechunk_state_benchmarks.jl | 6 +++--- benchmark/soapysdr_benchmarks.jl | 10 +++++----- 5 files changed, 30 insertions(+), 11 deletions(-) diff --git a/benchmark/benchmarks.jl b/benchmark/benchmarks.jl index 168daab..4bfa295 100644 --- a/benchmark/benchmarks.jl +++ b/benchmark/benchmarks.jl @@ -1,12 +1,31 @@ using BenchmarkTools using SignalChannels using Pkg +using FixedSizeArrays: FixedSizeMatrixDefault # Get package version for backward compatibility in benchmarks const PACKAGE_VERSION = Pkg.Types.read_project( joinpath(dirname(@__DIR__), "Project.toml") ).version +# Detect the backing-array-type parameter API (SignalChannel{T,N,M}). +# The new API defaults to Matrix{T}; the old API is FixedSizeMatrixDefault-only. +# AirspeedVelocity runs this (head) script against both the PR and the baseline, +# so we feed each version its native zero-copy buffer type to keep the +# comparison fair and runnable on both. +const HAS_BACKING_TYPE_PARAM = try + SignalChannel{ComplexF32,1,Matrix{ComplexF32}} + true +catch + false +end + +# Wrap a plain array as the buffer type native to the current API: +# Matrix{T} on the new API (returned as-is, zero-copy), FixedSizeMatrixDefault{T} +# on the old API. +as_buffer(a::AbstractMatrix) = + HAS_BACKING_TYPE_PARAM ? a : FixedSizeMatrixDefault{eltype(a)}(a) + const SUITE = BenchmarkGroup() # Include individual benchmark files diff --git a/benchmark/channel_benchmarks.jl b/benchmark/channel_benchmarks.jl index 261bdd5..fe37c41 100644 --- a/benchmark/channel_benchmarks.jl +++ b/benchmark/channel_benchmarks.jl @@ -19,12 +19,12 @@ function setup_channel_benchmark(num_samples::Int, buffer_size::Int) SignalChannel{ComplexF32}(num_samples, 1, buffer_size) end - data = Matrix{ComplexF32}(rand(ComplexF32, num_samples, 1)) + data = as_buffer(rand(ComplexF32, num_samples, 1)) return (ch, data) end # Benchmark: push data through the channel and drain output -function run_channel_benchmark!(ch, data::Matrix{ComplexF32}, num_items::Int) +function run_channel_benchmark!(ch, data::AbstractMatrix{ComplexF32}, num_items::Int) # Producer task producer = Threads.@spawn begin for _ in 1:num_items diff --git a/benchmark/rechunk_benchmarks.jl b/benchmark/rechunk_benchmarks.jl index 0d814fa..085c297 100644 --- a/benchmark/rechunk_benchmarks.jl +++ b/benchmark/rechunk_benchmarks.jl @@ -24,7 +24,7 @@ function setup_pipeline(input_size::Int, output_size::Int, output_channel_size:: # Create the rechunk pipeline - this spawns the task but it blocks waiting for input output = rechunk(input, output_size, output_channel_size) - buffers = [Matrix{ComplexF32}(zeros(ComplexF32, input_size, 1)) for _ in 1:NUM_BUFFERS] + buffers = [as_buffer(zeros(ComplexF32, input_size, 1)) for _ in 1:NUM_BUFFERS] return (input, output, buffers) end diff --git a/benchmark/rechunk_state_benchmarks.jl b/benchmark/rechunk_state_benchmarks.jl index 87f2065..a7beb3d 100644 --- a/benchmark/rechunk_state_benchmarks.jl +++ b/benchmark/rechunk_state_benchmarks.jl @@ -36,7 +36,7 @@ if RECHUNK_STATE_SUPPORTED max_output_buffers = cld(total_input_samples, output_size) + 2 state = RechunkState{T,N}(output_size, max_output_buffers, max_output_buffers) - inputs = [Matrix{T}(rand(T, input_size, N)) for _ in 1:num_buffers] + inputs = [as_buffer(rand(T, input_size, N)) for _ in 1:num_buffers] return (state, inputs) end @@ -192,7 +192,7 @@ if RECHUNK_STATE_SUPPORTED # Inner function with Val{N} for compile-time specialization function _setup_passthrough(::Type{T}, size::Int, ::Val{N}, num_buffers::Int) where {T,N} state = RechunkState{T,N}(size, num_buffers + 2, num_buffers + 2) - inputs = [Matrix{T}(rand(T, size, N)) for _ in 1:num_buffers] + inputs = [as_buffer(rand(T, size, N)) for _ in 1:num_buffers] return (state, inputs) end @@ -205,7 +205,7 @@ if RECHUNK_STATE_SUPPORTED function _setup_near_passthrough(::Type{T}, size::Int, ::Val{N}, num_buffers::Int) where {T,N} # Output is 1 sample smaller, so copies are required state = RechunkState{T,N}(size - 1, num_buffers + 100, num_buffers + 100) - inputs = [Matrix{T}(rand(T, size, N)) for _ in 1:num_buffers] + inputs = [as_buffer(rand(T, size, N)) for _ in 1:num_buffers] return (state, inputs) end diff --git a/benchmark/soapysdr_benchmarks.jl b/benchmark/soapysdr_benchmarks.jl index 0313684..28bebbb 100644 --- a/benchmark/soapysdr_benchmarks.jl +++ b/benchmark/soapysdr_benchmarks.jl @@ -63,7 +63,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER end # Pre-allocate test data - data = Matrix{ComplexF32}(zeros(ComplexF32, chunk_size, 1)) + data = as_buffer(zeros(ComplexF32, chunk_size, 1)) for i in 1:chunk_size data[i, 1] = ComplexF32(i / chunk_size, 0.5f0) end @@ -110,7 +110,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER # Benchmark: Single put! operation on SignalChannel # Should show zero or constant allocations per call - function benchmark_single_put!(channel, data::Matrix{ComplexF32}) + function benchmark_single_put!(channel, data::AbstractMatrix{ComplexF32}) put!(channel, data) return nothing end @@ -155,7 +155,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER tx_stats_channel, tx_warning_channel = stream_data(device_args, config, tx_channel) # Create test data buffer - test_data = Matrix{ComplexF32}(zeros(ComplexF32, chunk_size, 1)) + test_data = as_buffer(zeros(ComplexF32, chunk_size, 1)) for i in 1:chunk_size test_data[i, 1] = ComplexF32(i / chunk_size, (chunk_size - i) / chunk_size) end @@ -188,7 +188,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER function benchmark_loopback!( tx_channel, rx_channel, - data::Matrix{ComplexF32}, + data::AbstractMatrix{ComplexF32}, num_buffers::Int, ) # Start producer @@ -218,7 +218,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER function benchmark_loopback_single_take!( tx_channel, rx_channel, - data::Matrix{ComplexF32}, + data::AbstractMatrix{ComplexF32}, ) # Send one buffer via TX put!(tx_channel, data) From 111984b37b3146ce8e5a6db557e2203b0795cb4e Mon Sep 17 00:00:00 2001 From: Soeren Schoenbrod Date: Mon, 13 Jul 2026 13:59:43 +0000 Subject: [PATCH 3/4] fix: store matching buffers by reference in put! (zero-copy) The two-method put! (put!(::SignalChannel{T,N,M}, ::M) fast path plus an AbstractMatrix fallback) failed to dispatch to the fast path: M is a diagonal type variable shared with the channel type, so Julia did not consider the ::M method more specific than the ::AbstractMatrix fallback. Every put! hit the fallback and copied the buffer via M(v), allocating one buffer per call and regressing channel/rechunk throughput 2-500x (caught by the PR benchmark). Replace it with a single put! method plus a convert-style _as_backing(::Type{M}, v) helper (modeled on Base.convert's (::Type{T}, ::T) / (::Type{T}, x) pair), which dispatches the identity case reliably. Matching buffers are now stored by reference (0 allocations); only genuinely foreign matrix types are constructed into M. Add a regression test asserting take!(chan) === put! input. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/signal_channel.jl | 15 ++++++++------- test/backing_type.jl | 15 +++++++++++++++ 2 files changed, 23 insertions(+), 7 deletions(-) diff --git a/src/signal_channel.jl b/src/signal_channel.jl index 51331ca..12c80b5 100644 --- a/src/signal_channel.jl +++ b/src/signal_channel.jl @@ -222,16 +222,17 @@ is converted into an `M` via the `M(v)` constructor, which allocates a new buffe # Throws - `ArgumentError`: If matrix dimensions don't match the channel configuration """ -# Fast path: value is already the backing type — store it by reference (zero-copy). -function Base.put!(c::SignalChannel{T,N,M}, v::M) where {T,N,M} - _check_put_dims(c, v) - Base.put!(c.channel, v) -end +# Coerce a matrix to the backing type `M`. Modeled on `Base.convert`'s +# `(::Type{T}, ::T)` / `(::Type{T}, x)` pair so the identity case dispatches +# reliably (unlike a `v::M` method on `put!`, where `M` is a diagonal type +# variable shared with the channel type and is not treated as more specific +# than the `AbstractMatrix` fallback). +@inline _as_backing(::Type{M}, v::M) where {M<:AbstractMatrix} = v +@inline _as_backing(::Type{M}, v::AbstractMatrix) where {M<:AbstractMatrix} = M(v)::M -# Fallback: convert any other matrix into the backing type via its constructor. function Base.put!(c::SignalChannel{T,N,M}, v::AbstractMatrix{T}) where {T,N,M} _check_put_dims(c, v) - Base.put!(c.channel, M(v)::M) + Base.put!(c.channel, _as_backing(M, v)) end # Delegate Base methods to the underlying channel diff --git a/test/backing_type.jl b/test/backing_type.jl index 2d6f483..c031c29 100644 --- a/test/backing_type.jl +++ b/test/backing_type.jl @@ -37,6 +37,21 @@ const FSM = FixedSizeMatrixDefault @test all(received .== 1.5) end + @testset "put! stores a matching buffer by reference (zero-copy)" begin + # Guards against dispatch regressions that would copy on every put!. + # Default Matrix-backed channel with a Matrix buffer: + chan = SignalChannel{ComplexF32,1}(64, 4) + data = rand(ComplexF32, 64, 1) + put!(chan, data) + @test take!(chan) === data + + # Explicit FixedSizeArrays backing with a matching FSM buffer: + fchan = SignalChannel{Float64,1,FSM{Float64}}(64, 4) + fdata = FSM{Float64}(fill(1.0, 64, 1)) + put!(fchan, fdata) + @test take!(fchan) === fdata + end + @testset "similar preserves backing type" begin chan = SignalChannel{ComplexF32,2,FSM{ComplexF32}}(256, 8) s = similar(chan) From 80a21802a28e2cc663208131ff093e5d07d85e47 Mon Sep 17 00:00:00 2001 From: Soeren Schoenbrod Date: Mon, 13 Jul 2026 21:21:25 +0000 Subject: [PATCH 4/4] fix: require PipeChannels 1.0.1 for the iterate/close race fix PipeChannels 1.0.1 fixes a data-loss race where a consumer iterating/take!-ing could drop the final item when close() raced with the last put!. SignalChannels' zero-copy put! makes that window more likely to hit (e.g. flaky tee output), so require >= 1.0.1 (which excludes the fix-less 1.0.0). Co-Authored-By: Claude Opus 4.8 (1M context) --- Project.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Project.toml b/Project.toml index ee08c1f..a987c78 100644 --- a/Project.toml +++ b/Project.toml @@ -20,7 +20,7 @@ SignalChannelsSoapySDRExt = "SoapySDR" DSP = "0.7, 0.8" FixedSizeArrays = "1" LiveLayoutUnicodePlots = "0.1.0" -PipeChannels = "0.1.1" +PipeChannels = "1.0.1" SoapyLoopback_jll = "0.1.1" SoapySDR = "0.5" UnicodePlots = "3"