[ceph.git] / ceph / src / tools / rbd_mirror / LeaderWatcher.h

// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#ifndef CEPH_RBD_MIRROR_LEADER_WATCHER_H
#define CEPH_RBD_MIRROR_LEADER_WATCHER_H

#include <list>
#include <memory>
#include <string>

#include "common/AsyncOpTracker.h"
#include "librbd/ManagedLock.h"
#include "librbd/Watcher.h"
#include "librbd/managed_lock/Types.h"
#include "librbd/watcher/Types.h"
#include "Instances.h"
#include "tools/rbd_mirror/instances/Types.h"
#include "tools/rbd_mirror/leader_watcher/Types.h"

namespace librbd {
class ImageCtx;
namespace asio { struct ContextWQ; }
} // namespace librbd

namespace rbd {
namespace mirror {

template <typename> struct Threads;

template <typename ImageCtxT = librbd::ImageCtx>
class LeaderWatcher : protected librbd::Watcher {
  using librbd::Watcher::unregister_watch; // Silence overloaded virtual warning
public:
  static LeaderWatcher* create(Threads<ImageCtxT> *threads,
                               librados::IoCtx &io_ctx,
                               leader_watcher::Listener *listener) {
    return new LeaderWatcher(threads, io_ctx, listener);
  }

  LeaderWatcher(Threads<ImageCtxT> *threads, librados::IoCtx &io_ctx,
                leader_watcher::Listener *listener);
  ~LeaderWatcher() override;

  int init();
  void shut_down();

  void init(Context *on_finish);
  void shut_down(Context *on_finish);

  bool is_blocklisted() const;
  bool is_leader() const;
  bool is_releasing_leader() const;
  bool get_leader_instance_id(std::string *instance_id) const;
  void release_leader();
  void list_instances(std::vector<std::string> *instance_ids);

  std::string get_instance_id();

private:
  /**
   * @verbatim
   *
   *  <uninitialized> <------------------------------ WAIT_FOR_TASKS
   *     | (init)            ^                                ^
   *     v                   *                                |
   *  CREATE_OBJECT  * * * * *  (error)               UNREGISTER_WATCH
   *     |                   *                                ^
   *     v                   *                                |
   *  REGISTER_WATCH * * * * *                        SHUT_DOWN_LEADER_LOCK
   *     |                                                    ^
   *     |           (no leader heartbeat and acquire failed) |
   *     | BREAK_LOCK <-------------------------------------\ |
   *     |    |                 (no leader heartbeat)       | | (shut down)
   *     |    |  /----------------------------------------\ | |
   *     |    |  |              (lock_released received)    | |
   *     |    |  |  /-------------------------------------\ | |
   *     |    |  |  |                   (lock_acquired or | | |
   *     |    |  |  |                 heartbeat received) | | |
   *     |    |  |  |       (ENOENT)        /-----------\ | | |
   *     |    |  |  |  * * * * * * * * * *  |           | | | |
   *     v    v  v  v  v  (error)        *  v           | | | |
   *  ACQUIRE_LEADER_LOCK  * * * * *> GET_LOCKER ---> <secondary>
   *     |                   *                           ^
   * ....|...................*....................  .....|.....................
   * .   v                   *                   .  .    |       post_release .
   * .INIT_INSTANCES * * * * *                   .  .NOTIFY_LOCK_RELEASED     .
   * .   |                                       .  .....^.....................
   * .   v                                       .       |
   * .NOTIFY_LISTENER                            .   RELEASE_LEADER_LOCK
   * .   |                                       .       ^
   * .   v                                       .  .....|.....................
   * .NOTIFY_LOCK_ACQUIRED                       .  .    |                    .
   * .   |                          post_acquire .  .SHUT_DOWN_INSTANCES      .
   * ....|........................................  .    ^                    .
   *     v                                          .    |                    .
   *  <leader> -----------------------------------> .NOTIFY_LISTENER          .
   *            (shut_down, release_leader,         .             pre_release .
   *             notify error)                      ...........................
   * @endverbatim
   */

  struct InstancesListener : public instances::Listener {
    LeaderWatcher* leader_watcher;

    InstancesListener(LeaderWatcher* leader_watcher)
      : leader_watcher(leader_watcher) {
    }

    void handle_added(const InstanceIds& instance_ids) override {
      leader_watcher->m_listener->handle_instances_added(instance_ids);
    }

    void handle_removed(const InstanceIds& instance_ids) override {
      leader_watcher->m_listener->handle_instances_removed(instance_ids);
    }
  };

  class LeaderLock : public librbd::ManagedLock<ImageCtxT> {
  public:
    typedef librbd::ManagedLock<ImageCtxT> Parent;

    LeaderLock(librados::IoCtx& ioctx, librbd::AsioEngine& asio_engine,
               const std::string& oid, LeaderWatcher *watcher,
               bool blocklist_on_break_lock,
               uint32_t blocklist_expire_seconds)
      : Parent(ioctx, asio_engine, oid, watcher,
               librbd::managed_lock::EXCLUSIVE, blocklist_on_break_lock,
               blocklist_expire_seconds),
        watcher(watcher) {
    }

    bool is_leader() const {
      std::lock_guard locker{Parent::m_lock};
      return Parent::is_state_post_acquiring() || Parent::is_state_locked();
    }

    bool is_releasing_leader() const {
      std::lock_guard locker{Parent::m_lock};
      return Parent::is_state_pre_releasing();
    }

  protected:
    void post_acquire_lock_handler(int r, Context *on_finish) {
      if (r == 0) {
        // lock is owned at this point
	std::lock_guard locker{Parent::m_lock};
        Parent::set_state_post_acquiring();
      }
      watcher->handle_post_acquire_leader_lock(r, on_finish);
    }
    void pre_release_lock_handler(bool shutting_down,
                                  Context *on_finish) {
      watcher->handle_pre_release_leader_lock(on_finish);
    }
    void post_release_lock_handler(bool shutting_down, int r,
                                   Context *on_finish) {
      watcher->handle_post_release_leader_lock(r, on_finish);
    }
  private:
    LeaderWatcher *watcher;
  };

  struct HandlePayloadVisitor : public boost::static_visitor<void> {
    LeaderWatcher *leader_watcher;
    Context *on_notify_ack;

    HandlePayloadVisitor(LeaderWatcher *leader_watcher, Context *on_notify_ack)
      : leader_watcher(leader_watcher), on_notify_ack(on_notify_ack) {
    }

    template <typename Payload>
    inline void operator()(const Payload &payload) const {
      leader_watcher->handle_payload(payload, on_notify_ack);
    }
  };

  struct C_GetLocker : public Context {
    LeaderWatcher *leader_watcher;
    librbd::managed_lock::Locker locker;

    C_GetLocker(LeaderWatcher *leader_watcher)
      : leader_watcher(leader_watcher) {
    }

    void finish(int r) override {
      leader_watcher->handle_get_locker(r, locker);
    }
  };

  typedef void (LeaderWatcher<ImageCtxT>::*TimerCallback)();

  struct C_TimerGate : public Context {
    LeaderWatcher *leader_watcher;

    bool leader = false;
    TimerCallback timer_callback = nullptr;

    C_TimerGate(LeaderWatcher *leader_watcher)
      : leader_watcher(leader_watcher) {
    }

    void finish(int r) override {
      leader_watcher->m_timer_gate = nullptr;
      leader_watcher->execute_timer_task(leader, timer_callback);
    }
  };

  Threads<ImageCtxT> *m_threads;
  leader_watcher::Listener *m_listener;

  InstancesListener m_instances_listener;
  mutable ceph::mutex m_lock;
  uint64_t m_notifier_id;
  std::string m_instance_id;
  LeaderLock *m_leader_lock;
  Context *m_on_finish = nullptr;
  Context *m_on_shut_down_finish = nullptr;
  uint64_t m_acquire_attempts = 0;
  int m_ret_val = 0;
  Instances<ImageCtxT> *m_instances = nullptr;
  librbd::managed_lock::Locker m_locker;

  bool m_blocklisted = false;

  AsyncOpTracker m_timer_op_tracker;
  Context *m_timer_task = nullptr;
  C_TimerGate *m_timer_gate = nullptr;

  librbd::watcher::NotifyResponse m_heartbeat_response;

  bool is_leader(ceph::mutex &m_lock) const;
  bool is_releasing_leader(ceph::mutex &m_lock) const;

  void cancel_timer_task();
  void schedule_timer_task(const std::string &name,
                           int delay_factor, bool leader,
                           TimerCallback callback, bool shutting_down);
  void execute_timer_task(bool leader, TimerCallback timer_callback);

  void create_leader_object();
  void handle_create_leader_object(int r);

  void register_watch();
  void handle_register_watch(int r);

  void shut_down_leader_lock();
  void handle_shut_down_leader_lock(int r);

  void unregister_watch();
  void handle_unregister_watch(int r);

  void wait_for_tasks();
  void handle_wait_for_tasks();

  void break_leader_lock();
  void handle_break_leader_lock(int r);

  void schedule_get_locker(bool reset_leader, uint32_t delay_factor);
  void get_locker();
  void handle_get_locker(int r, librbd::managed_lock::Locker& locker);

  void schedule_acquire_leader_lock(uint32_t delay_factor);
  void acquire_leader_lock();
  void handle_acquire_leader_lock(int r);

  void release_leader_lock();
  void handle_release_leader_lock(int r);

  void init_instances();
  void handle_init_instances(int r);

  void shut_down_instances();
  void handle_shut_down_instances(int r);

  void notify_listener();
  void handle_notify_listener(int r);

  void notify_lock_acquired();
  void handle_notify_lock_acquired(int r);

  void notify_lock_released();
  void handle_notify_lock_released(int r);

  void notify_heartbeat();
  void handle_notify_heartbeat(int r);

  void handle_post_acquire_leader_lock(int r, Context *on_finish);
  void handle_pre_release_leader_lock(Context *on_finish);
  void handle_post_release_leader_lock(int r, Context *on_finish);

  void handle_notify(uint64_t notify_id, uint64_t handle,
                     uint64_t notifier_id, bufferlist &bl) override;

  void handle_rewatch_complete(int r) override;

  void handle_heartbeat(Context *on_ack);
  void handle_lock_acquired(Context *on_ack);
  void handle_lock_released(Context *on_ack);

  void handle_payload(const leader_watcher::HeartbeatPayload &payload,
                      Context *on_notify_ack);
  void handle_payload(const leader_watcher::LockAcquiredPayload &payload,
                      Context *on_notify_ack);
  void handle_payload(const leader_watcher::LockReleasedPayload &payload,
                      Context *on_notify_ack);
  void handle_payload(const leader_watcher::UnknownPayload &payload,
                      Context *on_notify_ack);
};

} // namespace mirror
} // namespace rbd

#endif // CEPH_RBD_MIRROR_LEADER_WATCHER_H
Commit	Line	Data
7c673cae FG	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_LEADER_WATCHER_H
	5	#define CEPH_RBD_MIRROR_LEADER_WATCHER_H
	6
	7	#include <list>
	8	#include <memory>
	9	#include <string>
	10
	11	#include "common/AsyncOpTracker.h"
	12	#include "librbd/ManagedLock.h"
	13	#include "librbd/Watcher.h"
	14	#include "librbd/managed_lock/Types.h"
	15	#include "librbd/watcher/Types.h"
	16	#include "Instances.h"
11fdf7f2	17	#include "tools/rbd_mirror/instances/Types.h"
7c673cae FG	18	#include "tools/rbd_mirror/leader_watcher/Types.h"
7c673cae FG	19
f67539c2 TL	20	namespace librbd {
	21	class ImageCtx;
	22	namespace asio { struct ContextWQ; }
	23	} // namespace librbd
7c673cae FG	24
	25	namespace rbd {
	26	namespace mirror {
	27
	28	template <typename> struct Threads;
	29
	30	template <typename ImageCtxT = librbd::ImageCtx>
	31	class LeaderWatcher : protected librbd::Watcher {
11fdf7f2	32	using librbd::Watcher::unregister_watch; // Silence overloaded virtual warning
7c673cae	33	public:
11fdf7f2 TL	34	static LeaderWatcher* create(Threads<ImageCtxT> *threads,
	35	librados::IoCtx &io_ctx,
	36	leader_watcher::Listener *listener) {
	37	return new LeaderWatcher(threads, io_ctx, listener);
	38	}
7c673cae FG	39
7c673cae FG	40	LeaderWatcher(Threads<ImageCtxT> *threads, librados::IoCtx &io_ctx,
11fdf7f2	41	leader_watcher::Listener *listener);
7c673cae FG	42	~LeaderWatcher() override;
	43
	44	int init();
	45	void shut_down();
	46
	47	void init(Context *on_finish);
	48	void shut_down(Context *on_finish);
	49
f67539c2	50	bool is_blocklisted() const;
7c673cae FG	51	bool is_leader() const;
	52	bool is_releasing_leader() const;
	53	bool get_leader_instance_id(std::string *instance_id) const;
	54	void release_leader();
	55	void list_instances(std::vector<std::string> *instance_ids);
	56
31f18b77 FG	57	std::string get_instance_id();
31f18b77 FG	58
7c673cae FG	59	private:
	60	/**
	61	* @verbatim
	62	*
	63	* <uninitialized> <------------------------------ WAIT_FOR_TASKS
11fdf7f2 TL	64	* \| (init) ^ ^
	65	* v * \|
	66	* CREATE_OBJECT * * * * * (error) UNREGISTER_WATCH
	67	* \| * ^
	68	* v * \|
9f95a23c TL	69	* REGISTER_WATCH * * * * * SHUT_DOWN_LEADER_LOCK
9f95a23c TL	70	* \| ^
7c673cae FG	71	* \| (no leader heartbeat and acquire failed) \|
	72	* \| BREAK_LOCK <-------------------------------------\ \|
	73	* \| \| (no leader heartbeat) \| \| (shut down)
	74	* \| \| /----------------------------------------\ \| \|
	75	* \| \| \| (lock_released received) \| \|
	76	* \| \| \| /-------------------------------------\ \| \|
	77	* \| \| \| \| (lock_acquired or \| \| \|
	78	* \| \| \| \| heartbeat received) \| \| \|
	79	* \| \| \| \| (ENOENT) /-----------\ \| \| \|
	80	* \| \| \| \| * * * * * * * * * * \| \| \| \| \|
	81	* v v v v v (error) * v \| \| \| \|
	82	* ACQUIRE_LEADER_LOCK * * * * *> GET_LOCKER ---> <secondary>
	83	* \| * ^
	84	* ....\|...................*.................... .....\|.....................
	85	* . v * . . \| post_release .
11fdf7f2 TL	86	* .INIT_INSTANCES * * * * * . .NOTIFY_LOCK_RELEASED .
	87	* . \| . .....^.....................
	88	* . v . \|
	89	* .NOTIFY_LISTENER . RELEASE_LEADER_LOCK
7c673cae FG	90	* . \| . ^
7c673cae FG	91	* . v . .....\|.....................
11fdf7f2 TL	92	* .NOTIFY_LOCK_ACQUIRED . . \| .
11fdf7f2 TL	93	* . \| post_acquire . .SHUT_DOWN_INSTANCES .
7c673cae FG	94	* ....\|........................................ . ^ .
	95	* v . \| .
	96	* <leader> -----------------------------------> .NOTIFY_LISTENER .
	97	* (shut_down, release_leader, . pre_release .
	98	* notify error) ...........................
	99	* @endverbatim
	100	*/
	101
11fdf7f2 TL	102	struct InstancesListener : public instances::Listener {
	103	LeaderWatcher* leader_watcher;
	104
	105	InstancesListener(LeaderWatcher* leader_watcher)
	106	: leader_watcher(leader_watcher) {
	107	}
	108
	109	void handle_added(const InstanceIds& instance_ids) override {
	110	leader_watcher->m_listener->handle_instances_added(instance_ids);
	111	}
	112
	113	void handle_removed(const InstanceIds& instance_ids) override {
	114	leader_watcher->m_listener->handle_instances_removed(instance_ids);
	115	}
	116	};
	117
7c673cae FG	118	class LeaderLock : public librbd::ManagedLock<ImageCtxT> {
	119	public:
	120	typedef librbd::ManagedLock<ImageCtxT> Parent;
	121
f67539c2	122	LeaderLock(librados::IoCtx& ioctx, librbd::AsioEngine& asio_engine,
7c673cae	123	const std::string& oid, LeaderWatcher *watcher,
f67539c2 TL	124	bool blocklist_on_break_lock,
	125	uint32_t blocklist_expire_seconds)
	126	: Parent(ioctx, asio_engine, oid, watcher,
	127	librbd::managed_lock::EXCLUSIVE, blocklist_on_break_lock,
	128	blocklist_expire_seconds),
7c673cae FG	129	watcher(watcher) {
	130	}
	131
	132	bool is_leader() const {
9f95a23c	133	std::lock_guard locker{Parent::m_lock};
7c673cae FG	134	return Parent::is_state_post_acquiring() \|\| Parent::is_state_locked();
	135	}
	136
	137	bool is_releasing_leader() const {
9f95a23c	138	std::lock_guard locker{Parent::m_lock};
7c673cae FG	139	return Parent::is_state_pre_releasing();
	140	}
	141
	142	protected:
	143	void post_acquire_lock_handler(int r, Context *on_finish) {
	144	if (r == 0) {
	145	// lock is owned at this point
9f95a23c	146	std::lock_guard locker{Parent::m_lock};
7c673cae FG	147	Parent::set_state_post_acquiring();
	148	}
	149	watcher->handle_post_acquire_leader_lock(r, on_finish);
	150	}
	151	void pre_release_lock_handler(bool shutting_down,
	152	Context *on_finish) {
	153	watcher->handle_pre_release_leader_lock(on_finish);
	154	}
	155	void post_release_lock_handler(bool shutting_down, int r,
	156	Context *on_finish) {
	157	watcher->handle_post_release_leader_lock(r, on_finish);
	158	}
	159	private:
	160	LeaderWatcher *watcher;
	161	};
	162
	163	struct HandlePayloadVisitor : public boost::static_visitor<void> {
	164	LeaderWatcher *leader_watcher;
	165	Context *on_notify_ack;
	166
	167	HandlePayloadVisitor(LeaderWatcher leader_watcher, Context on_notify_ack)
	168	: leader_watcher(leader_watcher), on_notify_ack(on_notify_ack) {
	169	}
	170
	171	template <typename Payload>
	172	inline void operator()(const Payload &payload) const {
	173	leader_watcher->handle_payload(payload, on_notify_ack);
	174	}
	175	};
	176
	177	struct C_GetLocker : public Context {
	178	LeaderWatcher *leader_watcher;
	179	librbd::managed_lock::Locker locker;
	180
	181	C_GetLocker(LeaderWatcher *leader_watcher)
	182	: leader_watcher(leader_watcher) {
	183	}
	184
	185	void finish(int r) override {
	186	leader_watcher->handle_get_locker(r, locker);
	187	}
	188	};
	189
	190	typedef void (LeaderWatcher<ImageCtxT>::*TimerCallback)();
	191
	192	struct C_TimerGate : public Context {
	193	LeaderWatcher *leader_watcher;
	194
	195	bool leader = false;
	196	TimerCallback timer_callback = nullptr;
	197
	198	C_TimerGate(LeaderWatcher *leader_watcher)
	199	: leader_watcher(leader_watcher) {
	200	}
	201
	202	void finish(int r) override {
	203	leader_watcher->m_timer_gate = nullptr;
	204	leader_watcher->execute_timer_task(leader, timer_callback);
	205	}
	206	};
	207
	208	Threads<ImageCtxT> *m_threads;
11fdf7f2	209	leader_watcher::Listener *m_listener;
7c673cae	210
11fdf7f2	211	InstancesListener m_instances_listener;
9f95a23c	212	mutable ceph::mutex m_lock;
7c673cae	213	uint64_t m_notifier_id;
11fdf7f2	214	std::string m_instance_id;
7c673cae FG	215	LeaderLock *m_leader_lock;
	216	Context *m_on_finish = nullptr;
	217	Context *m_on_shut_down_finish = nullptr;
11fdf7f2	218	uint64_t m_acquire_attempts = 0;
7c673cae	219	int m_ret_val = 0;
7c673cae FG	220	Instances<ImageCtxT> *m_instances = nullptr;
	221	librbd::managed_lock::Locker m_locker;
	222
f67539c2	223	bool m_blocklisted = false;
9f95a23c	224
7c673cae FG	225	AsyncOpTracker m_timer_op_tracker;
	226	Context *m_timer_task = nullptr;
	227	C_TimerGate *m_timer_gate = nullptr;
	228
	229	librbd::watcher::NotifyResponse m_heartbeat_response;
	230
9f95a23c TL	231	bool is_leader(ceph::mutex &m_lock) const;
9f95a23c TL	232	bool is_releasing_leader(ceph::mutex &m_lock) const;
7c673cae FG	233
	234	void cancel_timer_task();
	235	void schedule_timer_task(const std::string &name,
	236	int delay_factor, bool leader,
	237	TimerCallback callback, bool shutting_down);
	238	void execute_timer_task(bool leader, TimerCallback timer_callback);
	239
	240	void create_leader_object();
	241	void handle_create_leader_object(int r);
	242
	243	void register_watch();
	244	void handle_register_watch(int r);
	245
	246	void shut_down_leader_lock();
	247	void handle_shut_down_leader_lock(int r);
	248
	249	void unregister_watch();
	250	void handle_unregister_watch(int r);
	251
	252	void wait_for_tasks();
	253	void handle_wait_for_tasks();
	254
	255	void break_leader_lock();
	256	void handle_break_leader_lock(int r);
	257
	258	void schedule_get_locker(bool reset_leader, uint32_t delay_factor);
	259	void get_locker();
	260	void handle_get_locker(int r, librbd::managed_lock::Locker& locker);
	261
	262	void schedule_acquire_leader_lock(uint32_t delay_factor);
	263	void acquire_leader_lock();
	264	void handle_acquire_leader_lock(int r);
	265
	266	void release_leader_lock();
	267	void handle_release_leader_lock(int r);
	268
7c673cae FG	269	void init_instances();
	270	void handle_init_instances(int r);
	271
	272	void shut_down_instances();
	273	void handle_shut_down_instances(int r);
	274
	275	void notify_listener();
	276	void handle_notify_listener(int r);
	277
	278	void notify_lock_acquired();
	279	void handle_notify_lock_acquired(int r);
	280
	281	void notify_lock_released();
	282	void handle_notify_lock_released(int r);
	283
	284	void notify_heartbeat();
	285	void handle_notify_heartbeat(int r);
	286
7c673cae FG	287	void handle_post_acquire_leader_lock(int r, Context *on_finish);
	288	void handle_pre_release_leader_lock(Context *on_finish);
	289	void handle_post_release_leader_lock(int r, Context *on_finish);
	290
	291	void handle_notify(uint64_t notify_id, uint64_t handle,
	292	uint64_t notifier_id, bufferlist &bl) override;
	293
91327a77 AA	294	void handle_rewatch_complete(int r) override;
91327a77 AA	295
7c673cae FG	296	void handle_heartbeat(Context *on_ack);
	297	void handle_lock_acquired(Context *on_ack);
	298	void handle_lock_released(Context *on_ack);
	299
	300	void handle_payload(const leader_watcher::HeartbeatPayload &payload,
	301	Context *on_notify_ack);
	302	void handle_payload(const leader_watcher::LockAcquiredPayload &payload,
	303	Context *on_notify_ack);
	304	void handle_payload(const leader_watcher::LockReleasedPayload &payload,
	305	Context *on_notify_ack);
	306	void handle_payload(const leader_watcher::UnknownPayload &payload,
	307	Context *on_notify_ack);
	308	};
	309
	310	} // namespace mirror
	311	} // namespace rbd
	312
	313	#endif // CEPH_RBD_MIRROR_LEADER_WATCHER_H