ceph: set dn offset when spliced

We want to assign an offset when the dentry goes from null to linked, which
is always done by splice_dentry().  Notably, we should NOT assign an
offset when a dentry is first created and is still null.

BUG if we try to splice a non-null dentry (we shouldn't).

Signed-off-by: Sage Weil <sage@newdream.net>

diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index 0b0a39d05ca64733e9fd8c07f71339899cacd1f7..d3bb8132a1aac021a04623a249c4290f7f83eff9 100644 (file)
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -128,7 +128,8 @@ more:
        dentry = list_entry(p, struct dentry, d_u.d_child);
        di = ceph_dentry(dentry);
        while (1) {
-               dout(" p %p/%p d_subdirs %p/%p\n", p->prev, p->next,
+               dout(" p %p/%p %s d_subdirs %p/%p\n", p->prev, p->next,
+                    d_unhashed(dentry) ? "!hashed" : "hashed",
                     parent->d_subdirs.prev, parent->d_subdirs.next);
                if (p == &parent->d_subdirs) {
                        fi->at_end = 1;
@@ -571,7 +572,6 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
                    !is_root_ceph_dentry(dir, dentry) &&
                    (ci->i_ceph_flags & CEPH_I_COMPLETE) &&
                    (__ceph_caps_issued_mask(ci, CEPH_CAP_FILE_SHARED, 1))) {
-                       di->offset = ci->i_max_offset++;
                        spin_unlock(&dir->i_lock);
                        dout(" dir %p complete, -ENOENT\n", dir);
                        d_add(dentry, NULL);
@@ -984,8 +984,9 @@ static int ceph_d_revalidate(struct dentry *dentry, struct nameidata *nd)
 {
        struct inode *dir = dentry->d_parent->d_inode;
 
-       dout("d_revalidate %p '%.*s' inode %p\n", dentry,
-            dentry->d_name.len, dentry->d_name.name, dentry->d_inode);
+       dout("d_revalidate %p '%.*s' inode %p offset %lld\n", dentry,
+            dentry->d_name.len, dentry->d_name.name, dentry->d_inode,
+            ceph_dentry(dentry)->offset);
 
        /* always trust cached snapped dentries, snapdir dentry */
        if (ceph_snap(dir) != CEPH_NOSNAP) {
@@ -1177,8 +1178,8 @@ void ceph_dentry_lru_touch(struct dentry *dn)
        struct ceph_dentry_info *di = ceph_dentry(dn);
        struct ceph_mds_client *mdsc;
 
-       dout("dentry_lru_touch %p %p '%.*s'\n", di, dn,
-            dn->d_name.len, dn->d_name.name);
+       dout("dentry_lru_touch %p %p '%.*s' (offset %lld)\n", di, dn,
+            dn->d_name.len, dn->d_name.name, di->offset);
        if (di) {
                mdsc = &ceph_sb_to_client(dn->d_sb)->mdsc;
                spin_lock(&mdsc->dentry_lru_lock);

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index aa22a0bce52fbb6e9d75ecfd0c928eab31ee6d26..1bcf98bd02550f2b7ee125dbe11fc8cf51ee7ed0 100644 (file)
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -803,6 +803,37 @@ out_unlock:
        return;
 }
 
+/*
+ * Set dentry's directory position based on the current dir's max, and
+ * order it in d_subdirs, so that dcache_readdir behaves.
+ */
+static void ceph_set_dentry_offset(struct dentry *dn)
+{
+       struct dentry *dir = dn->d_parent;
+       struct inode *inode = dn->d_parent->d_inode;
+       struct ceph_dentry_info *di;
+
+       BUG_ON(!inode);
+
+       di = ceph_dentry(dn);
+
+       spin_lock(&inode->i_lock);
+       if ((ceph_inode(inode)->i_ceph_flags & CEPH_I_COMPLETE) == 0) {
+               spin_unlock(&inode->i_lock);
+               return;
+       }
+       di->offset = ceph_inode(inode)->i_max_offset++;
+       spin_unlock(&inode->i_lock);
+
+       spin_lock(&dcache_lock);
+       spin_lock(&dn->d_lock);
+       list_move_tail(&dir->d_subdirs, &dn->d_u.d_child);
+       dout("set_dentry_offset %p %lld (%p %p)\n", dn, di->offset,
+            dn->d_u.d_child.prev, dn->d_u.d_child.next);
+       spin_unlock(&dn->d_lock);
+       spin_unlock(&dcache_lock);
+}
+
 /*
  * splice a dentry to an inode.
  * caller must hold directory i_mutex for this to be safe.
@@ -816,6 +847,8 @@ static struct dentry *splice_dentry(struct dentry *dn, struct inode *in,
 {
        struct dentry *realdn;
 
+       BUG_ON(dn->d_inode);
+
        /* dn must be unhashed */
        if (!d_unhashed(dn))
                d_drop(dn);
@@ -837,47 +870,16 @@ static struct dentry *splice_dentry(struct dentry *dn, struct inode *in,
                dn = realdn;
        } else {
                BUG_ON(!ceph_dentry(dn));
-
                dout("dn %p attached to %p ino %llx.%llx\n",
                     dn, dn->d_inode, ceph_vinop(dn->d_inode));
        }
        if ((!prehash || *prehash) && d_unhashed(dn))
                d_rehash(dn);
+       ceph_set_dentry_offset(dn);
 out:
        return dn;
 }
 
-/*
- * Set dentry's directory position based on the current dir's max, and
- * order it in d_subdirs, so that dcache_readdir behaves.
- */
-static void ceph_set_dentry_offset(struct dentry *dn)
-{
-       struct dentry *dir = dn->d_parent;
-       struct inode *inode = dn->d_parent->d_inode;
-       struct ceph_dentry_info *di;
-
-       BUG_ON(!inode);
-
-       di = ceph_dentry(dn);
-
-       spin_lock(&inode->i_lock);
-       if ((ceph_inode(inode)->i_ceph_flags & CEPH_I_COMPLETE) == 0) {
-               spin_unlock(&inode->i_lock);
-               return;
-       }
-       di->offset = ceph_inode(inode)->i_max_offset++;
-       spin_unlock(&inode->i_lock);
-
-       spin_lock(&dcache_lock);
-       spin_lock(&dn->d_lock);
-       list_move_tail(&dir->d_subdirs, &dn->d_u.d_child);
-       dout("set_dentry_offset %p %lld (%p %p)\n", dn, di->offset,
-            dn->d_u.d_child.prev, dn->d_u.d_child.next);
-       spin_unlock(&dn->d_lock);
-       spin_unlock(&dcache_lock);
-}
-
 /*
  * Incorporate results into the local cache.  This is either just
  * one inode, or a directory, dentry, and possibly linked-to inode (e.g.,
@@ -1030,6 +1032,9 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                        ceph_invalidate_dentry_lease(dn);
 
                        /* take overwritten dentry's readdir offset */
+                       dout("dn %p gets %p offset %lld (old offset %lld)\n",
+                            req->r_old_dentry, dn, ceph_dentry(dn)->offset,
+                            ceph_dentry(req->r_old_dentry)->offset);
                        ceph_dentry(req->r_old_dentry)->offset =
                                ceph_dentry(dn)->offset;
 
@@ -1074,7 +1079,6 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                                goto done;
                        }
                        req->r_dentry = dn;  /* may have spliced */
-                       ceph_set_dentry_offset(dn);
                        igrab(in);
                } else if (ceph_ino(in) == vino.ino &&
                           ceph_snap(in) == vino.snap) {
@@ -1117,7 +1121,6 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                        err = PTR_ERR(dn);
                        goto done;
                }
-               ceph_set_dentry_offset(dn);
                req->r_dentry = dn;  /* may have spliced */
                igrab(in);
                rinfo->head->is_dentry = 1;  /* fool notrace handlers */