asyncProcess.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_libdaq
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief daq::AsyncProcess class declaration
 */
#include <daq/process/asyncProcess.hpp>
 
#include <ostream>
 
#include <boost/asio/read_until.hpp>
#include <boost/process/args.hpp>
#include <log4cplus/loggingmacros.h>
 
#include <daq/log4cplus.hpp>
 
namespace daq {
 
namespace bp = boost::process;
 
std::ostream& operator<<(std::ostream& os, AsyncProcessIf const& proc) {
    if (auto pid = proc.GetPid(); pid) {
        os << '[' << *pid << ']';
    } else {
        os << "[ <not running> ]";
    }
 
    for (auto const& arg : proc.GetArguments()) {
        os << ' ' << arg;
    }
    return os;
}
 
AsyncProcess::AsyncProcess(boost::asio::io_context& ctx, std::vector<std::string> args)
    : m_io_ctx(ctx), m_args(std::move(args)), m_stdout{ctx}, m_stderr{ctx} {
    if (m_args.size() < 1) {
        throw std::invalid_argument("No arguments provided");
    }
    m_args[0] = bp::search_path(m_args[0]).native();
}
 
AsyncProcess::~AsyncProcess() noexcept {
}
 
boost::future<int> AsyncProcess::Initiate() {
    assert(!m_proc.valid());
 
    // on_exit handler has the critical role of triggering the clean shutdown of in-progress
    // async operations.
    //
    // There may be scheduled, but not yet executed, completion handlers for PIPE reads so we cannot
    // simply cancel or close PIPES without risk loosing data.
    //
    // To ensure all output is read we close PIPE write end so that when read is exhausted EOF will
    // be emitted.
    //
    // Observed: on_exit handler not invoked.
    m_proc = bp::child(
        bp::args = m_args,
        bp::std_out > m_stdout.pipe,
        bp::std_err > m_stderr.pipe,
        m_io_ctx,  // io_context is required for on_exit handler to function.
        bp::on_exit = [this](int exit, const std::error_code& ec_in) {
            // After process is terminated PIPEs will eventually reach EOF which triggers
            // the completion AsyncProcess future. For now we signal the completion of the process
            // by setting value.
            m_result = {exit, ec_in};
            CheckCompleted();
        });
    m_pid = m_proc.id();
    AsyncReadStream(m_stdout);
    AsyncReadStream(m_stderr);
    return m_promise.get_future();
}
 
std::error_code AsyncProcess::Abort() noexcept {
    if (!IsRunning()) {
        return std::make_error_code(std::errc::no_such_process);
    }
    // Terminate process hard, this triggers even handlers and normal unwind.
    std::error_code ec;
    m_proc.terminate(ec);
    return ec;
}
 
std::optional<pid_t> AsyncProcess::GetPid() const noexcept {
    if (m_proc.valid()) {
        return m_pid;
    }
    return std::nullopt;
}
 
std::error_code AsyncProcess::Signal(int sig) noexcept {
    if (!IsRunning()) {
        return std::make_error_code(std::errc::no_such_process);
    }
 
    pid_t id = m_proc.id();
    int err = kill(id, sig);
    return std::make_error_code(static_cast<std::errc>(err));
}
 
bool AsyncProcess::IsRunning() const noexcept {
    // warning: It is not safe to invoke child::running() due to bug in boost so we infer the
    // running state based on whether exit_handler has been invoked or not. This is not 100%
    // accurate but will work for most use-cases.
    return m_proc.valid() && !m_result.has_value();
}
 
void AsyncProcess::AsyncReadStream(AsyncProcess::Pipe& pipe) {
    // Process pipe content 'line-buffered'
    boost::asio::async_read_until(pipe.pipe,
                                  pipe.buffer,
                                  '\n',
                                  [this, &pipe](const boost::system::error_code& ec, std::size_t) {
                                      if (ec) {
                                          // Expected error is 'EOF', but we don't make a
                                          // distinction at this point.
                                          //
                                          // There might still be data in buffer from previous reads
                                          // that do not end in newline, so we read all of it and
                                          // send it out to connected slots.
                                          std::istream is(&pipe.buffer);
                                          std::string remaining(pipe.buffer.size(), '\0');
                                          is.read(remaining.data(), remaining.size());
                                          pipe.signal(m_pid, remaining);
 
                                          // Close pipe to signal we're done.
                                          pipe.pipe.close();
                                          CheckCompleted();
 
                                          // We're done, do not schedule another read.
                                          return;
                                      }
 
                                      // Signal any slots.
                                      std::istream is(&pipe.buffer);
                                      std::string line;
                                      std::getline(is, line);
                                      // note: getline consumes '\n' which we don't want since we
                                      // don't want to modify source at all.
                                      line.push_back('\n');
                                      pipe.signal(m_pid, line);
 
                                      AsyncReadStream(pipe);
                                  });
}
 
void AsyncProcess::CheckCompleted() {
    // Note that we cannot check if process is alive with m_proc.is_running() as this
    if (!m_result.has_value() || m_stdout.pipe.is_open() || m_stderr.pipe.is_open()) {
        // Not all read operations have completed.
        return;
    }
    if (m_result->ec) {
        m_promise.set_exception(std::system_error(m_result->ec));
    } else {
        m_promise.set_value(m_result->exit_code);
    }
}
 
LogCaptureLast::LogCaptureLast(log4cplus::Logger logger, std::size_t num_lines)
    : m_logger(logger), m_buffer(num_lines) {
}
 
void LogCaptureLast::operator()(pid_t pid, std::string const& line) noexcept {
    LOG4CPLUS_INFO(m_logger, '[' << pid << "]: " << Trim(line));
    m_buffer.push_back(line);
}
 
std::ostream& operator<<(std::ostream& os, LogCaptureLast const& lines) {
    for (auto const& line : lines.Lines()) {
        // note: Line is unmodified from program and ends with \n
        os << line;
    }
    return os;
}
 
}  // namespace daq