CoreBroker.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/JsonBuilder.hpp"
#include "ActionMessage.hpp"
#include "BasicHandleInfo.hpp"
#include "Broker.hpp"
#include "BrokerBase.hpp"
#include "FederateIdExtra.hpp"
#include "HandleManager.hpp"
#include "TimeDependencies.hpp"
#include "UnknownHandleManager.hpp"
#include "gmlc/concurrency/DelayedObjects.hpp"
#include "gmlc/concurrency/TriggerVariable.hpp"
#include "gmlc/containers/AirLock.hpp"
#include "gmlc/containers/DualStringMappedVector.hpp"
#include "gmlc/containers/SimpleQueue.hpp"
 
#include <any>
#include <array>
#include <atomic>
#include <deque>
#include <functional>
#include <map>
#include <memory>
#include <string>
#include <thread>
#include <tuple>
#include <unordered_map>
#include <utility>
#include <vector>
 
namespace helics {
 
enum class ConnectionState : std::uint8_t {
    CONNECTED = 0,
    INIT_REQUESTED = 1,
    OPERATING = 2,
    ERROR_STATE = 40,
    REQUEST_DISCONNECT = 48,
    DISCONNECTED = 50
};
 
enum class QueryReuse : std::uint8_t;
 
struct BasicFedInfo {
    const std::string name;  
    GlobalFederateId global_id;  
    route_id route;  
    GlobalBrokerId parent;  
    ConnectionState state{ConnectionState::CONNECTED};
    bool nonCounting{false};  
    bool observer{false};  
    bool dynamic{false};  
    bool reentrant{false};  
    explicit BasicFedInfo(std::string_view fedname): name(fedname) {}
};
 
class BasicBrokerInfo {
  public:
    const std::string name;  
 
    GlobalBrokerId global_id;  
    route_id route;  
    GlobalBrokerId parent;  
    ConnectionState state{ConnectionState::CONNECTED};
    bool _hasTimeDependency{false};
    bool _core{false};  
    bool _nonLocal{false};  
    bool _route_key{false};  
    bool _sent_disconnect_ack{false};  
    bool _disable_ping{false};  
    bool _observer{false};  
    bool initIterating{false};  
    std::string routeInfo;  
    explicit BasicBrokerInfo(std::string_view brokerName): name(brokerName) {}
};
 
class TimeCoordinator;
class Logger;
class TimeoutMonitor;
 
class CoreBroker: public Broker, public BrokerBase {
  protected:
    bool _gateway = false;  
  private:
    std::atomic<bool> _isRoot{false};  
    bool isRootc{false};
    bool connectionEstablished{false};  
    bool initIterating{false};  
    int routeCount = 1;  
    gmlc::containers::
        DualStringMappedVector<BasicFedInfo, GlobalFederateId, reference_stability::unstable>
            mFederates;
    gmlc::containers::
        DualStringMappedVector<BasicBrokerInfo, GlobalBrokerId, reference_stability::unstable>
            mBrokers;
    std::string mPreviousLocalBrokerIdentifier;
 
    HandleManager handles;  
    UnknownHandleManager unknownHandles;  
    std::vector<std::pair<std::string, GlobalFederateId>> delayedDependencies;
    std::unordered_map<GlobalFederateId, LocalFederateId> global_id_translation;
    std::unordered_map<GlobalFederateId, route_id> routing_table;
    std::unordered_map<std::string, route_id> knownExternalEndpoints;
    std::unordered_map<std::string, std::string> global_values;  
    std::unordered_map<std::string, std::int64_t> renamers;  
    std::mutex name_mutex_;  
    std::atomic<int> queryCounter{1};  // counter for active queries going to the local API
    bool force_connection{false};
    gmlc::concurrency::DelayedObjects<std::string> activeQueries;  
    std::vector<std::tuple<fileops::JsonMapBuilder, std::vector<ActionMessage>, QueryReuse>>
        mapBuilders;
    std::deque<std::pair<int32_t, decltype(std::chrono::steady_clock::now())>> queryTimeouts;
 
    std::vector<ActionMessage> earlyMessages;  
    gmlc::concurrency::TriggerVariable disconnection;  
    std::unique_ptr<TimeoutMonitor> timeoutMon;
    std::atomic<uint16_t> nextAirLock{0};  
    std::array<gmlc::containers::AirLock<std::any>, 3> dataAirlocks;
    // variables for a time logging federate
    std::string
        mTimeMonitorFederate;  
    GlobalFederateId mTimeMonitorFederateId{};  
    GlobalFederateId mTimeMonitorLocalFederateId{};
    Time mTimeMonitorPeriod{timeZero};  
    Time mTimeMonitorLastLogTime{Time::minVal()};  
    Time mTimeMonitorCurrentTime{Time::minVal()};  
    std::atomic<double> simTime{mInvalidSimulationTime};  
    Time mNextTimeBarrier{Time::maxVal()};  
  private:
    virtual void processCommand(ActionMessage&& command) override;
    void processPriorityCommand(ActionMessage&& command) override;
 
    void processBrokerConfigureCommands(ActionMessage& cmd);
    gmlc::containers::SimpleQueue<ActionMessage> delayTransmitQueue;
    /* function to transmit the delayed messages*/
    void transmitDelayedMessages();
    void routeMessage(ActionMessage& cmd, GlobalFederateId dest);
    void routeMessage(ActionMessage&& cmd, GlobalFederateId dest);
    void routeMessage(const ActionMessage& cmd);
    void routeMessage(ActionMessage&& cmd);
    void transmitToParent(ActionMessage&& cmd);
    void propagateError(ActionMessage&& cmd);
    void broadcast(ActionMessage& cmd);
    
    route_id fillMessageRouteInformation(ActionMessage& mess);
 
    void executeInitializationOperations(bool iterating);
    uint16_t getNextAirlockIndex();
    bool verifyBrokerKey(ActionMessage& mess) const;
    bool verifyBrokerKey(std::string_view key) const;
 
  public:
    virtual bool connect() override final;
    virtual void disconnect() override final;
    void unregister();
    virtual void processDisconnect(bool skipUnregister = false) override final;
    virtual bool isConnected() const override final;
    virtual void setAsRoot() override final;
    virtual bool isRoot() const override final { return _isRoot; };
 
    virtual bool isOpenToNewFederates() const override;
 
    virtual void setLoggingCallback(
        std::function<void(int, std::string_view, std::string_view)> logFunction) override final;
 
    virtual bool waitForDisconnect(
        std::chrono::milliseconds msToWait = std::chrono::milliseconds(0)) const override final;
 
    virtual void setTimeBarrier(Time barrierTime) override final;
 
    virtual void clearTimeBarrier() override final;
 
    virtual void globalError(int32_t errorCode, std::string_view errorString) override final;
 
  private:
    virtual bool brokerConnect() = 0;
    virtual void brokerDisconnect() = 0;
 
  protected:
    virtual void transmit(route_id route, const ActionMessage& command) = 0;
    virtual void transmit(route_id route, ActionMessage&& command) = 0;
    virtual void addRoute(route_id rid, int interfaceId, std::string_view routeInfo) = 0;
    virtual void removeRoute(route_id rid) = 0;
 
  public:
    explicit CoreBroker(bool setAsRootBroker = false) noexcept;
    explicit CoreBroker(std::string_view broker_name);
    virtual ~CoreBroker();
    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;
 
    bool allInitReady() const;
    ConnectionState getAllConnectionState() const;
 
    void setIdentifier(std::string_view name);
    virtual const std::string& getIdentifier() const override final { return identifier; }
    virtual const std::string& getAddress() const override final;
    virtual void setLoggingLevel(int logLevel) 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 = HELICS_SEQUENCING_MODE_FAST) override final;
    virtual void setGlobal(std::string_view valueName, std::string_view value) override final;
    virtual void sendCommand(std::string_view target,
                             std::string_view commandStr,
                             HelicsSequencingModes mode) override final;
    virtual void makeConnections(const std::string& file) override final;
    virtual void linkEndpoints(std::string_view source, std::string_view target) override final;
    virtual void dataLink(std::string_view publication, std::string_view input) 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 addAlias(std::string_view interfaceKey, std::string_view alias) override final;
 
  protected:
    virtual std::shared_ptr<helicsCLI11App> generateCLI() override;
 
    virtual double getSimulationTime() const override;
 
  private:
    int getCountableFederates() const;
    void checkDependencies();
 
    void connectInterfaces(
        const BasicHandleInfo& origin,
        uint32_t originFlags,
        const BasicHandleInfo& target,
        uint32_t targetFlags,
        std::pair<action_message_def::action_t, action_message_def::action_t> actions);
 
    void findAndNotifyInputTargets(BasicHandleInfo& handleInfo, const std::string& key);
    void findAndNotifyPublicationTargets(BasicHandleInfo& handleInfo, const std::string& key);
 
    void findAndNotifyFilterTargets(BasicHandleInfo& handleInfo, const std::string& key);
    void findAndNotifyEndpointTargets(BasicHandleInfo& handleInfo, const std::string& key);
 
    void findRegexMatch(const std::string& target,
                        InterfaceType type,
                        GlobalHandle handle,
                        uint16_t flags);
    void processDisconnectCommand(ActionMessage& command);
    void disconnectTiming(ActionMessage& command);
    void processBrokerDisconnect(ActionMessage& command, BasicBrokerInfo* brk);
    void processError(ActionMessage& command);
    void disconnectBroker(BasicBrokerInfo& brk);
    void markAsDisconnected(GlobalBrokerId brkid);
    void checkInFlightQueries(GlobalBrokerId brkid);
    void checkForNamedInterface(ActionMessage& command);
    void removeNamedTarget(ActionMessage& command);
    void processQueryCommand(ActionMessage& cmd);
    void processQuery(ActionMessage& message);
    void processInitCommand(ActionMessage& cmd);
    void checkQueryTimeouts();
    void processQueryResponse(const ActionMessage& message);
    void processLocalQuery(const ActionMessage& message);
    std::string generateQueryAnswer(std::string_view request, bool force_ordering);
    std::string quickBrokerQueries(std::string_view request) const;
    void processCommandInstruction(ActionMessage& message);
    void processLocalCommandInstruction(ActionMessage& message);
    std::string getNameList(std::string_view gidString) const;
    route_id getRoute(GlobalFederateId fedid) const;
    route_id getRoute(int32_t fedid) const { return getRoute(GlobalFederateId(fedid)); }
 
    const BasicBrokerInfo* getBrokerById(GlobalBrokerId brokerid) const;
 
    BasicBrokerInfo* getBrokerById(GlobalBrokerId brokerid);
 
    void addLocalInfo(BasicHandleInfo& handleInfo, const ActionMessage& message);
    void addPublication(ActionMessage& message);
    void addInput(ActionMessage& message);
    void addEndpoint(ActionMessage& message);
    void addFilter(ActionMessage& message);
    void addTranslator(ActionMessage& message);
    void addDataSink(ActionMessage& message);
 
    bool checkInterfaceCreation(ActionMessage& message, InterfaceType type);
    // Handle the registration of new brokers
    void brokerRegistration(ActionMessage&& command);
    void sendBrokerErrorAck(ActionMessage& command, std::int32_t errorCode);
    // Helper function for linking interfaces
    void linkInterfaces(ActionMessage& command);
    // Handle the registration of new federates
    void fedRegistration(ActionMessage&& command);
    void sendFedErrorAck(ActionMessage& command, std::int32_t errorCode);
    //   bool updateSourceFilterOperator (ActionMessage &m);
    void initializeMapBuilder(std::string_view request,
                              std::uint16_t index,
                              QueryReuse reuse,
                              bool force_ordering);
 
    std::string generateGlobalStatus(fileops::JsonMapBuilder& builder);
    void sendErrorToImmediateBrokers(int errorCode);
    void sendDisconnect(action_message_def::action_t disconnectType);
    std::string generateFederationSummary() const;
    void labelAsDisconnected(GlobalBrokerId brkid);
 
    void processTimeMonitorMessage(ActionMessage& message);
    void loadTimeMonitor(bool firstLoad, std::string_view newFederate);
    void generateTimeBarrier(ActionMessage& message);
    int generateMapObjectCounter() const;
    std::string generateRename(std::string_view name);
    friend class TimeoutMonitor;
};
 
}  // namespace helics