From: Adam C. Emerson <aemerson@redhat.com>
Date: Tue, 7 Aug 2018 20:40:57 +0000 (-0400)
Subject: monc: Asifoact MonClient
X-Git-Tag: v15.1.0~843^2~8
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=73f5249f3f14886831a9da5fb7731b0c227b8d69;p=ceph.git

monc: Asifoact MonClient

Of course now everyone has to feed an io_context into the MonClient.

Signed-off-by: Adam C. Emerson <aemerson@redhat.com>
---

diff --git a/src/ceph_fuse.cc b/src/ceph_fuse.cc
index 5ac18116f12c..4f3cb7837532 100644
--- a/src/ceph_fuse.cc
+++ b/src/ceph_fuse.cc
@@ -16,7 +16,9 @@
 #include <sys/utsname.h>
 #include <iostream>
 #include <string>
+#include <optional>
 
+#include "common/async/context_pool.h"
 #include "common/config.h"
 #include "common/errno.h"
 
@@ -44,6 +46,8 @@
 
 #define dout_context g_ceph_context
 
+ceph::async::io_context_pool icp;
+
 static void fuse_usage()
 {
   const char* argv[] = {
@@ -223,7 +227,8 @@ int main(int argc, const char **argv, const char *envp[]) {
     int tester_r = 0;
     void *tester_rp = nullptr;
 
-    MonClient *mc = new MonClient(g_ceph_context);
+    icp.start(cct->_conf.get_val<std::uint64_t>("client_asio_thread_count"));
+    MonClient *mc = new MonClient(g_ceph_context, icp);
     int r = mc->build_initial_monmap();
     if (r == -EINVAL) {
       cerr << "failed to generate initial mon list" << std::endl;
@@ -305,6 +310,7 @@ int main(int argc, const char **argv, const char *envp[]) {
     client->unmount();
     cfuse->finalize();
   out_shutdown:
+    icp.stop();
     client->shutdown();
   out_init_failed:
     unregister_async_signal_handler(SIGHUP, sighup_handler);
diff --git a/src/ceph_mds.cc b/src/ceph_mds.cc
index 38e673755a95..4e6714ac44e4 100644
--- a/src/ceph_mds.cc
+++ b/src/ceph_mds.cc
@@ -20,6 +20,7 @@
 #include <iostream>
 #include <string>
 
+#include "common/async/context_pool.h"
 #include "include/ceph_features.h"
 #include "include/compat.h"
 #include "include/random.h"
@@ -176,7 +177,8 @@ int main(int argc, const char **argv)
   register_async_signal_handler(SIGHUP, sighup_handler);
   
   // get monmap
-  MonClient mc(g_ceph_context);
+  ceph::async::io_context_pool ctxpool(2);
+  MonClient mc(g_ceph_context, ctxpool);
   if (mc.build_initial_monmap() < 0)
     forker.exit(1);
   global_init_chdir(g_ceph_context);
@@ -184,7 +186,7 @@ int main(int argc, const char **argv)
   msgr->start();
 
   // start mds
-  mds = new MDSDaemon(g_conf()->name.get_id().c_str(), msgr, &mc);
+  mds = new MDSDaemon(g_conf()->name.get_id().c_str(), msgr, &mc, ctxpool);
 
   // in case we have to respawn...
   mds->orig_argc = argc;
@@ -215,6 +217,7 @@ int main(int argc, const char **argv)
   shutdown_async_signal_handler();
 
  shutdown:
+  ctxpool.stop();
   // yuck: grab the mds lock, so we can be sure that whoever in *mds
   // called shutdown finishes what they were doing.
   mds->mds_lock.lock();
@@ -238,4 +241,3 @@ int main(int argc, const char **argv)
 
   return 0;
 }
-
diff --git a/src/ceph_osd.cc b/src/ceph_osd.cc
index 3e03e8632396..18101dee1221 100644
--- a/src/ceph_osd.cc
+++ b/src/ceph_osd.cc
@@ -664,7 +664,10 @@ flushjournal_out:
 
   srand(time(NULL) + getpid());
 
-  MonClient mc(g_ceph_context);
+  ceph::async::io_context_pool poolctx(
+    cct->_conf.get_val<std::uint64_t>("osd_asio_thread_count"));
+
+  MonClient mc(g_ceph_context, poolctx);
   if (mc.build_initial_monmap() < 0)
     return -1;
   global_init_chdir(g_ceph_context);
@@ -685,7 +688,8 @@ flushjournal_out:
 		   ms_objecter,
 		   &mc,
 		   data_path,
-		   journal_path);
+		   journal_path,
+		   poolctx);
 
   int err = osdptr->pre_init();
   if (err < 0) {
@@ -740,6 +744,7 @@ flushjournal_out:
   shutdown_async_signal_handler();
 
   // done
+  poolctx.stop();
   delete osdptr;
   delete ms_public;
   delete ms_hb_front_client;
diff --git a/src/ceph_syn.cc b/src/ceph_syn.cc
index 50e26f2815a0..e3ca13287020 100644
--- a/src/ceph_syn.cc
+++ b/src/ceph_syn.cc
@@ -18,6 +18,7 @@
 
 #include "common/config.h"
 
+#include "common/async/context_pool.h"
 #include "client/SyntheticClient.h"
 #include "client/Client.h"
 
@@ -50,7 +51,8 @@ int main(int argc, const char **argv, char *envp[])
   pick_addresses(g_ceph_context, CEPH_PICK_ADDRESS_PUBLIC);
 
   // get monmap
-  MonClient mc(g_ceph_context);
+  ceph::async::io_context_pool  poolctx(1);
+  MonClient mc(g_ceph_context, poolctx);
   if (mc.build_initial_monmap() < 0)
     return -1;
 
@@ -64,7 +66,7 @@ int main(int argc, const char **argv, char *envp[])
     messengers[i] = Messenger::create_client_messenger(g_ceph_context,
 						       "synclient");
     messengers[i]->bind(g_conf()->public_addr);
-    mclients[i] = new MonClient(g_ceph_context);
+    mclients[i] = new MonClient(g_ceph_context, poolctx);
     mclients[i]->build_initial_monmap();
     auto client = new StandaloneClient(messengers[i], mclients[i]);
     client->set_filer_flags(syn_filer_flags);
@@ -79,6 +81,8 @@ int main(int argc, const char **argv, char *envp[])
        ++p)
     (*p)->start_thread();
 
+  poolctx.stop();
+
   //cout << "waiting for client(s) to finish" << std::endl;
   while (!clients.empty()) {
     Client *client = clients.front();
@@ -99,4 +103,3 @@ int main(int argc, const char **argv, char *envp[])
   }
   return 0;
 }
-
diff --git a/src/client/Client.cc b/src/client/Client.cc
index f3f6cacc9eba..ec0fb160b047 100644
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -44,6 +44,7 @@
 
 #include "common/config.h"
 #include "common/version.h"
+#include "common/async/waiter.h"
 
 #include "mon/MonClient.h"
 
@@ -124,6 +125,8 @@
 
 #define DEBUG_GETATTR_CAPS (CEPH_CAP_XATTR_SHARED)
 
+namespace bs = boost::system;
+
 void client_flush_set_callback(void *p, ObjectCacher::ObjectSet *oset)
 {
   Client *client = static_cast<Client*>(p);
@@ -5643,22 +5646,22 @@ int Client::authenticate()
 
 int Client::fetch_fsmap(bool user)
 {
-  int r;
   // Retrieve FSMap to enable looking up daemon addresses.  We need FSMap
   // rather than MDSMap because no one MDSMap contains all the daemons, and
   // a `tell` can address any daemon.
   version_t fsmap_latest;
+  boost::system::error_code ec;
   do {
-    C_SaferCond cond;
-    monclient->get_version("fsmap", &fsmap_latest, NULL, &cond);
+    ceph::async::waiter<bs::error_code, version_t, version_t> w;
+    monclient->get_version("fsmap", w);
     client_lock.unlock();
-    r = cond.wait();
+    std::tie(ec, fsmap_latest, std::ignore) = w.wait();
     client_lock.lock();
-  } while (r == -EAGAIN);
+  } while (ec == boost::system::errc::resource_unavailable_try_again);
 
-  if (r < 0) {
-    lderr(cct) << "Failed to learn FSMap version: " << cpp_strerror(r) << dendl;
-    return r;
+  if (ec) {
+    lderr(cct) << "Failed to learn FSMap version: " << ec << dendl;
+    return ceph::from_error_code(ec);
   }
 
   ldout(cct, 10) << __func__ << " learned FSMap version " << fsmap_latest << dendl;
@@ -14595,7 +14598,7 @@ mds_rank_t Client::_get_random_up_mds() const
 
 
 StandaloneClient::StandaloneClient(Messenger *m, MonClient *mc)
-    : Client(m, mc, new Objecter(m->cct, m, mc, NULL, 0, 0))
+  : Client(m, mc, new Objecter(m->cct, m, mc, nullptr, 0, 0))
 {
   monclient->set_messenger(m);
   objecter->set_client_incarnation(0);
diff --git a/src/common/ceph_context.h b/src/common/ceph_context.h
index 6b6e9482b997..ae590865945c 100644
--- a/src/common/ceph_context.h
+++ b/src/common/ceph_context.h
@@ -25,6 +25,8 @@
 #include <typeinfo>
 #include <typeindex>
 
+#include <boost/intrusive_ptr.hpp>
+
 #include "include/any.h"
 
 #include "common/cmdparse.h"
@@ -363,4 +365,15 @@ private:
 };
 #endif	// WITH_SEASTAR
 
+inline void intrusive_ptr_add_ref(CephContext* cct)
+{
+  cct->get();
+}
+
+inline void intrusive_ptr_release(CephContext* cct)
+{
+  cct->put();
+}
+
+
 #endif
diff --git a/src/common/options.cc b/src/common/options.cc
index 1f76931f8054..19b172830fd5 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -5224,6 +5224,18 @@ std::vector<Option> get_global_options() {
     Option("debug_heartbeat_testing_span", Option::TYPE_INT, Option::LEVEL_DEV)
     .set_default(0)
     .set_description("Override 60 second periods for testing only"),
+
+    Option("librados_thread_count", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
+    .set_default(2)
+    .set_min(1)
+    .set_description("Size of thread pool for Objecter")
+    .add_tag("client"),
+
+    Option("osd_asio_thread_count", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
+    .set_default(2)
+    .set_min(1)
+    .set_description("Size of thread pool for ASIO completions")
+    .add_tag("osd")
   });
 }
 
@@ -7933,6 +7945,12 @@ std::vector<Option> get_mds_options() {
      .set_flag(Option::FLAG_RUNTIME)
      .set_description("max snapshots per directory")
      .set_long_description("maximum number of snapshots that can be created per directory"),
+
+    Option("mds_asio_thread_count", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
+    .set_default(2)
+    .set_min(1)
+    .set_description("Size of thread pool for ASIO completions")
+    .add_tag("mds")
   });
 }
 
@@ -8169,6 +8187,12 @@ std::vector<Option> get_mds_client_options() {
     Option("debug_allow_any_pool_priority", Option::TYPE_BOOL, Option::LEVEL_DEV)
     .set_default(false)
     .set_description("Allow any pool priority to be set to test conversion to new range"),
+
+    Option("client_asio_thread_count", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
+    .set_default(2)
+    .set_min(1)
+    .set_description("Size of thread pool for ASIO completions")
+    .add_tag("client")
   });
 }
 
diff --git a/src/global/global_init.cc b/src/global/global_init.cc
index 6962f42a543e..b56d61f3800b 100644
--- a/src/global/global_init.cc
+++ b/src/global/global_init.cc
@@ -12,6 +12,7 @@
  *
  */
 
+#include "common/async/context_pool.h"
 #include "common/ceph_argparse.h"
 #include "common/code_environment.h"
 #include "common/config.h"
@@ -327,13 +328,16 @@ global_init(const std::map<std::string,std::string> *defaults,
     // make sure our mini-session gets legacy values
     g_conf().apply_changes(nullptr);
 
-    MonClient mc_bootstrap(g_ceph_context);
+    ceph::async::io_context_pool cp(1);
+    MonClient mc_bootstrap(g_ceph_context, cp);
     if (mc_bootstrap.get_monmap_and_config() < 0) {
+      cp.stop();
       g_ceph_context->_log->flush();
       cerr << "failed to fetch mon config (--no-mon-config to skip)"
 	   << std::endl;
       _exit(1);
     }
+    cp.stop();
   }
 
   // Expand metavariables. Invoke configuration observers. Open log file.
@@ -393,16 +397,6 @@ global_init(const std::map<std::string,std::string> *defaults,
   return boost::intrusive_ptr<CephContext>{g_ceph_context, false};
 }
 
-void intrusive_ptr_add_ref(CephContext* cct)
-{
-  cct->get();
-}
-
-void intrusive_ptr_release(CephContext* cct)
-{
-  cct->put();
-}
-
 void global_print_banner(void)
 {
   output_ceph_version();
diff --git a/src/global/global_init.h b/src/global/global_init.h
index fe6f557bb253..f4c9a4a8c826 100644
--- a/src/global/global_init.h
+++ b/src/global/global_init.h
@@ -20,11 +20,10 @@
 #include <map>
 #include <boost/intrusive_ptr.hpp>
 #include "include/ceph_assert.h"
+#include "common/ceph_context.h"
 #include "common/code_environment.h"
 #include "common/common_init.h"
 
-class CephContext;
-
 /*
  * global_init is the first initialization function that
  * daemons and utility programs need to call. It takes care of a lot of
@@ -40,9 +39,6 @@ global_init(
   const char *data_dir_option = 0,
   bool run_pre_init = true);
 
-void intrusive_ptr_add_ref(CephContext* cct);
-void intrusive_ptr_release(CephContext* cct);
-
 // just the first half; enough to get config parsed but doesn't start up the
 // cct or log.
 void global_pre_init(const std::map<std::string,std::string> *defaults,
diff --git a/src/libcephfs.cc b/src/libcephfs.cc
index 8a6b14d1ed8d..b98fa82c6b35 100644
--- a/src/libcephfs.cc
+++ b/src/libcephfs.cc
@@ -20,6 +20,7 @@
 #include "auth/Crypto.h"
 #include "client/Client.h"
 #include "librados/RadosClient.h"
+#include "common/async/context_pool.h"
 #include "common/ceph_argparse.h"
 #include "common/common_init.h"
 #include "common/config.h"
@@ -36,6 +37,7 @@
 #define DEFAULT_UMASK 002
 
 static mode_t umask_cb(void *);
+ceph::async::io_context_pool icp;
 
 struct ceph_mount_info
 {
@@ -83,8 +85,9 @@ public:
       cct->_log->start();
     }
 
+    icp.start(cct->_conf.get_val<std::uint64_t>("client_asio_thread_count"));
     {
-      MonClient mc_bootstrap(cct);
+      MonClient mc_bootstrap(cct, icp);
       ret = mc_bootstrap.get_monmap_and_config();
       if (ret < 0)
 	return ret;
@@ -93,7 +96,7 @@ public:
     common_init_finish(cct);
 
     //monmap
-    monclient = new MonClient(cct);
+    monclient = new MonClient(cct, icp);
     ret = -CEPHFS_ERROR_MON_MAP_BUILD; //defined in libcephfs.h;
     if (monclient->build_initial_monmap() < 0)
       goto fail;
@@ -202,6 +205,7 @@ public:
       delete messenger;
       messenger = nullptr;
     }
+    icp.stop();
     if (monclient) {
       delete monclient;
       monclient = nullptr;
diff --git a/src/librados/RadosClient.cc b/src/librados/RadosClient.cc
index 372419816d8b..4992cc23490d 100644
--- a/src/librados/RadosClient.cc
+++ b/src/librados/RadosClient.cc
@@ -28,6 +28,7 @@
 #include "common/ceph_json.h"
 #include "common/errno.h"
 #include "common/ceph_json.h"
+#include "common/async/waiter.h"
 #include "include/buffer.h"
 #include "include/stringify.h"
 #include "include/util.h"
@@ -56,12 +57,14 @@
 #undef dout_prefix
 #define dout_prefix *_dout << "librados: "
 
+using ceph::async::waiter;
+
 librados::RadosClient::RadosClient(CephContext *cct_)
   : Dispatcher(cct_->get()),
     cct_deleter{cct_, [](CephContext *p) {p->put();}},
     conf(cct_->_conf),
     state(DISCONNECTED),
-    monclient(cct_),
+    monclient(cct_, poolctx),
     mgrclient(cct_, nullptr, &monclient.monmap),
     messenger(NULL),
     instance_id(0),
@@ -233,7 +236,7 @@ int librados::RadosClient::connect()
   }
 
   {
-    MonClient mc_bootstrap(cct);
+    MonClient mc_bootstrap(cct, poolctx);
     err = mc_bootstrap.get_monmap_and_config();
     if (err < 0)
       return err;
@@ -241,6 +244,8 @@ int librados::RadosClient::connect()
 
   common_init_finish(cct);
 
+  poolctx.start(cct->_conf.get_val<std::uint64_t>("librados_thread_count"));
+
   // get monmap
   err = monclient.build_initial_monmap();
   if (err < 0)
@@ -382,6 +387,7 @@ void librados::RadosClient::shutdown()
     messenger->wait();
   }
   ldout(cct, 1) << "shutdown" << dendl;
+  poolctx.finish();
 }
 
 int librados::RadosClient::watch_flush()
@@ -843,7 +849,20 @@ void librados::RadosClient::mon_command_async(const vector<string>& cmd,
                                               Context *on_finish)
 {
   std::lock_guard l{lock};
-  monclient.start_mon_command(cmd, inbl, outbl, outs, on_finish);
+  monclient.start_mon_command(cmd, inbl,
+			      [outs, outbl,
+			       on_finish = std::unique_ptr<Context>(on_finish)]
+			      (boost::system::error_code e,
+			       std::string&& s,
+			       ceph::bufferlist&& b) mutable {
+				if (outs)
+				  *outs = std::move(s);
+				if (outbl)
+				  *outbl = std::move(b);
+				if (on_finish)
+				  on_finish.release()->complete(
+				    ceph::from_error_code(e));
+			      });
 }
 
 int librados::RadosClient::mgr_command(const vector<string>& cmd,
@@ -897,36 +916,32 @@ int librados::RadosClient::mon_command(int rank, const vector<string>& cmd,
 				       const bufferlist &inbl,
 				       bufferlist *outbl, string *outs)
 {
-  ceph::mutex mylock = ceph::make_mutex("RadosClient::mon_command::mylock");
-  ceph::condition_variable cond;
-  bool done;
-  int rval;
-  {
-    std::lock_guard l{mylock};
-    monclient.start_mon_command(rank, cmd, inbl, outbl, outs,
-				new C_SafeCond(mylock, cond, &done, &rval));
-  }
-  std::unique_lock l{mylock};
-  cond.wait(l, [&done] { return done;});
-  return rval;
+  waiter<boost::system::error_code, std::string, ceph::bufferlist> w;
+  monclient.start_mon_command(rank, cmd, inbl, w);
+  auto&& [ec, s, bl] = w.wait();
+
+  if (outs)
+    *outs = std::move(s);
+  if (outbl)
+    *outbl = std::move(bl);
+
+  return ceph::from_error_code(ec);
 }
 
 int librados::RadosClient::mon_command(string name, const vector<string>& cmd,
 				       const bufferlist &inbl,
 				       bufferlist *outbl, string *outs)
 {
-  ceph::mutex mylock = ceph::make_mutex("RadosClient::mon_command::mylock");
-  ceph::condition_variable cond;
-  bool done;
-  int rval;
-  {
-    std::lock_guard l{mylock};
-    monclient.start_mon_command(name, cmd, inbl, outbl, outs,
-				new C_SafeCond(mylock, cond, &done, &rval));
-  }
-  std::unique_lock l{mylock};
-  cond.wait(l, [&done] { return done;});
-  return rval;
+  waiter<boost::system::error_code, std::string, ceph::bufferlist> w;
+  monclient.start_mon_command(name, cmd, inbl, w);
+  auto&& [ec, s, bl] = w.wait();
+
+  if (outs)
+    *outs = std::move(s);
+  if (outbl)
+    *outbl = std::move(bl);
+
+  return ceph::from_error_code(ec);
 }
 
 int librados::RadosClient::osd_command(int osd, vector<string>& cmd,
@@ -1169,3 +1184,24 @@ int librados::RadosClient::get_inconsistent_pgs(int64_t pool_id,
   }
   return 0;
 }
+
+namespace {
+const char *config_keys[] = {
+  "librados_thread_count",
+  NULL
+};
+}
+
+const char** librados::RadosClient::get_tracked_conf_keys() const
+{
+  return config_keys;
+}
+
+void librados::RadosClient::handle_conf_change(const ConfigProxy& conf,
+					       const std::set<std::string> &changed)
+{
+  if (changed.count("librados_thread_count")) {
+    poolctx.stop();
+    poolctx.start(conf.get_val<std::uint64_t>("librados_thread_count"));
+  }
+}
diff --git a/src/librados/RadosClient.h b/src/librados/RadosClient.h
index 0f2b153def4a..9cd3768d47cf 100644
--- a/src/librados/RadosClient.h
+++ b/src/librados/RadosClient.h
@@ -14,36 +14,43 @@
 #ifndef CEPH_LIBRADOS_RADOSCLIENT_H
 #define CEPH_LIBRADOS_RADOSCLIENT_H
 
+#include <functional>
+#include <memory>
+#include <string>
+
+#include "msg/Dispatcher.h"
+
+#include "common/async/context_pool.h"
 #include "common/config_fwd.h"
 #include "common/Cond.h"
 #include "common/Timer.h"
 #include "common/ceph_mutex.h"
 #include "common/ceph_time.h"
+#include "common/config_obs.h"
 #include "include/rados/librados.h"
 #include "include/rados/librados.hpp"
 #include "mon/MonClient.h"
 #include "mgr/MgrClient.h"
-#include "msg/Dispatcher.h"
 
 #include "IoCtxImpl.h"
 
-struct AuthAuthorizer;
 struct Context;
 class CephContext;
-struct Connection;
 class Message;
 class MLog;
 class Messenger;
 class AioCompletionImpl;
 
-class librados::RadosClient : public Dispatcher
+class librados::RadosClient : public Dispatcher,
+			      public md_config_obs_t
 {
   std::unique_ptr<CephContext,
 		  std::function<void(CephContext*)> > cct_deleter;
 
 public:
   using Dispatcher::cct;
-  const ConfigProxy& conf;
+  const ConfigProxy& conf{cct->_conf};
+  ceph::async::io_context_pool poolctx;
 private:
   enum {
     DISCONNECTED,
@@ -89,7 +96,7 @@ public:
 
   explicit RadosClient(CephContext *cct_);
   ~RadosClient() override;
-  int ping_monitor(string mon_id, string *result);
+  int ping_monitor(std::string mon_id, std::string *result);
   int connect();
   void shutdown();
 
@@ -177,6 +184,9 @@ public:
   mon_feature_t get_required_monitor_features() const;
 
   int get_inconsistent_pgs(int64_t pool_id, std::vector<std::string>* pgs);
+  const char** get_tracked_conf_keys() const override;
+  void handle_conf_change(const ConfigProxy& conf,
+                          const std::set <std::string> &changed) override;
 };
 
 #endif
diff --git a/src/libradosstriper/MultiAioCompletionImpl.cc b/src/libradosstriper/MultiAioCompletionImpl.cc
index 901bb1366640..acf9e0b6b8b1 100644
--- a/src/libradosstriper/MultiAioCompletionImpl.cc
+++ b/src/libradosstriper/MultiAioCompletionImpl.cc
@@ -58,13 +58,3 @@ void libradosstriper::MultiAioCompletionImpl::finish_adding_requests()
   if (!pending_safe)
     safe();
 }
-
-void intrusive_ptr_add_ref(libradosstriper::MultiAioCompletionImpl* ptr)
-{
-  ptr->get();
-}
-
-void intrusive_ptr_release(libradosstriper::MultiAioCompletionImpl* ptr)
-{
-  ptr->put();
-}
diff --git a/src/libradosstriper/MultiAioCompletionImpl.h b/src/libradosstriper/MultiAioCompletionImpl.h
index 32f7b9a84cf9..3ac3aae44920 100644
--- a/src/libradosstriper/MultiAioCompletionImpl.h
+++ b/src/libradosstriper/MultiAioCompletionImpl.h
@@ -20,7 +20,9 @@
 #include "common/ceph_mutex.h"
 #include "include/radosstriper/libradosstriper.hpp"
 
-struct libradosstriper::MultiAioCompletionImpl {
+namespace libradosstriper {
+
+struct MultiAioCompletionImpl {
 
   ceph::mutex lock = ceph::make_mutex("MultiAioCompletionImpl lock", false);
   ceph::condition_variable cond;
@@ -151,10 +153,17 @@ struct libradosstriper::MultiAioCompletionImpl {
   void complete_request(ssize_t r);
   void safe_request(ssize_t r);
   void finish_adding_requests();
-
 };
 
-void intrusive_ptr_add_ref(libradosstriper::MultiAioCompletionImpl*);
-void intrusive_ptr_release(libradosstriper::MultiAioCompletionImpl*);
+inline void intrusive_ptr_add_ref(MultiAioCompletionImpl* ptr)
+{
+  ptr->get();
+}
+
+inline void intrusive_ptr_release(MultiAioCompletionImpl* ptr)
+{
+  ptr->put();
+}
+}
 
 #endif // CEPH_LIBRADOSSTRIPERSTRIPER_MULTIAIOCOMPLETIONIMPL_H
diff --git a/src/libradosstriper/RadosStriperImpl.h b/src/libradosstriper/RadosStriperImpl.h
index 160db7b6f8d3..8226a9ba2a23 100644
--- a/src/libradosstriper/RadosStriperImpl.h
+++ b/src/libradosstriper/RadosStriperImpl.h
@@ -17,6 +17,8 @@
 
 #include <string>
 
+#include <boost/intrusive_ptr.hpp>
+
 #include "include/rados/librados.h"
 #include "include/rados/librados.hpp"
 #include "include/radosstriper/libradosstriper.h"
@@ -26,6 +28,7 @@
 #include "librados/IoCtxImpl.h"
 #include "librados/AioCompletionImpl.h"
 #include "common/RefCountedObj.h"
+#include "common/ceph_context.h"
 
 namespace libradosstriper {
 
diff --git a/src/mds/MDSDaemon.cc b/src/mds/MDSDaemon.cc
index 9e96338242b3..3d5894fc5f30 100644
--- a/src/mds/MDSDaemon.cc
+++ b/src/mds/MDSDaemon.cc
@@ -61,7 +61,8 @@
 #define dout_prefix *_dout << "mds." << name << ' '
 
 // cons/des
-MDSDaemon::MDSDaemon(std::string_view n, Messenger *m, MonClient *mc) :
+MDSDaemon::MDSDaemon(std::string_view n, Messenger *m, MonClient *mc,
+		     boost::asio::io_context& ioctx) :
   Dispatcher(m->cct),
   timer(m->cct, mds_lock),
   gss_ktfile_client(m->cct->_conf.get_val<std::string>("gss_ktab_client_file")),
@@ -69,6 +70,7 @@ MDSDaemon::MDSDaemon(std::string_view n, Messenger *m, MonClient *mc) :
   name(n),
   messenger(m),
   monc(mc),
+  ioctx(ioctx),
   mgrc(m->cct, m, &mc->monmap),
   log_client(m->cct, messenger, &mc->monmap, LogClient::NO_FLAGS),
   starttime(mono_clock::now())
@@ -841,10 +843,12 @@ void MDSDaemon::handle_mds_map(const cref_t<MMDSMap> &m)
 
     // Did I previously not hold a rank?  Initialize!
     if (mds_rank == NULL) {
-      mds_rank = new MDSRankDispatcher(whoami, mds_lock, clog,
-          timer, beacon, mdsmap, messenger, monc, &mgrc,
-          new LambdaContext([this](int r){respawn();}),
-          new LambdaContext([this](int r){suicide();}));
+      mds_rank = new MDSRankDispatcher(
+	whoami, mds_lock, clog,
+	timer, beacon, mdsmap, messenger, monc, &mgrc,
+	new LambdaContext([this](int r){respawn();}),
+	new LambdaContext([this](int r){suicide();}),
+	ioctx);
       dout(10) <<  __func__ << ": initializing MDS rank "
                << mds_rank->get_nodeid() << dendl;
       mds_rank->init();
diff --git a/src/mds/MDSDaemon.h b/src/mds/MDSDaemon.h
index e8084b10474d..2e5fba0aa89d 100644
--- a/src/mds/MDSDaemon.h
+++ b/src/mds/MDSDaemon.h
@@ -42,7 +42,9 @@ class MonClient;
 
 class MDSDaemon : public Dispatcher {
  public:
-  MDSDaemon(std::string_view n, Messenger *m, MonClient *mc);
+  MDSDaemon(std::string_view n, Messenger *m, MonClient *mc,
+	    boost::asio::io_context& ioctx);
+
   ~MDSDaemon() override;
 
   mono_time get_starttime() const {
@@ -135,6 +137,7 @@ class MDSDaemon : public Dispatcher {
 
   Messenger    *messenger;
   MonClient    *monc;
+  boost::asio::io_context& ioctx;
   MgrClient     mgrc;
   std::unique_ptr<MDSMap> mdsmap;
   LogClient    log_client;
diff --git a/src/mds/MDSRank.cc b/src/mds/MDSRank.cc
index a5296784295e..f2b2f9192eb8 100644
--- a/src/mds/MDSRank.cc
+++ b/src/mds/MDSRank.cc
@@ -484,8 +484,8 @@ MDSRank::MDSRank(
     MonClient *monc_,
     MgrClient *mgrc,
     Context *respawn_hook_,
-    Context *suicide_hook_)
-  :
+    Context *suicide_hook_,
+    boost::asio::io_context& ioctx) :
     whoami(whoami_), incarnation(0),
     mds_lock(mds_lock_), cct(msgr->cct), clog(clog_), timer(timer_),
     mdsmap(mdsmap_),
@@ -598,6 +598,7 @@ MDSRank::~MDSRank()
 
 void MDSRankDispatcher::init()
 {
+  ctxpool.start(cct->_conf.get_val<std::uint64_t>("mds_asio_thread_count"));
   objecter->init();
   messenger->add_dispatcher_head(objecter);
 
@@ -846,6 +847,7 @@ void MDSRankDispatcher::shutdown()
     objecter->shutdown();
 
   monc->shutdown();
+  ctxpool.finish();
 
   op_tracker.on_shutdown();
 
@@ -3502,9 +3504,10 @@ MDSRankDispatcher::MDSRankDispatcher(
     MonClient *monc_,
     MgrClient *mgrc,
     Context *respawn_hook_,
-    Context *suicide_hook_)
+    Context *suicide_hook_,
+    boost::asio::io_context& ictx)
   : MDSRank(whoami_, mds_lock_, clog_, timer_, beacon_, mdsmap_,
-            msgr, monc_, mgrc, respawn_hook_, suicide_hook_)
+            msgr, monc_, mgrc, respawn_hook_, suicide_hook_, ictx)
 {
     g_conf().add_observer(this);
 }
@@ -3681,6 +3684,7 @@ const char** MDSRankDispatcher::get_tracked_conf_keys() const
     "mds_recall_warning_decay_rate",
     "mds_request_load_average_decay_rate",
     "mds_session_cache_liveness_decay_rate",
+    "mds_asio_thread_count",
     NULL
   };
   return KEYS;
@@ -3722,6 +3726,11 @@ void MDSRankDispatcher::handle_conf_change(const ConfigProxy& conf, const std::s
     mdcache->handle_conf_change(changed, *mdsmap);
     purge_queue.handle_conf_change(changed, *mdsmap);
   }));
+
+  if (changed.count("mds_asio_thread_count")) {
+    ctxpool.stop();
+    ctxpool.start(conf.get_val<std::uint64_t>("mds_asio_thread_count"));
+  }
 }
 
 void MDSRank::get_task_status(std::map<std::string, std::string> *status) {
diff --git a/src/mds/MDSRank.h b/src/mds/MDSRank.h
index 1944e2f17bde..a2dc8928ea6a 100644
--- a/src/mds/MDSRank.h
+++ b/src/mds/MDSRank.h
@@ -17,6 +17,7 @@
 
 #include <string_view>
 
+#include "common/async/context_pool.h"
 #include "common/DecayCounter.h"
 #include "common/LogClient.h"
 #include "common/Timer.h"
@@ -178,6 +179,7 @@ class MDSRank {
 
     std::unique_ptr<MDSMap> &mdsmap; /* MDSDaemon::mdsmap */
 
+    ceph::async::io_context_pool ctxpool;
     Objecter     *objecter;
 
     // sub systems
@@ -342,7 +344,8 @@ class MDSRank {
         MonClient *monc_,
         MgrClient *mgrc,
         Context *respawn_hook_,
-        Context *suicide_hook_);
+        Context *suicide_hook_,
+	boost::asio::io_context& ictx);
 
   protected:
     ~MDSRank();
@@ -665,7 +668,8 @@ public:
       MonClient *monc_,
       MgrClient *mgrc,
       Context *respawn_hook_,
-      Context *suicide_hook_);
+      Context *suicide_hook_,
+      boost::asio::io_context& ictx);
 };
 
 // This utility for MDS and MDSRank dispatchers.
diff --git a/src/mgr/MgrContext.h b/src/mgr/MgrContext.h
index bec51a368718..c6e647365b50 100644
--- a/src/mgr/MgrContext.h
+++ b/src/mgr/MgrContext.h
@@ -17,6 +17,7 @@
 #include <memory>
 
 #include "common/ceph_json.h"
+#include "common/Cond.h"
 #include "mon/MonClient.h"
 
 class Command
diff --git a/src/mgr/MgrStandby.cc b/src/mgr/MgrStandby.cc
index 7d193b6575b9..f751b83ee2c4 100644
--- a/src/mgr/MgrStandby.cc
+++ b/src/mgr/MgrStandby.cc
@@ -38,7 +38,7 @@
 
 MgrStandby::MgrStandby(int argc, const char **argv) :
   Dispatcher(g_ceph_context),
-  monc{g_ceph_context},
+  monc{g_ceph_context, poolctx},
   client_messenger(Messenger::create(
 		     g_ceph_context,
 		     cct->_conf.get_val<std::string>("ms_type"),
@@ -115,6 +115,8 @@ int MgrStandby::init()
   client_messenger->add_dispatcher_tail(&client);
   client_messenger->start();
 
+  poolctx.start(2);
+
   // Initialize MonClient
   if (monc.build_initial_monmap() < 0) {
     client_messenger->shutdown();
@@ -169,6 +171,7 @@ int MgrStandby::init()
   // this method.
   monc.set_passthrough_monmap();
 
+  poolctx.start(1);
   client_t whoami = monc.get_global_id();
   client_messenger->set_myname(entity_name_t::MGR(whoami.v));
   monc.set_log_client(&log_client);
@@ -287,7 +290,20 @@ void MgrStandby::shutdown()
     }
 
     py_module_registry.shutdown();
-
+    // stop sending beacon first, i use monc to talk with monitors
+    timer.shutdown();
+    // client uses monc and objecter
+    client.shutdown();
+    mgrc.shutdown();
+    // Stop asio threads, so leftover events won't call into shut down
+    // monclient/objecter.
+    poolctx.finish();
+    // stop monc, so mon won't be able to instruct me to shutdown/activate after
+    // the active_mgr is stopped
+    monc.shutdown();
+    if (active_mgr) {
+      active_mgr->shutdown();
+    }
     // objecter is used by monc and active_mgr
     objecter.shutdown();
     // client_messenger is used by all of them, so stop it in the end
@@ -298,6 +314,7 @@ void MgrStandby::shutdown()
   // to touch references to the things we're about to tear down
   finisher.wait_for_empty();
   finisher.stop();
+  poolctx.finish();
 }
 
 void MgrStandby::respawn()
diff --git a/src/mgr/MgrStandby.h b/src/mgr/MgrStandby.h
index 98b9d0d22a78..09ce8428b1ab 100644
--- a/src/mgr/MgrStandby.h
+++ b/src/mgr/MgrStandby.h
@@ -16,6 +16,7 @@
 #define MGR_STANDBY_H_
 
 #include "auth/Auth.h"
+#include "common/async/context_pool.h"
 #include "common/Finisher.h"
 #include "common/Timer.h"
 #include "common/LogClient.h"
@@ -39,6 +40,7 @@ public:
 			  const std::set <std::string> &changed) override;
 
 protected:
+  ceph::async::io_context_pool poolctx;
   MonClient monc;
   std::unique_ptr<Messenger> client_messenger;
   Objecter objecter;
diff --git a/src/mon/AuthMonitor.cc b/src/mon/AuthMonitor.cc
index 3cb261b412a0..aee0e33a4eaf 100644
--- a/src/mon/AuthMonitor.cc
+++ b/src/mon/AuthMonitor.cc
@@ -1773,7 +1773,7 @@ bool AuthMonitor::_upgrade_format_to_dumpling()
     // set daemon profiles
     if ((p->first.is_osd() || p->first.is_mds()) &&
         mon_caps == "allow rwx") {
-      new_caps = string("allow profile ") + p->first.get_type_name();
+      new_caps = string("allow profile ") + std::string(p->first.get_type_name());
     }
 
     // update bootstrap keys
diff --git a/src/mon/MonClient.cc b/src/mon/MonClient.cc
index dbb4287a2a2e..63e3d5cfa3cf 100644
--- a/src/mon/MonClient.cc
+++ b/src/mon/MonClient.cc
@@ -26,6 +26,7 @@
 
 #include "messages/MMonGetMap.h"
 #include "messages/MMonGetVersion.h"
+#include "messages/MMonGetMap.h"
 #include "messages/MMonGetVersionReply.h"
 #include "messages/MMonMap.h"
 #include "messages/MConfig.h"
@@ -46,6 +47,7 @@
 #include "common/LogClient.h"
 
 #include "MonClient.h"
+#include "error_code.h"
 #include "MonMap.h"
 
 #include "auth/Auth.h"
@@ -58,14 +60,16 @@
 #undef dout_prefix
 #define dout_prefix *_dout << "monclient" << (_hunting() ? "(hunting)":"") << ": "
 
+namespace bs = boost::system;
 using std::string;
+using namespace std::literals;
 
-MonClient::MonClient(CephContext *cct_) :
+MonClient::MonClient(CephContext *cct_, boost::asio::io_context& service) :
   Dispatcher(cct_),
   AuthServer(cct_),
   messenger(NULL),
   timer(cct_, monc_lock),
-  finisher(cct_),
+  service(service),
   initialized(false),
   log_client(NULL),
   more_log_pending(false),
@@ -95,7 +99,7 @@ int MonClient::get_monmap()
 {
   ldout(cct, 10) << __func__ << dendl;
   std::unique_lock l(monc_lock);
-  
+
   sub.want("monmap", 0, 0);
   if (!_opened())
     _reopen_session();
@@ -432,13 +436,18 @@ void MonClient::handle_monmap(MMonMap *m)
 void MonClient::handle_config(MConfig *m)
 {
   ldout(cct,10) << __func__ << " " << *m << dendl;
-  finisher.queue(new LambdaContext([this, m](int r) {
-	cct->_conf.set_mon_vals(cct, m->config, config_cb);
-	if (config_notify_cb) {
-	  config_notify_cb();
-	}
-	m->put();
-      }));
+  // Take the sledgehammer approach to ensuring we don't depend on
+  // anything in MonClient.
+  boost::asio::defer(finish_strand,
+		    [m, cct = boost::intrusive_ptr<CephContext>(cct),
+		     config_notify_cb = config_notify_cb,
+		     config_cb = config_cb]() {
+		      cct->_conf.set_mon_vals(cct.get(), m->config, config_cb);
+		      if (config_notify_cb) {
+			config_notify_cb();
+		      }
+		      m->put();
+		    });
   got_config = true;
   map_cond.notify_all();
 }
@@ -478,7 +487,6 @@ int MonClient::init()
   messenger->add_dispatcher_head(this);
 
   timer.init();
-  finisher.start();
   schedule_tick();
 
   return 0;
@@ -490,10 +498,10 @@ void MonClient::shutdown()
   monc_lock.lock();
   stopping = true;
   while (!version_requests.empty()) {
-    version_requests.begin()->second->context->complete(-ECANCELED);
+    ceph::async::defer(std::move(version_requests.begin()->second),
+		       monc_errc::shutting_down, 0, 0);
     ldout(cct, 20) << __func__ << " canceling and discarding version request "
-		   << version_requests.begin()->second << dendl;
-    delete version_requests.begin()->second;
+		   << version_requests.begin()->first << dendl;
     version_requests.erase(version_requests.begin());
   }
   while (!mon_commands.empty()) {
@@ -511,8 +519,6 @@ void MonClient::shutdown()
   monc_lock.unlock();
 
   if (initialized) {
-    finisher.wait_for_empty();
-    finisher.stop();
     initialized = false;
   }
   monc_lock.lock();
@@ -696,8 +702,8 @@ void MonClient::_reopen_session(int rank)
 
   // throw out version check requests
   while (!version_requests.empty()) {
-    finisher.queue(version_requests.begin()->second->context, -EAGAIN);
-    delete version_requests.begin()->second;
+    ceph::async::defer(std::move(version_requests.begin()->second),
+		       monc_errc::session_reset, 0, 0);
     version_requests.erase(version_requests.begin());
   }
 
@@ -1081,10 +1087,9 @@ void MonClient::_send_command(MonCommand *r)
   if (r->is_tell()) {
     ++r->send_attempts;
     if (r->send_attempts > cct->_conf->mon_client_directed_command_retry) {
-      _finish_command(r, -ENXIO, "mon unavailable");
+      _finish_command(r, monc_errc::mon_unavailable, "mon unavailable", {});
       return;
     }
-
     // tell-style command
     if (monmap.min_mon_release >= ceph_release_t::octopus) {
       if (r->target_con) {
@@ -1094,7 +1099,7 @@ void MonClient::_send_command(MonCommand *r)
 	if (r->target_rank >= (int)monmap.size()) {
 	  ldout(cct, 10) << " target " << r->target_rank
 			 << " >= max mon " << monmap.size() << dendl;
-	  _finish_command(r, -ENOENT, "mon rank dne");
+	  _finish_command(r, monc_errc::rank_dne, "mon rank dne"sv, {});
 	  return;
 	}
 	r->target_con = messenger->connect_to_mon(
@@ -1103,7 +1108,7 @@ void MonClient::_send_command(MonCommand *r)
 	if (!monmap.contains(r->target_name)) {
 	  ldout(cct, 10) << " target " << r->target_name
 			 << " not present in monmap" << dendl;
-	  _finish_command(r, -ENOENT, "mon dne");
+	  _finish_command(r, monc_errc::mon_dne, "mon dne"sv, {});
 	  return;
 	}
 	r->target_con = messenger->connect_to_mon(
@@ -1138,7 +1143,7 @@ void MonClient::_send_command(MonCommand *r)
       if (r->target_rank >= (int)monmap.size()) {
 	ldout(cct, 10) << " target " << r->target_rank
 		       << " >= max mon " << monmap.size() << dendl;
-	_finish_command(r, -ENOENT, "mon rank dne");
+	_finish_command(r, monc_errc::rank_dne, "mon rank dne"sv, {});
 	return;
       }
       _reopen_session(r->target_rank);
@@ -1153,7 +1158,7 @@ void MonClient::_send_command(MonCommand *r)
       if (!monmap.contains(r->target_name)) {
 	ldout(cct, 10) << " target " << r->target_name
 		       << " not present in monmap" << dendl;
-	_finish_command(r, -ENOENT, "mon dne");
+	_finish_command(r, monc_errc::mon_dne, "mon dne"sv, {});
 	return;
       }
       _reopen_session(monmap.get_rank(r->target_name));
@@ -1224,9 +1229,10 @@ void MonClient::handle_mon_command_ack(MMonCommandAck *ack)
   }
 
   ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
-  if (r->poutbl)
-    r->poutbl->claim(ack->get_data());
-  _finish_command(r, ack->r, ack->rs);
+  auto ec = ack->r < 0 ? bs::error_code(-ack->r, mon_category())
+    : bs::error_code();
+  _finish_command(r, ec, ack->rs,
+		  std::move(ack->get_data()));
   ack->put();
 }
 
@@ -1251,9 +1257,9 @@ void MonClient::handle_command_reply(MCommandReply *reply)
   }
 
   ldout(cct, 10) << __func__ << " " << r->tid << " " << r->cmd << dendl;
-  if (r->poutbl)
-    r->poutbl->claim(reply->get_data());
-  _finish_command(r, reply->r, reply->rs);
+  auto ec = reply->r < 0 ? bs::error_code(-reply->r, mon_category())
+    : bs::error_code();
+  _finish_command(r, ec, reply->rs, std::move(reply->get_data()));
   reply->put();
 }
 
@@ -1270,19 +1276,17 @@ int MonClient::_cancel_mon_command(uint64_t tid)
   ldout(cct, 10) << __func__ << " tid " << tid << dendl;
 
   MonCommand *cmd = it->second;
-  _finish_command(cmd, -ETIMEDOUT, "");
+  _finish_command(cmd, monc_errc::timed_out, "timed out"sv, {});
   return 0;
 }
 
-void MonClient::_finish_command(MonCommand *r, int ret, string rs)
+void MonClient::_finish_command(MonCommand *r, bs::error_code ret,
+				std::string_view rs, ceph::buffer::list&& bl)
 {
-  ldout(cct, 10) << __func__ << " " << r->tid << " = " << ret << " " << rs << dendl;
-  if (r->prval)
-    *(r->prval) = ret;
-  if (r->prs)
-    *(r->prs) = rs;
-  if (r->onfinish)
-    finisher.queue(r->onfinish, ret);
+  ldout(cct, 10) << __func__ << " " << r->tid << " = " << ret << " " << rs
+		 << dendl;
+  ceph::async::defer(std::move(r->onfinish), ret, std::string(rs),
+		     std::move(bl));
   if (r->target_con) {
     r->target_con->mark_down();
   }
@@ -1290,117 +1294,8 @@ void MonClient::_finish_command(MonCommand *r, int ret, string rs)
   delete r;
 }
 
-void MonClient::start_mon_command(const std::vector<string>& cmd,
-                                  const ceph::buffer::list& inbl,
-                                  ceph::buffer::list *outbl, string *outs,
-                                  Context *onfinish)
-{
-  ldout(cct,10) << __func__ << " cmd=" << cmd << dendl;
-  std::lock_guard l(monc_lock);
-  if (!initialized || stopping) {
-    if (onfinish) {
-      onfinish->complete(-ECANCELED);
-    }
-    return;
-  }
-  MonCommand *r = new MonCommand(++last_mon_command_tid);
-  r->cmd = cmd;
-  r->inbl = inbl;
-  r->poutbl = outbl;
-  r->prs = outs;
-  r->onfinish = onfinish;
-  if (cct->_conf->rados_mon_op_timeout > 0) {
-    class C_CancelMonCommand : public Context
-    {
-      uint64_t tid;
-      MonClient *monc;
-      public:
-      C_CancelMonCommand(uint64_t tid, MonClient *monc) : tid(tid), monc(monc) {}
-      void finish(int r) override {
-	monc->_cancel_mon_command(tid);
-      }
-    };
-    r->ontimeout = new C_CancelMonCommand(r->tid, this);
-    timer.add_event_after(cct->_conf->rados_mon_op_timeout, r->ontimeout);
-  }
-  mon_commands[r->tid] = r;
-  _send_command(r);
-}
-
-void MonClient::start_mon_command(const string &mon_name,
-                                  const std::vector<string>& cmd,
-                                  const ceph::buffer::list& inbl,
-                                  ceph::buffer::list *outbl, string *outs,
-                                  Context *onfinish)
-{
-  ldout(cct,10) << __func__ << " mon." << mon_name << " cmd=" << cmd << dendl;
-  std::lock_guard l(monc_lock);
-  if (!initialized || stopping) {
-    if (onfinish) {
-      onfinish->complete(-ECANCELED);
-    }
-    return;
-  }
-  MonCommand *r = new MonCommand(++last_mon_command_tid);
-
-  // detect/tolerate mon *rank* passed as a string
-  string err;
-  int rank = strict_strtoll(mon_name.c_str(), 10, &err);
-  if (err.size() == 0 && rank >= 0) {
-    ldout(cct,10) << __func__ << " interpreting name '" << mon_name
-		  << "' as rank " << rank << dendl;
-    r->target_rank = rank;
-  } else {
-    r->target_name = mon_name;
-  }
-  r->cmd = cmd;
-  r->inbl = inbl;
-  r->poutbl = outbl;
-  r->prs = outs;
-  r->onfinish = onfinish;
-  mon_commands[r->tid] = r;
-  _send_command(r);
-}
-
-void MonClient::start_mon_command(int rank,
-                                  const std::vector<string>& cmd,
-                                  const ceph::buffer::list& inbl,
-                                  ceph::buffer::list *outbl, string *outs,
-                                  Context *onfinish)
-{
-  ldout(cct,10) << __func__ << " rank " << rank << " cmd=" << cmd << dendl;
-  std::lock_guard l(monc_lock);
-  if (!initialized || stopping) {
-    if (onfinish) {
-      onfinish->complete(-ECANCELED);
-    }
-    return;
-  }
-  MonCommand *r = new MonCommand(++last_mon_command_tid);
-  r->target_rank = rank;
-  r->cmd = cmd;
-  r->inbl = inbl;
-  r->poutbl = outbl;
-  r->prs = outs;
-  r->onfinish = onfinish;
-  mon_commands[r->tid] = r;
-  _send_command(r);
-}
-
 // ---------
 
-void MonClient::get_version(string map, version_t *newest, version_t *oldest, Context *onfinish)
-{
-  version_req_d *req = new version_req_d(onfinish, newest, oldest);
-  ldout(cct, 10) << "get_version " << map << " req " << req << dendl;
-  std::lock_guard l(monc_lock);
-  auto m = ceph::make_message<MMonGetVersion>();
-  m->what = map;
-  m->handle = ++version_req_id;
-  version_requests[m->handle] = req;
-  _send_mon_message(std::move(m));
-}
-
 void MonClient::handle_get_version_reply(MMonGetVersionReply* m)
 {
   ceph_assert(ceph_mutex_is_locked(monc_lock));
@@ -1409,15 +1304,12 @@ void MonClient::handle_get_version_reply(MMonGetVersionReply* m)
     ldout(cct, 0) << __func__ << " version request with handle " << m->handle
 		  << " not found" << dendl;
   } else {
-    version_req_d *req = iter->second;
-    ldout(cct, 10) << __func__ << " finishing " << req << " version " << m->version << dendl;
+    auto req = std::move(iter->second);
+    ldout(cct, 10) << __func__ << " finishing " << iter->first << " version "
+		   << m->version << dendl;
     version_requests.erase(iter);
-    if (req->newest)
-      *req->newest = m->version;
-    if (req->oldest)
-      *req->oldest = m->oldest_version;
-    finisher.queue(req->context, 0);
-    delete req;
+    ceph::async::defer(std::move(req), bs::error_code(),
+		       m->version, m->oldest_version);
   }
   m->put();
 }
@@ -1587,7 +1479,8 @@ int MonClient::handle_auth_bad_method(
 	    allowed_methods,
 	    allowed_modes);
 	  if (r < 0) {
-	    _finish_command(i.second, r, "auth failed");
+	    auto ec = bs::error_code(-r, mon_category());
+	    _finish_command(i.second, ec, "auth failed"sv, {});
 	  }
 	  return r;
 	}
@@ -1996,3 +1889,97 @@ void MonClient::register_config_callback(md_config_t::config_callback fn) {
 md_config_t::config_callback MonClient::get_config_callback() {
   return config_cb;
 }
+
+class monc_error_category : public ceph::converting_category {
+public:
+  monc_error_category(){}
+  const char* name() const noexcept override;
+  std::string message(int ev) const override;
+  bs::error_condition default_error_condition(int ev) const noexcept
+    override;
+  bool equivalent(int ev, const bs::error_condition& c) const
+    noexcept override;
+  using ceph::converting_category::equivalent;
+  int from_code(int ev) const noexcept override;
+};
+
+const char* monc_error_category::name() const noexcept {
+  return "monc";
+}
+
+std::string monc_error_category::message(int ev) const {
+  if (ev == 0)
+    return "No error";
+
+  switch (static_cast<monc_errc>(ev)) {
+  case monc_errc::shutting_down: // Command failed due to MonClient shutting down
+    return "Command failed due to MonClient shutting down";
+  case monc_errc::session_reset:
+    return "Monitor session was reset";
+  case monc_errc::rank_dne:
+    return "Requested monitor rank does not exist";
+  case monc_errc::mon_dne:
+    return "Requested monitor does not exist";
+  case monc_errc::timed_out:
+    return "Monitor operation timed out";
+  case monc_errc::mon_unavailable:
+    return "Monitor unavailable";
+  }
+
+  return "Unknown error";
+}
+
+bs::error_condition monc_error_category::default_error_condition(int ev) const noexcept {
+  switch (static_cast<monc_errc>(ev)) {
+  case monc_errc::shutting_down:
+    return bs::errc::operation_canceled;
+  case monc_errc::session_reset:
+    return bs::errc::resource_unavailable_try_again;
+  case monc_errc::rank_dne:
+    [[fallthrough]];
+  case monc_errc::mon_dne:
+    return ceph::errc::not_in_map;
+  case monc_errc::timed_out:
+    return bs::errc::timed_out;
+  case monc_errc::mon_unavailable:
+    return bs::errc::no_such_device;
+  }
+  return { ev, *this };
+}
+
+bool monc_error_category::equivalent(int ev, const bs::error_condition& c) const noexcept {
+  switch (static_cast<monc_errc>(ev)) {
+  case monc_errc::rank_dne:
+    [[fallthrough]];
+  case monc_errc::mon_dne:
+      return c == bs::errc::no_such_file_or_directory;
+  default:
+    return default_error_condition(ev) == c;
+  }
+}
+
+int monc_error_category::from_code(int ev) const noexcept {
+  if (ev == 0)
+    return 0;
+
+  switch (static_cast<monc_errc>(ev)) {
+  case monc_errc::shutting_down:
+    return -ECANCELED;
+  case monc_errc::session_reset:
+    return -EAGAIN;
+  case monc_errc::rank_dne:
+    [[fallthrough]];
+  case monc_errc::mon_dne:
+    return -ENOENT;
+  case monc_errc::timed_out:
+    return -ETIMEDOUT;
+  case monc_errc::mon_unavailable:
+    return -ENXIO;
+  }
+  return -EDOM;
+}
+
+const bs::error_category& monc_category() noexcept {
+  static const monc_error_category c;
+  return c;
+}
diff --git a/src/mon/MonClient.h b/src/mon/MonClient.h
index 52c55fb18f32..2d97440df8f6 100644
--- a/src/mon/MonClient.h
+++ b/src/mon/MonClient.h
@@ -27,9 +27,10 @@
 #include "MonMap.h"
 #include "MonSub.h"
 
+#include "common/async/completion.h"
 #include "common/Timer.h"
-#include "common/Finisher.h"
 #include "common/config.h"
+#include "messages/MMonGetVersion.h"
 
 #include "auth/AuthClient.h"
 #include "auth/AuthServer.h"
@@ -37,11 +38,8 @@
 class MMonMap;
 class MConfig;
 class MMonGetVersionReply;
-struct MMonSubscribeAck;
 class MMonCommandAck;
-struct MAuthReply;
 class LogClient;
-class AuthAuthorizer;
 class AuthClientHandler;
 class AuthRegistry;
 class KeyRing;
@@ -141,7 +139,6 @@ private:
 
 struct MonClientPinger : public Dispatcher,
 			 public AuthClient {
-
   ceph::mutex lock = ceph::make_mutex("MonClientPinger::lock");
   ceph::condition_variable ping_recvd_cond;
   std::string *result;
@@ -240,11 +237,52 @@ struct MonClientPinger : public Dispatcher,
   }
 };
 
+const boost::system::error_category& monc_category() noexcept;
+
+enum class monc_errc {
+  shutting_down = 1, // Command failed due to MonClient shutting down
+  session_reset, // Monitor session was reset
+  rank_dne, // Requested monitor rank does not exist
+  mon_dne, // Requested monitor does not exist
+  timed_out, // Monitor operation timed out
+  mon_unavailable // Monitor unavailable
+};
+
+namespace boost {
+namespace system {
+template<>
+struct is_error_code_enum<::monc_errc> {
+  static const bool value = true;
+};
+}
+}
+
+//  explicit conversion:
+inline boost::system::error_code make_error_code(monc_errc e) noexcept {
+  return { static_cast<int>(e), monc_category() };
+}
+
+// implicit conversion:
+inline boost::system::error_condition make_error_condition(monc_errc e)
+  noexcept {
+  return { static_cast<int>(e), monc_category() };
+}
+
+const boost::system::error_category& monc_category() noexcept;
 
 class MonClient : public Dispatcher,
 		  public AuthClient,
 		  public AuthServer /* for mgr, osd, mds */ {
+  static constexpr auto dout_subsys = ceph_subsys_monc;
 public:
+  // Error, Newest, Oldest
+  using VersionSig = void(boost::system::error_code, version_t, version_t);
+  using VersionCompletion = ceph::async::Completion<VersionSig>;
+
+  using CommandSig = void(boost::system::error_code, std::string,
+			  ceph::buffer::list);
+  using CommandCompletion = ceph::async::Completion<CommandSig>;
+
   MonMap monmap;
   std::map<std::string,std::string> config_mgr;
 
@@ -259,7 +297,8 @@ private:
 
   mutable ceph::mutex monc_lock = ceph::make_mutex("MonClient::monc_lock");
   SafeTimer timer;
-  Finisher finisher;
+  boost::asio::io_context& service;
+  boost::asio::io_context::strand finish_strand{service};
 
   bool initialized;
   bool stopping = false;
@@ -304,7 +343,6 @@ private:
   double reopen_interval_multiplier;
 
   Dispatcher *handle_authentication_dispatcher = nullptr;
-  
   bool _opened() const;
   bool _hunting() const;
   void _start_hunting();
@@ -361,7 +399,7 @@ public:
     bool more,
     uint32_t auth_method,
     const ceph::buffer::list& bl,
-    ceph::buffer::list *reply);
+    ceph::buffer::list *reply) override;
 
   void set_entity_name(EntityName name) { entity_name = name; }
   void set_handle_authentication_dispatcher(Dispatcher *d) {
@@ -410,12 +448,12 @@ public:
     std::lock_guard l(monc_lock);
     return sub.inc_want(what, start, flags);
   }
-  
+
   std::unique_ptr<KeyRing> keyring;
   std::unique_ptr<RotatingKeyRing> rotating_secrets;
 
  public:
-  explicit MonClient(CephContext *cct_);
+  MonClient(CephContext *cct_, boost::asio::io_context& service);
   MonClient(const MonClient &) = delete;
   MonClient& operator=(const MonClient &) = delete;
   ~MonClient() override;
@@ -511,66 +549,206 @@ private:
   struct MonCommand {
     // for tell only
     std::string target_name;
-    int target_rank;
+    int target_rank = -1;
     ConnectionRef target_con;
     std::unique_ptr<MonConnection> target_session;
     unsigned send_attempts = 0;  ///< attempt count for legacy mons
     utime_t last_send_attempt;
-
     uint64_t tid;
     std::vector<std::string> cmd;
     ceph::buffer::list inbl;
-    ceph::buffer::list *poutbl;
-    std::string *prs;
-    int *prval;
-    Context *onfinish, *ontimeout;
-
-    explicit MonCommand(uint64_t t)
-      : target_rank(-1),
-	tid(t),
-	poutbl(NULL), prs(NULL), prval(NULL), onfinish(NULL), ontimeout(NULL)
-    {}
+    std::unique_ptr<CommandCompletion> onfinish;
+    std::optional<boost::asio::steady_timer> cancel_timer;
+
+    MonCommand(MonClient& monc, uint64_t t, std::unique_ptr<CommandCompletion> onfinish)
+      : tid(t), onfinish(std::move(onfinish)) {
+      auto timeout = ceph::maybe_timespan(monc.cct->_conf->rados_mon_op_timeout);
+      if (timeout) {
+	cancel_timer.emplace(monc.service, *timeout);
+	cancel_timer->async_wait(
+          [this, &monc](boost::system::error_code ec) {
+	    if (ec)
+	      return;
+	    std::scoped_lock l(monc.monc_lock);
+	    monc._cancel_mon_command(tid);
+	  });
+      }
+    }
 
     bool is_tell() const {
       return target_name.size() || target_rank >= 0;
     }
   };
+  friend MonCommand;
   std::map<uint64_t,MonCommand*> mon_commands;
 
   void _send_command(MonCommand *r);
   void _check_tell_commands();
   void _resend_mon_commands();
   int _cancel_mon_command(uint64_t tid);
-  void _finish_command(MonCommand *r, int ret, std::string rs);
+  void _finish_command(MonCommand *r, boost::system::error_code ret, std::string_view rs,
+		       bufferlist&& bl);
   void _finish_auth();
   void handle_mon_command_ack(MMonCommandAck *ack);
   void handle_command_reply(MCommandReply *reply);
 
 public:
-  void start_mon_command(const std::vector<std::string>& cmd, const ceph::buffer::list& inbl,
-			ceph::buffer::list *outbl, std::string *outs,
-			Context *onfinish);
+  template<typename CompletionToken>
+  auto start_mon_command(const std::vector<std::string>& cmd,
+                         const ceph::buffer::list& inbl,
+			 CompletionToken&& token) {
+    ldout(cct,10) << __func__ << " cmd=" << cmd << dendl;
+    boost::asio::async_completion<CompletionToken, CommandSig> init(token);
+    {
+      std::scoped_lock l(monc_lock);
+      auto h = CommandCompletion::create(service.get_executor(),
+					 std::move(init.completion_handler));
+      if (!initialized || stopping) {
+	ceph::async::post(std::move(h), monc_errc::shutting_down, std::string{},
+			  bufferlist{});
+      } else {
+	auto r = new MonCommand(*this, ++last_mon_command_tid, std::move(h));
+	r->cmd = cmd;
+	r->inbl = inbl;
+	mon_commands.emplace(r->tid, r);
+	_send_command(r);
+      }
+    }
+    return init.result.get();
+  }
+
+  template<typename CompletionToken>
+  auto start_mon_command(int mon_rank, const std::vector<std::string>& cmd,
+			 const ceph::buffer::list& inbl, CompletionToken&& token) {
+    ldout(cct,10) << __func__ << " cmd=" << cmd << dendl;
+    boost::asio::async_completion<CompletionToken, CommandSig> init(token);
+    {
+      std::scoped_lock l(monc_lock);
+      auto h = CommandCompletion::create(service.get_executor(),
+					 std::move(init.completion_handler));
+      if (!initialized || stopping) {
+	ceph::async::post(std::move(h), monc_errc::shutting_down, std::string{},
+			  bufferlist{});
+      } else {
+	auto r = new MonCommand(*this, ++last_mon_command_tid, std::move(h));
+	r->target_rank = mon_rank;
+	r->cmd = cmd;
+	r->inbl = inbl;
+	mon_commands.emplace(r->tid, r);
+	_send_command(r);
+      }
+    }
+    return init.result.get();
+  }
+
+  template<typename CompletionToken>
+  auto start_mon_command(const std::string& mon_name,
+                         const std::vector<std::string>& cmd,
+			 const ceph::buffer::list& inbl,
+			 CompletionToken&& token) {
+    ldout(cct,10) << __func__ << " cmd=" << cmd << dendl;
+    boost::asio::async_completion<CompletionToken, CommandSig> init(token);
+    {
+      std::scoped_lock l(monc_lock);
+      auto h = CommandCompletion::create(service.get_executor(),
+					 std::move(init.completion_handler));
+      if (!initialized || stopping) {
+	ceph::async::post(std::move(h), monc_errc::shutting_down, std::string{},
+			  bufferlist{});
+      } else {
+	auto r = new MonCommand(*this, ++last_mon_command_tid, std::move(h));
+	// detect/tolerate mon *rank* passed as a string
+	std::string err;
+	int rank = strict_strtoll(mon_name.c_str(), 10, &err);
+	if (err.size() == 0 && rank >= 0) {
+	  ldout(cct,10) << __func__ << " interpreting name '" << mon_name
+			<< "' as rank " << rank << dendl;
+	  r->target_rank = rank;
+	} else {
+	  r->target_name = mon_name;
+	}
+	r->cmd = cmd;
+	r->inbl = inbl;
+	mon_commands.emplace(r->tid, r);
+	_send_command(r);
+      }
+    }
+    return init.result.get();
+  }
+
+  class ContextVerter {
+    std::string* outs;
+    ceph::bufferlist* outbl;
+    Context* onfinish;
+
+  public:
+    ContextVerter(std::string* outs, ceph::bufferlist* outbl, Context* onfinish)
+      : outs(outs), outbl(outbl), onfinish(onfinish) {}
+    ~ContextVerter() = default;
+    ContextVerter(const ContextVerter&) = default;
+    ContextVerter& operator =(const ContextVerter&) = default;
+    ContextVerter(ContextVerter&&) = default;
+    ContextVerter& operator =(ContextVerter&&) = default;
+
+    void operator()(boost::system::error_code e,
+		    std::string s,
+		    ceph::bufferlist bl) {
+      if (outs)
+	*outs = std::move(s);
+      if (outbl)
+	*outbl = std::move(bl);
+      if (onfinish)
+	onfinish->complete(ceph::from_error_code(e));
+    }
+  };
+
+  void start_mon_command(const vector<string>& cmd, const bufferlist& inbl,
+			 bufferlist *outbl, string *outs,
+			 Context *onfinish) {
+    start_mon_command(cmd, inbl, ContextVerter(outs, outbl, onfinish));
+  }
   void start_mon_command(int mon_rank,
-                         const std::vector<std::string>& cmd, const ceph::buffer::list& inbl,
-                         ceph::buffer::list *outbl, std::string *outs,
-                         Context *onfinish);
-  void start_mon_command(const std::string &mon_name,  ///< mon name, with mon. prefix
-                         const std::vector<std::string>& cmd, const ceph::buffer::list& inbl,
-                         ceph::buffer::list *outbl, std::string *outs,
-                         Context *onfinish);
+			 const vector<string>& cmd, const bufferlist& inbl,
+			 bufferlist *outbl, string *outs,
+			 Context *onfinish) {
+    start_mon_command(mon_rank, cmd, inbl, ContextVerter(outs, outbl, onfinish));
+  }
+  void start_mon_command(const string &mon_name,  ///< mon name, with mon. prefix
+			 const vector<string>& cmd, const bufferlist& inbl,
+			 bufferlist *outbl, string *outs,
+			 Context *onfinish) {
+    start_mon_command(mon_name, cmd, inbl, ContextVerter(outs, outbl, onfinish));
+  }
+
 
   // version requests
 public:
   /**
    * get latest known version(s) of cluster map
    *
-   * @param map std::string name of map (e.g., 'osdmap')
-   * @param newest pointer where newest map version will be stored
-   * @param oldest pointer where oldest map version will be stored
-   * @param onfinish context that will be triggered on completion
-   * @return (via context) 0 on success, -EAGAIN if we need to resubmit our request
+   * @param map string name of map (e.g., 'osdmap')
+   * @param token context that will be triggered on completion
+   * @return (via Completion) {} on success,
+   *         boost::system::errc::resource_unavailable_try_again if we need to
+   *         resubmit our request
    */
-  void get_version(std::string map, version_t *newest, version_t *oldest, Context *onfinish);
+  template<typename CompletionToken>
+  auto get_version(std::string&& map, CompletionToken&& token) {
+    boost::asio::async_completion<CompletionToken, VersionSig> init(token);
+    {
+      std::scoped_lock l(monc_lock);
+      auto m = ceph::make_message<MMonGetVersion>();
+      m->what = std::move(map);
+      m->handle = ++version_req_id;
+      version_requests.emplace(m->handle,
+			       VersionCompletion::create(
+				 service.get_executor(),
+				 std::move(init.completion_handler)));
+      _send_mon_message(m);
+    }
+    return init.result.get();
+  }
+
   /**
    * Run a callback within our lock, with a reference
    * to the MonMap
@@ -589,16 +767,10 @@ public:
   md_config_t::config_callback get_config_callback();
 
 private:
-  struct version_req_d {
-    Context *context;
-    version_t *newest, *oldest;
-    version_req_d(Context *con, version_t *n, version_t *o) : context(con),newest(n), oldest(o) {}
-  };
 
-  std::map<ceph_tid_t, version_req_d*> version_requests;
+  std::map<ceph_tid_t, std::unique_ptr<VersionCompletion>> version_requests;
   ceph_tid_t version_req_id;
   void handle_get_version_reply(MMonGetVersionReply* m);
-
   md_config_t::config_callback config_cb;
   std::function<void(void)> config_notify_cb;
 };
diff --git a/src/mount/conf.cc b/src/mount/conf.cc
index 228f53e72187..b78aa562f9bf 100644
--- a/src/mount/conf.cc
+++ b/src/mount/conf.cc
@@ -6,6 +6,7 @@
 #include <cstring>
 #include <map>
 
+#include "common/async/context_pool.h"
 #include "common/ceph_context.h"
 #include "common/ceph_argparse.h"
 #include "global/global_init.h"
@@ -40,7 +41,8 @@ extern "C" void mount_ceph_get_config_info(const char *config_file,
   conf.parse_env(cct->get_module_type()); // environment variables override
   conf.apply_changes(nullptr);
 
-  MonClient monc = MonClient(cct.get());
+  ceph::async::io_context_pool ioc(1);
+  MonClient monc = MonClient(cct.get(), ioc);
   err = monc.build_initial_monmap();
   if (err)
     goto scrape_keyring;
diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index bba59d7073b2..e23c63141ae5 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -220,7 +220,7 @@ CompatSet OSD::get_osd_compat_set() {
   return compat;
 }
 
-OSDService::OSDService(OSD *osd) :
+OSDService::OSDService(OSD *osd, ceph::async::io_context_pool& poolctx) :
   osd(osd),
   cct(osd->cct),
   whoami(osd->whoami), store(osd->store),
@@ -248,6 +248,7 @@ OSDService::OSDService(OSD *osd) :
   last_recalibrate(ceph_clock_now()),
   promote_max_objects(0),
   promote_max_bytes(0),
+  poolctx(poolctx),
   objecter(new Objecter(osd->client_messenger->cct, osd->objecter_messenger, osd->monc, NULL, 0, 0)),
   m_objecter_finishers(cct->_conf->osd_objecter_finishers),
   watch_timer(osd->client_messenger->cct, watch_lock),
@@ -2112,7 +2113,8 @@ OSD::OSD(CephContext *cct_, ObjectStore *store_,
 	 Messenger *hb_back_serverm,
 	 Messenger *osdc_messenger,
 	 MonClient *mc,
-	 const std::string &dev, const std::string &jdev) :
+	 const std::string &dev, const std::string &jdev,
+	 ceph::async::io_context_pool& poolctx) :
   Dispatcher(cct_),
   tick_timer(cct, osd_lock),
   tick_timer_without_osd_lock(cct, tick_timer_lock),
@@ -2159,7 +2161,7 @@ OSD::OSD(CephContext *cct_, ObjectStore *store_,
   up_thru_wanted(0),
   requested_full_first(0),
   requested_full_last(0),
-  service(this)
+  service(this, poolctx)
 {
 
   if (!gss_ktfile_client.empty()) {
@@ -6184,12 +6186,12 @@ bool OSD::ms_handle_refused(Connection *con)
   return true;
 }
 
-struct C_OSD_GetVersion : public Context {
+struct CB_OSD_GetVersion {
   OSD *osd;
-  uint64_t oldest, newest;
-  explicit C_OSD_GetVersion(OSD *o) : osd(o), oldest(0), newest(0) {}
-  void finish(int r) override {
-    if (r >= 0)
+  explicit CB_OSD_GetVersion(OSD *o) : osd(o) {}
+  void operator ()(boost::system::error_code ec, version_t newest,
+		   version_t oldest) {
+    if (!ec)
       osd->_got_mon_epochs(oldest, newest);
   }
 };
@@ -6209,8 +6211,7 @@ void OSD::start_boot()
   set_state(STATE_PREBOOT);
   dout(10) << "start_boot - have maps " << superblock.oldest_map
 	   << ".." << superblock.newest_map << dendl;
-  C_OSD_GetVersion *c = new C_OSD_GetVersion(this);
-  monc->get_version("osdmap", &c->newest, &c->oldest, c);
+  monc->get_version("osdmap", CB_OSD_GetVersion(this));
 }
 
 void OSD::_got_mon_epochs(epoch_t oldest, epoch_t newest)
@@ -9790,6 +9791,10 @@ void OSD::handle_conf_change(const ConfigProxy& conf,
     dout(0) << __func__ << ": scrub interval change" << dendl;
   }
   check_config();
+  if (changed.count("osd_asio_thread_count")) {
+    service.poolctx.stop();
+    service.poolctx.start(conf.get_val<std::uint64_t>("osd_asio_thread_count"));
+  }
 }
 
 void OSD::update_log_config()
diff --git a/src/osd/OSD.h b/src/osd/OSD.h
index 7713c208f8a6..84e81c95e9ee 100644
--- a/src/osd/OSD.h
+++ b/src/osd/OSD.h
@@ -19,6 +19,7 @@
 
 #include "msg/Dispatcher.h"
 
+#include "common/async/context_pool.h"
 #include "common/Timer.h"
 #include "common/WorkQueue.h"
 #include "common/AsyncReserver.h"
@@ -511,6 +512,7 @@ public:
   void promote_throttle_recalibrate();
 
   // -- Objecter, for tiering reads/writes from/to other OSDs --
+  ceph::async::io_context_pool& poolctx;
   Objecter *objecter;
   int m_objecter_finishers;
   vector<Finisher*> objecter_finishers;
@@ -898,7 +900,7 @@ public:
   void dump_live_pgids();
 #endif
 
-  explicit OSDService(OSD *osd);
+  OSDService(OSD *osd, ceph::async::io_context_pool& poolctx);
   ~OSDService();
 };
 
@@ -1827,7 +1829,7 @@ protected:
 
   void send_full_update();
   
-  friend struct C_OSD_GetVersion;
+  friend struct CB_OSD_GetVersion;
 
   // -- alive --
   epoch_t up_thru_wanted;
@@ -2001,7 +2003,8 @@ private:
       Messenger *hb_front_server,
       Messenger *hb_back_server,
       Messenger *osdc_messenger,
-      MonClient *mc, const std::string &dev, const std::string &jdev);
+      MonClient *mc, const std::string &dev, const std::string &jdev,
+      ceph::async::io_context_pool& poolctx);
   ~OSD() override;
 
   // static bits
diff --git a/src/osdc/Objecter.cc b/src/osdc/Objecter.cc
index 37dc01d0144d..181c841edda2 100644
--- a/src/osdc/Objecter.cc
+++ b/src/osdc/Objecter.cc
@@ -45,6 +45,7 @@
 #include "messages/MWatchNotify.h"
 
 
+#include "common/Cond.h"
 #include "common/config.h"
 #include "common/perf_counters.h"
 #include "common/scrub_types.h"
@@ -1437,13 +1438,15 @@ void Objecter::emit_blacklist_events(const OSDMap &old_osd_map,
 
 // op pool check
 
-void Objecter::C_Op_Map_Latest::finish(int r)
+void Objecter::CB_Op_Map_Latest::operator()(boost::system::error_code e,
+					    version_t latest, version_t)
 {
-  if (r == -EAGAIN || r == -ECANCELED)
+  if (e == boost::system::errc::resource_unavailable_try_again ||
+      e == boost::system::errc::operation_canceled)
     return;
 
   lgeneric_subdout(objecter->cct, objecter, 10)
-    << "op_map_latest r=" << r << " tid=" << tid
+    << "op_map_latest r=" << e << " tid=" << tid
     << " latest " << latest << dendl;
 
   Objecter::unique_lock wl(objecter->rwlock);
@@ -1584,8 +1587,7 @@ void Objecter::_send_op_map_check(Op *op)
   if (check_latest_map_ops.count(op->tid) == 0) {
     op->get();
     check_latest_map_ops[op->tid] = op;
-    C_Op_Map_Latest *c = new C_Op_Map_Latest(this, op->tid);
-    monc->get_version("osdmap", &c->latest, NULL, c);
+    monc->get_version("osdmap", CB_Op_Map_Latest(this, op->tid));
   }
 }
 
@@ -1603,9 +1605,12 @@ void Objecter::_op_cancel_map_check(Op *op)
 
 // linger pool check
 
-void Objecter::C_Linger_Map_Latest::finish(int r)
+void Objecter::CB_Linger_Map_Latest::operator()(boost::system::error_code e,
+						version_t latest,
+						version_t)
 {
-  if (r == -EAGAIN || r == -ECANCELED) {
+  if (e == boost::system::errc::resource_unavailable_try_again ||
+      e == boost::system::errc::operation_canceled) {
     // ignore callback; we will retry in resend_mon_ops()
     return;
   }
@@ -1675,8 +1680,7 @@ void Objecter::_send_linger_map_check(LingerOp *op)
   if (check_latest_map_lingers.count(op->linger_id) == 0) {
     op->get();
     check_latest_map_lingers[op->linger_id] = op;
-    C_Linger_Map_Latest *c = new C_Linger_Map_Latest(this, op->linger_id);
-    monc->get_version("osdmap", &c->latest, NULL, c);
+    monc->get_version("osdmap", CB_Linger_Map_Latest(this, op->linger_id));
   }
 }
 
@@ -1695,9 +1699,11 @@ void Objecter::_linger_cancel_map_check(LingerOp *op)
 
 // command pool check
 
-void Objecter::C_Command_Map_Latest::finish(int r)
+void Objecter::CB_Command_Map_Latest::operator()(boost::system::error_code e,
+						 version_t latest, version_t)
 {
-  if (r == -EAGAIN || r == -ECANCELED) {
+  if (e == boost::system::errc::resource_unavailable_try_again ||
+      e == boost::system::errc::operation_canceled) {
     // ignore callback; we will retry in resend_mon_ops()
     return;
   }
@@ -1750,8 +1756,7 @@ void Objecter::_send_command_map_check(CommandOp *c)
   if (check_latest_map_commands.count(c->tid) == 0) {
     c->get();
     check_latest_map_commands[c->tid] = c;
-    C_Command_Map_Latest *f = new C_Command_Map_Latest(this, c->tid);
-    monc->get_version("osdmap", &f->latest, NULL, f);
+    monc->get_version("osdmap", CB_Command_Map_Latest(this, c->tid));
   }
 }
 
@@ -1930,16 +1935,15 @@ void Objecter::wait_for_osd_map()
   cond.wait(mlock, [&done] { return done; });
 }
 
-struct C_Objecter_GetVersion : public Context {
+struct CB_Objecter_GetVersion {
   Objecter *objecter;
-  uint64_t oldest, newest;
   Context *fin;
-  C_Objecter_GetVersion(Objecter *o, Context *c)
-    : objecter(o), oldest(0), newest(0), fin(c) {}
-  void finish(int r) override {
-    if (r >= 0) {
+  CB_Objecter_GetVersion(Objecter *o, Context *c) : objecter(o), fin(c) {}
+  void operator()(boost::system::error_code e, version_t newest, version_t oldest) {
+    if (!e) {
       objecter->get_latest_version(oldest, newest, fin);
-    } else if (r == -EAGAIN) { // try again as instructed
+    } else if (e == boost::system::errc::resource_unavailable_try_again) {
+      // try again as instructed
       objecter->wait_for_latest_osdmap(fin);
     } else {
       // it doesn't return any other error codes!
@@ -1951,8 +1955,7 @@ struct C_Objecter_GetVersion : public Context {
 void Objecter::wait_for_latest_osdmap(Context *fin)
 {
   ldout(cct, 10) << __func__ << dendl;
-  C_Objecter_GetVersion *c = new C_Objecter_GetVersion(this, fin);
-  monc->get_version("osdmap", &c->newest, &c->oldest, c);
+  monc->get_version("osdmap", CB_Objecter_GetVersion(this, fin));
 }
 
 void Objecter::get_latest_version(epoch_t oldest, epoch_t newest, Context *fin)
@@ -2230,24 +2233,20 @@ void Objecter::resend_mon_ops()
   for (map<ceph_tid_t, Op*>::iterator p = check_latest_map_ops.begin();
        p != check_latest_map_ops.end();
        ++p) {
-    C_Op_Map_Latest *c = new C_Op_Map_Latest(this, p->second->tid);
-    monc->get_version("osdmap", &c->latest, NULL, c);
+    monc->get_version("osdmap", CB_Op_Map_Latest(this, p->second->tid));
   }
 
   for (map<uint64_t, LingerOp*>::iterator p = check_latest_map_lingers.begin();
        p != check_latest_map_lingers.end();
        ++p) {
-    C_Linger_Map_Latest *c
-      = new C_Linger_Map_Latest(this, p->second->linger_id);
-    monc->get_version("osdmap", &c->latest, NULL, c);
+    monc->get_version("osdmap", CB_Linger_Map_Latest(this, p->second->linger_id));
   }
 
   for (map<uint64_t, CommandOp*>::iterator p
 	 = check_latest_map_commands.begin();
        p != check_latest_map_commands.end();
        ++p) {
-    C_Command_Map_Latest *c = new C_Command_Map_Latest(this, p->second->tid);
-    monc->get_version("osdmap", &c->latest, NULL, c);
+    monc->get_version("osdmap", CB_Command_Map_Latest(this, p->second->tid));
   }
 }
 
diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index f669ce31c5fb..d989bd71b431 100644
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -1537,22 +1537,18 @@ public:
     }
   };
 
-  struct C_Op_Map_Latest : public Context {
+  struct CB_Op_Map_Latest {
     Objecter *objecter;
     ceph_tid_t tid;
-    version_t latest;
-    C_Op_Map_Latest(Objecter *o, ceph_tid_t t) : objecter(o), tid(t),
-						 latest(0) {}
-    void finish(int r) override;
+    CB_Op_Map_Latest(Objecter *o, ceph_tid_t t) : objecter(o), tid(t) {}
+    void operator()(boost::system::error_code err, version_t latest, version_t);
   };
 
-  struct C_Command_Map_Latest : public Context {
+  struct CB_Command_Map_Latest {
     Objecter *objecter;
     uint64_t tid;
-    version_t latest;
-    C_Command_Map_Latest(Objecter *o, ceph_tid_t t) :  objecter(o), tid(t),
-						       latest(0) {}
-    void finish(int r) override;
+    CB_Command_Map_Latest(Objecter *o, ceph_tid_t t) :  objecter(o), tid(t) {}
+    void operator()(boost::system::error_code err, version_t latest, version_t);
   };
 
   struct C_Stat : public Context {
@@ -1896,13 +1892,11 @@ public:
     }
   };
 
-  struct C_Linger_Map_Latest : public Context {
+  struct CB_Linger_Map_Latest {
     Objecter *objecter;
     uint64_t linger_id;
-    version_t latest;
-    C_Linger_Map_Latest(Objecter *o, uint64_t id) :
-      objecter(o), linger_id(id), latest(0) {}
-    void finish(int r) override;
+    CB_Linger_Map_Latest(Objecter *o, uint64_t id) : objecter(o), linger_id(id) {}
+    void operator()(boost::system::error_code err, version_t latest, version_t);
   };
 
   // -- osd sessions --
diff --git a/src/test/crimson/test_socket.cc b/src/test/crimson/test_socket.cc
index ec306700bd7f..557a0cf6f067 100644
--- a/src/test/crimson/test_socket.cc
+++ b/src/test/crimson/test_socket.cc
@@ -32,7 +32,7 @@ class SocketFactoryBase
  public:
   virtual ~SocketFactoryBase() = default;
 
-  virtual future<> bind_accept() {
+  virtual seastar::future<> bind_accept() {
     return this->container().invoke_on_all([] (auto& factory) {
       entity_addr_t addr;
       addr.parse(server_addr, nullptr);
@@ -77,7 +77,7 @@ class SocketFactoryBase
     });
   }
 
-  future<> shutdown() {
+  seastar::future<> shutdown() {
     return this->container().invoke_on_all([] (auto& factory) {
       if (factory.listener) {
         factory.listener.value().abort_accept();
@@ -86,22 +86,22 @@ class SocketFactoryBase
     });
   }
 
-  future<> stop() { return seastar::now(); }
+  seastar::future<> stop() { return seastar::now(); }
 
-  static future<SocketFRef> connect() {
+  static seastar::future<SocketFRef> connect() {
     entity_addr_t addr;
     addr.parse(server_addr, nullptr);
     return Socket::connect(addr);
   }
 
  protected:
-  virtual future<> handle_server_socket(SocketFRef&& socket) = 0;
+  virtual seastar::future<> handle_server_socket(SocketFRef&& socket) = 0;
 };
 
 class AcceptTest final
     : public SocketFactoryBase<AcceptTest> {
  public:
-  future<> handle_server_socket(SocketFRef&& socket) override {
+  seastar::future<> handle_server_socket(SocketFRef&& socket) override {
     return seastar::sleep(100ms
     ).then([socket = std::move(socket)] () mutable {
       return socket->close()
@@ -110,7 +110,7 @@ class AcceptTest final
   }
 };
 
-future<> test_refused() {
+seastar::future<> test_refused() {
   logger.info("test_refused()...");
   return AcceptTest::connect().discard_result(
   ).then([] {
@@ -127,7 +127,7 @@ future<> test_refused() {
   });
 }
 
-future<> test_bind_same() {
+seastar::future<> test_bind_same() {
   logger.info("test_bind_same()...");
   return crimson::net::create_sharded<AcceptTest>().then(
     [] (AcceptTest* factory) {
@@ -157,7 +157,7 @@ future<> test_bind_same() {
   });
 }
 
-future<> test_accept() {
+seastar::future<> test_accept() {
   logger.info("test_accept()");
   return crimson::net::create_sharded<AcceptTest>(
   ).then([] (AcceptTest* factory) {
@@ -187,25 +187,25 @@ class SocketFactory final
   const seastar::shard_id target_shard;
   seastar::promise<SocketFRef> socket_promise;
 
-  future<> bind_accept() override {
+  seastar::future<> bind_accept() override {
     return SocketFactoryBase<SocketFactory>::bind_accept();
   }
 
-  future<SocketFRef> get_accepted() {
+  seastar::future<SocketFRef> get_accepted() {
     return socket_promise.get_future();
   }
 
  public:
   SocketFactory(seastar::shard_id shard) : target_shard{shard} {}
 
-  future<> handle_server_socket(SocketFRef&& socket) override {
+  seastar::future<> handle_server_socket(SocketFRef&& socket) override {
     return container().invoke_on(target_shard,
         [socket = std::move(socket)] (auto& factory) mutable {
       factory.socket_promise.set_value(std::move(socket));
     });
   }
 
-  static future<tuple<SocketFRef, SocketFRef>> get_sockets() {
+static seastar::future<tuple<SocketFRef, SocketFRef>> get_sockets() {
     return crimson::net::create_sharded<SocketFactory>(seastar::engine().cpu_id()
     ).then([] (SocketFactory* factory) {
       return factory->bind_accept().then([factory] {
@@ -241,7 +241,7 @@ class Connection {
     data[DATA_SIZE - 1] = DATA_TAIL;
   }
 
-  future<> dispatch_write(unsigned round = 0, bool force_shut = false) {
+  seastar::future<> dispatch_write(unsigned round = 0, bool force_shut = false) {
     return seastar::repeat([this, round, force_shut] {
       if (round != 0 && round <= write_count) {
         return seastar::futurize_apply([this, force_shut] {
@@ -265,7 +265,7 @@ class Connection {
     });
   }
 
-  future<> dispatch_write_unbounded() {
+  seastar::future<> dispatch_write_unbounded() {
     return dispatch_write(
     ).then([] {
       ceph_abort();
@@ -282,7 +282,7 @@ class Connection {
     });
   }
 
-  future<> dispatch_read(unsigned round = 0, bool force_shut = false) {
+  seastar::future<> dispatch_read(unsigned round = 0, bool force_shut = false) {
     return seastar::repeat([this, round, force_shut] {
       if (round != 0 && round <= read_count) {
         return seastar::futurize_apply([this, force_shut] {
@@ -318,7 +318,7 @@ class Connection {
     });
   }
 
-  future<> dispatch_read_unbounded() {
+  seastar::future<> dispatch_read_unbounded() {
     return dispatch_read(
     ).then([] {
       ceph_abort();
@@ -339,12 +339,12 @@ class Connection {
     socket->shutdown();
   }
 
-  future<> close() {
+  seastar::future<> close() {
     return socket->close();
   }
 
  public:
-  static future<> dispatch_rw_bounded(SocketFRef&& socket, bool is_client,
+  static seastar::future<> dispatch_rw_bounded(SocketFRef&& socket, bool is_client,
                                       unsigned round, bool force_shut = false) {
     return seastar::smp::submit_to(is_client ? 0 : 1,
         [socket = std::move(socket), round, force_shut] () mutable {
@@ -365,7 +365,7 @@ class Connection {
     });
   }
 
-  static future<> dispatch_rw_unbounded(SocketFRef&& socket, bool is_client,
+  static seastar::future<> dispatch_rw_unbounded(SocketFRef&& socket, bool is_client,
                                         bool preemptive_shut = false) {
     return seastar::smp::submit_to(is_client ? 0 : 1,
         [socket = std::move(socket), preemptive_shut, is_client] () mutable {
@@ -393,7 +393,7 @@ class Connection {
   }
 };
 
-future<> test_read_write() {
+seastar::future<> test_read_write() {
   logger.info("test_read_write()...");
   return SocketFactory::get_sockets().then([] (auto sockets) {
     auto [client_socket, server_socket] = std::move(sockets);
@@ -407,7 +407,7 @@ future<> test_read_write() {
   });
 }
 
-future<> test_unexpected_down() {
+seastar::future<> test_unexpected_down() {
   logger.info("test_unexpected_down()...");
   return SocketFactory::get_sockets().then([] (auto sockets) {
     auto [client_socket, server_socket] = std::move(sockets);
@@ -421,7 +421,7 @@ future<> test_unexpected_down() {
   });
 }
 
-future<> test_shutdown_propagated() {
+seastar::future<> test_shutdown_propagated() {
   logger.info("test_shutdown_propagated()...");
   return SocketFactory::get_sockets().then([] (auto sockets) {
     auto [client_socket, server_socket] = std::move(sockets);
@@ -437,7 +437,7 @@ future<> test_shutdown_propagated() {
   });
 }
 
-future<> test_preemptive_down() {
+seastar::future<> test_preemptive_down() {
   logger.info("test_preemptive_down()...");
   return SocketFactory::get_sockets().then([] (auto sockets) {
     auto [client_socket, server_socket] = std::move(sockets);
diff --git a/src/test/mon/test-mon-msg.cc b/src/test/mon/test-mon-msg.cc
index ad7ab6c8e23a..4c551cab8109 100644
--- a/src/test/mon/test-mon-msg.cc
+++ b/src/test/mon/test-mon-msg.cc
@@ -20,6 +20,7 @@
 
 #include "global/global_init.h"
 #include "global/global_context.h"
+#include "common/async/context_pool.h"
 #include "common/ceph_argparse.h"
 #include "common/version.h"
 #include "common/dout.h"
@@ -51,6 +52,7 @@ class MonClientHelper : public Dispatcher
 {
 protected:
   CephContext *cct;
+  ceph::async::io_context_pool poolctx;
   Messenger *msg;
   MonClient monc;
 
@@ -63,8 +65,9 @@ public:
   explicit MonClientHelper(CephContext *cct_)
     : Dispatcher(cct_),
       cct(cct_),
+      poolctx(1),
       msg(NULL),
-      monc(cct_)
+      monc(cct_, poolctx)
   { }
 
 
diff --git a/src/test/mon/test_mon_workloadgen.cc b/src/test/mon/test_mon_workloadgen.cc
index 15792a63af58..d62b0d7104a7 100644
--- a/src/test/mon/test_mon_workloadgen.cc
+++ b/src/test/mon/test_mon_workloadgen.cc
@@ -39,6 +39,7 @@
 #include "mon/MonClient.h"
 #include "msg/Dispatcher.h"
 #include "msg/Messenger.h"
+#include "common/async/context_pool.h"
 #include "common/Timer.h"
 #include "common/ceph_argparse.h"
 #include "global/global_init.h"
@@ -82,6 +83,7 @@ class TestStub : public Dispatcher
 {
  protected:
   MessengerRef messenger;
+  ceph::async::io_context_pool poolctx;
   MonClient monc;
 
   ceph::mutex lock;
@@ -163,6 +165,7 @@ class TestStub : public Dispatcher
     monc.shutdown();
     timer.shutdown();
     messenger->shutdown();
+    poolctx.finish();
     return 0;
   }
 
@@ -177,7 +180,7 @@ class TestStub : public Dispatcher
 
   TestStub(CephContext *cct, string who)
     : Dispatcher(cct),
-      monc(cct),
+      monc(cct, poolctx),
       lock(ceph::make_mutex(who.append("::lock"))),
       timer(cct, lock),
       do_shutdown(false),
@@ -244,6 +247,7 @@ class ClientStub : public TestStub
 
   int init() override {
     int err;
+    poolctx.start(1);
     err = monc.build_initial_monmap();
     if (err < 0) {
       derr << "ClientStub::" << __func__ << " ERROR: build initial monmap: "
diff --git a/src/test/osd/TestOSDScrub.cc b/src/test/osd/TestOSDScrub.cc
index 34a3b8d81afa..810d77521bb7 100644
--- a/src/test/osd/TestOSDScrub.cc
+++ b/src/test/osd/TestOSDScrub.cc
@@ -22,6 +22,7 @@
 #include <stdio.h>
 #include <signal.h>
 #include <gtest/gtest.h>
+#include "common/async/context_pool.h"
 #include "osd/OSD.h"
 #include "os/ObjectStore.h"
 #include "mon/MonClient.h"
@@ -41,8 +42,9 @@ public:
       Messenger *hb_front_server,
       Messenger *hb_back_server,
       Messenger *osdc_messenger,
-      MonClient *mc, const std::string &dev, const std::string &jdev) :
-      OSD(cct_, store_, id, internal, external, hb_front_client, hb_back_client, hb_front_server, hb_back_server, osdc_messenger, mc, dev, jdev)
+      MonClient *mc, const std::string &dev, const std::string &jdev,
+      ceph::async::io_context_pool& ictx) :
+      OSD(cct_, store_, id, internal, external, hb_front_client, hb_back_client, hb_front_server, hb_back_server, osdc_messenger, mc, dev, jdev, ictx)
   {
   }
 
@@ -52,6 +54,7 @@ public:
 };
 
 TEST(TestOSDScrub, scrub_time_permit) {
+  ceph::async::io_context_pool icp(1);
   ObjectStore *store = ObjectStore::create(g_ceph_context,
              g_conf()->osd_objectstore,
              g_conf()->osd_data,
@@ -63,9 +66,9 @@ TEST(TestOSDScrub, scrub_time_permit) {
   ms->set_cluster_protocol(CEPH_OSD_PROTOCOL);
   ms->set_default_policy(Messenger::Policy::stateless_server(0));
   ms->bind(g_conf()->public_addr);
-  MonClient mc(g_ceph_context);
+  MonClient mc(g_ceph_context, icp);
   mc.build_initial_monmap();
-  TestOSDScrub* osd = new TestOSDScrub(g_ceph_context, store, 0, ms, ms, ms, ms, ms, ms, ms, &mc, "", "");
+  TestOSDScrub* osd = new TestOSDScrub(g_ceph_context, store, 0, ms, ms, ms, ms, ms, ms, ms, &mc, "", "", icp);
 
   g_ceph_context->_conf.set_val("osd_scrub_begin_hour", "0");
   g_ceph_context->_conf.set_val("osd_scrub_end_hour", "24");
diff --git a/src/tools/cephfs/MDSUtility.cc b/src/tools/cephfs/MDSUtility.cc
index 839b4aea0502..3e867aa735f2 100644
--- a/src/tools/cephfs/MDSUtility.cc
+++ b/src/tools/cephfs/MDSUtility.cc
@@ -25,7 +25,7 @@ MDSUtility::MDSUtility() :
   waiting_for_mds_map(NULL),
   inited(false)
 {
-  monc = new MonClient(g_ceph_context);
+  monc = new MonClient(g_ceph_context, poolctx);
   messenger = Messenger::create_client_messenger(g_ceph_context, "mds");
   fsmap = new FSMap();
   objecter = new Objecter(g_ceph_context, messenger, monc, NULL, 0, 0);
@@ -52,6 +52,7 @@ int MDSUtility::init()
   if (r < 0)
     return r;
 
+  poolctx.start(1);
   messenger->start();
 
   objecter->set_client_incarnation(0);
@@ -125,6 +126,7 @@ void MDSUtility::shutdown()
   monc->shutdown();
   messenger->shutdown();
   messenger->wait();
+  poolctx.finish();
 }
 
 
diff --git a/src/tools/cephfs/MDSUtility.h b/src/tools/cephfs/MDSUtility.h
index e5097ec48fe2..09f1918ba445 100644
--- a/src/tools/cephfs/MDSUtility.h
+++ b/src/tools/cephfs/MDSUtility.h
@@ -20,6 +20,7 @@
 #include "msg/Dispatcher.h"
 #include "msg/Messenger.h"
 #include "auth/Auth.h"
+#include "common/async/context_pool.h"
 #include "common/Finisher.h"
 #include "common/Timer.h"
 
@@ -38,6 +39,7 @@ protected:
 
   ceph::mutex lock = ceph::make_mutex("MDSUtility::lock");
   Finisher finisher;
+  ceph::async::io_context_pool poolctx;
 
   Context *waiting_for_mds_map;