feat!: default to Matrix buffers with optional backing array type

Project.toml

@@ -5,7 +5,6 @@ authors = ["Soeren Schoenbrod <soeren.schoenbrod@outlook.de>"]
 
 [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"
@@ -21,17 +20,18 @@ 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"
 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"]

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

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

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 = 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::FixedSizeMatrixDefault{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

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 = [as_buffer(zeros(ComplexF32, input_size, 1)) for _ in 1:NUM_BUFFERS]
 
     return (input, output, buffers)
 end

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 = [as_buffer(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 = [as_buffer(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 = [as_buffer(rand(T, size, N)) for _ in 1:num_buffers]
         return (state, inputs)
     end
 

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 = as_buffer(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::AbstractMatrix{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 = 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
@@ -189,7 +188,7 @@ if SOAPYSDR_AVAILABLE && HAS_LOOPBACK_DRIVER
     function benchmark_loopback!(
         tx_channel,
         rx_channel,
-        data::FixedSizeMatrixDefault{ComplexF32},
+        data::AbstractMatrix{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::AbstractMatrix{ComplexF32},
     )
         # Send one buffer via TX
         put!(tx_channel, data)

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)

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,

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