03_multi_thread_writer.cpp

1#include "../../support/joining_thread.hpp"2// Multiple producers: one thread owns each writer; one TwoBufQueue gathers all writers.3// Only one thread may call write on a writer (checked internally); thread_id identifies the source.4// read_layer1_thread_count enables multiple layer-1 readers, allowing concurrency at both ends.56#include "../example_support.hpp"7#include <array>8#include <cstring>9#include <thread>10using namespace TwoBufQueue;1112int main() try {13    //---------------1. Initialize 4 writers (unique thread_id) and 2 readers-14    // read_layer1_thread_count = 2: split the four writers evenly; each layer-1 reader copies two writers into layer 2 concurrently.15    constexpr std::size_t writer_count = 4, per_writer = 1000, read_layer1_thread_count = 2;16    auto w0 = make_writer<Writer<>>(WriterSourceConfig{.thread_id = 0});17    auto w1 = make_writer<Writer<>>(WriterSourceConfig{.thread_id = 1});18    auto w2 = make_writer<Writer<>>(WriterSourceConfig{.thread_id = 2});19    auto w3 = make_writer<Writer<>>(WriterSourceConfig{.thread_id = 3});2021    // Route callbacks by thread_id and verify each source's ordering and count.22    std::array<std::uint64_t, writer_count> next{};23    std::array<std::size_t, writer_count> received{};24    bool valid = true;25    auto rcvCallback = [&](std::int32_t thread_id, const std::byte* data, std::size_t size) noexcept -> void {26        if (thread_id < 0 || static_cast<std::size_t>(thread_id) >= writer_count ||27            size != sizeof(std::uint64_t)) { valid = false; return; }28        std::uint64_t value = 0;29        std::memcpy(&value, data, sizeof(value));30        const auto id = static_cast<std::size_t>(thread_id);31        if (value != next[id]++) valid = false;32        ++received[id];33    };3435    ::TwoBufQueue::TwoBufQueue queue(36        rcvCallback,37        {&w0, &w1, &w2, &w3},                                        // Multiple writers.38        TwoBufQueueOptions{.read_layer1_thread_count = read_layer1_thread_count}); // Multiple readers.3940    //---------------2. Each writer thread sends per_writer records----------41    std::array<bool, writer_count> written{};42    {43        std::array<OwnedWriter<Writer<>>*, writer_count> writers{&w0, &w1, &w2, &w3};44        std::array<brq_support::JoiningThread, writer_count> threads;45        for (std::size_t i = 0; i < writer_count; ++i)46            threads[i] = brq_support::JoiningThread([&, i] {47                written[i] = true;48                for (std::size_t j = 0; j < per_writer; ++j)49                    if (!writers[i]->write(static_cast<std::uint64_t>(j))) { written[i] = false; break; }50                writers[i]->finish();51            });52    } // Join every writer thread when leaving this scope.5354    //---------------3. Finish----------------------------------------------55    queue.stop();  // Consume all remaining records and stop every reader thread.5657    //---------------4. Verify each writer's ordering and record count-------58    bool ok = valid;59    for (std::size_t i = 0; i < writer_count; ++i)60        ok = ok && written[i] && received[i] == per_writer;61    example::require(ok && queue.error_code() == ReadError::none, "multi-thread writer mismatch");62    std::puts("PASS: 4 writer threads, 2 read threads, 4000 records");63} catch (const std::exception& error) { std::fprintf(stderr, "%s\n", error.what()); return 1; }