clock_offset.hpp

Go to the documentation of this file.

/*
   ____ _            _      ___   __  __          _
  / ___| | ___   ___| | __ / _ \ / _|/ _|___  ___| |_
 | |   | |/ _ \ / __| |/ /| | | | |_| |_/ __|/ _ \ __|
 | |___| | (_) | (__|   < | |_| |  _|  _\__ \  __/ |_
  \____|_|\___/ \___|_|\_\ \___/|_| |_| |___/\___|\__|
 
Pure, socket-free clock-offset math and cross-agent consensus, shared by the
broker exchange (source A) and the bus ping/pong (source B) that measure
`Agent::clock_offset()`. No ZMQ, no Agent dependency: callers pass in
timestamps and a `now`, exactly like Mads::TopicStats, so the NTP-style
four-timestamp estimate, the domain-consensus rule, and the passive skew
tracker used by `mads top` are all directly unit-testable with synthetic
values -- no sockets, no real sleeps.
 
Author(s): Paolo Bosetti
*/
#pragma once
 
#include <chrono>
#include <cstdint>
#include <deque>
#include <map>
#include <mutex>
#include <string>
#include <vector>
 
namespace Mads {
 
struct ClockSample {
  int64_t t1 = 0, t2 = 0, t3 = 0, t4 = 0;
  int64_t local_elapsed_us = 0;
};
 
enum class ClockSource { None = 0, Broker = 1, Peer = 2 };
 
struct ClockOffsetResult {
  int64_t offset_us = 0;
  int64_t delay_us = 0;
  int64_t jitter_us = 0;
  size_t samples = 0;
  uint8_t hops = 0;
  ClockSource source = ClockSource::None;
  std::string origin_agent_id;
  uint64_t seq = 0;
  std::chrono::steady_clock::time_point measured_at{};
  bool valid = false;
 
  std::string clock_ref() const {
    return origin_agent_id + ":" + std::to_string(seq);
  }
};
 
ClockOffsetResult estimate(const ClockSample &s);
 
class ClockOffsetEstimator {
public:
  explicit ClockOffsetEstimator(size_t keep = 8);
 
  void add(const ClockSample &s);
 
  ClockOffsetResult best() const;
 
  ClockOffsetResult median() const;
 
  void reset();
 
private:
  int64_t jitter_us() const;
  size_t _keep;
  std::vector<ClockOffsetResult> _results;
};
 
class ClockConsensus {
public:
  explicit ClockConsensus(
      std::chrono::milliseconds stale = std::chrono::seconds(30));
 
  void record(const std::string &domain, const ClockOffsetResult &r,
              std::chrono::steady_clock::time_point now =
                  std::chrono::steady_clock::now());
 
  ClockOffsetResult adopted(const std::string &domain,
                            std::chrono::steady_clock::time_point now =
                                std::chrono::steady_clock::now()) const;
 
  bool is_winner(const std::string &domain, const std::string &agent_id,
                std::chrono::steady_clock::time_point now =
                    std::chrono::steady_clock::now()) const;
 
  void clear();
 
private:
  mutable std::mutex _mtx;
  std::chrono::milliseconds _stale;
  // domain -> (origin_agent_id -> its latest announced/measured result).
  std::map<std::string, std::map<std::string, ClockOffsetResult>> _domains;
};
 
struct HostSkewStat {
  std::string host;
  int64_t min_skew_us = 0;
  bool has_value = false;
  size_t samples = 0;
  std::chrono::steady_clock::time_point last_seen{};
};
 
class HostSkewStats {
public:
  explicit HostSkewStats(
      std::chrono::milliseconds window = std::chrono::seconds(10));
 
  void record(const std::string &host, int64_t skew_us,
              std::chrono::steady_clock::time_point now =
                  std::chrono::steady_clock::now());
 
  std::vector<HostSkewStat>
  snapshot(std::chrono::steady_clock::time_point now =
               std::chrono::steady_clock::now()) const;
 
  void clear();
 
private:
  struct Entry {
    std::deque<std::pair<std::chrono::steady_clock::time_point, int64_t>>
        events;
    std::chrono::steady_clock::time_point last_seen{};
  };
 
  mutable std::mutex _mtx;
  std::chrono::milliseconds _window;
  std::map<std::string, Entry> _entries;
};
 
} // namespace Mads