buffering_reader.hpp

Go to the documentation of this file.

#pragma once

 
#include <pulp/runtime/wake_signal.hpp>
 
#include <vector>
#include <atomic>
#include <chrono>
#include <thread>
#include <functional>
#include <cstring>
#include <algorithm>
#include <limits>
 
namespace pulp::audio {

class BufferingReader {
public:
    using ReadCallback = std::function<int(float* dest, int num_frames, int num_channels)>;
 
    BufferingReader() = default;
 
    ~BufferingReader() {
        stop();
    }
 
    BufferingReader(const BufferingReader&) = delete;
    BufferingReader& operator=(const BufferingReader&) = delete;

    void set_read_callback(ReadCallback cb) { read_callback_ = std::move(cb); }

    void start(int num_channels, int buffer_frames = 44100) {
        stop();
 
        if (num_channels <= 0 || buffer_frames <= 0 ||
            num_channels > std::numeric_limits<int>::max() / buffer_frames ||
            num_channels > std::numeric_limits<int>::max() / kChunkFrames) {
            num_channels_ = 0;
            buffer_size_ = 0;
            write_pos_ = 0;
            read_pos_ = 0;
            available_.store(0);
            finished_.store(false);
            running_.store(false);
            return;
        }
 
        const auto sample_capacity =
            static_cast<int64_t>(buffer_frames) * static_cast<int64_t>(num_channels);
        num_channels_ = num_channels;
        buffer_size_ = static_cast<int>(sample_capacity);
        buffer_.resize(static_cast<size_t>(buffer_size_), 0.0f);
        write_pos_ = 0;
        read_pos_ = 0;
        available_.store(0);
        finished_.store(false);
        running_.store(true);
 
        thread_ = std::thread([this] { background_loop(); });
    }

    void stop() {
        running_.store(false);
        wake_.notify();
        if (thread_.joinable()) thread_.join();
        buffer_.clear();
    }

    int read(float* dest, int num_frames, int num_channels) {
        if (dest == nullptr || num_frames <= 0) return 0;
        if (num_channels != num_channels_ || buffer_.empty()) return 0;
        if (num_channels <= 0 ||
            num_frames > std::numeric_limits<int>::max() / num_channels) return 0;
 
        int samples_requested = num_frames * num_channels;
        int avail = available_.load(std::memory_order_acquire);
        int samples_to_read = std::min(samples_requested, avail);
 
        if (samples_to_read <= 0) {
            std::memset(dest, 0, static_cast<size_t>(samples_requested) * sizeof(float));
            return 0;
        }
 
        // Copy from ring buffer
        int first_chunk = std::min(samples_to_read, buffer_size_ - read_pos_);
        std::memcpy(dest, buffer_.data() + read_pos_,
                     static_cast<size_t>(first_chunk) * sizeof(float));
 
        int second_chunk = samples_to_read - first_chunk;
        if (second_chunk > 0) {
            std::memcpy(dest + first_chunk, buffer_.data(),
                         static_cast<size_t>(second_chunk) * sizeof(float));
        }
 
        read_pos_ = (read_pos_ + samples_to_read) % buffer_size_;
        available_.fetch_sub(samples_to_read, std::memory_order_release);
 
        // Wake the background thread if it is parked waiting for space. No
        // lock, and a system call only while it is actually asleep.
        wake_.notify();
 
        // Zero-fill any remaining samples
        int frames_read = samples_to_read / num_channels;
        if (frames_read < num_frames) {
            std::memset(dest + samples_to_read, 0,
                         static_cast<size_t>((num_frames - frames_read) * num_channels) * sizeof(float));
        }
 
        return frames_read;
    }

    bool is_finished() const { return finished_.load(); }

    bool is_running() const { return running_.load(); }

    int frames_available() const {
        return num_channels_ > 0 ? available_.load() / num_channels_ : 0;
    }
 
private:
    ReadCallback read_callback_;
    std::vector<float> buffer_;
    int buffer_size_ = 0;
    int num_channels_ = 0;
    int write_pos_ = 0;
    int read_pos_ = 0;
    std::atomic<int> available_{0};
    std::atomic<bool> running_{false};
    std::atomic<bool> finished_{false};
    std::thread thread_;
    // Audio thread (read) and control thread (stop) -> the background thread.
    runtime::WakeSignal wake_;
    static constexpr int kChunkFrames = 1024;
    // Only bounds the cost of a missed wake; refills are driven by read().
    static constexpr auto kIdleBackstop = std::chrono::seconds(1);
 
    void background_loop() {
        std::vector<float> temp(static_cast<size_t>(kChunkFrames * num_channels_));
 
        while (running_.load()) {
            // Sleep while the buffer is full; read() wakes this thread when it
            // frees space, and stop() when the reader shuts down. A full,
            // unread buffer therefore costs no wakeups at all.
            wake_.wait_for(
                [this] {
                    return !running_.load() ||
                           available_.load(std::memory_order_acquire) < buffer_size_;
                },
                kIdleBackstop);
 
            if (!running_.load()) break;
 
            int space = buffer_size_ - available_.load(std::memory_order_acquire);
            int frames_to_read = std::min(kChunkFrames, space / std::max(1, num_channels_));
 
            if (frames_to_read <= 0) continue;
 
            int frames_got = 0;
            if (read_callback_) {
                frames_got = read_callback_(temp.data(), frames_to_read, num_channels_);
            }
 
            if (frames_got <= 0) {
                finished_.store(true);
                running_.store(false);
                return;
            }
            if (frames_got > frames_to_read) frames_got = frames_to_read;
 
            int samples = frames_got * num_channels_;
            int first_chunk = std::min(samples, buffer_size_ - write_pos_);
            std::memcpy(buffer_.data() + write_pos_, temp.data(),
                         static_cast<size_t>(first_chunk) * sizeof(float));
 
            int second_chunk = samples - first_chunk;
            if (second_chunk > 0) {
                std::memcpy(buffer_.data(), temp.data() + first_chunk,
                             static_cast<size_t>(second_chunk) * sizeof(float));
            }
 
            write_pos_ = (write_pos_ + samples) % buffer_size_;
            available_.fetch_add(samples, std::memory_order_release);
        }
    }
};
 
} // namespace pulp::audio