| helics 3.6.1

BrokerBase.hpp
Go to the documentation of this file.
1 /*
2 Copyright (c) 2017-2025,
3 Battelle Memorial Institute; Lawrence Livermore National Security, LLC; Alliance for Sustainable
4 Energy, LLC. See the top-level NOTICE for additional details. All rights reserved.
5 SPDX-License-Identifier: BSD-3-Clause
6 */
7 #pragma once
14 #include "ActionMessage.hpp"
15 #include "FederateIdExtra.hpp"
16 #include "gmlc/containers/BlockingPriorityQueue.hpp"
17
18 #include <atomic>
19 #include <limits>
20 #include <memory>
21 #include <nlohmann/json_fwd.hpp>
22 #include <string>
23 #include <thread>
24 #include <utility>
25 #include <vector>
26
27 namespace spdlog {
28 class logger;
29 }
30
31 namespace helics {
32 class BaseTimeCoordinator;
33 class helicsCLI11App;
34 class ProfilerBuffer;
35 class LogBuffer;
36 class LogManager;
39 class BrokerBase {
40 protected:
41 static constexpr double mInvalidSimulationTime{-98763.2};
43 std::atomic<GlobalBrokerId> global_id{parent_broker_id};
44 GlobalBrokerId global_broker_id_local{};
46 GlobalBrokerId higher_broker_id{0};
48 std::atomic<int32_t> maxLogLevel{HELICS_LOG_LEVEL_NO_PRINT};
49
51 int32_t minFederateCount{1};
53 int32_t minBrokerCount{0};
54 int32_t maxFederateCount{(std::numeric_limits<int32_t>::max)()};
55 int32_t maxBrokerCount{(std::numeric_limits<int32_t>::max)()};
57 int32_t minChildCount{0};
58 int32_t maxIterationCount{10000};
59 Time tickTimer{5.0};
60 Time timeout{30.0};
61 Time networkTimeout{-1.0};
62 Time queryTimeout{15.0};
64 Time errorDelay{0.0};
65 Time grantTimeout{-1.0};
66 Time maxCoSimDuration{-1.0};
67 std::string identifier;
68 std::string brokerKey;
70 // address is mutable since during initial phases it may not be fixed so to maintain a
71 // consistent public interface for extracting it this variable may need to be updated in a
72 // constant function
73 mutable std::string address;
74
75 std::thread queueProcessingThread;
77 std::atomic<bool> haltOperations{false};
79 bool restrictive_time_policy{false};
81 bool terminate_on_error{false};
83 bool debugging{false};
85 bool observer{false};
87 bool globalTime{false};
89 bool asyncTime{false};
91 bool dynamicFederation{false};
93 bool disableDynamicSources{false};
94
95 private:
97 std::atomic<bool> mainLoopIsRunning{false};
99 bool dumplog{false};
102 bool queueDisabled{false};
104 bool disable_timer{false};
106 std::atomic<std::size_t> messageCounter{0};
107
108 protected:
109 std::unique_ptr<BaseTimeCoordinator> timeCoord;
110 gmlc::containers::BlockingPriorityQueue<ActionMessage> actionQueue;
111 std::shared_ptr<LogManager> mLogManager;
113 enum class BrokerState : int16_t {
114 CREATED = -10,
115 CONFIGURING = -7,
116 CONFIGURED = -6,
117 CONNECTING = -4,
118 CONNECTED = -3,
119 INITIALIZING = -1,
120 OPERATING = 0,
121 CONNECTED_ERROR = 3,
122 TERMINATING = 4,
123 TERMINATING_ERROR = 5,
124 TERMINATED = 6,
125 ERRORED = 7,
126 };
127
128 enum class TickForwardingReasons : uint32_t {
129 NONE = 0,
130 NO_COMMS = 0x01,
131 PING_RESPONSE = 0x02,
132 QUERY_TIMEOUT = 0x04,
133 GRANT_TIMEOUT = 0x08,
134 DISCONNECT_TIMEOUT = 0x10
135 };
136 bool noAutomaticID{false};
137 bool hasTimeDependency{false};
139 bool enteredExecutionMode{false};
140 bool waitingForBrokerPingReply{false};
141 bool hasFilters{false};
142
143 bool no_ping{false};
144 bool uuid_like{false};
146 bool useJsonSerialization{false};
147 bool enable_profiling{false};
148 bool allowRemoteControl{true};
150 bool errorOnUnmatchedConnections{false};
151 bool globalDisconnect{false};
154 decltype(std::chrono::steady_clock::now()) errorTimeStart;
156 decltype(std::chrono::steady_clock::now()) disconnectTime;
157 std::atomic<int> lastErrorCode{0};
158 std::string lastErrorString;
159 std::string configString;
160 bool fileInUse{false};
161
162 private:
164 std::shared_ptr<ProfilerBuffer> prBuff;
165
167 bool forwardTick{false};
169 uint32_t forwardingReasons{0U};
171 std::atomic<BrokerState> brokerState{BrokerState::CREATED};
172
173 public:
174 explicit BrokerBase(bool DisableQueue = false) noexcept;
175 explicit BrokerBase(std::string_view broker_name, bool DisableQueue = false);
176
177 virtual ~BrokerBase();
178
182 void loadInfoFromToml(const std::string& toml, bool runArgParser = true);
183
187 void loadInfoFromJson(const std::string& json, bool runArgParser = true);
188
192 int parseArgs(int argc, char* argv[]);
196 int parseArgs(std::vector<std::string> args);
200 int parseArgs(std::string_view initializationString);
203 virtual void configureBase();
204
206 void addActionMessage(const ActionMessage& message);
208 void addActionMessage(ActionMessage&& message);
209
215 void setLoggerFunction(
216 std::function<void(int level, std::string_view identifier, std::string_view message)>
217 logFunction);
219 void logFlush();
221 bool isRunning() const { return mainLoopIsRunning.load(); }
223 void setLogLevel(int32_t level);
228 void setLogLevels(int32_t consoleLevel, int32_t fileLevel);
230 GlobalBrokerId getGlobalId() const { return global_id.load(); }
231
232 private:
234 void queueProcessingLoop();
237 action_message_def::action_t commandProcessor(ActionMessage& command);
238
240 std::shared_ptr<helicsCLI11App> generateBaseCLI();
242 void baseConfigure(ActionMessage& command);
243
246 void addActionMessage(ActionMessage&& message) const;
247
248 protected:
250 static bool isReasonForTick(std::uint32_t code, TickForwardingReasons reason)
251 {
252 return ((static_cast<std::uint32_t>(reason) & code) != 0);
253 }
255 void setTickForwarding(TickForwardingReasons reason, bool value = true);
256 BrokerState getBrokerState() const { return brokerState.load(); }
257 bool setBrokerState(BrokerState newState);
258 bool transitionBrokerState(BrokerState expectedState, BrokerState newState);
260 virtual void processDisconnect(bool skipUnregister = false) = 0;
263 virtual bool tryReconnect() = 0;
266 virtual void processCommand(ActionMessage&& cmd) = 0;
272 virtual void processPriorityCommand(ActionMessage&& command) = 0;
273
278 bool sendToLogger(GlobalFederateId federateID,
279 int logLevel,
280 std::string_view name,
281 std::string_view message,
282 bool fromRemote = false) const;
284 void saveProfilingData(std::string_view message);
286 void writeProfilingData();
288 void generateNewIdentifier();
290 virtual std::string generateLocalAddressString() const = 0;
292 virtual std::shared_ptr<helicsCLI11App> generateCLI();
294 void setErrorState(int eCode, std::string_view estring);
296 void setLoggingFile(std::string_view lfile);
298 bool getFlagValue(int32_t flag) const;
300 virtual double getSimulationTime() const { return mInvalidSimulationTime; }
302 std::pair<bool, std::vector<std::string_view>> processBaseCommands(ActionMessage& command);
304 void addBaseInformation(nlohmann::json& base, bool hasParent) const;
305
306 public:
308 std::function<void(int, std::string_view, std::string_view)> getLoggingCallback() const;
310 void joinAllThreads();
312 std::size_t currentMessageCounter() const
313 {
314 return messageCounter.load(std::memory_order_acquire);
315 }
316 friend class TimeoutMonitor;
317 friend const std::string& brokerStateName(BrokerState state);
318 };
319
322 const std::string& brokerStateName(BrokerBase::BrokerState state);
323
324 } // namespace helics
helics::ActionMessage
Definition: ActionMessage.hpp:30
helics::BrokerBase
Definition: BrokerBase.hpp:39
helics::BrokerBase::queueProcessingThread
std::thread queueProcessingThread
Definition: BrokerBase.hpp:75
helics::BrokerBase::setLoggingFile
void setLoggingFile(std::string_view lfile)
Definition: BrokerBase.cpp:640
helics::BrokerBase::disconnectTime
decltype(std::chrono::steady_clock::now()) disconnectTime
time when the disconnect started
Definition: BrokerBase.hpp:156
helics::BrokerBase::debugging
bool debugging
flag indicating operation in a user debugging mode
Definition: BrokerBase.hpp:83
helics::BrokerBase::setLogLevels
void setLogLevels(int32_t consoleLevel, int32_t fileLevel)
Definition: BrokerBase.cpp:766
helics::BrokerBase::brokerStateName
friend const std::string & brokerStateName(BrokerState state)
Definition: BrokerBase.cpp:1186
helics::BrokerBase::configString
std::string configString
storage for a config file location
Definition: BrokerBase.hpp:159
helics::BrokerBase::setLoggerFunction
void setLoggerFunction(std::function< void(int level, std::string_view identifier, std::string_view message)> logFunction)
Definition: BrokerBase.cpp:747
helics::BrokerBase::configureBase
virtual void configureBase()
Definition: BrokerBase.cpp:510
helics::BrokerBase::observer
bool observer
flag indicating that the broker is an observer only
Definition: BrokerBase.hpp:85
helics::BrokerBase::minFederateCount
int32_t minFederateCount
Definition: BrokerBase.hpp:51
helics::BrokerBase::identifier
std::string identifier
an identifier for the broker
Definition: BrokerBase.hpp:67
helics::BrokerBase::joinAllThreads
void joinAllThreads()
Definition: BrokerBase.cpp:97
helics::BrokerBase::lastErrorCode
std::atomic< int > lastErrorCode
storage for last error code
Definition: BrokerBase.hpp:157
helics::BrokerBase::isReasonForTick
static bool isReasonForTick(std::uint32_t code, TickForwardingReasons reason)
Definition: BrokerBase.hpp:250
helics::BrokerBase::no_ping
bool no_ping
indicator that the broker is not very responsive to ping requests
Definition: BrokerBase.hpp:143
helics::BrokerBase::higher_broker_id
GlobalBrokerId higher_broker_id
the id code of the broker 1 level about this broker
Definition: BrokerBase.hpp:46
helics::BrokerBase::BrokerState
BrokerState
Definition: BrokerBase.hpp:113
helics::BrokerBase::BrokerState::CONFIGURED
@ CONFIGURED
the broker itself has been configured and is ready to connect
helics::BrokerBase::BrokerState::TERMINATING_ERROR
@ TERMINATING_ERROR
the termination process has started while in an error state
helics::BrokerBase::BrokerState::OPERATING
@ OPERATING
normal operating conditions
helics::BrokerBase::BrokerState::CONNECTED_ERROR
@ CONNECTED_ERROR
error state but still connected
helics::BrokerBase::BrokerState::INITIALIZING
@ INITIALIZING
the enter initialization process has started
helics::BrokerBase::BrokerState::CREATED
@ CREATED
the broker has been created
helics::BrokerBase::BrokerState::CONNECTING
@ CONNECTING
the connection process has started
helics::BrokerBase::BrokerState::TERMINATED
@ TERMINATED
the termination process has started
helics::BrokerBase::BrokerState::CONNECTED
@ CONNECTED
the connection process has completed
helics::BrokerBase::BrokerState::CONFIGURING
@ CONFIGURING
the broker is in the processing of configuring
helics::BrokerBase::BrokerState::TERMINATING
@ TERMINATING
the termination process has started
helics::BrokerBase::BrokerState::ERRORED
@ ERRORED
an error was encountered
helics::BrokerBase::timeout
Time timeout
timeout to wait to establish a broker connection before giving up
Definition: BrokerBase.hpp:60
helics::BrokerBase::setTickForwarding
void setTickForwarding(TickForwardingReasons reason, bool value=true)
Definition: BrokerBase.cpp:1079
helics::BrokerBase::saveProfilingData
void saveProfilingData(std::string_view message)
Definition: BrokerBase.cpp:591
helics::BrokerBase::brokerKey
std::string brokerKey
Definition: BrokerBase.hpp:68
helics::BrokerBase::isRunning
bool isRunning() const
Definition: BrokerBase.hpp:221
helics::BrokerBase::sendToLogger
bool sendToLogger(GlobalFederateId federateID, int logLevel, std::string_view name, std::string_view message, bool fromRemote=false) const
Definition: BrokerBase.cpp:558
helics::BrokerBase::enable_profiling
bool enable_profiling
indicator that profiling is enabled
Definition: BrokerBase.hpp:147
helics::BrokerBase::networkTimeout
Time networkTimeout
timeout to establish a socket connection before giving up
Definition: BrokerBase.hpp:61
helics::BrokerBase::maxLogLevel
std::atomic< int32_t > maxLogLevel
Definition: BrokerBase.hpp:48
helics::BrokerBase::processCommand
virtual void processCommand(ActionMessage &&cmd)=0
helics::BrokerBase::errorOnUnmatchedConnections
bool errorOnUnmatchedConnections
error if there are unmatched connections on init
Definition: BrokerBase.hpp:150
helics::BrokerBase::currentMessageCounter
std::size_t currentMessageCounter() const
Definition: BrokerBase.hpp:312
helics::BrokerBase::haltOperations
std::atomic< bool > haltOperations
flag indicating that no further message should be processed
Definition: BrokerBase.hpp:77
helics::BrokerBase::mLogManager
std::shared_ptr< LogManager > mLogManager
object to handle the logging considerations
Definition: BrokerBase.hpp:111
helics::BrokerBase::getGlobalId
GlobalBrokerId getGlobalId() const
Definition: BrokerBase.hpp:230
helics::BrokerBase::minChildCount
int32_t minChildCount
Definition: BrokerBase.hpp:57
helics::BrokerBase::timeCoord
std::unique_ptr< BaseTimeCoordinator > timeCoord
object managing the time control
Definition: BrokerBase.hpp:109
helics::BrokerBase::loadInfoFromJson
void loadInfoFromJson(const std::string &json, bool runArgParser=true)
Definition: BrokerBase.cpp:394
helics::BrokerBase::generateLocalAddressString
virtual std::string generateLocalAddressString() const =0
helics::BrokerBase::logFlush
void logFlush()
Definition: BrokerBase.cpp:758
helics::BrokerBase::getLoggingCallback
std::function< void(int, std::string_view, std::string_view)> getLoggingCallback() const
Definition: BrokerBase.cpp:90
helics::BrokerBase::asyncTime
bool asyncTime
flag indicating the use of async time keeping
Definition: BrokerBase.hpp:89
helics::BrokerBase::address
std::string address
network location of the broker
Definition: BrokerBase.hpp:73
helics::BrokerBase::queryTimeout
Time queryTimeout
Definition: BrokerBase.hpp:62
helics::BrokerBase::disableDynamicSources
bool disableDynamicSources
flag disabling dynamic data sources
Definition: BrokerBase.hpp:93
helics::BrokerBase::terminate_on_error
bool terminate_on_error
flag indicating that the federation should halt on any error
Definition: BrokerBase.hpp:81
helics::BrokerBase::restrictive_time_policy
bool restrictive_time_policy
flag indicating the broker should use a conservative time policy
Definition: BrokerBase.hpp:79
helics::BrokerBase::maxCoSimDuration
Time maxCoSimDuration
the maximum lifetime (wall clock time) of the co-simulation
Definition: BrokerBase.hpp:66
helics::BrokerBase::generateCLI
virtual std::shared_ptr< helicsCLI11App > generateCLI()
Definition: BrokerBase.cpp:105
helics::BrokerBase::hasTimeDependency
bool hasTimeDependency
Definition: BrokerBase.hpp:137
helics::BrokerBase::noAutomaticID
bool noAutomaticID
the broker should not automatically generate an ID
Definition: BrokerBase.hpp:136
helics::BrokerBase::allowRemoteControl
bool allowRemoteControl
Definition: BrokerBase.hpp:148
helics::BrokerBase::uuid_like
bool uuid_like
will be set to true if the name looks like a uuid
Definition: BrokerBase.hpp:144
helics::BrokerBase::enteredExecutionMode
bool enteredExecutionMode
flag indicating that the broker has entered execution mode
Definition: BrokerBase.hpp:139
helics::BrokerBase::tryReconnect
virtual bool tryReconnect()=0
Definition: BrokerBase.cpp:865
helics::BrokerBase::addBaseInformation
void addBaseInformation(nlohmann::json &base, bool hasParent) const
Definition: BrokerBase.cpp:731
helics::BrokerBase::errorDelay
Time errorDelay
time to delay before terminating after error state
Definition: BrokerBase.hpp:64
helics::BrokerBase::waitingForBrokerPingReply
bool waitingForBrokerPingReply
flag indicating we are waiting for a ping reply
Definition: BrokerBase.hpp:140
helics::BrokerBase::setErrorState
void setErrorState(int eCode, std::string_view estring)
Definition: BrokerBase.cpp:615
helics::BrokerBase::lastErrorString
std::string lastErrorString
storage for last error string
Definition: BrokerBase.hpp:158
helics::BrokerBase::writeProfilingData
void writeProfilingData()
Definition: BrokerBase.cpp:600
helics::BrokerBase::getSimulationTime
virtual double getSimulationTime() const
Definition: BrokerBase.hpp:300
helics::BrokerBase::useJsonSerialization
bool useJsonSerialization
Definition: BrokerBase.hpp:146
helics::BrokerBase::generateNewIdentifier
void generateNewIdentifier()
Definition: BrokerBase.cpp:585
helics::BrokerBase::minBrokerCount
int32_t minBrokerCount
Definition: BrokerBase.hpp:53
helics::BrokerBase::processBaseCommands
std::pair< bool, std::vector< std::string_view > > processBaseCommands(ActionMessage &command)
Definition: BrokerBase.cpp:658
helics::BrokerBase::processPriorityCommand
virtual void processPriorityCommand(ActionMessage &&command)=0
helics::BrokerBase::getFlagValue
bool getFlagValue(int32_t flag) const
Definition: BrokerBase.cpp:645
helics::BrokerBase::actionQueue
gmlc::containers::BlockingPriorityQueue< ActionMessage > actionQueue
primary routing queue
Definition: BrokerBase.hpp:110
helics::BrokerBase::hasFilters
bool hasFilters
flag indicating filters come through the broker
Definition: BrokerBase.hpp:141
helics::BrokerBase::global_id
std::atomic< GlobalBrokerId > global_id
Definition: BrokerBase.hpp:43
helics::BrokerBase::loadInfoFromToml
void loadInfoFromToml(const std::string &toml, bool runArgParser=true)
Definition: BrokerBase.cpp:476
helics::BrokerBase::parseArgs
int parseArgs(int argc, char *argv[])
Definition: BrokerBase.cpp:312
helics::BrokerBase::grantTimeout
Time grantTimeout
timeout for triggering diagnostic action waiting for a time grant
Definition: BrokerBase.hpp:65
helics::BrokerBase::setLogLevel
void setLogLevel(int32_t level)
Definition: BrokerBase.cpp:753
helics::BrokerBase::tickTimer
Time tickTimer
the length of each heartbeat tick
Definition: BrokerBase.hpp:59
helics::BrokerBase::dynamicFederation
bool dynamicFederation
flag indicating that the broker supports dynamic federates
Definition: BrokerBase.hpp:91
helics::BrokerBase::processDisconnect
virtual void processDisconnect(bool skipUnregister=false)=0
helics::BrokerBase::maxIterationCount
int32_t maxIterationCount
the maximum number of iterative loops that are allowed
Definition: BrokerBase.hpp:58
helics::BrokerBase::globalDisconnect
bool globalDisconnect
Definition: BrokerBase.hpp:151
helics::BrokerBase::global_broker_id_local
GlobalBrokerId global_broker_id_local
Definition: BrokerBase.hpp:44
helics::BrokerBase::errorTimeStart
decltype(std::chrono::steady_clock::now()) errorTimeStart
time when the error condition started; related to the errorDelay
Definition: BrokerBase.hpp:154
helics::BrokerBase::addActionMessage
void addActionMessage(const ActionMessage &message)
Definition: BrokerBase.cpp:772
helics::BrokerBase::globalTime
bool globalTime
flag indicating that the broker should use a global time coordinator
Definition: BrokerBase.hpp:87
helics::GlobalBrokerId
Definition: GlobalFederateId.hpp:30
helics::GlobalFederateId
Definition: GlobalFederateId.hpp:75
helics::TimeoutMonitor
Definition: TimeoutMonitor.h:27
HELICS_LOG_LEVEL_NO_PRINT
@ HELICS_LOG_LEVEL_NO_PRINT
Definition: helics_enums.h:206
helics::action_message_def::action_t
action_t
Definition: ActionMessageDefintions.hpp:20
helics
the main namespace for the helics co-simulation library User functions will be in the helics namespac...
Definition: AsyncFedCallInfo.hpp:14
helics::parent_broker_id
constexpr GlobalBrokerId parent_broker_id
Definition: GlobalFederateId.hpp:67
helics::brokerStateName
const std::string & brokerStateName(BrokerBase::BrokerState state)
Definition: BrokerBase.cpp:1186
helics::Time
TimeRepresentation< count_time< 9 > > Time
Definition: helicsTime.hpp:27