Merge remote-tracking branch 'gh/wip-mds' into next

13 files changed, 209 insertions, 76 deletions

diff --git a/src/mds/CInode.cc b/src/mds/CInode.cc
index c12930837df..af70b681ffc 100644
--- a/src/mds/CInode.cc
+++ b/src/mds/CInode.cc
@@ -130,6 +130,7 @@ ostream& operator<<(ostream& out, CInode& in)
   if (in.state_test(CInode::STATE_DIRTYPARENT)) out << " dirtyparent";
   if (in.is_freezing_inode()) out << " FREEZING=" << in.auth_pin_freeze_allowance;
   if (in.is_frozen_inode()) out << " FROZEN";
+  if (in.is_frozen_auth_pin()) out << " FROZEN_AUTHPIN";
 
   inode_t *pi = in.get_projected_inode();
   if (pi->is_truncating())
@@ -1862,7 +1863,8 @@ void CInode::add_waiter(uint64_t tag, Context *c)
   // wait on the directory?
   //  make sure its not the inode that is explicitly ambiguous|freezing|frozen
   if (((tag & WAIT_SINGLEAUTH) && !state_test(STATE_AMBIGUOUSAUTH)) ||
-      ((tag & WAIT_UNFREEZE) && !is_frozen_inode() && !is_freezing_inode())) {
+      ((tag & WAIT_UNFREEZE) &&
+       !is_frozen_inode() && !is_freezing_inode() && !is_frozen_auth_pin())) {
     dout(15) << "passing waiter up tree" << dendl;
     parent->dir->add_waiter(tag, c);
     return;
@@ -1885,8 +1887,10 @@ bool CInode::freeze_inode(int auth_pin_allowance)
 
   dout(10) << "freeze_inode - frozen" << dendl;
   assert(auth_pins == auth_pin_allowance);
-  get(PIN_FROZEN);
-  state_set(STATE_FROZEN);
+  if (!state_test(STATE_FROZEN)) {
+    get(PIN_FROZEN);
+    state_set(STATE_FROZEN);
+  }
   return true;
 }
 
@@ -1904,10 +1908,34 @@ void CInode::unfreeze_inode(list<Context*>& finished)
   take_waiting(WAIT_UNFREEZE, finished);
 }
 
+void CInode::unfreeze_inode()
+{
+    list<Context*> finished;
+    unfreeze_inode(finished);
+    mdcache->mds->queue_waiters(finished);
+}
+
+void CInode::freeze_auth_pin()
+{
+  assert(state_test(CInode::STATE_FROZEN));
+  state_set(CInode::STATE_FROZENAUTHPIN);
+}
+
+void CInode::unfreeze_auth_pin()
+{
+  assert(state_test(CInode::STATE_FROZENAUTHPIN));
+  state_clear(CInode::STATE_FROZENAUTHPIN);
+  if (!state_test(STATE_FREEZING|STATE_FROZEN)) {
+    list<Context*> finished;
+    take_waiting(WAIT_UNFREEZE, finished);
+    mdcache->mds->queue_waiters(finished);
+  }
+}
 
 // auth_pins
 bool CInode::can_auth_pin() {
-  if (is_freezing_inode() || is_frozen_inode()) return false;
+  if (is_freezing_inode() || is_frozen_inode() || is_frozen_auth_pin())
+    return false;
   if (parent)
     return parent->can_auth_pin();
   return true;
diff --git a/src/mds/CInode.h b/src/mds/CInode.h
index b76b52414c9..e43ecf50fa3 100644
--- a/src/mds/CInode.h
+++ b/src/mds/CInode.h
@@ -181,6 +181,7 @@ public:
   static const int STATE_DIRTYPARENT =  (1<<14);
   static const int STATE_DIRTYRSTAT =  (1<<15);
   static const int STATE_STRAYPINNED = (1<<16);
+  static const int STATE_FROZENAUTHPIN = (1<<17);
 
   // -- waiters --
   static const uint64_t WAIT_DIR         = (1<<0);
@@ -856,6 +857,7 @@ public:
   // -- freeze --
   bool is_freezing_inode() { return state_test(STATE_FREEZING); }
   bool is_frozen_inode() { return state_test(STATE_FROZEN); }
+  bool is_frozen_auth_pin() { return state_test(STATE_FROZENAUTHPIN); }
   bool is_frozen();
   bool is_frozen_dir();
   bool is_freezing();
@@ -864,7 +866,10 @@ public:
    * auth_pins it is itself holding/responsible for. */
   bool freeze_inode(int auth_pin_allowance=0);
   void unfreeze_inode(list<Context*>& finished);
+  void unfreeze_inode();
 
+  void freeze_auth_pin();
+  void unfreeze_auth_pin();
 
   // -- reference counting --
   void bad_put(int by) {
diff --git a/src/mds/Capability.h b/src/mds/Capability.h
index f3743281c90..6fe67f45b1d 100644
--- a/src/mds/Capability.h
+++ b/src/mds/Capability.h
@@ -297,7 +297,8 @@ public:
     int newpending = other.pending | pending();
     if (other.issued & ~newpending)
       issue(other.issued | newpending);
-    issue(newpending);
+    else
+      issue(newpending);
     last_issue_stamp = other.last_issue_stamp;
 
     client_follows = other.client_follows;
@@ -311,7 +312,8 @@ public:
     int newpending = pending();
     if (otherissued & ~newpending)
       issue(otherissued | newpending);
-    issue(newpending);
+    else
+      issue(newpending);
 
     // wanted
     _wanted = _wanted | otherwanted;
diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index c29ac34ec85..ee4799e18f8 100644
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -174,7 +174,8 @@ bool Locker::acquire_locks(MDRequest *mdr,
 			   set<SimpleLock*> &rdlocks,
 			   set<SimpleLock*> &wrlocks,
 			   set<SimpleLock*> &xlocks,
-			   map<SimpleLock*,int> *remote_wrlocks)
+			   map<SimpleLock*,int> *remote_wrlocks,
+			   CInode *auth_pin_freeze)
 {
   if (mdr->done_locking &&
       !mdr->is_slave()) {  // not on slaves!  master requests locks piecemeal.
@@ -196,6 +197,8 @@ bool Locker::acquire_locks(MDRequest *mdr,
     // augment xlock with a versionlock?
     if ((*p)->get_type() == CEPH_LOCK_DN) {
       CDentry *dn = (CDentry*)(*p)->get_parent();
+      if (!dn->is_auth())
+	continue;
 
       if (xlocks.count(&dn->versionlock))
 	continue;  // we're xlocking the versionlock too; don't wrlock it!
@@ -213,6 +216,8 @@ bool Locker::acquire_locks(MDRequest *mdr,
     if ((*p)->get_type() > CEPH_LOCK_IVERSION) {
       // inode version lock?
       CInode *in = (CInode*)(*p)->get_parent();
+      if (!in->is_auth())
+	continue;
       if (mdr->is_master()) {
 	// master.  wrlock versionlock so we can pipeline inode updates to journal.
 	wrlocks.insert(&in->versionlock);
@@ -282,11 +287,12 @@ bool Locker::acquire_locks(MDRequest *mdr,
       continue;
     
     if (!object->is_auth()) {
+      if (!mdr->locks.empty())
+	mds->locker->drop_locks(mdr);
       if (object->is_ambiguous_auth()) {
 	// wait
 	dout(10) << " ambiguous auth, waiting to authpin " << *object << dendl;
 	object->add_waiter(MDSCacheObject::WAIT_SINGLEAUTH, new C_MDS_RetryRequest(mdcache, mdr));
-	mds->locker->drop_locks(mdr);
 	mdr->drop_local_auth_pins();
 	return false;
       }
@@ -331,7 +337,9 @@ bool Locker::acquire_locks(MDRequest *mdr,
 	dout(10) << " req remote auth_pin of " << **q << dendl;
 	MDSCacheObjectInfo info;
 	(*q)->set_object_info(info);
-	req->get_authpins().push_back(info);      
+	req->get_authpins().push_back(info);
+	if (*q == auth_pin_freeze)
+	  (*q)->set_object_info(req->get_authpin_freeze());
 	mdr->pin(*q);
       }
       mds->send_message_mds(req, p->first);
@@ -845,8 +853,8 @@ void Locker::try_eval(MDSCacheObject *p, int mask)
     return;
   }
 
-  if (p->is_auth() && !p->can_auth_pin()) {
-    dout(7) << "try_eval can't auth_pin, waiting on " << *p << dendl;
+  if (p->is_auth() && p->is_frozen()) {
+    dout(7) << "try_eval frozen, waiting on " << *p << dendl;
     p->add_waiter(MDSCacheObject::WAIT_UNFREEZE, new C_Locker_Eval(this, p, mask));
     return;
   }
@@ -3920,6 +3928,7 @@ void Locker::local_wrlock_grab(LocalLock *lock, Mutation *mut)
   dout(7) << "local_wrlock_grab  on " << *lock
 	  << " on " << *lock->get_parent() << dendl;  
   
+  assert(lock->get_parent()->is_auth());
   assert(lock->can_wrlock());
   assert(!mut->wrlocks.count(lock));
   lock->get_wrlock(mut->get_client());
@@ -3932,6 +3941,7 @@ bool Locker::local_wrlock_start(LocalLock *lock, MDRequest *mut)
   dout(7) << "local_wrlock_start  on " << *lock
 	  << " on " << *lock->get_parent() << dendl;  
   
+  assert(lock->get_parent()->is_auth());
   if (lock->can_wrlock()) {
     assert(!mut->wrlocks.count(lock));
     lock->get_wrlock(mut->get_client());
@@ -3963,6 +3973,7 @@ bool Locker::local_xlock_start(LocalLock *lock, MDRequest *mut)
   dout(7) << "local_xlock_start  on " << *lock
 	  << " on " << *lock->get_parent() << dendl;  
   
+  assert(lock->get_parent()->is_auth());
   if (!lock->can_xlock_local()) {
     lock->add_waiter(SimpleLock::WAIT_WR|SimpleLock::WAIT_STABLE, new C_MDS_RetryRequest(mdcache, mut));
     return false;
@@ -4397,8 +4408,12 @@ void Locker::handle_file_lock(ScatterLock *lock, MLock *m)
     if (lock->get_state() == LOCK_MIX_LOCK ||
 	lock->get_state() == LOCK_MIX_LOCK2 ||
 	lock->get_state() == LOCK_MIX_EXCL ||
-	lock->get_state() == LOCK_MIX_TSYN)
+	lock->get_state() == LOCK_MIX_TSYN) {
       lock->decode_locked_state(m->get_data());
+      // replica is waiting for AC_LOCKFLUSHED, eval_gather() should not
+      // delay calling scatter_writebehind().
+      lock->clear_flushed();
+    }
 
     if (lock->is_gathering()) {
       dout(7) << "handle_file_lock " << *in << " from " << from
diff --git a/src/mds/Locker.h b/src/mds/Locker.h
index a1cf59e3185..b3b9919e7fd 100644
--- a/src/mds/Locker.h
+++ b/src/mds/Locker.h
@@ -88,7 +88,8 @@ public:
 		     set<SimpleLock*> &rdlocks,
 		     set<SimpleLock*> &wrlocks,
 		     set<SimpleLock*> &xlocks,
-		     map<SimpleLock*,int> *remote_wrlocks=NULL);
+		     map<SimpleLock*,int> *remote_wrlocks=NULL,
+		     CInode *auth_pin_freeze=NULL);
 
   void cancel_locking(Mutation *mut, set<CInode*> *pneed_issue);
   void drop_locks(Mutation *mut, set<CInode*> *pneed_issue=0);
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 5f0ba16defd..585364d3f05 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -5928,6 +5928,7 @@ void MDCache::handle_cache_expire(MCacheExpire *m)
 	continue;
       }
       assert(!(parent_dir->is_auth() && parent_dir->is_exporting()) ||
+	     migrator->get_export_state(parent_dir) <= Migrator::EXPORT_PREPPING ||
              (migrator->get_export_state(parent_dir) == Migrator::EXPORT_WARNING &&
                  !migrator->export_has_warned(parent_dir, from)));
 
@@ -6706,11 +6707,11 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req, Context *fin,     // wh
     
     // can we conclude ENOENT?
     if (dnl && dnl->is_null()) {
-      if (dn->lock.can_read(client)) {
-        dout(12) << "traverse: miss on null+readable dentry " << path[depth] << " " << *dn << dendl;
+      if (mds->locker->rdlock_try(&dn->lock, client, NULL)) {
+        dout(10) << "traverse: miss on null+readable dentry " << path[depth] << " " << *dn << dendl;
         return -ENOENT;
       } else {
-        dout(12) << "miss on dentry " << *dn << ", can't read due to lock" << dendl;
+        dout(10) << "miss on dentry " << *dn << ", can't read due to lock" << dendl;
         dn->lock.add_waiter(SimpleLock::WAIT_RD, _get_waiter(mdr, req, fin));
         return 1;
       }
@@ -6730,7 +6731,8 @@ int MDCache::path_traverse(MDRequest *mdr, Message *req, Context *fin,     // wh
 	} else {
           dout(7) << "remote link to " << dnl->get_remote_ino() << ", which i don't have" << dendl;
 	  assert(mdr);  // we shouldn't hit non-primary dentries doing a non-mdr traversal!
-          open_remote_ino(dnl->get_remote_ino(), _get_waiter(mdr, req, fin));
+          open_remote_ino(dnl->get_remote_ino(), _get_waiter(mdr, req, fin),
+			  (null_okay && depth == path.depth() - 1));
 	  if (mds->logger) mds->logger->inc(l_mds_trino);
           return 1;
         }        
@@ -7016,12 +7018,13 @@ CInode *MDCache::get_dentry_inode(CDentry *dn, MDRequest *mdr, bool projected)
 class C_MDC_RetryOpenRemoteIno : public Context {
   MDCache *mdcache;
   inodeno_t ino;
+  bool want_xlocked;
   Context *onfinish;
 public:
-  C_MDC_RetryOpenRemoteIno(MDCache *mdc, inodeno_t i, Context *c) :
-    mdcache(mdc), ino(i), onfinish(c) {}
+  C_MDC_RetryOpenRemoteIno(MDCache *mdc, inodeno_t i, Context *c, bool wx) :
+    mdcache(mdc), ino(i), want_xlocked(wx), onfinish(c) {}
   void finish(int r) {
-    mdcache->open_remote_ino(ino, onfinish);
+    mdcache->open_remote_ino(ino, onfinish, want_xlocked);
   }
 };
 
@@ -7031,19 +7034,20 @@ class C_MDC_OpenRemoteIno : public Context {
   inodeno_t ino;
   inodeno_t hadino;
   version_t hadv;
+  bool want_xlocked;
   Context *onfinish;
 public:
   vector<Anchor> anchortrace;
 
-  C_MDC_OpenRemoteIno(MDCache *mdc, inodeno_t i, inodeno_t hi, version_t hv, Context *c) :
-    mdcache(mdc), ino(i), hadino(hi), hadv(hv), onfinish(c) {}
-  C_MDC_OpenRemoteIno(MDCache *mdc, inodeno_t i, vector<Anchor>& at, Context *c) :
-    mdcache(mdc), ino(i), hadino(0), hadv(0), onfinish(c), anchortrace(at) {}
+  C_MDC_OpenRemoteIno(MDCache *mdc, inodeno_t i, bool wx, inodeno_t hi, version_t hv, Context *c) :
+    mdcache(mdc), ino(i), hadino(hi), hadv(hv), want_xlocked(wx), onfinish(c) {}
+  C_MDC_OpenRemoteIno(MDCache *mdc, inodeno_t i, bool wx, vector<Anchor>& at, Context *c) :
+    mdcache(mdc), ino(i), hadino(0), hadv(0), want_xlocked(wx), onfinish(c), anchortrace(at) {}
 
   void finish(int r) {
     assert(r == 0);
     if (r == 0)
-      mdcache->open_remote_ino_2(ino, anchortrace, hadino, hadv, onfinish);
+      mdcache->open_remote_ino_2(ino, anchortrace, want_xlocked, hadino, hadv, onfinish);
     else {
       onfinish->finish(r);
       delete onfinish;
@@ -7051,18 +7055,18 @@ public:
   }
 };
 
-void MDCache::open_remote_ino(inodeno_t ino, Context *onfinish, inodeno_t hadino, version_t hadv)
+void MDCache::open_remote_ino(inodeno_t ino, Context *onfinish, bool want_xlocked,
+			      inodeno_t hadino, version_t hadv)
 {
-  dout(7) << "open_remote_ino on " << ino << dendl;
+  dout(7) << "open_remote_ino on " << ino << (want_xlocked ? " want_xlocked":"") << dendl;
   
-  C_MDC_OpenRemoteIno *c = new C_MDC_OpenRemoteIno(this, ino, hadino, hadv, onfinish);
+  C_MDC_OpenRemoteIno *c = new C_MDC_OpenRemoteIno(this, ino, want_xlocked,
+						   hadino, hadv, onfinish);
   mds->anchorclient->lookup(ino, c->anchortrace, c);
 }
 
-void MDCache::open_remote_ino_2(inodeno_t ino,
-                                vector<Anchor>& anchortrace,
-				inodeno_t hadino, version_t hadv,
-                                Context *onfinish)
+void MDCache::open_remote_ino_2(inodeno_t ino, vector<Anchor>& anchortrace, bool want_xlocked,
+				inodeno_t hadino, version_t hadv, Context *onfinish)
 {
   dout(7) << "open_remote_ino_2 on " << ino
 	  << ", trace depth is " << anchortrace.size() << dendl;
@@ -7105,7 +7109,7 @@ void MDCache::open_remote_ino_2(inodeno_t ino,
 
   if (!in->dirfragtree.contains(frag)) {
     dout(10) << "frag " << frag << " not valid, requerying anchortable" << dendl;
-    open_remote_ino(ino, onfinish);
+    open_remote_ino(ino, onfinish, want_xlocked);
     return;
   }
 
@@ -7115,14 +7119,15 @@ void MDCache::open_remote_ino_2(inodeno_t ino,
     dout(10) << "opening remote dirfrag " << frag << " under " << *in << dendl;
     /* we re-query the anchortable just to avoid a fragtree update race */
     open_remote_dirfrag(in, frag,
-			new C_MDC_RetryOpenRemoteIno(this, ino, onfinish));
+			new C_MDC_RetryOpenRemoteIno(this, ino, onfinish, want_xlocked));
     return;
   }
 
   if (!dir && in->is_auth()) {
     if (in->is_frozen_dir()) {
       dout(7) << "traverse: " << *in << " is frozen_dir, waiting" << dendl;
-      in->parent->dir->add_waiter(CDir::WAIT_UNFREEZE, onfinish);
+      in->parent->dir->add_waiter(CDir::WAIT_UNFREEZE,
+				  new C_MDC_RetryOpenRemoteIno(this, ino, onfinish, want_xlocked));
       return;
     }
     dir = in->get_or_open_dirfrag(this, frag);
@@ -7144,20 +7149,22 @@ void MDCache::open_remote_ino_2(inodeno_t ino,
 		 << " in complete dir " << *dir
 		 << ", requerying anchortable"
 		 << dendl;
-	open_remote_ino(ino, onfinish, anchortrace[i].ino, anchortrace[i].updated);
+	open_remote_ino(ino, onfinish, want_xlocked,
+		        anchortrace[i].ino, anchortrace[i].updated);
       }
     } else {
       dout(10) << "need ino " << anchortrace[i].ino
 	       << ", fetching incomplete dir " << *dir
 	       << dendl;
-      dir->fetch(new C_MDC_OpenRemoteIno(this, ino, anchortrace, onfinish));
+      dir->fetch(new C_MDC_OpenRemoteIno(this, ino, want_xlocked, anchortrace, onfinish));
     }
   } else {
     // hmm, discover.
     dout(10) << "have remote dirfrag " << *dir << ", discovering " 
 	     << anchortrace[i].ino << dendl;
-    discover_ino(dir, anchortrace[i].ino, 
-		 new C_MDC_OpenRemoteIno(this, ino, anchortrace, onfinish));
+    discover_ino(dir, anchortrace[i].ino,
+	         new C_MDC_RetryOpenRemoteIno(this, ino, onfinish, want_xlocked),
+		 (want_xlocked && i == anchortrace.size() - 1));
   }
 }
 
@@ -7476,13 +7483,17 @@ void MDCache::request_finish(MDRequest *mdr)
 
 void MDCache::request_forward(MDRequest *mdr, int who, int port)
 {
-  dout(7) << "request_forward " << *mdr << " to mds." << who << " req " << *mdr << dendl;
-  
-  mds->forward_message_mds(mdr->client_request, who);  
-  mdr->client_request = 0;
+  if (mdr->client_request->get_source().is_client()) {
+    dout(7) << "request_forward " << *mdr << " to mds." << who << " req "
+            << *mdr->client_request << dendl;
+    mds->forward_message_mds(mdr->client_request, who);
+    mdr->client_request = 0;
+    if (mds->logger) mds->logger->inc(l_mds_fw);
+  } else {
+    dout(7) << "request_forward drop " << *mdr << " req " << *mdr->client_request
+            << " was from mds" << dendl;
+  }
   request_cleanup(mdr);
-
-  if (mds->logger) mds->logger->inc(l_mds_fw);
 }
 
 
diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 64290aa97b9..31c7467bf41 100644
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -701,11 +701,11 @@ public:
 
   void open_remote_dirfrag(CInode *diri, frag_t fg, Context *fin);
   CInode *get_dentry_inode(CDentry *dn, MDRequest *mdr, bool projected=false);
-  void open_remote_ino(inodeno_t ino, Context *fin, inodeno_t hadino=0, version_t hadv=0);
+  void open_remote_ino(inodeno_t ino, Context *fin, bool want_xlocked=false,
+		       inodeno_t hadino=0, version_t hadv=0);
   void open_remote_ino_2(inodeno_t ino,
-                         vector<Anchor>& anchortrace,
-			 inodeno_t hadino, version_t hadv,
-                         Context *onfinish);
+			 vector<Anchor>& anchortrace, bool want_xlocked,
+			 inodeno_t hadino, version_t hadv, Context *onfinish);
   void open_remote_dentry(CDentry *dn, bool projected, Context *fin);
   void _open_remote_dentry_finish(int r, CDentry *dn, bool projected, Context *fin);
 
diff --git a/src/mds/Migrator.cc b/src/mds/Migrator.cc
index b66b54cbc83..a804eab7731 100644
--- a/src/mds/Migrator.cc
+++ b/src/mds/Migrator.cc
@@ -1051,6 +1051,7 @@ void Migrator::finish_export_inode_caps(CInode *in)
     mds->send_message_client_counted(m, it->first);
   }
   in->clear_client_caps_after_export();
+  mds->locker->eval(in, CEPH_CAP_LOCKS);
 }
 
 void Migrator::finish_export_inode(CInode *in, utime_t now, list<Context*>& finished)
diff --git a/src/mds/Mutation.cc b/src/mds/Mutation.cc
index 6321ffc160a..a9c35134bc8 100644
--- a/src/mds/Mutation.cc
+++ b/src/mds/Mutation.cc
@@ -82,8 +82,39 @@ void Mutation::auth_unpin(MDSCacheObject *object)
   auth_pins.erase(object);
 }
 
+bool Mutation::freeze_auth_pin(CInode *inode)
+{
+  assert(!auth_pin_freeze || auth_pin_freeze == inode);
+  auth_pin_freeze = inode;
+  auth_pin(inode);
+  if (!inode->freeze_inode(1))
+    return false;
+
+  inode->freeze_auth_pin();
+  inode->unfreeze_inode();
+  return true;
+}
+
+void Mutation::unfreeze_auth_pin(CInode *inode)
+{
+  assert(auth_pin_freeze == inode);
+  assert(is_auth_pinned(inode));
+  if (inode->is_frozen_auth_pin())
+    inode->unfreeze_auth_pin();
+  else
+    inode->unfreeze_inode();
+  auth_pin_freeze = NULL;
+}
+
+bool Mutation::can_auth_pin(MDSCacheObject *object)
+{
+  return object->can_auth_pin() || (is_auth_pinned(object) && object == auth_pin_freeze);
+}
+
 void Mutation::drop_local_auth_pins()
 {
+  if (auth_pin_freeze)
+    unfreeze_auth_pin(auth_pin_freeze);
   for (set<MDSCacheObject*>::iterator it = auth_pins.begin();
        it != auth_pins.end();
        it++) {
diff --git a/src/mds/Mutation.h b/src/mds/Mutation.h
index cba6223864e..37cc764254d 100644
--- a/src/mds/Mutation.h
+++ b/src/mds/Mutation.h
@@ -50,6 +50,7 @@ struct Mutation {
   // auth pins
   set< MDSCacheObject* > remote_auth_pins;
   set< MDSCacheObject* > auth_pins;
+  CInode *auth_pin_freeze;
   
   // held locks
   set< SimpleLock* > rdlocks;  // always local.
@@ -81,12 +82,14 @@ struct Mutation {
     : attempt(0),
       ls(0),
       slave_to_mds(-1),
+      auth_pin_freeze(NULL),
       locking(NULL),
       done_locking(false), committing(false), aborted(false), killed(false) { }
   Mutation(metareqid_t ri, __u32 att=0, int slave_to=-1)
     : reqid(ri), attempt(att),
       ls(0),
       slave_to_mds(slave_to), 
+      auth_pin_freeze(NULL),
       locking(NULL),
       done_locking(false), committing(false), aborted(false), killed(false) { }
   virtual ~Mutation() {
@@ -120,6 +123,9 @@ struct Mutation {
   bool is_auth_pinned(MDSCacheObject *object);
   void auth_pin(MDSCacheObject *object);
   void auth_unpin(MDSCacheObject *object);
+  bool freeze_auth_pin(CInode *inode);
+  void unfreeze_auth_pin(CInode *inode);
+  bool can_auth_pin(MDSCacheObject *object);
   void drop_local_auth_pins();
   void add_projected_inode(CInode *in);
   void pop_and_dirty_projected_inodes();
diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index d5548a8493c..72fd7da2305 100644
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -1487,6 +1487,7 @@ void Server::handle_slave_auth_pin(MDRequest *mdr)
 
   // build list of objects
   list<MDSCacheObject*> objects;
+  CInode *auth_pin_freeze = NULL;
   bool fail = false;
 
   for (vector<MDSCacheObjectInfo>::iterator p = mdr->slave_request->get_authpins().begin();
@@ -1500,6 +1501,8 @@ void Server::handle_slave_auth_pin(MDRequest *mdr)
     }
 
     objects.push_back(object);
+    if (*p == mdr->slave_request->get_authpin_freeze())
+      auth_pin_freeze = dynamic_cast<CInode*>(object);
   }
   
   // can we auth pin them?
@@ -1512,8 +1515,7 @@ void Server::handle_slave_auth_pin(MDRequest *mdr)
 	fail = true;
 	break;
       }
-      if (!mdr->is_auth_pinned(*p) &&
-	  !(*p)->can_auth_pin()) {
+      if (!mdr->can_auth_pin(*p)) {
 	// wait
 	dout(10) << " waiting for authpinnable on " << **p << dendl;
 	(*p)->add_waiter(CDir::WAIT_UNFREEZE, new C_MDS_RetryRequest(mdcache, mdr));
@@ -1527,6 +1529,22 @@ void Server::handle_slave_auth_pin(MDRequest *mdr)
   if (fail) {
     mdr->drop_local_auth_pins();  // just in case
   } else {
+    /* handle_slave_rename_prep() call freeze_inode() to wait for all other operations
+     * on the source inode to complete. This happens after all locks for the rename
+     * operation are acquired. But to acquire locks, we need auth pin locks' parent
+     * objects first. So there is an ABBA deadlock if someone auth pins the source inode
+     * after locks are acquired and before Server::handle_slave_rename_prep() is called.
+     * The solution is freeze the inode and prevent other MDRequests from getting new
+     * auth pins.
+     */
+    if (auth_pin_freeze) {
+      dout(10) << " freezing auth pin on " << *auth_pin_freeze << dendl;
+      if (!mdr->freeze_auth_pin(auth_pin_freeze)) {
+	auth_pin_freeze->add_waiter(CInode::WAIT_FROZEN, new C_MDS_RetryRequest(mdcache, mdr));
+	mds->mdlog->flush();
+	return;
+      }
+    }
     for (list<MDSCacheObject*>::iterator p = objects.begin();
 	 p != objects.end();
 	 ++p) {
@@ -1923,7 +1941,8 @@ CInode* Server::rdlock_path_pin_ref(MDRequest *mdr, int n,
     //   do NOT proceed if freezing, as cap release may defer in that case, and
     //   we could deadlock when we try to lock @ref.
     // if we're already auth_pinned, continue; the release has already been processed.
-    if (ref->is_frozen() || (ref->is_freezing() && !mdr->is_auth_pinned(ref))) {
+    if (ref->is_frozen() || ref->is_frozen_auth_pin() ||
+	(ref->is_freezing() && !mdr->is_auth_pinned(ref))) {
       dout(7) << "waiting for !frozen/authpinnable on " << *ref << dendl;
       ref->add_waiter(CInode::WAIT_UNFREEZE, new C_MDS_RetryRequest(mdcache, mdr));
       /* If we have any auth pins, this will deadlock.
@@ -5202,25 +5221,27 @@ void Server::handle_client_rename(MDRequest *mdr)
     wrlocks.insert(&straydn->get_dir()->inode->nestlock);
   }
 
-  // xlock versionlock on srci if remote?
-  //  this ensures it gets safely remotely auth_pinned, avoiding deadlock;
-  //  strictly speaking, having the slave node freeze the inode is 
-  //  otherwise sufficient for avoiding conflicts with inode locks, etc.
-  if (!srcdn->is_auth() && srcdnl->is_primary())  // xlock versionlock on srci if there are any witnesses
-    xlocks.insert(&srci->versionlock);
-
   // xlock versionlock on dentries if there are witnesses.
   //  replicas can't see projected dentry linkages, and will get
   //  confused if we try to pipeline things.
   if (!witnesses.empty()) {
-    if (srcdn->is_projected())
-      xlocks.insert(&srcdn->versionlock);
-    if (destdn->is_projected())
-      xlocks.insert(&destdn->versionlock);
-    // also take rdlock on all ancestor dentries for destdn.  this ensures that the
-    // destdn can be traversed to by the witnesses.
-    for (int i=0; i<(int)desttrace.size(); i++) 
-      xlocks.insert(&desttrace[i]->versionlock);
+    // take xlock on all projected ancestor dentries for srcdn and destdn.
+    // this ensures the srcdn and destdn can be traversed to by the witnesses.
+    for (int i= 0; i<(int)srctrace.size(); i++) {
+      if (srctrace[i]->is_auth() && srctrace[i]->is_projected())
+	  xlocks.insert(&srctrace[i]->versionlock);
+    }
+    for (int i=0; i<(int)desttrace.size(); i++) {
+      if (desttrace[i]->is_auth() && desttrace[i]->is_projected())
+	  xlocks.insert(&desttrace[i]->versionlock);
+    }
+    // xlock srci and oldin's primary dentries, so witnesses can call
+    // open_remote_ino() with 'want_locked=true' when the srcdn or destdn
+    // is traversed.
+    if (srcdnl->is_remote())
+      xlocks.insert(&srci->get_projected_parent_dn()->lock);
+    if (destdnl->is_remote())
+      xlocks.insert(&oldin->get_projected_parent_dn()->lock);
   }
 
   // we need to update srci's ctime.  xlock its least contended lock to do that...
@@ -5244,7 +5265,9 @@ void Server::handle_client_rename(MDRequest *mdr)
   // 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, &remote_wrlocks))
+  CInode *auth_pin_freeze = !srcdn->is_auth() && srcdnl->is_primary() ? srci : NULL;
+  if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks,
+				  &remote_wrlocks, auth_pin_freeze))
     return;
 
   if (oldin &&
@@ -5994,9 +6017,7 @@ void Server::handle_slave_rename_prep(MDRequest *mdr)
 
   // am i srcdn auth?
   if (srcdn->is_auth()) {
-    if (srcdnl->is_primary() && 
-	!srcdnl->get_inode()->is_freezing_inode() &&
-	!srcdnl->get_inode()->is_frozen_inode()) {
+    if (srcdnl->is_primary()) {
       // set ambiguous auth for srci
       /*
        * NOTE: we don't worry about ambiguous cache expire as we do
@@ -6013,7 +6034,13 @@ void Server::handle_slave_rename_prep(MDRequest *mdr)
       int allowance = 2; // 1 for the mdr auth_pin, 1 for the link lock
       allowance += srcdnl->get_inode()->is_dir(); // for the snap lock
       dout(10) << " freezing srci " << *srcdnl->get_inode() << " with allowance " << allowance << dendl;
-      if (!srcdnl->get_inode()->freeze_inode(allowance)) {
+      bool frozen_inode = srcdnl->get_inode()->freeze_inode(allowance);
+
+      // unfreeze auth pin after freezing the inode to avoid queueing waiters
+      if (srcdnl->get_inode()->is_frozen_auth_pin())
+	mdr->unfreeze_auth_pin(srcdnl->get_inode());
+
+      if (!frozen_inode) {
 	srcdnl->get_inode()->add_waiter(CInode::WAIT_FROZEN, new C_MDS_RetryRequest(mdcache, mdr));
 	return;
       }
@@ -6181,8 +6208,7 @@ void Server::_commit_slave_rename(MDRequest *mdr, int r,
       destdnl->get_inode()->take_waiting(CInode::WAIT_SINGLEAUTH, finished);
 
       // unfreeze
-      assert(destdnl->get_inode()->is_frozen_inode() ||
-             destdnl->get_inode()->is_freezing_inode());
+      assert(destdnl->get_inode()->is_frozen_inode());
       destdnl->get_inode()->unfreeze_inode(finished);
 
       mds->queue_waiters(finished);
@@ -6205,8 +6231,7 @@ void Server::_commit_slave_rename(MDRequest *mdr, int r,
       destdnl->get_inode()->take_waiting(CInode::WAIT_SINGLEAUTH, finished);
       
       // unfreeze
-      assert(destdnl->get_inode()->is_frozen_inode() ||
-	     destdnl->get_inode()->is_freezing_inode());
+      assert(destdnl->get_inode()->is_frozen_inode());
       destdnl->get_inode()->unfreeze_inode(finished);
       
       mds->queue_waiters(finished);
diff --git a/src/mds/mdstypes.h b/src/mds/mdstypes.h
index db4dbf1ac61..22e754eb2a1 100644
--- a/src/mds/mdstypes.h
+++ b/src/mds/mdstypes.h
@@ -1250,6 +1250,13 @@ public:
   }
 };
 
+inline bool operator==(const MDSCacheObjectInfo& l, const MDSCacheObjectInfo& r) {
+  if (l.ino || r.ino)
+    return l.ino == r.ino && l.snapid == r.snapid;
+  else
+    return l.dirfrag == r.dirfrag && l.dname == r.dname;
+}
+
 WRITE_CLASS_ENCODER(MDSCacheObjectInfo)
 
 
diff --git a/src/messages/MMDSSlaveRequest.h b/src/messages/MMDSSlaveRequest.h
index 4f2bb5948bd..03ec582c49e 100644
--- a/src/messages/MMDSSlaveRequest.h
+++ b/src/messages/MMDSSlaveRequest.h
@@ -112,6 +112,7 @@ public:
 
   int get_lock_type() { return lock_type; }
   MDSCacheObjectInfo &get_object_info() { return object_info; }
+  MDSCacheObjectInfo &get_authpin_freeze() { return object_info; }
 
   vector<MDSCacheObjectInfo>& get_authpins() { return authpins; }