CommsInterface.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 "NetworkBrokerData.hpp"
#include "gmlc/concurrency/TriggerVariable.hpp"
#include "gmlc/concurrency/TripWire.hpp"
#include "gmlc/containers/BlockingPriorityQueue.hpp"
#include "helics/core/ActionMessage.hpp"
 
#include <functional>
#include <memory>
#include <string>
#include <thread>
#include <utility>
 
namespace helics {
 
class CommsInterface {
  public:
    enum class thread_generation {
        single,  
        dual  
    };
    CommsInterface() = default;
    explicit CommsInterface(thread_generation threads);
    virtual ~CommsInterface();
 
    virtual void loadNetworkInfo(const NetworkBrokerData& netInfo);
    void loadTargetInfo(std::string_view localTarget,
                        std::string_view brokerTarget,
                        gmlc::networking::InterfaceNetworks targetNetwork =
                            gmlc::networking::InterfaceNetworks::LOCAL);
    void transmit(route_id rid, const ActionMessage& cmd);
    void transmit(route_id rid, ActionMessage&& cmd);
    void addRoute(route_id rid, std::string_view routeInfo);
    void removeRoute(route_id rid);
    bool connect();
 
    void disconnect();
 
    bool reconnect();
    void setName(const std::string& commName);
 
    void setRequireBrokerConnection(bool requireBrokerConnection);
 
    void setCallback(std::function<void(ActionMessage&&)> callback);
    void setLoggingCallback(
        std::function<void(int level, std::string_view name, std::string_view message)> callback);
    void setMessageSize(int maxMsgSize, int maxCount);
    bool isConnected() const;
 
    void setTimeout(std::chrono::milliseconds timeOut);
    virtual void setFlag(std::string_view flag, bool val);
    void setServerMode(bool serverActive);
 
    void logWarning(std::string_view message) const;
    void logError(std::string_view message) const;
    void logMessage(std::string_view message) const;
 
  protected:
    enum class ConnectionStatus : int {
 
        STARTUP = -1,  
        CONNECTED = 0,  
        RECONNECTING = 1,  
        TERMINATED = 2,  
        ERRORED = 4  
    };
 
  private:
    std::atomic<ConnectionStatus> rxStatus{ConnectionStatus::STARTUP};
 
  protected:
    gmlc::concurrency::TriggerVariable rxTrigger;
 
    std::string name;  
    std::string localTargetAddress;  
    std::string brokerTargetAddress;  
    std::string brokerName;  
    std::string brokerInitString;
 
  private:
    std::string randomID;  
    std::atomic<ConnectionStatus> txStatus{ConnectionStatus::STARTUP};
    gmlc::concurrency::TriggerVariable txTrigger;
    std::atomic<bool> operating{false};  
    const bool singleThread{false};  
 
  protected:
    bool mRequireBrokerConnection{
        false};  
    bool serverMode{true};  
    bool autoBroker{false};  
    bool useJsonSerialization{false};  
    bool observer{false};  
    std::chrono::milliseconds connectionTimeout{4000};
    int maxMessageSize = 16 * 1024;  
    int maxMessageCount = 512;  
    std::atomic<bool> requestDisconnect{false};  
    std::function<void(ActionMessage&&)>
        ActionCallback;  
    std::function<void(int level, std::string_view name, std::string_view message)>
        loggingCallback;  
    gmlc::containers::BlockingPriorityQueue<std::pair<route_id, ActionMessage>>
        txQueue;  
    // closing the files or connection can take some time so there is a need for inter-thread
    // communication to not spit out warning messages if it is in the process of disconnecting
    std::atomic<bool> disconnecting{
        false};  
    gmlc::networking::InterfaceNetworks interfaceNetwork{
        gmlc::networking::InterfaceNetworks::LOCAL};
 
  private:
    std::thread queue_transmitter;  
    std::thread queue_watcher;  
    std::mutex threadSyncLock;  
    virtual void queue_rx_function() = 0;  
    virtual void queue_tx_function() = 0;  
    virtual void closeTransmitter();  
    virtual void closeReceiver();  
    virtual void reconnectTransmitter();  
    virtual void reconnectReceiver();  
  protected:
    void setTxStatus(ConnectionStatus status);
    void setRxStatus(ConnectionStatus status);
    ConnectionStatus getRxStatus() const { return rxStatus.load(); }
    ConnectionStatus getTxStatus() const { return txStatus.load(); }
    bool propertyLock();
    void propertyUnLock();
    void join_tx_rx_thread();
    const std::string& getRandomID() const { return randomID; }
 
  private:
    gmlc::concurrency::TripWireDetector
        tripDetector;  
};
 
namespace CommFactory {
    class CommBuilder {
      public:
        virtual std::unique_ptr<CommsInterface> build() = 0;
    };
 
    template<class CommTYPE>
    class CommTypeBuilder final: public CommBuilder {
      public:
        static_assert(std::is_base_of<CommsInterface, CommTYPE>::value,
                      "Type does not inherit from helics::CommsInterface");
 
        using comm_build_type = CommTYPE;
        virtual std::unique_ptr<CommsInterface> build() override
        {
            return std::make_unique<CommTYPE>();
        }
    };
 
    void defineCommBuilder(std::shared_ptr<CommBuilder> builder,
                           std::string_view commTypeName,
                           int code);
 
    template<class CommTYPE>
    std::shared_ptr<CommBuilder> addCommType(std::string_view commTypeName, int code)
    {
        auto bld = std::make_shared<CommTypeBuilder<CommTYPE>>();
        std::shared_ptr<CommBuilder> cbld = std::static_pointer_cast<CommBuilder>(bld);
        defineCommBuilder(cbld, commTypeName, code);
        return cbld;
    }
 
    std::unique_ptr<CommsInterface> create(CoreType type);
    std::unique_ptr<CommsInterface> create(std::string_view type);
 
}  // namespace CommFactory
 
template<class X>
class ConditionalChangeOnDestroy {
  private:
    std::atomic<X>& aref;
    X fval;
    X expectedValue;
 
  public:
    ConditionalChangeOnDestroy(std::atomic<X>& var, X finalValue, X expValue):
        aref(var), fval(std::move(finalValue)), expectedValue(std::move(expValue))
    {
    }
    ~ConditionalChangeOnDestroy() { aref.compare_exchange_strong(expectedValue, fval); }
};
 
}  // namespace helics