librbd: avoid blocking AIO API methods

Enqueue all AIO API methods within the new librbd thread pool to
reduce the possibility of any blocking operations. To maintain
backwards compatibility with the legacy return codes of the API's
AIO methods, it's still possible to block attempting to acquire
the snap_lock.

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

diff --git a/src/librbd/librbd.cc b/src/librbd/librbd.cc
index 6d59872f30386fe6cdb55e14ef42d8785fc5d9d4..d72795df8bd82e7bc3de6e64dce1b32fb6ff7755 100644 (file)
--- a/src/librbd/librbd.cc
+++ b/src/librbd/librbd.cc
@@ -51,6 +51,87 @@ using ceph::bufferlist;
 using librados::snap_t;
 using librados::IoCtx;
 
+namespace {
+
+class C_AioReadWQ : public Context {
+public:
+  C_AioReadWQ(librbd::ImageCtx *ictx, uint64_t off, size_t len,
+              char *buf, bufferlist *pbl, librbd::AioCompletion *c,
+              int op_flags)
+    : m_ictx(ictx), m_off(off), m_len(len), m_buf(buf), m_pbl(pbl), m_comp(c),
+      m_op_flags(op_flags) {
+  }
+protected:
+  virtual void finish(int r) {
+    librbd::aio_read(m_ictx, m_off, m_len, m_buf, m_pbl, m_comp, m_op_flags);
+  }
+private:
+  librbd::ImageCtx *m_ictx;
+  uint64_t m_off;
+  uint64_t m_len;
+  char *m_buf;
+  bufferlist *m_pbl;
+  librbd::AioCompletion *m_comp;
+  int m_op_flags;
+};
+
+class C_AioWriteWQ : public Context {
+public:
+  C_AioWriteWQ(librbd::ImageCtx *ictx, uint64_t off, size_t len,
+               const char *buf, librbd::AioCompletion *c, int op_flags)
+    : m_ictx(ictx), m_off(off), m_len(len), m_buf(buf), m_comp(c),
+      m_op_flags(op_flags) {
+  }
+protected:
+  virtual void finish(int r) {
+    aio_write(m_ictx, m_off, m_len, m_buf, m_comp, m_op_flags);
+  }
+private:
+  librbd::ImageCtx *m_ictx;
+  uint64_t m_off;
+  uint64_t m_len;
+  const char *m_buf;
+  librbd::AioCompletion *m_comp;
+  int m_op_flags;
+};
+
+class C_AioDiscardWQ : public Context {
+public:
+  C_AioDiscardWQ(librbd::ImageCtx *ictx, uint64_t off, uint64_t len,
+                 librbd::AioCompletion *c)
+    : m_ictx(ictx), m_off(off), m_len(len), m_comp(c) {
+  }
+protected:
+  virtual void finish(int r) {
+    aio_discard(m_ictx, m_off, m_len, m_comp);
+  }
+private:
+  librbd::ImageCtx *m_ictx;
+  uint64_t m_off;
+  uint64_t m_len;
+  librbd::AioCompletion *m_comp;
+};
+
+class C_AioFlushWQ : public Context {
+public:
+  C_AioFlushWQ(librbd::ImageCtx *ictx, librbd::AioCompletion *c)
+    : m_ictx(ictx), m_comp(c) {
+  }
+protected:
+  virtual void finish(int r) {
+    aio_flush(m_ictx, m_comp);
+  }
+private:
+  librbd::ImageCtx *m_ictx;
+  librbd::AioCompletion *m_comp;
+};
+
+librbd::AioCompletion* get_aio_completion(librbd::RBD::AioCompletion *comp) {
+  return reinterpret_cast<librbd::AioCompletion *>(comp->pc);
+}
+
+} // anonymous namespace
+
 namespace librbd {
   ProgressContext::~ProgressContext()
   {
@@ -751,10 +832,10 @@ namespace librbd {
       tracepoint(librbd, aio_write_exit, -EINVAL);
       return -EINVAL;
     }
-    int r = librbd::aio_write(ictx, off, len, bl.c_str(),
-                            (librbd::AioCompletion *)c->pc, 0);
-    tracepoint(librbd, aio_write_exit, r);
-    return r;
+    ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, bl.c_str(),
+                                                 get_aio_completion(c), 0));
+    tracepoint(librbd, aio_write_exit, 0);
+    return 0;
   }
 
   int Image::aio_write2(uint64_t off, size_t len, bufferlist& bl,
@@ -767,19 +848,21 @@ namespace librbd {
       tracepoint(librbd, aio_write_exit, -EINVAL);
       return -EINVAL;
     }
-    int r = librbd::aio_write(ictx, off, len, bl.c_str(),
-                            (librbd::AioCompletion *)c->pc, op_flags);
-    tracepoint(librbd, aio_write_exit, r);
-    return r;
+    ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, bl.c_str(),
+                                                 get_aio_completion(c),
+                                                 op_flags));
+    tracepoint(librbd, aio_write_exit, 0);
+    return 0;
   }
 
   int Image::aio_discard(uint64_t off, uint64_t len, RBD::AioCompletion *c)
   {
     ImageCtx *ictx = (ImageCtx *)ctx;
     tracepoint(librbd, aio_discard_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, off, len, c->pc);
-    int r = librbd::aio_discard(ictx, off, len, (librbd::AioCompletion *)c->pc);
-    tracepoint(librbd, aio_discard_exit, r);
-    return r;
+    ictx->aio_work_queue->queue(new C_AioDiscardWQ(ictx, off, len,
+                                                   get_aio_completion(c)));
+    tracepoint(librbd, aio_discard_exit, 0);
+    return 0;
   }
 
   int Image::aio_read(uint64_t off, size_t len, bufferlist& bl,
@@ -789,9 +872,10 @@ namespace librbd {
     tracepoint(librbd, aio_read_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, off, len, bl.c_str(), c->pc);
     ldout(ictx->cct, 10) << "Image::aio_read() buf=" << (void *)bl.c_str() << "~"
                         << (void *)(bl.c_str() + len - 1) << dendl;
-    int r = librbd::aio_read(ictx, off, len, NULL, &bl, (librbd::AioCompletion *)c->pc, 0);
-    tracepoint(librbd, aio_read_exit, r);
-    return r;
+    ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, NULL, &bl,
+                                                get_aio_completion(c), 0));
+    tracepoint(librbd, aio_read_exit, 0);
+    return 0;
   }
 
   int Image::aio_read2(uint64_t off, size_t len, bufferlist& bl,
@@ -802,9 +886,11 @@ namespace librbd {
                ictx->read_only, off, len, bl.c_str(), c->pc, op_flags);
     ldout(ictx->cct, 10) << "Image::aio_read() buf=" << (void *)bl.c_str() << "~"
                         << (void *)(bl.c_str() + len - 1) << dendl;
-    int r = librbd::aio_read(ictx, off, len, NULL, &bl, (librbd::AioCompletion *)c->pc, op_flags);
-    tracepoint(librbd, aio_read_exit, r);
-    return r;
+    ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, NULL, &bl,
+                                                get_aio_completion(c),
+                                               op_flags));
+    tracepoint(librbd, aio_read_exit, 0);
+    return 0;
   }
 
   int Image::flush()
@@ -820,9 +906,9 @@ namespace librbd {
   {
     ImageCtx *ictx = (ImageCtx *)ctx;
     tracepoint(librbd, aio_flush_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, c->pc);
-    int r = librbd::aio_flush(ictx, (librbd::AioCompletion *)c->pc);
-    tracepoint(librbd, aio_flush_exit, r);
-    return r;
+    ictx->aio_work_queue->queue(new C_AioFlushWQ(ictx, get_aio_completion(c)));
+    tracepoint(librbd, aio_flush_exit, 0);
+    return 0;
   }
 
   int Image::invalidate_cache()
@@ -1738,10 +1824,10 @@ extern "C" int rbd_aio_write(rbd_image_t image, uint64_t off, size_t len,
   librbd::ImageCtx *ictx = (librbd::ImageCtx *)image;
   librbd::RBD::AioCompletion *comp = (librbd::RBD::AioCompletion *)c;
   tracepoint(librbd, aio_write_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, off, len, buf, comp->pc);
-  int r = librbd::aio_write(ictx, off, len, buf,
-                          (librbd::AioCompletion *)comp->pc, 0);
-  tracepoint(librbd, aio_write_exit, r);
-  return r;
+  ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, buf,
+                                               get_aio_completion(comp), 0));
+  tracepoint(librbd, aio_write_exit, 0);
+  return 0;
 }
 
 extern "C" int rbd_aio_write2(rbd_image_t image, uint64_t off, size_t len,
@@ -1751,10 +1837,11 @@ extern "C" int rbd_aio_write2(rbd_image_t image, uint64_t off, size_t len,
   librbd::RBD::AioCompletion *comp = (librbd::RBD::AioCompletion *)c;
   tracepoint(librbd, aio_write2_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(),
              ictx->read_only, off, len, buf, comp->pc, op_flags);
-  int r = librbd::aio_write(ictx, off, len, buf,
-                          (librbd::AioCompletion *)comp->pc, op_flags);
-  tracepoint(librbd, aio_write_exit, r);
-  return r;
+  ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, buf,
+                                               get_aio_completion(comp),
+                                               op_flags));
+  tracepoint(librbd, aio_write_exit, 0);
+  return 0;
 }
 
 
@@ -1764,9 +1851,10 @@ extern "C" int rbd_aio_discard(rbd_image_t image, uint64_t off, uint64_t len,
   librbd::ImageCtx *ictx = (librbd::ImageCtx *)image;
   librbd::RBD::AioCompletion *comp = (librbd::RBD::AioCompletion *)c;
   tracepoint(librbd, aio_discard_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, off, len, comp->pc);
-  int r = librbd::aio_discard(ictx, off, len, (librbd::AioCompletion *)comp->pc);
-  tracepoint(librbd, aio_discard_exit, r);
-  return r;
+  ictx->aio_work_queue->queue(new C_AioDiscardWQ(ictx, off, len,
+                                                 get_aio_completion(comp)));
+  tracepoint(librbd, aio_discard_exit, 0);
+  return 0;
 }
 
 extern "C" int rbd_aio_read(rbd_image_t image, uint64_t off, size_t len,
@@ -1775,10 +1863,10 @@ extern "C" int rbd_aio_read(rbd_image_t image, uint64_t off, size_t len,
   librbd::ImageCtx *ictx = (librbd::ImageCtx *)image;
   librbd::RBD::AioCompletion *comp = (librbd::RBD::AioCompletion *)c;
   tracepoint(librbd, aio_read_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, off, len, buf, comp->pc);
-  int r = librbd::aio_read(ictx, off, len, buf, NULL,
-                         (librbd::AioCompletion *)comp->pc, 0);
-  tracepoint(librbd, aio_read_exit, r);
-  return r;
+  ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, buf, NULL,
+                                              get_aio_completion(comp), 0));
+  tracepoint(librbd, aio_read_exit, 0);
+  return 0;
 }
 
 extern "C" int rbd_aio_read2(rbd_image_t image, uint64_t off, size_t len,
@@ -1788,10 +1876,11 @@ extern "C" int rbd_aio_read2(rbd_image_t image, uint64_t off, size_t len,
   librbd::RBD::AioCompletion *comp = (librbd::RBD::AioCompletion *)c;
   tracepoint(librbd, aio_read2_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(),
              ictx->read_only, off, len, buf, comp->pc, op_flags);
-  int r = librbd::aio_read(ictx, off, len, buf, NULL,
-                         (librbd::AioCompletion *)comp->pc, op_flags);
-  tracepoint(librbd, aio_read_exit, r);
-  return r;
+  ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, buf, NULL,
+                                              get_aio_completion(comp),
+                                             op_flags));
+  tracepoint(librbd, aio_read_exit, 0);
+  return 0;
 }
 
 extern "C" int rbd_flush(rbd_image_t image)
@@ -1808,9 +1897,9 @@ extern "C" int rbd_aio_flush(rbd_image_t image, rbd_completion_t c)
   librbd::ImageCtx *ictx = (librbd::ImageCtx *)image;
   librbd::RBD::AioCompletion *comp = (librbd::RBD::AioCompletion *)c;
   tracepoint(librbd, aio_flush_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only, comp->pc);
-  int r = librbd::aio_flush(ictx, (librbd::AioCompletion *)comp->pc);
-  tracepoint(librbd, aio_flush_exit, r);
-  return r;
+  ictx->aio_work_queue->queue(new C_AioFlushWQ(ictx, get_aio_completion(comp)));
+  tracepoint(librbd, aio_flush_exit, 0);
+  return 0;
 }
 
 extern "C" int rbd_invalidate_cache(rbd_image_t image)