ezmsg.sigproc.util.threaded_filt#

Split scipy IIR filtering across threads along a non-sample axis.

Each channel of an IIR filter is an independent recurrence, so splitting the non-sample axes across threads changes nothing about the arithmetic any one channel sees – the result is bit-identical to the single-threaded call, not merely close. That is what makes this usable in a pipeline whose whole point is that offline output matches online output exactly.

The win is real but strictly large-chunk. scipy’s sosfilt/lfilter are single-threaded C loops, so the speedup tracks chunk bytes – measured end-to-end through FilterTransformer, order-4 SOS, three independent reps:

0.25 MiB 0.50x 0.82x 0.79x loss 0.50 MiB 0.99x 1.07x 1.18x break-even 1.00 MiB 1.65x 1.74x 2.25x win 2.00 MiB 2.28x 2.36x 2.43x win 8.00 MiB ~3.3x win

32.00 MiB ~3.4x win

The ratio depends on total bytes rather than on the channel/sample split: at a fixed 1 MiB, 256x512, 512x256 and 1024x128 all land within 1.53-1.63x. Below ~0.5 MiB the dispatch cost dominates and threading is a loss – hence DEFAULT_MIN_BYTES and the hard gate in should_thread(). Online chunk sizes stay single-threaded.

The pool is module-level and created once, so a graph with many filter units shares one bounded set of workers rather than spawning a pool per unit, and no per-chunk thread creation ever happens. Note that this is explicit threading we control; it is unrelated to the implicit threading inside BLAS or scipy.fft, which is governed by environment variables and can nest badly underneath a pool like this one. scipy’s IIR filters touch neither.

Module Attributes

DEFAULT_MIN_BYTES

Chunks smaller than this (in bytes) are filtered single-threaded.

DEFAULT_MAX_WORKERS

Upper bound on pool size.

Functions

filt_threaded(filt_func, coefs, data, axis_idx, zi, zi_axis_offset, min_bytes=1048576)[source]#

Apply filt_func in parallel over blocks of a non-sample axis.

Parameters:

  • filt_func (Callable) – scipy.signal.sosfilt or scipy.signal.lfilter.

  • coefs (tuple) – Positional coefficient args for filt_func.

  • data (NDArray) – Input array; the sample axis is axis_idx.

  • zi (NDArray) – Filter state.

  • zi_axis_offset (int) – How far zi’s axes are shifted relative to data’s. 1 for SOS (leading n_sections axis), 0 for BA.

  • min_bytes (int) – Threshold below which this falls back to one call.

  • axis_idx (int)

Returns:

(y, zf), bit-identical to the equivalent single-threaded call.

Return type:

tuple[NDArray, NDArray]

get_pool()[source]#

The shared worker pool, created on first use and closed at interpreter exit.

Return type:

ThreadPoolExecutor

should_thread(data, axis_idx, min_bytes=1048576)[source]#

Whether splitting data across threads is likely to pay for itself.

Parameters:

  • data (NDArray)

  • axis_idx (int)

  • min_bytes (int)

Return type:

bool

shutdown_pool()[source]#

Release the shared pool. Registered with atexit; safe to call directly.

Return type:

None