BrokerBase.hpp

Go to the documentation of this file.

/*
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 "ActionMessage.hpp"
#include "FederateIdExtra.hpp"
#include "gmlc/containers/BlockingPriorityQueue.hpp"
 
#include <atomic>
#include <limits>
#include <memory>
#include <string>
#include <thread>
#include <utility>
#include <vector>
 
namespace spdlog {
class logger;
}
namespace Json {
class Value;
}
 
namespace helics {
class BaseTimeCoordinator;
class helicsCLI11App;
class ProfilerBuffer;
class LogBuffer;
class LogManager;
class BrokerBase {
  protected:
    static constexpr double mInvalidSimulationTime{-98763.2};
    std::atomic<GlobalBrokerId> global_id{parent_broker_id};
    GlobalBrokerId global_broker_id_local{};  
    GlobalBrokerId higher_broker_id{0};  
    std::atomic<int32_t> maxLogLevel{HELICS_LOG_LEVEL_NO_PRINT};
 
    int32_t minFederateCount{1};
    int32_t minBrokerCount{0};
    int32_t maxFederateCount{(std::numeric_limits<int32_t>::max)()};
    int32_t maxBrokerCount{(std::numeric_limits<int32_t>::max)()};
    int32_t minChildCount{0};
    int32_t maxIterationCount{10000};  
    Time tickTimer{5.0};  
    Time timeout{30.0};  
    Time networkTimeout{-1.0};  
    Time queryTimeout{15.0};  
    Time errorDelay{0.0};  
    Time grantTimeout{-1.0};  
    Time maxCoSimDuration{-1.0};  
    std::string identifier;  
    std::string brokerKey;  
    // address is mutable since during initial phases it may not be fixed so to maintain a
    // consistent public interface for extracting it this variable may need to be updated in a
    // constant function
    mutable std::string address;  
 
    std::thread queueProcessingThread;  
    std::atomic<bool> haltOperations{false};
    bool restrictive_time_policy{false};
    bool terminate_on_error{false};
    bool debugging{false};
    bool observer{false};
    bool globalTime{false};
    bool asyncTime{false};
    bool dynamicFederation{false};
    bool disableDynamicSources{false};
 
  private:
    std::atomic<bool> mainLoopIsRunning{false};
    bool dumplog{false};
    bool queueDisabled{false};
    bool disable_timer{false};
    std::atomic<std::size_t> messageCounter{0};
 
  protected:
    std::unique_ptr<BaseTimeCoordinator> timeCoord;  
    gmlc::containers::BlockingPriorityQueue<ActionMessage> actionQueue;  
    std::shared_ptr<LogManager> mLogManager;  
    enum class BrokerState : int16_t {
        CREATED = -10,  
        CONFIGURING = -7,  
        CONFIGURED = -6,  
        CONNECTING = -4,  
        CONNECTED = -3,  
        INITIALIZING = -1,  
        OPERATING = 0,  
        CONNECTED_ERROR = 3,  
        TERMINATING = 4,  
        TERMINATING_ERROR = 5,  
        TERMINATED = 6,  
        ERRORED = 7,  
    };
 
    enum class TickForwardingReasons : uint32_t {
        NONE = 0,
        NO_COMMS = 0x01,
        PING_RESPONSE = 0x02,
        QUERY_TIMEOUT = 0x04,
        GRANT_TIMEOUT = 0x08,
        DISCONNECT_TIMEOUT = 0x10
    };
    bool noAutomaticID{false};  
    bool hasTimeDependency{false};  
    bool enteredExecutionMode{false};
    bool waitingForBrokerPingReply{false};  
    bool hasFilters{false};  
 
    bool no_ping{false};  
    bool uuid_like{false};  
    bool useJsonSerialization{false};
    bool enable_profiling{false};  
    bool allowRemoteControl{true};  
    bool errorOnUnmatchedConnections{false};
    decltype(std::chrono::steady_clock::now()) errorTimeStart;
    decltype(std::chrono::steady_clock::now()) disconnectTime;
    std::atomic<int> lastErrorCode{0};  
    std::string lastErrorString;  
  private:
    std::shared_ptr<ProfilerBuffer> prBuff;
 
    bool forwardTick{false};
    uint32_t forwardingReasons{0U};
    std::atomic<BrokerState> brokerState{BrokerState::CREATED};
 
  public:
    explicit BrokerBase(bool DisableQueue = false) noexcept;
    explicit BrokerBase(std::string_view broker_name, bool DisableQueue = false);
 
    virtual ~BrokerBase();
    int parseArgs(int argc, char* argv[]);
    int parseArgs(std::vector<std::string> args);
    int parseArgs(std::string_view initializationString);
    virtual void configureBase();
 
    void addActionMessage(const ActionMessage& message);
    void addActionMessage(ActionMessage&& message);
 
    void setLoggerFunction(
        std::function<void(int level, std::string_view identifier, std::string_view message)>
            logFunction);
    void logFlush();
    bool isRunning() const { return mainLoopIsRunning.load(); }
    void setLogLevel(int32_t level);
    void setLogLevels(int32_t consoleLevel, int32_t fileLevel);
    GlobalBrokerId getGlobalId() const { return global_id.load(); }
 
  private:
    void queueProcessingLoop();
    action_message_def::action_t commandProcessor(ActionMessage& command);
 
    std::shared_ptr<helicsCLI11App> generateBaseCLI();
    void baseConfigure(ActionMessage& command);
 
    void addActionMessage(ActionMessage&& message) const;
 
  protected:
    static bool isReasonForTick(std::uint32_t code, TickForwardingReasons reason)
    {
        return ((static_cast<std::uint32_t>(reason) & code) != 0);
    }
    void setTickForwarding(TickForwardingReasons reason, bool value = true);
    BrokerState getBrokerState() const { return brokerState.load(); }
    bool setBrokerState(BrokerState newState);
    bool transitionBrokerState(BrokerState expectedState, BrokerState newState);
    virtual void processDisconnect(bool skipUnregister = false) = 0;
    virtual bool tryReconnect() = 0;
    virtual void processCommand(ActionMessage&& cmd) = 0;
    virtual void processPriorityCommand(ActionMessage&& command) = 0;
 
    bool sendToLogger(GlobalFederateId federateID,
                      int logLevel,
                      std::string_view name,
                      std::string_view message,
                      bool fromRemote = false) const;
    void saveProfilingData(std::string_view message);
    void writeProfilingData();
    void generateNewIdentifier();
    virtual std::string generateLocalAddressString() const = 0;
    virtual std::shared_ptr<helicsCLI11App> generateCLI();
    void setErrorState(int eCode, std::string_view estring);
    void setLoggingFile(std::string_view lfile);
    bool getFlagValue(int32_t flag) const;
    virtual double getSimulationTime() const { return mInvalidSimulationTime; }
    std::pair<bool, std::vector<std::string_view>> processBaseCommands(ActionMessage& command);
    void addBaseInformation(Json::Value& base, bool hasParent) const;
 
  public:
    std::function<void(int, std::string_view, std::string_view)> getLoggingCallback() const;
    void joinAllThreads();
    std::size_t currentMessageCounter() const
    {
        return messageCounter.load(std::memory_order_acquire);
    }
    friend class TimeoutMonitor;
    friend const std::string& brokerStateName(BrokerState state);
};
 
const std::string& brokerStateName(BrokerBase::BrokerState state);
 
}  // namespace helics