mds: find a new head for the batch ops when the head is dead

If the batch head request is already dead and then we need to choose
a new batch head anyways and release the reference for the current
batch head request. Else it will be reported as slow request.

Fixes: https://tracker.ceph.com/issues/65536
Signed-off-by: Xiubo Li <xiubli@redhat.com>

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index f65f81d217e1d6ded054589f803d66edf6d4e11e..9cc2eb8d9d68b688734b6d5d9647de58a5463823 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -9811,7 +9811,25 @@ void MDCache::request_forward(const MDRequestRef& mdr, mds_rank_t who, int port)
 void MDCache::dispatch_request(const MDRequestRef& mdr)
 {
   if (mdr->dead) {
-    dout(20) << __func__ << ": dead " << *mdr << dendl;
+    dout(10) << __func__ << ": dead " << *mdr << 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->client_request && 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);
+        dout(10) << __func__ << ": mask '" << mask
+                 << "' batch head is dead and there is no follower" << dendl;
+        return;
+      }
+      dout(10) << __func__ << ": mask '" << mask
+               << "' batch head is dead and queue a new one "
+               << *new_batch_head << dendl;
+      mds->finisher->queue(new C_MDS_RetryRequest(this, new_batch_head));
+      return;
+    }
     return;
   }
   if (mdr->client_request) {

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 0bb30664ac5f669847d4f8cc587604a13d8dfe2e..14faeb3ecc950dc4d969f31764a86bd7e4c408c5 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -2647,9 +2647,14 @@ void Server::dispatch_client_request(const MDRequestRef& mdr)
       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->batch_op_map->erase(it);
+        dout(10) << __func__ << ": mask '" << mask
+                 << "' batch head is killed and there is no follower" << dendl;
+        return;
       }
+      dout(10) << __func__ << ": mask '" << mask
+               << "' batch head is killed and queue a new one "
+               << *new_batch_head << dendl;
       mds->finisher->queue(new C_MDS_RetryRequest(mdcache, new_batch_head));
       return;
     } else {