mds: include authpinned objects in remote authpin request

Server::handle_slave_auth_pin() may drop old authpins if it encounters
object that is not authpinable. So it is better to include objects that
have already been authpinned in the remote authpin reuqest.

Signed-off-by: Yan, Zheng <zheng.z.yan@intel.com>

diff --git a/src/mds/Locker.cc b/src/mds/Locker.cc
index 4f6140dfa6281ade9fd28f4eb5338be4f0cb6eb4..2f0f5bdb234ccb2ca3f96bfd5a95a8956ecb9397 100644 (file)
--- a/src/mds/Locker.cc
+++ b/src/mds/Locker.cc
@@ -187,12 +187,13 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
   client_t client = mdr->get_client();
 
   set<SimpleLock*, SimpleLock::ptr_lt> sorted;  // sort everything we will lock
-  set<SimpleLock*> mustpin = xlocks;            // items to authpin
+  set<MDSCacheObject*> mustpin;            // items to authpin
 
   // xlocks
   for (set<SimpleLock*>::iterator p = xlocks.begin(); p != xlocks.end(); ++p) {
     dout(20) << " must xlock " << **p << " " << *(*p)->get_parent() << dendl;
     sorted.insert(*p);
+    mustpin.insert((*p)->get_parent());
 
     // augment xlock with a versionlock?
     if ((*p)->get_type() == CEPH_LOCK_DN) {
@@ -232,26 +233,28 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
 
   // wrlocks
   for (set<SimpleLock*>::iterator p = wrlocks.begin(); p != wrlocks.end(); ++p) {
-    dout(20) << " must wrlock " << **p << " " << *(*p)->get_parent() << dendl;
+    MDSCacheObject *object = (*p)->get_parent();
+    dout(20) << " must wrlock " << **p << " " << *object << dendl;
     sorted.insert(*p);
-    if ((*p)->get_parent()->is_auth())
-      mustpin.insert(*p);
-    else if (!(*p)->get_parent()->is_auth() &&
+    if (object->is_auth())
+      mustpin.insert(object);
+    else if (!object->is_auth() &&
             !(*p)->can_wrlock(client) &&  // we might have to request a scatter
             !mdr->is_slave()) {           // if we are slave (remote_wrlock), the master already authpinned
-      dout(15) << " will also auth_pin " << *(*p)->get_parent()
+      dout(15) << " will also auth_pin " << *object
               << " in case we need to request a scatter" << dendl;
-      mustpin.insert(*p);
+      mustpin.insert(object);
     }
   }
 
   // remote_wrlocks
   if (remote_wrlocks) {
     for (map<SimpleLock*,int>::iterator p = remote_wrlocks->begin(); p != remote_wrlocks->end(); ++p) {
+      MDSCacheObject *object = p->first->get_parent();
       dout(20) << " must remote_wrlock on mds." << p->second << " "
-              << *p->first << " " << *(p->first)->get_parent() << dendl;
+              << *p->first << " " << *object << dendl;
       sorted.insert(p->first);
-      mustpin.insert(p->first);
+      mustpin.insert(object);
     }
   }
 
@@ -259,15 +262,16 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
   for (set<SimpleLock*>::iterator p = rdlocks.begin();
         p != rdlocks.end();
        ++p) {
-    dout(20) << " must rdlock " << **p << " " << *(*p)->get_parent() << dendl;
+    MDSCacheObject *object = (*p)->get_parent();
+    dout(20) << " must rdlock " << **p << " " << *object << dendl;
     sorted.insert(*p);
-    if ((*p)->get_parent()->is_auth())
-      mustpin.insert(*p);
-    else if (!(*p)->get_parent()->is_auth() &&
+    if (object->is_auth())
+      mustpin.insert(object);
+    else if (!object->is_auth() &&
             !(*p)->can_rdlock(client)) {      // we might have to request an rdlock
-      dout(15) << " will also auth_pin " << *(*p)->get_parent()
+      dout(15) << " will also auth_pin " << *object
               << " in case we need to request a rdlock" << dendl;
-      mustpin.insert(*p);
+      mustpin.insert(object);
     }
   }
 
@@ -276,10 +280,10 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
   map<int, set<MDSCacheObject*> > mustpin_remote;  // mds -> (object set)
   
   // can i auth pin them all now?
-  for (set<SimpleLock*>::iterator p = mustpin.begin();
+  for (set<MDSCacheObject*>::iterator p = mustpin.begin();
        p != mustpin.end();
        ++p) {
-    MDSCacheObject *object = (*p)->get_parent();
+    MDSCacheObject *object = *p;
 
     dout(10) << " must authpin " << *object << dendl;
 
@@ -315,10 +319,10 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
   }
 
   // ok, grab local auth pins
-  for (set<SimpleLock*>::iterator p = mustpin.begin();
+  for (set<MDSCacheObject*>::iterator p = mustpin.begin();
        p != mustpin.end();
        ++p) {
-    MDSCacheObject *object = (*p)->get_parent();
+    MDSCacheObject *object = *p;
     if (mdr->is_auth_pinned(object)) {
       dout(10) << " already auth_pinned " << *object << dendl;
     } else if (object->is_auth()) {
@@ -329,6 +333,16 @@ bool Locker::acquire_locks(MDRequestRef& mdr,
 
   // request remote auth_pins
   if (!mustpin_remote.empty()) {
+    for (set<MDSCacheObject*>::iterator p = mdr->remote_auth_pins.begin();
+        p != mdr->remote_auth_pins.end();
+        ++p) {
+      if (mustpin.count(*p)) {
+       int auth = (*p)->authority().first;
+       map<int, set<MDSCacheObject*> >::iterator q = mustpin_remote.find(auth);
+       if (q != mustpin_remote.end())
+         q->second.insert(*p);
+      }
+    }
     for (map<int, set<MDSCacheObject*> >::iterator p = mustpin_remote.begin();
         p != mustpin_remote.end();
         ++p) {