librados: fix async aio completion wakeupwip-5919

For aio flush, we register a wait on the most recent write.  The write
completion code, however, was *only* waking the waiter if they were waiting
on that write, without regard to previous writes (completed or not).
For example, we might have 6 and 7 outstanding and wait on 7.  If they
finish in order all is well, but if 7 finishes first we do the flush
completion early.  Similarly, if we

 - start 6
 - start 7
 - finish 7
 - flush; wait on 7
 - finish 6

we can hang forever.

Fix by doing any completions that are prior to the oldest pending write in
the aio write completion handler.

Refs: #5919

Signed-off-by: Sage Weil <sage@inktank.com>
Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

1 files changed, 17 insertions, 12 deletions

diff --git a/src/librados/IoCtxImpl.cc b/src/librados/IoCtxImpl.cc
index b5da027e8f6..ce9743a54b3 100644
--- a/src/librados/IoCtxImpl.cc
+++ b/src/librados/IoCtxImpl.cc
@@ -82,22 +82,25 @@ void librados::IoCtxImpl::complete_aio_write(AioCompletionImpl *c)
   aio_write_list_lock.Lock();
   assert(c->io == this);
   c->aio_write_list_item.remove_myself();
-  // queue async flush waiters
-  map<tid_t, std::list<AioCompletionImpl*> >::iterator waiters =
-    aio_write_waiters.find(c->aio_write_seq);
-  if (waiters != aio_write_waiters.end()) {
-    ldout(client->cct, 20) << "found " << waiters->second.size()
-			   << " waiters" << dendl;
+
+  map<tid_t, std::list<AioCompletionImpl*> >::iterator waiters = aio_write_waiters.begin();
+  while (waiters != aio_write_waiters.end()) {
+    if (!aio_write_list.empty() &&
+	aio_write_list.front()->aio_write_seq <= waiters->first) {
+      ldout(client->cct, 20) << " next outstanding write is " << aio_write_list.front()->aio_write_seq
+			     << " <= waiter " << waiters->first
+			     << ", stopping" << dendl;
+      break;
+    }
+    ldout(client->cct, 20) << " waking waiters on seq " << waiters->first << dendl;
     for (std::list<AioCompletionImpl*>::iterator it = waiters->second.begin();
 	 it != waiters->second.end(); ++it) {
       client->finisher.queue(new C_AioCompleteAndSafe(*it));
       (*it)->put();
     }
-    aio_write_waiters.erase(waiters);
-  } else {
-    ldout(client->cct, 20) << "found no waiters for tid "
-			   << c->aio_write_seq << dendl;
+    aio_write_waiters.erase(waiters++);
   }
+
   aio_write_cond.Signal();
   aio_write_list_lock.Unlock();
   put();
@@ -109,11 +112,13 @@ void librados::IoCtxImpl::flush_aio_writes_async(AioCompletionImpl *c)
 			 << " completion " << c << dendl;
   Mutex::Locker l(aio_write_list_lock);
   tid_t seq = aio_write_seq;
-  ldout(client->cct, 20) << "flush_aio_writes_async waiting on tid "
-			 << seq << dendl;
   if (aio_write_list.empty()) {
+    ldout(client->cct, 20) << "flush_aio_writes_async no writes. (tid "
+			   << seq << ")" << dendl;
     client->finisher.queue(new C_AioCompleteAndSafe(c));
   } else {
+    ldout(client->cct, 20) << "flush_aio_writes_async " << aio_write_list.size()
+			   << " writes in flight; waiting on tid " << seq << dendl;
     c->get();
     aio_write_waiters[seq].push_back(c);
   }