streaming_sample_source.hpp

Go to the documentation of this file.

#pragma once

 
#include <atomic>
#include <cstdint>
#include <functional>
#include <thread>
 
#include <pulp/audio/buffer.hpp>
#include <pulp/runtime/wake_signal.hpp>
#include <pulp/audio/planar_audio_ring_buffer.hpp>
 
namespace pulp::audio {

class FrameReaderStopToken {
public:
    FrameReaderStopToken() noexcept = default;
 
    explicit FrameReaderStopToken(
        const std::atomic<bool>& stop_requested) noexcept
        : stop_requested_(&stop_requested) {}
 
    bool stop_requested() const noexcept {
        return stop_requested_ != nullptr &&
               stop_requested_->load(std::memory_order_acquire);
    }
 
private:
    const std::atomic<bool>* stop_requested_ = nullptr;
};

using FrameReader =
    std::function<std::uint64_t(std::uint64_t start_frame,
                                BufferView<float> dest,
                                std::uint64_t frames)>;
 
enum class FrameReaderStopMode : std::uint8_t {
    JoinOnly,
    Cooperative,
};
 
using StoppableFrameReader =
    std::function<std::uint64_t(std::uint64_t start_frame,
                                BufferView<float> dest,
                                std::uint64_t frames,
                                FrameReaderStopToken stop_token)>;
 
struct FrameReaderBinding {
    StoppableFrameReader read;
    FrameReaderStopMode stop_mode = FrameReaderStopMode::JoinOnly;
};
 
enum class StreamingUnderrunPolicy : std::uint8_t {
    PreservePosition = 0,
    AdvancePosition,
};
 
struct StreamingSampleSourceConfig {
    std::uint32_t channels = 0;          
    std::uint64_t total_frames = 0;      
    std::uint32_t sample_rate = 0;       

    std::uint64_t preload_frames = 0;

    std::uint64_t ring_capacity_frames = 0;

    std::uint64_t read_chunk_frames = 8192;

    bool loop = false;
    std::uint64_t loop_start = 0;        
    std::uint64_t loop_end = 0;          

    bool start_background_thread = true;

    StreamingUnderrunPolicy underrun_policy =
        StreamingUnderrunPolicy::PreservePosition;

    std::uint64_t max_read_ahead_frames = 0;
};
 
class StreamingSampleSource {
public:
    StreamingSampleSource() = default;
    ~StreamingSampleSource();
 
    StreamingSampleSource(const StreamingSampleSource&) = delete;
    StreamingSampleSource& operator=(const StreamingSampleSource&) = delete;

    bool prepare(const StreamingSampleSourceConfig& config, FrameReader reader);
    bool prepare(const StreamingSampleSourceConfig& config,
                 FrameReaderBinding reader);

    void release() noexcept;

    bool reset() noexcept;
 
    // ── Audio thread (RT-safe after prepare) ────────────────────────────────

    std::uint64_t pull(BufferView<float> dest, std::uint64_t frames) noexcept;

    bool finished() const noexcept;

    std::uint64_t position() const noexcept {
        return play_pos_.load(std::memory_order_relaxed);
    }

    bool fully_resident() const noexcept { return fully_resident_; }
 
    bool prepared() const noexcept { return prepared_; }
    std::uint32_t channels() const noexcept { return channels_; }
    std::uint64_t total_frames() const noexcept { return total_frames_; }
 
    // ── Background refill ───────────────────────────────────────────────────

    std::uint64_t pump_background() noexcept;
 
    struct Stats {
        std::uint64_t underrun_frames = 0;   
        std::uint64_t ring_available_frames = 0;
        std::uint64_t streamed_frames = 0;   
        std::uint64_t read_errors = 0;
        bool streaming_active = false;       
    };
    Stats stats() const noexcept;
 
private:
    struct DeadlinePolicy {
        bool enabled = false;
        std::uint64_t max_read_ahead_frames = 0;
        std::atomic<std::uint64_t> ring_start_frame{0};
 
        void configure(const StreamingSampleSourceConfig& config,
                       std::uint64_t initial_frame) noexcept;
        void clear() noexcept;
        std::uint64_t effective_end(std::uint64_t total_frames,
                                    std::uint64_t eos_frame) const noexcept;
        void apply_underrun(std::atomic<std::uint64_t>& play_pos,
                            std::uint64_t next_position) const noexcept;
        bool zero_is_backpressure() const noexcept;
        std::uint64_t producer_cursor(std::uint64_t play_pos,
                                      std::uint64_t reader_pos) const noexcept;
        bool discard_stale(PlanarAudioRingBuffer& ring,
                           std::uint64_t play_pos) noexcept;
        void consumed(std::uint64_t ring_start,
                      std::uint64_t frames) noexcept;
        bool align_reader(PlanarAudioRingBuffer& ring,
                          std::uint64_t play_pos,
                          std::uint64_t& reader_pos) noexcept;
        std::uint64_t read_ahead_room(std::uint64_t total_frames,
                                      std::uint64_t play_pos,
                                      std::uint64_t reader_pos) const noexcept;
        void publish_start_if_empty(PlanarAudioRingBuffer& ring,
                                    std::uint64_t frame) noexcept;
    };
 
    void stop_thread() noexcept;
    void reader_loop() noexcept;
    void notify_reader() noexcept;
    // Copy [src_start, src_start+n) from the resident preload window into the
    // [dest_offset, dest_offset+n) region of dest. Audio thread.
    void copy_from_preload(BufferView<float>& dest, std::uint64_t dest_offset,
                           std::uint64_t src_start, std::uint64_t n) noexcept;
 
    // Immutable after prepare.
    std::uint32_t channels_ = 0;
    std::uint64_t total_frames_ = 0;
    std::uint32_t sample_rate_ = 0;
    std::uint64_t preload_valid_ = 0;
    std::uint64_t read_chunk_ = 0;
    bool loop_ = false;
    std::uint64_t loop_start_ = 0;
    std::uint64_t loop_end_ = 0;
    bool fully_resident_ = false;
    bool prepared_ = false;
    bool use_thread_ = false;
    DeadlinePolicy deadline_;
 
    Buffer<float> preload_;          
    PlanarAudioRingBuffer ring_;     
    Buffer<float> read_scratch_;     
    FrameReaderBinding reader_;
    std::atomic<bool> reader_stop_requested_{false};
 
    std::atomic<std::uint64_t> play_pos_{0};    
    std::atomic<std::uint64_t> reader_pos_{0};  
    // Effective end-of-stream in source frames: starts at total_frames_ and is
    // shrunk by the background reader to the realized length if the FrameReader
    // signals an early end/error mid-stream (a short/zero return before the
    // declared total). The audio thread reads it to terminate one-shot playback
    // and report finished() at the real end instead of stalling forever waiting
    // for frames that will never arrive. Written only by the background reader,
    // read by the audio thread — hence atomic (total_frames_ itself stays an
    // immutable-after-prepare plain member that the audio thread can read freely).
    std::atomic<std::uint64_t> eos_frame_{0};
    std::atomic<std::uint64_t> streamed_frames_{0};
    std::atomic<std::uint64_t> read_errors_{0};
    std::atomic<std::uint64_t> underrun_frames_{0};  
 
    std::thread thread_;
    std::atomic<bool> run_{false};
    // pull() (audio thread) and stop_thread() -> the background reader.
    runtime::WakeSignal wake_;
};
 
}  // namespace pulp::audio