librbd: new rbd_non_blocking_aio config option

Setting this option to false reverts librbd to legacy behavior
where AIO operations could potentially block.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>
(cherry picked from commit 769cad12716b85d87eacc1069dd9f5c21cad3915)

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 381d8fa487713c98d07cb2a108dbb8bdac3691a4..aee900d4bb03feb68fc94887c591d0cea8e36be9 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -860,6 +860,7 @@ OPTION(rados_osd_op_timeout, OPT_DOUBLE, 0) // how many seconds to wait for a re
 
 OPTION(rbd_op_threads, OPT_INT, 1)
 OPTION(rbd_op_thread_timeout, OPT_INT, 60)
+OPTION(rbd_non_blocking_aio, OPT_BOOL, true) // process AIO ops from a worker thread to prevent blocking
 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/librbd.cc b/src/librbd/librbd.cc
index 80d3faedfedd4b0436e0aa1d0b49b8e1b2b5d9f4..a99f7d4725c84e766344693315b72531e79e59aa 100644 (file)
--- a/src/librbd/librbd.cc
+++ b/src/librbd/librbd.cc
@@ -85,7 +85,7 @@ public:
   }
 protected:
   virtual void finish(int r) {
-    aio_write(m_ictx, m_off, m_len, m_buf, m_comp, m_op_flags);
+    librbd::aio_write(m_ictx, m_off, m_len, m_buf, m_comp, m_op_flags);
   }
 private:
   librbd::ImageCtx *m_ictx;
@@ -104,7 +104,7 @@ public:
   }
 protected:
   virtual void finish(int r) {
-    aio_discard(m_ictx, m_off, m_len, m_comp);
+    librbd::aio_discard(m_ictx, m_off, m_len, m_comp);
   }
 private:
   librbd::ImageCtx *m_ictx;
@@ -120,13 +120,51 @@ public:
   }
 protected:
   virtual void finish(int r) {
-    aio_flush(m_ictx, m_comp);
+    librbd::aio_flush(m_ictx, m_comp);
   }
 private:
   librbd::ImageCtx *m_ictx;
   librbd::AioCompletion *m_comp;
 };
 
+void submit_aio_read(librbd::ImageCtx *ictx, uint64_t off, size_t len,
+                     char *buf, bufferlist *pbl, librbd::AioCompletion *c,
+                     int op_flags) {
+  if (ictx->cct->_conf->rbd_non_blocking_aio) {
+    ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, buf, pbl, c,
+                                                op_flags));
+  } else {
+    librbd::aio_read(ictx, off, len, buf, pbl, c, op_flags);
+  }
+}
+
+void submit_aio_write(librbd::ImageCtx *ictx, uint64_t off, size_t len,
+                      const char *buf, librbd::AioCompletion *c, int op_flags) {
+  if (ictx->cct->_conf->rbd_non_blocking_aio) {
+    ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, buf, c,
+                                                 op_flags));
+  } else {
+    librbd::aio_write(ictx, off, len, buf, c, op_flags);
+  }
+}
+
+void submit_aio_discard(librbd::ImageCtx *ictx, uint64_t off, uint64_t len,
+                        librbd::AioCompletion *c) {
+  if (ictx->cct->_conf->rbd_non_blocking_aio) {
+    ictx->aio_work_queue->queue(new C_AioDiscardWQ(ictx, off, len, c));
+  } else {
+    librbd::aio_discard(ictx, off, len, c);
+  }
+}
+
+void submit_aio_flush(librbd::ImageCtx *ictx, librbd::AioCompletion *c) {
+  if (ictx->cct->_conf->rbd_non_blocking_aio) {
+    ictx->aio_work_queue->queue(new C_AioFlushWQ(ictx, c));
+  } else {
+    librbd::aio_flush(ictx, c);
+  }
+}
+
 librbd::AioCompletion* get_aio_completion(librbd::RBD::AioCompletion *comp) {
   return reinterpret_cast<librbd::AioCompletion *>(comp->pc);
 }
@@ -791,8 +829,7 @@ namespace librbd {
       tracepoint(librbd, aio_write_exit, -EINVAL);
       return -EINVAL;
     }
-    ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, bl.c_str(),
-                                                 get_aio_completion(c), 0));
+    submit_aio_write(ictx, off, len, bl.c_str(), get_aio_completion(c), 0);
     tracepoint(librbd, aio_write_exit, 0);
     return 0;
   }
@@ -807,9 +844,8 @@ namespace librbd {
       tracepoint(librbd, aio_write_exit, -EINVAL);
       return -EINVAL;
     }
-    ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, bl.c_str(),
-                                                 get_aio_completion(c),
-                                                 op_flags));
+    submit_aio_write(ictx, off, len, bl.c_str(), get_aio_completion(c),
+                     op_flags);
     tracepoint(librbd, aio_write_exit, 0);
     return 0;
   }
@@ -818,8 +854,7 @@ namespace librbd {
   {
     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);
-    ictx->aio_work_queue->queue(new C_AioDiscardWQ(ictx, off, len,
-                                                   get_aio_completion(c)));
+    submit_aio_discard(ictx, off, len, get_aio_completion(c));
     tracepoint(librbd, aio_discard_exit, 0);
     return 0;
   }
@@ -831,8 +866,7 @@ 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;
-    ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, NULL, &bl,
-                                                get_aio_completion(c), 0));
+    submit_aio_read(ictx, off, len, NULL, &bl, get_aio_completion(c), 0);
     tracepoint(librbd, aio_read_exit, 0);
     return 0;
   }
@@ -845,9 +879,7 @@ 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;
-    ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, NULL, &bl,
-                                                get_aio_completion(c),
-                                               op_flags));
+    submit_aio_read(ictx, off, len, NULL, &bl, get_aio_completion(c), op_flags);
     tracepoint(librbd, aio_read_exit, 0);
     return 0;
   }
@@ -865,7 +897,7 @@ 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);
-    ictx->aio_work_queue->queue(new C_AioFlushWQ(ictx, get_aio_completion(c)));
+    submit_aio_flush(ictx, get_aio_completion(c));
     tracepoint(librbd, aio_flush_exit, 0);
     return 0;
   }
@@ -1699,8 +1731,7 @@ 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);
-  ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, buf,
-                                               get_aio_completion(comp), 0));
+  submit_aio_write(ictx, off, len, buf, get_aio_completion(comp), 0);
   tracepoint(librbd, aio_write_exit, 0);
   return 0;
 }
@@ -1712,9 +1743,7 @@ 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);
-  ictx->aio_work_queue->queue(new C_AioWriteWQ(ictx, off, len, buf,
-                                               get_aio_completion(comp),
-                                               op_flags));
+  submit_aio_write(ictx, off, len, buf, get_aio_completion(comp), op_flags);
   tracepoint(librbd, aio_write_exit, 0);
   return 0;
 }
@@ -1726,8 +1755,7 @@ 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);
-  ictx->aio_work_queue->queue(new C_AioDiscardWQ(ictx, off, len,
-                                                 get_aio_completion(comp)));
+  submit_aio_discard(ictx, off, len, get_aio_completion(comp));
   tracepoint(librbd, aio_discard_exit, 0);
   return 0;
 }
@@ -1738,8 +1766,7 @@ 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);
-  ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, buf, NULL,
-                                              get_aio_completion(comp), 0));
+  submit_aio_read(ictx, off, len, buf, NULL, get_aio_completion(comp), 0);
   tracepoint(librbd, aio_read_exit, 0);
   return 0;
 }
@@ -1751,9 +1778,8 @@ 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);
-  ictx->aio_work_queue->queue(new C_AioReadWQ(ictx, off, len, buf, NULL,
-                                              get_aio_completion(comp),
-                                             op_flags));
+  submit_aio_read(ictx, off, len, buf, NULL, get_aio_completion(comp),
+                  op_flags);
   tracepoint(librbd, aio_read_exit, 0);
   return 0;
 }
@@ -1772,7 +1798,7 @@ 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);
-  ictx->aio_work_queue->queue(new C_AioFlushWQ(ictx, get_aio_completion(comp)));
+  submit_aio_flush(ictx, get_aio_completion(comp));
   tracepoint(librbd, aio_flush_exit, 0);
   return 0;
 }