--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_RBD_MIRROR_POOL_REPLAYER_H
+#define CEPH_RBD_MIRROR_POOL_REPLAYER_H
+
+#include "common/AsyncOpTracker.h"
+#include "common/Cond.h"
+#include "common/Mutex.h"
+#include "common/WorkQueue.h"
+#include "include/rados/librados.hpp"
+
+#include "ClusterWatcher.h"
+#include "LeaderWatcher.h"
+#include "PoolWatcher.h"
+#include "ImageDeleter.h"
+#include "types.h"
+#include "tools/rbd_mirror/service_daemon/Types.h"
+
+#include <set>
+#include <map>
+#include <memory>
+#include <atomic>
+#include <string>
+
+class AdminSocketHook;
+
+namespace librbd { class ImageCtx; }
+
+namespace rbd {
+namespace mirror {
+
+template <typename> class InstanceReplayer;
+template <typename> class InstanceWatcher;
+template <typename> class ServiceDaemon;
+template <typename> struct Threads;
+
+/**
+ * Controls mirroring for a single remote cluster.
+ */
+class PoolReplayer {
+public:
+ PoolReplayer(Threads<librbd::ImageCtx> *threads,
+ ServiceDaemon<librbd::ImageCtx>* service_daemon,
+ ImageDeleter<>* image_deleter,
+ int64_t local_pool_id, const peer_t &peer,
+ const std::vector<const char*> &args);
+ ~PoolReplayer();
+ PoolReplayer(const PoolReplayer&) = delete;
+ PoolReplayer& operator=(const PoolReplayer&) = delete;
+
+ bool is_blacklisted() const;
+ bool is_leader() const;
+ bool is_running() const;
+
+ void init();
+ void shut_down();
+
+ void run();
+
+ void print_status(Formatter *f, stringstream *ss);
+ void start();
+ void stop(bool manual);
+ void restart();
+ void flush();
+ void release_leader();
+
+private:
+ struct PoolWatcherListener : public PoolWatcher<>::Listener {
+ PoolReplayer *pool_replayer;
+ bool local;
+
+ PoolWatcherListener(PoolReplayer *pool_replayer, bool local)
+ : pool_replayer(pool_replayer), local(local) {
+ }
+
+ void handle_update(const std::string &mirror_uuid,
+ ImageIds &&added_image_ids,
+ ImageIds &&removed_image_ids) override {
+ pool_replayer->handle_update((local ? "" : mirror_uuid),
+ std::move(added_image_ids),
+ std::move(removed_image_ids));
+ }
+ };
+
+ void handle_update(const std::string &mirror_uuid,
+ ImageIds &&added_image_ids,
+ ImageIds &&removed_image_ids);
+
+ int init_rados(const std::string &cluster_name,
+ const std::string &client_name,
+ const std::string &description, RadosRef *rados_ref,
+ bool strip_cluster_overrides);
+
+ void handle_post_acquire_leader(Context *on_finish);
+ void handle_pre_release_leader(Context *on_finish);
+
+ void init_local_pool_watcher(Context *on_finish);
+ void handle_init_local_pool_watcher(int r, Context *on_finish);
+
+ void init_remote_pool_watcher(Context *on_finish);
+
+ void shut_down_pool_watchers(Context *on_finish);
+ void handle_shut_down_pool_watchers(int r, Context *on_finish);
+
+ void wait_for_update_ops(Context *on_finish);
+ void handle_wait_for_update_ops(int r, Context *on_finish);
+
+ void handle_update_leader(const std::string &leader_instance_id);
+
+ Threads<librbd::ImageCtx> *m_threads;
+ ServiceDaemon<librbd::ImageCtx>* m_service_daemon;
+ ImageDeleter<>* m_image_deleter;
+ int64_t m_local_pool_id = -1;
+ peer_t m_peer;
+ std::vector<const char*> m_args;
+
+ mutable Mutex m_lock;
+ Cond m_cond;
+ std::atomic<bool> m_stopping = { false };
+ bool m_manual_stop = false;
+ bool m_blacklisted = false;
+
+ RadosRef m_local_rados;
+ RadosRef m_remote_rados;
+
+ librados::IoCtx m_local_io_ctx;
+ librados::IoCtx m_remote_io_ctx;
+
+ PoolWatcherListener m_local_pool_watcher_listener;
+ std::unique_ptr<PoolWatcher<> > m_local_pool_watcher;
+
+ PoolWatcherListener m_remote_pool_watcher_listener;
+ std::unique_ptr<PoolWatcher<> > m_remote_pool_watcher;
+
+ std::unique_ptr<InstanceReplayer<librbd::ImageCtx>> m_instance_replayer;
+
+ std::string m_asok_hook_name;
+ AdminSocketHook *m_asok_hook = nullptr;
+
+ service_daemon::CalloutId m_callout_id = service_daemon::CALLOUT_ID_NONE;
+
+ class PoolReplayerThread : public Thread {
+ PoolReplayer *m_pool_replayer;
+ public:
+ PoolReplayerThread(PoolReplayer *pool_replayer)
+ : m_pool_replayer(pool_replayer) {
+ }
+ void *entry() override {
+ m_pool_replayer->run();
+ return 0;
+ }
+ } m_pool_replayer_thread;
+
+ class LeaderListener : public LeaderWatcher<>::Listener {
+ public:
+ LeaderListener(PoolReplayer *pool_replayer)
+ : m_pool_replayer(pool_replayer) {
+ }
+
+ protected:
+ void post_acquire_handler(Context *on_finish) override {
+ m_pool_replayer->handle_post_acquire_leader(on_finish);
+ }
+
+ void pre_release_handler(Context *on_finish) override {
+ m_pool_replayer->handle_pre_release_leader(on_finish);
+ }
+
+ void update_leader_handler(
+ const std::string &leader_instance_id) override {
+ m_pool_replayer->handle_update_leader(leader_instance_id);
+ }
+
+ private:
+ PoolReplayer *m_pool_replayer;
+ } m_leader_listener;
+
+ std::unique_ptr<LeaderWatcher<> > m_leader_watcher;
+ std::unique_ptr<InstanceWatcher<librbd::ImageCtx> > m_instance_watcher;
+ AsyncOpTracker m_update_op_tracker;
+};
+
+} // namespace mirror
+} // namespace rbd
+
+#endif // CEPH_RBD_MIRROR_POOL_REPLAYER_H