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)

 Conflicts:
	src/mds/MDCache.cc
	src/mds/Server.cc
	Missing commit 99e58f58547("mds: use const qualifier for
	MDRequestRef").

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 392da0b10cb5c0248f59a2ef5dc119b1644dd225..d7228e12823ab954bdce162438fb9702e98af7e1 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -9857,6 +9857,19 @@ void MDCache::request_cleanup(MDRequestRef& mdr)
 {
   dout(15) << "request_cleanup " << *mdr << dendl;
 
+  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 54b8d27f3a74633c970cd0aad620e66fa5f5abdd..04b4a07da8cc93c585d701af5cdd5021092f32d4 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -2655,21 +2655,9 @@ void Server::dispatch_client_request(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;
-      }
-      mdr = std::move(new_batch_head);
-    } 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);