asyncProcess.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_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 definition
 */
#ifndef DAQ_ASYNC_PROCESS_HPP
#define DAQ_ASYNC_PROCESS_HPP
#include <daq/config.hpp>
 
#include <iosfwd>
#include <optional>
#include <string>
#include <vector>
 
#include <boost/circular_buffer.hpp>
#include <boost/process.hpp>
#include <boost/signals2/signal.hpp>
#include <boost/thread/future.hpp>
#include <log4cplus/logger.h>
 
namespace daq {
 
/**
 * Interface to asynchronous process.
 */
class AsyncProcessIf {
public:
    virtual ~AsyncProcessIf(){};
 
    /**
     * Initiates async operation by executing the specified process.
     *
     * This can only be called once.
     */
    [[nodiscard]] virtual boost::future<int> Initiate() = 0;
 
    /**
     * Get PID.
     *
     * @returns pid of process if process is running, otherwise nullopt.
     */
    virtual std::optional<pid_t> GetPid() const noexcept = 0;
 
    /**
     * Aborts the operation by terminating process which completes the operation.
     * If process is not running this will not do anything.
     *
     * @returns Error code of operation. If process is not running, because it has not been started
     * for with AsyncProcess::Initiate() for example, it returns std::errc::no_such_process.
     */
    virtual std::error_code Abort() noexcept = 0;
 
    /**
     * Send signal to process. Unlike Abort() this can be used to gracefully terminate application
     * by sending e.g. SIGTERM.
     *
     * @param sig Signal to send.
     * @returns Error code of operation. If process is not running, because it has not been started
     * for with AsyncProcess::Initiate() for example, it returns std::errc::no_such_process.
     */
    virtual std::error_code Signal(int sig) noexcept = 0;
 
    /**
     * @return arguments used when launching process.
     */
    virtual std::vector<std::string> const& GetArguments() const noexcept = 0;
 
    /**
     * @returns true if process is running.
     * @returns false otherwise.
     */
    virtual bool IsRunning() const noexcept = 0;
 
    /**
     * @name Signals
     */
    /// @{
    /**
     * Signal type for stdout/stderr signals.
     */
    using SigOutStream = boost::signals2::signal<void(pid_t, std::string const&)>;
 
    /**
     * Connect slot to line-buffered stdout signal.
     *
     * Signal is invoked for every line read from process.
     *
     * @returns connection object.
     */
    virtual boost::signals2::connection ConnectStdout(SigOutStream::slot_type const& slot) = 0;
 
    /**
     * Connect slot to line-buffered stderr signal.
     *
     * Signal is invoked for every line read from process.
     *
     * @returns connection object.
     */
    virtual boost::signals2::connection ConnectStderr(SigOutStream::slot_type const& slot) = 0;
    /// @}
};
 
/**
 * Formats proc representation in the form `[<pid>] <args>`
 *
 * @param os Ostream to format to.
 * @param proc instance to format.
 */
std::ostream& operator<<(std::ostream& os, AsyncProcessIf const& proc);
 
/**
 * Represents a subprocess as an asynchronous operation.
 *
 * Once constructed the operation is initiated (only once) with `Initiate()` which starts the
 * process and returns a boost::future object that will receive exit code when process terminates
 * *and* all output has been read.
 *
 * @note All completion handlers have been executed and no signals will be emitted after future has
 * received the value or exception, it is safe to delete AsyncProcess after this.
 *
 * Operation can be aborted with `Abort()` which will terminate process and set future with
 * exceptional result.
 *
 * boost::process is pretty buggy so be very careful making changes to this.
 * Examples:
 * - Do not check if process is running with child::is_running() after `on_exit` has been executed.
 *   This cause exit codes to be lost [https://github.com/boostorg/process/issues/187]
 *
 * @ingroup daq_common_libdaq
 */
class AsyncProcess : public virtual AsyncProcessIf {
public:
    /**
     * Constructor.
     *
     * @note Does not start the process or any other asynchronous operations. This is done in
     * AsyncProcess::Initiate().
     *
     * @param ctx io_context instance to use.
     * @param args Command line arguments. First argument specify the file to be executed.
     */
    explicit AsyncProcess(boost::asio::io_context& ctx, std::vector<std::string> args);
    virtual ~AsyncProcess() noexcept;
 
    /**
     * Starts process and asynchronous operations that read stdout and stderr.
     *
     * This can only be called once.
     *
     * @note Caller is responsible for keeping AsyncProcess alive until result is set on future.
     *
     * @returns Future that will receive process exit code @a after process has terminated and all
     * output has been read.
     */
    [[nodiscard]] boost::future<int> Initiate() override;
    std::optional<pid_t> GetPid() const noexcept override;
 
    /**
     * Aborts the operation by terminating process which completes the operation.
     * If process is not running this will not do anything.
     *
     * @returns Error code of operation. If process is not running, because it has not been started
     * for with AsyncProcess::Initiate() for example, it returns std::errc::no_such_process.
     */
    std::error_code Abort() noexcept override;
 
    /**
     * Send signal to process. Unlike Abort() this can be used to gracefully terminate application
     * by sending e.g. SIGTERM.
     *
     * @param sig Signal to send.
     * @returns Error code of operation. If process is not running, because it has not been started
     * for with AsyncProcess::Initiate() for example, it returns std::errc::no_such_process.
     */
    std::error_code Signal(int sig) noexcept override;
 
    /**
     * @returns true if process is running.
     * @returns false otherwise.
     */
    bool IsRunning() const noexcept override;
 
    std::vector<std::string> const& GetArguments() const noexcept override {
        return m_args;
    }
 
    /**
     * @name Signals
     */
    /// @{
    using AsyncProcessIf::SigOutStream;
    boost::signals2::connection ConnectStdout(SigOutStream::slot_type const& slot) override {
        return m_stdout.signal.connect(slot);
    }
    boost::signals2::connection ConnectStderr(SigOutStream::slot_type const& slot) override {
        return m_stderr.signal.connect(slot);
    }
    /// @}
 
protected:
private:
    /**
     * Describes process result.
     */
    struct Result {
        /**
         * If process was started this contains the exit_code.
         */
        int exit_code;
        /**
         * If process start fails this will be non-zero.
         */
        std::error_code ec;
    };
    struct Pipe {
        boost::process::async_pipe pipe;
        boost::asio::streambuf buffer = boost::asio::streambuf();
        SigOutStream signal = {};
    };
    /**
     * Checks if operation is completed and sets promise if it is.
     *
     * Three async operations may complete in any order and the future should be set only when all
     * three complete.
     *
     * - stdout PIPE reads. Completes when read get EOF.
     * - stderr PIPE reads. Completes when read get EOF.
     * - on_exit handler, completed automatically when process exits.
     *
     * Each of these must invoke CheckCompleted() which will trigger completion of future only after
     * all operations have completed.
     */
    void CheckCompleted();
    /**
     * Recursive async read initator */
    void AsyncReadStream(Pipe&);
 
    boost::asio::io_context& m_io_ctx;
    std::vector<std::string> m_args;
    /** @name PIPEs and buffers
     * @{
     */
    Pipe m_stdout;
    Pipe m_stderr;
    /** @} */
 
    boost::process::child m_proc;
    pid_t m_pid;
    std::optional<Result> m_result;
    boost::promise<int> m_promise;
};
 
/**
 * Logs output to logger and keeps last N lines in circular buffer for later retrival.
 */
class LogCaptureLast {
public:
    LogCaptureLast(log4cplus::Logger logger, std::size_t num_lines);
    void operator()(pid_t pid, std::string const& line) noexcept;
 
    auto Lines() const -> boost::circular_buffer<std::string> {
        return m_buffer;
    }
 
 
private:
    log4cplus::Logger m_logger;
    boost::circular_buffer<std::string> m_buffer;
};
 
std::ostream& operator<<(std::ostream& os, LogCaptureLast const& lines);
 
}  // namespace daq
 
#endif  // #ifndef DAQ_ASYNC_PROCESS_HPP