librbd: add new fail method to AioCompletion

Helper method to handle passing fatal errors generated within
librbd (not from the OSDs) back to the client.

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

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"
@@ -63,11 +64,11 @@ namespace librbd {
     }
   }
 
-  void AioCompletion::complete() {
+  void AioCompletion::complete(CephContext *cct) {
     tracepoint(librbd, aio_complete_enter, this, rval);
     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_rd_latency, elapsed); break;
@@ -78,12 +79,14 @@ 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;
     }
 
     // note: possible for image to be closed after op marked finished
-    async_op.finish_op();
+    if (async_op.started()) {
+      async_op.finish_op();
+    }
 
     if (complete_cb) {
       complete_cb(rbd_comp, complete_arg);
@@ -93,6 +96,17 @@ namespace librbd {
     tracepoint(librbd, aio_complete_exit);
   }
 
+  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() "
@@ -109,7 +123,7 @@ namespace librbd {
     int count = --pending_count;
     if (!count && blockers == 0) {
       finalize(cct, rval);
-      complete();
+      complete(cct);
     }
     put_unlock();
   }

src/librbd/AioCompletion.h

@@ -98,7 +98,9 @@ namespace librbd {
       }
     }
 
-    void complete();
+    void fail(CephContext *cct, int r);
+
+    void complete(CephContext *cct);
 
     void set_complete_cb(void *cb_arg, callback_t cb) {
       complete_cb = cb;
@@ -145,7 +147,7 @@ namespace librbd {
       --blockers;
       if (pending_count == 0 && blockers == 0) {
         finalize(cct, rval);
-        complete();
+        complete(cct);
       }
     }
   };

src/librbd/AsyncOperation.h

@@ -26,6 +26,10 @@ public:
     assert(!m_xlist_item.is_on_list());
   }
 
+  inline bool started() const {
+    return m_xlist_item.is_on_list();
+  }
+
   void start_op(ImageCtx &image_ctx);
   void finish_op();