Source · fairmq/tools/
Process.cxxsource
Described in Helpers in fair::mq::tools.
1/********************************************************************************2 * Copyright (C) 2017-2025 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#include <fairmq/tools/Process.h>10#include <fairmq/tools/Strings.h>11 12#include <boost/asio.hpp>13#ifdef FAIRMQ_BOOST_PROCESS_V1_HEADER14#include <boost/process/v1.hpp>15#else16#include <boost/process.hpp>17#endif18#include <chrono>19#include <csignal> // kill, signals20#include <iostream>21#include <mutex>22#include <sstream>23#include <stdexcept>24#include <thread>25#include <utility>26 27using namespace std;28#ifdef FAIRMQ_BOOST_PROCESS_V1_HEADER29namespace bp = boost::process::v1;30#else31namespace bp = boost::process;32#endif33namespace ba = boost::asio;34namespace bs = boost::system;35 36class LinePrinter37{38 public:39 LinePrinter(stringstream& out, string prefix)40 : fOut(out)41 , fPrefix(std::move(prefix))42 {}43 44 // prints line with prefix on both cout and output stream45 void Print(const string& line)46 {47 // Serialize: the standard allows concurrent insertion on std::cout,48 // but libstdc++ maintains the formatted-output state (e.g.49 // ios_base::width) with plain reads and writes that constitute data50 // races; the lock also keeps whole lines from interleaving.51 static mutex sCoutMutex;52 {53 lock_guard<mutex> lock(sCoutMutex);54 cout << fair::mq::tools::ToString(fPrefix, line, "\n") << flush;55 }56 fOut << fPrefix << line << endl;57 }58 59 private:60 stringstream& fOut;61 const string fPrefix;62};63 64namespace fair::mq::tools65{66 67execute_result execute(const string& cmd, const string& prefix, const string& input, int sig)68{69 return execute(cmd, prefix, input, sig, chrono::milliseconds::zero());70}71 72execute_result execute(const string& cmd,73 const string& prefix,74 const string& input,75 int sig,76 chrono::milliseconds timeout)77{78 execute_result result;79 stringstream out;80 81 LinePrinter p(out, prefix);82 83 p.Print(cmd);84 85 ba::io_context ioc;86 87 // containers for std_in88 ba::const_buffer inputBuffer(ba::buffer(input));89 bp::async_pipe inputPipe(ioc);90 // containers for std_out91 ba::streambuf outputBuffer;92 bp::async_pipe outputPipe(ioc);93 // containers for std_err94 ba::streambuf errorBuffer;95 bp::async_pipe errorPipe(ioc);96 97 const string delimiter = "\n";98 ba::steady_timer inputTimer(ioc);99 inputTimer.expires_after(std::chrono::milliseconds(1000)); // NOLINT100 ba::steady_timer signalTimer(ioc);101 signalTimer.expires_after(std::chrono::milliseconds(2000)); // NOLINT102 103 // child process104 bp::child c(cmd, bp::std_out > outputPipe, bp::std_err > errorPipe, bp::std_in < inputPipe);105 int pid = c.id();106 p.Print(ToString("fair::mq::tools::execute: pid: ", pid));107 108 // kill the child when the deadline expires; cancel the deadline when both109 // output pipes are closed (the child has exited)110 ba::steady_timer deadlineTimer(ioc);111 int openPipes = 2;112 auto onPipeClosed = [&]() {113 if (--openPipes == 0) {114 deadlineTimer.cancel();115 }116 };117 if (timeout > chrono::milliseconds::zero()) {118 deadlineTimer.expires_after(timeout);119 deadlineTimer.async_wait([&](const bs::error_code& ec) {120 if (!ec) {121 p.Print(ToString("fair::mq::tools::execute: timeout after ", timeout.count(), " ms, killing pid ", pid));122 kill(pid, SIGKILL);123 }124 });125 }126 127 // handle std_in with a delay128 if (!input.empty()) {129 inputTimer.async_wait([&](const bs::error_code& ec1) {130 if (!ec1) {131 ba::async_write(inputPipe, inputBuffer, [&](const bs::error_code& ec2, size_t /* n */) {132 if (!ec2) {133 // inputPipe.async_close();134 } else {135 p.Print(ToString("error in boost::asio::async_write: ", ec2.message()));136 }137 });138 } else {139 p.Print(ToString("error in boost::asio::deadline_timer.async_wait: ", ec1.message()));140 }141 });142 }143 144 if (sig != -1) {145 signalTimer.async_wait([&](const bs::error_code& ec1) {146 if (!ec1) {147 kill(pid, sig);148 } else {149 p.Print(ToString("error in boost::asio::deadline_timer.async_wait: ", ec1.message()));150 }151 });152 }153 154 // handle std_out line by line155 function<void(const bs::error_code&, size_t)> onStdOut = [&](const bs::error_code& ec, size_t /* n */) {156 if (!ec) {157 istream is(&outputBuffer);158 string line;159 getline(is, line);160 161 p.Print(line);162 163 ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);164 } else {165 if (ec != ba::error::eof) {166 p.Print(ec.message());167 }168 onPipeClosed();169 }170 };171 ba::async_read_until(outputPipe, outputBuffer, delimiter, onStdOut);172 173 // handle std_err line by line174 function<void(const bs::error_code&, size_t)> onStdErr = [&](const bs::error_code& ec, size_t /* n */) {175 if (!ec) {176 istream is(&errorBuffer);177 string line;178 getline(is, line);179 180 p.Print(ToString("error: ", line));181 182 ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);183 } else {184 if (ec != ba::error::eof) {185 p.Print(ec.message());186 }187 onPipeClosed();188 }189 };190 ba::async_read_until(errorPipe, errorBuffer, delimiter, onStdErr);191 192 ioc.run();193 c.wait();194 195 result.exit_code = c.exit_code();196 197 p.Print(ToString("fair::mq::tools::execute: exit code: ", result.exit_code));198 result.console_out = out.str();199 200 return result;201}202 203} // namespace fair::mq::tools