mds: enable lock cache for openc/unlink requests

when client has 'CEPH_CAP_FILE_EXCL|CEPH_CAP_DIR_UNLINK' caps of a
directory inode, it can asynchronously unlink file from the directory
if the corresponding dentry is a primary link.

when client has 'CEPH_CAP_FILE_EXCL|CEPH_CAP_DIR_CREATE' caps of a
directory inode, it can asynchronously create new file in the directory
as long as no file with the same name exists.

Signed-off-by: "Yan, Zheng" <zyan@redhat.com>

diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index 70de4793bd53fc51e78a437ded550fc38d812330..c3796738c4619568b142d02870cd6074bf8aee4a 100644 (file)
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -362,10 +362,16 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
     }
     
     if (!object->is_auth()) {
-      ceph_assert(!mdr->lock_cache);
+      if (mdr->lock_cache) { // debug
+       ceph_assert(mdr->lock_cache->opcode == CEPH_MDS_OP_UNLINK);
+       CDentry *dn = mdr->dn[0].back();
+       ceph_assert(dn->get_projected_linkage()->is_remote());
+      }
+
       if (object->is_ambiguous_auth()) {
        // wait
        dout(10) << " ambiguous auth, waiting to authpin " << *object << dendl;
+       mdr->disable_lock_cache();
        drop_locks(mdr.get());
        mdr->drop_local_auth_pins();
        marker.message = "waiting for single auth, object is being migrated";
@@ -387,12 +393,20 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
        } else {
          ceph_assert(0 == "unknown type of lock parent");
        }
-       ceph_assert(dir->get_inode() == mdr->lock_cache->get_dir_inode());
-       /* forcibly auth pin if lock cache is used */
-       continue;
+       if (dir->get_inode() == mdr->lock_cache->get_dir_inode()) {
+         // forcibly auth pin if there is lock cache on parent dir
+         continue;
+       }
+
+       { // debug
+         ceph_assert(mdr->lock_cache->opcode == CEPH_MDS_OP_UNLINK);
+         CDentry *dn = mdr->dn[0].back();
+         ceph_assert(dn->get_projected_linkage()->is_remote());
+       }
       }
 
       // wait
+      mdr->disable_lock_cache();
       drop_locks(mdr.get());
       mdr->drop_local_auth_pins();
       if (auth_pin_nonblocking) {
@@ -868,7 +882,7 @@ void Locker::invalidate_lock_caches(SimpleLock *lock)
   }
 }
 
-void Locker::create_lock_cache(MDRequestRef& mdr, CInode *diri)
+void Locker::create_lock_cache(MDRequestRef& mdr, CInode *diri, file_layout_t *dir_layout)
 {
   if (mdr->lock_cache)
     return;
@@ -955,6 +969,8 @@ void Locker::create_lock_cache(MDRequestRef& mdr, CInode *diri)
   }
 
   auto lock_cache = new MDLockCache(cap, opcode);
+  if (dir_layout)
+    lock_cache->set_dir_layout(*dir_layout);
   cap->set_lock_cache_allowed(get_cap_bit_for_lock_cache(opcode));
 
   for (auto dir : dfv) {

diff --git a/src/mds/Locker.h b/src/mds/Locker.h
index f0963fd49e3edbfb5dc8ba144f7f060896eb971c..2a3f51cf621909ff7545cc723b1b99db85825a4b 100644 (file)
--- a/src/mds/Locker.h
+++ b/src/mds/Locker.h
@@ -68,7 +68,7 @@ public:
   void drop_locks_for_fragment_unfreeze(MutationImpl *mut);
 
   int get_cap_bit_for_lock_cache(int op);
-  void create_lock_cache(MDRequestRef& mdr, CInode *diri);
+  void create_lock_cache(MDRequestRef& mdr, CInode *diri, file_layout_t *dir_layout=nullptr);
   bool find_and_attach_lock_cache(MDRequestRef& mdr, CInode *diri);
   void invalidate_lock_caches(CDir *dir);
   void invalidate_lock_caches(SimpleLock *lock);

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index c61690902979d2f643da1d644e04cc988e4b13fb..b241315d49cedc0b718408aee8fbf9ae807cf8c3 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -8111,7 +8111,13 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
     return 1;
   }
 
-  if (rdlock_snap) {
+  if (flags & MDS_TRAVERSE_CHECK_LOCKCACHE)
+    mds->locker->find_and_attach_lock_cache(mdr, cur);
+
+  if (mdr && mdr->lock_cache) {
+    if (flags & MDS_TRAVERSE_WANT_DIRLAYOUT)
+      mdr->dir_layout = mdr->lock_cache->get_dir_layout();
+  } else if (rdlock_snap) {
     int n = (flags & MDS_TRAVERSE_RDLOCK_SNAP2) ? 1 : 0;
     if ((n == 0 && !(mdr->locking_state & MutationImpl::SNAP_LOCKED)) ||
        (n == 1 && !(mdr->locking_state & MutationImpl::SNAP2_LOCKED))) {
@@ -8129,8 +8135,7 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
 
   MutationImpl::LockOpVec lov;
 
-  unsigned depth = 0;
-  while (depth < path.depth()) {
+  for (unsigned depth = 0; depth < path.depth(); ) {
     dout(12) << "traverse: path seg depth " << depth << " '" << path[depth]
             << "' snapid " << snapid << dendl;
     
@@ -8236,13 +8241,17 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
       if (rdlock_path) {
        lov.clear();
        if (xlock_dentry && depth == path.depth() - 1) {
-         lov.add_wrlock(&cur->filelock);
-         lov.add_wrlock(&cur->nestlock);
-         if (rdlock_authlock)
-           lov.add_rdlock(&cur->authlock);
-
+         if (depth > 0 || !mdr->lock_cache) {
+           lov.add_wrlock(&cur->filelock);
+           lov.add_wrlock(&cur->nestlock);
+           if (rdlock_authlock)
+             lov.add_rdlock(&cur->authlock);
+         }
          lov.add_xlock(&dn->lock);
        } else {
+         // force client to flush async dir operation if necessary
+         if (cur->filelock.is_cached())
+           lov.add_wrlock(&cur->filelock);
          lov.add_rdlock(&dn->lock);
        }
        if (!mds->locker->acquire_locks(mdr, lov)) {
@@ -8356,12 +8365,17 @@ int MDCache::path_traverse(MDRequestRef& mdr, MDSContextFactory& cf,
            if (rdlock_path) {
              lov.clear();
              if (xlock_dentry) {
-               lov.add_wrlock(&cur->filelock);
-               lov.add_wrlock(&cur->nestlock);
-               if (rdlock_authlock)
-                 lov.add_rdlock(&cur->authlock);
+               if (depth > 0 || !mdr->lock_cache) {
+                 lov.add_wrlock(&cur->filelock);
+                 lov.add_wrlock(&cur->nestlock);
+                 if (rdlock_authlock)
+                   lov.add_rdlock(&cur->authlock);
+               }
                lov.add_xlock(&dn->lock);
              } else {
+               // force client to flush async dir operation if necessary
+               if (cur->filelock.is_cached())
+                 lov.add_wrlock(&cur->filelock);
                lov.add_rdlock(&dn->lock);
              }
              if (!mds->locker->acquire_locks(mdr, lov)) {

diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index b781642de995d73efa08fdc8747951c2bbfaa1e2..c21a40dacaa27d3ddf5394628681e619f83054a8 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -121,6 +121,7 @@ static const int MDS_TRAVERSE_WANT_DIRLAYOUT        = (1 << 6);
 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);
 
 
 // flags for predirty_journal_parents()

diff --git a/src/mds/Mutation.h b/src/mds/Mutation.h
index fd72d6bd8f4bf008fa83a783900df85151cc92b7..0b080d8a3d9c79f32126c4fca52526d1853c1dbd 100644 (file)
--- a/src/mds/Mutation.h
+++ b/src/mds/Mutation.h
@@ -138,6 +138,11 @@ public:
   lock_set locks;  // full ordering
 
   MDLockCache* lock_cache = nullptr;
+  bool lock_cache_disabled = false;
+
+  void disable_lock_cache() {
+    lock_cache_disabled = true;
+  }
 
   lock_iterator emplace_lock(SimpleLock *l, unsigned f=0, mds_rank_t t=MDS_RANK_NONE) {
     last_locked = l;
@@ -493,6 +498,7 @@ struct MDLockCache : public MutationImpl {
   CInode *diri;
   Capability *client_cap;
   int opcode;
+  file_layout_t dir_layout;
 
   elist<MDLockCache*>::item item_cap_lock_cache;
 
@@ -517,6 +523,13 @@ struct MDLockCache : public MutationImpl {
   }
 
   CInode *get_dir_inode() { return diri; }
+  void set_dir_layout(file_layout_t& layout) {
+    dir_layout = layout;
+  }
+  const file_layout_t& get_dir_layout() const {
+    return dir_layout;
+  }
+
   void attach_locks();
   void attach_dirfrags(std::vector<CDir*>&& dfv);
   void detach_all();

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 8a1454371cc716c30e6bfa93666fe9c177ef40bc..76a3f7fe32b2119130a6a5a12e2e15300ec08e7d 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -2995,6 +2995,7 @@ void Server::handle_slave_auth_pin_ack(MDRequestRef& mdr, const cref_t<MMDSSlave
   mds_rank_t from = mds_rank_t(ack->get_source().num());
 
   if (ack->is_req_blocked()) {
+    mdr->disable_lock_cache();
     // slave auth pin is blocked, drop locks to avoid deadlock
     mds->locker->drop_locks(mdr.get(), nullptr);
     return;
@@ -3411,6 +3412,8 @@ CDentry* Server::rdlock_path_xlock_dentry(MDRequestRef& mdr,
   int flags = MDS_TRAVERSE_RDLOCK_SNAP | MDS_TRAVERSE_RDLOCK_PATH |
              MDS_TRAVERSE_WANT_DENTRY | MDS_TRAVERSE_XLOCK_DENTRY |
              MDS_TRAVERSE_WANT_AUTH;
+  if (refpath.depth() == 1 && !mdr->lock_cache_disabled)
+    flags |= MDS_TRAVERSE_CHECK_LOCKCACHE;
   if (create)
     flags |= MDS_TRAVERSE_RDLOCK_AUTHLOCK;
   if (want_layout)
@@ -4296,6 +4299,9 @@ void Server::handle_client_openc(MDRequestRef& mdr)
   if (!check_fragment_space(mdr, dir))
     return;
 
+  if (mdr->dn[0].size() == 1)
+    mds->locker->create_lock_cache(mdr, diri, &mdr->dir_layout);
+
   // create inode.
   CInode *in = prepare_new_inode(mdr, dn->get_dir(), inodeno_t(req->head.ino),
                                 req->head.args.open.mode | S_IFREG, &layout);
@@ -5836,6 +5842,7 @@ void Server::handle_client_mknod(MDRequestRef& mdr)
   if ((mode & S_IFMT) == 0)
     mode |= S_IFREG;
 
+  mdr->disable_lock_cache();
   CDentry *dn = rdlock_path_xlock_dentry(mdr, true, false, S_ISREG(mode));
   if (!dn)
     return;
@@ -5920,6 +5927,7 @@ void Server::handle_client_mkdir(MDRequestRef& mdr)
 {
   const cref_t<MClientRequest> &req = mdr->client_request;
 
+  mdr->disable_lock_cache();
   CDentry *dn = rdlock_path_xlock_dentry(mdr, true);
   if (!dn)
     return;
@@ -6000,6 +6008,7 @@ void Server::handle_client_mkdir(MDRequestRef& mdr)
 
 void Server::handle_client_symlink(MDRequestRef& mdr)
 {
+  mdr->disable_lock_cache();
   CDentry *dn = rdlock_path_xlock_dentry(mdr, true);
   if (!dn)
     return;
@@ -6699,6 +6708,8 @@ void Server::handle_client_unlink(MDRequestRef& mdr)
   // rmdir or unlink?
   bool rmdir = (req->get_op() == CEPH_MDS_OP_RMDIR);
 
+  if (rmdir)
+    mdr->disable_lock_cache();
   CDentry *dn = rdlock_path_xlock_dentry(mdr, false, true);
   if (!dn)
     return;
@@ -6831,6 +6842,9 @@ void Server::handle_client_unlink(MDRequestRef& mdr)
       return;  // we're waiting for a witness.
   }
 
+  if (!rmdir && dnl->is_primary() && mdr->dn[0].size() == 1)
+    mds->locker->create_lock_cache(mdr, diri);
+
   // ok!
   if (dnl->is_remote() && !dnl->get_inode()->is_auth()) 
     _link_remote(mdr, false, dn, dnl->get_inode());