Merge pull request #17352 from liewegas/wip-21171

os/bluestore: drop deferred_submit_lock, fix aio leak

Reviewed-by: xie xingguo <xie.xingguo@zte.com.cn>
Reviewed-by: Pan Liu <liupan1111@gmail.com>

src/os/bluestore/BlueStore.cc

@@ -8646,9 +8646,16 @@ void BlueStore::deferred_try_submit()
     osrs.push_back(&osr);
   }
   for (auto& osr : osrs) {
-    if (osr->deferred_pending && !osr->deferred_running) {
-      _deferred_submit_unlock(osr.get());
-      deferred_lock.lock();
+    if (osr->deferred_pending) {
+      if (!osr->deferred_running) {
+	_deferred_submit_unlock(osr.get());
+	deferred_lock.lock();
+      } else {
+	dout(20) << __func__ << "  osr " << osr << " already has running"
+		 << dendl;
+      }
+    } else {
+      dout(20) << __func__ << "  osr " << osr << " has no pending" << dendl;
     }
   }
 }
@@ -8705,8 +8712,6 @@ void BlueStore::_deferred_submit_unlock(OpSequencer *osr)
     ++i;
   }
 
-  // demote to deferred_submit_lock, then drop that too
-  std::lock_guard<std::mutex> l(deferred_submit_lock);
   deferred_lock.unlock();
   bdev->aio_submit(&b->ioc);
 }
@@ -8722,6 +8727,7 @@ void BlueStore::_deferred_aio_finish(OpSequencer *osr)
     assert(osr->deferred_running == b);
     osr->deferred_running = nullptr;
     if (!osr->deferred_pending) {
+      dout(20) << __func__ << " dequeueing" << dendl;
       auto q = deferred_queue.iterator_to(*osr);
       deferred_queue.erase(q);
     } else if (deferred_aggressive) {
@@ -8729,6 +8735,8 @@ void BlueStore::_deferred_aio_finish(OpSequencer *osr)
       deferred_finisher.queue(new FunctionContext([&](int) {
 	    deferred_try_submit();
 	  }));
+    } else {
+      dout(20) << __func__ << " leaving queued, more pending" << dendl;
     }
   }
 

src/os/bluestore/BlueStore.h

@@ -1841,7 +1841,7 @@ private:
   interval_set<uint64_t> bluefs_extents;  ///< block extents owned by bluefs
   interval_set<uint64_t> bluefs_extents_reclaiming; ///< currently reclaiming
 
-  std::mutex deferred_lock, deferred_submit_lock;
+  std::mutex deferred_lock;
   std::atomic<uint64_t> deferred_seq = {0};
   deferred_osr_queue_t deferred_queue; ///< osr's with deferred io pending
   int deferred_queue_size = 0;         ///< num txc's queued across all osrs

src/os/bluestore/aio.cc

@@ -39,15 +39,16 @@ int aio_queue_t::submit_batch(aio_iter begin, aio_iter end,
 
   aio_iter cur = begin;
   struct iocb *piocb[aios_size];
-  int r, pos = 0;
+  int left = 0;
   while (cur != end) {
     cur->priv = priv;
-    *(piocb+pos) = &cur->iocb;
-    ++pos;
+    *(piocb+left) = &cur->iocb;
+    ++left;
     ++cur;
   }
-  while (true) {
-    r = io_submit(ctx, pos, piocb);
+  int done = 0;
+  while (left > 0) {
+    int r = io_submit(ctx, left, piocb + done);
     if (r < 0) {
       if (r == -EAGAIN && attempts-- > 0) {
 	usleep(delay);
@@ -55,10 +56,13 @@ int aio_queue_t::submit_batch(aio_iter begin, aio_iter end,
 	(*retries)++;
 	continue;
       }
+      return r;
     }
-    break;
+    assert(r > 0);
+    done += r;
+    left -= r;
   }
-  return r;
+  return done;
 }
 
 int aio_queue_t::get_next_completed(int timeout_ms, aio_t **paio, int max)