mds: add request to batch_op before taking auth pins and locks

MDS does not dispatch non-head requests in batch_op. So non-head
requests in batch_op should not hold any authpins and locks. Otherwise
head request may wait for these authpins/locks.

Fixes: https://tracker.ceph.com/issues/44785
Signed-off-by: "Yan, Zheng" <zyan@redhat.com>

3 files changed, 40 insertions, 20 deletions

diff --git a/src/mds/Mutation.cc b/src/mds/Mutation.cc
index acb9db21867..16f1a7383ab 100644
--- a/src/mds/Mutation.cc
+++ b/src/mds/Mutation.cc
@@ -418,12 +418,22 @@ bool MDRequestImpl::is_queued_for_replay() const
   return client_request ? client_request->is_queued_for_replay() : false;
 }
 
-bool MDRequestImpl::is_batch_op()
+bool MDRequestImpl::can_batch()
 {
-  return (client_request->get_op() == CEPH_MDS_OP_LOOKUP &&
-      client_request->get_filepath().depth() == 1) ||
-    (client_request->get_op() == CEPH_MDS_OP_GETATTR &&
-     client_request->get_filepath().depth() == 0);
+  if (num_auth_pins || num_remote_auth_pins || lock_cache || !locks.empty())
+    return false;
+
+  auto op = client_request->get_op();
+  auto& path = client_request->get_filepath();
+  if (op == CEPH_MDS_OP_GETATTR) {
+    if (path.depth() == 0)
+      return true;
+  } else if (op == CEPH_MDS_OP_LOOKUP) {
+    if (path.depth() == 1 && !path.is_last_snap())
+      return true;
+  }
+
+  return false;
 }
 
 std::unique_ptr<BatchOp> MDRequestImpl::release_batch_op()
diff --git a/src/mds/Mutation.h b/src/mds/Mutation.h
index 9b44ed088dc..4e8454e1d7d 100644
--- a/src/mds/Mutation.h
+++ b/src/mds/Mutation.h
@@ -384,7 +384,7 @@ struct MDRequestImpl : public MutationImpl {
   bool is_queued_for_replay() const;
   int compare_paths();
 
-  bool is_batch_op();
+  bool can_batch();
   bool is_batch_head() {
     return batch_op_map != nullptr;
   }
diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 6db9f8d28d0..c621d3334d9 100644
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -3741,37 +3741,47 @@ void Server::handle_client_getattr(MDRequestRef& mdr, bool is_lookup)
   if (mask & CEPH_STAT_RSTAT)
     want_auth = true; // set want_auth for CEPH_STAT_RSTAT mask
 
-  CInode *ref = rdlock_path_pin_ref(mdr, want_auth, false);
-  if (!ref)
-    return;
+  if (!mdr->is_batch_head() && mdr->can_batch()) {
+    CF_MDS_MDRContextFactory cf(mdcache, mdr, false);
+    int r = mdcache->path_traverse(mdr, cf, mdr->get_filepath(),
+				   (want_auth ? MDS_TRAVERSE_WANT_AUTH : 0),
+				   &mdr->dn[0], &mdr->in[0]);
+    if (r > 0)
+      return; // delayed
 
-  mdr->getattr_caps = mask;
-
-  if (mdr->snapid == CEPH_NOSNAP && !mdr->is_batch_head() && mdr->is_batch_op()) {
-    if (!is_lookup) {
-      mdr->pin(ref);
-      auto em = ref->batch_ops.emplace(std::piecewise_construct, std::forward_as_tuple(mask), std::forward_as_tuple());
+    if (r < 0) {
+      // fall-thru. let rdlock_path_pin_ref() check again.
+    } else if (is_lookup) {
+      CDentry* dn = mdr->dn[0].back();
+      mdr->pin(dn);
+      auto em = dn->batch_ops.emplace(std::piecewise_construct, std::forward_as_tuple(mask), std::forward_as_tuple());
       if (em.second) {
 	em.first->second = std::make_unique<Batch_Getattr_Lookup>(this, mdr);
       } else {
-	dout(20) << __func__ << ": GETATTR op, wait for previous same getattr ops to respond. " << *mdr << dendl;
+	dout(20) << __func__ << ": LOOKUP op, wait for previous same getattr ops to respond. " << *mdr << dendl;
 	em.first->second->add_request(mdr);
 	return;
       }
     } else {
-      CDentry* dn = mdr->dn[0].back();
-      mdr->pin(dn);
-      auto em = dn->batch_ops.emplace(std::piecewise_construct, std::forward_as_tuple(mask), std::forward_as_tuple());
+      CInode *in = mdr->in[0];
+      mdr->pin(in);
+      auto em = in->batch_ops.emplace(std::piecewise_construct, std::forward_as_tuple(mask), std::forward_as_tuple());
       if (em.second) {
 	em.first->second = std::make_unique<Batch_Getattr_Lookup>(this, mdr);
       } else {
-	dout(20) << __func__ << ": LOOKUP op, wait for previous same getattr ops to respond. " << *mdr << dendl;
+	dout(20) << __func__ << ": GETATTR op, wait for previous same getattr ops to respond. " << *mdr << dendl;
 	em.first->second->add_request(mdr);
 	return;
       }
     }
   }
 
+  CInode *ref = rdlock_path_pin_ref(mdr, want_auth, false);
+  if (!ref)
+    return;
+
+  mdr->getattr_caps = mask;
+
   /*
    * if client currently holds the EXCL cap on a field, do not rdlock
    * it; client's stat() will result in valid info if _either_ EXCL