mds: commit new dirfrag before splitting it

Commit 6e013cd (properly set COMPLETE flag when merging dirfrags)
tries solving the issue that new dirfrag's COMPLETE flag gets lost
if MDS splits the new dirfrag, then the fragment operation gets
rolled back. It records the original dirfrag's COMPLETE flag when
EFragment PREPARE event is encountered. If the fragment operation
needs to rollback, The COMPLETE flag is journaled in corresponding
EFragment ROLLBACK event. This is problematic when the ROLLBACK
event and the "mkdir" event belong to different log segments. After
the log segment that contains the "mkdir" event is trimmed, the
dirfrag can not be considered as complete.

The fix is commit new dirfrag before splitting it. After dirfrag is
committed to object store, losing COMPLETE flag is not a big deal.

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

src/mds/CDir.h

@@ -219,6 +219,7 @@ class CDir : public MDSCacheObject {
   void log_mark_dirty();
   void mark_clean();
 
+  bool is_new() { return item_new.is_on_list(); }
   void mark_new(LogSegment *ls);
 
 public:

src/mds/MDCache.cc

@@ -11325,12 +11325,25 @@ void MDCache::fragment_mark_and_complete(list<CDir*>& dirs)
        ++p) {
     CDir *dir = *p;
 
+    bool ready = true;
     if (!dir->is_complete()) {
       dout(15) << " fetching incomplete " << *dir << dendl;
-      dir->fetch(gather.new_sub(),
-		 true);  // ignore authpinnability
-    } 
-    else if (!dir->state_test(CDir::STATE_DNPINNEDFRAG)) {
+      dir->fetch(gather.new_sub(), true);  // ignore authpinnability
+      ready = false;
+    }
+    if (dir->get_frag() == frag_t() && dir->is_new()) {
+      // The COMPLETE flag gets lost if we fragment a new dirfrag, then rollback
+      // the operation. To avoid CDir::fetch() complaining about missing object,
+      // we commit new dirfrag first.
+      dout(15) << " committing new " << *dir << dendl;
+      assert(dir->is_dirty());
+      dir->commit(0, gather.new_sub(), true);
+      ready = false;
+    }
+    if (!ready)
+      continue;
+
+    if (!dir->state_test(CDir::STATE_DNPINNEDFRAG)) {
       dout(15) << " marking " << *dir << dendl;
       for (CDir::map_t::iterator p = dir->items.begin();
 	   p != dir->items.end();
@@ -11342,8 +11355,7 @@ void MDCache::fragment_mark_and_complete(list<CDir*>& dirs)
       }
       dir->state_set(CDir::STATE_DNPINNEDFRAG);
       dir->auth_unpin(dir);
-    }
-    else {
+    } else {
       dout(15) << " already marked " << *dir << dendl;
     }
   }
@@ -11833,15 +11845,8 @@ void MDCache::add_uncommitted_fragment(dirfrag_t basedirfrag, int bits, list<fra
   uf.bits = bits;
   uf.ls = ls;
   ls->uncommitted_fragments.insert(basedirfrag);
-  if (rollback) {
+  if (rollback)
     uf.rollback.swap(*rollback);
-    // preserve COMPLETE flag for newly created dirfrag
-    if (bits > 0 && basedirfrag.frag == frag_t()) {
-      CDir *dir = get_dirfrag(basedirfrag);
-      if (dir && dir->is_complete())
-	uf.complete = true;
-    }
-  }
 }
 
 void MDCache::finish_uncommitted_fragment(dirfrag_t basedirfrag, int op)
@@ -11927,8 +11932,6 @@ void MDCache::rollback_uncommitted_fragments()
 	dir->set_version(rollback.fnode.version);
 	dir->fnode = rollback.fnode;
 
-	if (uf.complete)
-	  dir->mark_complete();
 	dir->_mark_dirty(ls);
 
 	if (!(dir->fnode.rstat == dir->fnode.accounted_rstat)) {

src/mds/MDCache.h

@@ -957,12 +957,11 @@ class MDCache {
   struct ufragment {
     int bits;
     bool committed;
-    bool complete;
     LogSegment *ls;
     list<Context*> waiters;
     list<frag_t> old_frags;
     bufferlist rollback;
-    ufragment() : bits(0), committed(false), complete(false), ls(NULL) {}
+    ufragment() : bits(0), committed(false), ls(NULL) {}
   };
   map<dirfrag_t, ufragment> uncommitted_fragments;