C++输出流的管道安全透明缓冲实现方案问询

问题背景 #

我有两个程序，需要向cout及其他文件输出流写入大量类似A=123 B=456 ...data...的短行数据（每行约50字符）。由于涉及管道操作，必须保证写入的原子性——最初采用每行写入后立即刷新，但现在刷新操作已经成为两个程序的主要性能瓶颈。

我希望所有输出流（包括cout）能将通过<<接收的内容缓冲至接近PIPE_BUF的大小再刷新，且不拆分单行内容：如果新内容会导致缓冲区溢出，则先刷新当前缓冲区，再写入新内容。

曾尝试调用std::cout.rdbuf()->pubsetbuf(new char[PIPE_BUF], PIPE_BUF - 200)但未生效，需要实现类似示例代码的效果：只需一次性配置流，后续写入操作自动执行缓冲逻辑，无需每次手动判断。

解决方案：自定义streambuf实现 #

标准库默认缓冲无法满足「不拆分单行+按PIPE_BUF大小批量刷新」的需求，因此需要自定义streambuf类，重写缓冲逻辑，实现自动的行级原子性与批量刷新。

完整实现代码 #

#include <iostream>
#include <fstream>
#include <limits.h> // PIPE_BUF
#include <cstring>
#include <fmt/format.h>
using fmt::format;

// 自定义缓冲流，保证单行不拆分，缓冲接近PIPE_BUF再刷新
class AtomicLineBuf : public std::streambuf {
private:
    char* buffer_;
    const size_t buf_size_; // 缓冲区大小，设为PIPE_BUF - 预留空间
    size_t current_len_ = 0; // 当前缓冲区已用长度

    // 刷新缓冲区到目标流
    int flush_buffer() {
        if (current_len_ == 0) return 0;
        int result = sync();
        current_len_ = 0;
        return result;
    }

protected:
    // 当缓冲区不够时调用，处理新字符
    int overflow(int c) override {
        // 遇到换行符时，写入后立即刷新，保证整行原子输出
        if (c == '\n') {
            buffer_[current_len_++] = static_cast<char>(c);
            flush_buffer();
            return c;
        }

        // 新字符加入后将溢出缓冲区，先刷新
        if (current_len_ + 1 >= buf_size_) {
            flush_buffer();
        }

        // 写入新字符到缓冲区
        buffer_[current_len_++] = static_cast<char>(c);
        return c;
    }

    // 同步缓冲区到目标设备
    int sync() override {
        if (current_len_ > 0) {
            // 写入所有缓冲数据
            std::streamsize written = std::fwrite(buffer_, 1, current_len_, stdout);
            if (written != current_len_) {
                return -1;
            }
            std::fflush(stdout);
        }
        return 0;
    }

public:
    explicit AtomicLineBuf(size_t buf_size = PIPE_BUF - 200)
        : buf_size_(buf_size), buffer_(new char[buf_size_]) {}

    ~AtomicLineBuf() override {
        flush_buffer();
        delete[] buffer_;
    }

    // 手动触发刷新
    void force_flush() {
        flush_buffer();
    }
};

// 全局配置cout为自定义缓冲
void setup_atomic_cout() {
    static AtomicLineBuf custom_buf;
    std::cout.rdbuf(&custom_buf);
    // 关闭C++流与C stdio的同步，提升写入速度
    std::ios_base::sync_with_stdio(false);
    // 解除cout与cin的绑定，避免不必要的自动刷新
    std::cout.tie(nullptr);
}

// 为自定义文件流配置缓冲
void setup_atomic_stream(std::ofstream& stream) {
    static AtomicLineBuf custom_buf;
    stream.rdbuf(&custom_buf);
}

int main() {
    // 配置cout的自定义缓冲
    setup_atomic_cout();

    // 示例：为自定义文件流配置缓冲
    // std::ofstream log_file("output.log");
    // setup_atomic_stream(log_file);

    std::srand(42);
    for (int l = 0; l < 300'000; l++) {
        int A = rand() % 1000, B = rand() % 1000, L = rand() % 40, c = rand() % 26;
        std::string text = format("A={} B={} {}\n", A, B, std::string(L, 'a' + c));
        // 直接写入，缓冲逻辑自动处理
        std::cout << text;
    }

    // 程序结束前确保所有数据刷新
    std::cout.flush();
    return 0;
}

测试脚本（验证原子性） #

#!/bin/bash
rm -f test.log
# 启动8个进程同时写入同一文件，模拟管道/多进程写入场景
for _ in {0..7}; do ./writer >>test.log & done && wait
# 检查是否有行被拆分（所有行必须以"A="开头）
grep -q --invert-match "^A" test.log && echo "数据被拆分（MANGLED）" || echo "数据完整（OK）"

关键说明 #

1. 核心逻辑：
   • 缓冲区设为PIPE_BUF - 200，预留空间避免边界溢出问题
   • 遇到换行符立即刷新，保证整行原子写入
   • 新内容将导致缓冲区溢出时，先刷新再写入新内容
2. pubsetbuf失效原因：
   标准库std::cout默认的streambuf实现（如GCC的stdout_buf）可能忽略pubsetbuf调用，因此必须替换为自定义缓冲
3. 性能优化：
   • 关闭std::ios_base::sync_with_stdio(false)解除C++流与C stdio的同步，提升写入速度
   • std::cout.tie(nullptr)解除cout与cin的绑定，避免不必要的自动刷新

内容的提问来源于stack exchange，提问作者elbrunovsky