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/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     RWLock::RLocker watch_locker(m_watch_lock);
  52     return m_watch_handle;
  53   }
  54
  55   bool is_registered() const {
  56     RWLock::RLocker locker(m_watch_lock);
  57     return m_watch_state == WATCH_STATE_REGISTERED;
  58   }
  59   bool is_unregistered() const {
  60     RWLock::RLocker locker(m_watch_lock);
  61     return m_watch_state == WATCH_STATE_UNREGISTERED;
  62   }
  63
  64 protected:
  65   enum WatchState {
  66     WATCH_STATE_UNREGISTERED,
  67     WATCH_STATE_REGISTERING,
  68     WATCH_STATE_REGISTERED,
  69     WATCH_STATE_ERROR,
  70     WATCH_STATE_REWATCHING
  71   };
  72
  73   librados::IoCtx& m_ioctx;
  74   ContextWQ *m_work_queue;
  75   std::string m_oid;
  76   CephContext *m_cct;
  77   mutable RWLock m_watch_lock;
  78   uint64_t m_watch_handle;
  79   watcher::Notifier m_notifier;
  80   WatchState m_watch_state;
  81   AsyncOpTracker m_async_op_tracker;
  82
  83   void send_notify(bufferlist &payload,
  84                    watcher::NotifyResponse *response = nullptr,
  85                    Context *on_finish = nullptr);
  86
  87   virtual void handle_notify(uint64_t notify_id, uint64_t handle,
  88                              uint64_t notifier_id, bufferlist &bl) = 0;
  89
  90   virtual void handle_error(uint64_t cookie, int err);
  91
  92   void acknowledge_notify(uint64_t notify_id, uint64_t handle,
  93                           bufferlist &out);
  94
  95   virtual void handle_rewatch_complete(int r) { }
  96
  97 private:
  98   /**
  99    * @verbatim
 100    *
 101    * <start>
 102    *    |
 103    *    v
 104    * UNREGISTERED
 105    *    |
 106    *    | (register_watch)
 107    *    |
 108    * REGISTERING
 109    *    |
 110    *    v      (watch error)
 111    * REGISTERED * * * * * * * > ERROR
 112    *    |   ^                     |
 113    *    |   |                     | (rewatch)
 114    *    |   |                     v
 115    *    |   |                   REWATCHING
 116    *    |   |                     |
 117    *    |   |                     |
 118    *    |   \---------------------/
 119    *    |
 120    *    | (unregister_watch)
 121    *    |
 122    *    v
 123    * UNREGISTERED
 124    *    |
 125    *    v
 126    * <finish>
 127    *
 128    * @endverbatim
 129    */
 130
 131   struct WatchCtx : public librados::WatchCtx2 {
 132     Watcher &watcher;
 133
 134     WatchCtx(Watcher &parent) : watcher(parent) {}
 135
 136     void handle_notify(uint64_t notify_id,
 137                        uint64_t handle,
 138                        uint64_t notifier_id,
 139                        bufferlist& bl) override;
 140     void handle_error(uint64_t handle, int err) override;
 141   };
 142
 143   struct C_RegisterWatch : public Context {
 144     Watcher *watcher;
 145     Context *on_finish;
 146
 147     C_RegisterWatch(Watcher *watcher, Context *on_finish)
 148        : watcher(watcher), on_finish(on_finish) {
 149     }
 150     void finish(int r) override {
 151       watcher->handle_register_watch(r, on_finish);
 152     }
 153   };
 154
 155   WatchCtx m_watch_ctx;
 156   Context *m_unregister_watch_ctx = nullptr;
 157
 158   uint32_t m_blocked_count = 0;
 159
 160   void handle_register_watch(int r, Context *on_finish);
 161
 162   void rewatch();
 163   void handle_rewatch(int r);
 164
 165 };
 166
 167 } // namespace librbd
 168
 169 #endif // CEPH_LIBRBD_WATCHER_H