Boost async_pipe 不显示所有子进程输出

Question

我遇到了障碍。下面的代码有问题，但这只是一个演示；我想先把高层逻辑弄对。

两个启动应用程序在到达“就绪”状态之前输出了很多启动信息。在这种状态下，程序 A 已准备好通过标准输入进行用户输入。程序 B 只是通过网络连接进行监听——摄取和记录数据。

理想情况下，使用这个示例程序，我应该能够“实时”看到程序 B 的输出。但是在每次循环迭代中，什么都没有发生；我不确定它是否通过管道接收输入。

我之前使用bp::opstream 写入孩子的--Program A--stdin。我知道程序 A 是否通过其 async_pipe 接受了某些命令，Progam B show 还会显示一些日志记录信息（例如“trip”）。这些是窗口控制台应用程序，我使用 Boost C++ 作为子进程与它们进行交互。

有人知道发生了什么吗？

std::size_t read_loop(bp::async_pipe& p, mutable_buffer buf, boost::system::error_code &err)
{
    return p.read_some(buf, err);


}



void read_loop_async(bp::async_pipe& p, mutable_buffer buf, std::error_code &err) {
    p.async_read_some(buf, [&p, buf, &err](std::error_code ec, size_t n) {
        std::cout << "Received " << n << " bytes (" << ec.message() << "): '";
        std::cout.write(boost::asio::buffer_cast<char const*>(buf), n) << std::endl;
        err = ec;

        if (!ec)
            read_loop_async(p, buf, err);

    });
}


void write_pipe(bp::async_pipe&p, mutable_buffer buf)
{
    ba::async_write(p, buf, [](boost::system::error_code ec, std::size_t sz)
    {
        std::cout << "Size Written " << sz << " Ec: " << ec << " " << ec.message() << '\n';
    });

}


int main()
{

    bp::opstream sendToChild;
    string wd = "<---path-to-working-dir----->";
    ba::io_service ios;
    string bin = "<path-to-bin-and-name>";


    bp::async_pipe input_pipe(ios);
    bp::async_pipe output_pipe(ios);

    bp::child c(bin, "arg1", "arg2", "arg3", bp::std_out > output_pipe,
        bp::std_in < input_pipe, ios, bp::start_dir(wd.c_str()));

    size_t size = 8192;
    string input;

    vector <char> buffer(size);
    boost::system::error_code ec;

    std::error_code err;

    ios.run();
    while (1)
    {
        //show read whatever is available from the childs output_pipe
        read_loop_async(output_pipe, bp::buffer(buffer), err);



        cout << "\nBoot-> ";
        cin >> input;
        if (input == "1")
        {
            cout << "   send input to child: ";
            cin >> input;
            //send commands to the child, Program A
            //originally
            //sendToChild << input<< endl;
            write_pipe(input_pipe, bp::buffer(input));
        }
        if (input == "quit")
        {
            //sendToChild << input << endl;
            read_loop_async(output_pipe, bp::buffer(buffer), err);
            break;
        }

        ios.poll(ec);
        ios.restart();

    }

    c.join();
    cout << "done...";
    cin >> input;
}

这是我关注的链接： How to retrieve program output as soon as it printed?

Answer 1

嗯。有很多东西要解压。首先：

ios.run();

一直运行到子进程完成。如果子进程需要发送的输出超出缓冲区的容量，则很可能会出现死锁，因为您在执行ios.run() 之前没有消耗任何输出。

根据定义，下一个poll() 不执行任何操作，因为您没有先调用restart。幸运的是，您忽略了错误代码，接下来会发生 restart。

然后，您会遇到下一个问题，因为循环的下一次迭代以 another read_loop_async(output_pipe, bp::buffer(buffer), err); 开始，这意味着您有重叠的读取操作，这通常是被禁止的 (Undefined Behaviour)，但会运行无论如何都在这里进入 UB，因为您使用的是相同的缓冲区。

这本身就足以解释“丢失的数据”，因为，是的，你在同一个位置进行多次读取，所以一个会破坏另一个。也就是说，如果你可以推理它，因为你不能推理 UB。

奇怪的是，现在我的眼睛甚至发现了第三次调用 read_loop_async。这没有道理。顾名思义，read_loop_async 已经是一个循环：它在完成时调用自己：

    if (!ec)
        read_loop_async(p, buf, err);

因此，预计只有 1 次调用。似乎您没有理解async_* 启动函数总是立即返回（因为操作完成异步）。这也体现在您分配的事实中：

    err = ec;

其中err 是启动函数的引用参数。它不是那样工作的。该错误仅在完成时可用。由于您似乎并没有在读取循环之外使用它，所以我会放弃它。

然后是

        sendToChild << input << std::endl;

这绝对没有任何作用，因为 sendToChild 只是在字面上被声明，从未在其他地方使用过。

write_pipe 再次尝试使用async_ 启动，但它不能，因为它正在同步输入循环中使用。只是不要在那里使用异步。正如所写，它是 UB 的另一个来源，因为 buf 参数将指向一个在主函数中发生变异的 std::string 变量。所以，简化：

void write_pipe(bp::async_pipe& p, const_buffer buf) {
    error_code ec;
    auto       sz = write(p, buf, ec);
    std::cout << "Size Written " << sz << " Ec: " << ec << " " << ec.message() << '\n';
}

[注意它如何正确地将buf 标记为const_buffer。]

现在，可能修复 sendToChild 使用的问题

• 同时关闭管道（向子节点发送 EOF 信号）
• 打破输入循环

    if (input == "quit") {
        write_pipe(input_pipe, bp::buffer(input + "\n"));
        input_pipe.close();
        break;
    }

我将用 poll() 替换 ios.restart() 的东西 - 因为无论如何我们都没有太早使用 run()。

除上述之外，我将operator>> 替换为std::getline 调用，因为您很可能希望用户输入用Enter 键而不是空格分隔。我还添加了"\n"，就像您在sendToChild 行中所做的那样，因为它有助于使用使用行缓冲输入的简单测试子项进行演示。

现在，我们将使用它作为测试孩子：

bp::child c(bin, "-c",
            "time while read line; do echo \"$line\" | rev | xxd; done", //
            bp::std_out > output_pipe,
            bp::std_in < input_pipe, //
            ios,                     //
            bp::start_dir(wd));

这意味着我们的输入以反向和十六进制转储的形式回显，并在最后显示时间摘要。

有些固定的列表

Live On Coliru

#include <boost/asio.hpp>
#include <boost/process.hpp>
#include <boost/process/async.hpp>
#include <iostream>
namespace bp = boost::process;
using boost::asio::const_buffer;
using boost::asio::mutable_buffer;
using boost::system::error_code;

void read_loop_async(bp::async_pipe& p, mutable_buffer buf) {
    p.async_read_some(buf, [&p, buf](std::error_code ec, size_t n) {
        std::cout << "Received " << n << " bytes (" << ec.message() << "): '";
        std::cout.write(boost::asio::buffer_cast<char const*>(buf), n) << std::endl;

        if (!ec)
            read_loop_async(p, buf);
    });
}

void write_pipe(bp::async_pipe& p, const_buffer buf) {
    error_code ec;
    auto       sz = write(p, buf, ec);
    std::cout << "Size Written " << sz << " Ec: " << ec << " " << ec.message() << '\n';
}

int main() {
    std::string wd = "/home/sehe/Projects/stackoverflow";
    boost::asio::io_service ios;
    std::string bin = "/bin/bash";


    bp::async_pipe input_pipe(ios);
    bp::async_pipe output_pipe(ios);

    bp::child c(bin, "-c",
                "while read line; do echo \"$line\" | rev | xxd; done", //
                bp::std_out > output_pipe,
                bp::std_in < input_pipe, //
                ios,                     //
                bp::start_dir(wd));


    // Single invocation!
    std::vector<char> buffer(8192);
    read_loop_async(output_pipe, bp::buffer(buffer));

    std::cout << "\nBoot-> ";
    for (std::string input; getline(std::cin, input);
         std::cout << "\nBoot-> ") {
        if (input == "1") {
            std::cout << "   send input to child: ";
            if (getline(std::cin, input)) {
                write_pipe(input_pipe, bp::buffer(input + "\n"));
            }
        }
        if (input == "quit") {
            write_pipe(input_pipe, bp::buffer(input + "\n"));
            input_pipe.close();
            break;
        }

        ios.poll();
    }

    ios.run(); // effectively like `c.wait();` but async

    std::cout << "done...";
    // ignore until line end
    std::cin.ignore(std::numeric_limits<std::streamsize>::max(), '\n');
}

经过测试

g++ -std=c++20 -O2 -Wall -pedantic -pthread main.cpp -lboost_{system,filesystem} && ./a.out <<HERE
1
Hello world
Ignored
1
Bye world
quit
HERE

打印

Boot->    send input to child: Size Written 12 Ec: system:0 Success

Boot-> 
Boot->    send input to child: Size Written 10 Ec: system:0 Success

Boot-> Size Written 5 Ec: system:0 Success
Received 64 bytes (Success): '00000000: 646c 726f 7720 6f6c 6c65 480a            dlrow olleH.

Received 62 bytes (Success): '00000000: 646c 726f 7720 6579 420a                 dlrow eyB.

Received 57 bytes (Success): '00000000: 7469 7571 0a                             tiuq.

Received 0 bytes (End of file): '
done...

在我的系统上以交互方式更容易理解：

Answer 2

我最终编写了自己的课程。它是fork 和execv 的包装器。我尝试了各种解决方案，但并非一切都运作良好。有了这个类，我能够分拆“子进程”和“异步”读取它们的输出。我使用管道与子进程进行通信。我有一个工作线程来监视管道的读取端——使用select 并弹出 sendQueue 以将内容发送给孩子。我没有看到任何“潜在的”垮台，而不是不保护 - 使用我的使用 - 队列。与队列的推送/弹出没有真正的重叠。

我可以更好地管理两个队列中的所有字符串数据——效率更高，使用 std::move。

#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <signal.h>
#include <unistd.h>
#include <strings.h>
#include <sys/wait.h>
#include <sys/prctl.h>
#include <sys/types.h>
#include <sys/select.h>

#include <atomic>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <vector>
#include <iostream>
#include <algorithm>
#include <condition_variable>


class Process
{
public:
    Process(std::string bin, std::string pwd, std::vector<std::string> args);
    ~Process();

    void Send(std::string str, int whoami = 0); 
    std::string Receive();

    void start();
    void stop();


private:
    pid_t _pid = 0;  
    int pipeA[2] = {0};
    int pipeB[2] = {0};


    Process &operator = (Process const&);
    void taskLoop();


    std::string _pwd; 
    std::string _bin;
    std::vector<std::string> _args;
    

    std::thread _thread;
    
    std::atomic<bool> _stop = false;            /* Process obj stopped, not child */
    std::atomic<bool> _oneShotDone = false;     /* Prevents creating multiple children */ 
    std::atomic<bool> _childTerminated = false; /* Child terminated before parent */




    std::atomic<bool> _threadRun = false;
    std::queue<std::string> sendQueue;
    std::queue<std::string> recvQueue;
    

    /* Exiting mutex and cv, tells destructor when its ok*/
    std::mutex exitMutex;
    std::condition_variable cvExit;
};


    //TODO: check for empty args
Process::Process(std::string bin, std::string pwd, std::vector<std::string> args)
    {
    _bin = pwd+bin; 
    _pwd = pwd; 
    _args.push_back(_bin);

    for(auto &s: args) 
        _args.push_back(s);

    }


Process::~Process()
    {
    _threadRun = false;

    if(!_childTerminated) /*Kill child, child process is still alive */
        { 
        kill(_pid, SIGKILL);
        }

    
    close(pipeA[1]);
    close(pipeB[0]);
    }


static char* convert(const std::string & s)
    {
   char *pc = new char[s.size()+1];
   strcpy(pc, s.c_str());
   return pc; 
    }


void Process::Start()
    {
    if(_oneShotDone)
        return; //todo return error


    std::vector<char*> args;
    std::transform(_args.begin(), _args.end(), std::back_inserter(args), convert);
    args.push_back(0);

    signal(SIGPIPE, SIG_IGN);

    pipe(pipeA);
    pipe(pipeB);


    _pid = fork();
    if(_pid == ERROR)
        {
        perror("Process::start(): fork failed...");
        exit(EXIT_FAILURE);
        }
    else if(_pid == 0)
        {
        dup2(pipeA[0], 0);
        dup2(pipeB[1], 1);

        close(pipeA[1]);
        close(pipeB[0]);
        
        close(pipeA[0]);
        close(pipeB[1]);

        chdir(_pwd.c_str()); 
        execv(*args.data(), args.data());
        perror("execv");
        std::cout << "Process::child exited failure\n";
        exit(EXIT_FAILURE);
        }
    else
        {

        close(pipeA[0]);
        close(pipeB[1]);

        _oneShotDone = true;

        _threadRun = true;
        _thread = std::thread(&Process::read, this);
        _thread.detach();
        }
    } 



void Process::Send(std::string str, int whoami)
    {
    if(_stop)
        throw std::runtime_error("Process has stopped, can't communicate"); 

    if(!_oneShotDone) 
        throw std::runtime_error("Haven't started child process, can't communicate"); 
    
    sendQueue.push(str);
    }


std::string Process::Receive()
    {
    if(_stop)
        throw std::runtime_error("Process has stopped, can't communicate"); 

    if(!_oneShotDone) 
        throw std::runtime_error("Haven't started child process, can't communicate"); 
    
    if( recvQueue.empty())
        {
        return "";
        }
   
    std::string const& refString = RxQueue.front(); 
    std::string s(refString);
    recvQueue.pop();
    
    return s;
    }



void Process::read()
    {

    size_t ec = 0;
    char tbuffer[1024] = {0};
    std::string newline("\n"); 
        
    int& readFD = pipeB[0]; 
    int& writeFD = pipeA[1]; 

    int sel = 0;
    fd_set readFDS, writeFDS;

    while(_threadRun)
        {   
        FD_ZERO(&readFDS); 
        FD_ZERO(&writeFDS); 
   
        FD_SET(readFD, &readFDS);
        FD_SET(writeFD, &writeFDS);
        
        sel = select(std::max(writeFD, readFD)+1, &readFDS, &writeFDS, 0, 0);
    
        if( sel == ERROR)
            {perror("select"); continue;}
        if( sel)
            { 
            if(FD_ISSET(readFD, &readFDS))
                { /* READ first */
                
                bzero(tbuffer, 1024);  
                ec = read(readFD, tbuffer, 1024);
                if(ec == ERROR)
                    {  
                    if(errno==EPIPE)
                        {
                        std::cerr << "Child process closed pipe end"<<std::endl;
                        _childTerminated = true;
                        _stop = true;
                        _threadRun = false;
                        break;
                        }                
                    }

                tbuffer[ec] = '\0';
                
                std::string s(tbuffer);
                recvQueue.push(s);
                }

            if(FD_ISSET(writeFD, &writeFDS))
                { /* WRITE */
    
                if(sendQueue.empty())
                    {continue;}

                auto& strRef = sendQueue.front();
                ec  = write(writeFD, strRef.c_str(), strRef.length());
                sendQueue.pop();
                if(ec == ERROR)
                    {
                    if(errno=EPIPE)
                        {
                        std::cerr << "Child process closed pipe end"<<std::endl;
                        _childTerminated = true;
                        _stop = true;
                        _threadRun = false;
                        break;
                        }
                    }
                } /* WRITE end */
            } /* select end */
        }/* end loop */
    
    if(!_childTerminated) /* force child process kill */
        kill(_pid, SIGKILL); 
    }



void Process::Stop()
    {
    _stop = true;
    _threadRun = false;
    }