engine_grpc_server.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_GRPC_SERVER_H
#define ENGINE_GRPC_SERVER_H
 
#include <string>
#include <map>
#include <type_traits>
 
#include <grpcpp/grpcpp.h>
#include <grpcpp/health_check_service_interface.h>
#include <nlohmann/json.hpp>
 
#include "nrp_protobuf/engine_grpc.grpc.pb.h"
#include "nrp_general_library/engine_interfaces/datapack_controller.h"
#include "nrp_general_library/utils/time_utils.h"
#include "nrp_protobuf/config/cmake_constants.h"
#include "nrp_protobuf/proto_python_bindings/proto_field_ops.h"
#include "nrp_protobuf/proto_ops/protobuf_ops.h"
#include "nrp_protobuf/proto_ops/proto_ops_manager.h"
 
#include "nrp_grpc_engine_protocol/engine_server/engine_proto_wrapper.h"
 
 
using EngineGrpc::EngineGrpcService;
 
class EngineGrpcServer : public EngineGrpcService::Service
{
    public:
 
        using mutex_t = std::timed_mutex;
        using lock_t  = std::unique_lock<EngineGrpcServer::mutex_t>;
 
        EngineGrpcServer() = delete;
 
        EngineGrpcServer(const std::string serverAddress, EngineProtoWrapper* engineWrapper)
                :  _engineWrapper(engineWrapper)
        {
            this->_serverAddress   = serverAddress;
            this->_isServerRunning = false;
 
            grpc::EnableDefaultHealthCheckService(true);
        }
 
        ~EngineGrpcServer()
        {
            this->shutdownServer();
        }
 
        void startServer()
        {
            if(!this->_isServerRunning)
            {
                grpc::ServerBuilder builder;
                builder.AddListeningPort(_serverAddress, grpc::InsecureServerCredentials());
                builder.RegisterService(this);
                NRPLogger::debug("Using server address: "+ this->_serverAddress);
 
                this->_server = builder.BuildAndStart();
                // TODO Should we use a memory barrier here?
                this->_isServerRunning = true;
            }
        }
 
        void startServerAsync()
        {
            this->startServer();
        }
 
        void shutdownServer()
        {
            if(this->_isServerRunning)
            {
                this->_server->Shutdown();
                // TODO Should we use a memory barrier here?
                this->_isServerRunning = false;
            }
        }
 
         virtual bool initRunFlag() const  { return this->_engineWrapper->initRunFlag(); }
 
        virtual bool shutdownFlag() const  { return this->_engineWrapper->shutdownFlag(); }
 
        bool isServerRunning() const
        {
            return this->_isServerRunning;
        }
 
        const std::string serverAddress() const
        {
            return this->_serverAddress;
        }
 
    private:
 
        mutex_t _engineCallLock;
 
        std::string _serverAddress;
 
        bool _isServerRunning;
 
        std::unique_ptr<grpc::Server> _server;
 
        std::unique_ptr<EngineProtoWrapper> _engineWrapper;
 
        grpc::Status initialize(      grpc::ServerContext           * /*context*/,
                                const EngineGrpc::InitializeRequest * request,
                                      EngineGrpc::InitializeReply   * /*reply*/) override
        {
            try
            {
                EngineGrpcServer::lock_t lock(this->_engineCallLock);
 
                nlohmann::json requestJson = nlohmann::json::parse(request->json());
 
                // Run engine-specific initialization function
 
                this->_engineWrapper->initialize(requestJson);
            }
            catch(const std::exception &e)
            {
                return handleGrpcError("Error while executing initialization", e.what());
            }
            NRPLogger::debug("init command returns OK");
            return grpc::Status::OK;
        }
 
 
        grpc::Status reset(      grpc::ServerContext      * /*context*/,
                           const EngineGrpc::ResetRequest * /*request*/,
                                 EngineGrpc::ResetReply   * /*reply*/) override
        {
            NRP_LOGGER_TRACE("{} called", __FUNCTION__);
            try
            {
                EngineGrpcServer::lock_t lock(this->_engineCallLock);
 
                // Run engine-specific reset function
                this->_engineWrapper->reset();
            }
            catch(const std::exception &e)
            {
                return handleGrpcError("Error while executing initialization", e.what());
            }
 
            NRPLogger::debug("reset command returns OK");
            return grpc::Status::OK;
        }
 
        grpc::Status shutdown(      grpc::ServerContext         * /*context*/,
                              const EngineGrpc::ShutdownRequest * request,
                                    EngineGrpc::ShutdownReply   * /*reply*/) override
        {
            try
            {
                EngineGrpcServer::lock_t lock(this->_engineCallLock);
 
                // TODO: we don't have any instance where json data is actually sent to an engine for shutdown,
                //  it can be removed from EngineGrpc::ShutdownRequest definition
                nlohmann::json requestJson = nlohmann::json::parse(request->json());
 
                // Run engine-specifi shutdown function
 
                this->_engineWrapper->shutdown();
            }
            catch(const std::exception &e)
            {
                return handleGrpcError("Error while executing shutdown", e.what());
            }
 
            NRPLogger::debug("shutdown command returns OK");
            return grpc::Status::OK;
        }
 
        grpc::Status runLoopStep(      grpc::ServerContext            * /*context*/,
                                 const EngineGrpc::RunLoopStepRequest * request,
                                       EngineGrpc::RunLoopStepReply   * reply) override
        {
            try
            {
                EngineGrpcServer::lock_t lock(this->_engineCallLock);
 
                int64_t engineTime = (this->_engineWrapper->runLoopStep(SimulationTime(request->timestep()))).count();
 
                reply->set_enginetime(engineTime);
            }
            catch(const std::exception &e)
            {
                return handleGrpcError("Error while executing runLoopStep", e.what());
            }
 
            return grpc::Status::OK;
        }
 
        grpc::Status setDataPacks(      grpc::ServerContext             * /*context*/,
                                  const EngineGrpc::SetDataPacksRequest * request,
                                        EngineGrpc::SetDataPacksReply   * /*reply*/) override
        {
            try
            {
                EngineGrpcServer::lock_t lock(this->_engineCallLock);
 
                this->_engineWrapper->setDataPacks(*request);
            }
            catch(const std::exception &e)
            {
                return handleGrpcError("Error while executing setDataPack", e.what());
            }
 
            return grpc::Status::OK;
        }
 
        grpc::Status getDataPacks(      grpc::ServerContext             * /*context*/,
                                  const EngineGrpc::GetDataPacksRequest * request,
                                        EngineGrpc::GetDataPacksReply   * reply) override
        {
            try
            {
                EngineGrpcServer::lock_t lock(this->_engineCallLock);
 
                this->_engineWrapper->getDataPacks(*request, reply);
            }
            catch(const std::exception &e)
            {
                return handleGrpcError("Error while executing getDataPack", e.what());
            }
 
            return grpc::Status::OK;
        }
 
        grpc::Status handleGrpcError(const std::string & contextMessage, const std::string & errorMessage)
        {
            NRPLogger::error("context message: [ {} ]", contextMessage);
            NRPLogger::error("error message: [ {} ]", errorMessage);
 
            // Pass the error message to the client
 
            grpc::Status status(grpc::StatusCode::CANCELLED, errorMessage);
 
            return status;
        }
};
 
#endif // ENGINE_GRPC_SERVER_H