ceph/src/crimson/osd/watch.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #pragma once
   5
   6 #include <iterator>
   7 #include <map>
   8 #include <set>
   9
  10 #include <seastar/core/shared_ptr.hh>
  11
  12 #include "crimson/net/Connection.h"
  13 #include "crimson/osd/object_context.h"
  14 #include "crimson/osd/pg.h"
  15 #include "include/denc.h"
  16
  17 namespace crimson::osd {
  18
  19 class Notify;
  20 using NotifyRef = seastar::shared_ptr<Notify>;
  21
  22 // NOTE: really need to have this public. Otherwise `shared_from_this()`
  23 // will abort. According to cppreference.com:
  24 //
  25 //   "The constructors of std::shared_ptr detect the presence
  26 //   of an unambiguous and accessible (ie. public inheritance
  27 //   is mandatory) (since C++17) enable_shared_from_this base".
  28 //
  29 // I expect the `seastar::shared_ptr` shares this behaviour.
  30 class Watch : public seastar::enable_shared_from_this<Watch> {
  31   // this is a private tag for the public constructor that turns it into
  32   // de facto private one. The motivation behind the hack is make_shared
  33   // used by create().
  34   struct private_ctag_t{};
  35
  36   std::set<NotifyRef, std::less<>> in_progress_notifies;
  37   crimson::net::ConnectionRef conn;
  38   crimson::osd::ObjectContextRef obc;
  39
  40   watch_info_t winfo;
  41   entity_name_t entity_name;
  42   Ref<PG> pg;
  43
  44   seastar::timer<seastar::lowres_clock> timeout_timer;
  45
  46   seastar::future<> start_notify(NotifyRef);
  47   seastar::future<> send_notify_msg(NotifyRef);
  48   seastar::future<> send_disconnect_msg();
  49
  50   friend Notify;
  51   friend class WatchTimeoutRequest;
  52
  53 public:
  54   Watch(private_ctag_t,
  55         crimson::osd::ObjectContextRef obc,
  56         const watch_info_t& winfo,
  57         const entity_name_t& entity_name,
  58         Ref<PG> pg)
  59     : obc(std::move(obc)),
  60       winfo(winfo),
  61       entity_name(entity_name),
  62       pg(std::move(pg)),
  63       timeout_timer([this] {
  64         return do_watch_timeout();
  65       }) {
  66     assert(this->pg);
  67   }
  68   ~Watch();
  69
  70   seastar::future<> connect(crimson::net::ConnectionRef, bool);
  71   void disconnect();
  72   bool is_alive() const {
  73     return true;
  74   }
  75   bool is_connected() const {
  76     return static_cast<bool>(conn);
  77   }
  78   void got_ping(utime_t);
  79
  80   void discard_state();
  81
  82   seastar::future<> remove();
  83
  84   /// Call when notify_ack received on notify_id
  85   seastar::future<> notify_ack(
  86     uint64_t notify_id, ///< [in] id of acked notify
  87     const ceph::bufferlist& reply_bl); ///< [in] notify reply buffer
  88
  89   template <class... Args>
  90   static seastar::shared_ptr<Watch> create(Args&&... args) {
  91     return seastar::make_shared<Watch>(private_ctag_t{},
  92                                        std::forward<Args>(args)...);
  93   };
  94
  95   uint64_t get_watcher_gid() const {
  96     return entity_name.num();
  97   }
  98   auto get_pg() const {
  99     return pg;
 100   }
 101   auto& get_entity() const {
 102     return entity_name;
 103   }
 104   auto& get_cookie() const {
 105     return winfo.cookie;
 106   }
 107   auto& get_peer_addr() const {
 108     return winfo.addr;
 109   }
 110   void cancel_notify(const uint64_t notify_id);
 111   void do_watch_timeout();
 112 };
 113
 114 using WatchRef = seastar::shared_ptr<Watch>;
 115
 116 struct notify_reply_t {
 117   uint64_t watcher_gid;
 118   uint64_t watcher_cookie;
 119   ceph::bufferlist bl;
 120
 121   bool operator<(const notify_reply_t& rhs) const;
 122   DENC(notify_reply_t, v, p) {
 123     // there is no versioning / preamble
 124     denc(v.watcher_gid, p);
 125     denc(v.watcher_cookie, p);
 126     denc(v.bl, p);
 127   }
 128 };
 129 std::ostream &operator<<(std::ostream &out, const notify_reply_t &rhs);
 130
 131 class Notify : public seastar::enable_shared_from_this<Notify> {
 132   std::set<WatchRef> watchers;
 133   const notify_info_t ninfo;
 134   crimson::net::ConnectionRef conn;
 135   const uint64_t client_gid;
 136   const uint64_t user_version;
 137   bool complete{false};
 138   bool discarded{false};
 139   seastar::timer<seastar::lowres_clock> timeout_timer{
 140     [this] { do_notify_timeout(); }
 141   };
 142
 143   ~Notify();
 144
 145   /// (gid,cookie) -> reply_bl for everyone who acked the notify
 146   std::multiset<notify_reply_t> notify_replies;
 147
 148   uint64_t get_id() const { return ninfo.notify_id; }
 149
 150   /// Sends notify completion if watchers.empty() or timeout
 151   seastar::future<> send_completion(
 152     std::set<WatchRef> timedout_watchers = {});
 153
 154   /// Called on Notify timeout
 155   void do_notify_timeout();
 156
 157   Notify(crimson::net::ConnectionRef conn,
 158          const notify_info_t& ninfo,
 159          const uint64_t client_gid,
 160          const uint64_t user_version);
 161   template <class WatchIteratorT>
 162   Notify(WatchIteratorT begin,
 163          WatchIteratorT end,
 164          crimson::net::ConnectionRef conn,
 165          const notify_info_t& ninfo,
 166          const uint64_t client_gid,
 167          const uint64_t user_version);
 168   // this is a private tag for the public constructor that turns it into
 169   // de facto private one. The motivation behind the hack is make_shared
 170   // used by create_n_propagate factory.
 171   struct private_ctag_t{};
 172
 173   using ptr_t = seastar::shared_ptr<Notify>;
 174   friend bool operator<(const ptr_t& lhs, const ptr_t& rhs) {
 175     assert(lhs);
 176     assert(rhs);
 177     return lhs->get_id() < rhs->get_id();
 178   }
 179   friend bool operator<(const ptr_t& ptr, const uint64_t id) {
 180     assert(ptr);
 181     return ptr->get_id() < id;
 182   }
 183   friend bool operator<(const uint64_t id, const ptr_t& ptr) {
 184     assert(ptr);
 185     return id < ptr->get_id();
 186   }
 187
 188   friend Watch;
 189
 190 public:
 191   template <class... Args>
 192   Notify(private_ctag_t, Args&&... args) : Notify(std::forward<Args>(args)...) {
 193   }
 194
 195   template <class WatchIteratorT, class... Args>
 196   static seastar::future<> create_n_propagate(
 197     WatchIteratorT begin,
 198     WatchIteratorT end,
 199     Args&&... args);
 200
 201   seastar::future<> remove_watcher(WatchRef watch);
 202   seastar::future<> complete_watcher(WatchRef watch,
 203                                      const ceph::bufferlist& reply_bl);
 204 };
 205
 206
 207 template <class WatchIteratorT>
 208 Notify::Notify(WatchIteratorT begin,
 209                WatchIteratorT end,
 210                crimson::net::ConnectionRef conn,
 211                const notify_info_t& ninfo,
 212                const uint64_t client_gid,
 213                const uint64_t user_version)
 214   : watchers(begin, end),
 215     ninfo(ninfo),
 216     conn(std::move(conn)),
 217     client_gid(client_gid),
 218     user_version(user_version) {
 219   assert(!std::empty(watchers));
 220   if (ninfo.timeout) {
 221     timeout_timer.arm(std::chrono::seconds{ninfo.timeout});
 222   }
 223 }
 224
 225 template <class WatchIteratorT, class... Args>
 226 seastar::future<> Notify::create_n_propagate(
 227   WatchIteratorT begin,
 228   WatchIteratorT end,
 229   Args&&... args)
 230 {
 231   static_assert(
 232     std::is_same_v<typename std::iterator_traits<WatchIteratorT>::value_type,
 233                    crimson::osd::WatchRef>);
 234   if (begin == end) {
 235     auto notify = seastar::make_shared<Notify>(
 236       private_ctag_t{},
 237       std::forward<Args>(args)...);
 238     return notify->send_completion();
 239   } else {
 240     auto notify = seastar::make_shared<Notify>(
 241       private_ctag_t{},
 242       begin, end,
 243       std::forward<Args>(args)...);
 244     return seastar::do_for_each(begin, end, [=] (auto& watchref) {
 245       return watchref->start_notify(notify);
 246     });
 247   }
 248 }
 249
 250 } // namespace crimson::osd
 251
 252 WRITE_CLASS_DENC(crimson::osd::notify_reply_t)
 253
 254 #if FMT_VERSION >= 90000
 255 template <> struct fmt::formatter<crimson::osd::notify_reply_t> : fmt::ostream_formatter {};
 256 #endif