CommonCore.hpp

/*
Copyright (c) 2017-2024,
Battelle Memorial Institute; Lawrence Livermore National Security, LLC; Alliance for Sustainable
Energy, LLC.  See the top-level NOTICE for additional details. All rights reserved.
SPDX-License-Identifier: BSD-3-Clause
*/
#pragma once
 
#include "../common/GuardedTypes.hpp"
#include "../common/JsonBuilder.hpp"
#include "ActionMessage.hpp"
#include "BrokerBase.hpp"
#include "Core.hpp"
#include "FederateIdExtra.hpp"
#include "HandleManager.hpp"
#include "gmlc/concurrency/DelayedObjects.hpp"
#include "gmlc/concurrency/TriggerVariable.hpp"
#include "gmlc/containers/AirLock.hpp"
#include "gmlc/containers/DualStringMappedVector.hpp"
#include "gmlc/containers/MappedPointerVector.hpp"
#include "gmlc/containers/SimpleQueue.hpp"
#include "helics/helics-config.h"
#include "helicsTime.hpp"
 
#include "json/forwards.h"
#include <any>
#include <array>
#include <atomic>
#include <chrono>
#include <deque>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <tuple>
#include <unordered_map>
#include <utility>
#include <vector>
 
namespace helics {
class TestHandle;
class FederateState;
 
class BasicHandleInfo;
class FilterCoordinator;
class FilterInfo;
class FilterFederate;
class TranslatorFederate;
class TimeoutMonitor;
enum class InterfaceType : char;
enum class QueryReuse : std::uint8_t;
enum class OperatingState : std::uint8_t { OPERATING = 0, ERROR_STATE = 5, DISCONNECTED = 10 };
 
const std::string& stateString(OperatingState state);
 
class FedInfo {
  public:
    FederateState* fed = nullptr;
    OperatingState state{OperatingState::OPERATING};
 
    constexpr FedInfo() = default;
    constexpr explicit FedInfo(FederateState* newfed) noexcept: fed(newfed) {}
    FederateState* operator->() noexcept { return fed; }
    const FederateState* operator->() const noexcept { return fed; }
    operator bool() const noexcept { return (fed != nullptr); }
};
 
class CommonCore: public Core, public BrokerBase {
  public:
    CommonCore() noexcept;
    explicit CommonCore(bool arg) noexcept;
    explicit CommonCore(std::string_view coreName);
    virtual ~CommonCore() override;
    virtual void configure(std::string_view configureString) override final;
    virtual void configureFromArgs(int argc, char* argv[]) override final;
    virtual void configureFromVector(std::vector<std::string> args) override final;
    virtual bool isConfigured() const override final;
    virtual bool isOpenToNewFederates() const override final;
    virtual bool hasError() const override final;
    virtual void globalError(LocalFederateId federateID,
                             int errorCode,
                             std::string_view errorString) override final;
    virtual void localError(LocalFederateId federateID,
                            int errorCode,
                            std::string_view errorString) override final;
    virtual int getErrorCode() const override final;
    virtual std::string getErrorMessage() const override final;
    virtual void finalize(LocalFederateId federateID) override final;
    virtual bool enterInitializingMode(LocalFederateId federateID,
                                       IterationRequest request) override final;
    virtual void setCoreReadyToInit() override final;
    virtual iteration_time
        enterExecutingMode(LocalFederateId federateID,
                           IterationRequest iterate = NO_ITERATION) override final;
    virtual LocalFederateId registerFederate(std::string_view name,
                                             const CoreFederateInfo& info) override final;
    virtual const std::string& getFederateName(LocalFederateId federateID) const override final;
    virtual LocalFederateId getFederateId(std::string_view name) const override final;
    virtual int32_t getFederationSize() override final;
    virtual Time timeRequest(LocalFederateId federateID, Time next) override final;
    virtual iteration_time requestTimeIterative(LocalFederateId federateID,
                                                Time next,
                                                IterationRequest iterate) override final;
    virtual void processCommunications(LocalFederateId federateID,
                                       std::chrono::milliseconds msToWait) override final;
    virtual Time getCurrentTime(LocalFederateId federateID) const override final;
    virtual void
        setTimeProperty(LocalFederateId federateID, int32_t property, Time time) override final;
    virtual void setIntegerProperty(LocalFederateId federateID,
                                    int32_t property,
                                    int16_t propertyValue) override final;
    virtual Time getTimeProperty(LocalFederateId federateID, int32_t property) const override final;
    virtual int16_t getIntegerProperty(LocalFederateId federateID,
                                       int32_t property) const override final;
    virtual void setFlagOption(LocalFederateId federateID,
                               int32_t flag,
                               bool flagValue = true) override final;
    virtual bool getFlagOption(LocalFederateId federateID, int32_t flag) const override final;
 
    virtual InterfaceHandle registerPublication(LocalFederateId federateID,
                                                std::string_view key,
                                                std::string_view type,
                                                std::string_view units) override final;
    virtual InterfaceHandle getPublication(LocalFederateId federateID,
                                           std::string_view key) const override final;
    virtual InterfaceHandle registerInput(LocalFederateId federateID,
                                          std::string_view key,
                                          std::string_view type,
                                          std::string_view units) override final;
 
    virtual InterfaceHandle getInput(LocalFederateId federateID,
                                     std::string_view key) const override final;
 
    virtual const std::string& getHandleName(InterfaceHandle handle) const override final;
 
    virtual void setHandleOption(InterfaceHandle handle,
                                 int32_t option,
                                 int32_t option_value) override final;
 
    virtual int32_t getHandleOption(InterfaceHandle handle, int32_t option) const override final;
    virtual void closeHandle(InterfaceHandle handle) override final;
    virtual void removeTarget(InterfaceHandle handle,
                              std::string_view targetToRemove) override final;
    virtual void addDestinationTarget(InterfaceHandle handle,
                                      std::string_view dest,
                                      InterfaceType hint) override final;
    virtual void addSourceTarget(InterfaceHandle handle,
                                 std::string_view name,
                                 InterfaceType hint) override final;
    virtual const std::string& getDestinationTargets(InterfaceHandle handle) const override final;
 
    virtual const std::string& getSourceTargets(InterfaceHandle handle) const override final;
    virtual const std::string& getInjectionUnits(InterfaceHandle handle) const override final;
    virtual const std::string& getExtractionUnits(InterfaceHandle handle) const override final;
    virtual const std::string& getInjectionType(InterfaceHandle handle) const override final;
    virtual const std::string& getExtractionType(InterfaceHandle handle) const override final;
    virtual void setValue(InterfaceHandle handle, const char* data, uint64_t len) override final;
    virtual const std::shared_ptr<const SmallBuffer>& getValue(InterfaceHandle handle,
                                                               uint32_t* inputIndex) override final;
    virtual const std::vector<std::shared_ptr<const SmallBuffer>>&
        getAllValues(InterfaceHandle handle) override final;
    virtual const std::vector<InterfaceHandle>&
        getValueUpdates(LocalFederateId federateID) override final;
    virtual InterfaceHandle registerEndpoint(LocalFederateId federateID,
                                             std::string_view name,
                                             std::string_view type) override final;
 
    virtual InterfaceHandle registerTargetedEndpoint(LocalFederateId federateID,
                                                     std::string_view name,
                                                     std::string_view type) override final;
    virtual InterfaceHandle getEndpoint(LocalFederateId federateID,
                                        std::string_view name) const override final;
 
    virtual InterfaceHandle registerDataSink(LocalFederateId federateID,
                                             std::string_view name) override final;
 
    virtual InterfaceHandle getDataSink(LocalFederateId federateID,
                                        std::string_view name) const override final;
 
    virtual InterfaceHandle registerFilter(std::string_view filterName,
                                           std::string_view type_in,
                                           std::string_view type_out) override final;
    virtual InterfaceHandle registerCloningFilter(std::string_view filterName,
                                                  std::string_view type_in,
                                                  std::string_view type_out) override final;
    virtual InterfaceHandle registerTranslator(std::string_view translatorName,
                                               std::string_view endpointType,
                                               std::string_view units) override final;
    virtual InterfaceHandle getFilter(std::string_view name) const override final;
    virtual InterfaceHandle getTranslator(std::string_view name) const override final;
    virtual void addDependency(LocalFederateId federateID,
                               std::string_view federateName) override final;
    virtual void linkEndpoints(std::string_view source, std::string_view dest) override final;
    virtual void addAlias(std::string_view interfaceKey, std::string_view alias) override final;
    virtual void makeConnections(const std::string& file) override final;
    virtual void dataLink(std::string_view source, std::string_view target) override final;
    virtual void addSourceFilterToEndpoint(std::string_view filter,
                                           std::string_view endpoint) override final;
    virtual void addDestinationFilterToEndpoint(std::string_view filter,
                                                std::string_view endpoint) override final;
    virtual void
        send(InterfaceHandle sourceHandle, const void* data, uint64_t length) override final;
    virtual void sendAt(InterfaceHandle sourceHandle,
                        const void* data,
                        uint64_t length,
                        Time time) override final;
    virtual void sendTo(InterfaceHandle sourceHandle,
                        const void* data,
                        uint64_t length,
                        std::string_view destination) override final;
    virtual void sendToAt(InterfaceHandle sourceHandle,
                          const void* data,
                          uint64_t length,
                          std::string_view destination,
                          Time time) override final;
    virtual void sendMessage(InterfaceHandle sourceHandle,
                             std::unique_ptr<Message> message) override final;
    virtual uint64_t receiveCount(InterfaceHandle destination) override final;
    virtual std::unique_ptr<Message> receive(InterfaceHandle destination) override final;
    virtual std::unique_ptr<Message> receiveAny(LocalFederateId federateID,
                                                InterfaceHandle& endpoint_id) override final;
    virtual uint64_t receiveCountAny(LocalFederateId federateID) override final;
    virtual void logMessage(LocalFederateId federateID,
                            int logLevel,
                            std::string_view messageToLog) override final;
    virtual void setFilterOperator(InterfaceHandle filter,
                                   std::shared_ptr<FilterOperator> callback) override final;
    virtual void
        setTranslatorOperator(InterfaceHandle translator,
                              std::shared_ptr<TranslatorOperator> callbacks) override final;
    virtual void setFederateOperator(LocalFederateId federateID,
                                     std::shared_ptr<FederateOperator> callback) override;
    void setIdentifier(std::string_view name);
    virtual const std::string& getIdentifier() const override final;
    virtual const std::string& getAddress() const override final;
    const std::string& getFederateNameNoThrow(GlobalFederateId federateID) const noexcept;
 
    virtual void setLoggingLevel(int logLevel) override;
    virtual void setLoggingCallback(
        LocalFederateId federateID,
        std::function<void(int, std::string_view, std::string_view)> logFunction) override final;
 
    virtual void setLogFile(std::string_view lfile) override final;
 
    virtual std::string query(std::string_view target,
                              std::string_view queryStr,
                              HelicsSequencingModes mode) override;
    virtual void setQueryCallback(LocalFederateId federateID,
                                  std::function<std::string(std::string_view)> queryFunction,
                                  int order) override;
    virtual void setGlobal(std::string_view valueName, std::string_view value) override;
    virtual void sendCommand(std::string_view target,
                             std::string_view commandStr,
                             std::string_view source,
                             HelicsSequencingModes mode) override;
    virtual std::pair<std::string, std::string> getCommand(LocalFederateId federateID) override;
 
    virtual std::pair<std::string, std::string> waitCommand(LocalFederateId federateID) override;
 
    virtual bool connect() override final;
    virtual bool isConnected() const override final;
    virtual void disconnect() override final;
    virtual bool waitForDisconnect(
        std::chrono::milliseconds msToWait = std::chrono::milliseconds(0)) const override final;
    void unregister();
    virtual void processDisconnect(bool skipUnregister = false) override final;
 
    void checkInFlightQueriesForDisconnect();
 
    virtual void setInterfaceInfo(InterfaceHandle handle, std::string_view info) override final;
    virtual const std::string& getInterfaceInfo(InterfaceHandle handle) const override final;
 
    virtual void setInterfaceTag(InterfaceHandle handle,
                                 std::string_view tag,
                                 std::string_view value) override final;
    virtual const std::string& getInterfaceTag(InterfaceHandle handle,
                                               std::string_view tag) const override final;
 
    virtual void setFederateTag(LocalFederateId fid,
                                std::string_view tag,
                                std::string_view value) override final;
    virtual const std::string& getFederateTag(LocalFederateId fid,
                                              std::string_view tag) const override final;
 
  private:
    virtual bool brokerConnect() = 0;
    virtual void brokerDisconnect() = 0;
 
  protected:
    virtual void processCommand(ActionMessage&& command) override final;
 
    virtual void processPriorityCommand(ActionMessage&& command) override final;
 
    virtual void transmit(route_id rid, const ActionMessage& command) = 0;
    virtual void transmit(route_id rid, ActionMessage&& command) = 0;
    virtual void addRoute(route_id rid, int interfaceId, std::string_view routeInfo) = 0;
    virtual void removeRoute(route_id rid) = 0;
    FederateState* getFederateAt(LocalFederateId federateID) const;
    FederateState* getFederate(std::string_view federateName) const;
    FederateState* getHandleFederate(InterfaceHandle handle);
    const BasicHandleInfo* getHandleInfo(InterfaceHandle handle) const;
    const BasicHandleInfo* getLocalEndpoint(std::string_view name) const;
 
    bool allInitReady() const;
    bool allDisconnected() const;
    OperatingState minFederateState() const;
 
    virtual double getSimulationTime() const override;
 
  private:
    FederateState* getFederateCore(GlobalFederateId federateID);
    FederateState* getFederateCore(std::string_view federateName);
    FederateState* getHandleFederateCore(InterfaceHandle handle);
 
  private:
    std::atomic<double> simTime{BrokerBase::mInvalidSimulationTime};
    GlobalFederateId keyFed{};
    std::string prevIdentifier;  
    std::map<GlobalFederateId, route_id> routing_table;
    gmlc::containers::SimpleQueue<ActionMessage> delayTransmitQueue;
    std::unordered_map<std::string, route_id> knownExternalEndpoints;
    std::vector<std::pair<std::string, std::string>> tags;  
    std::unique_ptr<TimeoutMonitor> timeoutMon;
    void transmitDelayedMessages();
    void errorRespondDelayedMessages(std::string_view estring);
    void transmitDelayedMessages(GlobalFederateId source);
 
    void routeMessage(ActionMessage& cmd, GlobalFederateId dest);
    void routeMessage(const ActionMessage& cmd);
 
    void routeMessage(ActionMessage&& cmd, GlobalFederateId dest);
    void routeMessage(ActionMessage&& cmd);
    FederateState*
        checkNewInterface(LocalFederateId federateID, std::string_view key, InterfaceType type);
    void checkDependencies();
    void processQueryResponse(const ActionMessage& message);
    void checkQueryTimeouts();
    void processCommandsForCore(const ActionMessage& cmd);
    void processCoreConfigureCommands(ActionMessage& cmd);
    void processInitRequest(ActionMessage& cmd);
    void processExecRequest(ActionMessage& cmd);
    void processDisconnectCommand(ActionMessage& cmd);
 
    void processTimingTick(ActionMessage& cmd);
    void processQueryCommand(ActionMessage& cmd);
    void processLogAndErrorCommand(ActionMessage& cmd);
    void processLinkingCommand(ActionMessage& cmd);
    bool checkForLocalPublication(ActionMessage& cmd);
    uint16_t getNextAirlockIndex();
    void loadBasicJsonInfo(
        Json::Value& base,
        const std::function<void(Json::Value& fedval, const FedInfo& fed)>& fedLoader) const;
    void initializeMapBuilder(std::string_view request,
                              std::uint16_t index,
                              QueryReuse reuse,
                              bool force_ordering) const;
    std::string coreQuery(std::string_view queryStr, bool force_ordering) const;
 
    std::string quickCoreQueries(std::string_view queryStr) const;
 
    std::string filteredEndpointQuery(const FederateState* fed) const;
    void processCommandInstruction(ActionMessage& command);
 
  private:
    int32_t mGlobalFederationSize{0};  
    std::atomic<int16_t> delayInitCounter{0};
    bool filterTiming{false};  
    shared_guarded<gmlc::containers::MappedPointerVector<FederateState, std::string>> federates;
    gmlc::containers::DualStringMappedVector<FedInfo, GlobalFederateId> loopFederates;
 
    std::atomic<int32_t> messageCounter{54};
    ordered_guarded<HandleManager> handles;  
    HandleManager loopHandles;
    std::vector<std::pair<GlobalFederateId, int32_t>> timeBlocks;
    TranslatorFederate* translatorFed{nullptr};
    std::atomic<std::thread::id> translatorThread{std::thread::id{}};
    std::atomic<GlobalFederateId> translatorFedID;
 
    std::map<int32_t, std::vector<ActionMessage>> delayedTimingMessages;
 
    std::atomic<int> queryCounter{1};
    gmlc::concurrency::DelayedObjects<std::string> activeQueries;
    std::deque<std::pair<int32_t, decltype(std::chrono::steady_clock::now())>> queryTimeouts;
    mutable std::vector<std::tuple<fileops::JsonMapBuilder, std::vector<ActionMessage>, QueryReuse>>
        mapBuilders;
 
    FilterFederate* filterFed{nullptr};
    std::atomic<std::thread::id> filterThread{std::thread::id{}};
    std::atomic<GlobalFederateId> filterFedID;
    std::atomic<uint16_t> nextAirLock{0};  
    std::array<gmlc::containers::AirLock<std::any>, 4> dataAirlocks;
    gmlc::concurrency::TriggerVariable disconnection;  
    std::atomic<bool> initIterations{false};
 
  private:
    // generate a filter Federate
    void generateFilterFederate();
    // generate a translator Federate
    void generateTranslatorFederate();
    // generate a timing connection between the core and filter Federate
    void connectFilterTiming();
    bool hasTimeBlock(GlobalFederateId federateID);
    bool waitCoreRegistration();
    void generateMessages(ActionMessage& message,
                          const std::vector<std::pair<GlobalHandle, std::string_view>>& targets);
    void deliverMessage(ActionMessage& message);
    ActionMessage& processMessage(ActionMessage& message);
    const BasicHandleInfo& createBasicHandle(GlobalFederateId global_federateId,
                                             LocalFederateId local_federateId,
                                             InterfaceType HandleType,
                                             std::string_view key,
                                             std::string_view type,
                                             std::string_view units,
                                             uint16_t flags = 0);
 
    bool isLocal(GlobalFederateId global_fedid) const;
    route_id getRoute(GlobalFederateId global_fedid) const;
    void checkForNamedInterface(ActionMessage& command);
    void removeNamedTarget(ActionMessage& command);
    void setAsUsed(BasicHandleInfo* hand);
    void registerInterface(ActionMessage& command);
    void addTargetToInterface(ActionMessage& command);
    void removeTargetFromInterface(ActionMessage& command);
    void disconnectInterface(ActionMessage& command);
    void manageTimeBlocks(const ActionMessage& command);
 
    std::string federateQuery(const FederateState* fed,
                              std::string_view queryStr,
                              bool force_ordering) const;
 
    void sendErrorToFederates(int errorCode, std::string_view message);
    bool checkAndProcessDisconnect();
    void sendDisconnect(action_message_def::action_t disconnectType = CMD_STOP);
    void broadcastToFederates(ActionMessage& cmd);
    int generateMapObjectCounter() const;
    friend class TimeoutMonitor;
};
 
}  // namespace helics