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