engine_client_interface.h

Go to the documentation of this file.

/* * NRP Core - Backend infrastructure to synchronize simulations
 *
 * Copyright 2020-2023 NRP Team
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * This project has received funding from the European Union’s Horizon 2020
 * Framework Programme for Research and Innovation under the Specific Grant
 * Agreement No. 945539 (Human Brain Project SGA3).
 */
 
#ifndef ENGINE_CLIENT_INTERFACE_H
#define ENGINE_CLIENT_INTERFACE_H
 
#include "nrp_general_library/process_launchers/process_launcher.h"
#include "nrp_general_library/utils/fixed_string.h"
#include "nrp_general_library/utils/ptr_templates.h"
#include "nrp_general_library/utils/time_utils.h"
#include "nrp_general_library/utils/json_schema_utils.h"
#include "nrp_general_library/datapack_interface/datapack_interface.h"
 
#include <set>
#include <vector>
#include <future>
 
class EngineClientInterface;
class EngineLauncherInterface;
 
 
class EngineClientInterface
        : public PtrTemplates<EngineClientInterface>
{
    public:
        explicit EngineClientInterface(ProcessLauncherInterface::unique_ptr &&launcher);
        virtual ~EngineClientInterface();
 
        virtual const std::string engineName() const = 0;
 
        virtual const nlohmann::json &engineConfig() const = 0;
 
        virtual nlohmann::json &engineConfig() = 0;
 
        virtual const std::vector<std::string> engineProcStartParams() const = 0;
 
        virtual pid_t launchEngine();
 
        virtual void initialize() = 0;
 
        virtual void reset() = 0;
 
        virtual void shutdown() = 0;
 
        virtual SimulationTime getEngineTimestep() const = 0;
 
        virtual SimulationTime getEngineTime() const = 0;
 
        virtual const std::string engineSchema() const = 0;
 
        virtual void runLoopStepAsync(SimulationTime timeStep) = 0;
 
        virtual void runLoopStepAsyncGet(SimulationTime timeOut) = 0;
 
        virtual void sendDataPacksToEngine(const datapacks_set_t &dataPacks) = 0;
 
        virtual datapacks_vector_t getDataPacksFromEngine(const datapack_identifiers_set_t &datapackIdentifiers) = 0;
 
protected:
 
        ProcessLauncherInterface::unique_ptr _process;
};
 
using EngineClientInterfaceSharedPtr = EngineClientInterface::shared_ptr;
using EngineClientInterfaceConstSharedPtr = EngineClientInterface::const_shared_ptr;
 
class EngineLauncherInterface
        : public PtrTemplates<EngineLauncherInterface>
{
    public:
        using engine_type_t = decltype(DataPackIdentifier::Type);
 
        EngineLauncherInterface(const engine_type_t &engineType);
        virtual ~EngineLauncherInterface() = default;
 
        const engine_type_t &engineType() const;
        virtual EngineClientInterfaceSharedPtr launchEngine(nlohmann::json  &engineConfig, ProcessLauncherInterface::unique_ptr &&launcher) = 0;
 
    private:
        const engine_type_t _engineType;
};
 
using EngineLauncherInterfaceSharedPtr = EngineLauncherInterface::shared_ptr;
using EngineLauncherInterfaceConstSharedPtr = EngineLauncherInterface::const_shared_ptr;
 
template<class ENGINE, const char *SCHEMA>
class EngineClient
        : public EngineClientInterface
{
    public:
        using engine_t = ENGINE;
 
        template<const char *ENGINE_TYPE>
        class EngineLauncher
                : public EngineLauncherInterface
        {
            public:
                EngineLauncher()
                    : EngineLauncherInterface(ENGINE_TYPE)
                {}
 
                EngineLauncher(const engine_type_t &engineType)
                    : EngineLauncherInterface(engineType)
                {}
 
                ~EngineLauncher() override = default;
 
                EngineClientInterfaceSharedPtr launchEngine(nlohmann::json &engineConfig, ProcessLauncherInterface::unique_ptr &&launcher) override
                {
                    EngineClientInterfaceSharedPtr engine(new ENGINE(engineConfig, std::move(launcher)));
 
                    switch (engine->launchEngine())
                    {
                        case 0: { // TODO: process not forked (error)
                                NRPLogger::error(
                                    "\"{}\" Engine (type: \"{}\") could NOT be launched.", engine->engineName(), this->engineType());
                            }; break;
                        case -1: { // process not forked (empty launcher, expected behaviour)
                                NRPLogger::info(
                                    "\"{}\" EngineClient (type: \"{}\") won't launch an EngineServer. Assume it's already been launched.", engine->engineName(), this->engineType());
                            }; break;
                        default: { // success
                                NRPLogger::info(
                                    "\"{}\" Engine (type: \"{}\") launched successfully", engine->engineName(), this->engineType());
                            }; break;
                    };
 
                    return engine;
                }
        };
 
        EngineClient(nlohmann::json  &engineConfig, ProcessLauncherInterface::unique_ptr &&launcher)
            : EngineClientInterface(std::move(launcher)),
              engineConfig_(engineConfig)
        {
            // validate engine config
            json_utils::validateJson(this->engineConfig(), this->engineSchema());
 
            // setting process start and env params to an empty vector since this can't be done from json schema
            setDefaultProperty<std::vector<std::string>>("EngineProcStartParams", std::vector<std::string>());
            setDefaultProperty<std::vector<std::string>>("EngineEnvParams", std::vector<std::string>());
            setDefaultProperty<nlohmann::json>("EngineExtraConfigs", nlohmann::json(json::value_t::object));
        }
 
        ~EngineClient() override = default;
 
        const std::string engineName() const override final
        { return this->engineConfig().at("EngineName"); }
 
        SimulationTime getEngineTimestep() const override final
        {
            // We need to cast floating-point seconds to integers with units of SimulationTime type
            return toSimulationTime<float, std::ratio<1>>(this->engineConfig().at("EngineTimestep"));
        }
 
        const nlohmann::json &engineConfig() const override final
        {   return engineConfig_;   }
 
        nlohmann::json &engineConfig() override final
        {   return engineConfig_;   }
 
        const std::string engineSchema() const override final
        { return schema;   }
 
        SimulationTime getEngineTime() const override
        {
            return this->_engineTime;
        }
 
        void runLoopStepAsync(SimulationTime timeStep) override
        {
            if(this->_loopStepThread.valid())
            {
                throw NRPException::logCreate("Engine \"" + this->engineName() + "\" runLoopStepAsync has overrun");
            }
 
            this->_loopStepThread = std::async(std::launch::async, std::bind(&EngineClient::runLoopStepCallback, this, timeStep));
        }
 
        void runLoopStepAsyncGet(SimulationTime timeOut) override
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            // If thread state is invalid, loop thread has completed and runLoopStepAsyncGet was called once before
            if(!this->_loopStepThread.valid())
                return;
 
            // Wait until timeOut has passed
            if(timeOut > SimulationTime::zero())
            {
                if(this->_loopStepThread.wait_for(timeOut) != std::future_status::ready)
                    throw NRPException::logCreate("Engine \"" + this->engineName() + "\" loop is taking too long to complete");
            }
 
            // Retrieve the engine time returned from the loop step
 
            this->_engineTime = this->_loopStepThread.get();
        }
 
    protected:
 
        virtual void resetEngineTime()
        {
            this->_engineTime = SimulationTime(0);
        }
 
        template<class T>
        void setDefaultProperty(std::string key, T value)
        {
            json_utils::setDefault<T>(this->engineConfig(), key, value);
        }
 
        virtual SimulationTime runLoopStepCallback(SimulationTime timeStep) = 0;
 
    private:
 
        nlohmann::json engineConfig_;
        std::string schema = std::string(SCHEMA);
        std::future<SimulationTime> _loopStepThread;
 
        SimulationTime _engineTime = SimulationTime::zero();
};
 
#endif // ENGINE_CLIENT_INTERFACE_H