main.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_dpm_server
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief daqDpmServer entrypoint.
 */
#include <daq/config.hpp>
#include <iostream>
 
#include <boost/asio.hpp>
#include <boost/program_options.hpp>
#include <ciiLogManager.hpp>
#include <log4cplus/configurator.h>
#include <log4cplus/logger.h>
#include <log4cplus/loggingmacros.h>
#include <rad/ioExecutor.hpp>
#include <rad/logger.hpp>
#include <rad/mal/publisher.hpp>
#include <rad/mal/replier.hpp>
#include <rad/oldbAsyncWriter.hpp>
 
#include <daq/conversion.hpp>
#include <daq/dpm/dpmService.hpp>
#include <daq/dpm/scheduler.hpp>
#include <daq/error/report.hpp>
#include <daq/status.hpp>
 
#include <daqif/uri.hpp>
 
#include "configManager.hpp"
 
namespace error {
/**
 * Error codes
 */
enum {
    /** Invalid program arguments */
    InvalidProgramArgument = 1,
    Unknown = 255,
};
}  // namespace error
 
namespace po = boost::program_options;
namespace daq::dpm {
 
int Entrypoint(log4cplus::Logger const& logger, ConfigManager& cfg_mgr) {
    boost::asio::io_context io_ctx;
    rad::IoExecutor exec(io_ctx);
    Configuration const& config = cfg_mgr.GetConfig();
 
    auto ws_path = config.workspace;
    if (ws_path.is_relative()) {
        ws_path = config.dataroot / ws_path;
    }
 
    // Construction also initializes/verifies workspace and may throw.
    WorkspaceImpl workspace(ws_path);
 
    // Scheduler
    SchedulerOptions scheduler_options;
    scheduler_options.concurrency_limits.daq = config.limit_daq;
    scheduler_options.concurrency_limits.merge = config.limit_merge;
    scheduler_options.concurrency_limits.net_send = config.limit_net_send;
    scheduler_options.concurrency_limits.net_receive = config.limit_net_receive;
 
    DaqControllerOptions daq_controller_options;
    daq_controller_options.merge_bin = config.merge_bin;
    daq_controller_options.rsync_bin = config.rsync_bin;
 
    DaqControllerImpl::RsyncFactory rsync_factory =
        [](boost::asio::io_context& io_ctx,
           std::string source,  // source
           std::string dest,    // dest
           RsyncOptions const& opts,
           RsyncAsyncProcess::DryRun dry_run) -> std::unique_ptr<RsyncAsyncProcessIf> {
        return std::make_unique<RsyncAsyncProcess>(
            io_ctx, std::move(source), std::move(dest), opts, dry_run);
    };
 
    DaqControllerImpl::ProcFactory proc_factory = [](boost::asio::io_context& io_ctx,
                                                     std::vector<std::string> args) {
        return std::make_unique<AsyncProcess>(io_ctx, std::move(args));
    };
 
    SchedulerImpl scheduler(
        exec,
        workspace,
        [&](std::unique_ptr<DaqWorkspace> workspace,
            Resources& resources) -> std::unique_ptr<DaqController> {
            return std::make_unique<DaqControllerImpl>(exec,
                                                       std::move(workspace),
                                                       resources,
                                                       rsync_factory,
                                                       proc_factory,
                                                       daq_controller_options);
        },
        scheduler_options);
 
    // MAL variables that have an empty value if --no-ipc is used.
    std::optional<rad::cii::Replier> replier;
    std::shared_ptr<elt::mal::Mal> mal;
    std::unique_ptr<rad::cii::Publisher<daqif::DaqStatus>> daq_status_publisher;
    std::unique_ptr<rad::cii::Publisher<daqif::InternalDaqStatus>> daq_dpm_status_publisher;
    std::optional<rad::OldbAsyncWriter> oldb_writer;
    if (!config.no_ipc) {
        mal = elt::mal::loadMal("zpb", {});
        auto dpm_service = std::make_shared<DpmService>(exec, *mal, workspace, scheduler);
        /*
         *  Load CII/MAL middleware here because it resets
         *  the log4cplus configuration!
         */
        elt::mal::CiiFactory& factory = elt::mal::CiiFactory::getInstance();
        factory.registerMal("zpb", mal);
 
        /*
         * Register CII/MAL replier
         */
        replier.emplace(elt::mal::Uri(config.rr_uri));
        replier->RegisterService<daqif::AsyncDpmControl>("dpm", dpm_service);
        replier->RegisterService<daqif::AsyncDpmDaqControl>("daq", dpm_service);
        daq_status_publisher = std::make_unique<rad::cii::Publisher<daqif::DaqStatus>>(
            daqif::MakeServiceUri(config.ps_uri, "daq/status"));
        daq_dpm_status_publisher = std::make_unique<rad::cii::Publisher<daqif::InternalDaqStatus>>(
            daqif::MakeServiceUri(config.ps_uri, "internal/daq/status"));
        {
            auto daq_status_sample = daq_status_publisher->CreateTopic();
            auto daq_dpm_status_sample = daq_dpm_status_publisher->CreateTopic();
            scheduler.ConnectStatus([&,
                                     daq_status_sample = std::move(daq_status_sample),
                                     daq_dpm_status_sample =
                                         std::move(daq_dpm_status_sample)](Status const& status) {
                try {
                    LOG4CPLUS_TRACE(logger, "Publishing DAQ status update: " << status);
                    *daq_status_sample << status;
                    daq_status_publisher->Publish(*daq_status_sample);
                    *daq_dpm_status_sample << status;
                    daq_dpm_status_publisher->Publish(*daq_dpm_status_sample);
                } catch (std::exception const& e) {
                    LOG4CPLUS_ERROR(logger, fmt::format("Failed to publish status: {}", e.what()));
                }
            });
        }
        // Write configuration
        {
            oldb_writer.emplace(config.db_timeout, std::chrono::milliseconds(250));
            auto prefix = config.db_prefix;
            if (prefix.empty()) {
                throw std::invalid_argument("OLDB prefix is invalid (empty)");
            }
            if (prefix.back() != '/') {
                prefix.push_back('/');
            }
 
            cfg_mgr.Visit([&](daq::config::Manager::CiiValue const& param) {
                oldb_writer->Set(prefix + param.metadata.canonical_name, param.value);
            });
            oldb_writer->StartWriter();
        }
    } else {
        LOG4CPLUS_INFO(logger,
                       "Note: Running in standalone mode without interprocess communication."
                       " Stop process using CTRL-C/SIGINT");
    }
 
    // Run until requested to exit
    LOG4CPLUS_INFO(logger, fmt::format("Application daqDpmServer started as '{}'", config.name));
    scheduler.Start();
    if (config.poll_once) {
        // First poll may initiate async activities, after which we Stop scheduler and let it run to
        // completion.
        LOG4CPLUS_INFO(logger,
                       "Note: Polling once due to (--poll-once) to start operations, wait for "
                       "those to complete and then then exit");
        io_ctx.poll();
        scheduler.Stop();
        io_ctx.restart();
    } else {
        io_ctx.run();
    }
    return 0;
}
}  // namespace daq::dpm
 
int main(int argc, char* argv[]) {
    try {
        auto log_initializer = rad::LogInitializer();
        log4cplus::Logger logger = log4cplus::Logger::getInstance(daq::dpm::LOGGER_NAME);
        try {
            using daq::dpm::Configuration;
            // @todo Uses default configuration at the moment. Load from command line and
            // configuration file and environment variables as well.
            daq::dpm::ConfigManager cfg_mgr(logger);
            if (!cfg_mgr.ParseArguments(argc, argv)) {
                // --help was invoked
                return 0;
            }
 
            if (!cfg_mgr.GetConfig().config_file.empty()) {
                cfg_mgr.LoadConfig();
            }
            elt::log::CiiLogManager::SetApplicationName(cfg_mgr.GetConfig().name);
            auto level = log4cplus::getLogLevelManager().toString(cfg_mgr.GetConfig().log_level);
            auto* daq_logs = std::getenv("DAQ_LOGS");
            elt::log::CiiLogManager::Configure(rad::GetDefaultLogProperties(
                daq_logs == nullptr ? std::string()
                                    : fmt::format("{}/{}.log", daq_logs, cfg_mgr.GetConfig().name),
                level));
 
            if (auto log_cfg = cfg_mgr.GetConfig().log_config; !log_cfg.empty()) {
                auto resolved = rad::Helper::FindFile(log_cfg);
                if (resolved.empty()) {
                    LOG4CPLUS_ERROR(logger,
                                    "Configured log property file not found: '"
                                        << log_cfg
                                        << "', $CFGPATH=" << rad::Helper::GetEnvVar("CFGPATH"));
                    return EXIT_FAILURE;
                }
                elt::log::CiiLogManager::Configure(resolved);
            } else {
                log4cplus::Logger::getRoot().setLogLevel(cfg_mgr.GetConfig().log_level);
            }
 
            int rc = daq::dpm::Entrypoint(logger, cfg_mgr);
            LOG4CPLUS_INFO(logger, "Application terminating with code " << rc);
            return rc;
        } catch (po::error const& e) {
            daq::error::ReportNestedExceptions(std::cerr, e);
            return error::InvalidProgramArgument;
        } catch (std::exception const& ex) {
            LOG4CPLUS_ERROR(logger, daq::error::NestedExceptionReporter(ex));
            return error::Unknown;
        }
    } catch (...) {
        std::cerr << "Unknown error during log initialization\n";
        return error::Unknown;
    }
}