gpu_audio_transport.hpp

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <chrono>
#include <cstdint>
#include <memory>
#include <mutex>
#include <semaphore>
#include <thread>
#include <vector>
 
#include <pulp/audio/buffer.hpp>
#include <pulp/audio/planar_audio_ring_buffer.hpp>
#include <pulp/audio/workgroup.hpp>
#include <pulp/gpu_audio/gpu_audio_capability.hpp>
#include <pulp/gpu_audio/gpu_audio_node.hpp>
#include <pulp/gpu_audio/gpu_audio_status.hpp>
 
namespace pulp::gpu_audio {
 
namespace detail {
struct RealtimeGpuHealth;
struct GpuAudioTransportDiagnostics;
struct SharedIoCompletionNotifier;
class WorkerWake;
} // namespace detail

class GpuAudioTransport {
  public:
    // The node's descriptor is the single source of truth for channels, block
    // size, and latency. Config only carries transport knobs.
    struct Config {
        uint32_t ring_blocks = 4; // ring capacity, in blocks (>= latency+2)
        // Spawn an internal non-RT worker thread that drives pump(). By default
        // the worker polls the input ring, so the RT path stays fully decoupled
        // and lock-free. When false, the caller drives pump() (deterministic
        // tests / custom worker integration).
        bool run_worker_thread = false;
        // Opt-in: the RT process() posts a semaphore after each input write,
        // and a provider that installs a completion notifier posts it when a
        // completion may be ready. The worker waits on it, with the poll
        // interval as a fallback timeout only, instead of sleeping a full
        // quantum. Posting never allocates or blocks, but while the worker is
        // parked it is a kernel call (a Mach semaphore signal on macOS, the
        // platform's semaphore wake elsewhere) on the audio thread every
        // block, so it is off by default and the default RT path makes no
        // syscall. No shipped provider
        // installs a completion notifier yet, so today the worker still finds
        // completions by polling. Only meaningful when run_worker_thread is true.
        bool wake_on_write = false;
        // Optional macOS Audio Workgroup for the owned submission worker. The
        // handle is borrowed by the transport and must remain valid until
        // release(). This is an experiment switch: Dawn encode/submit still
        // runs on the worker and is not claimed to be realtime-safe.
        void* audio_workgroup = nullptr;
        bool join_audio_workgroup = false;
        GpuAudioRecoveryPolicy recovery_policy = GpuAudioRecoveryPolicy::Manual;
        std::uint64_t reacquire_release_budget_ns = 250'000'000;
        std::uint32_t reacquire_max_attempts = 3;
        std::uint64_t reacquire_backoff_ns = 100'000'000;
    };
 
    struct Stats {
        std::uint64_t produced_blocks = 0;      // blocks the worker completed
        std::uint64_t miss_blocks = 0;          // RT reads with no ready output
        std::uint64_t input_dropped_frames = 0; // RT writes lost to a full input ring
        // Wet blocks the worker dropped to realign the stream after a miss. Each
        // miss already emitted a substitute (dry/fallback) block for that timeline
        // slot, so its late-arriving wet counterpart is redundant; dropping it
        // keeps effective latency pinned at latency_blocks instead of creeping one
        // block per miss (which would comb-filter dry against wet).
        std::uint64_t resynced_blocks = 0;
        // Worker-observed wall time. The staged path measures one process_block
        // call, including any blocking readback. The shared path measures one
        // service call that reports progress; it may cover several completions.
        // Neither interval is GPU execution time, CPU consumption, or an audio
        // deadline measurement. last is the latest recorded interval; avg is an
        // EWMA. Both remain zero until the worker first reports progress.
        double last_block_us = 0.0;
        double avg_block_us = 0.0;
        bool worker_workgroup_joined = false;
        std::uint64_t worker_workgroup_join_failures = 0;
        // Appended: the owned worker runs at QOS_CLASS_USER_INTERACTIVE
        // (macOS; false when it joined a host workgroup instead, or off Apple).
        bool worker_interactive_qos = false;
    };

    struct DeliverySnapshot {
        std::uint64_t gpu_blocks = 0; // callback-side GPU path selected ready output
        // A generic node may implement process_gpu() using CPU work or internal
        // fallback. A ring delivery alone cannot establish GPU execution.
        std::uint64_t worker_output_blocks = 0;
        std::uint64_t cpu_fallback_blocks = 0;
        std::uint64_t silence_blocks = 0;
        std::uint64_t passthrough_blocks = 0;
        std::uint64_t priming_blocks = 0;
        std::uint64_t invalid_blocks = 0;
    };
 
    GpuAudioTransport();
    ~GpuAudioTransport();
 
    GpuAudioTransport(const GpuAudioTransport&) = delete;
    GpuAudioTransport& operator=(const GpuAudioTransport&) = delete;

    bool prepare(GpuAudioNode* node, const Config& config);
    void release() noexcept;
 
    bool is_prepared() const noexcept {
        return prepared_;
    }
    uint32_t latency_samples() const noexcept {
        return latency_blocks_ * block_size_;
    }

    void process(const audio::BufferView<const float>& input, audio::BufferView<float>& output,
                 uint32_t n) noexcept;

    void process_offline(const audio::BufferView<const float>& input,
                         audio::BufferView<float>& output, uint32_t n) noexcept;

    void pump(uint32_t max_blocks = 0) noexcept;
 
    Stats stats() const noexcept;

    GpuAudioStatus status_snapshot() const noexcept;

    DeliverySnapshot delivery_snapshot() const noexcept;

    GpuAudioCapabilityReport capability_report() const noexcept;
 
  private:
    // Undo the worker-pipelining opt-in prepare() bound, if any. Host thread,
    // no worker running.
    void release_pipeline_hosting() noexcept;
    using TrialDeliveryFn = void (*)(void*, std::uint64_t, std::uint8_t, std::uint64_t,
                                     std::uint64_t, std::uint64_t) noexcept;
    friend bool configure_gpu_audio_transport_trial_observer(GpuAudioTransport&, void*,
                                                             TrialDeliveryFn) noexcept;
 
    GpuAudioNode* node_ = nullptr;
    // Derived from the node descriptor + config at prepare(); authoritative for
    // the RT path so it never calls the (allocating) descriptor().
    uint32_t channels_ = 0;
    uint32_t block_size_ = 0;
    uint32_t sample_rate_ = 0;
    uint32_t latency_blocks_ = 0;
    uint32_t ring_blocks_ = 0;
    // Fails closed until prepare() copies the node's declared policy in.
    MissPolicy miss_policy_ = MissPolicy::Silence;
    bool prepared_ = false;
 
    void reset_staged_transport_state() noexcept;
    // Callback timeline only. Records whether `sequence`'s input block was
    // dropped and returns whether the block due now (`latency_blocks` earlier)
    // was.
    bool record_dropped_slot(std::uint64_t sequence, bool dropped) noexcept;
    void process_shared(const audio::BufferView<const float>&, audio::BufferView<float>&,
                        std::uint32_t, std::uint64_t, bool input_valid,
                        std::uint64_t callback_start_ns, bool count_delivery) noexcept;
    void process_realtime_position(const audio::BufferView<const float>&, audio::BufferView<float>&,
                                   std::uint32_t, std::uint64_t, bool input_valid,
                                   std::uint64_t callback_start_ns) noexcept;
    void process_offline_position(const audio::BufferView<const float>&, audio::BufferView<float>&,
                                  std::uint32_t, std::uint64_t, bool input_valid) noexcept;
    void process_invalid_position(audio::BufferView<float>&, std::uint64_t, bool offline) noexcept;
    void publish_trial_delivery(std::uint64_t sequence, std::uint8_t disposition,
                                std::uint64_t callback_start_ns, std::uint64_t callback_end_ns,
                                std::uint64_t result_visible_ns) noexcept;
    void record_delivery(std::uint8_t disposition, bool worker_output) noexcept;
 
    audio::PlanarAudioRingBuffer input_ring_;
    audio::PlanarAudioRingBuffer output_ring_;
 
    audio::Buffer<float> worker_in_;
    audio::Buffer<float> worker_out_;
    // Callback-owned scratch for rejected views. Zero input advances every
    // stateful delay/history position without reading the malformed input.
    audio::Buffer<float> rejected_input_;
    audio::Buffer<float> rejected_output_;
    std::vector<const float*> rejected_input_ptrs_;
    std::vector<float*> rejected_output_ptrs_;
    // Stable channel-pointer arrays for the worker views (BufferView holds the
    // array by reference, so it must outlive the views).
    std::vector<float*> in_fptrs_;
    std::vector<const float*> in_cptrs_;
    std::vector<float*> out_fptrs_;
 
    std::atomic<std::uint64_t> produced_blocks_{0};
    std::atomic<std::uint64_t> miss_blocks_{0};
    std::atomic<std::uint64_t> input_dropped_blocks_{0}; // whole-block input drops
    std::atomic<std::uint64_t> resynced_blocks_{0};      // late wet blocks dropped to realign
    std::atomic<std::uint64_t> delivery_gpu_{0}, delivery_worker_{0}, delivery_fallback_{0},
        delivery_silence_{0}, delivery_passthrough_{0}, delivery_priming_{0}, delivery_invalid_{0};
    // Resync debt: output slots a miss already substituted for, whose late wet
    // counterparts must still be dropped to realign the stream. Incremented on a
    // miss, decremented as those blocks are drained. Touched ONLY by process() on
    // the audio thread, so it needs no atomicity — counting misses explicitly
    // (rather than inferring "excess ring depth") keeps the drain immune to a
    // worker that races a fresh block in before the RT read of the same call.
    std::uint64_t blocks_owed_ = 0;
    // One flag per callback slot over a window of latency_blocks + 1, sized at
    // prepare(): set when that slot's input block was dropped on a full input
    // ring. Its wet block will never exist, so the callback `latency_blocks`
    // later substitutes without reading the ring and without adding resync
    // debt; reading there would return the next slot's block early and shift
    // the stream ahead of the declared latency. Written and read only on the
    // callback timeline.
    std::vector<std::uint8_t> dropped_slots_;
    // Per-block worker timing, published as integer nanoseconds for lock-free
    // reads by the UI thread (double isn't reliably lock-free).
    std::atomic<std::uint64_t> last_block_ns_{0};
    std::atomic<std::uint64_t> avg_block_ns_{0};
 
    // Optional internal worker. The worker polls the input ring; the RT path
    // never touches these.
    void worker_loop() noexcept;
    std::thread worker_;
    std::atomic<bool> worker_running_{false};
    std::chrono::microseconds poll_interval_{200};
    void* audio_workgroup_ = nullptr;
    bool join_audio_workgroup_ = false;
    std::atomic<bool> worker_workgroup_joined_{false};
    std::atomic<std::uint64_t> worker_workgroup_join_failures_{0};
    std::atomic<bool> worker_interactive_qos_{false};
 
    // Wake-on-write (Config::wake_on_write, opt-in). The RT process() posts
    // `wake_` after each input write, and an installed completion notifier
    // posts it from the provider's callback thread; the worker waits on it
    // (bounded by poll_interval_) instead of always sleeping the full quantum.
    // A post never allocates or blocks, but signals the kernel while the
    // worker is parked. When off, worker_loop() sleeps the poll interval.
    bool wake_on_write_ = false;
    std::unique_ptr<detail::WorkerWake> wake_; // allocated at construction
    // Provider completion events, routed to `wake_` while the owned worker
    // runs. Installed at prepare(), cleared in release() after the join.
    bool (*realtime_gpu_set_notifier_)(void*,
                                       detail::SharedIoCompletionNotifier) noexcept = nullptr;
    bool completion_notifier_installed_ = false;
    static void notify_worker(void* self) noexcept;
 
    // Private callback-side GPU path captured at prepare() from the concrete
    // node. These are plain function pointers so the RT process path does not
    // allocate, lock, or touch a public base-class extension.
    void* realtime_gpu_context_ = nullptr;
    std::uint8_t (*realtime_gpu_process_)(void*, const audio::BufferView<const float>&,
                                          audio::BufferView<float>&, std::uint32_t, std::uint64_t,
                                          bool, std::uint64_t) noexcept = nullptr;
    std::uint32_t (*realtime_gpu_service_)(void*, std::uint64_t) noexcept = nullptr;
    bool (*realtime_gpu_fence_)(void*) noexcept = nullptr;
    void (*realtime_gpu_delivered_)(void*, std::uint64_t, std::uint8_t, std::uint64_t,
                                    std::uint64_t) noexcept = nullptr;
    // Serialized non-RT health read for the same context. The worker
    // republishes every node health sample, and prepare()/release() publish
    // the prepared configuration, into `diagnostics_`; status_snapshot() and
    // capability_report() read only those copies. Allocated at construction.
    bool (*realtime_gpu_health_)(const void*, detail::RealtimeGpuHealth&) noexcept = nullptr;
    bool sample_staged_health_ = false;
    std::unique_ptr<detail::GpuAudioTransportDiagnostics> diagnostics_;
    void refresh_node_health() noexcept;
    void publish_prepared_configuration() noexcept;
 
    // Worker-only re-acquire steward (Config::recovery).
    void steward_reacquire() noexcept;
    bool (*realtime_gpu_reprepare_)(void*, std::uint64_t) noexcept = nullptr;
    GpuAudioRecoveryPolicy recovery_policy_ = GpuAudioRecoveryPolicy::Manual;
    std::uint64_t reacquire_budget_ns_ = 0;
    std::uint32_t reacquire_max_attempts_ = 0;
    std::uint64_t reacquire_backoff_ns_ = 0;
    std::uint32_t reacquire_round_ = 0;   // failed attempts in this outage
    std::uint64_t outage_started_ns_ = 0; // worker's first sight of the fence
    std::uint64_t next_reacquire_ns_ = 0;
    // A success armed resume_pending_from_ns_; once the callback clears it the
    // re-acquired epoch has delivered GPU audio, even if a new fault latched
    // before the worker saw the healthy state.
    bool resume_armed_ = false;
    std::atomic<std::uint64_t> reacquire_attempts_{0};
    std::atomic<std::uint64_t> reacquire_successes_{0};
    std::atomic<std::uint64_t> last_reacquire_to_gpu_ns_{0};
    // Outage start of a successful re-acquire whose first GPU block the
    // callback has not delivered yet; the callback measures and clears it.
    std::atomic<std::uint64_t> resume_pending_from_ns_{0};
    std::uint64_t callback_sequence_ = 0; // one monotonic RT/offline callback timeline
    bool realtime_gpu_fenced_for_offline_ = false;
    // Optional node hook bracketing each offline run; `realtime_gpu_offline_`
    // is set by its first block and cleared by the next realtime callback.
    void (*realtime_gpu_offline_render_)(void*, bool) noexcept = nullptr;
    bool realtime_gpu_offline_ = false;
 
    // Synchronous (offline) drive. When `synchronous_` is set, the background
    // worker yields and process_offline() pumps the node inline under
    // `pump_mutex_`; the realtime process() clears the flag so the worker
    // resumes. `pump_mutex_` only ever contends between the offline path and the
    // worker (both off the audio thread) — process() never takes it, so the
    // realtime path stays lock-free.
    std::atomic<bool> synchronous_{false};
    std::mutex pump_mutex_;
 
    // Host-only P4 trial observer. The observer is installed before the
    // callback starts and must itself be realtime-safe. The default runtime
    // path leaves it null, so this adds no hot-path work for normal users.
    void* trial_observer_context_ = nullptr;
    TrialDeliveryFn trial_observer_ = nullptr;
 
    // True only while the hosted node pipelines its worker
    // (worker_pipeline_blocks > 0 after the opt-in at prepare()). release()
    // opts such a node back out; any other node is never touched there, so a
    // host may destroy it before release().
    bool pipeline_hosted_ = false;
};
 
} // namespace pulp::gpu_audio