1 files changed, 121 insertions, 0 deletions
diff --git a/src/crimson/osd/heartbeat.h b/src/crimson/osd/heartbeat.h
new file mode 100644
index 00000000..b5eb0f7c
--- /dev/null
+++ b/src/crimson/osd/heartbeat.h
@@ -0,0 +1,121 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <cstdint>
+#include <seastar/core/future.hh>
+#include "common/ceph_time.h"
+#include "crimson/net/Dispatcher.h"
+#include "crimson/net/Fwd.h"
+
+class MOSDPing;
+class OSDMapService;
+
+namespace ceph::mon {
+  class Client;
+}
+
+template<typename Message> using Ref = boost::intrusive_ptr<Message>;
+
+class Heartbeat : public ceph::net::Dispatcher {
+public:
+  using osd_id_t = int;
+
+  Heartbeat(int whoami,
+	    uint32_t nonce,
+	    const OSDMapService& service,
+	    ceph::mon::Client& monc);
+
+  seastar::future<> start(entity_addrvec_t front,
+			  entity_addrvec_t back);
+  seastar::future<> stop();
+
+  seastar::future<> add_peer(osd_id_t peer, epoch_t epoch);
+  seastar::future<> update_peers(int whoami);
+  seastar::future<> remove_peer(osd_id_t peer);
+
+  seastar::future<> send_heartbeats();
+  seastar::future<> send_failures();
+
+  const entity_addrvec_t& get_front_addrs() const;
+  const entity_addrvec_t& get_back_addrs() const;
+
+  // Dispatcher methods
+  seastar::future<> ms_dispatch(ceph::net::ConnectionRef conn,
+				MessageRef m) override;
+
+private:
+  seastar::future<> handle_osd_ping(ceph::net::ConnectionRef conn,
+				    Ref<MOSDPing> m);
+  seastar::future<> handle_ping(ceph::net::ConnectionRef conn,
+				Ref<MOSDPing> m);
+  seastar::future<> handle_reply(ceph::net::ConnectionRef conn,
+				 Ref<MOSDPing> m);
+  seastar::future<> handle_you_died();
+
+  seastar::future<> send_still_alive(osd_id_t, const entity_addrvec_t&);
+
+  using osds_t = std::vector<osd_id_t>;
+  /// remove down OSDs
+  /// @return peers not needed in this epoch
+  seastar::future<osds_t> remove_down_peers();
+  /// add enough reporters for fast failure detection
+  void add_reporter_peers(int whoami);
+
+  seastar::future<> start_messenger(ceph::net::Messenger* msgr,
+				    const entity_addrvec_t& addrs);
+private:
+  const int whoami;
+  const uint32_t nonce;
+  ceph::net::Messenger* front_msgr = nullptr;
+  ceph::net::Messenger* back_msgr = nullptr;
+  const OSDMapService& service;
+  ceph::mon::Client& monc;
+
+  seastar::timer<seastar::lowres_clock> timer;
+  // use real_clock so it can be converted to utime_t
+  using clock = ceph::coarse_real_clock;
+
+  struct reply_t {
+    clock::time_point deadline;
+    // one sent over front conn, another sent over back conn
+    uint8_t unacknowledged = 0;
+  };
+  struct PeerInfo {
+    /// peer connection (front)
+    ceph::net::ConnectionRef con_front;
+    /// peer connection (back)
+    ceph::net::ConnectionRef con_back;
+    /// time we sent our first ping request
+    clock::time_point first_tx;
+    /// last time we sent a ping request
+    clock::time_point last_tx;
+    /// last time we got a ping reply on the front side
+    clock::time_point last_rx_front;
+    /// last time we got a ping reply on the back side
+    clock::time_point last_rx_back;
+    /// most recent epoch we wanted this peer
+    epoch_t epoch;
+    /// history of inflight pings, arranging by timestamp we sent
+    std::map<utime_t, reply_t> ping_history;
+
+    bool is_unhealthy(clock::time_point now) const;
+    bool is_healthy(clock::time_point now) const;
+  };
+  using peers_map_t = std::map<osd_id_t, PeerInfo>;
+  peers_map_t peers;
+
+  // osds which are considered failed
+  // osd_id => when was the last time that both front and back pings were acked
+  //           use for calculating how long the OSD has been unresponsive
+  using failure_queue_t = std::map<osd_id_t, clock::time_point>;
+  failure_queue_t failure_queue;
+  struct failure_info_t {
+    clock::time_point failed_since;
+    entity_addrvec_t addrs;
+  };
+  // osds we've reported to monior as failed ones, but they are not marked down
+  // yet
+  std::map<osd_id_t, failure_info_t> failure_pending;
+};