15_handoff_queue.cpp

1// HandoffQueue in SPSC mode: the main thread produces; one built-in Reader invokes the consumer callback.23#include "TwoBufQueue/src/base_components/handoff_queue/handoff_queue.hpp"4#include <atomic>5#include <cstdint>6#include <cstdio>7#include <stdexcept>8#include <thread>9using namespace TwoBufQueue;1011struct Message {12    int value = 0;13    bool reset(std::uint64_t) noexcept { value = 0; return true; }14};1516// HandoffQueue's built-in Reader thread calls this consumer logic.17class Consumer {18public:19    // true completes processing; false makes the Reader cancel its claim and retry later.20    bool process(Message& message) noexcept {21        if (!can_process_.load(std::memory_order_acquire)) {22            deferred_attempts_.fetch_add(1, std::memory_order_release);23            std::printf("defer value=%d\n", message.value);24            return false;25        }26        total_.fetch_add(message.value, std::memory_order_relaxed);27        received_.fetch_add(1, std::memory_order_relaxed);28        std::printf("processed value=%d\n", message.value);29        return true;30    }31    void allow_processing() noexcept { can_process_.store(true, std::memory_order_release); }32    int deferred_attempts() const noexcept { return deferred_attempts_.load(std::memory_order_acquire); }33    int received() const noexcept { return received_.load(std::memory_order_acquire); }34    int total() const noexcept { return total_.load(std::memory_order_acquire); }3536private:37    std::atomic<bool> can_process_{false};38    std::atomic<int> deferred_attempts_{0};39    std::atomic<int> received_{0};40    std::atomic<int> total_{0};41};4243// Retry only capacity backpressure in this example; report other acquisition errors to the caller.44template<class TQueue>45auto acquire_until_available(TQueue& queue) {46    for (;;) {47        auto acquired = queue.try_acquire();48        if (acquired) return acquired;49        if (acquired.status != AcquireStatus::QueueFull &&50            acquired.status != AcquireStatus::PoolLimit)51            throw std::runtime_error("acquire failed");52        std::this_thread::yield();53    }54}5556int main() try {57    //---------------1. Initialize the pool, consumer and queue--------------58    constexpr std::size_t pool_min = 2, pool_max = 8, queue_capacity = 8;59    WarmPool<Message> pool(pool_min, pool_max);60    Consumer consumer;6162    HandoffOptions options;63    options.mode = HandoffMode::SPSC;64    // For single-threaded debugging, use HandoffMode::Sync: no Reader thread; send()/retry_pending() invoke the callback in their calling thread.65    options.queue_capacity = queue_capacity;66    options.cancel_claim_mode = CancelClaimMode::Keep;67    auto callback = [&consumer](Message& message) noexcept { return consumer.process(message); };68    HandoffQueue queue(pool, options, callback);6970    //---------------2. Abandon an unsent resource---------------------------71    auto unused = acquire_until_available(queue);72    if (!queue.discard(unused.loan)) throw std::runtime_error("discard failed");7374    //---------------3. Produce: borrow, write and send----------------------75    for (int value : {10, 20, 30}) {76        auto acquired = acquire_until_available(queue);77        acquired.loan->value = value;78        if (!queue.send(acquired.loan)) throw std::runtime_error("send failed");79        // send notifies the Reader; the consumer callback runs on the Reader thread.80    }8182    //---------------4. Once ready, let the Reader retry and drain-----------83    while (consumer.deferred_attempts() == 0) std::this_thread::yield();84    const bool deferred = consumer.received() == 0;85    consumer.allow_processing();86    queue.close();87    while (!queue.is_drained()) std::this_thread::yield();88    queue.stop();8990    if (!deferred || consumer.received() != 3 || consumer.total() != 60 ||91        !pool.all_released()) throw std::runtime_error("delivery mismatch");92    std::puts("PASS: discard, callback retry, 3 messages summed to 60");93} catch (const std::exception& error) {94    std::fprintf(stderr, "%s\n", error.what());95    return 1;96}