agent.hpp

Go to the documentation of this file.

/*
     _                    _          _
    / \   __ _  ___ _ __ | |_    ___| | __ _ ___ ___
   / _ \ / _` |/ _ \ '_ \| __|  / __| |/ _` / __/ __|
  / ___ \ (_| |  __/ | | | |_  | (__| | (_| \__ \__ \
 /_/   \_\__, |\___|_| |_|\__|  \___|_|\__,_|___/___/
         |___/
 
Base class for all agents. This class is used to define the basic
functionalities provided by all agents. Each agent subclass must implement the
pure virtual functions defined in this class (currently none)
 
Author(s): Paolo Bosetti
*/
 
#ifndef AGENT_HPP
#define AGENT_HPP
 
#if defined _WIN32 && !defined NOMINMAX
#define NOMINMAX
#endif
 
#include "mads.hpp"
#include <nlohmann/json.hpp>
#ifdef _WIN32
#include <winsock2.h>
#else
#include <unistd.h>
#endif
#ifndef _MSC_VER
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wunknown-attributes"
#endif
#include <toml++/toml.hpp>
#ifndef _MSC_VER
#pragma GCC diagnostic pop
#endif
#include <iostream>
#include <regex>
#include <string>
#include <string_view>
#include <thread>
#include <future>
#include <zmq.hpp>
#include <zmq_addon.hpp>
#include <mutex>
#include <condition_variable>
#include <optional>
#include <span>
#include <atomic>
#include <memory>
#include "clock_offset.hpp"
#include "curve.hpp"
#include "exec_path.hpp"
#include "socket_monitor.hpp"
#include "topic_match.hpp"
 
#ifndef HOST_NAME_MAX
#define HOST_NAME_MAX 255
#endif
 
#ifndef MADS_AGENT_NO_INFO
#include <rang.hpp>
using namespace rang;
#endif
 
namespace Mads {
 
template<typename T> struct SharedLatest {
  std::mutex mtx;
  std::condition_variable cv;
  std::optional<T> value;
};
 
class LazyPayload {
public:
  LazyPayload() = default;
  static LazyPayload from_text(std::string text) {
    LazyPayload p;
    p._text = std::move(text);
    return p;
  }
  static LazyPayload from_doc(nlohmann::json doc) {
    LazyPayload p;
    p._doc = std::move(doc);
    return p;
  }
  const std::string &text() const {
    if (!_text)
      _text = _doc ? _doc->dump() : std::string();
    return *_text;
  }
  const nlohmann::json &doc() const {
    if (!_doc) {
      if (_text && !_text->empty())
        _doc = nlohmann::json::parse(*_text);
      else
        _doc = nlohmann::json();
    }
    return *_doc;
  }
 
private:
  mutable std::optional<std::string> _text;
  mutable std::optional<nlohmann::json> _doc;
};
 
struct ClockPeerObservation {
  std::string responder_agent_id;
  std::string responder_name;
  std::string responder_hostname;
  std::string responder_domain;
  ClockSample sample;
  ClockOffsetResult responder_adopted;
};
 
class Agent; // forward declaration
 
std::unique_ptr<Agent> start_agent(std::string name, std::string settings_uri,
                                   std::map<std::string, std::string> crypto_settings = {});
 
class Agent {
 
/*
  ____  _        _   _      
 / ___|| |_ __ _| |_(_) ___ 
 \___ \| __/ _` | __| |/ __|
  ___) | || (_| | |_| | (__ 
 |____/ \__\__,_|\__|_|\___|
                            
*/
 
private:
  void setup_curve_on(zmq::socket_t &socket);
 
  std::tuple<std::string, std::string, double>
  query_broker(std::string uri, std::string name,
               int timeout = DEFAULT_SETTINGS_TIMEOUT_MS);
 
public:
/*
  _     _  __                      _      
 | |   (_)/ _| ___  ___ _   _  ___| | ___ 
 | |   | | |_ / _ \/ __| | | |/ __| |/ _ \
 | |___| |  _|  __/ (__| |_| | (__| |  __/
 |_____|_|_|  \___|\___|\__, |\___|_|\___|
                        |___/             
*/
 
  Agent(std::string name, std::string settings_uri);
 
 
  void init(std::string name, std::string settings_uri, bool crypto = false, std::filesystem::path const &key_dir = "", bool install_watchdog = true);
 
 
  void init(bool crypto = false, bool install_watchdog = true);
 
  void fetch_settings(bool crypto = false);
 
  // Destructor
  virtual ~Agent();
 
  void install_loop_watchdog(uint8_t max_count = 3);
 
 
/*
  ____       _   _   _                 
 / ___|  ___| |_| |_(_)_ __   __ _ ___ 
 \___ \ / _ \ __| __| | '_ \ / _` / __|
  ___) |  __/ |_| |_| | | | | (_| \__ \
 |____/ \___|\__|\__|_|_| |_|\__, |___/
                             |___/     
*/
 
  virtual void load_settings();
 
 
  void save_settings(const std::string path = SETTINGS_PATH);
 
 
  nlohmann::json get_settings();
 
 
#ifndef MADS_AGENT_NO_INFO
  virtual void info(std::ostream &out = std::cout);
#endif
 
 
/*
   ____                            _   _             
  / ___|___  _ __  _ __   ___  ___| |_(_) ___  _ __  
 | |   / _ \| '_ \| '_ \ / _ \/ __| __| |/ _ \| '_ \ 
 | |__| (_) | | | | | | |  __/ (__| |_| | (_) | | | |
  \____\___/|_| |_|_| |_|\___|\___|\__|_|\___/|_| |_|
                                                     
*/
 
  void connect(std::chrono::milliseconds delay = std::chrono::milliseconds(250));
 
  bool wait_for_connection(std::chrono::milliseconds timeout);
 
  Mads::LinkState link_state() const;
 
  void disconnect();
 
 
  void shutdown();
 
 
  void set_cross(bool cross);
  
 
  void enable_remote_control(bool threaded = false);
  
  inline void enable_threaded_remote_control() {
    enable_remote_control(true);
  }
 
 
  void register_event(const event_type event = event_type::marker,
                      const nlohmann::json &info = nlohmann::json(),
                      const std::string &info_name = "info");
 
 
  void publish(nlohmann::json payload, std::string topic = "");
 
 
  void publish(const char *payload, size_t len,
               nlohmann::json meta = nlohmann::json{{"format", "raw"}},
               std::string topic = "");
 
  
  void publish(const std::vector<unsigned char> &payload,
               nlohmann::json meta = nlohmann::json{{"format", "raw"}},
               std::string topic = "");
 
 
  message_type receive(bool dont_block = false);
 
  bool receive_raw_message(std::string &topic, std::vector<std::string> &parts,
                           bool dont_block = false);
 
  void publish_raw_message(const std::string &topic,
                           const std::vector<std::string> &parts);
 
 
  /*
    _                      
   | |    ___   ___  _ __  
   | |   / _ \ / _ \| '_ \ 
   | |__| (_) | (_) | |_) |
   |_____\___/ \___/| .__/ 
                    |_|    
  */
 
  using loop_fun_t = std::function<std::chrono::nanoseconds()>;
  void loop(loop_fun_t const &lambda,
            std::chrono::nanoseconds duration);
 
 
  void loop(loop_fun_t const &lambda);
 
 
  void enable_high_res_loop(bool on = true,
                            std::chrono::nanoseconds spin_margin =
                                std::chrono::microseconds(200));
 
  bool high_res_loop() const;
 
 
  void remote_control(std::string payload_str);
 
 
  /*
       _                                        
      / \   ___ ___ ___  ___ ___  ___  _ __ ___ 
     / _ \ / __/ __/ _ \/ __/ __|/ _ \| '__/ __|
    / ___ \ (_| (_|  __/\__ \__ \ (_) | |  \__ \
   /_/   \_\___\___\___||___/___/\___/|_|  |___/
                                                
  */
 
   void set_agent_id(std::string id);
 
 
  
  std::string get_agent_id();
 
  void set_sub_endpoint(std::string endpoint) { _sub_endpoint = endpoint; }
 
  std::string sub_endpoint() const { return _sub_endpoint; }
 
  void set_pub_endpoint(std::string endpoint) { _pub_endpoint = endpoint; }
 
  std::string pub_endpoint() const { return _pub_endpoint; }
 
  void set_pub_topic(std::string topic);
 
  std::string pub_topic() const { return _pub_topic; }
 
  void set_sub_topic(std::vector<std::string> topics) { _sub_topic = topics; }
  
  std::vector<std::string> sub_topic() const { return _sub_topic; }
 
 
  std::map<std::string, std::string> status();
 
 
  std::string name();
 
 
  std::tuple<std::string, std::string> last_message();
 
 
  std::tuple<std::string, nlohmann::json> last_json();
 
 
  std::string last_topic();
 
 
  std::tuple<std::string, std::string, std::vector<unsigned char>> last_blob();
 
 
  std::tuple<std::string_view, std::string_view,
             std::span<const unsigned char>>
  last_blob_view() const;
 
 
  size_t dropped_messages() const;
 
 
  bool settings_are_local() const;
 
 
  bool is_connected();
 
 
  int settings_timeout();
 
 
  void set_settings_timeout(int to);
 
  void set_settings_timeout(std::chrono::milliseconds to);
 
 
  int receive_timeout();
 
 
  void set_receive_timeout(int to);
 
  void set_receive_timeout(std::chrono::milliseconds to);
 
 
  bool restart();
 
  std::shared_ptr<Mads::Runtime> runtime() const { return _runtime; }
 
  bool running() const { return keep_running(); }
 
  void set_runtime(std::shared_ptr<Mads::Runtime> runtime);
 
 
  std::filesystem::path attachment_path();
 
 
  bool is_crypto();
 
 
  void setup_crypto(Mads::auth_verbose verbose = auth_verbose::off);
 
 
  std::unique_ptr<CurveAuth> *curve_auth();
 
 
  std::filesystem::path key_dir();
 
 
  void set_key_dir(const std::filesystem::path &path);
 
  std::string settings_uri();
 
  
  [[deprecated("conflate is disabled; use set_delivery(Delivery::LastKnownValue)")]]
  void set_conflate(bool conflate);
  [[deprecated("conflate is disabled; use delivery()")]]
  bool conflate();
 
  void set_high_watermark(int i = 1000);
  int high_watermark();
 
  void set_delivery(Delivery d);
 
  Delivery delivery() const;
 
  void set_wire_format(WireFormat fmt);
 
  WireFormat wire_format() const;
 
  void set_compression(Compression c);
 
  Compression compression() const;
 
  static void install_signal_handlers();
 
 
  /*
    ____ _            _      ___   __  __          _
   / ___| | ___   ___| | __ / _ \ / _|/ _|___  ___| |_
  | |   | |/ _ \ / __| |/ /| | | | |_| |_/ __|/ _ \ __|
  | |___| | (_) | (__|   < | |_| |  _|  _\__ \  __/ |_
   \____|_|\___/ \___|_|\_\ \___/|_| |_| |___/\___|\__|
 
  */
 
  Mads::ClockOffsetResult measure_clock_offset(size_t samples = 5,
                                               int timeout_ms = 1000);
 
  std::vector<Mads::ClockPeerObservation> broadcast_clock_probe(
      std::chrono::milliseconds window = std::chrono::milliseconds(300));
 
  Mads::ClockOffsetResult clock_offset() const;
 
  std::string clock_domain() const;
 
  double timecode_fps = MADS_FPS;
  Mads::auth_verbose auth_verbose = auth_verbose::off;
  std::string server_key_name = "broker";
  std::string client_key_name = "client";
 
  /*
    ____       _            _
   |  _ \ _ __(_)_   ____ _| |_ ___
   | |_) | '__| \ \ / / _` | __/ _ \
   |  __/| |  | |\ V / (_| | ||  __/
   |_|   |_|  |_| \_/ \__,_|\__\___|
 
  */
 
protected:
  bool keep_running() const {
    return _runtime->running() && !_stopping.load();
  }
 
  void connect_pub(std::chrono::milliseconds delay = std::chrono::milliseconds(0));
 
 
  void connect_sub();
 
 
  bool receive_raw(zmq::multipart_t &message, bool dont_block = false);
 
  bool _topic_matches_subscription(const std::string &topic) const;
 
  static std::tuple<std::string, std::string, std::string> split_URL(const std::string &url);
 
  void _apply_socket_options();
 
  void _start_io_thread();
 
  void _configure_clock_sync();
 
  bool _clock_wants_sync() const;
 
  void _start_clock_thread();
 
  void _run_peer_measurement();
 
  void _handle_clocksync_message(const nlohmann::json &msg);
 
  void _announce_clock_offset(const Mads::ClockOffsetResult &r);
 
  Mads::ClockOffsetResult
  _stamp_clock_result(Mads::ClockOffsetResult r, Mads::ClockSource source,
                      uint8_t hops);
 
  std::string _clock_agent_identity() const;
 
  // Member variables
  std::string _hostname;
  std::string _name;
  std::string _settings_uri;
  std::string _raw_settings;
  toml::table _config;
  std::string _pub_endpoint, _sub_endpoint;
  std::string _pub_topic;
  std::string _agent_id;
  std::vector<std::string> _sub_topic;
  // Subset of _sub_topic containing a '+'/'#' wildcard token (P2), computed
  // once by connect_sub(). Empty for every agent using only literal
  // sub_topic entries, which keeps the receive-time filter a single cheap
  // emptiness check in that -- the common -- case, adding no overhead.
  std::vector<std::string> _wildcard_sub_topic;
  zmq::context_t _context;
  zmq::socket_t _publisher;
  zmq::socket_t _subscriber;
  // Attached before each socket's connect()/bind() (ZMQ_DEVELOPMENT.md
  // §2.1); stopped in shutdown() before the sockets are closed.
  Mads::SocketMonitor _pub_monitor;
  Mads::SocketMonitor _sub_monitor;
  // A single LazyPayload per message is shared between _last_message and
  // _status so the lazy text/object caches are shared and never duplicated.
  std::map<std::string, std::shared_ptr<LazyPayload>> _status;
  std::tuple<std::string, std::shared_ptr<LazyPayload>> _last_message;
  std::tuple<std::string, std::string, std::vector<unsigned char>> _last_blob;
  mutable std::mutex _message_state_mutex;
  bool _cross = false;
  bool _connected = false;
  int _receive_timeout = DEFAULT_RECEIVE_TIMEOUT_MS;
  int _settings_timeout = 0;
  bool _init_done = false;
  bool _settings_fetched = false;
  bool _restart = false;
  std::shared_ptr<Mads::Runtime> _runtime = std::make_shared<Mads::Runtime>();
  // Per-agent stop request: set by shutdown()/disconnect(), cleared by
  // connect(). Keeps an individual agent's teardown from stopping the other
  // agents that share its Runtime.
  std::atomic<bool> _stopping{false};
  bool _remote_controlled = false;
  std::chrono::nanoseconds _time_step = std::chrono::nanoseconds(0);
  bool _high_res_loop = false;
  std::chrono::nanoseconds _spin_margin = std::chrono::microseconds(200);
  double _timecode_offset = 0.0;
  std::filesystem::path _attachment_path;
  bool _crypto = false;
  bool _conflate = false;
  std::unique_ptr<CurveAuth> _curve_auth = nullptr;
  std::filesystem::path _key_dir;
  bool _last_value_only = false;
  bool _shutdown_done = false;
  SharedLatest<zmq::multipart_t> _latest_message;
  // The agent's one socket thread (ZMQ_DEVELOPMENT.md §4.1). It always polls
  // both socket monitors' PAIR sockets -- which used to cost a thread each --
  // and additionally owns _subscriber exclusively whenever LKV delivery
  // and/or threaded remote control need it consumed off the application
  // thread. A single thread rather than one per feature: two threads calling
  // recv() on the same (non-thread-safe) ZMQ socket is undefined behaviour,
  // and with LKV and threaded remote control both enabled it also meant a
  // message landed on whichever thread's recv() call won the race.
  std::thread _io_thread;
  // Whether _io_thread polls _subscriber at all. Set by connect_sub() only
  // once the socket is fully subscribed, and read by _io_thread, so it must
  // be atomic even though it never changes after that.
  std::atomic<bool> _io_reads_subscriber{false};
  std::thread _watchdog_thread;
  // Delayed startup-event publisher: owned (not detached) so shutdown() can
  // wake it via _event_cv and join it before the sockets close.
  std::thread _startup_event_thread;
  std::mutex _event_mtx;
  std::condition_variable _event_cv;
  // ZMQ sockets are not thread-safe; the event thread publishes concurrently
  // with the owner thread, so sends on _publisher are serialized.
  std::mutex _publish_mutex;
  std::atomic<bool> _watchdog_stop{false};
  bool _rc_owns_socket = false;
  WireFormat _wire_format = WireFormat::Json;
  Compression _compression = Compression::Auto;
  std::atomic<size_t> _dropped_messages{0};
  nlohmann::json _settings_json; // cached JSON projection of settings
 
  // ---- Clock offset (see clock_offset.hpp) --------------------------
  Mads::ClockSource _clock_source = Mads::ClockSource::Broker;
  bool _clock_sync_responder = true;
  int _clock_interval_ms = 0;
  int _clock_announce_ms = 5000;
  bool _clock_correction = false;
  // Per-clock-domain adoption (§2 of the design): every agent, regardless
  // of its own clock_source, records what it hears so clock_offset() can
  // return the domain's winner rather than only this agent's own
  // measurement.
  Mads::ClockConsensus _clock_consensus{std::chrono::seconds(30)};
  // This agent's own last measurement (source Broker or Peer), guarded
  // separately from _clock_consensus's own internal mutex since it is
  // read/written by measure_clock_offset()/_run_peer_measurement() (the
  // clock thread or init()) and read by clock_offset() (any thread).
  mutable std::mutex _clock_mtx;
  Mads::ClockOffsetResult _own_clock_measurement;
  uint64_t _clock_seq = 0;
  std::thread _clock_thread;
  // Pending broadcast_clock_probe() collection state: the send instant (for
  // local_elapsed_us) and the pongs gathered so far. Single-flight by
  // design (see broadcast_clock_probe()'s doc comment).
  mutable std::mutex _clocksync_mtx;
  std::chrono::steady_clock::time_point _clocksync_probe_sent_at{};
  std::vector<Mads::ClockPeerObservation> _clocksync_pongs;
  uint64_t _clocksync_probe_seq = 0;
  // Per-initiator rate limit for this agent's own ping responses (§4.1 of
  // the design): guarded by _clocksync_mtx too.
  std::map<std::string, std::chrono::steady_clock::time_point>
      _clocksync_last_reply;
 
public:
  bool dummy = false;
};
 
} // namespace Mads
 
#endif // AGENT_HPP