mds: trim log segment after finishing uncommitted fragments

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

4 files changed, 38 insertions, 22 deletions

diff --git a/src/mds/LogSegment.h b/src/mds/LogSegment.h
index 723267da116..624c3bc2395 100644
--- a/src/mds/LogSegment.h
+++ b/src/mds/LogSegment.h
@@ -56,6 +56,7 @@ class LogSegment {
 
   map<int, hash_set<version_t> > pending_commit_tids;  // mdstable
   set<metareqid_t> uncommitted_masters;
+  set<dirfrag_t> uncommitted_fragments;
 
   // client request ids
   map<int, tid_t> last_client_tids;
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index f973ec7b731..1296e8abb1a 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -11177,7 +11177,7 @@ void MDCache::dispatch_fragment_dir(MDRequest *mdr)
   mdr->ls->dirty_dirfrag_dirfragtree.push_back(&diri->item_dirty_dirfrag_dirfragtree);
   mdr->add_updated_lock(&diri->dirfragtreelock);
 
-  add_uncommitted_fragment(dirfrag_t(diri->ino(), info.basefrag), info.bits, old_frags);
+  add_uncommitted_fragment(dirfrag_t(diri->ino(), info.basefrag), info.bits, old_frags, mdr->ls);
   mds->mdlog->submit_entry(le, new C_MDC_FragmentPrep(this, mdr));
   mds->mdlog->flush();
 }
@@ -11364,13 +11364,16 @@ void MDCache::handle_fragment_notify(MMDSFragmentNotify *notify)
   notify->put();
 }
 
-void MDCache::add_uncommitted_fragment(dirfrag_t basedirfrag, int bits, list<frag_t>& old_frags)
+void MDCache::add_uncommitted_fragment(dirfrag_t basedirfrag, int bits,
+				       list<frag_t>& old_frags, LogSegment *ls)
 {
   dout(10) << "add_uncommitted_fragment: base dirfrag " << basedirfrag << " bits " << bits << dendl;
   assert(!uncommitted_fragments.count(basedirfrag));
   ufragment& uf = uncommitted_fragments[basedirfrag];
   uf.old_frags.swap(old_frags);
   uf.bits = bits;
+  uf.ls = ls;
+  ls->uncommitted_fragments.insert(basedirfrag);
 }
 
 void MDCache::finish_uncommitted_fragment(dirfrag_t basedirfrag, int op)
@@ -11379,10 +11382,13 @@ void MDCache::finish_uncommitted_fragment(dirfrag_t basedirfrag, int op)
 	   << " op " << EFragment::op_name(op) << dendl;
   if (uncommitted_fragments.count(basedirfrag)) {
     ufragment& uf = uncommitted_fragments[basedirfrag];
-    if (op != EFragment::OP_FINISH && !uf.old_frags.empty())
+    if (op != EFragment::OP_FINISH && !uf.old_frags.empty()) {
       uf.committed = true;
-    else
+    } else {
+      uf.ls->uncommitted_fragments.erase(basedirfrag);
+      mds->queue_waiters(uf.waiters);
       uncommitted_fragments.erase(basedirfrag);
+    }
   }
 }
 
@@ -11395,8 +11401,10 @@ void MDCache::rollback_uncommitted_fragment(dirfrag_t basedirfrag, list<frag_t>&
     if (!uf.old_frags.empty()) {
       uf.old_frags.swap(old_frags);
       uf.committed = true;
-    } else
+    } else {
+      uf.ls->uncommitted_fragments.erase(basedirfrag);
       uncommitted_fragments.erase(basedirfrag);
+    }
   }
 }
 
diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 50ae258cc2d..cd6ff2a2a7a 100644
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -946,8 +946,10 @@ private:
   struct ufragment {
     int bits;
     bool committed;
+    LogSegment *ls;
     list<frag_t> old_frags;
-    ufragment() : bits(0), committed(false) {}
+    list<Context*> waiters;
+    ufragment() : bits(0), committed(false), ls(NULL) {}
   };
   map<dirfrag_t, ufragment> uncommitted_fragments;
 
@@ -970,16 +972,7 @@ private:
   CDir *force_dir_fragment(CInode *diri, frag_t fg);
   void get_force_dirfrag_bound_set(vector<dirfrag_t>& dfs, set<CDir*>& bounds);
 
-
-  friend class EFragment;
-
   bool can_fragment(CInode *diri, list<CDir*>& dirs);
-
-public:
-  void split_dir(CDir *dir, int byn);
-  void merge_dir(CInode *diri, frag_t fg);
-
-private:
   void fragment_freeze_dirs(list<CDir*>& dirs, C_GatherBuilder &gather);
   void fragment_mark_and_complete(list<CDir*>& dirs);
   void fragment_frozen(list<CDir*>& dirs, frag_t basefrag, int bits);
@@ -990,10 +983,7 @@ private:
   void _fragment_committed(dirfrag_t f, list<CDir*>& resultfrags);
   void _fragment_finish(dirfrag_t f, list<CDir*>& resultfrags);
 
-public:
-  void rollback_uncommitted_fragments();
-private:
-
+  friend class EFragment;
   friend class C_MDC_FragmentFrozen;
   friend class C_MDC_FragmentMarking;
   friend class C_MDC_FragmentPrep;
@@ -1003,9 +993,18 @@ private:
 
   void handle_fragment_notify(MMDSFragmentNotify *m);
 
-  void add_uncommitted_fragment(dirfrag_t basedirfrag, int bits, list<frag_t>& old_frag);
-  void finish_uncommitted_fragment(dirfrag_t basedirfrag);
+  void add_uncommitted_fragment(dirfrag_t basedirfrag, int bits, list<frag_t>& old_frags,
+                                LogSegment *ls);
+  void finish_uncommitted_fragment(dirfrag_t basedirfrag, int op);
   void rollback_uncommitted_fragment(dirfrag_t basedirfrag, list<frag_t>& old_frags);
+public:
+  void wait_for_uncommitted_fragment(dirfrag_t dirfrag, Context *c) {
+    assert(uncommitted_fragments.count(dirfrag));
+    uncommitted_fragments[dirfrag].waiters.push_back(c);
+  }
+  void split_dir(CDir *dir, int byn);
+  void merge_dir(CInode *diri, frag_t fg);
+  void rollback_uncommitted_fragments();
 
   // -- updates --
   //int send_inode_updates(CInode *in);
diff --git a/src/mds/journal.cc b/src/mds/journal.cc
index df730db4d5c..81d9ab7d2fe 100644
--- a/src/mds/journal.cc
+++ b/src/mds/journal.cc
@@ -119,6 +119,14 @@ void LogSegment::try_to_expire(MDS *mds, C_GatherBuilder &gather_bld)
     mds->mdcache->wait_for_uncommitted_master(*p, gather_bld.new_sub());
   }
 
+  // uncommitted fragments
+  for (set<dirfrag_t>::iterator p = uncommitted_fragments.begin();
+       p != uncommitted_fragments.end();
+       ++p) {
+    dout(10) << "try_to_expire waiting for uncommitted fragment " << *p << dendl;
+    mds->mdcache->wait_for_uncommitted_fragment(*p, gather_bld.new_sub());
+  }
+
   // nudge scatterlocks
   for (elist<CInode*>::iterator p = dirty_dirfrag_dir.begin(); !p.end(); ++p) {
     CInode *in = *p;
@@ -2390,7 +2398,7 @@ void EFragment::replay(MDS *mds)
 
   switch (op) {
   case OP_PREPARE:
-    mds->mdcache->add_uncommitted_fragment(dirfrag_t(ino, basefrag), bits, orig_frags);
+    mds->mdcache->add_uncommitted_fragment(dirfrag_t(ino, basefrag), bits, orig_frags, _segment);
     // fall-thru
   case OP_ONESHOT:
     if (in)