mds: trim log segment after finishing uncommitted fragments

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

4 files changed, 25 insertions, 5 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 9f97f31fcbf..bde5735ea89 100644
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -11191,7 +11191,7 @@ void MDCache::dispatch_fragment_dir(MDRequest *mdr)
   mut->add_updated_lock(&diri->nestlock);
   */
 
-  add_uncommitted_fragment(dirfrag_t(diri->ino(), info.basefrag), info.bits, le->orig_frags);
+  add_uncommitted_fragment(dirfrag_t(diri->ino(), info.basefrag), info.bits, le->orig_frags, mdr->ls);
   mds->mdlog->submit_entry(le, new C_MDC_FragmentPrep(this, mdr));
   mds->mdlog->flush();
 }
@@ -11379,13 +11379,15 @@ void MDCache::handle_fragment_notify(MMDSFragmentNotify *notify)
 }
 
 void MDCache::add_uncommitted_fragment(dirfrag_t basedirfrag, int bits, list<frag_t>& old_frags,
-				       bufferlist *rollback)
+				       LogSegment *ls, bufferlist *rollback)
 {
   dout(10) << "add_uncommitted_fragment: base dirfrag " << basedirfrag << " bits " << bits << dendl;
   assert(!uncommitted_fragments.count(basedirfrag));
   ufragment& uf = uncommitted_fragments[basedirfrag];
   uf.old_frags = old_frags;
   uf.bits = bits;
+  uf.ls = ls;
+  ls->uncommitted_fragments.insert(basedirfrag);
   if (rollback)
     uf.rollback.swap(*rollback);
 }
@@ -11399,6 +11401,8 @@ void MDCache::finish_uncommitted_fragment(dirfrag_t basedirfrag, int op)
     if (op != EFragment::OP_FINISH && !uf.old_frags.empty()) {
       uf.committed = true;
     } else {
+      uf.ls->uncommitted_fragments.erase(basedirfrag);
+      mds->queue_waiters(uf.waiters);
       uncommitted_fragments.erase(basedirfrag);
     }
   }
@@ -11414,6 +11418,7 @@ void MDCache::rollback_uncommitted_fragment(dirfrag_t basedirfrag, list<frag_t>&
       uf.old_frags.swap(old_frags);
       uf.committed = true;
     } else {
+      uf.ls->uncommitted_fragments.erase(basedirfrag);
       uncommitted_fragments.erase(basedirfrag);
     }
   }
diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index b9e7cfa823f..f5c83d5b18d 100644
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -946,9 +946,11 @@ private:
   struct ufragment {
     int bits;
     bool committed;
+    LogSegment *ls;
+    list<Context*> waiters;
     list<frag_t> old_frags;
     bufferlist rollback;
-    ufragment() : bits(0), committed(false) {}
+    ufragment() : bits(0), committed(false), ls(NULL) {}
   };
   map<dirfrag_t, ufragment> uncommitted_fragments;
 
@@ -993,10 +995,14 @@ private:
   void handle_fragment_notify(MMDSFragmentNotify *m);
 
   void add_uncommitted_fragment(dirfrag_t basedirfrag, int bits, list<frag_t>& old_frag,
-				bufferlist *rollback=NULL);
+				LogSegment *ls, bufferlist *rollback=NULL);
   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();
diff --git a/src/mds/journal.cc b/src/mds/journal.cc
index ece1156fc58..41a79f9fb38 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, &rollback);
+    mds->mdcache->add_uncommitted_fragment(dirfrag_t(ino, basefrag), bits, orig_frags, _segment, &rollback);
     // fall-thru
   case OP_ONESHOT:
     if (in)