update sources to v12.1.0

[ceph.git] / ceph / src / mds / CDir.cc

diff --git a/ceph/src/mds/CDir.cc b/ceph/src/mds/CDir.cc
index 03ba644631ecc431eb02faaaea934d8de42a9db0..6561d2fb257c4f6eecdaf8a3d2bfc7741e687fae 100644 (file)
--- a/ceph/src/mds/CDir.cc
+++ b/ceph/src/mds/CDir.cc
@@ -328,7 +328,9 @@ CDentry* CDir::add_null_dentry(const string& dname,
   CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), first, last);
   if (is_auth()) 
     dn->state_set(CDentry::STATE_AUTH);
-  cache->lru.lru_insert_mid(dn);
+
+  cache->bottom_lru.lru_insert_mid(dn);
+  dn->state_set(CDentry::STATE_BOTTOMLRU);
 
   dn->dir = this;
   dn->version = get_projected_version();
@@ -369,7 +371,12 @@ CDentry* CDir::add_primary_dentry(const string& dname, CInode *in,
   CDentry* dn = new CDentry(dname, inode->hash_dentry_name(dname), first, last);
   if (is_auth()) 
     dn->state_set(CDentry::STATE_AUTH);
-  cache->lru.lru_insert_mid(dn);
+  if (is_auth() || !inode->is_stray()) {
+    cache->lru.lru_insert_mid(dn);
+  } else {
+    cache->bottom_lru.lru_insert_mid(dn);
+    dn->state_set(CDentry::STATE_BOTTOMLRU);
+  }
 
   dn->dir = this;
   dn->version = get_projected_version();
@@ -482,7 +489,10 @@ void CDir::remove_dentry(CDentry *dn)
   if (dn->is_dirty())
     dn->mark_clean();
 
-  cache->lru.lru_remove(dn);
+  if (dn->state_test(CDentry::STATE_BOTTOMLRU))
+    cache->bottom_lru.lru_remove(dn);
+  else
+    cache->lru.lru_remove(dn);
   delete dn;
 
   // unpin?
@@ -503,6 +513,12 @@ void CDir::link_remote_inode(CDentry *dn, inodeno_t ino, unsigned char d_type)
 
   dn->get_linkage()->set_remote(ino, d_type);
 
+  if (dn->state_test(CDentry::STATE_BOTTOMLRU)) {
+    cache->bottom_lru.lru_remove(dn);
+    cache->lru.lru_insert_mid(dn);
+    dn->state_clear(CDentry::STATE_BOTTOMLRU);
+  }
+
   if (dn->last == CEPH_NOSNAP) {
     num_head_items++;
     num_head_null--;
@@ -522,6 +538,13 @@ void CDir::link_primary_inode(CDentry *dn, CInode *in)
   in->set_primary_parent(dn);
 
   link_inode_work(dn, in);
+
+  if (dn->state_test(CDentry::STATE_BOTTOMLRU) &&
+      (is_auth() || !inode->is_stray())) {
+    cache->bottom_lru.lru_remove(dn);
+    cache->lru.lru_insert_mid(dn);
+    dn->state_clear(CDentry::STATE_BOTTOMLRU);
+  }
   
   if (dn->last == CEPH_NOSNAP) {
     num_head_items++;
@@ -557,7 +580,7 @@ void CDir::link_inode_work( CDentry *dn, CInode *in)
     in->move_to_realm(inode->find_snaprealm());
 }
 
-void CDir::unlink_inode(CDentry *dn)
+void CDir::unlink_inode(CDentry *dn, bool adjust_lru)
 {
   if (dn->get_linkage()->is_primary()) {
     dout(12) << "unlink_inode " << *dn << " " << *dn->get_linkage()->get_inode() << dendl;
@@ -567,6 +590,12 @@ void CDir::unlink_inode(CDentry *dn)
 
   unlink_inode_work(dn);
 
+  if (adjust_lru && !dn->state_test(CDentry::STATE_BOTTOMLRU)) {
+    cache->lru.lru_remove(dn);
+    cache->bottom_lru.lru_insert_mid(dn);
+    dn->state_set(CDentry::STATE_BOTTOMLRU);
+  }
+
   if (dn->last == CEPH_NOSNAP) {
     num_head_items--;
     num_head_null++;
@@ -683,7 +712,7 @@ void CDir::remove_null_dentries() {
 void CDir::try_remove_dentries_for_stray()
 {
   dout(10) << __func__ << dendl;
-  assert(inode->inode.nlink == 0);
+  assert(get_parent_dir()->inode->is_stray());
 
   // clear dirty only when the directory was not snapshotted
   bool clear_dirty = !inode->snaprealm;
@@ -726,15 +755,6 @@ void CDir::try_remove_dentries_for_stray()
     mark_clean();
 }
 
-void CDir::touch_dentries_bottom() {
-  dout(12) << "touch_dentries_bottom " << *this << dendl;
-
-  for (CDir::map_t::iterator p = items.begin();
-       p != items.end();
-       ++p)
-    inode->mdcache->touch_dentry_bottom(p->second);
-}
-
 bool CDir::try_trim_snap_dentry(CDentry *dn, const set<snapid_t>& snaps)
 {
   assert(dn->last != CEPH_NOSNAP);
@@ -849,12 +869,19 @@ void CDir::steal_dentry(CDentry *dn)
   dn->dir = this;
 }
 
-void CDir::prepare_old_fragment(bool replay)
+void CDir::prepare_old_fragment(map<string_snap_t, std::list<MDSInternalContextBase*> >& dentry_waiters, bool replay)
 {
   // auth_pin old fragment for duration so that any auth_pinning
   // during the dentry migration doesn't trigger side effects
   if (!replay && is_auth())
     auth_pin(this);
+
+  if (!waiting_on_dentry.empty()) {
+    for (auto p = waiting_on_dentry.begin(); p != waiting_on_dentry.end(); ++p)
+      dentry_waiters[p->first].swap(p->second);
+    waiting_on_dentry.clear();
+    put(PIN_DNWAITER);
+  }
 }
 
 void CDir::prepare_new_fragment(bool replay)
@@ -863,6 +890,7 @@ void CDir::prepare_new_fragment(bool replay)
     _freeze_dir();
     mark_complete();
   }
+  inode->add_dirfrag(this);
 }
 
 void CDir::finish_old_fragment(list<MDSInternalContextBase*>& waiters, bool replay)
@@ -940,7 +968,8 @@ void CDir::split(int bits, list<CDir*>& subs, list<MDSInternalContextBase*>& wai
     fragstatdiff.add_delta(fnode.accounted_fragstat, fnode.fragstat);
   dout(10) << " rstatdiff " << rstatdiff << " fragstatdiff " << fragstatdiff << dendl;
 
-  prepare_old_fragment(replay);
+  map<string_snap_t, std::list<MDSInternalContextBase*> > dentry_waiters;
+  prepare_old_fragment(dentry_waiters, replay);
 
   // create subfrag dirs
   int n = 0;
@@ -966,7 +995,6 @@ void CDir::split(int bits, list<CDir*>& subs, list<MDSInternalContextBase*>& wai
     dout(10) << " subfrag " << *p << " " << *f << dendl;
     subfrags[n++] = f;
     subs.push_back(f);
-    inode->add_dirfrag(f);
 
     f->set_dir_auth(get_dir_auth());
     f->prepare_new_fragment(replay);
@@ -984,6 +1012,16 @@ void CDir::split(int bits, list<CDir*>& subs, list<MDSInternalContextBase*>& wai
     f->steal_dentry(dn);
   }
 
+  for (auto& p : dentry_waiters) {
+    frag_t subfrag = inode->pick_dirfrag(p.first.name);
+    int n = (subfrag.value() & (subfrag.mask() ^ frag.mask())) >> subfrag.mask_shift();
+    CDir *f = subfrags[n];
+
+    if (f->waiting_on_dentry.empty())
+      f->get(PIN_DNWAITER);
+    f->waiting_on_dentry[p.first].swap(p.second);
+  }
+
   // FIXME: handle dirty old rstat
 
   // fix up new frag fragstats
@@ -1028,6 +1066,8 @@ void CDir::merge(list<CDir*>& subs, list<MDSInternalContextBase*>& waiters, bool
   version_t rstat_version = inode->get_projected_inode()->rstat.version;
   version_t dirstat_version = inode->get_projected_inode()->dirstat.version;
 
+  map<string_snap_t, std::list<MDSInternalContextBase*> > dentry_waiters;
+
   for (auto dir : subs) {
     dout(10) << " subfrag " << dir->get_frag() << " " << *dir << dendl;
     assert(!dir->is_auth() || dir->is_complete() || replay);
@@ -1038,7 +1078,7 @@ void CDir::merge(list<CDir*>& subs, list<MDSInternalContextBase*>& waiters, bool
       fragstatdiff.add_delta(dir->fnode.accounted_fragstat, dir->fnode.fragstat,
                             &touched_mtime, &touched_chattr);
 
-    dir->prepare_old_fragment(replay);
+    dir->prepare_old_fragment(dentry_waiters, replay);
 
     // steal dentries
     while (!dir->items.empty()) 
@@ -1059,12 +1099,18 @@ void CDir::merge(list<CDir*>& subs, list<MDSInternalContextBase*>& waiters, bool
 
     // merge state
     state_set(dir->get_state() & MASK_STATE_FRAGMENT_KEPT);
-    dir_auth = dir->dir_auth;
 
     dir->finish_old_fragment(waiters, replay);
     inode->close_dirfrag(dir->get_frag());
   }
 
+  if (!dentry_waiters.empty()) {
+    get(PIN_DNWAITER);
+    for (auto& p : dentry_waiters) {
+      waiting_on_dentry[p.first].swap(p.second);
+    }
+  }
+
   if (is_auth() && !replay)
     mark_complete();
 
@@ -1425,9 +1471,11 @@ void CDir::fetch(MDSInternalContextBase *c, const string& want_dn, bool ignore_a
   }
 
   // unlinked directory inode shouldn't have any entry
-  if (inode->inode.nlink == 0 && !inode->snaprealm) {
+  if (!inode->is_base() && get_parent_dir()->inode->is_stray() &&
+      !inode->snaprealm) {
     dout(7) << "fetch dirfrag for unlinked directory, mark complete" << dendl;
     if (get_version() == 0) {
+      assert(inode->is_auth());
       set_version(1);
 
       if (state_test(STATE_REJOINUNDEF)) {
@@ -2003,14 +2051,6 @@ void CDir::commit(version_t want, MDSInternalContextBase *c, bool ignore_authpin
   assert(is_auth());
   assert(ignore_authpinnability || can_auth_pin());
 
-  if (inode->inode.nlink == 0 && !inode->snaprealm) {
-    dout(7) << "commit dirfrag for unlinked directory, mark clean" << dendl;
-    try_remove_dentries_for_stray();
-    if (c)
-      cache->mds->queue_waiter(c);
-    return;
-  }
-
   // note: queue up a noop if necessary, so that we always
   // get an auth_pin.
   if (!c)
@@ -2268,9 +2308,10 @@ void CDir::_committed(int r, version_t v)
   if (r < 0) {
     // the directory could be partly purged during MDS failover
     if (r == -ENOENT && committed_version == 0 &&
-       inode->inode.nlink == 0 && inode->snaprealm) {
-      inode->state_set(CInode::STATE_MISSINGOBJS);
+       !inode->is_base() && get_parent_dir()->inode->is_stray()) {
       r = 0;
+      if (inode->snaprealm)
+       inode->state_set(CInode::STATE_MISSINGOBJS);
     }
     if (r < 0) {
       dout(1) << "commit error " << r << " v " << v << dendl;
@@ -2361,7 +2402,8 @@ void CDir::_committed(int r, version_t v)
   } 
 
   // try drop dentries in this dirfrag if it's about to be purged
-  if (inode->inode.nlink == 0 && inode->snaprealm)
+  if (!inode->is_base() && get_parent_dir()->inode->is_stray() &&
+      inode->snaprealm)
     cache->maybe_eval_stray(inode, true);
 
   // unpin if we kicked the last waiter.