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>

diff --git a/src/client/Client.cc b/src/client/Client.cc
index f5c97a128949c0d73d05f9a11dc97ba600d72f32..abadc8bf296c8ab0697829b4e256967d8b1e04ce 100644 (file)
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -6254,8 +6254,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;
@@ -6278,6 +6282,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();
@@ -6285,13 +6291,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;
   }