mds: allow delay in evaluating stray

Add a new parameter 'delay' to MDCache::eval_stray(). If 'delay'
is true, MDCache::eval_stray() adds the stray dentry to a delayed
list. Delayed stray dentries are processed in MDCache::trim().

This change is required by later commit that evaluates stray when
reference to cache object is released.

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

3 files changed, 29 insertions, 11 deletions

diff --git a/src/mds/CDentry.h b/src/mds/CDentry.h
index 0d2445a525f..c540a810456 100644
--- a/src/mds/CDentry.h
+++ b/src/mds/CDentry.h
@@ -146,6 +146,7 @@ protected:
 
 public:
   elist<CDentry*>::item item_dirty;
+  elist<CDentry*>::item item_stray;
 
 protected:
   int auth_pins, nested_auth_pins;
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 7bbbacee234..4773e113dad 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -127,7 +127,8 @@ long g_num_caps = 0;
 set<int> SimpleLock::empty_gather_set;
 
 
-MDCache::MDCache(MDS *m)
+MDCache::MDCache(MDS *m) :
+  delayed_eval_stray(member_offset(CDentry, item_stray))
 {
   mds = m;
   migrator = new Migrator(mds, this);
@@ -6018,8 +6019,15 @@ bool MDCache::trim(int max)
   }
   dout(7) << "trim max=" << max << "  cur=" << lru.lru_get_size() << dendl;
 
-  map<int, MCacheExpire*> expiremap;
+  // process delayed eval_stray()
+  for (elist<CDentry*>::iterator p = delayed_eval_stray.begin(); !p.end(); ) {
+    CDentry *dn = *p;
+    ++p;
+    dn->item_stray.remove_myself();
+    eval_stray(dn);
+  }
 
+  map<int, MCacheExpire*> expiremap;
   bool is_standby_replay = mds->is_standby_replay();
   int unexpirable = 0;
   list<CDentry*> unexpirables;
@@ -9148,7 +9156,7 @@ struct C_MDC_EvalStray : public Context {
   }
 };
 
-void MDCache::eval_stray(CDentry *dn)
+void MDCache::eval_stray(CDentry *dn, bool delay)
 {
   dout(10) << "eval_stray " << *dn << dendl;
   CDentry::linkage_t *dnl = dn->get_projected_linkage();
@@ -9212,7 +9220,11 @@ void MDCache::eval_stray(CDentry *dn)
       dout(20) << " too many dn refs" << dendl;
       return;
     }
-    purge_stray(dn);
+    if (delay) {
+      if (!dn->item_stray.is_on_list())
+	delayed_eval_stray.push_back(&dn->item_stray);
+    } else
+      purge_stray(dn);
   }
   else if (in->inode.nlink >= 1) {
     // trivial reintegrate?
@@ -9382,7 +9394,9 @@ void MDCache::purge_stray(CDentry *dn)
   dn->get(CDentry::PIN_PURGING);
   in->state_set(CInode::STATE_PURGING);
 
-  
+  if (dn->item_stray.is_on_list())
+    dn->item_stray.remove_myself();
+
   // CHEAT.  there's no real need to journal our intent to purge, since
   // that is implicit in the dentry's presence and non-use in the stray
   // dir.  on recovery, we'll need to re-eval all strays anyway.
diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 1673ad8076a..3cd85825275 100644
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -19,6 +19,7 @@
 
 #include "include/types.h"
 #include "include/filepath.h"
+#include "include/elist.h"
 
 #include "CInode.h"
 #include "CDentry.h"
@@ -867,18 +868,20 @@ public:
 
   // -- stray --
 public:
+  elist<CDentry*> delayed_eval_stray;
+
   void scan_stray_dir();
-  void eval_stray(CDentry *dn);
+  void eval_stray(CDentry *dn, bool delay=false);
   void eval_remote(CDentry *dn);
 
-  void maybe_eval_stray(CInode *in) {
+  void maybe_eval_stray(CInode *in, bool delay=false) {
     if (in->inode.nlink > 0 || in->is_base())
       return;
     CDentry *dn = in->get_projected_parent_dn();
-    if (dn->get_projected_linkage()->is_primary() &&
-	dn->get_dir()->get_inode()->is_stray() &&
-	!dn->is_replicated())
-      eval_stray(dn);
+    if (!dn->state_test(CDentry::STATE_PURGING) &&
+	dn->get_projected_linkage()->is_primary() &&
+	dn->get_dir()->get_inode()->is_stray())
+      eval_stray(dn, delay);
   }
 protected:
   void fetch_backtrace(inodeno_t ino, int64_t pool, bufferlist& bl, Context *fin);