FederateState.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 "ActionMessage.hpp"
#include "BasicHandleInfo.hpp"
#include "CoreTypes.hpp"
#include "InterfaceInfo.hpp"
#include "core-data.hpp"
#include "gmlc/containers/BlockingQueue.hpp"
#include "helicsTime.hpp"
 
#include <algorithm>
#include <atomic>
#include <chrono>
#include <deque>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <thread>
#include <tuple>
#include <utility>
#include <vector>
 
namespace helics {
class SubscriptionInfo;
class PublicationInfo;
class EndpointInfo;
class FilterInfo;
class CommonCore;
class CoreFederateInfo;
 
class TimeCoordinator;
class MessageTimer;
class LogManager;
 
constexpr Time startupTime = Time::minVal();
constexpr Time initialTime{-1000000.0};
 
enum class TimeSynchronizationMethod : uint8_t { DISTRIBUTED = 0, GLOBAL = 1, ASYNC = 2 };
 
class FederateState {
  public:
    FederateState(const std::string& fedName, const CoreFederateInfo& fedInfo);
    // the destructor is defined so some classes linked with unique ptrs don't have to be defined in
    // the header
    FederateState(const FederateState&) = delete;
    FederateState& operator=(const FederateState&) = delete;
    ~FederateState();
 
  private:
    const std::string name;  
    std::unique_ptr<TimeCoordinator> timeCoord;
 
  public:
    LocalFederateId local_id;  
    std::atomic<GlobalFederateId> global_id;  
  private:
    std::atomic<FederateStates> state{FederateStates::CREATED};
    bool only_transmit_on_change{false};  
    bool realtime{false};  
    bool observer{false};  
    bool reentrant{false};  
    bool mSourceOnly{false};  
    bool mCallbackBased{false};  
    bool strict_input_type_checking{false};
    bool ignore_unit_mismatch{false};  
    bool mSlowResponding{false};
    bool mAllowRemoteControl{true};
    InterfaceInfo interfaceInformation;  
    std::unique_ptr<LogManager> mLogManager;
    int maxLogLevel{HELICS_LOG_LEVEL_NO_PRINT};
 
  public:
    std::atomic<bool> init_transmitted{false};  
    int indexGroup{0};
 
  private:
    bool wait_for_current_time{false};
    bool ignore_time_mismatch_warnings{false};
    bool mProfilerActive{false};
    bool mLocalProfileCapture{false};
    int errorCode{0};  
    CommonCore* mParent{nullptr};  
    std::string errorString;  
    decltype(std::chrono::steady_clock::now()) start_clock_time;
    Time rt_lag{timeZero};  
    Time rt_lead{timeZero};  
    Time grantTimeOutPeriod{timeZero};  
    std::int32_t realTimeTimerIndex{-1};  
    std::int32_t grantTimeoutTimeIndex{-1};  
  public:
    std::atomic<bool> initRequested{false};
    // temporary
    std::atomic<bool> requestingMode{false};
 
    std::atomic<bool> initIterating{false};
 
  private:
    bool iterating{false};  
    bool timeGranted_mode{false};  
    bool terminate_on_error{false};  
    IterationRequest lastIterationRequest{IterationRequest::NO_ITERATIONS};
    TimeSynchronizationMethod timeMethod{TimeSynchronizationMethod::DISTRIBUTED};
    std::uint32_t mGrantCount{0};  // this is intended to allow wrapping
    std::shared_ptr<MessageTimer> mTimer;
    gmlc::containers::BlockingQueue<ActionMessage> queue;
    gmlc::containers::BlockingQueue<std::pair<std::string, std::string>> commandQueue;
    std::atomic<uint16_t> interfaceFlags{0};
    std::map<GlobalFederateId, std::deque<ActionMessage>> delayQueues;
    std::vector<InterfaceHandle> events;  
    std::vector<InterfaceHandle> eventMessages;  
    std::vector<GlobalFederateId> delayedFederates;  
    Time time_granted{startupTime};  
    Time allowed_send_time{startupTime};  
    Time minimumReceiveTime{startupTime};  
    mutable std::atomic_flag processing = ATOMIC_FLAG_INIT;  
 
    std::vector<std::function<std::string(std::string_view)>> queryCallbacks;
    std::shared_ptr<FederateOperator> fedCallbacks;  
    std::vector<std::pair<std::string, std::string>> tags;  
    std::atomic<bool> queueProcessing{false};
    Time nextValueTime() const;
    Time nextMessageTime() const;
 
    void setState(FederateStates newState);
 
    bool messageShouldBeDelayed(const ActionMessage& cmd) const noexcept;
    void addFederateToDelay(GlobalFederateId gid);
    void generateConfig(Json::Value& base) const;
 
  public:
    void reset(const CoreFederateInfo& fedInfo);
    const std::string& getIdentifier() const { return name; }
    FederateStates getState() const;
    InterfaceInfo& interfaces() { return interfaceInformation; }
    const InterfaceInfo& interfaces() const { return interfaceInformation; }
 
    uint64_t getQueueSize(InterfaceHandle hid) const;
    uint64_t getQueueSize() const;
    int32_t getCurrentIteration() const;
    std::unique_ptr<Message> receive(InterfaceHandle hid);
    std::unique_ptr<Message> receiveAny(InterfaceHandle& hid);
    const std::shared_ptr<const SmallBuffer>& getValue(InterfaceHandle handle,
                                                       uint32_t* inputIndex);
 
    const std::vector<std::shared_ptr<const SmallBuffer>>& getAllValues(InterfaceHandle handle);
 
    std::pair<SmallBuffer, Time> getPublishedValue(InterfaceHandle handle);
    void setParent(CommonCore* coreObject) { mParent = coreObject; }
    void setProperties(const ActionMessage& cmd);
    void setInterfaceProperty(const ActionMessage& cmd);
    void setProperty(int timeProperty, Time propertyVal);
    void setProperty(int intProperty, int propertyVal);
    void setOptionFlag(int optionFlag, bool value);
    Time getTimeProperty(int timeProperty) const;
    bool getOptionFlag(int optionFlag) const;
    int32_t getHandleOption(InterfaceHandle handle, char iType, int32_t option) const;
    uint16_t getInterfaceFlags() const { return interfaceFlags.load(); }
    int getIntegerProperty(int intProperty) const;
    int publicationCount() const;
    int endpointCount() const;
    int inputCount() const;
    void spinlock() const
    {
        while (processing.test_and_set()) {
            ;  // spin
        }
    }
    void sleeplock() const
    {
        if (!processing.test_and_set()) {
            return;
        }
        // spin for 10000 tries
        for (int ii = 0; ii < 10000; ++ii) {
            if (!processing.test_and_set()) {
                return;
            }
        }
        while (processing.test_and_set()) {
            std::this_thread::yield();
        }
    }
    void lock() { sleeplock(); }
 
    bool try_lock() const { return !processing.test_and_set(); }
    void unlock() const { processing.clear(); }
    int loggingLevel() const;
 
    void setTag(std::string_view tag, std::string_view value);
    const std::string& getTag(std::string_view tag) const;
    const std::pair<std::string, std::string>& getTagByIndex(size_t index) const
    {
        return tags[index];
    }
    auto tagCount() const { return tags.size(); }
    bool isCallbackFederate() const { return mCallbackBased; }
 
  private:
    MessageProcessingResult processQueue() noexcept;
 
    MessageProcessingResult processDelayQueue() noexcept;
 
    std::optional<MessageProcessingResult>
        checkProcResult(std::tuple<FederateStates, MessageProcessingResult, bool>& proc_result,
                        ActionMessage& cmd);
    MessageProcessingResult processActionMessage(ActionMessage& cmd);
 
    void processDataConnectionMessage(ActionMessage& cmd);
 
    void processDataMessage(ActionMessage& cmd);
    void timeoutCheck(ActionMessage& cmd);
    void processLoggingMessage(ActionMessage& cmd);
    void fillEventVectorUpTo(Time currentTime);
    void fillEventVectorInclusive(Time currentTime);
    void fillEventVectorNextIteration(Time currentTime);
    void addDependency(GlobalFederateId fedToDependOn);
    void addDependent(GlobalFederateId fedThatDependsOnThis);
    void resetDependency(GlobalFederateId gid);
 
    int checkInterfaces();
    std::string processQueryActual(std::string_view query) const;
    void generateProfilingMessage(bool enterHelicsCode);
    void generateProfilingMarker();
    void updateMaxLogLevel();
 
    void callbackProcessing() noexcept;
    void callbackReturnResult(FederateStates lastState,
                              MessageProcessingResult result,
                              FederateStates newState) noexcept;
    void initCallbackProcessing();
    void execCallbackProcessing(IterationResult result);
    void updateDataForExecEntry(MessageProcessingResult result, IterationRequest iterate);
    void updateDataForTimeReturn(MessageProcessingResult result,
                                 Time nextTime,
                                 IterationRequest iterate);
 
  public:
    Time grantedTime() const { return time_granted; }
    Time nextAllowedSendTime() const { return allowed_send_time; }
    const std::vector<InterfaceHandle>& getEvents() const;
    std::vector<GlobalFederateId> getDependencies() const;
    std::vector<GlobalFederateId> getDependents() const;
    const std::string& lastErrorString() const { return errorString; }
    int lastErrorCode() const noexcept { return errorCode; }
    void setCoreObject(CommonCore* parent);
    // the next 5 functions are the processing functions that actually process the queue
    IterationResult waitSetup();
    IterationResult enterInitializingMode(IterationRequest request);
    iteration_time enterExecutingMode(IterationRequest iterate, bool sendRequest = false);
    iteration_time requestTime(Time nextTime, IterationRequest iterate, bool sendRequest = false);
    std::vector<GlobalHandle> getSubscribers(InterfaceHandle handle);
 
    std::vector<std::pair<GlobalHandle, std::string_view>>
        getMessageDestinations(InterfaceHandle handle);
 
    MessageProcessingResult genericUnspecifiedQueueProcess(bool busyReturn);
    void finalize();
    void processCommunications(std::chrono::milliseconds period);
    void addAction(const ActionMessage& action);
    void addAction(ActionMessage&& action);
    std::optional<ActionMessage> processPostTerminationAction(const ActionMessage& action);
 
    void forceProcessMessage(ActionMessage& action);
 
    void logMessage(int level,
                    std::string_view logMessageSource,
                    std::string_view message,
                    bool fromRemote = false) const;
 
    void setLogger(std::function<void(int, std::string_view, std::string_view)> logFunction);
 
    void setCallbackOperator(std::shared_ptr<FederateOperator> fed)
    {
        fedCallbacks = std::move(fed);
    }
 
    void setQueryCallback(std::function<std::string(std::string_view)> queryCallbackFunction,
                          int order)
    {
        order = std::clamp(order, 1, 10);
 
        if (static_cast<int>(queryCallbacks.size()) < order) {
            queryCallbacks.resize(order);
        }
        queryCallbacks[order - 1] = std::move(queryCallbackFunction);
    }
    std::string processQuery(std::string_view query, bool force_ordering = false) const;
    bool checkAndSetValue(InterfaceHandle pub_id, const char* data, uint64_t len);
 
    void routeMessage(const ActionMessage& msg);
 
    void routeMessage(ActionMessage&& msg);
    void createInterface(InterfaceType htype,
                         InterfaceHandle handle,
                         std::string_view key,
                         std::string_view type,
                         std::string_view units,
                         uint16_t flags);
    void closeInterface(InterfaceHandle handle, InterfaceType type);
    void sendCommand(ActionMessage& command);
 
    std::pair<std::string, std::string> getCommand();
    std::pair<std::string, std::string> waitCommand();
};
 
}  // namespace helics