16_handoff_queue_mpmc.cpp

1#include "../../support/joining_thread.hpp"2// HandoffQueue in MPMC mode: multiple producers send; built-in Reader threads invoke callbacks concurrently.3// Consumer implements business processing only; no extra reader thread. Accumulate shared results with atomics.45#include "TwoBufQueue/src/base_components/handoff_queue/handoff_queue.hpp"6#include <atomic>7#include <cstdint>8#include <cstdio>9#include <stdexcept>10#include <thread>11#include <vector>12using namespace TwoBufQueue;1314struct Message {15    int value = 0;16    bool reset(std::uint64_t) noexcept { value = 0; return true; }17};1819class Consumer {20public:21    // true completes processing; this example does not demonstrate cancellation and retry after false.22    bool process(Message& message) noexcept {23        total_.fetch_add(message.value, std::memory_order_relaxed);24        return true; // The built-in Reader then calls finish to return the resource.25    }26    int total() const noexcept { return total_.load(std::memory_order_relaxed); }2728private:29    std::atomic<int> total_{0};30};3132// Retry only capacity backpressure; other acquisition errors terminate this example's thread.33template<class TQueue>34auto acquire_until_available(TQueue& queue) {35    for (;;) {36        auto acquired = queue.try_acquire();37        if (acquired) return acquired;38        if (acquired.status != AcquireStatus::QueueFull &&39            acquired.status != AcquireStatus::PoolLimit) std::terminate();40        std::this_thread::yield();41    }42}4344int main() try {45    //---------------1. Initialize the pool, consumer and queue--------------46    constexpr std::size_t pool_min = 2, pool_max = 16, queue_capacity = 32, reader_count = 2;47    constexpr int producer_count = 2, per_producer = 100;48    WarmPool<Message> pool(pool_min, pool_max);49    Consumer consumer;5051    HandoffOptions options;52    options.mode = HandoffMode::MPMC;53    options.queue_capacity = queue_capacity;54    options.reader_count = reader_count;55    options.reader_mode = ReaderMode::WaitSignal;56    options.cancel_claim_mode = CancelClaimMode::Keep;57    auto callback = [&consumer](Message& message) noexcept { return consumer.process(message); };58    HandoffQueue queue(pool, options, callback); // Construction starts reader_count reader threads.5960    //---------------2. Each producer borrows, writes and sends--------------61    {62        std::vector<brq_support::JoiningThread> producers;63        for (int i = 0; i < producer_count; ++i) {64            producers.emplace_back([&queue] {65                for (int value = 1; value <= per_producer; ++value) {66                    auto acquired = acquire_until_available(queue);67                    acquired.loan->value = value;68                    if (!queue.send(acquired.loan)) std::terminate();69                    // Do not access the resource after send.70                }71            });72        }73    } // Destruction of jthread waits for every producer to finish.7475    //---------------3. Drain and verify------------------------------------76    queue.close();77    while (!queue.is_drained()) std::this_thread::yield();78    queue.stop(); // Wait until the built-in Reader threads exit.7980    const int expected = producer_count * (per_producer * (per_producer + 1) / 2);81    if (consumer.total() != expected || !pool.all_released())82        throw std::runtime_error("delivery mismatch");83    std::puts("PASS: 2 producers + 2 consumers, 200 records summed");84} catch (const std::exception& error) {85    std::fprintf(stderr, "%s\n", error.what());86    return 1;87}