From: Jason Dillaman <dillaman@redhat.com>
Date: Wed, 8 Apr 2015 21:24:08 +0000 (-0400)
Subject: librbd: add task pool / work queue for requests / callbacks
X-Git-Tag: v9.0.2~116^2~7
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=afb896d91f886b647baf38f7ec94cc3739f6d2a9;p=ceph.git

librbd: add task pool / work queue for requests / callbacks

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

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index e79eeaab801f..026e5a310101 100644
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -869,6 +869,8 @@ OPTION(journal_discard, OPT_BOOL, false) //using ssd disk as journal, whether su
 OPTION(rados_mon_op_timeout, OPT_DOUBLE, 0) // how many seconds to wait for a response from the monitor before returning an error from a rados operation. 0 means on limit.
 OPTION(rados_osd_op_timeout, OPT_DOUBLE, 0) // how many seconds to wait for a response from osds before returning an error from a rados operation. 0 means no limit.
 
+OPTION(rbd_op_threads, OPT_INT, 1)
+OPTION(rbd_op_thread_timeout, OPT_INT, 60)
 OPTION(rbd_cache, OPT_BOOL, true) // whether to enable caching (writeback unless rbd_cache_max_dirty is 0)
 OPTION(rbd_cache_writethrough_until_flush, OPT_BOOL, true) // whether to make writeback caching writethrough until flush is called, to be sure the user of librbd will send flushs so that writeback is safe
 OPTION(rbd_cache_size, OPT_LONGLONG, 32<<20)         // cache size in bytes
diff --git a/src/librbd/ImageCtx.cc b/src/librbd/ImageCtx.cc
index 46ea1c0066cc..39ae8e7bd2d6 100644
--- a/src/librbd/ImageCtx.cc
+++ b/src/librbd/ImageCtx.cc
@@ -34,6 +34,23 @@ using librados::snap_t;
 using librados::IoCtx;
 
 namespace librbd {
+
+namespace {
+
+class ThreadPoolSingleton : public ThreadPool {
+public:
+  ThreadPoolSingleton(CephContext *cct)
+    : ThreadPool(cct, "librbd::thread_pool", cct->_conf->rbd_op_threads,
+                 "rbd_op_threads") {
+    start();
+  }
+  virtual ~ThreadPoolSingleton() {
+    stop();
+  }
+};
+
+} // anonymous namespace
+
   const string ImageCtx::METADATA_CONF_PREFIX = "conf_";
 
   ImageCtx::ImageCtx(const string &image_name, const string &image_id,
@@ -67,7 +84,7 @@ namespace librbd {
       object_cacher(NULL), writeback_handler(NULL), object_set(NULL),
       readahead(),
       total_bytes_read(0), copyup_finisher(NULL),
-      object_map(*this)
+      object_map(*this), aio_work_queue(NULL)
   {
     md_ctx.dup(p);
     data_ctx.dup(p);
@@ -76,6 +93,13 @@ namespace librbd {
 
     memset(&header, 0, sizeof(header));
     memset(&layout, 0, sizeof(layout));
+
+    ThreadPoolSingleton *thread_pool_singleton;
+    cct->lookup_or_create_singleton_object<ThreadPoolSingleton>(
+      thread_pool_singleton, "librbd::thread_pool");
+    aio_work_queue = new ContextWQ("librbd::aio_work_queue",
+                                   cct->_conf->rbd_op_thread_timeout,
+                                   thread_pool_singleton);
   }
 
   ImageCtx::~ImageCtx() {
@@ -97,6 +121,8 @@ namespace librbd {
       copyup_finisher = NULL;
     }
     delete[] format_string;
+
+    delete aio_work_queue;
   }
 
   int ImageCtx::init() {
diff --git a/src/librbd/ImageCtx.h b/src/librbd/ImageCtx.h
index 1e48e02c46f4..16d99e67197a 100644
--- a/src/librbd/ImageCtx.h
+++ b/src/librbd/ImageCtx.h
@@ -16,6 +16,7 @@
 #include "common/Readahead.h"
 #include "common/RWLock.h"
 #include "common/snap_types.h"
+#include "common/WorkQueue.h"
 #include "include/atomic.h"
 #include "include/buffer.h"
 #include "include/rbd/librbd.hpp"
@@ -130,6 +131,8 @@ namespace librbd {
 
     xlist<AsyncResizeRequest*> async_resize_reqs;
 
+    ContextWQ *aio_work_queue;
+
     // Configuration
     static const string METADATA_CONF_PREFIX;
     bool cache;
diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index 94dbb73ff1ba..fbe9e5a58d32 100644
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -2827,6 +2827,8 @@ reprotect_and_return_err:
       }
     }
 
+    ictx->aio_work_queue->drain();
+
     ictx->cancel_async_requests();
     ictx->readahead.wait_for_pending();
     if (ictx->object_cacher) {