05_seq_writer.cpp

1#include "../../support/joining_thread.hpp"2// SeqWriter prefixes each record with a fixed-width little-endian sequence number; consumers can use it to check ordering.3// Each producer owns a SeqWriter; shared atomic fetch_add assigns globally unique sequence numbers.4// Preserve order within each Writer; records from different Writers may interleave without a global ordering guarantee.56#include "../example_support.hpp"7#include <array>8#include <atomic>9#include <cstring>10#include <thread>11using namespace TwoBufQueue;1213int main() try {14    //---------------1. Initialize 4 SeqWriters and TwoBufQueue---------------15    // SeqWriter<T> uses T for the sequence number; its callback takes one extra leading seq argument.16    using MyWriter = SeqWriter<std::uint32_t>;17    constexpr std::size_t writer_count = 4;18    constexpr std::uint32_t per_writer = 1000;19    std::atomic<std::uint32_t> next_seq{0};20    auto w0 = make_writer<MyWriter>(WriterSourceConfig{.thread_id = 0});21    auto w1 = make_writer<MyWriter>(WriterSourceConfig{.thread_id = 1});22    auto w2 = make_writer<MyWriter>(WriterSourceConfig{.thread_id = 2});23    auto w3 = make_writer<MyWriter>(WriterSourceConfig{.thread_id = 3});2425    std::array<std::size_t, writer_count> received{};26    std::array<std::uint32_t, writer_count> last_seq{};27    std::array<bool, writer_count * per_writer> seen{};28    bool valid = true;29    auto rcvCallback = [&](std::int32_t thread_id, std::uint32_t seq, const std::byte* data, std::size_t size) noexcept -> void {30        if (thread_id < 0 || thread_id >= static_cast<std::int32_t>(writer_count) ||31            size != sizeof(std::uint32_t) || seq >= seen.size() || seen[seq]) {32            valid = false;33            return;34        }35        std::uint32_t value = 0;36        std::memcpy(&value, data, sizeof(value));37        const auto id = static_cast<std::size_t>(thread_id);38        if (value != seq * 2 || (received[id] != 0 && seq <= last_seq[id])) valid = false;39        seen[seq] = true;40        last_seq[id] = seq;41        ++received[id];42    };4344    ::TwoBufQueue::TwoBufQueue queue(rcvCallback, {&w0, &w1, &w2, &w3});4546    //---------------2. Each producer writes through its Writer-------------47    std::array<bool, writer_count> written{};48    {49        std::array<OwnedWriter<MyWriter>*, writer_count> writers{&w0, &w1, &w2, &w3};50        std::array<brq_support::JoiningThread, writer_count> threads;51        for (std::size_t i = 0; i < writer_count; ++i)52            threads[i] = brq_support::JoiningThread([&, i] {53                written[i] = true;54                for (std::uint32_t n = 0; n < per_writer; ++n) {55                    const auto seq = next_seq.fetch_add(1, std::memory_order_relaxed);56                    if (!writers[i]->write(seq, static_cast<std::uint32_t>(seq * 2))) {57                        written[i] = false;58                        break;59                    }60                }61                writers[i]->finish(); // Each producer thread finishes its own Writer.62            });63    } // Join all producer threads before stopping the consumer.6465    //---------------3. Finish----------------------------------------------66    queue.stop();      // Wait for every record to be consumed and stop the reader threads.6768    //---------------4. Verify----------------------------------------------69    bool ok = valid && next_seq.load(std::memory_order_relaxed) == writer_count * per_writer;70    for (std::size_t i = 0; i < writer_count; ++i)71        ok = ok && written[i] && received[i] == per_writer;72    example::require(ok && queue.error_code() == ReadError::none, "seq delivery mismatch");73    std::puts("PASS: 4 writer threads, 4000 unique seq records, ordered per writer");74} catch (const std::exception& error) { std::fprintf(stderr, "%s\n", error.what()); return 1; }