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   /// (gid,cookie) -> reply_bl for everyone who acked the notify
 144   std::multiset<notify_reply_t> notify_replies;
 145
 146   uint64_t get_id() const { return ninfo.notify_id; }
 147
 148   /// Sends notify completion if watchers.empty() or timeout
 149   seastar::future<> send_completion(
 150     std::set<WatchRef> timedout_watchers = {});
 151
 152   /// Called on Notify timeout
 153   void do_notify_timeout();
 154
 155   Notify(crimson::net::ConnectionRef conn,
 156          const notify_info_t& ninfo,
 157          const uint64_t client_gid,
 158          const uint64_t user_version);
 159   template <class WatchIteratorT>
 160   Notify(WatchIteratorT begin,
 161          WatchIteratorT end,
 162          crimson::net::ConnectionRef conn,
 163          const notify_info_t& ninfo,
 164          const uint64_t client_gid,
 165          const uint64_t user_version);
 166   // this is a private tag for the public constructor that turns it into
 167   // de facto private one. The motivation behind the hack is make_shared
 168   // used by create_n_propagate factory.
 169   struct private_ctag_t{};
 170
 171   using ptr_t = seastar::shared_ptr<Notify>;
 172   friend bool operator<(const ptr_t& lhs, const ptr_t& rhs) {
 173     assert(lhs);
 174     assert(rhs);
 175     return lhs->get_id() < rhs->get_id();
 176   }
 177   friend bool operator<(const ptr_t& ptr, const uint64_t id) {
 178     assert(ptr);
 179     return ptr->get_id() < id;
 180   }
 181   friend bool operator<(const uint64_t id, const ptr_t& ptr) {
 182     assert(ptr);
 183     return id < ptr->get_id();
 184   }
 185
 186   friend Watch;
 187
 188 public:
 189   template <class... Args>
 190   Notify(private_ctag_t, Args&&... args) : Notify(std::forward<Args>(args)...) {
 191   }
 192
 193   template <class WatchIteratorT, class... Args>
 194   static seastar::future<> create_n_propagate(
 195     WatchIteratorT begin,
 196     WatchIteratorT end,
 197     Args&&... args);
 198
 199   seastar::future<> remove_watcher(WatchRef watch);
 200   seastar::future<> complete_watcher(WatchRef watch,
 201                                      const ceph::bufferlist& reply_bl);
 202 };
 203
 204
 205 template <class WatchIteratorT>
 206 Notify::Notify(WatchIteratorT begin,
 207                WatchIteratorT end,
 208                crimson::net::ConnectionRef conn,
 209                const notify_info_t& ninfo,
 210                const uint64_t client_gid,
 211                const uint64_t user_version)
 212   : watchers(begin, end),
 213     ninfo(ninfo),
 214     conn(std::move(conn)),
 215     client_gid(client_gid),
 216     user_version(user_version) {
 217   assert(!std::empty(watchers));
 218   if (ninfo.timeout) {
 219     timeout_timer.arm(std::chrono::seconds{ninfo.timeout});
 220   }
 221 }
 222
 223 template <class WatchIteratorT, class... Args>
 224 seastar::future<> Notify::create_n_propagate(
 225   WatchIteratorT begin,
 226   WatchIteratorT end,
 227   Args&&... args)
 228 {
 229   static_assert(
 230     std::is_same_v<typename std::iterator_traits<WatchIteratorT>::value_type,
 231                    crimson::osd::WatchRef>);
 232   if (begin == end) {
 233     auto notify = seastar::make_shared<Notify>(
 234       private_ctag_t{},
 235       std::forward<Args>(args)...);
 236     return notify->send_completion();
 237   } else {
 238     auto notify = seastar::make_shared<Notify>(
 239       private_ctag_t{},
 240       begin, end,
 241       std::forward<Args>(args)...);
 242     return seastar::do_for_each(begin, end, [=] (auto& watchref) {
 243       return watchref->start_notify(notify);
 244     });
 245   }
 246 }
 247
 248 } // namespace crimson::osd
 249
 250 WRITE_CLASS_DENC(crimson::osd::notify_reply_t)
 251
 252 #if FMT_VERSION >= 90000
 253 template <> struct fmt::formatter<crimson::osd::notify_reply_t> : fmt::ostream_formatter {};
 254 #endif