mds: properly update freeze_tree_state when fragmenting dirfrags

The bug was introduced by commit 6de90905 'mds: cleanup nested auth
pins tracking'

Signed-off-by: "Yan, Zheng" <zyan@redhat.com>

diff --git a/src/mds/CDir.cc b/src/mds/CDir.cc
index 3973fd6b432a6eccdca09c019f1224c1f07fc8f9..856bc70b592fa2a10ec77cf013c580e9683a2c00 100644 (file)
--- a/src/mds/CDir.cc
+++ b/src/mds/CDir.cc
@@ -1036,6 +1036,7 @@ void CDir::split(int bits, list<CDir*>& subs, MDSInternalContextBase::vec& waite
     subs.push_back(f);
 
     f->set_dir_auth(get_dir_auth());
+    f->freeze_tree_state = freeze_tree_state;
     f->prepare_new_fragment(replay);
     f->init_fragment_pins();
   }
@@ -1091,16 +1092,14 @@ void CDir::merge(list<CDir*>& subs, MDSInternalContextBase::vec& waiters, bool r
 {
   dout(10) << "merge " << subs << dendl;
 
-  mds_authority_t new_auth = CDIR_AUTH_DEFAULT;
+  set_dir_auth(subs.front()->get_dir_auth());
+  freeze_tree_state = subs.front()->freeze_tree_state;
+
   for (auto dir : subs) {
-    if (dir->get_dir_auth() != CDIR_AUTH_DEFAULT &&
-       dir->get_dir_auth() != new_auth) {
-      ceph_assert(new_auth == CDIR_AUTH_DEFAULT);
-      new_auth = dir->get_dir_auth();
-    }
+    ceph_assert(get_dir_auth() == dir->get_dir_auth());
+    ceph_assert(freeze_tree_state == dir->freeze_tree_state);
   }
 
-  set_dir_auth(new_auth);
   prepare_new_fragment(replay);
 
   nest_info_t rstatdiff;
@@ -2670,6 +2669,11 @@ void CDir::set_dir_auth(const mds_authority_t &a)
   if (!was_subtree && is_subtree_root()) {
     dout(10) << " new subtree root, adjusting auth_pins" << dendl;
 
+    if (freeze_tree_state) {
+      // only by CDir::_freeze_tree()
+      ceph_assert(is_freezing_tree_root());
+    }
+
     inode->num_subtree_roots++;   
     
     // unpin parent of frozen dir/tree?
@@ -2903,12 +2907,6 @@ void CDir::_freeze_tree()
   }
   freeze_tree_state->frozen = true;
 
-  // twiddle state
-  if (state_test(STATE_FREEZINGTREE)) {
-    state_clear(STATE_FREEZINGTREE);   // actually, this may get set again by next context?
-    --num_freezing_trees;
-  }
-
   if (is_auth()) {
     mds_authority_t auth;
     bool was_subtree = is_subtree_root();
@@ -2946,6 +2944,12 @@ void CDir::_freeze_tree()
     );
   }
 
+  // twiddle state
+  if (state_test(STATE_FREEZINGTREE)) {
+    state_clear(STATE_FREEZINGTREE);
+    --num_freezing_trees;
+  }
+
   state_set(STATE_FROZENTREE);
   ++num_frozen_trees;
   get(PIN_FROZEN);

diff --git a/src/mds/MDCache.cc b/src/mds/MDCache.cc
index 71a6c828e10049a4f660989b4ec734ef150aa7c7..0a961fd21871e104c16441bbc920ba400c2bdf60 100644 (file)
--- a/src/mds/MDCache.cc
+++ b/src/mds/MDCache.cc
@@ -10915,13 +10915,15 @@ void MDCache::adjust_dir_fragments(CInode *diri,
 
     // are my constituent bits subtrees?  if so, i will be too.
     // (it's all or none, actually.)
-    bool any_subtree = false;
+    bool any_subtree = false, any_non_subtree = false;
     for (CDir *dir : srcfrags) {
-      if (dir->is_subtree_root()) {
+      if (dir->is_subtree_root())
        any_subtree = true;
-       break;
-      }
+      else
+       any_non_subtree = true;
     }
+    ceph_assert(!any_subtree || !any_non_subtree);
+
     set<CDir*> new_bounds;
     if (any_subtree)  {
       for (CDir *dir : srcfrags) {
@@ -11101,12 +11103,29 @@ void MDCache::merge_dir(CInode *diri, frag_t frag)
 
 void MDCache::fragment_freeze_dirs(list<CDir*>& dirs)
 {
-  for (list<CDir*>::iterator p = dirs.begin(); p != dirs.end(); ++p) {
-    CDir *dir = *p;
+  bool any_subtree = false, any_non_subtree = false;
+  for (CDir* dir : dirs) {
     dir->auth_pin(dir);  // until we mark and complete them
     dir->state_set(CDir::STATE_FRAGMENTING);
     dir->freeze_dir();
     ceph_assert(dir->is_freezing_dir());
+
+    if (dir->is_subtree_root())
+      any_subtree = true;
+    else
+      any_non_subtree = true;
+  }
+
+  if (any_subtree && any_non_subtree) {
+    // either all dirfrags are subtree roots or all are not.
+    for (CDir *dir : dirs) {
+      if (dir->is_subtree_root()) {
+       ceph_assert(dir->state_test(CDir::STATE_AUXSUBTREE));
+      } else {
+       dir->state_set(CDir::STATE_AUXSUBTREE);
+       adjust_subtree_auth(dir, mds->get_nodeid());
+      }
+    }
   }
 }