mds: rdlock_path_xlock_dentry supports returning auth target inode

`Server::handle_client_openc` will call `Server::handle_client_open`
if the target inode exists. `Server::handle_client_open requires`
the target inode to be auth if client sets `O_WRONLY` or `O_TRUNC`,
but `Server::rdlock_path_xlock_dentry` can not ensure that. The assertion
that the target inode is auth in `Server::handle_client_open` may fail
and it will crash mds. So add a parameter to to support returning
auth target inode if it exists and forward the request if necessary.
The tail dentry may not be auth anymore when an auth target inode
is provided as required by the caller, because we can't require
both of them to be auth in one mds.
This bug will only be triggered in kernel client, because ceph-fuse
client will try to lookup the target before sending a create request
to mds and it can send an open request instead if the target exists.
But kernel client will send the create request to mds without lookup
first because `atomic open` is supported to improve performance.

Fixes: https://tracker.ceph.com/issues/58411
Signed-off-by: Zhansong Gao <zhsgao@hotmail.com>

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 101b0b55b37a6b89f94d3ac9abeb2e396867c6a4..8d9cf6f91301b9a73632aa0961bda297ea7e1fed 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -8277,6 +8277,7 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
   bool forward = !discover;
   bool path_locked = (flags & MDS_TRAVERSE_PATH_LOCKED);
   bool want_dentry = (flags & MDS_TRAVERSE_WANT_DENTRY);
+  bool want_inode = (flags & MDS_TRAVERSE_WANT_INODE);
   bool want_auth = (flags & MDS_TRAVERSE_WANT_AUTH);
   bool rdlock_snap = (flags & (MDS_TRAVERSE_RDLOCK_SNAP | MDS_TRAVERSE_RDLOCK_SNAP2));
   bool rdlock_path = (flags & MDS_TRAVERSE_RDLOCK_PATH);
@@ -8336,7 +8337,9 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
   if (pin)
     *pin = cur;
 
+  CInode *target_inode = nullptr;
   MutationImpl::LockOpVec lov;
+  int r;
 
   for (unsigned depth = 0; depth < path.depth(); ) {
     dout(12) << "traverse: path seg depth " << depth << " '" << path[depth]
@@ -8369,6 +8372,8 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
          pdnvec->clear();   // do not confuse likes of rdlock_path_pin_ref();
        return -CEPHFS_ENOENT;
       }
+      if (depth == path.depth() - 1)
+       target_inode = cur;
       mdr->snapid = snapid;
       depth++;
       continue;
@@ -8414,18 +8419,11 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
     }
     */
 
-    if (want_auth && want_dentry && depth == path.depth() - 1) {
-      if (curdir->is_ambiguous_auth()) {
-       dout(10) << "waiting for single auth on " << *curdir << dendl;
-       curdir->add_waiter(CInode::WAIT_SINGLEAUTH, cf.build());
-       return 1;
-      }
-      if (!curdir->is_auth()) {
-       dout(10) << "fw to auth for " << *curdir << dendl;
-       request_forward(mdr, curdir->authority().first);
-       return 2;
-      }
-    }
+    // Defer the auth check until the target inode is determined not to exist
+    // if want_inode is true.
+    if (want_auth && want_dentry && !want_inode && depth == path.depth() - 1 &&
+        (r = maybe_request_forward_to_auth(mdr, cf, curdir)) != 0)
+      return r;
 
     // Before doing dirfrag->dn lookup, compare with DamageTable's
     // record of which dentries were unreadable
@@ -8441,9 +8439,19 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
       if (dn->state_test(CDentry::STATE_PURGING))
        return -CEPHFS_ENOENT;
 
+      CDentry::linkage_t *dnl = dn->get_projected_linkage();
+      // If an auth check was deferred before and the target inode is found
+      // not to exist now, do the auth check here if necessary.
+      if (want_auth && want_dentry && want_inode && depth == path.depth() - 1 &&
+         dnl->is_null() && (r = maybe_request_forward_to_auth(mdr, cf, dn)) != 0)
+       return r;
+
       if (rdlock_path) {
        lov.clear();
-       if (xlock_dentry && depth == path.depth() - 1) {
+       // do not xlock the tail dentry if target inode exists and caller wants it
+       if (xlock_dentry && (dnl->is_null() || !want_inode) &&
+           depth == path.depth() - 1) {
+         ceph_assert(dn->is_auth());
          if (depth > 0 || !mdr->lock_cache) {
            lov.add_wrlock(&cur->filelock);
            lov.add_wrlock(&cur->nestlock);
@@ -8476,7 +8484,6 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
       if (pdnvec)
        pdnvec->push_back(dn);
 
-      CDentry::linkage_t *dnl = dn->get_projected_linkage();
       // can we conclude CEPHFS_ENOENT?
       if (dnl->is_null()) {
        dout(10) << "traverse: null+readable dentry at " << *dn << dendl;
@@ -8516,7 +8523,7 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
 
       cur = in;
 
-      if (rdlock_snap && !(want_dentry && depth == path.depth() - 1)) {
+      if (rdlock_snap && !(want_dentry && !want_inode && depth == path.depth() - 1)) {
        lov.clear();
        lov.add_rdlock(&cur->snaplock);
        if (!mds->locker->acquire_locks(mdr, lov)) {
@@ -8525,6 +8532,9 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
        }
       }
 
+      if (depth == path.depth() - 1)
+       target_inode = cur;
+
       // add to trace, continue.
       touch_inode(cur);
       if (pin)
@@ -8610,7 +8620,6 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
       }
     } else {
       // dirfrag/dentry is not mine.
-      mds_authority_t dauth = curdir->authority();
 
       if (forward &&
          mdr && mdr->client_request &&
@@ -8631,39 +8640,35 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
       if (forward) {
         // forward
         dout(7) << "traverse: not auth for " << path << " in " << *curdir << dendl;
-       
-       if (curdir->is_ambiguous_auth()) {
-         // wait
-         dout(7) << "traverse: waiting for single auth in " << *curdir << dendl;
-         curdir->add_waiter(CDir::WAIT_SINGLEAUTH, cf.build());
-         return 1;
-       } 
 
-       dout(7) << "traverse: forwarding, not auth for " << *curdir << dendl;
+       r = maybe_request_forward_to_auth(mdr, cf, curdir);
+       ceph_assert(r != 0);
 
-        request_forward(mdr, dauth.first);
+       if (r == 2 && mds->logger)
+         mds->logger->inc(l_mds_traverse_forward);
 
-       if (mds->logger) mds->logger->inc(l_mds_traverse_forward);
-       return 2;
+       return r;
       }
     }
 
     ceph_abort();  // i shouldn't get here
   }
 
-  if (want_auth && !want_dentry) {
-    if (cur->is_ambiguous_auth()) {
-      dout(10) << "waiting for single auth on " << *cur << dendl;
-      cur->add_waiter(CInode::WAIT_SINGLEAUTH, cf.build());
-      return 1;
-    }
-    if (!cur->is_auth()) {
-      dout(10) << "fw to auth for " << *cur << dendl;
-      request_forward(mdr, cur->authority().first);
-      return 2;
+  if (path.depth() == 0) {
+    dout(7) << "no tail dentry, base " << *cur << dendl;
+    if (want_dentry && !want_inode) {
+      return -CEPHFS_ENOENT;
     }
+    target_inode = cur;
   }
-  
+
+  if (target_inode) {
+    dout(7) << "found target " << *target_inode << dendl;
+    if (want_auth && !(want_dentry && !want_inode) &&
+       (r = maybe_request_forward_to_auth(mdr, cf, target_inode)) != 0)
+      return r;
+  }
+
   // success.
   if (mds->logger) mds->logger->inc(l_mds_traverse_hit);
   dout(10) << "path_traverse finish on snapid " << snapid << dendl;
@@ -8681,6 +8686,22 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
   return 0;
 }
 
+int MDCache::maybe_request_forward_to_auth(MDRequestRef& mdr, MDSContextFactory& cf,
+                                          MDSCacheObject *p)
+{
+  if (p->is_ambiguous_auth()) {
+    dout(7) << "waiting for single auth on " << *p << dendl;
+    p->add_waiter(CInode::WAIT_SINGLEAUTH, cf.build());
+    return 1;
+  }
+  if (!p->is_auth()) {
+    dout(7) << "fw to auth for " << *p << dendl;
+    request_forward(mdr, p->authority().first);
+    return 2;
+  }
+  return 0;
+}
+
 CInode *MDCache::cache_traverse(const filepath& fp)
 {
   dout(10) << "cache_traverse " << fp << dendl;

diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 7b544998b3c12c95c37c43084dd50f1ecd1b24f8..cb359c8ef4742cc8d210b96155d70722ba61bcba 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -129,6 +129,7 @@ static const int MDS_TRAVERSE_RDLOCK_PATH   = (1 << 7);
 static const int MDS_TRAVERSE_XLOCK_DENTRY     = (1 << 8);
 static const int MDS_TRAVERSE_RDLOCK_AUTHLOCK  = (1 << 9);
 static const int MDS_TRAVERSE_CHECK_LOCKCACHE  = (1 << 10);
+static const int MDS_TRAVERSE_WANT_INODE       = (1 << 11);
 
 
 // flags for predirty_journal_parents()
@@ -810,8 +811,13 @@ class MDCache {
    * dentry is encountered.
    * MDS_TRAVERSE_WANT_DENTRY: Caller wants tail dentry. Add a null dentry if
    * tail dentry does not exist. return 0 even tail dentry is null.
+   * MDS_TRAVERSE_WANT_INODE: Caller only wants target inode if it exists, or
+   * wants tail dentry if target inode does not exist and MDS_TRAVERSE_WANT_DENTRY
+   * is also set.
    * MDS_TRAVERSE_WANT_AUTH: Always forward request to auth MDS of target inode
    * or auth MDS of tail dentry (MDS_TRAVERSE_WANT_DENTRY is set).
+   * MDS_TRAVERSE_XLOCK_DENTRY: Caller wants to xlock tail dentry if MDS_TRAVERSE_WANT_INODE
+   * is not set or (MDS_TRAVERSE_WANT_INODE is set but target inode does not exist)
    *
    * @param pdnvec Data return parameter -- on success, contains a
    * vector of dentries. On failure, is either empty or contains the
@@ -829,6 +835,9 @@ class MDCache {
                    const filepath& path, int flags,
                    std::vector<CDentry*> *pdnvec, CInode **pin=nullptr);
 
+  int maybe_request_forward_to_auth(MDRequestRef& mdr, MDSContextFactory& cf,
+                                   MDSCacheObject *p);
+
   CInode *cache_traverse(const filepath& path);
 
   void open_remote_dirfrag(CInode *diri, frag_t fg, MDSContext *fin);

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index bbfa55144f5700967c4e2b8b76895f6955421a03..1a79b97461ac63d2f3bce1f0965c4c5148093082 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -3585,12 +3585,18 @@ CInode* Server::rdlock_path_pin_ref(MDRequestRef& mdr,
 
 /** rdlock_path_xlock_dentry
  * traverse path to the directory that could/would contain dentry.
- * make sure i am auth for that dentry, forward as necessary.
- * create null dentry in place (or use existing if okexist).
+ * make sure i am auth for that dentry (or target inode if it exists and authexist),
+ * forward as necessary. create null dentry in place (or use existing if okexist).
  * get rdlocks on traversed dentries, xlock on new dentry.
+ *
+ * set authexist true if caller requires the target inode to be auth when it exists.
+ * the tail dentry is not always auth any more if authexist because it is impossible
+ * to ensure tail dentry and target inode are both auth in one mds. the tail dentry
+ * will not be xlocked too if authexist and the target inode exists.
  */
 CDentry* Server::rdlock_path_xlock_dentry(MDRequestRef& mdr,
-                                         bool create, bool okexist, bool want_layout)
+                                         bool create, bool okexist, bool authexist,
+                                         bool want_layout)
 {
   const filepath& refpath = mdr->get_filepath();
   dout(10) << "rdlock_path_xlock_dentry " << *mdr << " " << refpath << dendl;
@@ -3628,6 +3634,8 @@ CDentry* Server::rdlock_path_xlock_dentry(MDRequestRef& mdr,
     flags |= MDS_TRAVERSE_CHECK_LOCKCACHE;
   if (create)
     flags |= MDS_TRAVERSE_RDLOCK_AUTHLOCK;
+  if (authexist)
+    flags |= MDS_TRAVERSE_WANT_INODE;
   if (want_layout)
     flags |= MDS_TRAVERSE_WANT_DIRLAYOUT;
   int r = mdcache->path_traverse(mdr, cf, refpath, flags, &mdr->dn[0]);
@@ -4453,7 +4461,7 @@ void Server::handle_client_openc(MDRequestRef& mdr)
   }
 
   bool excl = req->head.args.open.flags & CEPH_O_EXCL;
-  CDentry *dn = rdlock_path_xlock_dentry(mdr, true, !excl, true);
+  CDentry *dn = rdlock_path_xlock_dentry(mdr, true, !excl, true, true);
   if (!dn)
     return;
 
@@ -4465,7 +4473,7 @@ void Server::handle_client_openc(MDRequestRef& mdr)
   CDentry::linkage_t *dnl = dn->get_projected_linkage();
   if (!excl && !dnl->is_null()) {
     // it existed.
-    mds->locker->xlock_downgrade(&dn->lock, mdr.get());
+    ceph_assert(mdr.get()->is_rdlocked(&dn->lock));
 
     MutationImpl::LockOpVec lov;
     lov.add_rdlock(&dnl->get_inode()->snaplock);
@@ -6801,7 +6809,7 @@ void Server::handle_client_mknod(MDRequestRef& mdr)
     mode |= S_IFREG;
 
   mdr->disable_lock_cache();
-  CDentry *dn = rdlock_path_xlock_dentry(mdr, true, false, S_ISREG(mode));
+  CDentry *dn = rdlock_path_xlock_dentry(mdr, true, false, false, S_ISREG(mode));
   if (!dn)
     return;
 

diff --git a/src/mds/Server.h b/src/mds/Server.h
index a2d8f69670871d043fb824fa10d406124aa6bc68..cae6951baeb012eb4804ddcd3bfa1b778633d20c 100644 (file)
--- a/src/mds/Server.h
+++ b/src/mds/Server.h
@@ -191,7 +191,8 @@ public:
   CInode* rdlock_path_pin_ref(MDRequestRef& mdr, bool want_auth,
                              bool no_want_auth=false);
   CDentry* rdlock_path_xlock_dentry(MDRequestRef& mdr, bool create,
-                                   bool okexist=false, bool want_layout=false);
+                                   bool okexist=false, bool authexist=false,
+                                   bool want_layout=false);
   std::pair<CDentry*, CDentry*>
            rdlock_two_paths_xlock_destdn(MDRequestRef& mdr, bool xlock_srcdn);