[ceph.git] / ceph / src / tools / rbd_mirror / image_replayer / snapshot / Replayer.h

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

#ifndef RBD_MIRROR_IMAGE_REPLAYER_SNAPSHOT_REPLAYER_H
#define RBD_MIRROR_IMAGE_REPLAYER_SNAPSHOT_REPLAYER_H

#include "tools/rbd_mirror/image_replayer/Replayer.h"
#include "common/ceph_mutex.h"
#include "common/AsyncOpTracker.h"
#include "cls/rbd/cls_rbd_types.h"
#include "librbd/mirror/snapshot/Types.h"
#include "tools/rbd_mirror/image_replayer/TimeRollingMean.h"
#include <boost/accumulators/accumulators.hpp>
#include <boost/accumulators/statistics/stats.hpp>
#include <boost/accumulators/statistics/rolling_mean.hpp>
#include <string>
#include <type_traits>

namespace librbd {

struct ImageCtx;
namespace snapshot { template <typename I> class Replay; }

} // namespace librbd

namespace rbd {
namespace mirror {

template <typename> struct InstanceWatcher;
class PoolMetaCache;
template <typename> struct Threads;

namespace image_replayer {

struct ReplayerListener;

namespace snapshot {

template <typename> class EventPreprocessor;
template <typename> class ReplayStatusFormatter;
template <typename> class StateBuilder;

template <typename ImageCtxT>
class Replayer : public image_replayer::Replayer {
public:
  static Replayer* create(
      Threads<ImageCtxT>* threads,
      InstanceWatcher<ImageCtxT>* instance_watcher,
      const std::string& local_mirror_uuid,
      PoolMetaCache* pool_meta_cache,
      StateBuilder<ImageCtxT>* state_builder,
      ReplayerListener* replayer_listener) {
    return new Replayer(threads, instance_watcher, local_mirror_uuid,
                        pool_meta_cache, state_builder, replayer_listener);
  }

  Replayer(
      Threads<ImageCtxT>* threads,
      InstanceWatcher<ImageCtxT>* instance_watcher,
      const std::string& local_mirror_uuid,
      PoolMetaCache* pool_meta_cache,
      StateBuilder<ImageCtxT>* state_builder,
      ReplayerListener* replayer_listener);
  ~Replayer();

  void destroy() override {
    delete this;
  }

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

  void flush(Context* on_finish) override;

  bool get_replay_status(std::string* description, Context* on_finish) override;

  bool is_replaying() const override {
    std::unique_lock locker{m_lock};
    return (m_state == STATE_REPLAYING || m_state == STATE_IDLE);
  }

  bool is_resync_requested() const override {
    std::unique_lock locker{m_lock};
    return m_resync_requested;
  }

  int get_error_code() const override {
    std::unique_lock locker(m_lock);
    return m_error_code;
  }

  std::string get_error_description() const override {
    std::unique_lock locker(m_lock);
    return m_error_description;
  }

private:
  /**
   * @verbatim
   *
   * <init>
   *    |
   *    v
   * REGISTER_LOCAL_UPDATE_WATCHER
   *    |
   *    v
   * REGISTER_REMOTE_UPDATE_WATCHER
   *    |
   *    v
   * LOAD_LOCAL_IMAGE_META <----------------------------\
   *    |                                               |
   *    v (skip if not needed)                          |
   * REFRESH_LOCAL_IMAGE                                |
   *    |                                               |
   *    v (skip if not needed)                          |
   * REFRESH_REMOTE_IMAGE                               |
   *    |                                               |
   *    | (unused non-primary snapshot)                 |
   *    |\--------------> PRUNE_NON_PRIMARY_SNAPSHOT---/|
   *    |                                               |
   *    | (interrupted sync)                            |
   *    |\--------------> GET_LOCAL_IMAGE_STATE ------\ |
   *    |                                             | |
   *    | (new snapshot)                              | |
   *    |\--------------> COPY_SNAPSHOTS              | |
   *    |                       |                     | |
   *    |                       v                     | |
   *    |                 GET_REMOTE_IMAGE_STATE      | |
   *    |                       |                     | |
   *    |                       v                     | |
   *    |                 CREATE_NON_PRIMARY_SNAPSHOT | |
   *    |                       |                     | |
   *    |                       v (skip if not needed)| |
   *    |                 UPDATE_MIRROR_IMAGE_STATE   | |
   *    |                       |                     | |
   *    |                       |/--------------------/ |
   *    |                       |                       |
   *    |                       v                       |
   *    |                 REQUEST_SYNC                  |
   *    |                       |                       |
   *    |                       v                       |
   *    |                 COPY_IMAGE                    |
   *    |                       |                       |
   *    |                       v                       |
   *    |                 APPLY_IMAGE_STATE             |
   *    |                       |                       |
   *    |                       v                       |
   *    |                 UPDATE_NON_PRIMARY_SNAPSHOT   |
   *    |                       |                       |
   *    |                       v                       |
   *    |                 NOTIFY_IMAGE_UPDATE           |
   *    |                       |                       |
   *    | (interrupted unlink)  v                       |
   *    |\--------------> UNLINK_PEER                   |
   *    |                       |                       |
   *    |                       v                       |
   *    |                 NOTIFY_LISTENER               |
   *    |                       |                       |
   *    |                       \----------------------/|
   *    |                                               |
   *    | (remote demoted)                              |
   *    \---------------> NOTIFY_LISTENER               |
   *    |                     |                         |
   *    |/--------------------/                         |
   *    |                                               |
   *    |   (update notification)                       |
   * <idle> --------------------------------------------/
   *    |
   *    v
   * <shut down>
   *    |
   *    v
   * UNREGISTER_REMOTE_UPDATE_WATCHER
   *    |
   *    v
   * UNREGISTER_LOCAL_UPDATE_WATCHER
   *    |
   *    v
   * WAIT_FOR_IN_FLIGHT_OPS
   *    |
   *    v
   * <finish>
   *
   * @endverbatim
   */

  enum State {
    STATE_INIT,
    STATE_REPLAYING,
    STATE_IDLE,
    STATE_COMPLETE
  };

  struct C_UpdateWatchCtx;
  struct DeepCopyHandler;

  Threads<ImageCtxT>* m_threads;
  InstanceWatcher<ImageCtxT>* m_instance_watcher;
  std::string m_local_mirror_uuid;
  PoolMetaCache* m_pool_meta_cache;
  StateBuilder<ImageCtxT>* m_state_builder;
  ReplayerListener* m_replayer_listener;

  mutable ceph::mutex m_lock;

  State m_state = STATE_INIT;

  Context* m_on_init_shutdown = nullptr;

  bool m_resync_requested = false;
  int m_error_code = 0;
  std::string m_error_description;

  C_UpdateWatchCtx* m_update_watch_ctx = nullptr;
  uint64_t m_local_update_watcher_handle = 0;
  uint64_t m_remote_update_watcher_handle = 0;
  bool m_image_updated = false;

  AsyncOpTracker m_in_flight_op_tracker;

  uint64_t m_local_snap_id_start = 0;
  uint64_t m_local_snap_id_end = CEPH_NOSNAP;
  cls::rbd::MirrorSnapshotNamespace m_local_mirror_snap_ns;
  uint64_t m_local_object_count = 0;

  std::string m_remote_mirror_peer_uuid;
  uint64_t m_remote_snap_id_start = 0;
  uint64_t m_remote_snap_id_end = CEPH_NOSNAP;
  cls::rbd::MirrorSnapshotNamespace m_remote_mirror_snap_ns;

  librbd::mirror::snapshot::ImageState m_image_state;
  DeepCopyHandler* m_deep_copy_handler = nullptr;

  TimeRollingMean m_bytes_per_second;

  uint64_t m_snapshot_bytes = 0;
  boost::accumulators::accumulator_set<
    uint64_t, boost::accumulators::stats<
      boost::accumulators::tag::rolling_mean>> m_bytes_per_snapshot{
    boost::accumulators::tag::rolling_window::window_size = 2};

  uint32_t m_pending_snapshots = 0;

  bool m_remote_image_updated = false;
  bool m_updating_sync_point = false;
  bool m_sync_in_progress = false;

  void load_local_image_meta();
  void handle_load_local_image_meta(int r);

  void refresh_local_image();
  void handle_refresh_local_image(int r);

  void refresh_remote_image();
  void handle_refresh_remote_image(int r);

  void scan_local_mirror_snapshots(std::unique_lock<ceph::mutex>* locker);
  void scan_remote_mirror_snapshots(std::unique_lock<ceph::mutex>* locker);

  void prune_non_primary_snapshot(uint64_t snap_id);
  void handle_prune_non_primary_snapshot(int r);

  void copy_snapshots();
  void handle_copy_snapshots(int r);

  void get_remote_image_state();
  void handle_get_remote_image_state(int r);

  void get_local_image_state();
  void handle_get_local_image_state(int r);

  void create_non_primary_snapshot();
  void handle_create_non_primary_snapshot(int r);

  void update_mirror_image_state();
  void handle_update_mirror_image_state(int r);

  void request_sync();
  void handle_request_sync(int r);

  void copy_image();
  void handle_copy_image(int r);
  void handle_copy_image_progress(uint64_t object_number,
                                  uint64_t object_count);
  void handle_copy_image_read(uint64_t bytes_read);

  void apply_image_state();
  void handle_apply_image_state(int r);

  void update_non_primary_snapshot(bool complete);
  void handle_update_non_primary_snapshot(bool complete, int r);

  void notify_image_update();
  void handle_notify_image_update(int r);

  void unlink_peer(uint64_t remote_snap_id);
  void handle_unlink_peer(int r);

  void finish_sync();

  void register_local_update_watcher();
  void handle_register_local_update_watcher(int r);

  void register_remote_update_watcher();
  void handle_register_remote_update_watcher(int r);

  void unregister_remote_update_watcher();
  void handle_unregister_remote_update_watcher(int r);

  void unregister_local_update_watcher();
  void handle_unregister_local_update_watcher(int r);

  void wait_for_in_flight_ops();
  void handle_wait_for_in_flight_ops(int r);

  void handle_image_update_notify();

  void handle_replay_complete(int r, const std::string& description);
  void handle_replay_complete(std::unique_lock<ceph::mutex>* locker,
                              int r, const std::string& description);
  void notify_status_updated();

  bool is_replay_interrupted();
  bool is_replay_interrupted(std::unique_lock<ceph::mutex>* lock);

};

} // namespace snapshot
} // namespace image_replayer
} // namespace mirror
} // namespace rbd

extern template class rbd::mirror::image_replayer::snapshot::Replayer<librbd::ImageCtx>;

#endif // RBD_MIRROR_IMAGE_REPLAYER_SNAPSHOT_REPLAYER_H
Commit	Line	Data
9f95a23c TL	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t --
	2	// vim: ts=8 sw=2 smarttab
	3
	4	#ifndef RBD_MIRROR_IMAGE_REPLAYER_SNAPSHOT_REPLAYER_H
	5	#define RBD_MIRROR_IMAGE_REPLAYER_SNAPSHOT_REPLAYER_H
	6
	7	#include "tools/rbd_mirror/image_replayer/Replayer.h"
	8	#include "common/ceph_mutex.h"
	9	#include "common/AsyncOpTracker.h"
	10	#include "cls/rbd/cls_rbd_types.h"
	11	#include "librbd/mirror/snapshot/Types.h"
1911f103 TL	12	#include "tools/rbd_mirror/image_replayer/TimeRollingMean.h"
	13	#include <boost/accumulators/accumulators.hpp>
	14	#include <boost/accumulators/statistics/stats.hpp>
	15	#include <boost/accumulators/statistics/rolling_mean.hpp>
9f95a23c TL	16	#include <string>
	17	#include <type_traits>
	18
	19	namespace librbd {
	20
	21	struct ImageCtx;
	22	namespace snapshot { template <typename I> class Replay; }
	23
	24	} // namespace librbd
	25
	26	namespace rbd {
	27	namespace mirror {
	28
	29	template <typename> struct InstanceWatcher;
	30	class PoolMetaCache;
	31	template <typename> struct Threads;
	32
	33	namespace image_replayer {
	34
	35	struct ReplayerListener;
	36
	37	namespace snapshot {
	38
	39	template <typename> class EventPreprocessor;
	40	template <typename> class ReplayStatusFormatter;
	41	template <typename> class StateBuilder;
	42
	43	template <typename ImageCtxT>
	44	class Replayer : public image_replayer::Replayer {
	45	public:
	46	static Replayer* create(
	47	Threads<ImageCtxT>* threads,
	48	InstanceWatcher<ImageCtxT>* instance_watcher,
	49	const std::string& local_mirror_uuid,
	50	PoolMetaCache* pool_meta_cache,
	51	StateBuilder<ImageCtxT>* state_builder,
	52	ReplayerListener* replayer_listener) {
	53	return new Replayer(threads, instance_watcher, local_mirror_uuid,
	54	pool_meta_cache, state_builder, replayer_listener);
	55	}
	56
	57	Replayer(
	58	Threads<ImageCtxT>* threads,
	59	InstanceWatcher<ImageCtxT>* instance_watcher,
	60	const std::string& local_mirror_uuid,
	61	PoolMetaCache* pool_meta_cache,
	62	StateBuilder<ImageCtxT>* state_builder,
	63	ReplayerListener* replayer_listener);
	64	~Replayer();
	65
	66	void destroy() override {
	67	delete this;
	68	}
	69
	70	void init(Context* on_finish) override;
	71	void shut_down(Context* on_finish) override;
	72
	73	void flush(Context* on_finish) override;
	74
	75	bool get_replay_status(std::string* description, Context* on_finish) override;
	76
	77	bool is_replaying() const override {
	78	std::unique_lock locker{m_lock};
	79	return (m_state == STATE_REPLAYING \|\| m_state == STATE_IDLE);
80	}
81
82	bool is_resync_requested() const override {
83	std::unique_lock locker{m_lock};
84	return m_resync_requested;
85	}
86
87	int get_error_code() const override {
88	std::unique_lock locker(m_lock);
89	return m_error_code;
90	}
91
92	std::string get_error_description() const override {
93	std::unique_lock locker(m_lock);
94	return m_error_description;
95	}
96
97	private:
98	/**
99	* @verbatim
100	*
101	* <init>
102	* \|
103	* v
104	* REGISTER_LOCAL_UPDATE_WATCHER
105	* \|
106	* v
107	* REGISTER_REMOTE_UPDATE_WATCHER
108	* \|
109	* v
110	* LOAD_LOCAL_IMAGE_META <----------------------------\
111	* \| \|
112	* v (skip if not needed) \|
113	* REFRESH_LOCAL_IMAGE \|
114	* \| \|
115	* v (skip if not needed) \|
116	* REFRESH_REMOTE_IMAGE \|
117	* \| \|
1911f103 TL	118	* \| (unused non-primary snapshot) \|
	119	* \|\--------------> PRUNE_NON_PRIMARY_SNAPSHOT---/\|
	120	* \| \|
9f95a23c TL	121	* \| (interrupted sync) \|
	122	* \|\--------------> GET_LOCAL_IMAGE_STATE ------\ \|
	123	* \| \| \|
	124	* \| (new snapshot) \| \|
	125	* \|\--------------> COPY_SNAPSHOTS \| \|
1911f103 TL	126	* \| \| \| \|
1911f103 TL	127	* \| v \| \|
9f95a23c	128	* \| GET_REMOTE_IMAGE_STATE \| \|
1911f103 TL	129	* \| \| \| \|
1911f103 TL	130	* \| v \| \|
9f95a23c	131	* \| CREATE_NON_PRIMARY_SNAPSHOT \| \|
1911f103	132	* \| \| \| \|
7f7e6c64 TL	133	* \| v (skip if not needed)\| \|
	134	* \| UPDATE_MIRROR_IMAGE_STATE \| \|
	135	* \| \| \| \|
1911f103 TL	136	* \| \|/--------------------/ \|
	137	* \| \| \|
	138	* \| v \|
9f95a23c	139	* \| REQUEST_SYNC \|
1911f103 TL	140	* \| \| \|
1911f103 TL	141	* \| v \|
9f95a23c	142	* \| COPY_IMAGE \|
1911f103 TL	143	* \| \| \|
1911f103 TL	144	* \| v \|
9f95a23c	145	* \| APPLY_IMAGE_STATE \|
1911f103 TL	146	* \| \| \|
1911f103 TL	147	* \| v \|
9f95a23c	148	* \| UPDATE_NON_PRIMARY_SNAPSHOT \|
1911f103 TL	149	* \| \| \|
1911f103 TL	150	* \| v \|
9f95a23c	151	* \| NOTIFY_IMAGE_UPDATE \|
1911f103 TL	152	* \| \| \|
	153	* \| (interrupted unlink) v \|
	154	* \|\--------------> UNLINK_PEER \|
	155	* \| \| \|
	156	* \| v \|
9f95a23c	157	* \| NOTIFY_LISTENER \|
1911f103 TL	158	* \| \| \|
1911f103 TL	159	* \| \----------------------/\|
9f95a23c TL	160	* \| \|
	161	* \| (remote demoted) \|
	162	* \---------------> NOTIFY_LISTENER \|
	163	* \| \| \|
	164	* \|/--------------------/ \|
	165	* \| \|
	166	* \| (update notification) \|
	167	* <idle> --------------------------------------------/
	168	* \|
	169	* v
	170	* <shut down>
	171	* \|
	172	* v
	173	* UNREGISTER_REMOTE_UPDATE_WATCHER
	174	* \|
	175	* v
	176	* UNREGISTER_LOCAL_UPDATE_WATCHER
	177	* \|
	178	* v
	179	* WAIT_FOR_IN_FLIGHT_OPS
	180	* \|
	181	* v
	182	* <finish>
	183	*
	184	* @endverbatim
	185	*/
	186
	187	enum State {
	188	STATE_INIT,
	189	STATE_REPLAYING,
	190	STATE_IDLE,
	191	STATE_COMPLETE
	192	};
	193
	194	struct C_UpdateWatchCtx;
1911f103	195	struct DeepCopyHandler;
9f95a23c TL	196
	197	Threads<ImageCtxT>* m_threads;
	198	InstanceWatcher<ImageCtxT>* m_instance_watcher;
	199	std::string m_local_mirror_uuid;
	200	PoolMetaCache* m_pool_meta_cache;
	201	StateBuilder<ImageCtxT>* m_state_builder;
	202	ReplayerListener* m_replayer_listener;
	203
	204	mutable ceph::mutex m_lock;
	205
	206	State m_state = STATE_INIT;
	207
	208	Context* m_on_init_shutdown = nullptr;
	209
	210	bool m_resync_requested = false;
	211	int m_error_code = 0;
	212	std::string m_error_description;
	213
ec96510d	214	C_UpdateWatchCtx* m_update_watch_ctx = nullptr;
9f95a23c TL	215	uint64_t m_local_update_watcher_handle = 0;
	216	uint64_t m_remote_update_watcher_handle = 0;
	217	bool m_image_updated = false;
	218
	219	AsyncOpTracker m_in_flight_op_tracker;
	220
	221	uint64_t m_local_snap_id_start = 0;
	222	uint64_t m_local_snap_id_end = CEPH_NOSNAP;
	223	cls::rbd::MirrorSnapshotNamespace m_local_mirror_snap_ns;
	224	uint64_t m_local_object_count = 0;
	225
	226	std::string m_remote_mirror_peer_uuid;
	227	uint64_t m_remote_snap_id_start = 0;
	228	uint64_t m_remote_snap_id_end = CEPH_NOSNAP;
	229	cls::rbd::MirrorSnapshotNamespace m_remote_mirror_snap_ns;
	230
	231	librbd::mirror::snapshot::ImageState m_image_state;
1911f103 TL	232	DeepCopyHandler* m_deep_copy_handler = nullptr;
	233
	234	TimeRollingMean m_bytes_per_second;
	235
	236	uint64_t m_snapshot_bytes = 0;
	237	boost::accumulators::accumulator_set<
	238	uint64_t, boost::accumulators::stats<
	239	boost::accumulators::tag::rolling_mean>> m_bytes_per_snapshot{
	240	boost::accumulators::tag::rolling_window::window_size = 2};
	241
	242	uint32_t m_pending_snapshots = 0;
9f95a23c TL	243
	244	bool m_remote_image_updated = false;
	245	bool m_updating_sync_point = false;
	246	bool m_sync_in_progress = false;
	247
	248	void load_local_image_meta();
	249	void handle_load_local_image_meta(int r);
	250
	251	void refresh_local_image();
	252	void handle_refresh_local_image(int r);
	253
	254	void refresh_remote_image();
	255	void handle_refresh_remote_image(int r);
	256
	257	void scan_local_mirror_snapshots(std::unique_lock<ceph::mutex>* locker);
	258	void scan_remote_mirror_snapshots(std::unique_lock<ceph::mutex>* locker);
	259
1911f103 TL	260	void prune_non_primary_snapshot(uint64_t snap_id);
	261	void handle_prune_non_primary_snapshot(int r);
	262
9f95a23c TL	263	void copy_snapshots();
	264	void handle_copy_snapshots(int r);
	265
	266	void get_remote_image_state();
	267	void handle_get_remote_image_state(int r);
	268
	269	void get_local_image_state();
	270	void handle_get_local_image_state(int r);
	271
	272	void create_non_primary_snapshot();
	273	void handle_create_non_primary_snapshot(int r);
	274
7f7e6c64 TL	275	void update_mirror_image_state();
	276	void handle_update_mirror_image_state(int r);
	277
9f95a23c TL	278	void request_sync();
	279	void handle_request_sync(int r);
	280
	281	void copy_image();
	282	void handle_copy_image(int r);
	283	void handle_copy_image_progress(uint64_t object_number,
	284	uint64_t object_count);
1911f103	285	void handle_copy_image_read(uint64_t bytes_read);
9f95a23c TL	286
	287	void apply_image_state();
	288	void handle_apply_image_state(int r);
	289
	290	void update_non_primary_snapshot(bool complete);
	291	void handle_update_non_primary_snapshot(bool complete, int r);
	292
	293	void notify_image_update();
	294	void handle_notify_image_update(int r);
	295
1911f103	296	void unlink_peer(uint64_t remote_snap_id);
9f95a23c TL	297	void handle_unlink_peer(int r);
	298
	299	void finish_sync();
	300
	301	void register_local_update_watcher();
	302	void handle_register_local_update_watcher(int r);
	303
	304	void register_remote_update_watcher();
	305	void handle_register_remote_update_watcher(int r);
	306
	307	void unregister_remote_update_watcher();
	308	void handle_unregister_remote_update_watcher(int r);
	309
	310	void unregister_local_update_watcher();
	311	void handle_unregister_local_update_watcher(int r);
	312
	313	void wait_for_in_flight_ops();
	314	void handle_wait_for_in_flight_ops(int r);
	315
	316	void handle_image_update_notify();
	317
	318	void handle_replay_complete(int r, const std::string& description);
	319	void handle_replay_complete(std::unique_lock<ceph::mutex>* locker,
	320	int r, const std::string& description);
	321	void notify_status_updated();
	322
	323	bool is_replay_interrupted();
	324	bool is_replay_interrupted(std::unique_lock<ceph::mutex>* lock);
	325
	326	};
	327
	328	} // namespace snapshot
	329	} // namespace image_replayer
	330	} // namespace mirror
	331	} // namespace rbd
	332
	333	extern template class rbd::mirror::image_replayer::snapshot::Replayer<librbd::ImageCtx>;
	334
	335	#endif // RBD_MIRROR_IMAGE_REPLAYER_SNAPSHOT_REPLAYER_H