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 1f1bc11216754cfa873996e823f07de48ab7e80e..1655a79651859f691d46ae662988d043acb8a8c1 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -6542,6 +6542,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());
@@ -6555,22 +6576,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 ae4d7f39d9cc78169878961665007ef10a890e66..9c1c0af7b80dd0d229162fa3acb3d7871e7c3951 100644 (file)
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -971,6 +971,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 bde9d236e12a32a8e68c76fa240ccf51d550a9b6..b0195bb970c6ff7d36e30c2dd7ac320fa847aa66 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -3106,6 +3106,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;
 
@@ -3375,6 +3379,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;
     } 
@@ -4144,6 +4159,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;