client: move dir_{release,ordered}_count into class Inode

We close Inode::dir when it's empty. Once closing the dir, we lose
track of {release,ordered}_count. This causes direcotry to be wrongly
marked as complete. (dir is trimmed to empty in the middle of readdir)

Signed-off-by: Yan, Zheng <zyan@redhat.com>
(cherry picked from commit 235fcf68f5d96a4d9d6cc260d12da912fa7ea4a8)

Signed-off-by: Greg Farnum <gfarnum@redhat.com

diff --git a/src/client/Client.cc b/src/client/Client.cc
index 9755446c400fdd33bf7b2f4fe998b51a376fb237..476a62f93ed71d89f76d68d2e39e3d6e7dc85f45 100644 (file)
--- a/src/client/Client.cc
+++ b/src/client/Client.cc
@@ -713,11 +713,11 @@ void Client::trim_dentry(Dentry *dn)
                 << " in dir " << hex << dn->dir->parent_inode->ino 
                 << dendl;
   if (dn->inode) {
-    dn->dir->release_count++;
-    if (dn->dir->parent_inode->flags & I_COMPLETE) {
-      ldout(cct, 10) << " clearing (I_COMPLETE|I_DIR_ORDERED) on "
-                    << *dn->dir->parent_inode << dendl;
-      dn->dir->parent_inode->flags &= ~(I_COMPLETE | I_DIR_ORDERED);
+    Inode *diri = dn->dir->parent_inode;
+    diri->dir_release_count++;
+    if (diri->flags & I_COMPLETE) {
+      ldout(cct, 10) << " clearing (I_COMPLETE|I_DIR_ORDERED) on " << *diri << dendl;
+      diri->flags &= ~(I_COMPLETE | I_DIR_ORDERED);
     }
   }
   unlink(dn, false, false);  // drop dir, drop dentry
@@ -1002,19 +1002,20 @@ Dentry *Client::insert_dentry_inode(Dir *dir, const string& dname, LeaseStat *dl
     InodeRef tmp_ref(in);
     if (old_dentry) {
       if (old_dentry->dir != dir) {
-       old_dentry->dir->ordered_count++;
-       if (old_dentry->dir->parent_inode->flags & I_DIR_ORDERED) {
-         ldout(cct, 10) << " clearing I_DIR_ORDERED on "
-                        << *old_dentry->dir->parent_inode << dendl;
-         old_dentry->dir->parent_inode->flags &= ~I_DIR_ORDERED;
+       Inode *old_diri = old_dentry->dir->parent_inode;
+       old_diri->dir_ordered_count++;
+       if (old_diri->flags & I_DIR_ORDERED) {
+         ldout(cct, 10) << " clearing I_DIR_ORDERED on " << *old_diri << dendl;
+         old_diri->flags &= ~I_DIR_ORDERED;
        }
       }
       unlink(old_dentry, dir == old_dentry->dir, false);  // drop dentry, keep dir open if its the same dir
     }
-    dir->ordered_count++;
-    if (dir->parent_inode->flags & I_DIR_ORDERED) {
-       ldout(cct, 10) << " clearing I_DIR_ORDERED on " << *dir->parent_inode << dendl;
-       dir->parent_inode->flags &= ~I_DIR_ORDERED;
+    Inode *diri = dir->parent_inode;
+    diri->dir_ordered_count++;
+    if (diri->flags & I_DIR_ORDERED) {
+       ldout(cct, 10) << " clearing I_DIR_ORDERED on " << *diri << dendl;
+       diri->flags &= ~I_DIR_ORDERED;
     }
     dn = link(dir, dname, in, dn);
   }
@@ -1136,6 +1137,12 @@ void Client::insert_readdir_results(MetaRequest *request, MetaSession *session,
                   << ", hash_order=" << hash_order << ", offset " << readdir_offset
                   << ", readdir_start " << readdir_start << dendl;
 
+    if (fg.is_leftmost() && readdir_offset == 2) {
+      dirp->release_count = diri->dir_release_count;
+      dirp->ordered_count = diri->dir_ordered_count;
+      dirp->start_shared_gen = diri->shared_gen;
+    }
+
     dirp->buffer_frag = fg;
 
     _readdir_drop_dirp_buffer(dirp);
@@ -1222,10 +1229,11 @@ Inode* Client::insert_trace(MetaRequest *request, MetaSession *session)
 
     Dentry *d = request->dentry();
     if (d && d->dir) {
-      d->dir->release_count++;
-      if (d->dir->parent_inode->flags & I_COMPLETE) {
-       ldout(cct, 10) << " clearing (I_COMPLETE|I_DIR_ORDERED) on " << *d->dir->parent_inode << dendl;
-       d->dir->parent_inode->flags &= ~(I_COMPLETE | I_DIR_ORDERED);
+      Inode *diri = d->dir->parent_inode;
+      diri->dir_release_count++;
+      if (diri->flags & I_COMPLETE) {
+       ldout(cct, 10) << " clearing (I_COMPLETE|I_DIR_ORDERED) on " << *diri << dendl;
+       diri->flags &= ~(I_COMPLETE | I_DIR_ORDERED);
       }
     }
 
@@ -1304,7 +1312,7 @@ Inode* Client::insert_trace(MetaRequest *request, MetaSession *session)
       if (diri->dir && diri->dir->dentries.count(dname)) {
        Dentry *dn = diri->dir->dentries[dname];
        if (dn->inode) {
-         diri->dir->ordered_count++;
+         diri->dir_ordered_count++;
          if (diri->flags & I_DIR_ORDERED) {
            ldout(cct, 10) << " clearing I_DIR_ORDERED on " << *diri << dendl;
            diri->flags &= ~I_DIR_ORDERED;
@@ -6801,11 +6809,6 @@ int Client::_opendir(Inode *in, dir_result_t **dirpp, int uid, int gid)
     return -ENOTDIR;
   *dirpp = new dir_result_t(in);
   opened_dirs.insert(*dirpp);
-  if (in->dir) {
-    (*dirpp)->release_count = in->dir->release_count;
-    (*dirpp)->ordered_count = in->dir->ordered_count;
-  }
-  (*dirpp)->start_shared_gen = in->shared_gen;
   (*dirpp)->owner_uid = uid;
   (*dirpp)->owner_gid = gid;
   ldout(cct, 3) << "_opendir(" << in->ino << ") = " << 0 << " (" << *dirpp << ")" << dendl;
@@ -7222,10 +7225,9 @@ int Client::readdir_r_cb(dir_result_t *d, add_dirent_cb_t cb, void *p)
       continue;
     }
 
-    if (diri->dir &&
-       diri->shared_gen == dirp->start_shared_gen &&
-       diri->dir->release_count == dirp->release_count) {
-      if (diri->dir->ordered_count == dirp->ordered_count) {
+    if (diri->shared_gen == dirp->start_shared_gen &&
+       diri->dir_release_count == dirp->release_count) {
+      if (diri->dir_ordered_count == dirp->ordered_count) {
        ldout(cct, 10) << " marking (I_COMPLETE|I_DIR_ORDERED) on " << *diri << dendl;
        diri->flags |= I_COMPLETE | I_DIR_ORDERED;
       } else {

diff --git a/src/client/Dir.h b/src/client/Dir.h
index a7f484d45f3c97abc5e6a9a5c23fd66ca79a7de1..608473beb961fde1abfcf7297f13001aa20974b6 100644 (file)
--- a/src/client/Dir.h
+++ b/src/client/Dir.h
@@ -8,10 +8,8 @@ class Dir {
   Inode    *parent_inode;  // my inode
   ceph::unordered_map<string, Dentry*> dentries;
   xlist<Dentry*> dentry_list;
-  uint64_t release_count;
-  uint64_t ordered_count;
 
-  explicit Dir(Inode* in) : release_count(0), ordered_count(0) { parent_inode = in; }
+  explicit Dir(Inode* in) { parent_inode = in; }
 
   bool is_empty() {  return dentries.empty(); }
 };

diff --git a/src/client/Inode.h b/src/client/Inode.h
index c92b103a15af88e04233c494f8a430e95dc94b3e..958e84911bb2766995a6d8529f0ce0ff8172ed51 100644 (file)
--- a/src/client/Inode.h
+++ b/src/client/Inode.h
@@ -226,8 +226,11 @@ struct Inode {
   }
 
   // about the dir (if this is one!)
+  Dir       *dir;     // if i'm a dir.
+  fragtree_t dirfragtree;
   set<int>  dir_contacts;
-  bool      dir_hashed, dir_replicated;
+  uint64_t dir_release_count, dir_ordered_count;
+  bool dir_hashed, dir_replicated;
 
   // per-mds caps
   map<mds_rank_t, Cap*> caps;            // mds -> Cap
@@ -256,10 +259,8 @@ struct Inode {
 
   int       _ref;      // ref count. 1 for each dentry, fh that links to me.
   int       ll_ref;   // separate ref count for ll client
-  Dir       *dir;     // if i'm a dir.
   set<Dentry*> dn_set;      // if i'm linked to a dentry.
   string    symlink;  // symlink content, if it's a symlink
-  fragtree_t dirfragtree;
   map<string,bufferptr> xattrs;
   map<frag_t,int> fragmap;  // known frag -> mds mappings
 
@@ -300,9 +301,8 @@ struct Inode {
       rdev(0), mode(0), uid(0), gid(0), nlink(0),
       size(0), truncate_seq(1), truncate_size(-1),
       time_warp_seq(0), max_size(0), version(0), xattr_version(0),
-      inline_version(0),
-      flags(0),
-      qtree(NULL),
+      inline_version(0), flags(0), qtree(NULL),
+      dir(0), dir_release_count(0), dir_ordered_count(0),
       dir_hashed(false), dir_replicated(false), auth_cap(NULL),
       cap_dirtier_uid(-1), cap_dirtier_gid(-1),
       dirty_caps(0), flushing_caps(0), shared_gen(0), cache_gen(0),
@@ -311,7 +311,7 @@ struct Inode {
       snaprealm(0), snaprealm_item(this),
       oset((void *)this, newlayout->pool_id, ino),
       reported_size(0), wanted_max_size(0), requested_max_size(0),
-      _ref(0), ll_ref(0), dir(0), dn_set(),
+      _ref(0), ll_ref(0), dn_set(),
       fcntl_locks(NULL), flock_locks(NULL),
       async_err(0)
   {