engine_json_nrp_client.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_JSON_NRP_CLIENT_H
#define ENGINE_JSON_NRP_CLIENT_H
 
#include "nrp_general_library/engine_interfaces/engine_client_interface.h"
#include "nrp_json_engine_protocol/config/engine_json_config.h"
#include "nrp_json_engine_protocol/nrp_client/engine_json_registration_server.h"
#include "nrp_json_engine_protocol/datapack_interfaces/json_datapack.h"
#include "nrp_general_library/utils/nrp_exceptions.h"
#include "nrp_general_library/utils/restclient_setup.h"
 
#include <nlohmann/json.hpp>
#include <list>
 
#include <chrono>
 
#include <iostream>
#include <restclient-cpp/restclient.h>
 
template<class ENGINE, const char* SCHEMA>
class EngineJSONNRPClient
        : public EngineClient<ENGINE, SCHEMA>
{
    public:
        EngineJSONNRPClient(nlohmann::json &config, ProcessLauncherInterface::unique_ptr &&launcher)
            : EngineClient<ENGINE, SCHEMA>(config, std::move(launcher)),
              _serverAddress(this->engineConfig().at("ServerAddress"))
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            RestClientSetup::ensureInstance();
        }
 
        EngineJSONNRPClient(const std::string &serverAddress, nlohmann::json &config, ProcessLauncherInterface::unique_ptr &&launcher)
            : EngineClient<ENGINE, SCHEMA>(config, std::move(launcher)),
              _serverAddress(serverAddress)
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            this->engineConfig()["ServerAddress"] = this->_serverAddress;
            RestClientSetup::ensureInstance();
        }
 
        virtual ~EngineJSONNRPClient() override = default;
 
        virtual pid_t launchEngine() override
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            this->startRegistrationServer();
 
            // Launch engine process.
            // enginePID == -1 means that the launcher hasn't launched a process
            auto enginePID = this->EngineClientInterface::launchEngine();
 
            // Wait for engine to register itself if process launching has succeeded
            if(enginePID > 0 && !this->engineConfig().at("RegistrationServerAddress").empty())
            {
                const auto serverAddr = this->waitForRegistration(20, 1);
 
                if(serverAddr.empty())
                    throw NRPException::logCreate("Error while waiting for engine \"" + this->engineName() + "\" to register its address. Did not receive a reply");
 
                this->engineConfig()["ServerAddress"] = serverAddr;
                this->_serverAddress = serverAddr;
            }
 
            return enginePID;
        }
 
        virtual void sendDataPacksToEngine(const datapacks_set_t & dataPacks) override
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            // TODO Early return if no datapacks?
 
            // Convert datapacks to JSON format
            nlohmann::json request;
            for(auto curDataPack : dataPacks)
            {
                assert(curDataPack->engineName() == this->engineName());
 
                if(curDataPack->isEmpty())
                    throw NRPException::logCreate("Attempt to send empty datapack " + curDataPack->name() + " to Engine " + this->engineName());
                else if(curDataPack->type() != JsonDataPack::getType())
                {
                    throw NRPException::logCreate("Engine \"" +
                                                this->engineName() +
                                                "\" cannot handle datapack type '" +
                                                curDataPack->type() + "'");
                }
 
                const nlohmann::json & data = (dynamic_cast<const JsonDataPack *>(curDataPack.get()))->getData();
                const auto & name = curDataPack->name();
 
                request[name]["engine_name"] = curDataPack->engineName();
                request[name]["type"]        = curDataPack->type();
                request[name]["data"]        = data;
            }
 
            // Send updated datapacks to Engine JSON server
            EngineJSONNRPClient::sendRequest(this->_serverAddress + "/" + EngineJSONConfigConst::EngineServerSetDataPacksRoute.data(),
                                             EngineJSONConfigConst::EngineServerContentType.data(), request.dump(),
                                             "Engine server \"" + this->engineName() + "\" failed during datapack handling");
 
            // TODO: Check if engine has processed all sent datapacks
        }
 
        virtual const std::vector<std::string> engineProcStartParams() const override
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            std::vector<std::string> startParams = this->engineConfig().at("EngineProcStartParams");
 
            std::string name = this->engineConfig().at("EngineName");
            startParams.push_back(std::string("--") + EngineJSONConfigConst::EngineNameArg.data() + "=" + name);
 
            // Add JSON Server address (will be used by EngineJSONServer)
            std::string address = this->engineConfig().at("ServerAddress");
            startParams.push_back(std::string("--") + EngineJSONConfigConst::EngineServerAddrArg.data() + "=" + address);
 
            // Add JSON registration Server address (will be used by EngineJSONServer)
            std::string reg_address = this->engineConfig().at("RegistrationServerAddress");
            startParams.push_back(std::string("--") + EngineJSONConfigConst::EngineRegistrationServerAddrArg.data() + "=" + reg_address);
 
            return startParams;
        }
 
        virtual datapacks_vector_t getDataPacksFromEngine(const datapack_identifiers_set_t &requestedDataPackIds) override
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            nlohmann::json request;
            for(const auto &id : requestedDataPackIds)
            {
                if(this->engineName().compare(id.EngineName) == 0)
                {
                    request[id.Name] = {{"engine_name", id.EngineName}, {"type", id.Type}};
                }
 
            }
 
            // Post request to Engine JSON server
            const auto resp(EngineJSONNRPClient::sendRequest(this->_serverAddress + "/" + EngineJSONConfigConst::EngineServerGetDataPacksRoute.data(),
                                                             EngineJSONConfigConst::EngineServerContentType.data(), request.dump(),
                                                             "Engine server \"" + this->engineName() + "\" failed during datapack retrieval"));
 
            return this->getDataPackInterfacesFromJSON(resp);
        }
 
protected:
        nlohmann::json sendInitCommand(const nlohmann::json &data)
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            NRPLogger::debug("EngineJSONNRPClient::sendInitCommand [ data: {} ]", data.dump());
 
            // Post init request to Engine JSON server
            return sendRequest(this->_serverAddress + "/" + EngineJSONConfigConst::EngineServerInitializeRoute.data(),
                               EngineJSONConfigConst::EngineServerContentType.data(), data.dump(),
                               "Engine server \"" + this->engineName() + "\" failed during initialization");
        }
 
        nlohmann::json sendResetCommand(const nlohmann::json &data)
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            NRPLogger::debug("EngineJSONNRPClient::sendResetCommand [ data: {} ]", data.dump());
 
            // Post reset request to Engine JSON server
            return sendRequest(this->_serverAddress + "/" + EngineJSONConfigConst::EngineServerResetRoute.data(),
                               EngineJSONConfigConst::EngineServerContentType.data(), data.dump(),
                               "Engine server \"" + this->engineName() + "\" failed during reset");
        }
 
        nlohmann::json sendShutdownCommand(const nlohmann::json &data)
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            NRPLogger::debug("EngineJSONNRPClient::sendShutdownCommand [ data: {} ]", data.dump());
 
            // Post init request to Engine JSON server
            return sendRequest(this->_serverAddress + "/" + EngineJSONConfigConst::EngineServerShutdownRoute.data(),
                               EngineJSONConfigConst::EngineServerContentType.data(), data.dump(),
                               "Engine server \"" + this->engineName() + "\" failed during shutdown");
        }
 
        std::string waitForRegistration(unsigned int numTries, unsigned int waitTime) const
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            auto *pRegistrationServer = EngineJSONRegistrationServer::getInstance();
 
            assert(pRegistrationServer != nullptr);
            assert(pRegistrationServer->isRunning());
 
            // Try to retrieve engine address
            auto engineAddr = pRegistrationServer->requestEngine(this->engineName());
 
            while(engineAddr.empty() && numTries > 0)
            {
                // Continue to wait for engine address for 20s
                sleep(waitTime);
                engineAddr = pRegistrationServer->retrieveEngineAddress(this->engineName());
                --numTries;
            }
 
            // Close server if no additional clients are waiting for their engines to register
            if(pRegistrationServer->getNumWaitingEngines() == 0)
            {
                EngineJSONRegistrationServer::clearInstance();
            }
 
            return engineAddr;
        }
 
    private:
 
        const unsigned NUM_RETRIES_REG_SERVER = 2;
 
        std::string _serverAddress;
 
        static nlohmann::json sendRequest(const std::string &serverName, const std::string &contentType, const std::string &request, const std::string_view &exceptionMessage)
        {
            // Post request to Engine JSON server
            try
            {
                auto resp = RestClient::post(serverName, contentType, request);
                if(resp.code != 200)
                {
                    throw std::domain_error("Request failed with code " + std::to_string(resp.code) +
                                            " and message:" + "\n" + resp.body +
                                            "Client context message: " + exceptionMessage.data());
                }
 
                return nlohmann::json::parse(resp.body);
            }
            catch(std::exception &e)
            {
                throw NRPException::logCreate(e, "Communication with engine server failed");
            }
        }
 
        SimulationTime runLoopStepCallback(SimulationTime timeStep) override
        {
            nlohmann::json request;
            request[EngineJSONConfigConst::EngineTimeStepName.data()] = timeStep.count();
 
            // Post run loop request to Engine JSON server
            nlohmann::json resp(EngineJSONNRPClient::sendRequest(this->_serverAddress + "/" + EngineJSONConfigConst::EngineServerRunLoopStepRoute.data(),
                                                                 EngineJSONConfigConst::EngineServerContentType.data(), request.dump(),
                                                                 "Engine Server failed during loop execution"));
 
            // Get engine time from response
            SimulationTime engineTime;
            try
            {
                engineTime = SimulationTime(resp[EngineJSONConfigConst::EngineTimeName.data()]);
            }
            catch(std::exception &e)
            {
                throw NRPException::logCreate(e, "Error while parsing the return value of the run_step of \"" + this->engineName() + "\"");
            }
 
            if(engineTime < SimulationTime::zero())
                throw NRPException::logCreate("Error during execution of engine \"" + this->engineName() + "\"");
 
            return engineTime;
        }
 
        datapacks_vector_t getDataPackInterfacesFromJSON(const nlohmann::json &datapacks) const
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            datapacks_vector_t interfaces;
 
            for(auto curDataPackIterator = datapacks.begin(); curDataPackIterator != datapacks.end(); ++curDataPackIterator)
            {
                try
                {
                    DataPackIdentifier datapackID(curDataPackIterator.key(),
                                              (*curDataPackIterator)["engine_name"],
                                              (*curDataPackIterator)["type"]);
 
                    datapackID.EngineName = this->engineName();
                    interfaces.push_back(this->getSingleDataPackInterfaceFromJSON(curDataPackIterator, datapackID));
                }
                catch(std::exception &e)
                {
                    // TODO: Handle json datapack parsing error
                    throw NRPException::logCreate(e, "Failed to parse JSON DataPack Interface");
                }
            }
 
            return interfaces;
        }
 
        inline DataPackInterfaceConstSharedPtr getSingleDataPackInterfaceFromJSON(const nlohmann::json::const_iterator &datapackData, DataPackIdentifier &datapackID) const
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
 
            if(datapackID.Type == JsonDataPack::getType())
            {
                // Check whether the requested datapack has new data
                // A datapack that has no data will contain two JSON objects with "engine_name" and "type" keys (parts of datapack ID)
 
                if(datapackData->size() == 2)
                {
                    // There's no meaningful data in the datapack, so create an empty datapack with datapack ID only
 
                    return DataPackInterfaceSharedPtr(new DataPackInterface(std::move(datapackID)));
                }
 
                nlohmann::json * data = new nlohmann::json(std::move((*datapackData)["data"]));
 
                DataPackInterfaceSharedPtr newDataPack(new JsonDataPack(datapackID.Name, datapackID.EngineName, data));
                newDataPack->setEngineName(this->engineName());
 
                return newDataPack;
            }
            else
            {
                throw NRPException::logCreate("DataPack type \"" + datapackID.Type + "\" cannot be handled by the \"" + this->engineName() + "\" engine");
            }
        }
 
        void startRegistrationServer()
        {
            const std::string regServerAddressConfig = this->engineConfig().at("RegistrationServerAddress");
            const std::string regServerAddressActual = EngineJSONRegistrationServer::tryInstantiate(regServerAddressConfig, NUM_RETRIES_REG_SERVER);
 
            if(regServerAddressActual != regServerAddressConfig)
            {
                NRPLogger::warn("Registration server failed to bind to the specified address '{}'. Using '{}' instead.", regServerAddressConfig, regServerAddressActual);
 
                // Update the address in the config
                // This value will be passed to the forked process
 
                this->engineConfig()["RegistrationServerAddress"] = regServerAddressActual;
            }
        }
};
 
 
#endif //ENGINE_JSON_NRP_CLIENT_H