mds: try to choose a new batch head in request_clientup()

This will happen only for the client requests, not peer requests.
The 'mdr->killed' and 'mdr->dead' will always be set at the same
time when killing the client requests.

Fixes: https://tracker.ceph.com/issues/66124
Signed-off-by: Xiubo Li <xiubli@redhat.com>
(cherry picked from commit 73b266e53632de270af31588da497f7162301fc0)

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 0f17a29ab2c9e72af9edf3672f534c6c66814144..85852e0c526fd1523f28066782a3ce6f9bcd9d8f 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -9897,6 +9897,19 @@ void MDCache::request_cleanup(const MDRequestRef& mdr)
 
   mdr->dead = true;
 
+  if (mdr->killed && mdr->client_request && mdr->is_batch_head()) {
+    dout(10) << "request " << *mdr << " was killed and dead" << dendl;
+    //if the mdr is a "batch_op" and it has followers, pick a follower as
+    //the new "head of the batch ops" and go on processing the new one.
+    int mask = mdr->client_request->head.args.getattr.mask;
+    auto it = mdr->batch_op_map->find(mask);
+    auto new_batch_head = it->second->find_new_head();
+    if (!new_batch_head) {
+      mdr->batch_op_map->erase(it);
+    }
+    mds->finisher->queue(new C_MDS_RetryRequest(this, new_batch_head));
+  }
+
   if (mdr->has_more()) {
     if (mdr->more()->is_ambiguous_auth)
       mdr->clear_ambiguous_auth();

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 1df124631f6b38f50d3d1eda91bc1ffe2c51c126..bd8e1f11b1cfce5e0a607bb8ca387802ad8bfd09 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -2657,22 +2657,9 @@ void Server::dispatch_client_request(const MDRequestRef& mdr)
   ceph_assert(!mdr->has_more() || mdr->more()->waiting_on_peer.empty());
 
   if (mdr->killed) {
-    dout(10) << "request " << *mdr << " was killed" << dendl;
-    //if the mdr is a "batch_op" and it has followers, pick a follower as
-    //the new "head of the batch ops" and go on processing the new one.
-    if (mdr->is_batch_head()) {
-      int mask = mdr->client_request->head.args.getattr.mask;
-      auto it = mdr->batch_op_map->find(mask);
-      auto new_batch_head = it->second->find_new_head();
-      if (!new_batch_head) {
-       mdr->batch_op_map->erase(it);
-       return;
-      }
-      mds->finisher->queue(new C_MDS_RetryRequest(mdcache, new_batch_head));
-      return;
-    } else {
-      return;
-    }
+    // Should already be reset in request_cleanup().
+    ceph_assert(!mdr->is_batch_head());
+    return;
   } else if (mdr->aborted) {
     mdr->aborted = false;
     mdcache->request_kill(mdr);