librbd: sync flush should re-use existing async flush logic

qa/workunits/rbd/journal.sh

@@ -96,16 +96,17 @@ test_rbd_journal()
     rbd snap create ${image1}@test
     restore_commit_position ${journal1}
     # check that commit position is properly updated: the journal should contain
-    # 12 entries (10 AioWrite + 1 SnapCreate + 1 OpFinish) and commit
-    # position set to tid=11
+    # 14 entries (2 AioFlush + 10 AioWrite + 1 SnapCreate + 1 OpFinish) and
+    # commit position set to tid=14
     rbd journal inspect --image ${image1} --verbose | awk '
+      /AioFlush/          {a++}         # match: "event_type": "AioFlush",
       /AioWrite/          {w++}         # match: "event_type": "AioWrite",
       /SnapCreate/        {s++}         # match: "event_type": "SnapCreate",
       /OpFinish/          {f++}         # match: "event_type": "OpFinish",
-      /entries inspected/ {t=$1; e=$4}  # match: 12 entries inspected, 0 errors
+      /entries inspected/ {t=$1; e=$4}  # match: 14 entries inspected, 0 errors
                           {print}       # for diagnostic
       END                 {
-        if (w != 10 || s != 1 || f != 1 || t != 12 || e != 0) exit(1)
+        if (a != 2 || w != 10 || s != 1 || f != 1 || t != 14 || e != 0) exit(1)
       }
     '
 

src/librbd/ImageCtx.cc

@@ -374,8 +374,7 @@ struct C_InvalidateCache : public Context {
     plb.add_u64_counter(l_librbd_discard_bytes, "discard_bytes", "Discarded data");
     plb.add_time_avg(l_librbd_discard_latency, "discard_latency", "Discard latency");
     plb.add_u64_counter(l_librbd_flush, "flush", "Flushes");
-    plb.add_u64_counter(l_librbd_aio_flush, "aio_flush", "Async flushes");
-    plb.add_time_avg(l_librbd_aio_flush_latency, "aio_flush_latency", "Latency of async flushes");
+    plb.add_time_avg(l_librbd_flush_latency, "flush_latency", "Latency of flushes");
     plb.add_u64_counter(l_librbd_ws, "ws", "WriteSames");
     plb.add_u64_counter(l_librbd_ws_bytes, "ws_bytes", "WriteSame data");
     plb.add_time_avg(l_librbd_ws_latency, "ws_latency", "WriteSame latency");

src/librbd/LibrbdAdminSocketHook.cc

@@ -6,6 +6,7 @@
 #include "librbd/ImageCtx.h"
 #include "librbd/LibrbdAdminSocketHook.h"
 #include "librbd/internal.h"
+#include "librbd/io/ImageRequestWQ.h"
 
 #define dout_subsys ceph_subsys_rbd
 #undef dout_prefix
@@ -24,7 +25,7 @@ class FlushCacheCommand : public LibrbdAdminSocketCommand {
   explicit FlushCacheCommand(ImageCtx *ictx) : ictx(ictx) {}
 
   bool call(stringstream *ss) override {
-    int r = flush(ictx);
+    int r = ictx->io_work_queue->flush();
     if (r < 0) {
       *ss << "flush: " << cpp_strerror(r);
       return false;

src/librbd/internal.cc

@@ -2221,28 +2221,6 @@ int validate_pool(IoCtx &io_ctx, CephContext *cct) {
     return 0;
   }
 
-  int flush(ImageCtx *ictx)
-  {
-    CephContext *cct = ictx->cct;
-    ldout(cct, 20) << "flush " << ictx << dendl;
-
-    int r = ictx->state->refresh_if_required();
-    if (r < 0) {
-      return r;
-    }
-
-    ictx->user_flushed();
-    C_SaferCond ctx;
-    {
-      RWLock::RLocker owner_locker(ictx->owner_lock);
-      ictx->flush(&ctx);
-    }
-    r = ctx.wait();
-
-    ictx->perfcounter->inc(l_librbd_flush);
-    return r;
-  }
-
   int invalidate_cache(ImageCtx *ictx)
   {
     CephContext *cct = ictx->cct;

src/librbd/internal.h

@@ -30,9 +30,7 @@ enum {
   l_librbd_discard_bytes,
   l_librbd_discard_latency,
   l_librbd_flush,
-
-  l_librbd_aio_flush,
-  l_librbd_aio_flush_latency,
+  l_librbd_flush_latency,
   l_librbd_ws,
   l_librbd_ws_bytes,
   l_librbd_ws_latency,
@@ -205,7 +203,6 @@ namespace librbd {
   void readahead(ImageCtx *ictx,
                  const vector<pair<uint64_t,uint64_t> >& image_extents);
 
-  int flush(ImageCtx *ictx);
   int invalidate_cache(ImageCtx *ictx);
   int poll_io_events(ImageCtx *ictx, io::AioCompletion **comps, int numcomp);
   int metadata_list(ImageCtx *ictx, const string &last, uint64_t max, map<string, bufferlist> *pairs);

src/librbd/io/AioCompletion.cc

@@ -71,7 +71,7 @@ void AioCompletion::complete() {
   case AIO_TYPE_DISCARD:
     ictx->perfcounter->tinc(l_librbd_discard_latency, elapsed); break;
   case AIO_TYPE_FLUSH:
-    ictx->perfcounter->tinc(l_librbd_aio_flush_latency, elapsed); break;
+    ictx->perfcounter->tinc(l_librbd_flush_latency, elapsed); break;
   case AIO_TYPE_WRITESAME:
     ictx->perfcounter->tinc(l_librbd_ws_latency, elapsed); break;
   case AIO_TYPE_COMPARE_AND_WRITE:

src/librbd/io/ImageRequest.cc

@@ -750,7 +750,7 @@ void ImageFlushRequest<I>::send_request() {
     aio_comp->put();
   }
 
-  image_ctx.perfcounter->inc(l_librbd_aio_flush);
+  image_ctx.perfcounter->inc(l_librbd_flush);
 }
 
 template <typename I>

src/librbd/io/ImageRequestWQ.cc

@@ -194,6 +194,23 @@ ssize_t ImageRequestWQ<I>::compare_and_write(uint64_t off, uint64_t len,
   return len;
 }
 
+template <typename I>
+int ImageRequestWQ<I>::flush() {
+  CephContext *cct = m_image_ctx.cct;
+  ldout(cct, 20) << "ictx=" << &m_image_ctx << dendl;
+
+  C_SaferCond cond;
+  AioCompletion *c = AioCompletion::create(&cond);
+  aio_flush(c, false);
+
+  int r = cond.wait();
+  if (r < 0) {
+    return r;
+  }
+
+  return 0;
+}
+
 template <typename I>
 void ImageRequestWQ<I>::aio_read(AioCompletion *c, uint64_t off, uint64_t len,
 				 ReadResult &&read_result, int op_flags,

src/librbd/io/ImageRequestWQ.h

@@ -37,6 +37,7 @@ class ImageRequestWQ
   ssize_t compare_and_write(uint64_t off, uint64_t len,
                             bufferlist &&cmp_bl, bufferlist &&bl,
                             uint64_t *mismatch_off, int op_flags);
+  int flush();
 
   void aio_read(AioCompletion *c, uint64_t off, uint64_t len,
                 ReadResult &&read_result, int op_flags, bool native_async=true);

src/librbd/librbd.cc

@@ -1729,7 +1729,7 @@ namespace librbd {
   {
     ImageCtx *ictx = (ImageCtx *)ctx;
     tracepoint(librbd, flush_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only);
-    int r = librbd::flush(ictx);
+    int r = ictx->io_work_queue->flush();
     tracepoint(librbd, flush_exit, r);
     return r;
   }
@@ -3771,7 +3771,7 @@ extern "C" int rbd_flush(rbd_image_t image)
 {
   librbd::ImageCtx *ictx = (librbd::ImageCtx *)image;
   tracepoint(librbd, flush_enter, ictx, ictx->name.c_str(), ictx->snap_name.c_str(), ictx->read_only);
-  int r = librbd::flush(ictx);
+  int r = ictx->io_work_queue->flush();
   tracepoint(librbd, flush_exit, r);
   return r;
 }

src/test/librbd/test_internal.cc

@@ -905,7 +905,7 @@ TEST_F(TestInternal, WriteFullCopyup) {
   bl.append(std::string(1 << ictx->order, '1'));
   ASSERT_EQ((ssize_t)bl.length(),
             ictx->io_work_queue->write(0, bl.length(), bufferlist{bl}, 0));
-  ASSERT_EQ(0, librbd::flush(ictx));
+  ASSERT_EQ(0, ictx->io_work_queue->flush());
 
   ASSERT_EQ(0, create_snapshot("snap1", true));
 

src/tools/rbd/action/Bench.cc

@@ -207,6 +207,12 @@ int do_bench(librbd::Image& image, io_type_t io_type,
     return -EINVAL;
   }
 
+  int r = image.flush();
+  if (r < 0 && (r != -EROFS || io_type != IO_TYPE_READ)) {
+    std::cerr << "rbd: failed to flush: " << cpp_strerror(r) << std::endl;
+    return r;
+  }
+
   rbd_bencher b(&image, io_type, io_size);
 
   std::cout << "bench "
@@ -316,10 +322,13 @@ int do_bench(librbd::Image& image, io_type_t io_type,
     }
   }
   b.wait_for(0);
-  int r = image.flush();
-  if (r < 0) {
-    std::cerr << "Error flushing data at the end: " << cpp_strerror(r)
-              << std::endl;
+
+  if (io_type != IO_TYPE_READ) {
+    r = image.flush();
+    if (r < 0) {
+      std::cerr << "rbd: failed to flush at the end: " << cpp_strerror(r)
+                << std::endl;
+    }
   }
 
   utime_t now = ceph_clock_now();