librbd: invoke AioCompletion external callbacks via the op_work_queue

The op_work_queue is single-threaded and will ensure clients won't
receive concurrent callbacks. It is also a simpler approach than
the original lock-free queue implementation that had issues with
potential race conditions.

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

diff --git a/src/librbd/ImageCtx.cc b/src/librbd/ImageCtx.cc
index 74bd121bad7df3b650ef189d57e7404e0cdc9d17..a80f4dab1f3b87c816bdea09aac2caf2a695a976 100644 (file)
--- a/src/librbd/ImageCtx.cc
+++ b/src/librbd/ImageCtx.cc
@@ -122,7 +122,6 @@ public:
       operations(new Operations<>(*this)),
       exclusive_lock(nullptr), object_map(nullptr),
       io_work_queue(nullptr), op_work_queue(nullptr),
-      external_callback_completions(32),
       event_socket_completions(32),
       asok_hook(nullptr),
       trace_endpoint("librbd")

diff --git a/src/librbd/ImageCtx.h b/src/librbd/ImageCtx.h
index 79908cf18dc43d72067187b9b1dde2e569f2859f..fe54d19467eaa189d9944916ec3ccf964532b928 100644 (file)
--- a/src/librbd/ImageCtx.h
+++ b/src/librbd/ImageCtx.h
@@ -189,9 +189,6 @@ namespace librbd {
       io::AioCompletion*,
       boost::lockfree::allocator<ceph::allocator<void>>> Completions;
 
-    Completions external_callback_completions;
-    std::atomic<bool> external_callback_in_progress = {false};
-
     Completions event_socket_completions;
     EventSocket event_socket;
 

diff --git a/src/librbd/io/AioCompletion.cc b/src/librbd/io/AioCompletion.cc
index fa0548f15573cea2b960632782b61378e730aba3..6644b608bd93e501f0ef5fdcbf2342e86a8dc14a 100644 (file)
--- a/src/librbd/io/AioCompletion.cc
+++ b/src/librbd/io/AioCompletion.cc
@@ -240,28 +240,13 @@ ssize_t AioCompletion::get_return_value() {
 void AioCompletion::complete_external_callback() {
   // ensure librbd external users never experience concurrent callbacks
   // from multiple librbd-internal threads.
-  ictx->external_callback_completions.push(this);
-
-  while (true) {
-    if (ictx->external_callback_in_progress.exchange(true)) {
-      // another thread is concurrently invoking external callbacks
-      break;
-    }
-
-    AioCompletion* aio_comp;
-    while (ictx->external_callback_completions.pop(aio_comp)) {
-      aio_comp->complete_cb(aio_comp->rbd_comp, aio_comp->complete_arg);
-      aio_comp->complete_event_socket();
-      aio_comp->notify_callbacks_complete();
-    }
-
-    ictx->external_callback_in_progress.store(false);
-    if (ictx->external_callback_completions.empty()) {
-      // queue still empty implies we didn't have a race between the last failed
-      // pop and resetting the in-progress state
-      break;
-    }
-  }
+  get();
+  ictx->op_work_queue->queue(new LambdaContext([this](int r) {
+      complete_cb(rbd_comp, complete_arg);
+      complete_event_socket();
+      notify_callbacks_complete();
+      put();
+    }));
 }
 
 void AioCompletion::complete_event_socket() {