Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 16:46:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use Asio to launch processes in fair::mq::tools::execute

Browse Source
This commit is contained in:
Alexey Rybalchenko 2019-03-14 14:26:50 +01:00 committed by Dennis Klein
parent a8f1a4dfdb
commit 922f7e9a92
1 changed files with 73 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
fairmq/tools/Process.cxx
View File

@ -9,6 +9,8 @@
#include <fairmq/tools/Process.h> #include <fairmq/tools/Process.h>
#include <boost/process.hpp> #include <boost/process.hpp>
#include <boost/asio.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <iostream> #include <iostream>
#include <sstream> #include <sstream>
@ -17,6 +19,7 @@
using namespace std; using namespace std;
namespace bp = boost::process; namespace bp = boost::process;
namespace ba = boost::asio;
namespace fair namespace fair
{ {
@ -42,48 +45,92 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
stringstream printCmd; stringstream printCmd;
printCmd << prefix << " " << cmd << "\n"; printCmd << prefix << " " << cmd << "\n";
cout << printCmd.str() << flush; cout << printCmd.str() << flush;
out << prefix << cmd << endl; out << prefix << cmd << endl;
// Execute command and capture stdout, add prefix line by line ba::io_service ios;
bp::ipstream c_stdout;
bp::opstream c_stdin;
bp::child c(cmd, bp::std_out > c_stdout, bp::std_in < c_stdin);
while (c.valid() && !c.running()) { // containers for std_in
; ba::const_buffer inputBuffer(ba::buffer(input));
} bp::async_pipe inputPipe(ios);
// containers for std_out
ba::streambuf outputBuffer;
bp::async_pipe outputPipe(ios);
// containers for std_err
ba::streambuf errorBuffer;
bp::async_pipe errorPipe(ios);
if (!c.valid()) { const string delimiter = "\n";
throw runtime_error("Can't execute the given process."); ba::deadline_timer timer(ios, boost::posix_time::milliseconds(100));
}
// Optionally, write to stdin of the child // child process
bp::child c(cmd, bp::std_out > outputPipe, bp::std_err > errorPipe, bp::std_in < inputPipe, ios);
// handle std_in with a delay
if (input != "") { if (input != "") {
this_thread::sleep_for(chrono::milliseconds(100)); timer.async_wait([&](const boost::system::error_code& ec1) {
c_stdin << input; if (!ec1) {
c_stdin.flush(); ba::async_write(inputPipe, inputBuffer, [&](const boost::system::error_code& ec2, size_t /* n */) {
if (!ec2) {
inputPipe.async_close();
} else {
out << prefix << "error in boost::asio::async_write: " << ec2.value() << endl;
}
});
} else {
out << prefix << "error in boost::asio::deadline_timer.async_wait: " << ec1.value() << endl;
}
});
} }
string line; // handle std_out line by line
while (c.running() && getline(c_stdout, line)) { function<void(const boost::system::error_code&, size_t)> onStdOut = [&](const boost::system::error_code& ec, size_t /* n */) {
// print full line thread-safe if (!ec) {
stringstream printLine; istream is(&outputBuffer);
printLine << prefix << line << "\n"; string line;
cout << printLine.str() << flush; getline(is, line);
out << prefix << line << "\n"; stringstream printLine;
} printLine << prefix << line << "\n";
cout << printLine.str() << flush;
out << prefix << line << endl;
ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
} else {
outputPipe.async_close();
}
};
ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
// handle std_err line by line
function<void(const boost::system::error_code&, size_t)> onStdErr = [&](const boost::system::error_code& ec, size_t /* n */) {
if (!ec) {
istream is(&errorBuffer);
string line;
getline(is, line);
stringstream printLine;
printLine << prefix << line << "\n";
cerr << printLine.str() << flush;
out << prefix << "error: " << line << endl;
ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
} else {
errorPipe.async_close();
}
};
ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
ios.run();
c.wait(); c.wait();
// Capture exit code
result.exit_code = c.exit_code(); result.exit_code = c.exit_code();
out << prefix << " Exit code: " << result.exit_code << endl; stringstream exitCode;
exitCode << prefix << " Exit code: " << result.exit_code << "\n";
cout << exitCode.str() << flush;
out << prefix << " Exit code: " << result.exit_code << endl;
result.console_out = out.str(); result.console_out = out.str();
// Return result
return result; return result;
} }