ceph/src/librbd/Watcher.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifndef CEPH_LIBRBD_WATCHER_H
   5 #define CEPH_LIBRBD_WATCHER_H
   6
   7 #include "common/AsyncOpTracker.h"
   8 #include "common/ceph_mutex.h"
   9 #include "common/RWLock.h"
  10 #include "include/rados/librados.hpp"
  11 #include "librbd/watcher/Notifier.h"
  12 #include "librbd/watcher/Types.h"
  13 #include <string>
  14 #include <utility>
  15
  16 class ContextWQ;
  17
  18 namespace librbd {
  19
  20 namespace watcher { struct NotifyResponse; }
  21
  22 class Watcher {
  23 public:
  24   struct C_NotifyAck : public Context {
  25     Watcher *watcher;
  26     CephContext *cct;
  27     uint64_t notify_id;
  28     uint64_t handle;
  29     bufferlist out;
  30
  31     C_NotifyAck(Watcher *watcher, uint64_t notify_id, uint64_t handle);
  32     void finish(int r) override;
  33   };
  34
  35   Watcher(librados::IoCtx& ioctx, ContextWQ *work_queue,
  36           const std::string& oid);
  37   virtual ~Watcher();
  38
  39   void register_watch(Context *on_finish);
  40   virtual void unregister_watch(Context *on_finish);
  41   void flush(Context *on_finish);
  42
  43   bool notifications_blocked() const;
  44   virtual void block_notifies(Context *on_finish);
  45   void unblock_notifies();
  46
  47   std::string get_oid() const;
  48   void set_oid(const string& oid);
  49
  50   uint64_t get_watch_handle() const {
  51     std::shared_lock watch_locker{m_watch_lock};
  52     return m_watch_handle;
  53   }
  54
  55   bool is_registered() const {
  56     std::shared_lock locker{m_watch_lock};
  57     return is_registered(m_watch_lock);
  58   }
  59   bool is_unregistered() const {
  60     std::shared_lock locker{m_watch_lock};
  61     return is_unregistered(m_watch_lock);
  62   }
  63   bool is_blacklisted() const {
  64     std::shared_lock locker{m_watch_lock};
  65     return m_watch_blacklisted;
  66   }
  67
  68 protected:
  69   enum WatchState {
  70     WATCH_STATE_IDLE,
  71     WATCH_STATE_REGISTERING,
  72     WATCH_STATE_REWATCHING
  73   };
  74
  75   librados::IoCtx& m_ioctx;
  76   ContextWQ *m_work_queue;
  77   std::string m_oid;
  78   CephContext *m_cct;
  79   mutable ceph::shared_mutex m_watch_lock;
  80   uint64_t m_watch_handle;
  81   watcher::Notifier m_notifier;
  82
  83   WatchState m_watch_state;
  84   bool m_watch_blacklisted = false;
  85
  86   AsyncOpTracker m_async_op_tracker;
  87
  88   bool is_registered(const ceph::shared_mutex&) const {
  89     return (m_watch_state == WATCH_STATE_IDLE && m_watch_handle != 0);
  90   }
  91   bool is_unregistered(const ceph::shared_mutex&) const {
  92     return (m_watch_state == WATCH_STATE_IDLE && m_watch_handle == 0);
  93   }
  94
  95   void send_notify(bufferlist &payload,
  96                    watcher::NotifyResponse *response = nullptr,
  97                    Context *on_finish = nullptr);
  98
  99   virtual void handle_notify(uint64_t notify_id, uint64_t handle,
 100                              uint64_t notifier_id, bufferlist &bl) = 0;
 101
 102   virtual void handle_error(uint64_t cookie, int err);
 103
 104   void acknowledge_notify(uint64_t notify_id, uint64_t handle,
 105                           bufferlist &out);
 106
 107   virtual void handle_rewatch_complete(int r) { }
 108
 109 private:
 110   /**
 111    * @verbatim
 112    *
 113    * <start>
 114    *    |
 115    *    v
 116    * UNREGISTERED
 117    *    |
 118    *    | (register_watch)
 119    *    |
 120    * REGISTERING
 121    *    |
 122    *    v      (watch error)
 123    * REGISTERED * * * * * * * > ERROR
 124    *    |   ^                     |
 125    *    |   |                     | (rewatch)
 126    *    |   |                     v
 127    *    |   |                   REWATCHING
 128    *    |   |                     |
 129    *    |   |                     |
 130    *    |   \---------------------/
 131    *    |
 132    *    | (unregister_watch)
 133    *    |
 134    *    v
 135    * UNREGISTERED
 136    *    |
 137    *    v
 138    * <finish>
 139    *
 140    * @endverbatim
 141    */
 142
 143   struct WatchCtx : public librados::WatchCtx2 {
 144     Watcher &watcher;
 145
 146     WatchCtx(Watcher &parent) : watcher(parent) {}
 147
 148     void handle_notify(uint64_t notify_id,
 149                        uint64_t handle,
 150                        uint64_t notifier_id,
 151                        bufferlist& bl) override;
 152     void handle_error(uint64_t handle, int err) override;
 153   };
 154
 155   struct C_RegisterWatch : public Context {
 156     Watcher *watcher;
 157     Context *on_finish;
 158
 159     C_RegisterWatch(Watcher *watcher, Context *on_finish)
 160        : watcher(watcher), on_finish(on_finish) {
 161     }
 162     void finish(int r) override {
 163       watcher->handle_register_watch(r, on_finish);
 164     }
 165   };
 166
 167   WatchCtx m_watch_ctx;
 168   Context *m_unregister_watch_ctx = nullptr;
 169
 170   bool m_watch_error = false;
 171
 172   uint32_t m_blocked_count = 0;
 173
 174   void handle_register_watch(int r, Context *on_finish);
 175
 176   void rewatch();
 177   void handle_rewatch(int r);
 178   void handle_rewatch_callback(int r);
 179
 180 };
 181
 182 } // namespace librbd
 183
 184 #endif // CEPH_LIBRBD_WATCHER_H