[DNM] librbd: avoid blocking AIO API methods

PendingReleaseNotes

@@ -1,4 +1,11 @@
 
+v0.80.10
+--------
+
+-  The return code for librbd's rbd_aio_read and Image::aio_read API methods no
+  longer returns the number of bytes read upon success.  Instead, it returns 0
+  upon success and a negative value upon failure.
+
 v0.81
 -----
 

src/common/Throttle.cc

@@ -267,6 +267,12 @@ void SimpleThrottle::end_op(int r)
   m_cond.Signal();
 }
 
+bool SimpleThrottle::pending_error() const
+{
+  Mutex::Locker l(m_lock);
+  return (m_ret < 0);
+}
+
 int SimpleThrottle::wait_for_ret()
 {
   Mutex::Locker l(m_lock);

src/common/Throttle.h

@@ -76,9 +76,10 @@ class SimpleThrottle {
   ~SimpleThrottle();
   void start_op();
   void end_op(int r);
+  bool pending_error() const;
   int wait_for_ret();
 private:
-  Mutex m_lock;
+  mutable Mutex m_lock;
   Cond m_cond;
   uint64_t m_max;
   uint64_t m_current;

src/common/WorkQueue.h

@@ -433,4 +433,40 @@ class C_QueueInWQ : public Context {
   }
 };
 
+class ContextWQ : public ThreadPool::WorkQueue<Context> {
+public:
+  ContextWQ(const string &name, time_t ti, ThreadPool *tp)
+    : ThreadPool::WorkQueue<Context>(name, ti, 0, tp) {}
+
+  virtual bool _enqueue(Context *c) {
+    _queue.push_back(c);
+    return true;
+  }
+  virtual void _enqueue_front(Context *c) {
+    _queue.push_front(c);
+  }
+  virtual void _clear() {
+    _queue.clear();
+  }
+  virtual bool _empty() {
+    return _queue.empty();
+  }
+  virtual void _dequeue(Context *c) {
+    _queue.remove(c);
+  }
+  virtual Context *_dequeue() {
+    if (_queue.empty()) {
+      return NULL;
+    }
+    Context *c = _queue.front();
+    _queue.pop_front();
+    return c;
+  }
+  virtual void _process(Context *c) {
+    c->complete(0);
+  }
+private:
+  list<Context*> _queue;
+};
+
 #endif

src/common/config_opts.h

@@ -727,6 +727,8 @@ OPTION(journal_ignore_corruption, OPT_BOOL, false) // assume journal is not corr
 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, false) // whether to enable caching (writeback unless rbd_cache_max_dirty is 0)
 OPTION(rbd_cache_writethrough_until_flush, OPT_BOOL, false) // 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

src/librbd/AioCompletion.cc

@@ -5,6 +5,7 @@
 
 #include "common/ceph_context.h"
 #include "common/dout.h"
+#include "common/errno.h"
 
 #include "librbd/AioRequest.h"
 #include "librbd/internal.h"
@@ -25,7 +26,7 @@ namespace librbd {
     building = false;
     if (!pending_count) {
       finalize(cct, rval);
-      complete();
+      complete(cct);
     }
     lock.Unlock();
   }
@@ -54,6 +55,17 @@ namespace librbd {
     }
   }
 
+  void AioCompletion::fail(CephContext *cct, int r)
+  {
+    lderr(cct) << "AioCompletion::fail() " << this << ": " << cpp_strerror(r)
+               << dendl;
+    lock.Lock();
+    assert(pending_count == 0);
+    rval = r;
+    complete(cct);
+    put_unlock();
+  }
+
   void AioCompletion::complete_request(CephContext *cct, ssize_t r)
   {
     ldout(cct, 20) << "AioCompletion::complete_request() "
@@ -70,7 +82,7 @@ namespace librbd {
     int count = --pending_count;
     if (!count && !building) {
       finalize(cct, rval);
-      complete();
+      complete(cct);
     }
     put_unlock();
   }

src/librbd/AioCompletion.h

@@ -101,10 +101,10 @@ namespace librbd {
       start_time = ceph_clock_now(ictx->cct);
     }
 
-    void complete() {
+    void complete(CephContext *cct) {
       utime_t elapsed;
       assert(lock.is_locked());
-      elapsed = ceph_clock_now(ictx->cct) - start_time;
+      elapsed = ceph_clock_now(cct) - start_time;
       switch (aio_type) {
       case AIO_TYPE_READ:
 	ictx->perfcounter->tinc(l_librbd_aio_rd_latency, elapsed); break;
@@ -115,11 +115,11 @@ namespace librbd {
       case AIO_TYPE_FLUSH:
 	ictx->perfcounter->tinc(l_librbd_aio_flush_latency, elapsed); break;
       default:
-	lderr(ictx->cct) << "completed invalid aio_type: " << aio_type << dendl;
+	lderr(cct) << "completed invalid aio_type: " << aio_type << dendl;
 	break;
       }
 
-      {
+      if (ictx != NULL) {
         Mutex::Locker l(ictx->aio_lock);
         assert(ictx->pending_aio != 0);
         --ictx->pending_aio;
@@ -132,6 +132,7 @@ namespace librbd {
       done = true;
       cond.Signal();
     }
+    void fail(CephContext *cct, int r);
 
     void set_complete_cb(void *cb_arg, callback_t cb) {
       complete_cb = cb;

src/librbd/ImageCtx.cc

@@ -53,7 +53,11 @@ namespace librbd {
       id(image_id), parent(NULL),
       stripe_unit(0), stripe_count(0),
       object_cacher(NULL), writeback_handler(NULL), object_set(NULL),
-      pending_aio(0)
+      pending_aio(0),
+      thread_pool(cct, "librbd::thread_pool", cct->_conf->rbd_op_threads,
+                  "rbd_op_threads"),
+      aio_work_queue("librbd::aio_work_queue", cct->_conf->rbd_op_thread_timeout,
+                     &thread_pool)
   {
     md_ctx.dup(p);
     data_ctx.dup(p);
@@ -98,6 +102,8 @@ namespace librbd {
       object_set->return_enoent = true;
       object_cacher->start();
     }
+
+    thread_pool.start();
   }
 
   ImageCtx::~ImageCtx() {

src/librbd/ImageCtx.h

@@ -14,6 +14,7 @@
 #include "common/Mutex.h"
 #include "common/RWLock.h"
 #include "common/snap_types.h"
+#include "common/WorkQueue.h"
 #include "include/buffer.h"
 #include "include/rbd/librbd.hpp"
 #include "include/rbd_types.h"
@@ -95,6 +96,9 @@ namespace librbd {
     Cond pending_aio_cond;
     uint64_t pending_aio;
 
+    ThreadPool thread_pool;
+    ContextWQ aio_work_queue;
+
     /**
      * Either image_name or image_id must be set.
      * If id is not known, pass the empty std::string,