13_spsc_live_handoff_queue.cpp

1// SpscLiveHandoffQueue: the pop-based interface.23#include "TwoBufQueue/src/base_components/live_handoff_queue/spsc_live_handoff_queue.hpp"4#include "TwoBufQueue/src/utils/auto_reset_event.hpp"5#include <atomic>6#include <cassert>7#include <cstdint>8#include <cstdio>9#include <optional>10#include <stdexcept>11#include <thread>12#include <utility>13using namespace TwoBufQueue;1415struct Message {16    int value = 0;17    std::atomic<bool> ready{false};18    bool consumed = false; // Accessed only by the consumer.1920    bool reset(std::uint64_t) noexcept {21        value = 0;22        consumed = false;23        ready.store(false, std::memory_order_relaxed);24        return true;25    }26    void publish(int new_value) noexcept {27        value = new_value;28        ready.store(true, std::memory_order_release);29    }30};3132using Queue = SpscLiveHandoffQueue<Message>;3334// Consumer thread.35class Consumer {36public:37    explicit Consumer(Queue& queue)38        : queue_(queue), thread_([this] {39            for (;;) {40                wake_.wait();41                if (stopping_.load(std::memory_order_acquire)) return;42                consume();43            }44        }) {45        queue_.set_read_available_callback(&Consumer::queue_notify, this);46    }4748    ~Consumer() {49        queue_.set_read_available_callback(nullptr, nullptr);50        stopping_.store(true, std::memory_order_release);51        wake_.set(); thread_.join();52    }5354    // Queue callback: wake the consumer without waiting for its scan to finish.55    void notify() { wake_.set(); }5657private:58    static void queue_notify(void* context) noexcept {59        static_cast<Consumer*>(context)->notify();60    }61    // Consumer thread: process this round's completed records before this round's current resource.62    void consume() {63        // pop has removed the record. On temporary failure, retain and retry it before later records.64        if (pending_) {65            if (pending_->get() && !process(**pending_, "completed")) return;66            queue_.finish(*pending_);67            pending_.reset();68        }6970        std::size_t finish_count = 0;71        // With records, use the last record's next; otherwise use queue.current().72        const auto current = queue_.snapshot_for_read(finish_count);7374        for (std::size_t i = 0; i < finish_count; ++i) {75            auto record = queue_.try_pop();76            assert(record); // No other consumer can remove this round's fixed queue prefix.7778            // A begin record has an empty get(); skip business processing but still acknowledge it with finish.79            if (record.get() && !process(*record, "completed")) {80                pending_.emplace(record);81                return; // Stop this round; leave later records and current for the next round.82            }83            queue_.finish(record); // Return a completed resource, if any; do not access it afterward.84        }85        // Records enqueued during processing belong to the next round; do not reload the tail or producer current here.86        if (current) process(*current, "current"); // Read the current block only; do not finish it.87    }8889    // Business handler: true means handled or no work; false requests a later retry.90    bool process(Message& message, const char* phase) {91        if (message.consumed) return true;92        if (!message.ready.load(std::memory_order_acquire)) return false;93        std::printf("%s=%d\n", phase, message.value);94        message.consumed = true;95        return true;96    }9798    Queue& queue_;99    std::optional<decltype(std::declval<Queue&>().try_pop())> pending_;100    AutoResetEvent wake_;101    std::atomic<bool> stopping_{false};102    std::thread thread_;   // Construct last so every member accessed by the thread is already initialized.103};104105// Producer helper: notify the consumer after changing shared data.106template<class TModify>107void modify_and_notify(Queue& queue, TModify&& modify) {108    modify();109    queue.notify_read_available();110}111112// This example retries until a resource is available. Applications must handle timeouts, backpressure and other states.113Queue::AcquireResult acquire_until_available(Queue& queue) {114    auto result = queue.try_acquire();115    while (!result) {116        std::this_thread::yield();117        result = queue.try_acquire();118    }119    return result;120}121122int main() try {123    //---------------1. Initialize a pool and ordered handoff queue (16)------124    constexpr std::size_t pool_min = 2, pool_max = 8, queue_capacity = 16;125    Queue queue(pool_min, pool_max, queue_capacity);126    {127        Consumer consumer(queue); // The consumer runs independently; the main thread produces data.128129        //---------------2. Read current before A is sent------------------------130        auto acquired = acquire_until_available(queue); // Enqueue the {empty, A} begin record.131        modify_and_notify(queue, [&] { acquired.loan->publish(7); }); // Notify the consumer to read current = 7.132133        //---------------3. Complete A and move to B-----------------------------134        acquired = queue.send_and_acquire();            // Enqueue {A, B}; do not access A afterward.135        if (!acquired) acquired = acquire_until_available(queue); // The old block has been sent; keep trying to acquire the next block.136137        modify_and_notify(queue, [&] { acquired.loan->publish(9); }); // Notify the consumer to read the current block B.138139        //---------------4. Finish all work-------------------------------------140        if (!queue.send()) throw std::runtime_error("send failed"); // {B, empty}.141        queue.close();142        while (!queue.all_released()) std::this_thread::yield(); // notify does not wait; wait here until the resource returns to the pool.143    } // Stop and join the consumer before destroying the queue and pool.144    if (!queue.is_drained() || !queue.all_released()) throw std::runtime_error("drain failed");145    std::puts("PASS: current read, pending retry and resource return");146} catch (const std::exception& error) {147    std::fprintf(stderr, "%s\n", error.what());148    return 1;149}