From: Jason Dillaman <dillaman@redhat.com>
Date: Thu, 11 Jun 2020 17:58:35 +0000 (-0400)
Subject: librbd: simple ASIO execution context
X-Git-Tag: wip-pdonnell-testing-20200918.022351~998^2~4
X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=ac74b29b529d6198a40108ed665a9d0d3c8758e0;p=ceph-ci.git

librbd: simple ASIO execution context

This initial execution context wrapper will start "rbd_op_threads"
number of threads and bind them to an ASIO io_context to be used for
processing async callbacks.

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

diff --git a/src/librbd/AsioEngine.cc b/src/librbd/AsioEngine.cc
new file mode 100644
index 00000000000..ef1ec785409
--- /dev/null
+++ b/src/librbd/AsioEngine.cc
@@ -0,0 +1,52 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/AsioEngine.h"
+#include "common/dout.h"
+#include <boost/system/error_code.hpp>
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::AsioEngine: " \
+                           << this << " " << __func__ << ": "
+
+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);
+    });
+  }
+}
+
+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();
+
+  ldout(m_cct, 5) << "done" << dendl;
+}
+
+} // namespace librbd
diff --git a/src/librbd/AsioEngine.h b/src/librbd/AsioEngine.h
new file mode 100644
index 00000000000..5d530a9e37a
--- /dev/null
+++ b/src/librbd/AsioEngine.h
@@ -0,0 +1,44 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_ASIO_ENGINE_H
+#define CEPH_LIBRBD_ASIO_ENGINE_H
+
+#include "include/common_fwd.h"
+#include <optional>
+#include <thread>
+#include <vector>
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+
+namespace librbd {
+
+class AsioEngine {
+public:
+  explicit AsioEngine(CephContext* cct);
+  ~AsioEngine();
+
+  inline boost::asio::io_context& get_io_context() {
+    return m_io_context;
+  }
+
+private:
+  typedef std::vector<std::thread> Threads;
+
+  typedef boost::asio::executor_work_guard<
+    boost::asio::io_context::executor_type> WorkGuard;
+
+  CephContext* m_cct;
+  Threads m_threads;
+
+  boost::asio::io_context m_io_context;
+  std::optional<WorkGuard> m_work_guard;
+
+  void init();
+  void shut_down();
+
+};
+
+} // namespace librbd
+
+#endif // CEPH_LIBRBD_ASIO_ENGINE_H
diff --git a/src/librbd/CMakeLists.txt b/src/librbd/CMakeLists.txt
index 5f43300c7bc..0bbd73396eb 100644
--- a/src/librbd/CMakeLists.txt
+++ b/src/librbd/CMakeLists.txt
@@ -6,6 +6,7 @@ add_library(rbd_types STATIC
   WatchNotifyTypes.cc)
 
 set(librbd_internal_srcs
+  AsioEngine.cc
   AsyncObjectThrottle.cc
   AsyncRequest.cc
   DeepCopyRequest.cc
diff --git a/src/librbd/ImageCtx.cc b/src/librbd/ImageCtx.cc
index 8d71a8e061c..f8564005d88 100644
--- a/src/librbd/ImageCtx.cc
+++ b/src/librbd/ImageCtx.cc
@@ -11,6 +11,7 @@
 #include "common/WorkQueue.h"
 #include "common/Timer.h"
 
+#include "librbd/AsioEngine.h"
 #include "librbd/AsyncRequest.h"
 #include "librbd/ExclusiveLock.h"
 #include "librbd/internal.h"
@@ -90,6 +91,11 @@ 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_";
@@ -123,6 +129,7 @@ public:
       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),
       plugin_registry(new PluginRegistry<ImageCtx>(this)),
       external_callback_completions(32),
@@ -913,6 +920,11 @@ public:
     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_thread_pool_instance(CephContext *cct,
                                           ThreadPool **thread_pool,
                                           ContextWQ **op_work_queue) {
diff --git a/src/librbd/ImageCtx.h b/src/librbd/ImageCtx.h
index b2b15b7455c..d25f80464ac 100644
--- a/src/librbd/ImageCtx.h
+++ b/src/librbd/ImageCtx.h
@@ -32,6 +32,7 @@
 #include "librbd/AsyncRequest.h"
 #include "librbd/Types.h"
 
+#include <boost/asio/io_context.hpp>
 #include <boost/lockfree/policies.hpp>
 #include <boost/lockfree/queue.hpp>
 
@@ -42,6 +43,7 @@ class SafeTimer;
 
 namespace librbd {
 
+  struct AsioEngine;
   template <typename> class ExclusiveLock;
   template <typename> class ImageState;
   template <typename> class ImageWatcher;
@@ -181,6 +183,8 @@ 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;
 
@@ -342,6 +346,7 @@ 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_thread_pool_instance(CephContext *cct,
                                          ThreadPool **thread_pool,
                                          ContextWQ **op_work_queue);