librbd: fix block_completion race condition

If block completions finished before all of them were scheduled, the
pending_count could reach 0, and the callback could be called more
than once. This caused the reference counting to be off, so the next
call to the callback passed an invalid pointer, resulting in crashes
like this (or more mysterious ones):

common/Mutex.h: In function 'void Mutex::Lock(bool)', in thread '0x7f5f37e62700'
common/Mutex.h: 118: FAILED assert(r == 0)
 1: (librbd::RBD::AioCompletion::get_return_value()+0x19c) [0x7f5f408835bc]
 2: /usr/bin/kvm() [0x4629fd]
 3: (librbd::AioCompletion::complete_block(librbd::AioBlockCompletion*, long)+0x13c) [0x7f5f4088ab5c]
 4: (librbd::rados_aio_sparse_read_cb(void*, void*)+0x8d) [0x7f5f4088b40d]
 5: (librados::RadosClient::C_aio_sparse_read_Ack::finish(int)+0x12e) [0x7f5f4050beee]
 6: (Objecter::handle_osd_op_reply(MOSDOpReply*)+0x783) [0x7f5f40517863]
 7: (librados::RadosClient::_dispatch(Message*)+0x3c) [0x7f5f404ffe8c]
 8: (librados::RadosClient::ms_dispatch(Message*)+0x33) [0x7f5f404fff53]
 9: (SimpleMessenger::dispatch_entry()+0x69d) [0x7f5f405bf5bd]
 10: (SimpleMessenger::DispatchThread::entry()+0x1c) [0x7f5f4050052c]
 11: (()+0x6d8c) [0x7f5f40c9dd8c]
 12: (clone()+0x6d) [0x7f5f3e0b904d]

Signed-off-by: Josh Durgin <josh.durgin@dreamhost.com>

diff --git a/src/librbd.cc b/src/librbd.cc
index 8ab22a2d8744cb759aba6964a521f309ea28aed5..35f34c84c4962e56478cfe138b74ebfe2af3dc8d 100644 (file)
--- a/src/librbd.cc
+++ b/src/librbd.cc
@@ -140,12 +140,13 @@ namespace librbd {
     bool released;
 
     AioCompletion() : lock("AioCompletion::lock", true), done(false), rval(0), complete_cb(NULL), complete_arg(NULL),
-                     rbd_comp(NULL), pending_count(0), ref(1), released(false) {
+                     rbd_comp(NULL), pending_count(1), ref(1), released(false) {
       dout(20) << "AioCompletion::AioCompletion() this=" << (void *)this << dendl;
     }
     ~AioCompletion() {
       dout(20) << "AioCompletion::~AioCompletion() this=" << (void *)this << dendl;
     }
+
     int wait_for_complete() {
       dout(20) << "AioCompletion::wait_for_complete() this=" << (void *)this << dendl;
       lock.Lock();
@@ -163,6 +164,27 @@ namespace librbd {
       get();
     }
 
+    void finish_adding_completions() {
+      dout(20) << "AioCompletion::finish_adding_completions() this=" << (void *)this << dendl;
+      lock.Lock();
+      assert(pending_count);
+      int count = --pending_count;
+      if (!count) {
+       complete();
+      }
+      lock.Unlock();
+    }
+
+    void complete() {
+      dout(20) << "AioCompletion::complete() this=" << (void *)this << dendl;
+      assert(lock.is_locked());
+      if (complete_cb) {
+       complete_cb(rbd_comp, complete_arg);
+      }
+      done = true;
+      cond.Signal();
+    }
+
     void set_complete_cb(void *cb_arg, callback_t cb) {
       complete_cb = cb;
       complete_arg = cb_arg;
@@ -1263,10 +1285,7 @@ void AioCompletion::complete_block(AioBlockCompletion *block_completion, ssize_t
   assert(pending_count);
   int count = --pending_count;
   if (!count) {
-    if (complete_cb)
-      complete_cb(rbd_comp, complete_arg);
-    done = true;
-    cond.Signal();
+    complete();
   }
   put_unlock();
 }
@@ -1328,6 +1347,7 @@ int aio_write(ImageCtx *ictx, uint64_t off, size_t len, const char *buf,
   }
   r = 0;
 done:
+  c->finish_adding_completions();
   c->put();
   /* FIXME: cleanup all the allocated stuff */
   return r;
@@ -1392,6 +1412,7 @@ int aio_read(ImageCtx *ictx, uint64_t off, size_t len,
   }
   ret = total_read;
 done:
+  c->finish_adding_completions();
   c->put();
   return ret;
 }