mds: reset heartbeat when fetching or committing dentries

Fixes: https://tracker.ceph.com/issues/54345
Signed-off-by: Xiubo Li <xiubli@redhat.com>

diff --git a/src/mds/CDir.cc b/src/mds/CDir.cc
index ad7b16053a3effbf27aa162e26392c5ab43323e3..aee737f82b69b8bb21d6b9985ba5302d92488896 100644 (file)
--- a/src/mds/CDir.cc
+++ b/src/mds/CDir.cc
@@ -2106,6 +2106,8 @@ void CDir::_omap_fetched(bufferlist& hdrbl, map<string, bufferlist>& omap,
   };
   auto proc_nulls_and_waiters = [&](const string& str_key, const string_snap_t& key) {
     bool touch = false;
+    int count = 0;
+
     while (k_it != keys.rend()) {
       int cmp = k_it->compare(str_key);
       if (cmp < 0)
@@ -2123,10 +2125,14 @@ void CDir::_omap_fetched(bufferlist& hdrbl, map<string, bufferlist>& omap,
       last_name = std::string_view(k_it->c_str(), n_key.name.length());
       null_keys.emplace_back(std::move(n_key));
       ++k_it;
+
+      if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+        mdcache->mds->heartbeat_reset();
     }
     return touch;
   };
 
+  int count = 0;
   unsigned pos = omap.size() - 1;
   double rand_threshold = get_inode()->get_ephemeral_rand();
   for (auto p = omap.rbegin(); p != omap.rend(); ++p, --pos) {
@@ -2143,6 +2149,9 @@ void CDir::_omap_fetched(bufferlist& hdrbl, map<string, bufferlist>& omap,
       last_name = std::string_view(p->first.c_str(), key.name.length());
     }
 
+    if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+      mdcache->mds->heartbeat_reset();
+
     CDentry *dn = nullptr;
     try {
       dn = _load_dentry(
@@ -2204,6 +2213,9 @@ void CDir::_omap_fetched(bufferlist& hdrbl, map<string, bufferlist>& omap,
        dout(12) << "_fetched got null for key " << key << ", added " << *dn << dendl;
       }
       mdcache->touch_dentry(dn);
+
+      if (!(++count % mdcache->mds->heartbeat_reset_grace(2)))
+        mdcache->mds->heartbeat_reset();
     }
   }
 
@@ -2219,9 +2231,13 @@ void CDir::_omap_fetched(bufferlist& hdrbl, map<string, bufferlist>& omap,
   // open & force frags
   while (!undef_inodes.empty()) {
     CInode *in = undef_inodes.front();
+
     undef_inodes.pop_front();
     in->state_clear(CInode::STATE_REJOINUNDEF);
     mdcache->opened_undef_inode(in);
+
+    if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+      mdcache->mds->heartbeat_reset();
   }
 
   // dirty myself to remove stale snap dentries
@@ -2451,6 +2467,7 @@ void CDir::_omap_commit_ops(int r, int op_prio, int64_t metapool, version_t vers
     _rm.clear();
   };
 
+  int count = 0;
   for (auto &key : stales) {
     unsigned size = key.length() + sizeof(__u32);
     if (write_size + size > max_write_size)
@@ -2458,6 +2475,9 @@ void CDir::_omap_commit_ops(int r, int op_prio, int64_t metapool, version_t vers
 
     write_size += size;
     _rm.emplace(key);
+
+    if (!(++count % mdcache->mds->heartbeat_reset_grace(2)))
+      mdcache->mds->heartbeat_reset();
   }
 
   for (auto &key : to_remove) {
@@ -2467,6 +2487,9 @@ void CDir::_omap_commit_ops(int r, int op_prio, int64_t metapool, version_t vers
 
     write_size += size;
     _rm.emplace(std::move(key));
+
+    if (!(++count % mdcache->mds->heartbeat_reset_grace(2)))
+      mdcache->mds->heartbeat_reset();
   }
 
   bufferlist bl;
@@ -2492,6 +2515,9 @@ void CDir::_omap_commit_ops(int r, int op_prio, int64_t metapool, version_t vers
 
     write_size += size;
     _set[std::move(item.key)].swap(bl);
+
+    if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+      mdcache->mds->heartbeat_reset();
   }
 
   commit_one(true);
@@ -2570,6 +2596,7 @@ void CDir::_omap_commit(int op_prio)
     }
   };
 
+  int count = 0;
   if (state_test(CDir::STATE_FRAGMENTING) && is_new()) {
     ceph_assert(committed_version == 0);
     for (auto p = items.begin(); p != items.end(); ) {
@@ -2578,12 +2605,18 @@ void CDir::_omap_commit(int op_prio)
       if (dn->get_linkage()->is_null())
        continue;
       write_one(dn);
+
+      if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+        mdcache->mds->heartbeat_reset();
     }
   } else {
     for (auto p = dirty_dentries.begin(); !p.end(); ) {
       CDentry *dn = *p;
       ++p;
       write_one(dn);
+
+      if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+        mdcache->mds->heartbeat_reset();
     }
   }
 
@@ -2735,6 +2768,8 @@ void CDir::_committed(int r, version_t v)
   if (committed_version == get_version()) 
     mark_clean();
 
+  int count = 0;
+
   // dentries clean?
   for (auto p = dirty_dentries.begin(); !p.end(); ) {
     CDentry *dn = *p;
@@ -2772,11 +2807,14 @@ void CDir::_committed(int r, version_t v)
       dout(15) << " dir " << committed_version << " < dn " << dn->get_version() << " still dirty " << *dn << dendl;
       ceph_assert(dn->is_dirty());
     }
+
+    if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+      mdcache->mds->heartbeat_reset();
   }
 
   // finishers?
   bool were_waiters = !waiting_for_commit.empty();
-  
+
   auto it = waiting_for_commit.begin();
   while (it != waiting_for_commit.end()) {
     auto _it = it;
@@ -2792,7 +2830,10 @@ void CDir::_committed(int r, version_t v)
     mdcache->mds->queue_waiters(t);
     waiting_for_commit.erase(it);
     it = _it;
-  } 
+
+    if (!(++count % mdcache->mds->heartbeat_reset_grace()))
+      mdcache->mds->heartbeat_reset();
+  }
 
   // try drop dentries in this dirfrag if it's about to be purged
   if (!inode->is_base() && get_parent_dir()->inode->is_stray() &&