使用 c++ stdlib 从管道非阻塞读取

Question

在 2021 年，有没有办法使用 C++ 标准库的高级工具从管道进行非阻塞读取？ IE。 C++17（或者可能是 C++20）？在 Linux 上进行开发，但在一定程度上可移植到其他 *NIX 是可取的。 Boost 不是我手头问题的选择。

我很了解 POSIX 功能，即 poll() 和 open() 和 O_NONBLOCK。沿着这条路线，我将实现自己的缓冲并拆分成行，而且数据是二进制的（实际上这并不是看起来那么大的问题，只要它是 unicode-utf8 并且基本上只是通过一些报告，其他现有软件将正确呈现字符，例如 Markdown 到 HTML）。

我只是在问自己，我真的必须（再次）实施吗？还是有一些现成的解决方案可用，我只是忽略了？据我了解，std::basic_istream<..> 不太合适，因为它会尝试填充底层缓冲区，并且当管道中没有足够的数据时会阻塞。

对于背景：我正在通过两个管道从子进程中检索 STDIN 和 STDERR。我必须将这两个流逐行交错。这发生在专用的读取器线程中 - 但是，即使子进程进入活锁，该读取器线程也不能卡住。

Answer 1

C++ 库没有“管道”或其他操作系统特定资源的概念。

对于背景：我正在从子进程中检索 STDIN 和 STDERR 通过两个管道。我必须通过 线基。

我会为这种情况做些什么：子类std::streambuf 并覆盖underflow()。被覆盖的underflow() 分别实现从两个管道的非阻塞读取，为每个管道保留单独的缓冲区。并且std::streambuf 的缓冲区会被一个已完成的行填满，无论哪个管道成功地读取了已完成的行，首先，只要有一个。

有一个重载的std::istream 构造函数，它接受一个指向自定义std::streambuf 参数的指针。

您最终会得到一个看起来很普通的std::istream，您可以通过std::getline 得到任何一个能够生成完整行的管道，最终得到逐行交错输入。

Answer 2

产生两个std::thread，每个都从不同的管道读取。将std::getline 读取到单独的std::string。读完一行后，将std::string 放到受std::mutex 保护的std::vector<std::string>，然后通知condition_variable。在主线程中，您可以在condition_variable 上等待新事件，然后抓取mutex 并将std::vector<std::string> 中的所有行刷新到输出。

像这样：

#include <string>
#include <thread>
#include <iostream>
#include <mutex>
#include <vector>
#include <condition_variable>
#include <fstream>
#include <assert.h>

std::mutex g_lines_mutex;
std::condition_variable g_lines_notify;
std::vector<std::string> g_lines;

void outputter() {
    while (1) {
        std::unique_lock<std::mutex> guard(g_lines_mutex);
        if (g_lines.empty()) {
             g_lines_notify.wait(guard);
        }
        for (auto&& i : g_lines) {
            std::cout << "Read line: " << i << "\n";
        }
        g_lines.clear();
    }
}

void interleaver(const char *arg) {
    std::ifstream f(arg);
    std::string line;
    while (std::getline(f, line)) {
        {
            std::lock_guard<std::mutex> guard(g_lines_mutex);
            g_lines.emplace_back(std::move(line));
        }
        g_lines_notify.notify_one();
    }
}

int main(int argc, char *argv[]) {
    assert(argc == 3);
    std::array<std::thread, 3> t = {
        std::thread{ outputter },
        std::thread{ interleaver, argv[1] },
        std::thread{ interleaver, argv[2] },
    };
    for (auto&& i : t) {
        i.join();
    }
}

然后这样的程序编译并运行：

$ mkfifo fifo1 fifo2; 
$ ( exec 3> fifo1 ; exec 4> fifo2; while sleep 1; do echo 1 $(date) >&3 ; echo 2 $(date) >&4; done; ) &
$ g++ -pthread ./1.cpp && ./a.out fifo1 fifo2
Read line: 1 Sun, 01 Aug 2021 17:41:25 +0200
Read line: 2 Sun, 01 Aug 2021 17:41:25 +0200
Read line: 1 Sun, 01 Aug 2021 17:41:26 +0200
Read line: 2 Sun, 01 Aug 2021 17:41:26 +0200
Read line: 1 Sun, 01 Aug 2021 17:41:27 +0200
Read line: 2 Sun, 01 Aug 2021 17:41:27 +0200