buffered_content_source.hpp

Go to the documentation of this file.

#pragma once

 
#include <algorithm>
#include <atomic>
#include <cstdint>
#include <functional>
#include <thread>
#include <utility>
 
#include <pulp/audio/buffer.hpp>
#include <pulp/audio/streaming_sample_source.hpp>
#include <pulp/timeline/production_mode.hpp>
 
namespace pulp::playback {

using BufferedContentEpoch = std::uint64_t;

using BufferedContentProducer =
    std::function<std::uint64_t(BufferedContentEpoch epoch, std::uint64_t start_frame,
                                audio::BufferView<float> destination,
                                std::uint64_t frames, audio::FrameReaderStopToken stop_token)>;
 
struct BufferedContentSourceConfig {
    std::uint32_t channels = 0;
    std::uint64_t declared_frames = 0;
    std::uint32_t sample_rate = 0;
    std::uint64_t max_pull_frames = 0;
    std::uint64_t ring_capacity_frames = 0;
    std::uint64_t produce_chunk_frames = 4096;
    std::uint64_t preroll_frames = 0;
    std::uint64_t loop_start_frame = 0;
    std::uint64_t loop_end_frame = 0;
    BufferedContentEpoch initial_epoch = 1;
    bool start_background_thread = false;
};
 
class BufferedContentSource {
  public:
    BufferedContentSource() = default;
    ~BufferedContentSource() { release(); }
 
    BufferedContentSource(const BufferedContentSource&) = delete;
    BufferedContentSource& operator=(const BufferedContentSource&) = delete;
 
    struct Stats {
        std::uint64_t produced_frames = 0;
        std::uint64_t starved_frames = 0;
        std::uint64_t starvation_events = 0;
        std::uint64_t ring_underrun_frames = 0;
        std::uint64_t ring_available_frames = 0;
        std::uint64_t producer_errors = 0;
    };

    bool prepare(const timeline::ProductionDeclaration& declaration,
                 const BufferedContentSourceConfig& config, BufferedContentProducer producer) {
        release();
        if (declaration.mode != timeline::ProductionMode::Buffered)
            return false;
        if (timeline::validate_production_declaration(declaration) !=
            timeline::ProductionDeclarationErrorCode::None)
            return false;
        if (config.channels == 0 || config.declared_frames == 0 || config.sample_rate == 0 ||
            config.max_pull_frames == 0 || !producer)
            return false;
        if (config.initial_epoch == 0)
            return false;
        if (config.preroll_frames > config.declared_frames)
            return false;
        const bool declares_loop = config.loop_start_frame != 0 || config.loop_end_frame != 0;
        if (declares_loop && (config.loop_start_frame >= config.loop_end_frame ||
                              config.loop_end_frame > config.declared_frames))
            return false;
 
        const std::uint64_t declared_lookahead =
            std::max<std::uint64_t>(
                lookahead_frames(declaration, config.sample_rate), 1);
        if (config.ring_capacity_frames != 0 &&
            (config.ring_capacity_frames < config.max_pull_frames ||
             config.ring_capacity_frames < config.preroll_frames))
            return false;
 
        declaration_ = declaration;
        declared_frames_ = config.declared_frames;
        preroll_frames_ = config.preroll_frames;
        loop_start_frame_ = config.loop_start_frame;
        loop_end_frame_ = config.loop_end_frame;
        declared_lookahead_ = declared_lookahead;
        config_ = config;
        producer_ = std::move(producer);
        epoch_.store(config.initial_epoch, std::memory_order_relaxed);
        base_frame_ = 0;
        if (!start_stream()) {
            release();
            return false;
        }
        prepared_ = true;
        return true;
    }

    void release() noexcept {
        // Ask an in-flight producer call to cut short before the stream joins
        // its thread, so the wait stays bounded by the remainder of one chunk.
        producer_stop_requested_.store(true, std::memory_order_release);
        stream_.release();
        producer_stop_requested_.store(false, std::memory_order_release);
        declaration_ = {};
        declared_frames_ = 0;
        preroll_frames_ = 0;
        loop_start_frame_ = 0;
        loop_end_frame_ = 0;
        declared_lookahead_ = 0;
        config_ = {};
        producer_ = nullptr;
        epoch_.store(0, std::memory_order_relaxed);
        base_frame_ = 0;
        prepared_ = false;
        produced_frames_.store(0, std::memory_order_relaxed);
        starved_frames_.store(0, std::memory_order_relaxed);
        starvation_events_.store(0, std::memory_order_relaxed);
    }

    bool seek(std::uint64_t frame, BufferedContentEpoch epoch) {
        if (!prepared_)
            return false;
        if (epoch == 0 || epoch <= current_epoch())
            return false;
        if (frame >= declared_frames_)
            return false;
        if (preroll_frames_ > declared_frames_ - frame)
            return false;
 
        // Publish the epoch before teardown: any producer call still running
        // against the old ring becomes recognizable as superseded, and the
        // stop request keeps its remaining run inside the current chunk.
        epoch_.store(epoch, std::memory_order_release);
        producer_stop_requested_.store(true, std::memory_order_release);
        stream_.release();
        producer_stop_requested_.store(false, std::memory_order_release);
        base_frame_ = frame;
        if (!start_stream()) {
            release();
            return false;
        }
        return true;
    }

    bool loop_wrap(BufferedContentEpoch epoch) {
        if (!prepared_ || (loop_start_frame_ == 0 && loop_end_frame_ == 0))
            return false;
        return seek(loop_start_frame_, epoch);
    }

    void cancel_production() noexcept {
        if (!prepared_)
            return;
        producer_stop_requested_.store(true, std::memory_order_release);
        while (producer_calls_in_flight_.load(std::memory_order_acquire) != 0)
            std::this_thread::yield();
        producer_stop_requested_.store(false, std::memory_order_release);
    }

    std::uint64_t pull(audio::BufferView<float> destination, std::uint64_t frames) noexcept {
        frames = std::min<std::uint64_t>(frames, destination.num_samples());
        const std::uint64_t position = base_frame_ + stream_.position();
        const std::uint64_t remaining =
            declared_frames_ > position ? declared_frames_ - position : 0;
        const std::uint64_t expected = std::min(frames, remaining);
        const std::uint64_t produced = stream_.pull(destination, frames);
        produced_frames_.fetch_add(produced, std::memory_order_relaxed);
        if (produced < expected) {
            starved_frames_.fetch_add(expected - produced, std::memory_order_relaxed);
            starvation_events_.fetch_add(1, std::memory_order_relaxed);
        }
        return produced;
    }

    std::uint64_t pump_background() noexcept {
        return stream_.pump_background();
    }
 
    bool prepared() const noexcept {
        return prepared_;
    }
    const timeline::ProductionDeclaration& declaration() const noexcept {
        return declaration_;
    }
    std::uint64_t declared_frames() const noexcept {
        return declared_frames_;
    }

    BufferedContentEpoch current_epoch() const noexcept {
        return epoch_.load(std::memory_order_acquire);
    }

    std::uint64_t loop_start_frame() const noexcept {
        return loop_start_frame_;
    }
    std::uint64_t loop_end_frame() const noexcept {
        return loop_end_frame_;
    }

    std::uint64_t position() const noexcept {
        return base_frame_ + stream_.position();
    }

    bool exhausted() const noexcept {
        return position() >= declared_frames_;
    }
 
    Stats stats() const noexcept {
        const auto stream_stats = stream_.stats();
        Stats result;
        result.produced_frames = produced_frames_.load(std::memory_order_relaxed);
        result.starved_frames = starved_frames_.load(std::memory_order_relaxed);
        result.starvation_events = starvation_events_.load(std::memory_order_relaxed);
        result.ring_underrun_frames = stream_stats.underrun_frames;
        result.ring_available_frames = stream_stats.ring_available_frames;
        result.producer_errors = stream_stats.read_errors;
        return result;
    }

    static std::uint64_t lookahead_frames(const timeline::ProductionDeclaration& declaration,
                                          std::uint32_t sample_rate) noexcept {
        return declaration.lookahead_ms * static_cast<std::uint64_t>(sample_rate) / 1000u;
    }
 
  private:
    bool start_stream() {
        audio::StreamingSampleSourceConfig stream_config;
        stream_config.channels = config_.channels;
        stream_config.total_frames = declared_frames_ - base_frame_;
        stream_config.sample_rate = config_.sample_rate;
        // No resident head: buffered content is produced on demand, so every
        // frame must travel through the ring where it can be counted.
        stream_config.preload_frames = 0;
        stream_config.ring_capacity_frames =
            config_.ring_capacity_frames != 0
                ? config_.ring_capacity_frames
                : std::max({declared_lookahead_, config_.max_pull_frames, preroll_frames_});
        stream_config.read_chunk_frames = config_.produce_chunk_frames;
        stream_config.start_background_thread = config_.start_background_thread;
        // Generated content is tied to playhead deadlines. Once a frame has
        // been emitted as starvation silence, producing it later must not move
        // subsequent content out of time.
        stream_config.underrun_policy =
            audio::StreamingUnderrunPolicy::AdvancePosition;
        stream_config.max_read_ahead_frames =
            std::max(declared_lookahead_, preroll_frames_);
 
        audio::FrameReaderBinding binding;
        binding.read = [this](std::uint64_t stream_frame, audio::BufferView<float> destination,
                              std::uint64_t frames, audio::FrameReaderStopToken stop_token) {
            return produce_for_stream(stream_frame, destination, frames, stop_token);
        };
        if (!stream_.prepare(stream_config, std::move(binding)))
            return false;
        // The stream primes its ring synchronously during prepare; the
        // declared preroll is the part of that fill the source guarantees.
        if (preroll_frames_ > 0 &&
            stream_.stats().ring_available_frames < preroll_frames_) {
            stream_.release();
            return false;
        }
        return true;
    }

    std::uint64_t produce_for_stream(std::uint64_t stream_frame,
                                     audio::BufferView<float> destination,
                                     std::uint64_t frames,
                                     audio::FrameReaderStopToken stream_stop_token) {
        if (stream_stop_token.stop_requested())
            return 0;
        const BufferedContentEpoch call_epoch = epoch_.load(std::memory_order_acquire);
        producer_calls_in_flight_.fetch_add(1, std::memory_order_acq_rel);
        struct InFlight {
            std::atomic<std::uint32_t>& count;
            ~InFlight() {
                count.fetch_sub(1, std::memory_order_acq_rel);
            }
        } in_flight{producer_calls_in_flight_};
        const std::uint64_t produced =
            producer_(call_epoch, base_frame_ + stream_frame, destination, frames,
                      audio::FrameReaderStopToken{producer_stop_requested_});
        // The producer may ignore the stop token and return frames anyway; an
        // interrupted chunk publishes nothing either way.
        if (producer_stop_requested_.load(std::memory_order_acquire))
            return 0;
        return produced;
    }
 
    audio::StreamingSampleSource stream_;
    timeline::ProductionDeclaration declaration_;
    std::uint64_t declared_frames_ = 0;
    std::uint64_t preroll_frames_ = 0;
    std::uint64_t loop_start_frame_ = 0;
    std::uint64_t loop_end_frame_ = 0;
    std::uint64_t declared_lookahead_ = 0;
    BufferedContentSourceConfig config_{};
    BufferedContentProducer producer_;
    std::uint64_t base_frame_ = 0;
    bool prepared_ = false;
 
    std::atomic<BufferedContentEpoch> epoch_{0};
    std::atomic<bool> producer_stop_requested_{false};
    std::atomic<std::uint32_t> producer_calls_in_flight_{0};
 
    std::atomic<std::uint64_t> produced_frames_{0};
    std::atomic<std::uint64_t> starved_frames_{0};
    std::atomic<std::uint64_t> starvation_events_{0};
};
 
} // namespace pulp::playback