08_all_parameters.cpp

1// Writer / TwoBufQueue parameters: specify every template argument, constructor argument and configuration field.2// The main thread writes two records; TwoBufQueue's copy and callback threads run independently.3#include "../example_support.hpp"4#include <chrono>5using namespace TwoBufQueue;6using namespace std::chrono_literals;78int main() try {9    // 1. Writer<TPublishWritePtrCount, TIsWriteLen, TDataLen>10    // TPublishWritePtrCount = 4: publish after every four successful writes; 1 publishes each record. Must be positive.11    // flush() publishes a partial group; finish() and block switches also publish remaining data.12    // The uint32_t maximum publishes only on flush / block switch / finish; write does not decrement a publication counter.13    // Combine with is_scan_current = false below to reduce producer hot-path overhead.14    // TIsWriteLen = true adds a length prefix; false writes raw bytes and requires TwoBufQueueBatch.15    // TDataLen = uint32_t selects a four-byte prefix and the write_mem length argument's type.16    // Choose an unsigned integer type; payload length must fit it and the whole prefixed record must fit one source block.17    constexpr std::uint32_t publish_write_ptr_count = 4;18    constexpr bool is_write_len = true;19    using DataLen = std::uint32_t;20    using MyWriter = Writer<publish_write_ptr_count, is_write_len, DataLen>;2122    // 2. All WriterSourceConfig fields: source pool, source queue, source id and backpressure.23    const WriterSourceConfig source{24        .block_size = 4096,       // Bytes per source block, including record length prefixes.25        .pool_min = 2,            // Initially preallocated source blocks: 0 <= pool_min <= pool_max.26        .pool_max = 8,            // Maximum source block count, not message count; must be positive.27        .thread_id = 7,           // Nonnegative source identifier, not an OS thread id; Writer ids must be unique.28        .queue_capacity = 16,     // Source handoff-record capacity, including begin records; at least 2, independent of pool capacity.29        .backpressure = WriterBackpressureConfig{30            .mode = BackpressureMode::Block, // Wait when the next block is unavailable; Drop rejects the write immediately.31            .spin_count = 100,               // Block mode retries with pause this many times before timed waits.32            .wait_get_timeout = 2s,          // Total wait limit, including spinning; write returns false on timeout. Must be positive.33        },34    };3536    // 3. Construct the source pool, source queue and Writer explicitly; all must outlive TwoBufQueue.37    // WriterPool(min_count, max_count, factory, max_free_count): the last argument caps retained free blocks.38    constexpr std::size_t max_free_count = 8;39    WriterPool pool(source.pool_min, source.pool_max,40                    SpscBlockFactory<SpscWriteBlock>{source.block_size}, max_free_count);41    // make_writer_queue(pool, capacity, enable_stats): true records reasons for queue acquisition failure.42    constexpr bool enable_stats = true;43    auto source_queue = make_writer_queue(pool, source.queue_capacity, enable_stats);44    // Writer(queue, thread_id, backpressure): supply all three constructor arguments.45    MyWriter writer(&source_queue, source.thread_id, source.backpressure);46    // To avoid manually managing the source pool and queue, replace these three objects with:47    // auto writer = make_writer<MyWriter>(source); // Explicit WriterT; every source field is supplied above.48    // The resulting type is OwnedWriter<MyWriter>; use decltype(writer) as the queue's WriterT.4950    // 4. All TwoBufQueueOptions fields: output resources and readers; their capacities are independent of source.51    const TwoBufQueueOptions options{52        .output_block_size = 8192,    // Data bytes per output block; must hold the largest readable source-block range.53        .output_frame_capacity = 16,  // Maximum source ranges per output block; one frame can contain multiple records.54        .output_pool_min = 2,         // Initial output block count: 0 <= min <= max.55        .output_pool_max = 8,         // Maximum output block count; must be positive.56        .output_queue_capacity = 8,   // Output handoff queue slots; positive and independent of the output pool limit.57        .read_layer1_thread_count = 1,// Reader count, clamped to [1, Writer count]; assign Writers round-robin.58        .read_config = ReadConfig{59            .idle_spin_count = 100,          // Poll with pause this many times when no new completed block exists; zero waits immediately.60            .idle_rescan_interval = 1ms,     // Initial current-block scan interval; must be positive.61            .worker_cpu = -1,                // -1 disables CPU binding; nonnegative values select a logical CPU shared by all readers.62            .idle_rescan_max_interval = 16ms,// Maximum exponentially increased interval after consecutive empty scans; at least the initial interval.63            .finish_timeout = 2s,            // Total deadline for collecting source data during graceful stop; must be positive.64            .work_wait_min_interval = 1ms,   // Minimum ordinary wait; an event may wake it earlier. Must be positive.65            .finish_retry_interval = 1ms,    // Retry interval while graceful stop still has unread data; must be positive.66            .is_scan_current = true,         // false reads completed blocks only; tails arrive after a block switch or finish.67        },68        .cancel_claim_mode = CancelClaimMode::Keep, // Claim available blocks in original enqueue order; QueueTail requeues cancelled claims at the tail.69        .output_mode = HandoffMode::MPSC,           // Independent callback thread; multiple copy threads may share the output queue.70        .output_reader_mode = ReaderMode::WaitSignal, // Wait for a notification when empty; Polling continuously checks for work.71        .output_reader_count = 1,                  // Must be 1 for SPSC/MPSC; SPMC/MPMC allow 1..64.72        .output_retry_wait = 1ms,                  // WaitSignal retry delay after a rejected callback.73        .output_enable_stats = true,              // Enable output handoff queue statistics.74    };75    // Data resets the scan interval. Update scan timing when scanning, regardless of the callback's result.76    // flush / block switch / finish notifications wake only that source: read completed blocks, then check current immediately.77    // With is_scan_current = false, neither notifications nor timeouts check current; finish every Writer before stopping the queue.78    // flush does not wait for consumption. Scan timeouts visit all sources; space notifications force retries for all sources.79    // finish_timeout cannot interrupt a running business callback; callbacks must return promptly.80    // cancel_claim_mode affects the output HandoffQueue, not the Writer's ordered source queue.81    // Keep allows claimed blocks to run concurrently without ordered completion; HandoffOptions uses the same field name.82    // Per-record callbacks ignore return values and do not normally cancel; a false TwoBufQueueBatch callback retries using this mode.8384    // 5. TwoBufQueue<RecordCallback, WriterT>: supply both template arguments explicitly.85    // RecordCallback is the callback type; WriterT is the common type of every Writer passed to this queue.86    // Ordinary Writer callback arguments: source id, payload address and payload bytes, excluding the length prefix.87    // The callback must be noexcept; this per-record interface ignores its return value. data is valid only during the callback.88    // Callback that receives messages.89    auto on_record = [](std::int32_t thread_id, const std::byte* data, std::size_t size) noexcept {90        std::printf("writer=%d: %.*s\n", static_cast<int>(thread_id),91                    static_cast<int>(size), reinterpret_cast<const char*>(data));92    };93    // Single-Writer construction: TwoBufQueue(callback, writer reference, options); construction starts the reader.94    ::TwoBufQueue::TwoBufQueue<decltype(on_record), MyWriter> queue(on_record, writer, options);95    // For multiple Writers, replace only the second argument: queue(on_record, {&writer1, &writer2}, options).96    // Each Writer owns its source pool/queue, is used by one producer and has a distinct thread_id.97    // read_layer1_thread_count > 1 copies multiple Writers in parallel; independent threads invoke output callbacks.98    // A non-void TSyncContext enables per-record callbacks in the Writer/SeqWriter's producer thread; see 06_sync_writer.cpp.99    example::require(queue.started(), "reader failed to start");100101    // 6. Write only two records and explicitly publish before reaching four; finish the Writer, then drain the queue.102    constexpr char message[] = "hello";103    constexpr DataLen message_size = sizeof(message) - 1; // Exclude the terminating '\0'.104    const bool first = writer.write_mem(message, message_size);105    const bool second = writer.write('o', 'k');106    writer.flush();107    writer.finish();108    queue.stop();109    example::require(first && second && queue.error_code() == ReadError::none, "write/read failed");110    std::puts("PASS: all Writer and TwoBufQueue parameters specified");111} catch (const std::exception& error) { std::fprintf(stderr, "%s\n", error.what()); return 1; }