handoff_queue.hpp

1// HandoffQueue combines a resource pool with a producer-consumer queue, supporting SPSC, MPSC, SPMC, MPMC and Sync.2// Producers borrow pooled resources and populate them; completed consumption returns each resource to the pool.3// Producer: borrow from the pool (try_acquire) -> populate the resource -> hand it to the queue (send).4//                                          -> cannot produce now -> return the resource and reserved slot (discard).5// Consumer: claim a queued resource (try_claim) -> callback returns true -> acknowledge and return it (finish).6//                                              -> callback returns false -> cancel; cancel_claim_mode controls retry order.7// Differences from a conventional producer-consumer queue:8// 1. An external pool owns reusable resources; this example uses WarmPool. An upstream queue needs a separate adapter.9// 2. Claiming is not completion: cancel when processing is not possible; the same or another consumer can retry.10//    Keep prioritizes earlier enqueued claims; QueueTail returns them to the tail. Both retain the resource and slot and do not undo business effects.11// 3. try_acquire also reserves a queue slot, so a valid send cannot fail because the queue is full. Configure pool and queue capacities separately.12//    The producer exclusively owns an unsent resource and must stop accessing it after send; consumers handle handed-off resources only.13// 4. The built-in Reader claims, invokes callbacks and finishes/cancels. Applications supply bool(Resource&) noexcept.14// SPSC/MPSC use one built-in reader; SPMC/MPMC allow multiple readers. Choose continuous polling or signal-based waits.15//16// Compared with pop, claim/cancel: a. simplifies consumers by keeping pending resources in the queue;17//                               b. avoids losing a removed resource when a consumer exits unexpectedly;18//                               c. reorders cancelled claims in the queue so earlier eligible work can be claimed first.19#pragma once2021#include "pooled_resource_queue.hpp"22#include "handoff_reader.hpp"2324namespace TwoBufQueue {2526// Scheduling facade: compose the resource queue and consumer behind one producer/consumer interface.27// Owns no slots, queue locks, resource state or threads; the separate components perform those operations.28template<class TPool, class TCallback, class TResourceQueue = PooledResourceQueue<TPool>>29class HandoffQueue {30public:31    using Resource = typename TResourceQueue::Resource;32    using Loan = typename TResourceQueue::Loan;33    template<bool TC> using Ticket = typename TResourceQueue::template Ticket<TC>;34    using ProducerLoan = typename TResourceQueue::ProducerLoan;35    using Claim = typename TResourceQueue::Claim;36    using AcquireResult = typename TResourceQueue::AcquireResult;37    using NotifyCallback = void (*)(void*) noexcept;3839    HandoffQueue(TPool& pool, HandoffOptions options, TCallback callback)40        : options_(validate_handoff_options(options)),41          queue_(pool, options_.queue_capacity, options_.enable_stats, options_.cancel_claim_mode),42          reader_(queue_, options_, std::move(callback)) {}43    HandoffQueue(const HandoffQueue&) = delete;44    HandoffQueue& operator=(const HandoffQueue&) = delete;45    ~HandoffQueue() noexcept { close(); stop(); }4647    [[nodiscard]] AcquireResult try_acquire() noexcept { return queue_.try_acquire(); }48    bool send(ProducerLoan& loan) noexcept {49        if (!queue_.send(loan)) return false;50        notify_read_available();51        reader_.notify();52        return true;53    }54    bool discard(ProducerLoan& loan) noexcept { return queue_.discard(loan); }55    [[nodiscard]] Loan pop() noexcept { return queue_.pop(); }56    [[nodiscard]] Loan try_receive() noexcept { return queue_.try_receive(); }57    [[nodiscard]] Claim try_claim() noexcept { return queue_.try_claim(); }58    bool finish(Loan& loan) noexcept { return queue_.finish(loan); }59    bool finish(Claim& claim) noexcept { return queue_.finish(claim); }60    template<class TTicketType> [[nodiscard]] HandoffState state(const TTicketType& ticket) const noexcept {61        return queue_.state(ticket);62    }63    bool cancel_claim(Claim& claim) noexcept {64        if (!queue_.cancel_claim(claim)) return false;65        notify_read_available();66        if (options_.mode != HandoffMode::Sync) reader_.notify();67        return true;68    }69    // Producers must be quiescent while installing/removing this callback; notify only, never reenter the production interface.70    void set_read_available_callback(NotifyCallback callback, void* context) noexcept {71        notify_ = callback;72        notify_context_ = context;73    }74    void retry_pending() noexcept { reader_.process_pending(); }75    void close() noexcept { queue_.close(); }76    [[nodiscard]] bool is_closed() const noexcept { return queue_.is_closed(); }77    [[nodiscard]] bool is_drained() const noexcept {78        return queue_.is_drained() && !reader_.is_processing();79    }80    [[nodiscard]] std::size_t capacity() const noexcept { return queue_.capacity(); }81    [[nodiscard]] HandoffStats stats() const noexcept { return queue_.stats(); }82    [[nodiscard]] HandoffMode mode() const noexcept { return options_.mode; }83    bool start() noexcept { return reader_.start(); }84    void stop() noexcept { reader_.stop(); }85    [[nodiscard]] bool is_running() const noexcept { return reader_.is_running(); }8687private:88    void notify_read_available() noexcept {89        if (notify_) notify_(notify_context_);90    }9192    const HandoffOptions options_;93    TResourceQueue queue_;94    HandoffReader<TResourceQueue, TCallback> reader_;95    NotifyCallback notify_ = nullptr;96    void* notify_context_ = nullptr;97};9899template<class TPool, class TCallback>100HandoffQueue(TPool&, HandoffOptions, TCallback) -> HandoffQueue<TPool, TCallback>;101102} // namespace TwoBufQueue