14_spsc_live_handoff_queue_claim.cpp

1// SpscLiveHandoffQueue: the claim/cancel interface.2// Compared with pop, claim/cancel keeps pending resources in the queue, simplifies retries and avoids losing a popped resource if the consumer exits unexpectedly.34#include "TwoBufQueue/src/base_components/live_handoff_queue/spsc_live_handoff_queue.hpp"5#include "TwoBufQueue/src/utils/auto_reset_event.hpp"6#include <atomic>7#include <cassert>8#include <cstdint>9#include <cstdio>10#include <stdexcept>11#include <thread>12using namespace TwoBufQueue;1314struct Message {15    int value = 0;16    std::atomic<bool> ready{false};17    bool consumed = false; // Accessed only by the consumer.1819    bool reset(std::uint64_t) noexcept {20        value = 0;21        consumed = false;22        ready.store(false, std::memory_order_relaxed);23        return true;24    }25    void publish(int new_value) noexcept {26        value = new_value;27        ready.store(true, std::memory_order_release);28    }29};3031using Queue = SpscLiveHandoffQueue<Message>;3233// Consumer thread.34class Consumer {35public:36    explicit Consumer(Queue& queue)37        : queue_(queue), thread_([this] {38            for (;;) {39                wake_.wait();40                if (stopping_.load(std::memory_order_acquire)) return;41                consume();42            }43        }) {44        queue_.set_read_available_callback(&Consumer::queue_notify, this);45    }4647    ~Consumer() {48        queue_.set_read_available_callback(nullptr, nullptr);49        stopping_.store(true, std::memory_order_release);50        wake_.set(); thread_.join();51    }5253    // Queue callback: wake the consumer without waiting for its scan to finish.54    void notify() { wake_.set(); }55    void allow_processing() noexcept { can_process_.store(true, std::memory_order_release); }56    int cancelled() const noexcept { return cancelled_.load(std::memory_order_acquire); }57    int current_processed() const noexcept { return current_processed_.load(std::memory_order_acquire); }5859private:60    static void queue_notify(void* context) noexcept {61        static_cast<Consumer*>(context)->notify();62    }63    // Consumer thread: process this round's completed records before this round's current resource.64    void consume() {65        std::size_t finish_count = 0;66        // With records, use the last record's next; otherwise use queue.current().67        const auto current = queue_.snapshot_for_read(finish_count);6869        for (std::size_t i = 0; i < finish_count; ++i) {70            auto claim = queue_.try_claim();71            assert(claim); // No other consumer can remove this round's fixed queue prefix.7273            if (claim.get() && !process(*claim, false)) {74                queue_.cancel_claim(claim);75                cancelled_.fetch_add(1, std::memory_order_release);76                return; // Leave the head record in place for a later retry.77            }78            queue_.finish(claim); // Dequeue and return any completed resource; do not access it afterward.79        }80        // Records enqueued during processing belong to the next round; do not reload the tail or producer current here.81        if (current) (void)process(*current, true); // Read the current block only; do not finish it.82    }8384    // true means no pending data or handled; false means conditions are not ready, so retry later.85    bool process(Message& message, bool current) {86        if (!message.ready.load(std::memory_order_acquire)) return current; // Do not finish a block whose completion has not been published.87        if (message.consumed) return true;88        if (!can_process_.load(std::memory_order_acquire)) return false;89        std::printf("%s=%d\n", current ? "current" : "completed", message.value);90        message.consumed = true;91        if (current) current_processed_.fetch_add(1, std::memory_order_release);92        return true;93    }9495    Queue& queue_;96    std::atomic<bool> can_process_{false};97    std::atomic<int> cancelled_{0};98    std::atomic<int> current_processed_{0};99    AutoResetEvent wake_;100    std::atomic<bool> stopping_{false};101    std::thread thread_;   // Construct last so every member accessed by the thread is already initialized.102};103104// Producer helper: notify the consumer after changing shared data.105template<class TModify>106void modify_and_notify(Queue& queue, TModify&& modify) {107    modify();108    queue.notify_read_available();109}110111// This example retries until a resource is available. Applications must handle timeouts, backpressure and other states.112Queue::AcquireResult acquire_until_available(Queue& queue) {113    auto result = queue.try_acquire();114    while (!result) {115        if (result.status != LiveAcquireStatus::QueueFull &&116            result.status != LiveAcquireStatus::PoolLimit)117            throw std::runtime_error("acquire failed");118        std::this_thread::yield();119        result = queue.try_acquire();120    }121    return result;122}123124int main() try {125    //---------------1. Initialize a pool and ordered handoff queue (16)------126    constexpr std::size_t pool_min = 2, pool_max = 8, queue_capacity = 16;127    Queue queue(pool_min, pool_max, queue_capacity);128    {129        Consumer consumer(queue); // The consumer runs independently; the main thread produces data.130131        //---------------2. Read current before A is sent------------------------132        auto acquired = acquire_until_available(queue); // Enqueue the {empty, A} begin record.133        modify_and_notify(queue, [&] { acquired.loan->publish(7); }); // Notify the consumer to read current = 7.134135        //---------------3. Complete A and move to B-----------------------------136        acquired = queue.send_and_acquire();            // Enqueue {A, B}; do not access A afterward.137        if (!acquired) acquired = acquire_until_available(queue); // The old block has been sent; keep trying to acquire the next block.138139        // A is complete but cannot be processed yet; first verify that the consumer cancelled once.140        while (consumer.cancelled() == 0) std::this_thread::yield();141        consumer.allow_processing();142        modify_and_notify(queue, [&] { acquired.loan->publish(9); }); // Retry A, then read the current block B.143        while (consumer.current_processed() == 0) std::this_thread::yield();144145        //---------------4. Finish all work-------------------------------------146        if (!queue.send()) throw std::runtime_error("send failed"); // {B, empty}.147        queue.close();148        while (!queue.all_released()) std::this_thread::yield(); // notify does not wait; wait here until the resource returns to the pool.149    } // Stop and join the consumer before destroying the queue and pool.150    if (!queue.is_drained() || !queue.all_released()) throw std::runtime_error("drain failed");151    std::puts("PASS: current read, claim/cancel retry and resource return");152} catch (const std::exception& error) {153    std::fprintf(stderr, "%s\n", error.what());154    return 1;155}