[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;
  }

  std::string get_image_spec() const {
    std::unique_lock locker(m_lock);
    return m_image_spec;
  }

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;

  std::string m_image_spec;
  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_last_snapshot_sync_seconds = 0;

  uint64_t m_snapshot_bytes = 0;
  uint64_t m_last_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};
  utime_t m_snapshot_replay_start;

  uint32_t m_pending_snapshots = 0;

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

  PerfCounters *m_perf_counters = nullptr;

  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);

  void register_perf_counters();
  void unregister_perf_counters();
};

} // 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
	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"
	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>
	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	std::string get_image_spec() const {
	98	std::unique_lock locker(m_lock);
	99	return m_image_spec;
	100	}
	101
	102	private:
	103	/**
	104	* @verbatim
	105	*
	106	* <init>
	107	* \|
	108	* v
	109	* REGISTER_LOCAL_UPDATE_WATCHER
	110	* \|
	111	* v
	112	* REGISTER_REMOTE_UPDATE_WATCHER
	113	* \|
	114	* v
	115	* LOAD_LOCAL_IMAGE_META <----------------------------\
	116	* \| \|
	117	* v (skip if not needed) \|
	118	* REFRESH_LOCAL_IMAGE \|
	119	* \| \|
	120	* v (skip if not needed) \|
	121	* REFRESH_REMOTE_IMAGE \|
	122	* \| \|
	123	* \| (unused non-primary snapshot) \|
	124	* \|\--------------> PRUNE_NON_PRIMARY_SNAPSHOT---/\|
	125	* \| \|
	126	* \| (interrupted sync) \|
	127	* \|\--------------> GET_LOCAL_IMAGE_STATE ------\ \|
	128	* \| \| \|
	129	* \| (new snapshot) \| \|
	130	* \|\--------------> COPY_SNAPSHOTS \| \|
	131	* \| \| \| \|
	132	* \| v \| \|
	133	* \| GET_REMOTE_IMAGE_STATE \| \|
	134	* \| \| \| \|
	135	* \| v \| \|
	136	* \| CREATE_NON_PRIMARY_SNAPSHOT \| \|
	137	* \| \| \| \|
	138	* \| v (skip if not needed)\| \|
	139	* \| UPDATE_MIRROR_IMAGE_STATE \| \|
	140	* \| \| \| \|
	141	* \| \|/--------------------/ \|
	142	* \| \| \|
	143	* \| v \|
	144	* \| REQUEST_SYNC \|
	145	* \| \| \|
	146	* \| v \|
	147	* \| COPY_IMAGE \|
	148	* \| \| \|
	149	* \| v \|
	150	* \| APPLY_IMAGE_STATE \|
	151	* \| \| \|
	152	* \| v \|
	153	* \| UPDATE_NON_PRIMARY_SNAPSHOT \|
	154	* \| \| \|
	155	* \| v \|
	156	* \| NOTIFY_IMAGE_UPDATE \|
	157	* \| \| \|
	158	* \| (interrupted unlink) v \|
	159	* \|\--------------> UNLINK_PEER \|
	160	* \| \| \|
	161	* \| v \|
	162	* \| NOTIFY_LISTENER \|
	163	* \| \| \|
	164	* \| \----------------------/\|
	165	* \| \|
	166	* \| (remote demoted) \|
	167	* \---------------> NOTIFY_LISTENER \|
	168	* \| \| \|
	169	* \|/--------------------/ \|
	170	* \| \|
	171	* \| (update notification) \|
	172	* <idle> --------------------------------------------/
	173	* \|
	174	* v
	175	* <shut down>
	176	* \|
	177	* v
	178	* UNREGISTER_REMOTE_UPDATE_WATCHER
	179	* \|
	180	* v
	181	* UNREGISTER_LOCAL_UPDATE_WATCHER
	182	* \|
	183	* v
	184	* WAIT_FOR_IN_FLIGHT_OPS
	185	* \|
	186	* v
	187	* <finish>
	188	*
	189	* @endverbatim
	190	*/
	191
	192	enum State {
	193	STATE_INIT,
	194	STATE_REPLAYING,
	195	STATE_IDLE,
	196	STATE_COMPLETE
	197	};
	198
	199	struct C_UpdateWatchCtx;
	200	struct DeepCopyHandler;
	201
	202	Threads<ImageCtxT>* m_threads;
	203	InstanceWatcher<ImageCtxT>* m_instance_watcher;
	204	std::string m_local_mirror_uuid;
	205	PoolMetaCache* m_pool_meta_cache;
	206	StateBuilder<ImageCtxT>* m_state_builder;
	207	ReplayerListener* m_replayer_listener;
	208
	209	mutable ceph::mutex m_lock;
	210
	211	State m_state = STATE_INIT;
	212
	213	std::string m_image_spec;
	214	Context* m_on_init_shutdown = nullptr;
	215
	216	bool m_resync_requested = false;
	217	int m_error_code = 0;
	218	std::string m_error_description;
	219
	220	C_UpdateWatchCtx* m_update_watch_ctx = nullptr;
	221	uint64_t m_local_update_watcher_handle = 0;
	222	uint64_t m_remote_update_watcher_handle = 0;
	223	bool m_image_updated = false;
	224
	225	AsyncOpTracker m_in_flight_op_tracker;
	226
	227	uint64_t m_local_snap_id_start = 0;
	228	uint64_t m_local_snap_id_end = CEPH_NOSNAP;
	229	cls::rbd::MirrorSnapshotNamespace m_local_mirror_snap_ns;
	230	uint64_t m_local_object_count = 0;
	231
	232	std::string m_remote_mirror_peer_uuid;
	233	uint64_t m_remote_snap_id_start = 0;
	234	uint64_t m_remote_snap_id_end = CEPH_NOSNAP;
	235	cls::rbd::MirrorSnapshotNamespace m_remote_mirror_snap_ns;
	236
	237	librbd::mirror::snapshot::ImageState m_image_state;
	238	DeepCopyHandler* m_deep_copy_handler = nullptr;
	239
	240	TimeRollingMean m_bytes_per_second;
	241	uint64_t m_last_snapshot_sync_seconds = 0;
	242
	243	uint64_t m_snapshot_bytes = 0;
	244	uint64_t m_last_snapshot_bytes = 0;
	245
	246	boost::accumulators::accumulator_set<
	247	uint64_t, boost::accumulators::stats<
	248	boost::accumulators::tag::rolling_mean>> m_bytes_per_snapshot{
	249	boost::accumulators::tag::rolling_window::window_size = 2};
	250	utime_t m_snapshot_replay_start;
	251
	252	uint32_t m_pending_snapshots = 0;
	253
	254	bool m_remote_image_updated = false;
	255	bool m_updating_sync_point = false;
	256	bool m_sync_in_progress = false;
	257
	258	PerfCounters *m_perf_counters = nullptr;
	259
	260	void load_local_image_meta();
	261	void handle_load_local_image_meta(int r);
	262
	263	void refresh_local_image();
	264	void handle_refresh_local_image(int r);
	265
	266	void refresh_remote_image();
	267	void handle_refresh_remote_image(int r);
	268
	269	void scan_local_mirror_snapshots(std::unique_lock<ceph::mutex>* locker);
	270	void scan_remote_mirror_snapshots(std::unique_lock<ceph::mutex>* locker);
	271
	272	void prune_non_primary_snapshot(uint64_t snap_id);
	273	void handle_prune_non_primary_snapshot(int r);
	274
	275	void copy_snapshots();
	276	void handle_copy_snapshots(int r);
	277
	278	void get_remote_image_state();
	279	void handle_get_remote_image_state(int r);
	280
	281	void get_local_image_state();
	282	void handle_get_local_image_state(int r);
	283
	284	void create_non_primary_snapshot();
	285	void handle_create_non_primary_snapshot(int r);
	286
	287	void update_mirror_image_state();
	288	void handle_update_mirror_image_state(int r);
	289
	290	void request_sync();
	291	void handle_request_sync(int r);
	292
	293	void copy_image();
	294	void handle_copy_image(int r);
	295	void handle_copy_image_progress(uint64_t object_number,
	296	uint64_t object_count);
	297	void handle_copy_image_read(uint64_t bytes_read);
	298
	299	void apply_image_state();
	300	void handle_apply_image_state(int r);
	301
	302	void update_non_primary_snapshot(bool complete);
	303	void handle_update_non_primary_snapshot(bool complete, int r);
	304
	305	void notify_image_update();
	306	void handle_notify_image_update(int r);
	307
	308	void unlink_peer(uint64_t remote_snap_id);
	309	void handle_unlink_peer(int r);
	310
	311	void finish_sync();
	312
	313	void register_local_update_watcher();
	314	void handle_register_local_update_watcher(int r);
	315
	316	void register_remote_update_watcher();
	317	void handle_register_remote_update_watcher(int r);
	318
	319	void unregister_remote_update_watcher();
	320	void handle_unregister_remote_update_watcher(int r);
	321
	322	void unregister_local_update_watcher();
	323	void handle_unregister_local_update_watcher(int r);
	324
	325	void wait_for_in_flight_ops();
	326	void handle_wait_for_in_flight_ops(int r);
	327
	328	void handle_image_update_notify();
	329
	330	void handle_replay_complete(int r, const std::string& description);
	331	void handle_replay_complete(std::unique_lock<ceph::mutex>* locker,
	332	int r, const std::string& description);
	333	void notify_status_updated();
	334
	335	bool is_replay_interrupted();
	336	bool is_replay_interrupted(std::unique_lock<ceph::mutex>* lock);
	337
	338	void register_perf_counters();
	339	void unregister_perf_counters();
	340	};
	341
	342	} // namespace snapshot
	343	} // namespace image_replayer
	344	} // namespace mirror
	345	} // namespace rbd
	346
	347	extern template class rbd::mirror::image_replayer::snapshot::Replayer<librbd::ImageCtx>;
	348
	349	#endif // RBD_MIRROR_IMAGE_REPLAYER_SNAPSHOT_REPLAYER_H