--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "PoolReplayer.h"
+#include <boost/bind.hpp>
+#include "common/Formatter.h"
+#include "common/admin_socket.h"
+#include "common/ceph_argparse.h"
+#include "common/code_environment.h"
+#include "common/common_init.h"
+#include "common/debug.h"
+#include "common/errno.h"
+#include "include/stringify.h"
+#include "cls/rbd/cls_rbd_client.h"
+#include "global/global_context.h"
+#include "librbd/internal.h"
+#include "librbd/Utils.h"
+#include "librbd/Watcher.h"
+#include "librbd/api/Mirror.h"
+#include "InstanceReplayer.h"
+#include "InstanceWatcher.h"
+#include "LeaderWatcher.h"
+#include "ServiceDaemon.h"
+#include "Threads.h"
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_rbd_mirror
+#undef dout_prefix
+#define dout_prefix *_dout << "rbd::mirror::PoolReplayer: " \
+ << this << " " << __func__ << ": "
+
+using std::chrono::seconds;
+using std::map;
+using std::string;
+using std::unique_ptr;
+using std::vector;
+
+using librbd::cls_client::dir_get_name;
+using librbd::util::create_async_context_callback;
+
+namespace rbd {
+namespace mirror {
+
+namespace {
+
+const std::string SERVICE_DAEMON_LEADER_KEY("leader");
+const std::string SERVICE_DAEMON_LOCAL_COUNT_KEY("image_local_count");
+const std::string SERVICE_DAEMON_REMOTE_COUNT_KEY("image_remote_count");
+
+const std::vector<std::string> UNIQUE_PEER_CONFIG_KEYS {
+ {"monmap", "mon_host", "mon_dns_srv_name", "key", "keyfile", "keyring"}};
+
+class PoolReplayerAdminSocketCommand {
+public:
+ PoolReplayerAdminSocketCommand(PoolReplayer *pool_replayer)
+ : pool_replayer(pool_replayer) {
+ }
+ virtual ~PoolReplayerAdminSocketCommand() {}
+ virtual bool call(Formatter *f, stringstream *ss) = 0;
+protected:
+ PoolReplayer *pool_replayer;
+};
+
+class StatusCommand : public PoolReplayerAdminSocketCommand {
+public:
+ explicit StatusCommand(PoolReplayer *pool_replayer)
+ : PoolReplayerAdminSocketCommand(pool_replayer) {
+ }
+
+ bool call(Formatter *f, stringstream *ss) override {
+ pool_replayer->print_status(f, ss);
+ return true;
+ }
+};
+
+class StartCommand : public PoolReplayerAdminSocketCommand {
+public:
+ explicit StartCommand(PoolReplayer *pool_replayer)
+ : PoolReplayerAdminSocketCommand(pool_replayer) {
+ }
+
+ bool call(Formatter *f, stringstream *ss) override {
+ pool_replayer->start();
+ return true;
+ }
+};
+
+class StopCommand : public PoolReplayerAdminSocketCommand {
+public:
+ explicit StopCommand(PoolReplayer *pool_replayer)
+ : PoolReplayerAdminSocketCommand(pool_replayer) {
+ }
+
+ bool call(Formatter *f, stringstream *ss) override {
+ pool_replayer->stop(true);
+ return true;
+ }
+};
+
+class RestartCommand : public PoolReplayerAdminSocketCommand {
+public:
+ explicit RestartCommand(PoolReplayer *pool_replayer)
+ : PoolReplayerAdminSocketCommand(pool_replayer) {
+ }
+
+ bool call(Formatter *f, stringstream *ss) override {
+ pool_replayer->restart();
+ return true;
+ }
+};
+
+class FlushCommand : public PoolReplayerAdminSocketCommand {
+public:
+ explicit FlushCommand(PoolReplayer *pool_replayer)
+ : PoolReplayerAdminSocketCommand(pool_replayer) {
+ }
+
+ bool call(Formatter *f, stringstream *ss) override {
+ pool_replayer->flush();
+ return true;
+ }
+};
+
+class LeaderReleaseCommand : public PoolReplayerAdminSocketCommand {
+public:
+ explicit LeaderReleaseCommand(PoolReplayer *pool_replayer)
+ : PoolReplayerAdminSocketCommand(pool_replayer) {
+ }
+
+ bool call(Formatter *f, stringstream *ss) override {
+ pool_replayer->release_leader();
+ return true;
+ }
+};
+
+class PoolReplayerAdminSocketHook : public AdminSocketHook {
+public:
+ PoolReplayerAdminSocketHook(CephContext *cct, const std::string &name,
+ PoolReplayer *pool_replayer)
+ : admin_socket(cct->get_admin_socket()) {
+ std::string command;
+ int r;
+
+ command = "rbd mirror status " + name;
+ r = admin_socket->register_command(command, command, this,
+ "get status for rbd mirror " + name);
+ if (r == 0) {
+ commands[command] = new StatusCommand(pool_replayer);
+ }
+
+ command = "rbd mirror start " + name;
+ r = admin_socket->register_command(command, command, this,
+ "start rbd mirror " + name);
+ if (r == 0) {
+ commands[command] = new StartCommand(pool_replayer);
+ }
+
+ command = "rbd mirror stop " + name;
+ r = admin_socket->register_command(command, command, this,
+ "stop rbd mirror " + name);
+ if (r == 0) {
+ commands[command] = new StopCommand(pool_replayer);
+ }
+
+ command = "rbd mirror restart " + name;
+ r = admin_socket->register_command(command, command, this,
+ "restart rbd mirror " + name);
+ if (r == 0) {
+ commands[command] = new RestartCommand(pool_replayer);
+ }
+
+ command = "rbd mirror flush " + name;
+ r = admin_socket->register_command(command, command, this,
+ "flush rbd mirror " + name);
+ if (r == 0) {
+ commands[command] = new FlushCommand(pool_replayer);
+ }
+
+ command = "rbd mirror leader release " + name;
+ r = admin_socket->register_command(command, command, this,
+ "release rbd mirror leader " + name);
+ if (r == 0) {
+ commands[command] = new LeaderReleaseCommand(pool_replayer);
+ }
+ }
+
+ ~PoolReplayerAdminSocketHook() override {
+ for (Commands::const_iterator i = commands.begin(); i != commands.end();
+ ++i) {
+ (void)admin_socket->unregister_command(i->first);
+ delete i->second;
+ }
+ }
+
+ bool call(std::string command, cmdmap_t& cmdmap, std::string format,
+ bufferlist& out) override {
+ Commands::const_iterator i = commands.find(command);
+ assert(i != commands.end());
+ Formatter *f = Formatter::create(format);
+ stringstream ss;
+ bool r = i->second->call(f, &ss);
+ delete f;
+ out.append(ss);
+ return r;
+ }
+
+private:
+ typedef std::map<std::string, PoolReplayerAdminSocketCommand*> Commands;
+
+ AdminSocket *admin_socket;
+ Commands commands;
+};
+
+} // anonymous namespace
+
+PoolReplayer::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) :
+ m_threads(threads),
+ m_service_daemon(service_daemon),
+ m_image_deleter(image_deleter),
+ m_local_pool_id(local_pool_id),
+ m_peer(peer),
+ m_args(args),
+ m_lock(stringify("rbd::mirror::PoolReplayer ") + stringify(peer)),
+ m_local_pool_watcher_listener(this, true),
+ m_remote_pool_watcher_listener(this, false),
+ m_pool_replayer_thread(this),
+ m_leader_listener(this)
+{
+}
+
+PoolReplayer::~PoolReplayer()
+{
+ delete m_asok_hook;
+ shut_down();
+}
+
+bool PoolReplayer::is_blacklisted() const {
+ Mutex::Locker locker(m_lock);
+ return m_blacklisted;
+}
+
+bool PoolReplayer::is_leader() const {
+ Mutex::Locker locker(m_lock);
+ return m_leader_watcher && m_leader_watcher->is_leader();
+}
+
+bool PoolReplayer::is_running() const {
+ return m_pool_replayer_thread.is_started();
+}
+
+void PoolReplayer::init()
+{
+ assert(!m_pool_replayer_thread.is_started());
+
+ // reset state
+ m_stopping = false;
+ m_blacklisted = false;
+
+ dout(20) << "replaying for " << m_peer << dendl;
+ int r = init_rados(g_ceph_context->_conf->cluster,
+ g_ceph_context->_conf->name.to_str(),
+ "local cluster", &m_local_rados, false);
+ if (r < 0) {
+ m_callout_id = m_service_daemon->add_or_update_callout(
+ m_local_pool_id, m_callout_id, service_daemon::CALLOUT_LEVEL_ERROR,
+ "unable to connect to local cluster");
+ return;
+ }
+
+ r = init_rados(m_peer.cluster_name, m_peer.client_name,
+ std::string("remote peer ") + stringify(m_peer),
+ &m_remote_rados, true);
+ if (r < 0) {
+ m_callout_id = m_service_daemon->add_or_update_callout(
+ m_local_pool_id, m_callout_id, service_daemon::CALLOUT_LEVEL_ERROR,
+ "unable to connect to remote cluster");
+ return;
+ }
+
+ r = m_local_rados->ioctx_create2(m_local_pool_id, m_local_io_ctx);
+ if (r < 0) {
+ derr << "error accessing local pool " << m_local_pool_id << ": "
+ << cpp_strerror(r) << dendl;
+ return;
+ }
+
+ std::string local_mirror_uuid;
+ r = librbd::cls_client::mirror_uuid_get(&m_local_io_ctx,
+ &local_mirror_uuid);
+ if (r < 0) {
+ derr << "failed to retrieve local mirror uuid from pool "
+ << m_local_io_ctx.get_pool_name() << ": " << cpp_strerror(r) << dendl;
+ m_callout_id = m_service_daemon->add_or_update_callout(
+ m_local_pool_id, m_callout_id, service_daemon::CALLOUT_LEVEL_ERROR,
+ "unable to query local mirror uuid");
+ return;
+ }
+
+ r = m_remote_rados->ioctx_create(m_local_io_ctx.get_pool_name().c_str(),
+ m_remote_io_ctx);
+ if (r < 0) {
+ derr << "error accessing remote pool " << m_local_io_ctx.get_pool_name()
+ << ": " << cpp_strerror(r) << dendl;
+ m_callout_id = m_service_daemon->add_or_update_callout(
+ m_local_pool_id, m_callout_id, service_daemon::CALLOUT_LEVEL_WARNING,
+ "unable to access remote pool");
+ return;
+ }
+
+ dout(20) << "connected to " << m_peer << dendl;
+
+ m_instance_replayer.reset(InstanceReplayer<>::create(
+ m_threads, m_service_daemon, m_image_deleter, m_local_rados,
+ local_mirror_uuid, m_local_pool_id));
+ m_instance_replayer->init();
+ m_instance_replayer->add_peer(m_peer.uuid, m_remote_io_ctx);
+
+ m_instance_watcher.reset(InstanceWatcher<>::create(
+ m_local_io_ctx, m_threads->work_queue, m_instance_replayer.get()));
+ r = m_instance_watcher->init();
+ if (r < 0) {
+ derr << "error initializing instance watcher: " << cpp_strerror(r) << dendl;
+ m_callout_id = m_service_daemon->add_or_update_callout(
+ m_local_pool_id, m_callout_id, service_daemon::CALLOUT_LEVEL_ERROR,
+ "unable to initialize instance messenger object");
+ return;
+ }
+
+ m_leader_watcher.reset(new LeaderWatcher<>(m_threads, m_local_io_ctx,
+ &m_leader_listener));
+ r = m_leader_watcher->init();
+ if (r < 0) {
+ derr << "error initializing leader watcher: " << cpp_strerror(r) << dendl;
+ m_callout_id = m_service_daemon->add_or_update_callout(
+ m_local_pool_id, m_callout_id, service_daemon::CALLOUT_LEVEL_ERROR,
+ "unable to initialize leader messenger object");
+ return;
+ }
+
+ if (m_callout_id != service_daemon::CALLOUT_ID_NONE) {
+ m_service_daemon->remove_callout(m_local_pool_id, m_callout_id);
+ m_callout_id = service_daemon::CALLOUT_ID_NONE;
+ }
+
+ m_pool_replayer_thread.create("pool replayer");
+}
+
+void PoolReplayer::shut_down() {
+ m_stopping = true;
+ {
+ Mutex::Locker l(m_lock);
+ m_cond.Signal();
+ }
+ if (m_pool_replayer_thread.is_started()) {
+ m_pool_replayer_thread.join();
+ }
+ if (m_leader_watcher) {
+ m_leader_watcher->shut_down();
+ m_leader_watcher.reset();
+ }
+ if (m_instance_watcher) {
+ m_instance_watcher->shut_down();
+ m_instance_watcher.reset();
+ }
+ if (m_instance_replayer) {
+ m_instance_replayer->shut_down();
+ m_instance_replayer.reset();
+ }
+
+ assert(!m_local_pool_watcher);
+ assert(!m_remote_pool_watcher);
+ m_local_rados.reset();
+ m_remote_rados.reset();
+}
+
+int PoolReplayer::init_rados(const std::string &cluster_name,
+ const std::string &client_name,
+ const std::string &description,
+ RadosRef *rados_ref,
+ bool strip_cluster_overrides) {
+ rados_ref->reset(new librados::Rados());
+
+ // NOTE: manually bootstrap a CephContext here instead of via
+ // the librados API to avoid mixing global singletons between
+ // the librados shared library and the daemon
+ // TODO: eliminate intermingling of global singletons within Ceph APIs
+ CephInitParameters iparams(CEPH_ENTITY_TYPE_CLIENT);
+ if (client_name.empty() || !iparams.name.from_str(client_name)) {
+ derr << "error initializing cluster handle for " << description << dendl;
+ return -EINVAL;
+ }
+
+ CephContext *cct = common_preinit(iparams, CODE_ENVIRONMENT_LIBRARY,
+ CINIT_FLAG_UNPRIVILEGED_DAEMON_DEFAULTS);
+ cct->_conf->cluster = cluster_name;
+
+ // librados::Rados::conf_read_file
+ int r = cct->_conf->parse_config_files(nullptr, nullptr, 0);
+ if (r < 0) {
+ derr << "could not read ceph conf for " << description << ": "
+ << cpp_strerror(r) << dendl;
+ cct->put();
+ return r;
+ }
+
+ // preserve cluster-specific config settings before applying environment/cli
+ // overrides
+ std::map<std::string, std::string> config_values;
+ if (strip_cluster_overrides) {
+ // remote peer connections shouldn't apply cluster-specific
+ // configuration settings
+ for (auto& key : UNIQUE_PEER_CONFIG_KEYS) {
+ config_values[key] = cct->_conf->get_val<std::string>(key);
+ }
+ }
+
+ cct->_conf->parse_env();
+
+ // librados::Rados::conf_parse_env
+ std::vector<const char*> args;
+ env_to_vec(args, nullptr);
+ r = cct->_conf->parse_argv(args);
+ if (r < 0) {
+ derr << "could not parse environment for " << description << ":"
+ << cpp_strerror(r) << dendl;
+ cct->put();
+ return r;
+ }
+
+ if (!m_args.empty()) {
+ // librados::Rados::conf_parse_argv
+ args = m_args;
+ r = cct->_conf->parse_argv(args);
+ if (r < 0) {
+ derr << "could not parse command line args for " << description << ": "
+ << cpp_strerror(r) << dendl;
+ cct->put();
+ return r;
+ }
+ }
+
+ if (strip_cluster_overrides) {
+ // remote peer connections shouldn't apply cluster-specific
+ // configuration settings
+ for (auto& pair : config_values) {
+ auto value = cct->_conf->get_val<std::string>(pair.first);
+ if (pair.second != value) {
+ dout(0) << "reverting global config option override: "
+ << pair.first << ": " << value << " -> " << pair.second
+ << dendl;
+ cct->_conf->set_val_or_die(pair.first, pair.second);
+ }
+ }
+ }
+
+ if (!g_ceph_context->_conf->admin_socket.empty()) {
+ cct->_conf->set_val_or_die("admin_socket",
+ "$run_dir/$name.$pid.$cluster.$cctid.asok");
+ }
+
+ // disable unnecessary librbd cache
+ cct->_conf->set_val_or_die("rbd_cache", "false");
+ cct->_conf->apply_changes(nullptr);
+ cct->_conf->complain_about_parse_errors(cct);
+
+ r = (*rados_ref)->init_with_context(cct);
+ assert(r == 0);
+ cct->put();
+
+ r = (*rados_ref)->connect();
+ if (r < 0) {
+ derr << "error connecting to " << description << ": "
+ << cpp_strerror(r) << dendl;
+ return r;
+ }
+
+ return 0;
+}
+
+void PoolReplayer::run()
+{
+ dout(20) << "enter" << dendl;
+
+ while (!m_stopping) {
+ std::string asok_hook_name = m_local_io_ctx.get_pool_name() + " " +
+ m_peer.cluster_name;
+ if (m_asok_hook_name != asok_hook_name || m_asok_hook == nullptr) {
+ m_asok_hook_name = asok_hook_name;
+ delete m_asok_hook;
+
+ m_asok_hook = new PoolReplayerAdminSocketHook(g_ceph_context,
+ m_asok_hook_name, this);
+ }
+
+ Mutex::Locker locker(m_lock);
+ if ((m_local_pool_watcher && m_local_pool_watcher->is_blacklisted()) ||
+ (m_remote_pool_watcher && m_remote_pool_watcher->is_blacklisted())) {
+ m_blacklisted = true;
+ m_stopping = true;
+ break;
+ }
+
+ if (!m_stopping) {
+ m_cond.WaitInterval(m_lock, utime_t(1, 0));
+ }
+ }
+}
+
+void PoolReplayer::print_status(Formatter *f, stringstream *ss)
+{
+ dout(20) << "enter" << dendl;
+
+ if (!f) {
+ return;
+ }
+
+ Mutex::Locker l(m_lock);
+
+ f->open_object_section("pool_replayer_status");
+ f->dump_string("pool", m_local_io_ctx.get_pool_name());
+ f->dump_stream("peer") << m_peer;
+ f->dump_string("instance_id", m_instance_watcher->get_instance_id());
+
+ std::string leader_instance_id;
+ m_leader_watcher->get_leader_instance_id(&leader_instance_id);
+ f->dump_string("leader_instance_id", leader_instance_id);
+
+ bool leader = m_leader_watcher->is_leader();
+ f->dump_bool("leader", leader);
+ if (leader) {
+ std::vector<std::string> instance_ids;
+ m_leader_watcher->list_instances(&instance_ids);
+ f->open_array_section("instances");
+ for (auto instance_id : instance_ids) {
+ f->dump_string("instance_id", instance_id);
+ }
+ f->close_section();
+ }
+
+ f->dump_string("local_cluster_admin_socket",
+ reinterpret_cast<CephContext *>(m_local_io_ctx.cct())->_conf->
+ get_val<std::string>("admin_socket"));
+ f->dump_string("remote_cluster_admin_socket",
+ reinterpret_cast<CephContext *>(m_remote_io_ctx.cct())->_conf->
+ get_val<std::string>("admin_socket"));
+
+ f->open_object_section("sync_throttler");
+ m_instance_watcher->print_sync_status(f, ss);
+ f->close_section();
+
+ m_instance_replayer->print_status(f, ss);
+
+ f->close_section();
+ f->flush(*ss);
+}
+
+void PoolReplayer::start()
+{
+ dout(20) << "enter" << dendl;
+
+ Mutex::Locker l(m_lock);
+
+ if (m_stopping) {
+ return;
+ }
+
+ m_instance_replayer->start();
+}
+
+void PoolReplayer::stop(bool manual)
+{
+ dout(20) << "enter: manual=" << manual << dendl;
+
+ Mutex::Locker l(m_lock);
+ if (!manual) {
+ m_stopping = true;
+ m_cond.Signal();
+ return;
+ } else if (m_stopping) {
+ return;
+ }
+
+ m_instance_replayer->stop();
+}
+
+void PoolReplayer::restart()
+{
+ dout(20) << "enter" << dendl;
+
+ Mutex::Locker l(m_lock);
+
+ if (m_stopping) {
+ return;
+ }
+
+ m_instance_replayer->restart();
+}
+
+void PoolReplayer::flush()
+{
+ dout(20) << "enter" << dendl;
+
+ Mutex::Locker l(m_lock);
+
+ if (m_stopping || m_manual_stop) {
+ return;
+ }
+
+ m_instance_replayer->flush();
+}
+
+void PoolReplayer::release_leader()
+{
+ dout(20) << "enter" << dendl;
+
+ Mutex::Locker l(m_lock);
+
+ if (m_stopping || !m_leader_watcher) {
+ return;
+ }
+
+ m_leader_watcher->release_leader();
+}
+
+void PoolReplayer::handle_update(const std::string &mirror_uuid,
+ ImageIds &&added_image_ids,
+ ImageIds &&removed_image_ids) {
+ if (m_stopping) {
+ return;
+ }
+
+ dout(10) << "mirror_uuid=" << mirror_uuid << ", "
+ << "added_count=" << added_image_ids.size() << ", "
+ << "removed_count=" << removed_image_ids.size() << dendl;
+ Mutex::Locker locker(m_lock);
+ if (!m_leader_watcher->is_leader()) {
+ return;
+ }
+
+ m_service_daemon->add_or_update_attribute(
+ m_local_pool_id, SERVICE_DAEMON_LOCAL_COUNT_KEY,
+ m_local_pool_watcher->get_image_count());
+ if (m_remote_pool_watcher) {
+ m_service_daemon->add_or_update_attribute(
+ m_local_pool_id, SERVICE_DAEMON_REMOTE_COUNT_KEY,
+ m_remote_pool_watcher->get_image_count());
+ }
+
+ m_update_op_tracker.start_op();
+ Context *ctx = new FunctionContext([this](int r) {
+ dout(20) << "complete handle_update: r=" << r << dendl;
+ m_update_op_tracker.finish_op();
+ });
+
+ C_Gather *gather_ctx = new C_Gather(g_ceph_context, ctx);
+
+ for (auto &image_id : added_image_ids) {
+ // for now always send to myself (the leader)
+ std::string &instance_id = m_instance_watcher->get_instance_id();
+ m_instance_watcher->notify_image_acquire(instance_id, image_id.global_id,
+ gather_ctx->new_sub());
+ }
+
+ if (!mirror_uuid.empty()) {
+ for (auto &image_id : removed_image_ids) {
+ // for now always send to myself (the leader)
+ std::string &instance_id = m_instance_watcher->get_instance_id();
+ m_instance_watcher->notify_peer_image_removed(instance_id,
+ image_id.global_id,
+ mirror_uuid,
+ gather_ctx->new_sub());
+ }
+ }
+
+ gather_ctx->activate();
+}
+
+void PoolReplayer::handle_post_acquire_leader(Context *on_finish) {
+ dout(20) << dendl;
+
+ m_service_daemon->add_or_update_attribute(m_local_pool_id,
+ SERVICE_DAEMON_LEADER_KEY, true);
+ m_instance_watcher->handle_acquire_leader();
+ init_local_pool_watcher(on_finish);
+}
+
+void PoolReplayer::handle_pre_release_leader(Context *on_finish) {
+ dout(20) << dendl;
+
+ m_service_daemon->remove_attribute(m_local_pool_id, SERVICE_DAEMON_LEADER_KEY);
+ m_instance_watcher->handle_release_leader();
+ shut_down_pool_watchers(on_finish);
+}
+
+void PoolReplayer::init_local_pool_watcher(Context *on_finish) {
+ dout(20) << dendl;
+
+ Mutex::Locker locker(m_lock);
+ assert(!m_local_pool_watcher);
+ m_local_pool_watcher.reset(new PoolWatcher<>(
+ m_threads, m_local_io_ctx, m_local_pool_watcher_listener));
+
+ // ensure the initial set of local images is up-to-date
+ // after acquiring the leader role
+ auto ctx = new FunctionContext([this, on_finish](int r) {
+ handle_init_local_pool_watcher(r, on_finish);
+ });
+ m_local_pool_watcher->init(create_async_context_callback(
+ m_threads->work_queue, ctx));
+}
+
+void PoolReplayer::handle_init_local_pool_watcher(int r, Context *on_finish) {
+ dout(20) << "r=" << r << dendl;
+ if (r < 0) {
+ derr << "failed to retrieve local images: " << cpp_strerror(r) << dendl;
+ on_finish->complete(r);
+ return;
+ }
+
+ init_remote_pool_watcher(on_finish);
+}
+
+void PoolReplayer::init_remote_pool_watcher(Context *on_finish) {
+ dout(20) << dendl;
+
+ Mutex::Locker locker(m_lock);
+ assert(!m_remote_pool_watcher);
+ m_remote_pool_watcher.reset(new PoolWatcher<>(
+ m_threads, m_remote_io_ctx, m_remote_pool_watcher_listener));
+ m_remote_pool_watcher->init(create_async_context_callback(
+ m_threads->work_queue, on_finish));
+
+ m_cond.Signal();
+}
+
+void PoolReplayer::shut_down_pool_watchers(Context *on_finish) {
+ dout(20) << dendl;
+
+ {
+ Mutex::Locker locker(m_lock);
+ if (m_local_pool_watcher) {
+ Context *ctx = new FunctionContext([this, on_finish](int r) {
+ handle_shut_down_pool_watchers(r, on_finish);
+ });
+ ctx = create_async_context_callback(m_threads->work_queue, ctx);
+
+ auto gather_ctx = new C_Gather(g_ceph_context, ctx);
+ m_local_pool_watcher->shut_down(gather_ctx->new_sub());
+ if (m_remote_pool_watcher) {
+ m_remote_pool_watcher->shut_down(gather_ctx->new_sub());
+ }
+ gather_ctx->activate();
+ return;
+ }
+ }
+
+ on_finish->complete(0);
+}
+
+void PoolReplayer::handle_shut_down_pool_watchers(int r, Context *on_finish) {
+ dout(20) << "r=" << r << dendl;
+
+ {
+ Mutex::Locker locker(m_lock);
+ assert(m_local_pool_watcher);
+ m_local_pool_watcher.reset();
+
+ if (m_remote_pool_watcher) {
+ m_remote_pool_watcher.reset();
+ }
+ }
+ wait_for_update_ops(on_finish);
+}
+
+void PoolReplayer::wait_for_update_ops(Context *on_finish) {
+ dout(20) << dendl;
+
+ Mutex::Locker locker(m_lock);
+
+ Context *ctx = new FunctionContext([this, on_finish](int r) {
+ handle_wait_for_update_ops(r, on_finish);
+ });
+ ctx = create_async_context_callback(m_threads->work_queue, ctx);
+
+ m_update_op_tracker.wait_for_ops(ctx);
+}
+
+void PoolReplayer::handle_wait_for_update_ops(int r, Context *on_finish) {
+ dout(20) << "r=" << r << dendl;
+
+ assert(r == 0);
+
+ Mutex::Locker locker(m_lock);
+ m_instance_replayer->release_all(on_finish);
+}
+
+void PoolReplayer::handle_update_leader(const std::string &leader_instance_id) {
+ dout(20) << "leader_instance_id=" << leader_instance_id << dendl;
+
+ m_instance_watcher->handle_update_leader(leader_instance_id);
+}
+
+} // namespace mirror
+} // namespace rbd
Code Review / stor4nfv.git / blobdiff