From: Yan, Zheng <zheng.z.yan@intel.com>
Date: Fri, 28 Mar 2014 17:53:15 +0000 (+0800)
Subject: mds: commit new dirfrag before splitting it
X-Git-Tag: v0.79~52^2~1
X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=07ccc4e7bda67c4af0e4dc24aa0ecbf35016c3c3;p=ceph.git

mds: commit new dirfrag before splitting it

Commit 6e013cd6 (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>
---

diff --git a/src/mds/CDir.h b/src/mds/CDir.h
index 30edd28506ad3..50ac4424f25f8 100644
--- a/src/mds/CDir.h
+++ b/src/mds/CDir.h
@@ -219,6 +219,7 @@ public:
   void log_mark_dirty();
   void mark_clean();
 
+  bool is_new() { return item_new.is_on_list(); }
   void mark_new(LogSegment *ls);
 
 public:
diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index bffc7cc4a0d19..9e9a835ee5974 100644
--- a/src/mds/MDCache.cc
+++ b/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)) {
diff --git a/src/mds/MDCache.h b/src/mds/MDCache.h
index 7b46bd9e3ef43..422b94d90b256 100644
--- a/src/mds/MDCache.h
+++ b/src/mds/MDCache.h
@@ -957,12 +957,11 @@ private:
   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;