+++ /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