Client: check dir is still complete after dropping locks in _readdir_cache_cb

We drop the lock when invoking the callback, which means the directory
we're looking at might get dentries trimmed out of memory. Make sure that
hasn't happened after we get the lock back. If it *has* happened, fall back
to requesting the directory contents from the MDS. Update the dirp location
pointers after each entry to facilitate this.
Because this requires we update the dirp->at_cache_name value on every loop,
we rework the updating scheme a bit: to dereference the dn->name before
unlocking, so we know it's filled in; and since we update it on every loop
we don't need to refer to the previous dentry explicitly like we did before.

This should also handle racing file deletes: we get back a trace on
the removed dentry and that will clear the COMPLETE|ORDERED flags.

Fixes #12297

Signed-off-by: Greg Farnum <gfarnum@redhat.com>
(cherry picked from commit 62dd63761701a7e0f7ce39f4071dcabc19bb1cf4)

diff --git a/src/client/Client.cc b/src/client/Client.cc
index 63a9faa93a4ca1f1790dd39039928b981e5e0de0..ae38c9d02b0bb87ca6608034cdd288e08eba86b0 100644 (file)
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -6042,8 +6042,12 @@ int Client::_readdir_cache_cb(dir_result_t *dirp, add_dirent_cb_t cb, void *p)
     ++pd;
   }
 
-  string prev_name;
-  while (!pd.end()) {
+  string dn_name;
+  while (true) {
+    if (!dirp->inode->is_complete_and_ordered())
+      return -EAGAIN;
+    if (pd.end())
+      break;
     Dentry *dn = *pd;
     if (dn->inode == NULL) {
       ldout(cct, 15) << " skipping null '" << dn->name << "'" << dendl;
@@ -6066,6 +6070,8 @@ int Client::_readdir_cache_cb(dir_result_t *dirp, add_dirent_cb_t cb, void *p)
     if (pd.end())
       next_off = dir_result_t::END;
 
+    dn_name = dn->name; // fill in name while we have lock
+
     client_lock.Unlock();
     int r = cb(p, &de, &st, stmask, next_off);  // _next_ offset
     client_lock.Lock();
@@ -6073,13 +6079,12 @@ int Client::_readdir_cache_cb(dir_result_t *dirp, add_dirent_cb_t cb, void *p)
             << " = " << r
             << dendl;
     if (r < 0) {
-      dirp->next_offset = dn->offset;
-      dirp->at_cache_name = prev_name;
+      dirp->next_offset = next_off - 1;
       return r;
     }
 
-    prev_name = dn->name;
-    dirp->offset = next_off;
+    dirp->next_offset = dirp->offset = next_off;
+    dirp->at_cache_name = dn_name; // we successfully returned this one; update!
     if (r > 0)
       return r;
   }