ceph/src/tools/rbd_mirror/InstanceWatcher.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_RBD_MIRROR_INSTANCE_WATCHER_H
   5 #define CEPH_RBD_MIRROR_INSTANCE_WATCHER_H
   6
   7 #include <map>
   8 #include <memory>
   9 #include <set>
  10 #include <string>
  11 #include <vector>
  12
  13 #include "common/AsyncOpTracker.h"
  14 #include "librbd/Watcher.h"
  15 #include "librbd/managed_lock/Types.h"
  16 #include "tools/rbd_mirror/instance_watcher/Types.h"
  17
  18 namespace librbd {
  19
  20 class ImageCtx;
  21 template <typename> class ManagedLock;
  22
  23 }
  24
  25 namespace rbd {
  26 namespace mirror {
  27
  28 template <typename> class ImageSyncThrottler;
  29 template <typename> class InstanceReplayer;
  30 template <typename> struct Threads;
  31
  32 template <typename ImageCtxT = librbd::ImageCtx>
  33 class InstanceWatcher : protected librbd::Watcher {
  34 public:
  35   static void get_instances(librados::IoCtx &io_ctx,
  36                             std::vector<std::string> *instance_ids,
  37                             Context *on_finish);
  38   static void remove_instance(librados::IoCtx &io_ctx,
  39                               ContextWQ *work_queue,
  40                               const std::string &instance_id,
  41                               Context *on_finish);
  42
  43   static InstanceWatcher *create(
  44     librados::IoCtx &io_ctx, ContextWQ *work_queue,
  45     InstanceReplayer<ImageCtxT> *instance_replayer);
  46   void destroy() {
  47     delete this;
  48   }
  49
  50   InstanceWatcher(librados::IoCtx &io_ctx, ContextWQ *work_queue,
  51                   InstanceReplayer<ImageCtxT> *instance_replayer,
  52                   const std::string &instance_id);
  53   ~InstanceWatcher() override;
  54
  55   inline std::string &get_instance_id() {
  56     return m_instance_id;
  57   }
  58
  59   int init();
  60   void shut_down();
  61
  62   void init(Context *on_finish);
  63   void shut_down(Context *on_finish);
  64   void remove(Context *on_finish);
  65
  66   void notify_image_acquire(const std::string &instance_id,
  67                             const std::string &global_image_id,
  68                             Context *on_notify_ack);
  69   void notify_image_release(const std::string &instance_id,
  70                             const std::string &global_image_id,
  71                             Context *on_notify_ack);
  72   void notify_peer_image_removed(const std::string &instance_id,
  73                                  const std::string &global_image_id,
  74                                  const std::string &peer_mirror_uuid,
  75                                  Context *on_notify_ack);
  76
  77   void notify_sync_request(const std::string &sync_id, Context *on_sync_start);
  78   bool cancel_sync_request(const std::string &sync_id);
  79   void notify_sync_complete(const std::string &sync_id);
  80
  81   void print_sync_status(Formatter *f, stringstream *ss);
  82
  83   void cancel_notify_requests(const std::string &instance_id);
  84
  85   void handle_acquire_leader();
  86   void handle_release_leader();
  87   void handle_update_leader(const std::string &leader_instance_id);
  88
  89 private:
  90   /**
  91    * @verbatim
  92    *
  93    *       BREAK_INSTANCE_LOCK -------\
  94    *          ^                       |
  95    *          |               (error) |
  96    *       GET_INSTANCE_LOCKER  * * *>|
  97    *          ^ (remove)              |
  98    *          |                       |
  99    * <uninitialized> <----------------+---- WAIT_FOR_NOTIFY_OPS
 100    *    | (init)         ^            |        ^
 101    *    v        (error) *            |        |
 102    * REGISTER_INSTANCE * *     * * * *|* *> UNREGISTER_INSTANCE
 103    *    |                      *      |        ^
 104    *    v              (error) *      v        |
 105    * CREATE_INSTANCE_OBJECT  * *   * * * *> REMOVE_INSTANCE_OBJECT
 106    *    |                          *           ^
 107    *    v           (error)        *           |
 108    * REGISTER_WATCH  * * * * * * * *   * *> UNREGISTER_WATCH
 109    *    |                              *       ^
 110    *    v         (error)              *       |
 111    * ACQUIRE_LOCK  * * * * * * * * * * *    RELEASE_LOCK
 112    *    |                                      ^
 113    *    v       (shut_down)                    |
 114    * <watching> -------------------------------/
 115    *
 116    * @endverbatim
 117    */
 118
 119   struct C_NotifyInstanceRequest;
 120   struct C_SyncRequest;
 121
 122   typedef std::pair<std::string, std::string> Id;
 123
 124   struct HandlePayloadVisitor : public boost::static_visitor<void> {
 125     InstanceWatcher *instance_watcher;
 126     std::string instance_id;
 127     C_NotifyAck *on_notify_ack;
 128
 129     HandlePayloadVisitor(InstanceWatcher *instance_watcher,
 130                          const std::string &instance_id,
 131                          C_NotifyAck *on_notify_ack)
 132       : instance_watcher(instance_watcher), instance_id(instance_id),
 133         on_notify_ack(on_notify_ack) {
 134     }
 135
 136     template <typename Payload>
 137     inline void operator()(const Payload &payload) const {
 138       instance_watcher->handle_payload(instance_id, payload, on_notify_ack);
 139     }
 140   };
 141
 142   struct Request {
 143     std::string instance_id;
 144     uint64_t request_id;
 145     C_NotifyAck *on_notify_ack = nullptr;
 146
 147     Request(const std::string &instance_id, uint64_t request_id)
 148       : instance_id(instance_id), request_id(request_id) {
 149     }
 150
 151     inline bool operator<(const Request &rhs) const {
 152       return instance_id < rhs.instance_id ||
 153         (instance_id == rhs.instance_id && request_id < rhs.request_id);
 154     }
 155   };
 156
 157   Threads<ImageCtxT> *m_threads;
 158   InstanceReplayer<ImageCtxT> *m_instance_replayer;
 159   std::string m_instance_id;
 160
 161   mutable Mutex m_lock;
 162   librbd::ManagedLock<ImageCtxT> *m_instance_lock;
 163   Context *m_on_finish = nullptr;
 164   int m_ret_val = 0;
 165   bool m_removing = false;
 166   std::string m_leader_instance_id;
 167   librbd::managed_lock::Locker m_instance_locker;
 168   std::set<std::pair<std::string, C_NotifyInstanceRequest *>> m_notify_ops;
 169   AsyncOpTracker m_notify_op_tracker;
 170   uint64_t m_request_seq = 0;
 171   std::set<Request> m_requests;
 172   std::set<C_NotifyInstanceRequest *> m_suspended_ops;
 173   std::map<std::string, C_SyncRequest *> m_inflight_sync_reqs;
 174   ImageSyncThrottler<ImageCtxT> *m_image_sync_throttler = nullptr;
 175
 176   void register_instance();
 177   void handle_register_instance(int r);
 178
 179   void create_instance_object();
 180   void handle_create_instance_object(int r);
 181
 182   void register_watch();
 183   void handle_register_watch(int r);
 184
 185   void acquire_lock();
 186   void handle_acquire_lock(int r);
 187
 188   void release_lock();
 189   void handle_release_lock(int r);
 190
 191   void unregister_watch();
 192   void handle_unregister_watch(int r);
 193
 194   void remove_instance_object();
 195   void handle_remove_instance_object(int r);
 196
 197   void unregister_instance();
 198   void handle_unregister_instance(int r);
 199
 200   void wait_for_notify_ops();
 201   void handle_wait_for_notify_ops(int r);
 202
 203   void get_instance_locker();
 204   void handle_get_instance_locker(int r);
 205
 206   void break_instance_lock();
 207   void handle_break_instance_lock(int r);
 208
 209   void suspend_notify_request(C_NotifyInstanceRequest *req);
 210   bool unsuspend_notify_request(C_NotifyInstanceRequest *req);
 211   void unsuspend_notify_requests();
 212
 213   void notify_sync_complete(const Mutex& lock, const std::string &sync_id);
 214   void handle_notify_sync_request(C_SyncRequest *sync_ctx, int r);
 215   void handle_notify_sync_complete(C_SyncRequest *sync_ctx, int r);
 216
 217   void notify_sync_start(const std::string &instance_id,
 218                          const std::string &sync_id);
 219
 220   Context *prepare_request(const std::string &instance_id, uint64_t request_id,
 221                            C_NotifyAck *on_notify_ack);
 222   void complete_request(const std::string &instance_id, uint64_t request_id,
 223                         int r);
 224
 225   void handle_notify(uint64_t notify_id, uint64_t handle,
 226                      uint64_t notifier_id, bufferlist &bl) override;
 227
 228   void handle_image_acquire(const std::string &global_image_id,
 229                             Context *on_finish);
 230   void handle_image_release(const std::string &global_image_id,
 231                             Context *on_finish);
 232   void handle_peer_image_removed(const std::string &global_image_id,
 233                                  const std::string &peer_mirror_uuid,
 234                                  Context *on_finish);
 235
 236   void handle_sync_request(const std::string &instance_id,
 237                            const std::string &sync_id, Context *on_finish);
 238   void handle_sync_start(const std::string &instance_id,
 239                          const std::string &sync_id, Context *on_finish);
 240
 241   void handle_payload(const std::string &instance_id,
 242                       const instance_watcher::ImageAcquirePayload &payload,
 243                       C_NotifyAck *on_notify_ack);
 244   void handle_payload(const std::string &instance_id,
 245                       const instance_watcher::ImageReleasePayload &payload,
 246                       C_NotifyAck *on_notify_ack);
 247   void handle_payload(const std::string &instance_id,
 248                       const instance_watcher::PeerImageRemovedPayload &payload,
 249                       C_NotifyAck *on_notify_ack);
 250   void handle_payload(const std::string &instance_id,
 251                       const instance_watcher::SyncRequestPayload &payload,
 252                       C_NotifyAck *on_notify_ack);
 253   void handle_payload(const std::string &instance_id,
 254                       const instance_watcher::SyncStartPayload &payload,
 255                       C_NotifyAck *on_notify_ack);
 256   void handle_payload(const std::string &instance_id,
 257                       const instance_watcher::UnknownPayload &payload,
 258                       C_NotifyAck *on_notify_ack);
 259 };
 260
 261 } // namespace mirror
 262 } // namespace rbd
 263
 264 #endif // CEPH_RBD_MIRROR_INSTANCE_WATCHER_H