librbd: replace ImageCtx::get_work_queue with direct AsioEngine usage

The ImageCtx will contain a shared_ptr to its AsioEngine and previous
usage for get_work_queue now directly instantiate their own AsioEngine
instance.

Additionally, AsioEngine is now lighter weight by re-using the
io_context exposed via a neorados-wrapped librados API.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>

28 files changed, 177 insertions, 154 deletions

diff --git a/src/librbd/AsioEngine.cc b/src/librbd/AsioEngine.cc
index 99e381f0bed..55916ab654e 100644
--- a/src/librbd/AsioEngine.cc
+++ b/src/librbd/AsioEngine.cc
@@ -2,9 +2,11 @@
 // vim: ts=8 sw=2 smarttab
 
 #include "librbd/AsioEngine.h"
+#include "include/stringify.h"
+#include "include/neorados/RADOS.hpp"
+#include "include/rados/librados.hpp"
 #include "common/dout.h"
 #include "librbd/asio/ContextWQ.h"
-#include <boost/system/error_code.hpp>
 
 #define dout_subsys ceph_subsys_rbd
 #undef dout_prefix
@@ -13,43 +15,29 @@
 
 namespace librbd {
 
-AsioEngine::AsioEngine(CephContext* cct)
-  : m_cct(cct) {
-  init();
-}
-
-AsioEngine::~AsioEngine() {
-  shut_down();
-}
-
-void AsioEngine::init() {
-  auto thread_count = m_cct->_conf.get_val<uint64_t>("rbd_op_threads");
-  m_threads.reserve(thread_count);
-
-  // prevent IO context from exiting if no work is currently scheduled
-  m_work_guard.emplace(boost::asio::make_work_guard(m_io_context));
-
-  ldout(m_cct, 5) << "spawning " << thread_count << " threads" << dendl;
-  for (auto i = 0U; i < thread_count; i++) {
-    m_threads.emplace_back([=] {
-      boost::system::error_code ec;
-      m_io_context.run(ec);
-    });
+AsioEngine::AsioEngine(std::shared_ptr<librados::Rados> rados)
+  : m_rados_api(std::make_shared<neorados::RADOS>(
+      neorados::RADOS::make_with_librados(*rados))),
+    m_cct(m_rados_api->cct()),
+    m_io_context(m_rados_api->get_io_context()),
+    m_context_wq(std::make_unique<asio::ContextWQ>(m_io_context)) {
+  ldout(m_cct, 20) << dendl;
+
+  auto rados_threads = m_cct->_conf.get_val<uint64_t>("librados_thread_count");
+  auto rbd_threads = m_cct->_conf.get_val<uint64_t>("rbd_op_threads");
+  if (rbd_threads > rados_threads) {
+    // inherit the librados thread count -- but increase it if librbd wants to
+    // utilize more threads
+    m_cct->_conf.set_val("librados_thread_count", stringify(rbd_threads));
   }
-
-  m_work_queue = std::make_unique<asio::ContextWQ>(m_io_context);
 }
 
-void AsioEngine::shut_down() {
-  ldout(m_cct, 5) << "joining threads" << dendl;
-
-  m_work_guard.reset();
-  for (auto& thread : m_threads) {
-    thread.join();
-  }
-  m_threads.clear();
+AsioEngine::AsioEngine(librados::IoCtx& io_ctx)
+  : AsioEngine(std::make_shared<librados::Rados>(io_ctx)) {
+}
 
-  ldout(m_cct, 5) << "done" << dendl;
+AsioEngine::~AsioEngine() {
+  ldout(m_cct, 20) << dendl;
 }
 
 } // namespace librbd
diff --git a/src/librbd/AsioEngine.h b/src/librbd/AsioEngine.h
index 6c7f1a7b171..a06b611b49d 100644
--- a/src/librbd/AsioEngine.h
+++ b/src/librbd/AsioEngine.h
@@ -5,47 +5,50 @@
 #define CEPH_LIBRBD_ASIO_ENGINE_H
 
 #include "include/common_fwd.h"
+#include "include/rados/librados_fwd.hpp"
 #include <memory>
-#include <optional>
-#include <thread>
-#include <vector>
-#include <boost/asio/executor_work_guard.hpp>
 #include <boost/asio/io_context.hpp>
 
+namespace neorados { struct RADOS; }
+
 namespace librbd {
 
 namespace asio { struct ContextWQ; }
 
 class AsioEngine {
 public:
-  explicit AsioEngine(CephContext* cct);
+  explicit AsioEngine(std::shared_ptr<librados::Rados> rados);
+  explicit AsioEngine(librados::IoCtx& io_ctx);
   ~AsioEngine();
 
+  AsioEngine(AsioEngine&&) = delete;
+  AsioEngine(const AsioEngine&) = delete;
+  AsioEngine& operator=(const AsioEngine&) = delete;
+
+  inline neorados::RADOS& get_rados_api() {
+    return *m_rados_api;
+  }
+
   inline boost::asio::io_context& get_io_context() {
     return m_io_context;
   }
+  inline operator boost::asio::io_context&() {
+    return m_io_context;
+  }
+  inline boost::asio::io_context::executor_type get_executor() {
+    return m_io_context.get_executor();
+  }
 
   inline asio::ContextWQ* get_work_queue() {
-    return m_work_queue.get();
+    return m_context_wq.get();
   }
 
 private:
-  typedef std::vector<std::thread> Threads;
-
-  typedef boost::asio::executor_work_guard<
-    boost::asio::io_context::executor_type> WorkGuard;
-
+  std::shared_ptr<neorados::RADOS> m_rados_api;
   CephContext* m_cct;
-  Threads m_threads;
-
-  boost::asio::io_context m_io_context;
-  std::optional<WorkGuard> m_work_guard;
-
-  std::unique_ptr<asio::ContextWQ> m_work_queue;
-
-  void init();
-  void shut_down();
 
+  boost::asio::io_context& m_io_context;
+  std::unique_ptr<asio::ContextWQ> m_context_wq;
 };
 
 } // namespace librbd
diff --git a/src/librbd/CMakeLists.txt b/src/librbd/CMakeLists.txt
index f4c48c61817..5f9063042ad 100644
--- a/src/librbd/CMakeLists.txt
+++ b/src/librbd/CMakeLists.txt
@@ -234,10 +234,11 @@ target_link_libraries(librbd PRIVATE
   rbd_internal
   rbd_types
   journal
+  libneorados
   librados
-  cls_rbd_client 
-  cls_lock_client 
-  cls_journal_client 
+  cls_rbd_client
+  cls_lock_client
+  cls_journal_client
   ceph-common
   pthread
   ${CMAKE_DL_LIBS}
diff --git a/src/librbd/ImageCtx.cc b/src/librbd/ImageCtx.cc
index c5746f66eff..6b9e934f9c4 100644
--- a/src/librbd/ImageCtx.cc
+++ b/src/librbd/ImageCtx.cc
@@ -71,11 +71,6 @@ public:
   }
 };
 
-boost::asio::io_context& get_asio_engine_io_context(CephContext* cct) {
-  auto asio_engine_singleton = ImageCtx::get_asio_engine(cct);
-  return asio_engine_singleton->get_io_context();
-}
-
 } // anonymous namespace
 
   const string ImageCtx::METADATA_CONF_PREFIX = "conf_";
@@ -91,6 +86,7 @@ boost::asio::io_context& get_asio_engine_io_context(CephContext* cct) {
       read_only_flags(ro ? IMAGE_READ_ONLY_FLAG_USER : 0U),
       exclusive_locked(false),
       name(image_name),
+      asio_engine(std::make_shared<AsioEngine>(p)),
       image_watcher(NULL),
       journal(NULL),
       owner_lock(ceph::make_shared_mutex(util::unique_lock_name("librbd::ImageCtx::owner_lock", this))),
@@ -109,8 +105,7 @@ boost::asio::io_context& get_asio_engine_io_context(CephContext* cct) {
       state(new ImageState<>(this)),
       operations(new Operations<>(*this)),
       exclusive_lock(nullptr), object_map(nullptr),
-      io_context(get_asio_engine_io_context(cct)),
-      op_work_queue(nullptr),
+      op_work_queue(asio_engine->get_work_queue()),
       plugin_registry(new PluginRegistry<ImageCtx>(this)),
       external_callback_completions(32),
       event_socket_completions(32),
@@ -125,7 +120,6 @@ boost::asio::io_context& get_asio_engine_io_context(CephContext* cct) {
     // FIPS zeroization audit 20191117: this memset is not security related.
     memset(&header, 0, sizeof(header));
 
-    get_work_queue(cct, &op_work_queue);
     io_image_dispatcher = new io::ImageDispatcher<ImageCtx>(this);
     io_object_dispatcher = new io::ObjectDispatcher<ImageCtx>(this);
 
@@ -899,16 +893,6 @@ boost::asio::io_context& get_asio_engine_io_context(CephContext* cct) {
     journal_policy = policy;
   }
 
-  AsioEngine* ImageCtx::get_asio_engine(CephContext* cct) {
-    return &cct->lookup_or_create_singleton_object<AsioEngine>(
-      "librbd::AsioEngine", false, cct);
-  }
-
-  void ImageCtx::get_work_queue(CephContext *cct,
-                                asio::ContextWQ **op_work_queue) {
-    *op_work_queue = get_asio_engine(cct)->get_work_queue();
-  }
-
   void ImageCtx::get_timer_instance(CephContext *cct, SafeTimer **timer,
                                     ceph::mutex **timer_lock) {
     auto safe_timer_singleton =
diff --git a/src/librbd/ImageCtx.h b/src/librbd/ImageCtx.h
index b134117dc2c..43ed32cac35 100644
--- a/src/librbd/ImageCtx.h
+++ b/src/librbd/ImageCtx.h
@@ -8,6 +8,7 @@
 #include <atomic>
 #include <list>
 #include <map>
+#include <memory>
 #include <set>
 #include <string>
 #include <vector>
@@ -32,14 +33,15 @@
 #include "librbd/AsyncRequest.h"
 #include "librbd/Types.h"
 
-#include <boost/asio/io_context.hpp>
 #include <boost/lockfree/policies.hpp>
 #include <boost/lockfree/queue.hpp>
 
-class Finisher;
-class ThreadPool;
 class SafeTimer;
 
+namespace neorados {
+class RADOS;
+} // namespace neorados
+
 namespace librbd {
 
   struct AsioEngine;
@@ -109,6 +111,9 @@ namespace librbd {
     std::string name;
     cls::rbd::SnapshotNamespace snap_namespace;
     std::string snap_name;
+
+    std::shared_ptr<AsioEngine> asio_engine;
+
     IoCtx data_ctx, md_ctx;
     ImageWatcher<ImageCtx> *image_watcher;
     Journal<ImageCtx> *journal;
@@ -181,8 +186,6 @@ namespace librbd {
 
     xlist<operation::ResizeRequest<ImageCtx>*> resize_reqs;
 
-    boost::asio::io_context& io_context;
-
     io::ImageDispatcherInterface *io_image_dispatcher = nullptr;
     io::ObjectDispatcherInterface *io_object_dispatcher = nullptr;
 
@@ -344,9 +347,6 @@ namespace librbd {
     journal::Policy *get_journal_policy() const;
     void set_journal_policy(journal::Policy *policy);
 
-    static AsioEngine* get_asio_engine(CephContext* cct);
-    static void get_work_queue(CephContext *cct,
-                               asio::ContextWQ **op_work_queue);
     static void get_timer_instance(CephContext *cct, SafeTimer **timer,
                                    ceph::mutex **timer_lock);
   };
diff --git a/src/librbd/ImageState.cc b/src/librbd/ImageState.cc
index 8e831a6fdc7..96f57f87910 100644
--- a/src/librbd/ImageState.cc
+++ b/src/librbd/ImageState.cc
@@ -7,6 +7,7 @@
 #include "common/errno.h"
 #include "common/Cond.h"
 #include "common/WorkQueue.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/Utils.h"
 #include "librbd/asio/ContextWQ.h"
@@ -233,11 +234,11 @@ private:
 
 class QuiesceWatchers {
 public:
-  explicit QuiesceWatchers(CephContext *cct)
+  explicit QuiesceWatchers(CephContext *cct, asio::ContextWQ* work_queue)
     : m_cct(cct),
+      m_work_queue(work_queue),
       m_lock(ceph::make_mutex(util::unique_lock_name(
         "librbd::QuiesceWatchers::m_lock", this))) {
-    ImageCtx::get_work_queue(m_cct, &m_work_queue);
   }
 
   ~QuiesceWatchers() {
@@ -423,7 +424,8 @@ ImageState<I>::ImageState(I *image_ctx)
     m_lock(ceph::make_mutex(util::unique_lock_name("librbd::ImageState::m_lock", this))),
     m_last_refresh(0), m_refresh_seq(0),
     m_update_watchers(new ImageUpdateWatchers(image_ctx->cct)),
-    m_quiesce_watchers(new QuiesceWatchers(image_ctx->cct)) {
+    m_quiesce_watchers(new QuiesceWatchers(
+      image_ctx->cct, image_ctx->asio_engine->get_work_queue())) {
 }
 
 template <typename I>
diff --git a/src/librbd/api/Image.cc b/src/librbd/api/Image.cc
index d714aef3db5..db2a82eb5c6 100644
--- a/src/librbd/api/Image.cc
+++ b/src/librbd/api/Image.cc
@@ -7,6 +7,7 @@
 #include "common/errno.h"
 #include "common/Cond.h"
 #include "cls/rbd/cls_rbd_client.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/DeepCopyRequest.h"
 #include "librbd/ExclusiveLock.h"
 #include "librbd/ImageCtx.h"
@@ -666,7 +667,6 @@ int Image<I>::deep_copy(I *src, librados::IoCtx& dest_md_ctx,
 template <typename I>
 int Image<I>::deep_copy(I *src, I *dest, bool flatten,
                         ProgressContext &prog_ctx) {
-  CephContext *cct = src->cct;
   librados::snap_t snap_id_start = 0;
   librados::snap_t snap_id_end;
   {
@@ -674,15 +674,14 @@ int Image<I>::deep_copy(I *src, I *dest, bool flatten,
     snap_id_end = src->snap_id;
   }
 
-  asio::ContextWQ *op_work_queue;
-  ImageCtx::get_work_queue(cct, &op_work_queue);
+  AsioEngine asio_engine(src->md_ctx);
 
   C_SaferCond cond;
   SnapSeqs snap_seqs;
   deep_copy::ProgressHandler progress_handler{&prog_ctx};
   auto req = DeepCopyRequest<I>::create(
     src, dest, snap_id_start, snap_id_end, 0U, flatten, boost::none,
-    op_work_queue, &snap_seqs, &progress_handler, &cond);
+    asio_engine.get_work_queue(), &snap_seqs, &progress_handler, &cond);
   req->send();
   int r = cond.wait();
   if (r < 0) {
@@ -824,15 +823,15 @@ int Image<I>::remove(IoCtx& io_ctx, const std::string &image_name,
     // fall-through if trash isn't supported
   }
 
-  asio::ContextWQ *op_work_queue;
-  ImageCtx::get_work_queue(cct, &op_work_queue);
+  AsioEngine asio_engine(io_ctx);
 
   // might be a V1 image format that cannot be moved to the trash
   // and would not have been listed in the V2 directory -- or the OSDs
   // are too old and don't support the trash feature
   C_SaferCond cond;
   auto req = librbd::image::RemoveRequest<I>::create(
-    io_ctx, image_name, "", false, false, prog_ctx, op_work_queue, &cond);
+    io_ctx, image_name, "", false, false, prog_ctx,
+    asio_engine.get_work_queue(), &cond);
   req->send();
 
   return cond.wait();
diff --git a/src/librbd/api/Migration.cc b/src/librbd/api/Migration.cc
index 3d9d86020f1..85733c4cab1 100644
--- a/src/librbd/api/Migration.cc
+++ b/src/librbd/api/Migration.cc
@@ -8,6 +8,7 @@
 #include "common/errno.h"
 #include "common/Cond.h"
 #include "cls/rbd/cls_rbd_client.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/ExclusiveLock.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
@@ -835,12 +836,12 @@ int Migration<I>::abort() {
 
     ceph_assert(dst_image_ctx->ignore_migrating);
 
-    asio::ContextWQ *op_work_queue;
-    ImageCtx::get_work_queue(m_cct, &op_work_queue);
+    auto asio_engine = dst_image_ctx->asio_engine;
+
     C_SaferCond on_remove;
     auto req = librbd::image::RemoveRequest<>::create(
-      m_dst_io_ctx, dst_image_ctx, false, false, *m_prog_ctx, op_work_queue,
-      &on_remove);
+      m_dst_io_ctx, dst_image_ctx, false, false, *m_prog_ctx,
+      asio_engine->get_work_queue(), &on_remove);
     req->send();
     r = on_remove.wait();
 
@@ -1218,9 +1219,6 @@ int Migration<I>::create_dst_image() {
     }
   }
 
-  asio::ContextWQ *op_work_queue;
-  ImageCtx::get_work_queue(m_cct, &op_work_queue);
-
   ConfigProxy config{m_cct->_conf};
   api::Config<I>::apply_pool_overrides(m_dst_io_ctx, &config);
 
@@ -1240,7 +1238,8 @@ int Migration<I>::create_dst_image() {
     auto *req = image::CreateRequest<I>::create(
       config, m_dst_io_ctx, m_dst_image_name, m_dst_image_id, size,
       m_image_options, image::CREATE_FLAG_SKIP_MIRROR_ENABLE,
-      cls::rbd::MIRROR_IMAGE_MODE_JOURNAL, "", "", op_work_queue, &on_create);
+      cls::rbd::MIRROR_IMAGE_MODE_JOURNAL, "", "",
+      m_src_image_ctx->op_work_queue, &on_create);
     req->send();
   } else {
     r = util::create_ioctx(m_src_image_ctx->md_ctx, "destination image",
@@ -1253,7 +1252,8 @@ int Migration<I>::create_dst_image() {
     auto *req = image::CloneRequest<I>::create(
       config, parent_io_ctx, parent_spec.image_id, "", {}, parent_spec.snap_id,
       m_dst_io_ctx, m_dst_image_name, m_dst_image_id, m_image_options,
-      cls::rbd::MIRROR_IMAGE_MODE_JOURNAL, "", "", op_work_queue, &on_create);
+      cls::rbd::MIRROR_IMAGE_MODE_JOURNAL, "", "",
+      m_src_image_ctx->op_work_queue, &on_create);
     req->send();
   }
 
@@ -1759,12 +1759,12 @@ int Migration<I>::remove_src_image() {
 
   ceph_assert(m_src_image_ctx->ignore_migrating);
 
-  asio::ContextWQ *op_work_queue;
-  ImageCtx::get_work_queue(m_cct, &op_work_queue);
+  auto asio_engine = m_src_image_ctx->asio_engine;
+
   C_SaferCond on_remove;
   auto req = librbd::image::RemoveRequest<I>::create(
-      m_src_io_ctx, m_src_image_ctx, false, true, *m_prog_ctx, op_work_queue,
-      &on_remove);
+      m_src_io_ctx, m_src_image_ctx, false, true, *m_prog_ctx,
+      asio_engine->get_work_queue(), &on_remove);
   req->send();
   r = on_remove.wait();
 
diff --git a/src/librbd/api/Mirror.cc b/src/librbd/api/Mirror.cc
index adf8da97c56..1e34f34374c 100644
--- a/src/librbd/api/Mirror.cc
+++ b/src/librbd/api/Mirror.cc
@@ -8,6 +8,7 @@
 #include "common/dout.h"
 #include "common/errno.h"
 #include "cls/rbd/cls_rbd_client.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
 #include "librbd/Journal.h"
@@ -1940,8 +1941,7 @@ int Mirror<I>::image_info_list(
       break;
     }
 
-    asio::ContextWQ *op_work_queue;
-    ImageCtx::get_work_queue(cct, &op_work_queue);
+    AsioEngine asio_engine(io_ctx);
 
     for (auto &it : images) {
       auto &image_id = it.first;
@@ -1956,7 +1956,7 @@ int Mirror<I>::image_info_list(
       // need to call get_info for every image to retrieve promotion state
 
       mirror_image_info_t info;
-      r = image_get_info(io_ctx, op_work_queue, image_id, &info);
+      r = image_get_info(io_ctx, asio_engine.get_work_queue(), image_id, &info);
       if (r >= 0) {
         (*entries)[image_id] = std::make_pair(mode, info);
       }
diff --git a/src/librbd/api/Pool.cc b/src/librbd/api/Pool.cc
index 75f2dd3b529..890341e015c 100644
--- a/src/librbd/api/Pool.cc
+++ b/src/librbd/api/Pool.cc
@@ -9,6 +9,7 @@
 #include "common/Throttle.h"
 #include "cls/rbd/cls_rbd_client.h"
 #include "osd/osd_types.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/Utils.h"
 #include "librbd/api/Config.h"
@@ -251,11 +252,11 @@ int Pool<I>::init(librados::IoCtx& io_ctx, bool force) {
     return 0;
   }
 
-  asio::ContextWQ *op_work_queue;
-  ImageCtx::get_work_queue(cct, &op_work_queue);
+  AsioEngine asio_engine(io_ctx);
 
   C_SaferCond ctx;
-  auto req = image::ValidatePoolRequest<I>::create(io_ctx, op_work_queue, &ctx);
+  auto req = image::ValidatePoolRequest<I>::create(
+    io_ctx, asio_engine.get_work_queue(), &ctx);
   req->send();
 
   return ctx.wait();
diff --git a/src/librbd/api/Trash.cc b/src/librbd/api/Trash.cc
index 38a270fb8be..eb847e3900d 100644
--- a/src/librbd/api/Trash.cc
+++ b/src/librbd/api/Trash.cc
@@ -7,6 +7,7 @@
 #include "common/errno.h"
 #include "common/Cond.h"
 #include "cls/rbd/cls_rbd_client.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/ExclusiveLock.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
@@ -90,12 +91,12 @@ int enable_mirroring(IoCtx &io_ctx, const std::string &image_id) {
 
   ldout(cct, 10) << dendl;
 
-  asio::ContextWQ *op_work_queue;
-  ImageCtx::get_work_queue(cct, &op_work_queue);
+  AsioEngine asio_engine(io_ctx);
+
   C_SaferCond ctx;
   auto req = mirror::EnableRequest<I>::create(
     io_ctx, image_id, cls::rbd::MIRROR_IMAGE_MODE_JOURNAL, "", false,
-    op_work_queue, &ctx);
+    asio_engine.get_work_queue(), &ctx);
   req->send();
   r = ctx.wait();
   if (r < 0) {
@@ -534,12 +535,11 @@ int Trash<I>::remove(IoCtx &io_ctx, const std::string &image_id, bool force,
     return -EBUSY;
   }
 
-  asio::ContextWQ *op_work_queue;
-  ImageCtx::get_work_queue(cct, &op_work_queue);
+  AsioEngine asio_engine(io_ctx);
 
   C_SaferCond cond;
   auto req = librbd::trash::RemoveRequest<I>::create(
-      io_ctx, image_id, op_work_queue, force, prog_ctx, &cond);
+      io_ctx, image_id, asio_engine.get_work_queue(), force, prog_ctx, &cond);
   req->send();
 
   r = cond.wait();
diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index e3b9bc9ba90..d98e59fc84c 100644
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -23,6 +23,7 @@
 #include "cls/journal/cls_journal_types.h"
 #include "cls/journal/cls_journal_client.h"
 
+#include "librbd/AsioEngine.h"
 #include "librbd/ExclusiveLock.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
@@ -684,8 +685,7 @@ int validate_pool(IoCtx &io_ctx, CephContext *cct) {
       lderr(cct) << "Forced V1 image creation. " << dendl;
       r = create_v1(io_ctx, image_name.c_str(), size, order);
     } else {
-      asio::ContextWQ *op_work_queue;
-      ImageCtx::get_work_queue(cct, &op_work_queue);
+      AsioEngine asio_engine(io_ctx);
 
       ConfigProxy config{cct->_conf};
       api::Config<>::apply_pool_overrides(io_ctx, &config);
@@ -703,7 +703,8 @@ int validate_pool(IoCtx &io_ctx, CephContext *cct) {
       image::CreateRequest<> *req = image::CreateRequest<>::create(
         config, io_ctx, image_name, id, size, opts, create_flags,
         static_cast<cls::rbd::MirrorImageMode>(mirror_image_mode),
-        non_primary_global_image_id, primary_mirror_uuid, op_work_queue, &cond);
+        non_primary_global_image_id, primary_mirror_uuid,
+        asio_engine.get_work_queue(), &cond);
       req->send();
 
       r = cond.wait();
@@ -789,15 +790,15 @@ int validate_pool(IoCtx &io_ctx, CephContext *cct) {
     ConfigProxy config{reinterpret_cast<CephContext *>(c_ioctx.cct())->_conf};
     api::Config<>::apply_pool_overrides(c_ioctx, &config);
 
-    asio::ContextWQ *op_work_queue;
-    ImageCtx::get_work_queue(cct, &op_work_queue);
+    AsioEngine asio_engine(p_ioctx);
 
     C_SaferCond cond;
     auto *req = image::CloneRequest<>::create(
       config, p_ioctx, parent_id, p_snap_name,
       {cls::rbd::UserSnapshotNamespace{}}, CEPH_NOSNAP, c_ioctx, c_name,
       clone_id, c_opts, cls::rbd::MIRROR_IMAGE_MODE_JOURNAL,
-      non_primary_global_image_id, primary_mirror_uuid, op_work_queue, &cond);
+      non_primary_global_image_id, primary_mirror_uuid,
+      asio_engine.get_work_queue(), &cond);
     req->send();
 
     r = cond.wait();
diff --git a/src/test/librbd/CMakeLists.txt b/src/test/librbd/CMakeLists.txt
index e57842ef982..8189b46d755 100644
--- a/src/test/librbd/CMakeLists.txt
+++ b/src/test/librbd/CMakeLists.txt
@@ -126,22 +126,23 @@ add_executable(unittest_librbd
   ${unittest_librbd_srcs}
   $<TARGET_OBJECTS:common_texttable_obj>)
 target_compile_definitions(unittest_librbd PRIVATE "TEST_LIBRBD_INTERNALS")
-target_link_libraries(unittest_librbd
-  cls_rbd
-  cls_rbd_client
+add_dependencies(unittest_librbd
+  cls_journal
   cls_lock
-  cls_lock_client
+  cls_rbd)
+target_link_libraries(unittest_librbd
+  rbd_test
+  rbd_api
+  rbd_internal
+  rbd_test_mock
   journal
   journal_test_mock
-  cls_journal
+  cls_rbd_client
+  cls_lock_client
   cls_journal_client
+  rbd_types
   rados_test_stub
   librados
-  rbd_test
-  rbd_test_mock
-  rbd_api
-  rbd_internal
-  rbd_types
   ceph_immutable_object_cache_lib
   osdc
   ceph-common
@@ -159,6 +160,7 @@ target_link_libraries(ceph_test_librbd
   journal
   cls_journal_client
   cls_rbd_client
+  libneorados
   librados
   ${UNITTEST_LIBS}
   radostest)
diff --git a/src/test/librbd/deep_copy/test_mock_ImageCopyRequest.cc b/src/test/librbd/deep_copy/test_mock_ImageCopyRequest.cc
index 1b2172a51cc..f20867cd4ba 100644
--- a/src/test/librbd/deep_copy/test_mock_ImageCopyRequest.cc
+++ b/src/test/librbd/deep_copy/test_mock_ImageCopyRequest.cc
@@ -3,6 +3,7 @@
 
 #include "test/librbd/test_mock_fixture.h"
 #include "include/rbd/librbd.hpp"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
 #include "librbd/internal.h"
@@ -174,7 +175,10 @@ public:
 
   librbd::ImageCtx *m_src_image_ctx;
   librbd::ImageCtx *m_dst_image_ctx;
+
+  std::shared_ptr<librbd::AsioEngine> m_asio_engine;
   asio::ContextWQ *m_work_queue;
+
   librbd::SnapSeqs m_snap_seqs;
   SnapMap m_snap_map;
 
@@ -188,7 +192,9 @@ public:
     ASSERT_EQ(0, create_image_pp(rbd, m_ioctx, dst_image_name, m_image_size));
     ASSERT_EQ(0, open_image(dst_image_name, &m_dst_image_ctx));
 
-    librbd::ImageCtx::get_work_queue(m_src_image_ctx->cct, &m_work_queue);
+    m_asio_engine = std::make_shared<librbd::AsioEngine>(
+      m_src_image_ctx->md_ctx);
+    m_work_queue = m_asio_engine->get_work_queue();
   }
 
   void expect_get_image_size(librbd::MockTestImageCtx &mock_image_ctx,
diff --git a/src/test/librbd/deep_copy/test_mock_MetadataCopyRequest.cc b/src/test/librbd/deep_copy/test_mock_MetadataCopyRequest.cc
index d18e79f2c4d..ba59e3cdb7c 100644
--- a/src/test/librbd/deep_copy/test_mock_MetadataCopyRequest.cc
+++ b/src/test/librbd/deep_copy/test_mock_MetadataCopyRequest.cc
@@ -4,6 +4,7 @@
 #include "test/librbd/test_mock_fixture.h"
 #include "include/rbd/librbd.hpp"
 #include "include/stringify.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/deep_copy/MetadataCopyRequest.h"
 #include "librbd/image/GetMetadataRequest.h"
@@ -79,6 +80,8 @@ public:
 
   librbd::ImageCtx *m_src_image_ctx;
   librbd::ImageCtx *m_dst_image_ctx;
+
+  std::shared_ptr<librbd::AsioEngine> m_asio_engine;
   asio::ContextWQ *m_work_queue;
 
   void SetUp() override {
@@ -91,7 +94,9 @@ public:
     ASSERT_EQ(0, create_image_pp(rbd, m_ioctx, dst_image_name, m_image_size));
     ASSERT_EQ(0, open_image(dst_image_name, &m_dst_image_ctx));
 
-    librbd::ImageCtx::get_work_queue(m_src_image_ctx->cct, &m_work_queue);
+    m_asio_engine = std::make_shared<librbd::AsioEngine>(
+      m_src_image_ctx->md_ctx);
+    m_work_queue = m_asio_engine->get_work_queue();
   }
 
   void expect_get_metadata(MockGetMetadataRequest& mock_request,
diff --git a/src/test/librbd/deep_copy/test_mock_ObjectCopyRequest.cc b/src/test/librbd/deep_copy/test_mock_ObjectCopyRequest.cc
index 085b838917e..3c6db9057d2 100644
--- a/src/test/librbd/deep_copy/test_mock_ObjectCopyRequest.cc
+++ b/src/test/librbd/deep_copy/test_mock_ObjectCopyRequest.cc
@@ -5,6 +5,7 @@
 #include "include/interval_set.h"
 #include "include/rbd/librbd.hpp"
 #include "include/rbd/object_map_types.h"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
 #include "librbd/internal.h"
@@ -112,6 +113,8 @@ public:
 
   librbd::ImageCtx *m_src_image_ctx;
   librbd::ImageCtx *m_dst_image_ctx;
+
+  std::shared_ptr<librbd::AsioEngine> m_asio_engine;
   asio::ContextWQ *m_work_queue;
 
   SnapMap m_snap_map;
@@ -132,7 +135,9 @@ public:
     ASSERT_EQ(0, create_image_pp(rbd, m_ioctx, dst_image_name, m_image_size));
     ASSERT_EQ(0, open_image(dst_image_name, &m_dst_image_ctx));
 
-    librbd::ImageCtx::get_work_queue(m_src_image_ctx->cct, &m_work_queue);
+    m_asio_engine = std::make_shared<librbd::AsioEngine>(
+      m_src_image_ctx->md_ctx);
+    m_work_queue = m_asio_engine->get_work_queue();
   }
 
   bool is_fast_diff(librbd::MockImageCtx &mock_image_ctx) {
diff --git a/src/test/librbd/deep_copy/test_mock_SetHeadRequest.cc b/src/test/librbd/deep_copy/test_mock_SetHeadRequest.cc
index d8442d84b01..209339973ed 100644
--- a/src/test/librbd/deep_copy/test_mock_SetHeadRequest.cc
+++ b/src/test/librbd/deep_copy/test_mock_SetHeadRequest.cc
@@ -4,6 +4,7 @@
 #include "test/librbd/test_mock_fixture.h"
 #include "test/librados_test_stub/LibradosTestStub.h"
 #include "include/rbd/librbd.hpp"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
 #include "osdc/Striper.h"
@@ -97,6 +98,8 @@ public:
   typedef image::DetachParentRequest<MockTestImageCtx> MockDetachParentRequest;
 
   librbd::ImageCtx *m_image_ctx;
+
+  std::shared_ptr<librbd::AsioEngine> m_asio_engine;
   asio::ContextWQ *m_work_queue;
 
   void SetUp() override {
@@ -104,7 +107,9 @@ public:
 
     ASSERT_EQ(0, open_image(m_image_name, &m_image_ctx));
 
-    librbd::ImageCtx::get_work_queue(m_image_ctx->cct, &m_work_queue);
+    m_asio_engine = std::make_shared<librbd::AsioEngine>(
+      m_image_ctx->md_ctx);
+    m_work_queue = m_asio_engine->get_work_queue();
   }
 
   void expect_start_op(librbd::MockExclusiveLock &mock_exclusive_lock) {
diff --git a/src/test/librbd/deep_copy/test_mock_SnapshotCopyRequest.cc b/src/test/librbd/deep_copy/test_mock_SnapshotCopyRequest.cc
index f2e10cfde40..6a52898220a 100644
--- a/src/test/librbd/deep_copy/test_mock_SnapshotCopyRequest.cc
+++ b/src/test/librbd/deep_copy/test_mock_SnapshotCopyRequest.cc
@@ -3,6 +3,7 @@
 
 #include "test/librbd/test_mock_fixture.h"
 #include "include/rbd/librbd.hpp"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
 #include "librbd/Operations.h"
@@ -106,6 +107,8 @@ public:
 
   librbd::ImageCtx *m_src_image_ctx;
   librbd::ImageCtx *m_dst_image_ctx;
+
+  std::shared_ptr<librbd::AsioEngine> m_asio_engine;
   asio::ContextWQ *m_work_queue;
 
   librbd::SnapSeqs m_snap_seqs;
@@ -120,7 +123,9 @@ public:
     ASSERT_EQ(0, create_image_pp(rbd, m_ioctx, dst_image_name, m_image_size));
     ASSERT_EQ(0, open_image(dst_image_name, &m_dst_image_ctx));
 
-    librbd::ImageCtx::get_work_queue(m_src_image_ctx->cct, &m_work_queue);
+    m_asio_engine = std::make_shared<librbd::AsioEngine>(
+      m_src_image_ctx->md_ctx);
+    m_work_queue = m_asio_engine->get_work_queue();
   }
 
   void prepare_exclusive_lock(librbd::MockImageCtx &mock_image_ctx,
diff --git a/src/test/librbd/deep_copy/test_mock_SnapshotCreateRequest.cc b/src/test/librbd/deep_copy/test_mock_SnapshotCreateRequest.cc
index 01b55af04fd..af24f5a0cf2 100644
--- a/src/test/librbd/deep_copy/test_mock_SnapshotCreateRequest.cc
+++ b/src/test/librbd/deep_copy/test_mock_SnapshotCreateRequest.cc
@@ -4,6 +4,7 @@
 #include "test/librbd/test_mock_fixture.h"
 #include "test/librados_test_stub/LibradosTestStub.h"
 #include "include/rbd/librbd.hpp"
+#include "librbd/AsioEngine.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/ImageState.h"
 #include "librbd/Operations.h"
@@ -76,6 +77,8 @@ public:
   typedef SnapshotCreateRequest<librbd::MockTestImageCtx> MockSnapshotCreateRequest;
 
   librbd::ImageCtx *m_image_ctx;
+
+  std::shared_ptr<librbd::AsioEngine> m_asio_engine;
   asio::ContextWQ *m_work_queue;
 
   void SetUp() override {
@@ -83,7 +86,9 @@ public:
 
     ASSERT_EQ(0, open_image(m_image_name, &m_image_ctx));
 
-    librbd::ImageCtx::get_work_queue(m_image_ctx->cct, &m_work_queue);
+    m_asio_engine = std::make_shared<librbd::AsioEngine>(
+      m_image_ctx->md_ctx);
+    m_work_queue = m_asio_engine->get_work_queue();
   }
 
   void expect_start_op(librbd::MockExclusiveLock &mock_exclusive_lock) {
diff --git a/src/test/librbd/mock/MockImageCtx.h b/src/test/librbd/mock/MockImageCtx.h
index 738143153e6..ab8b0696a1f 100644
--- a/src/test/librbd/mock/MockImageCtx.h
+++ b/src/test/librbd/mock/MockImageCtx.h
@@ -62,6 +62,7 @@ struct MockImageCtx {
       lockers(image_ctx.lockers),
       exclusive_locked(image_ctx.exclusive_locked),
       lock_tag(image_ctx.lock_tag),
+      asio_engine(image_ctx.asio_engine),
       owner_lock(image_ctx.owner_lock),
       image_lock(image_ctx.image_lock),
       timestamp_lock(image_ctx.timestamp_lock),
@@ -249,6 +250,8 @@ struct MockImageCtx {
   bool exclusive_locked;
   std::string lock_tag;
 
+  std::shared_ptr<AsioEngine> asio_engine;
+
   librados::IoCtx md_ctx;
   librados::IoCtx data_ctx;
 
diff --git a/src/test/librbd/test_mock_DeepCopyRequest.cc b/src/test/librbd/test_mock_DeepCopyRequest.cc
index 93f37909fee..de6ceb64d5f 100644
--- a/src/test/librbd/test_mock_DeepCopyRequest.cc
+++ b/src/test/librbd/test_mock_DeepCopyRequest.cc
@@ -3,6 +3,7 @@
 
 #include "test/librbd/test_mock_fixture.h"
 #include "include/rbd/librbd.hpp"
+#include "librbd/AsioEngine.h"
 #include "librbd/DeepCopyRequest.h"
 #include "librbd/ImageState.h"
 #include "librbd/Operations.h"
@@ -150,6 +151,8 @@ public:
 
   librbd::ImageCtx *m_src_image_ctx;
   librbd::ImageCtx *m_dst_image_ctx;
+
+  std::shared_ptr<librbd::AsioEngine> m_asio_engine;
   librbd::asio::ContextWQ *m_work_queue;
 
   void SetUp() override {
@@ -160,7 +163,9 @@ public:
 
     ASSERT_EQ(0, open_image(m_image_name, &m_dst_image_ctx));
 
-    librbd::ImageCtx::get_work_queue(m_src_image_ctx->cct, &m_work_queue);
+    m_asio_engine = std::make_shared<librbd::AsioEngine>(
+      m_src_image_ctx->md_ctx);
+    m_work_queue = m_asio_engine->get_work_queue();
   }
 
   void TearDown() override {
diff --git a/src/test/rbd_mirror/CMakeLists.txt b/src/test/rbd_mirror/CMakeLists.txt
index 36a1603d9c5..1226735d546 100644
--- a/src/test/rbd_mirror/CMakeLists.txt
+++ b/src/test/rbd_mirror/CMakeLists.txt
@@ -57,7 +57,6 @@ add_dependencies(unittest_rbd_mirror
   cls_rbd)
 target_link_libraries(unittest_rbd_mirror
   rbd_mirror_test
-  rados_test_stub
   rbd_mirror_internal
   rbd_mirror_types
   rbd_api
@@ -69,6 +68,7 @@ target_link_libraries(unittest_rbd_mirror
   cls_lock_client
   cls_journal_client
   rbd_types
+  rados_test_stub
   librados
   osdc
   global
@@ -89,6 +89,7 @@ target_link_libraries(ceph_test_rbd_mirror
   cls_rbd_client
   cls_journal_client
   rbd_types
+  libneorados
   librados
   radostest-cxx
   ${UNITTEST_LIBS}
diff --git a/src/test/rbd_mirror/test_ImageReplayer.cc b/src/test/rbd_mirror/test_ImageReplayer.cc
index ce94e5a6d14..540a5a94fb7 100644
--- a/src/test/rbd_mirror/test_ImageReplayer.cc
+++ b/src/test/rbd_mirror/test_ImageReplayer.cc
@@ -170,7 +170,7 @@ public:
     m_global_image_id = get_global_image_id(m_remote_ioctx, m_remote_image_id);
 
     auto cct = reinterpret_cast<CephContext*>(m_local_ioctx.cct());
-    m_threads.reset(new Threads<>(cct));
+    m_threads.reset(new Threads<>(m_local_cluster));
 
     m_image_sync_throttler.reset(new Throttler<>(
         cct, "rbd_mirror_concurrent_image_syncs"));
diff --git a/src/test/rbd_mirror/test_fixture.cc b/src/test/rbd_mirror/test_fixture.cc
index e271364033d..23191da0398 100644
--- a/src/test/rbd_mirror/test_fixture.cc
+++ b/src/test/rbd_mirror/test_fixture.cc
@@ -72,8 +72,7 @@ void TestFixture::SetUp() {
   ASSERT_EQ(0, _rados->ioctx_create(_remote_pool_name.c_str(), m_remote_io_ctx));
   m_image_name = get_temp_image_name();
 
-  m_threads = new rbd::mirror::Threads<>(reinterpret_cast<CephContext*>(
-    m_local_io_ctx.cct()));
+  m_threads = new rbd::mirror::Threads<>(_rados);
 }
 
 void TestFixture::TearDown() {
diff --git a/src/tools/rbd_mirror/CMakeLists.txt b/src/tools/rbd_mirror/CMakeLists.txt
index 9c1a71c57d1..5a89b6c3c9a 100644
--- a/src/tools/rbd_mirror/CMakeLists.txt
+++ b/src/tools/rbd_mirror/CMakeLists.txt
@@ -77,6 +77,7 @@ target_link_libraries(rbd-mirror
   rbd_internal
   rbd_types
   journal
+  libneorados
   librados
   osdc
   cls_rbd_client
diff --git a/src/tools/rbd_mirror/Mirror.cc b/src/tools/rbd_mirror/Mirror.cc
index e24a78ae805..590336b12a4 100644
--- a/src/tools/rbd_mirror/Mirror.cc
+++ b/src/tools/rbd_mirror/Mirror.cc
@@ -487,12 +487,7 @@ Mirror::Mirror(CephContext *cct, const std::vector<const char*> &args) :
   m_local(new librados::Rados()),
   m_cache_manager_handler(new CacheManagerHandler(cct)),
   m_pool_meta_cache(new PoolMetaCache(cct)),
-  m_asok_hook(new MirrorAdminSocketHook(cct, this))
-{
-  m_threads =
-    &(cct->lookup_or_create_singleton_object<Threads<librbd::ImageCtx>>(
-	"rbd_mirror::threads", false, cct));
-  m_service_daemon.reset(new ServiceDaemon<>(m_cct, m_local, m_threads));
+  m_asok_hook(new MirrorAdminSocketHook(cct, this)) {
 }
 
 Mirror::~Mirror()
@@ -539,6 +534,10 @@ int Mirror::init()
     return r;
   }
 
+  m_threads = &(m_cct->lookup_or_create_singleton_object<
+    Threads<librbd::ImageCtx>>("rbd_mirror::threads", false, m_local));
+  m_service_daemon.reset(new ServiceDaemon<>(m_cct, m_local, m_threads));
+
   r = m_service_daemon->init();
   if (r < 0) {
     derr << "error registering service daemon: " << cpp_strerror(r) << dendl;
diff --git a/src/tools/rbd_mirror/Threads.cc b/src/tools/rbd_mirror/Threads.cc
index 702f26a0852..b0c76264119 100644
--- a/src/tools/rbd_mirror/Threads.cc
+++ b/src/tools/rbd_mirror/Threads.cc
@@ -11,8 +11,9 @@ namespace rbd {
 namespace mirror {
 
 template <typename I>
-Threads<I>::Threads(CephContext *cct) {
-  asio_engine = new librbd::AsioEngine(cct);
+Threads<I>::Threads(std::shared_ptr<librados::Rados>& rados) {
+  auto cct = static_cast<CephContext*>(rados->cct());
+  asio_engine = new librbd::AsioEngine(rados);
   work_queue = asio_engine->get_work_queue();
 
   timer = new SafeTimer(cct, timer_lock, true);
diff --git a/src/tools/rbd_mirror/Threads.h b/src/tools/rbd_mirror/Threads.h
index 91c086b1e31..6cbac09bc82 100644
--- a/src/tools/rbd_mirror/Threads.h
+++ b/src/tools/rbd_mirror/Threads.h
@@ -5,7 +5,9 @@
 #define CEPH_RBD_MIRROR_THREADS_H
 
 #include "include/common_fwd.h"
+#include "include/rados/librados_fwd.hpp"
 #include "common/ceph_mutex.h"
+#include <memory>
 
 class SafeTimer;
 class ThreadPool;
@@ -30,7 +32,7 @@ public:
   SafeTimer *timer = nullptr;
   ceph::mutex timer_lock = ceph::make_mutex("Threads::timer_lock");
 
-  explicit Threads(CephContext *cct);
+  explicit Threads(std::shared_ptr<librados::Rados>& rados);
   Threads(const Threads&) = delete;
   Threads& operator=(const Threads&) = delete;