09_async_logger.cpp

1// Copyright 2026 chuan lu2// SPDX-License-Identifier: Apache-2.03#include "../example_support.hpp"4#include <array>5#include <chrono>6#include <fstream>7#include <memory>8#include <string>9#include <thread>10#include <vector>11using namespace TwoBufQueue;1213int main(int argc, char** argv) try {14    // Usage: brq_tbq_09_async_logger [output-file] [--slow-sink] [--drop]15    const char* path = "twobufqueue-example.log";16    bool slow_sink = false, drop = false, path_set = false;17    for (int i = 1; i < argc; ++i) {18        if (std::string_view{argv[i]} == "--slow-sink") slow_sink = true;19        else if (std::string_view{argv[i]} == "--drop") drop = true;20        else if (!path_set && !std::string_view{argv[i]}.starts_with("--")) {21            path = argv[i]; path_set = true;22        } else throw std::invalid_argument("expected [output-file] [--slow-sink] [--drop]");23    }24    struct LogRecord {25        std::uint64_t sequence;26        std::uint32_t request_id;27        std::array<char, 80> text;28    };2930    //---------------1. Initialization--------------------------------------31    // File lifetime covers the callback threads. Only the callback writes it.32    const auto close_file = [](std::FILE* handle) noexcept { (void)std::fclose(handle); };33    using File = std::unique_ptr<std::FILE, decltype(close_file)>;34    File file(std::fopen(path, "wb"), close_file);35    example::require(static_cast<bool>(file), "cannot open log file");36    WriterSourceConfig source;37    source.block_size = 4096;38    source.pool_min = source.pool_max = 2; // Deliberately small, to expose pressure.39    source.backpressure.mode = drop ? BackpressureMode::Drop : BackpressureMode::Block;40    source.backpressure.wait_get_timeout = std::chrono::seconds{2};41    auto writer = make_writer<Writer<32>>(source);42    std::size_t written = 0;43    bool sink_error = false;44    std::thread::id callback_thread;45    const auto producer_thread = std::this_thread::get_id();46    auto rcvCallback = [&](std::int32_t thread_id, const std::byte* data,47                           std::size_t size) noexcept -> void {48        if (written == 0) callback_thread = std::this_thread::get_id();49        // Simulate a temporarily slow file system; not enabled in normal runs.50        if (slow_sink && written == 0) std::this_thread::sleep_for(std::chrono::milliseconds{20});51        LogRecord record{};52        if (thread_id != 0 || size != sizeof(record)) { sink_error = true; return; }53        std::memcpy(&record, data, sizeof(record)); // Callback data can be unaligned.54        if (!std::memchr(record.text.data(), '\0', record.text.size())) {55            sink_error = true; return;56        }57        if (std::fprintf(file.get(), "%llu request=%u %s\n",58                static_cast<unsigned long long>(record.sequence),59                static_cast<unsigned>(record.request_id), record.text.data()) < 0)60            sink_error = true; // noexcept callback records I/O errors; no implicit retry.61        ++written;62    };63    TwoBufQueueOptions options;64    options.output_block_size = 4096; // Can hold a complete source block.65    options.output_pool_min = options.output_pool_max = 2;66    options.output_queue_capacity = 2;67    options.read_config.is_scan_current = true; // flush can expose a partial block.68    ::TwoBufQueue::TwoBufQueue queue(rcvCallback, writer, options);6970    //---------------2. Producer writes-------------------------------------71    // Business work and record construction stay on the application thread.72    // File I/O runs on the independent callback thread, after layer-2 copying.73    std::vector<std::uint64_t> accepted;74    accepted.reserve(10000);75    std::size_t failed = 0;76    for (std::uint64_t sequence = 0; sequence != 10000; ++sequence) {77        LogRecord record{};78        record.sequence = sequence;79        record.request_id = static_cast<std::uint32_t>(sequence % 100);80        std::snprintf(record.text.data(), record.text.size(), "request completed");81        if (writer.write(record)) accepted.push_back(sequence);82        else ++failed; // Drop rejects this record; Block can also time out.83        if ((sequence + 1) % 1000 == 0) writer.flush();84    }85    writer.flush(); // Publish + notify; does not wait for callbacks or flush the file.8687    //---------------3. Finish----------------------------------------------88    writer.finish(); // Stop producing and hand off the final source block.89    queue.stop();    // Drain copying and callbacks before closing the file.90    const auto pressure = writer.backpressure_stats();91    const bool flush_ok = std::fflush(file.get()) == 0;92    const bool close_ok = std::fclose(file.release()) == 0;9394    //---------------4. Check the result------------------------------------95    example::require(queue.error_code() == ReadError::none, "queue did not drain cleanly");96    example::require(!sink_error && flush_ok && close_ok, "file I/O failed");97    example::require(written == accepted.size() && callback_thread != producer_thread,98                     "callback count or thread mismatch");99    std::ifstream saved(path, std::ios::binary);100    example::require(saved.is_open(), "cannot reopen saved log");101    std::string line;102    for (const auto sequence : accepted) {103        const auto expected = std::to_string(sequence) + " request=" +104            std::to_string(sequence % 100) + " request completed";105        example::require(static_cast<bool>(std::getline(saved, line)) && line == expected,106                         "saved record is missing, reordered or corrupted");107    }108    example::require(!std::getline(saved, line) && saved.eof(), "unexpected record or read error");109    std::printf("PASS: %zu saved, %zu rejected, pressure=%llu, blocked=%llu; file=%s\n",110        written, failed, static_cast<unsigned long long>(pressure.pressure_events),111        static_cast<unsigned long long>(pressure.blocked_events), path);112    example::require(drop || failed == 0, "blocking writer timed out; rejected records were reported");113} catch (const std::exception& error) {114    std::fprintf(stderr, "%s\n", error.what()); return 1;115}