mds: acquire anchor_create locks in caller

We're not allowed to call acquire_locks again once we've completed locking.
Do it in the caller.  Add a helper that will also take an rdlock to ensure
the anchor doesn't go away due to a racing operation.

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index ab399ba6057bd433ec9a053a3d7fce3f34aa0157..4307e01cf1e09f0c92500a538250191736d9e76c 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -6540,6 +6540,27 @@ public:
   }
 };
 
+void MDCache::anchor_create_prep_locks(MDRequest *mdr, CInode *in,
+                                      set<SimpleLock*>& rdlocks, set<SimpleLock*>& xlocks)
+{
+  dout(10) << "anchor_create_prep_locks " << *in << dendl;
+
+  if (in->is_anchored()) {
+    // caller may have already xlocked it.. if so, that will suffice!
+    if (xlocks.count(&in->linklock) == 0)
+      rdlocks.insert(&in->linklock);
+  } else {
+    xlocks.insert(&in->linklock);
+
+    // path components too!
+    CDentry *dn = in->get_projected_parent_dn();
+    while (dn) {
+      rdlocks.insert(&dn->lock);
+      dn = dn->get_dir()->get_inode()->get_parent_dn();
+    }
+  }
+}
+
 void MDCache::anchor_create(MDRequest *mdr, CInode *in, Context *onfinish)
 {
   assert(in->is_auth());
@@ -6553,22 +6574,6 @@ void MDCache::anchor_create(MDRequest *mdr, CInode *in, Context *onfinish)
     return;
   }
 
-  // rdlock path
-  set<SimpleLock*> rdlocks = mdr->rdlocks;
-  set<SimpleLock*> wrlocks = mdr->wrlocks;
-  set<SimpleLock*> xlocks = mdr->xlocks;
-  xlocks.insert(&in->linklock);
-
-  CDentry *dn = in->get_parent_dn();
-  while (dn) {
-    rdlocks.insert(&dn->lock);
-    dn = dn->get_dir()->get_inode()->get_parent_dn();
-  }
-  if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks)) {
-   dout(7) << "anchor_create waiting for locks  " << *in << dendl;
-   return;
-  }
-
   // wait
   in->add_waiter(CInode::WAIT_ANCHORED, onfinish);
 

diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 3bece87f4403a17312a0a0011e76322d5bafed82..625a8be48f0a1a7826366bf02d4ec3c8129a2385 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -969,6 +969,8 @@ public:
   
   // -- anchors --
 public:
+  void anchor_create_prep_locks(MDRequest *mdr, CInode *in, set<SimpleLock*>& rdlocks,
+                               set<SimpleLock*>& xlocks);
   void anchor_create(MDRequest *mdr, CInode *in, Context *onfinish);
   void anchor_destroy(CInode *in, Context *onfinish);
 protected:

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 6ad3fd39066c93438e4ffdc3bef8f71a98779dca..cf941f14a5be6ed828664ebd1c12b99714e4664a 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -3100,6 +3100,10 @@ void Server::handle_client_link(MDRequest *mdr)
   }
   
   xlocks.insert(&targeti->linklock);
+
+  // take any locks needed for anchor creation/verification
+  mds->mdcache->anchor_create_prep_locks(mdr, targeti, rdlocks, xlocks);
+
   if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
     return;
 
@@ -3369,6 +3373,17 @@ void Server::handle_slave_link_prep(MDRequest *mdr)
 
   // anchor?
   if (mdr->slave_request->get_op() == MMDSSlaveRequest::OP_LINKPREP) {
+    
+    set<SimpleLock*> rdlocks = mdr->rdlocks;
+    set<SimpleLock*> wrlocks = mdr->wrlocks;
+    set<SimpleLock*> xlocks = mdr->xlocks;
+
+    // take any locks needed for anchor creation/verification
+    mds->mdcache->anchor_create_prep_locks(mdr, targeti, rdlocks, xlocks);
+
+    if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
+      return;
+
     if (targeti->is_anchored()) {
       dout(7) << "target anchored already (nlink=" << targeti->inode.nlink << "), sweet" << dendl;
     } 
@@ -4157,6 +4172,9 @@ void Server::handle_client_rename(MDRequest *mdr)
   else
     rdlocks.insert(&srci->snaplock);
 
+  // take any locks needed for anchor creation/verification
+  mds->mdcache->anchor_create_prep_locks(mdr, srci, rdlocks, xlocks);
+
   if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
     return;