slot.hpp

Go to the documentation of this file.

#pragma once

 
#include <pulp/runtime/spsc_queue.hpp>
 
#include <atomic>
#include <cassert>
#include <cstddef>
#include <memory>
#include <thread>
#include <utility>
#include <vector>
 
namespace pulp::runtime {

template <typename T>
class Slot {
public:
    class ReadGuard {
    public:
        ReadGuard() noexcept = default;
        ReadGuard(const ReadGuard&) = delete;
        ReadGuard& operator=(const ReadGuard&) = delete;
 
        ReadGuard(ReadGuard&& other) noexcept
            : owner_(other.owner_), value_(other.value_) {
            other.owner_ = nullptr;
            other.value_ = nullptr;
        }
 
        ~ReadGuard() {
            if (owner_ != nullptr)
                owner_->readers_.fetch_sub(1, std::memory_order_seq_cst);
        }
 
        T* get() const noexcept { return value_; }
        T& operator*() const noexcept { return *value_; }
        T* operator->() const noexcept { return value_; }
        explicit operator bool() const noexcept { return value_ != nullptr; }
 
    private:
        friend class Slot;
        ReadGuard(const Slot* owner, T* value) noexcept : owner_(owner), value_(value) {}
 
        const Slot* owner_ = nullptr;
        T* value_ = nullptr;
    };
 
    Slot() = default;
    Slot(const Slot&) = delete;
    Slot& operator=(const Slot&) = delete;

    ~Slot() { wait_and_clear(); }

    [[nodiscard]] ReadGuard read() const noexcept {
        readers_.fetch_add(1, std::memory_order_seq_cst);
        // seq_cst pairs with the store in publish() and the load in
        // reclaim_if_quiescent(): either a publisher observes this pin and
        // declines to free, or this load observes the publisher's new pointer.
        T* value = live_raw_.load(std::memory_order_seq_cst);
        if (value == nullptr) {
            readers_.fetch_sub(1, std::memory_order_seq_cst);
            return ReadGuard{};
        }
        return ReadGuard{this, value};
    }

    void prepare_publish() {
        reclaim_if_quiescent();
        if (live_)
            retired_.reserve(retired_.size() + 1);
    }

    void publish_prepared(std::shared_ptr<T> next) noexcept {
        assert((!live_ || retired_.size() < retired_.capacity()) &&
               "Slot::publish_prepared requires prepare_publish");
        live_raw_.store(next.get(), std::memory_order_seq_cst);
        if (live_)
            retired_.push_back(std::move(live_));
        live_ = std::move(next);
        reclaim_if_quiescent();
    }
 
    void publish(std::shared_ptr<T> next) {
        // Reserve before changing live_raw_: allocation failure must leave the
        // currently-published pointer and owner coherent.
        prepare_publish();
        publish_prepared(std::move(next));
    }
 
    void publish(std::unique_ptr<T> next) { publish(std::shared_ptr<T>(std::move(next))); }

    void unpublish() { publish(std::shared_ptr<T>{}); }

    void reclaim_if_quiescent() {
        if (readers_.load(std::memory_order_seq_cst) == 0)
            retired_.clear();
    }

    void wait_and_clear() {
        while (readers_.load(std::memory_order_seq_cst) != 0)
            std::this_thread::yield();
        retired_.clear();
    }

    const std::shared_ptr<T>& live() const noexcept { return live_; }

    std::size_t retired_count() const noexcept { return retired_.size(); }

    bool has_value() const noexcept {
        return live_raw_.load(std::memory_order_acquire) != nullptr;
    }
 
private:
    // Readers dereference `live_raw_` only. `live_` and `retired_` keep the
    // pointed-to objects alive and are touched by the publisher thread alone.
    std::shared_ptr<T> live_;
    std::atomic<T*> live_raw_{nullptr};
    std::vector<std::shared_ptr<T>> retired_;
    mutable std::atomic<std::uint32_t> readers_{0};
};

template <typename T, std::size_t RetireCapacity = 8>
class Handoff {
public:
    Handoff() = default;
    Handoff(const Handoff&) = delete;
    Handoff& operator=(const Handoff&) = delete;

    ~Handoff() {
        delete pending_.exchange(nullptr, std::memory_order_acquire);
        while (T* raw = pop_retired_raw()) delete raw;
    }

    bool publish(std::unique_ptr<T> next) {
        if (!next) return false;
        // Release ordering so the value's contents happen-before the audio
        // thread's acquire in try_consume().
        T* raw = next.release();
        delete pending_.exchange(raw, std::memory_order_acq_rel);
        return true;
    }

    [[nodiscard]] std::unique_ptr<T> try_consume() noexcept {
        return std::unique_ptr<T>(pending_.exchange(nullptr, std::memory_order_acquire));
    }

    [[nodiscard]] bool retire(std::unique_ptr<T>& displaced) noexcept {
        if (!displaced) return true;  // nothing to park
        if (!retired_.try_push(displaced.get())) return false;
        (void)displaced.release();  // ownership transferred to the ring
        return true;
    }

    bool has_retire_capacity(std::size_t n = 1) const noexcept {
        return retired_.size_approx() + n <= RetireCapacity;
    }

    std::size_t drain_retired() {
        std::size_t freed = 0;
        while (T* raw = pop_retired_raw()) {
            std::unique_ptr<T> owned(raw);
            ++freed;
        }
        return freed;
    }

    std::unique_ptr<T> drain_retired_one() { return std::unique_ptr<T>(pop_retired_raw()); }
 
    bool has_pending() const noexcept {
        return pending_.load(std::memory_order_acquire) != nullptr;
    }
    bool has_retired() const noexcept { return retired_.size_approx() > 0; }
 
    static constexpr std::size_t retire_capacity() noexcept { return RetireCapacity; }
 
private:
    T* pop_retired_raw() {
        auto popped = retired_.try_pop();
        return popped ? *popped : nullptr;
    }
 
    // Producer -> consumer: latest value awaiting pickup.
    std::atomic<T*> pending_{nullptr};
    // Consumer -> producer: displaced values awaiting off-thread deletion.
    // A ring rather than a single slot so rapid swaps survive a missed drain.
    SpscQueue<T*, RetireCapacity> retired_;
};
 
}  // namespace pulp::runtime