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

Fix broken pipe errors in tools::execute

Browse Source
This commit is contained in:
Alexey Rybalchenko 2019-03-15 13:25:56 +01:00
parent 922f7e9a92
commit cdc1ba084c
1 changed files with 26 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
fairmq/tools/Process.cxx
View File

@ -20,6 +20,7 @@
using namespace std; using namespace std;
namespace bp = boost::process; namespace bp = boost::process;
namespace ba = boost::asio; namespace ba = boost::asio;
namespace bs = boost::system;
namespace fair namespace fair
{ {
@ -63,27 +64,29 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
ba::deadline_timer timer(ios, boost::posix_time::milliseconds(100)); ba::deadline_timer timer(ios, boost::posix_time::milliseconds(100));
// child process // child process
bp::child c(cmd, bp::std_out > outputPipe, bp::std_err > errorPipe, bp::std_in < inputPipe, ios); bp::child c(cmd, bp::std_out > outputPipe, bp::std_err > errorPipe, bp::std_in < inputPipe);
// handle std_in with a delay // handle std_in with a delay
if (input != "") { if (input != "") {
timer.async_wait([&](const boost::system::error_code& ec1) { timer.async_wait([&](const bs::error_code& ec1) {
if (!ec1) { if (!ec1) {
ba::async_write(inputPipe, inputBuffer, [&](const boost::system::error_code& ec2, size_t /* n */) { ba::async_write(inputPipe, inputBuffer, [&](const bs::error_code& ec2, size_t /* n */) {
if (!ec2) { if (!ec2) {
inputPipe.async_close(); // inputPipe.async_close();
} else { } else {
out << prefix << "error in boost::asio::async_write: " << ec2.value() << endl; cout << prefix << "error in boost::asio::async_write: " << ec2.message() << endl;
out << prefix << "error in boost::asio::async_write: " << ec2.message() << endl;
} }
}); });
} else { } else {
out << prefix << "error in boost::asio::deadline_timer.async_wait: " << ec1.value() << endl; cout << prefix << "error in boost::asio::deadline_timer.async_wait: " << ec1.message() << endl;
out << prefix << "error in boost::asio::deadline_timer.async_wait: " << ec1.message() << endl;
} }
}); });
} }
// handle std_out line by line // handle std_out line by line
function<void(const boost::system::error_code&, size_t)> onStdOut = [&](const boost::system::error_code& ec, size_t /* n */) { function<void(const bs::error_code&, size_t)> onStdOut = [&](const bs::error_code& ec, size_t /* n */) {
if (!ec) { if (!ec) {
istream is(&outputBuffer); istream is(&outputBuffer);
string line; string line;
@ -96,13 +99,18 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut); ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
} else { } else {
outputPipe.async_close(); if (ec == ba::error::eof) {
// outputPipe.async_close();
} else {
cout << prefix << ec.message() << endl;
out << prefix << ec.message() << endl;
}
} }
}; };
ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut); ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);
// handle std_err line by line // 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 */) { function<void(const bs::error_code&, size_t)> onStdErr = [&](const bs::error_code& ec, size_t /* n */) {
if (!ec) { if (!ec) {
istream is(&errorBuffer); istream is(&errorBuffer);
string line; string line;
@ -115,7 +123,12 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr); ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
} else { } else {
errorPipe.async_close(); if (ec == ba::error::eof) {
// errorPipe.async_close();
} else {
cout << prefix << ec.message() << endl;
out << prefix << ec.message() << endl;
}
} }
}; };
ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr); ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);
@ -124,11 +137,11 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
c.wait(); c.wait();
result.exit_code = c.exit_code(); result.exit_code = c.exit_code();
stringstream exitCode;
exitCode << prefix << " Exit code: " << result.exit_code << "\n"; stringstream exitCode;
exitCode << prefix << " fair::mq::tools::execute: exit code: " << result.exit_code << "\n";
cout << exitCode.str() << flush; cout << exitCode.str() << flush;
out << prefix << " Exit code: " << result.exit_code << endl; out << prefix << " fair::mq::tools::execute: exit code: " << result.exit_code << endl;
result.console_out = out.str(); result.console_out = out.str();
return result; return result;