make_resample_poly_stream#

Streaming polyphase resampler: resample a signal delivered in segments, equivalent to a one-shot resample_poly over the concatenated stream

Added in version 1.1.0.

struct ResamplePolyStreamParams#

Construction-time parameters for a streaming polyphase resampler object.

Aggregate with designated-initializer support, e.g. make_resample_poly_stream<float>(h, {.up = 3, .down = 2}, exec). A params struct is used so future options can be added without changing call sites.

Public Members

index_t up = 1#: Upsample factor (must be positive)

index_t down = 1#: Downsample factor (must be positive)

template<typename InType, typename FilterOp, typename Exec = cudaExecutor> auto matx::make_resample_poly_stream(const FilterOp &filter, const ResamplePolyStreamParams &params, Exec exec = {})#

Create a streaming polyphase resampler object.

The object resamples an arbitrarily long signal by params.up / params.down, delivered in segments of any (possibly varying) size. Feed segments via feed() and call flush() once at end of stream. The concatenation of the produced outputs equals a single one-shot resample_poly(signal, filter, params.up, params.down) over the whole signal.

The streamed outputs match the one-shot to floating-point tolerance but are not guaranteed bit-for-bit identical: resample_poly selects its kernel from the output length, and a streaming call (which computes a smaller window of outputs) can select a kernel with a different summation order than the full-length one-shot. Repeating a streaming run with the same segment sizes produces reproducible results given an identical GPU on the same system.

The object owns a small retained-history buffer sized from the filter and the reduced resampling ratio. No allocation scales with the segment size. The output buffer provided to feed() / flush() must be large enough to hold the maximum number of outputs that can be produced by a single call. This value is returned by max_output(max_input_segment_size). If the maximum segment size is known a priori, then a single output buffer can be allocated and reused for all calls.

Each feed() / flush() call writes the outputs to the front of the output buffer and returns the number written (which may be 0), so no dynamic memory is allocated during the call. Consume the produced region slice(out, {0}, {count}) before reusing the output buffer. A zero-length slice is not valid, so create and use the slice only when count > 0.

Template Parameters:

InType – Sample type of the input stream (as in make_tensor<T>)
FilterOp – Type of the filter operator (deduced)
Exec – Executor type (CUDA or host; deduced, cudaExecutor by default)

Parameters:

filter – FIR prototype filter (1D, length >= 1). The object materializes a copy of the filter at construction, so any filter operator (including a transform expression) is evaluated once and need not outlive the object.
params – Stream parameters; see ResamplePolyStreamParams
exec – Executor bound to this stream’s lifetime. All feed()/flush() work runs on it. For CUDA executors the retain buffer is stream-ordered device memory. For host executors the retain buffer is system-allocated memory. The executor (and, for CUDA, its stream) must outlive the object.

Returns:

A streaming object exposing feed(), flush(), max_output(), reset(), and history_len()

template<typename InType, typename FilterOp, typename Exec> class ResamplePolyStream#

Streaming polyphase resampler object; construct via make_resample_poly_stream().

Resamples an arbitrarily long signal delivered in segments of any (possibly varying) size. The only stream state is a small retained-history buffer; no allocation scales with the segment size. All work runs asynchronously on the executor bound at construction. An object serves one stream at a time and is not thread-safe.

Public Functions

ResamplePolyStream(const ResamplePolyStream&) = delete#: Streaming objects are not copyable: a copy would share the retained history buffer while tracking its state independently, corrupting both streams. Move construction/assignment transfers ownership and is allowed.

ResamplePolyStream &operator=(const ResamplePolyStream&) = delete#: Streaming objects are not copyable: a copy would share the retained history buffer while tracking its state independently, corrupting both streams. Move construction/assignment transfers ownership and is allowed.

ResamplePolyStream(ResamplePolyStream&&) = default#: Move-construct, transferring ownership of the stream state.

ResamplePolyStream &operator=(ResamplePolyStream&&) = default#: Move-assign, transferring ownership of the stream state.

inline index_t history_len() const#

Maximum number of trailing input samples retained between calls.

Informational; the caller does not size anything with it. feed() concatenates the retained history internally.

Returns:: Maximum retained history length in samples

inline index_t max_output(index_t new_len) const#

Upper bound on the outputs a single feed() or flush() call can produce.

Size one reusable output buffer with max_output(largest segment size); it is then large enough for every feed() of up to that many samples and for flush().

Parameters:: new_len – Largest segment size that will be passed to feed()
Returns:: Maximum outputs a single call may produce

inline void reset()#

Restart the stream for a new input signal.

Drops all retained history and clears the end-of-stream state set by flush().

template<typename InOp, typename OutTensor> inline index_t feed(const InOp &new_samples, OutTensor &out)#

Feed a segment of new samples and receive the number of outputs it produces.

Emits every not-yet-emitted output whose input samples have fully arrived; the per-call count varies with the resampling ratio and segment size (it can be zero). Outputs are written to the front of out; the return value is the number written. Consume slice(out, {0}, {count}) (the produced region) before reusing out. Runs asynchronously on the object’s executor. Throws matxInvalidParameter if called after flush(). Use reset() to start a new stream.

Template Parameters:

InOp – 1D input operator type (deduced)
OutTensor – 1D output tensor type (deduced)

Parameters:

new_samples – New signal samples (1D, non-empty). Any MatX operator is accepted and its lifecycle is run each call. A transform-valued segment (e.g. ifft(…)) therefore works but allocates and evaluates a per-call temporary; for hot streaming loops prefer a directly-evaluable segment (a tensor, view, or generator) or materialize once and reuse.
out – Output buffer with last-dim size >= max_output(input_segment_size); throws matxInvalidSize if smaller than the produced count

Returns:

Number of outputs written to the front of out (may be 0). A zero-length slice is not valid, so create and use slice(out, {0}, {count}) only when count > 0.

template<typename OutTensor> inline index_t flush(OutTensor &out)#

Emit the end-of-stream trailing outputs.

Produces the remaining outputs whose filter windows extend past the last input sample (computed with implicit zero padding on the right), matching the trailing outputs of the one-shot resample_poly. The first call emits the trailing outputs and ends the stream. Subsequent calls return 0, and feed() throws until reset() starts a new stream. A flush() that throws (for example, an undersized output buffer) does not end the stream and can be retried. Runs asynchronously on the object’s executor.

Template Parameters:: OutTensor – 1D output tensor type (deduced)
Parameters:: out – Output buffer with last-dim size >= max_output(input_segment_size); throws matxInvalidSize if smaller than the produced count
Returns:: Number of outputs written to the front of out (may be 0). A zero-length slice is not valid, so create and use slice(out, {0}, {count}) only when count > 0.

Examples#

auto resample_stream = make_resample_poly_stream<T>(h, {.up = up, .down = down}, exec);
auto output_frame = make_tensor<T>({resample_stream.max_output(a.chunk)});

index_t off = 0, nchunks = 0;
for (index_t g = 0; g < N; g += a.chunk) {
  const index_t in_length = std::min(a.chunk, N - g);
  auto in_chunk = slice(sig, {g}, {g + in_length});
  const index_t cnt = resample_stream.feed(in_chunk, output_frame); // outputs emitted

  // For real applications, consume slice(output_frame, {0}, {cnt}) here

  // For validation only, copy the frame into the full-length buffer.
  if (cnt > 0) {
    (slice(full, {off}, {off + cnt}) = slice(output_frame, {0}, {cnt})).run(exec);
  }
  off += cnt;
  ++nchunks;
}
// End of stream: flush the trailing (edge) outputs.
const index_t tcnt = resample_stream.flush(output_frame);
if (tcnt > 0) {
  (slice(full, {off}, {off + tcnt}) = slice(output_frame, {0}, {tcnt})).run(exec);
}
off += tcnt;

make_resample_poly_stream

Contents

make_resample_poly_stream#

Examples#