1 files changed, 108 insertions, 63 deletions

diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 62377ec37edf..d8858e96ab18 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -1,8 +1,7 @@
-#include "ceph_debug.h"
+#include <linux/ceph/ceph_debug.h>
 
 #include <linux/module.h>
 #include <linux/fs.h>
-#include <linux/smp_lock.h>
 #include <linux/slab.h>
 #include <linux/string.h>
 #include <linux/uaccess.h>
@@ -13,7 +12,8 @@
 #include <linux/pagevec.h>
 
 #include "super.h"
-#include "decode.h"
+#include "mds_client.h"
+#include <linux/ceph/decode.h>
 
 /*
  * Ceph inode operations
@@ -36,6 +36,13 @@ static void ceph_vmtruncate_work(struct work_struct *work);
 /*
  * find or create an inode, given the ceph ino number
  */
+static int ceph_set_ino_cb(struct inode *inode, void *data)
+{
+        ceph_inode(inode)->i_vino = *(struct ceph_vino *)data;
+        inode->i_ino = ceph_vino_to_ino(*(struct ceph_vino *)data);
+        return 0;
+}
+
 struct inode *ceph_get_inode(struct super_block *sb, struct ceph_vino vino)
 {
         struct inode *inode;
@@ -297,6 +304,8 @@ struct inode *ceph_alloc_inode(struct super_block *sb)
         ci->i_release_count = 0;
         ci->i_symlink = NULL;
 
+        memset(&ci->i_dir_layout, 0, sizeof(ci->i_dir_layout));
+
         ci->i_fragtree = RB_ROOT;
         mutex_init(&ci->i_fragtree_mutex);
 
@@ -346,6 +355,7 @@ struct inode *ceph_alloc_inode(struct super_block *sb)
         ci->i_rd_ref = 0;
         ci->i_rdcache_ref = 0;
         ci->i_wr_ref = 0;
+        ci->i_wb_ref = 0;
         ci->i_wrbuffer_ref = 0;
         ci->i_wrbuffer_ref_head = 0;
         ci->i_shared_gen = 0;
@@ -368,6 +378,15 @@ struct inode *ceph_alloc_inode(struct super_block *sb)
         return &ci->vfs_inode;
 }
 
+static void ceph_i_callback(struct rcu_head *head)
+{
+        struct inode *inode = container_of(head, struct inode, i_rcu);
+        struct ceph_inode_info *ci = ceph_inode(inode);
+
+        INIT_LIST_HEAD(&inode->i_dentry);
+        kmem_cache_free(ceph_inode_cachep, ci);
+}
+
 void ceph_destroy_inode(struct inode *inode)
 {
         struct ceph_inode_info *ci = ceph_inode(inode);
@@ -384,7 +403,7 @@ void ceph_destroy_inode(struct inode *inode)
          */
         if (ci->i_snap_realm) {
                 struct ceph_mds_client *mdsc =
-                        &ceph_sb_to_client(ci->vfs_inode.i_sb)->mdsc;
+                        ceph_sb_to_client(ci->vfs_inode.i_sb)->mdsc;
                 struct ceph_snap_realm *realm = ci->i_snap_realm;
 
                 dout(" dropping residual ref to snap realm %p\n", realm);
@@ -407,7 +426,7 @@ void ceph_destroy_inode(struct inode *inode)
         if (ci->i_xattrs.prealloc_blob)
                 ceph_buffer_put(ci->i_xattrs.prealloc_blob);
 
-        kmem_cache_free(ceph_inode_cachep, ci);
+        call_rcu(&inode->i_rcu, ceph_i_callback);
 }
 
 
@@ -470,7 +489,9 @@ void ceph_fill_file_time(struct inode *inode, int issued,
 
         if (issued & (CEPH_CAP_FILE_EXCL|
                       CEPH_CAP_FILE_WR|
-                      CEPH_CAP_FILE_BUFFER)) {
+                      CEPH_CAP_FILE_BUFFER|
+                      CEPH_CAP_AUTH_EXCL|
+                      CEPH_CAP_XATTR_EXCL)) {
                 if (timespec_compare(ctime, &inode->i_ctime) > 0) {
                         dout("ctime %ld.%09ld -> %ld.%09ld inc w/ cap\n",
                              inode->i_ctime.tv_sec, inode->i_ctime.tv_nsec,
@@ -510,7 +531,7 @@ void ceph_fill_file_time(struct inode *inode, int issued,
                         warn = 1;
                 }
         } else {
-                /* we have no write caps; whatever the MDS says is true */
+                /* we have no write|excl caps; whatever the MDS says is true */
                 if (ceph_seq_cmp(time_warp_seq, ci->i_time_warp_seq) >= 0) {
                         inode->i_ctime = *ctime;
                         inode->i_mtime = *mtime;
@@ -566,12 +587,17 @@ static int fill_inode(struct inode *inode,
 
         /*
          * provided version will be odd if inode value is projected,
-         * even if stable.  skip the update if we have a newer info
-         * (e.g., due to inode info racing form multiple MDSs), or if
-         * we are getting projected (unstable) inode info.
+         * even if stable.  skip the update if we have newer stable
+         * info (ours>=theirs, e.g. due to racing mds replies), unless
+         * we are getting projected (unstable) info (in which case the
+         * version is odd, and we want ours>theirs).
+         *   us   them
+         *   2    2     skip
+         *   3    2     skip
+         *   3    3     update
          */
         if (le64_to_cpu(info->version) > 0 &&
-            (ci->i_version & ~1) > le64_to_cpu(info->version))
+            (ci->i_version & ~1) >= le64_to_cpu(info->version))
                 goto no_change;
 
         issued = __ceph_caps_issued(ci, &implemented);
@@ -605,7 +631,14 @@ static int fill_inode(struct inode *inode,
                             le32_to_cpu(info->time_warp_seq),
                             &ctime, &mtime, &atime);
 
-        ci->i_max_size = le64_to_cpu(info->max_size);
+        /* only update max_size on auth cap */
+        if ((info->cap.flags & CEPH_CAP_FLAG_AUTH) &&
+            ci->i_max_size != le64_to_cpu(info->max_size)) {
+                dout("max_size %lld -> %llu\n", ci->i_max_size,
+                     le64_to_cpu(info->max_size));
+                ci->i_max_size = le64_to_cpu(info->max_size);
+        }
+
         ci->i_layout = info->layout;
         inode->i_blkbits = fls(le32_to_cpu(info->layout.fl_stripe_unit)) - 1;
 
@@ -666,6 +699,8 @@ static int fill_inode(struct inode *inode,
                 inode->i_op = &ceph_dir_iops;
                 inode->i_fop = &ceph_dir_fops;
 
+                ci->i_dir_layout = iinfo->dir_layout;
+
                 ci->i_files = le64_to_cpu(info->files);
                 ci->i_subdirs = le64_to_cpu(info->subdirs);
                 ci->i_rbytes = le64_to_cpu(info->rbytes);
@@ -680,13 +715,9 @@ static int fill_inode(struct inode *inode,
                     (issued & CEPH_CAP_FILE_EXCL) == 0 &&
                     (ci->i_ceph_flags & CEPH_I_COMPLETE) == 0) {
                         dout(" marking %p complete (empty)\n", inode);
-                        ci->i_ceph_flags |= CEPH_I_COMPLETE;
+                        /* ci->i_ceph_flags |= CEPH_I_COMPLETE; */
                         ci->i_max_offset = 2;
                 }
-
-                /* it may be better to set st_size in getattr instead? */
-                if (ceph_test_opt(ceph_sb_to_client(inode->i_sb), RBYTES))
-                        inode->i_size = ci->i_rbytes;
                 break;
         default:
                 pr_err("fill_inode %llx.%llx BAD mode 0%o\n",
@@ -827,13 +858,13 @@ static void ceph_set_dentry_offset(struct dentry *dn)
         di->offset = ceph_inode(inode)->i_max_offset++;
         spin_unlock(&inode->i_lock);
 
-        spin_lock(&dcache_lock);
-        spin_lock(&dn->d_lock);
+        spin_lock(&dir->d_lock);
+        spin_lock_nested(&dn->d_lock, DENTRY_D_LOCK_NESTED);
         list_move(&dn->d_u.d_child, &dir->d_subdirs);
         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);
+        spin_unlock(&dir->d_lock);
 }
 
 /*
@@ -865,8 +896,8 @@ static struct dentry *splice_dentry(struct dentry *dn, struct inode *in,
         } else if (realdn) {
                 dout("dn %p (%d) spliced with %p (%d) "
                      "inode %p ino %llx.%llx\n",
-                     dn, atomic_read(&dn->d_count),
-                     realdn, atomic_read(&realdn->d_count),
+                     dn, dn->d_count,
+                     realdn, realdn->d_count,
                      realdn->d_inode, ceph_vinop(realdn->d_inode));
                 dput(dn);
                 dn = realdn;
@@ -901,7 +932,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
         struct inode *in = NULL;
         struct ceph_mds_reply_inode *ininfo;
         struct ceph_vino vino;
-        struct ceph_client *client = ceph_sb_to_client(sb);
+        struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
         int i = 0;
         int err = 0;
 
@@ -965,7 +996,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
          */
         if (rinfo->head->is_dentry && !req->r_aborted &&
             (rinfo->head->is_target || strncmp(req->r_dentry->d_name.name,
-                                               client->mount_args->snapdir_name,
+                                               fsc->mount_options->snapdir_name,
                                                req->r_dentry->d_name.len))) {
                 /*
                  * lookup link rename   : null -> possibly existing inode
@@ -1007,9 +1038,6 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                         dout("fill_trace doing d_move %p -> %p\n",
                              req->r_old_dentry, dn);
 
-                        /* d_move screws up d_subdirs order */
-                        ceph_i_clear(dir, CEPH_I_COMPLETE);
-
                         d_move(req->r_old_dentry, dn);
                         dout(" src %p '%.*s' dst %p '%.*s'\n",
                              req->r_old_dentry,
@@ -1021,12 +1049,15 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                            rehashing bug in vfs_rename_dir */
                         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,
+                        /*
+                         * d_move() puts the renamed dentry at the end of
+                         * d_subdirs.  We need to assign it an appropriate
+                         * directory offset so we can behave when holding
+                         * I_COMPLETE.
+                         */
+                        ceph_set_dentry_offset(req->r_old_dentry);
+                        dout("dn %p gets new offset %lld\n", req->r_old_dentry, 
                              ceph_dentry(req->r_old_dentry)->offset);
-                        ceph_dentry(req->r_old_dentry)->offset =
-                                ceph_dentry(dn)->offset;
 
                         dn = req->r_old_dentry;  /* use old_dentry */
                         in = dn->d_inode;
@@ -1054,7 +1085,8 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                 ininfo = rinfo->targeti.in;
                 vino.ino = le64_to_cpu(ininfo->ino);
                 vino.snap = le64_to_cpu(ininfo->snapid);
-                if (!dn->d_inode) {
+                in = dn->d_inode;
+                if (!in) {
                         in = ceph_get_inode(sb, vino);
                         if (IS_ERR(in)) {
                                 pr_err("fill_trace bad get_inode "
@@ -1069,10 +1101,10 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                                 goto done;
                         }
                         req->r_dentry = dn;  /* may have spliced */
-                        igrab(in);
+                        ihold(in);
                 } else if (ceph_ino(in) == vino.ino &&
                            ceph_snap(in) == vino.snap) {
-                        igrab(in);
+                        ihold(in);
                 } else {
                         dout(" %p links to %p %llx.%llx, not %llx.%llx\n",
                              dn, in, ceph_ino(in), ceph_snap(in),
@@ -1112,7 +1144,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req,
                         goto done;
                 }
                 req->r_dentry = dn;  /* may have spliced */
-                igrab(in);
+                ihold(in);
                 rinfo->head->is_dentry = 1;  /* fool notrace handlers */
         }
 
@@ -1216,11 +1248,11 @@ retry_lookup:
                         goto retry_lookup;
                 } else {
                         /* reorder parent's d_subdirs */
-                        spin_lock(&dcache_lock);
-                        spin_lock(&dn->d_lock);
+                        spin_lock(&parent->d_lock);
+                        spin_lock_nested(&dn->d_lock, DENTRY_D_LOCK_NESTED);
                         list_move(&dn->d_u.d_child, &parent->d_subdirs);
                         spin_unlock(&dn->d_lock);
-                        spin_unlock(&dcache_lock);
+                        spin_unlock(&parent->d_lock);
                 }
 
                 di = dn->d_fsdata;
@@ -1296,7 +1328,7 @@ void ceph_queue_writeback(struct inode *inode)
         if (queue_work(ceph_inode_to_client(inode)->wb_wq,
                        &ceph_inode(inode)->i_wb_work)) {
                 dout("ceph_queue_writeback %p\n", inode);
-                igrab(inode);
+                ihold(inode);
         } else {
                 dout("ceph_queue_writeback %p failed\n", inode);
         }
@@ -1321,7 +1353,7 @@ void ceph_queue_invalidate(struct inode *inode)
         if (queue_work(ceph_inode_to_client(inode)->pg_inv_wq,
                        &ceph_inode(inode)->i_pg_inv_work)) {
                 dout("ceph_queue_invalidate %p\n", inode);
-                igrab(inode);
+                ihold(inode);
         } else {
                 dout("ceph_queue_invalidate %p failed\n", inode);
         }
@@ -1385,11 +1417,8 @@ static void ceph_invalidate_work(struct work_struct *work)
         spin_lock(&inode->i_lock);
         dout("invalidate_pages %p gen %d revoking %d\n", inode,
              ci->i_rdcache_gen, ci->i_rdcache_revoking);
-        if (ci->i_rdcache_gen == 0 ||
-            ci->i_rdcache_revoking != ci->i_rdcache_gen) {
-                BUG_ON(ci->i_rdcache_revoking > ci->i_rdcache_gen);
+        if (ci->i_rdcache_revoking != ci->i_rdcache_gen) {
                 /* nevermind! */
-                ci->i_rdcache_revoking = 0;
                 spin_unlock(&inode->i_lock);
                 goto out;
         }
@@ -1399,15 +1428,16 @@ static void ceph_invalidate_work(struct work_struct *work)
         ceph_invalidate_nondirty_pages(inode->i_mapping);
 
         spin_lock(&inode->i_lock);
-        if (orig_gen == ci->i_rdcache_gen) {
+        if (orig_gen == ci->i_rdcache_gen &&
+            orig_gen == ci->i_rdcache_revoking) {
                 dout("invalidate_pages %p gen %d successful\n", inode,
                      ci->i_rdcache_gen);
-                ci->i_rdcache_gen = 0;
-                ci->i_rdcache_revoking = 0;
+                ci->i_rdcache_revoking--;
                 check = 1;
         } else {
-                dout("invalidate_pages %p gen %d raced, gen now %d\n",
-                     inode, orig_gen, ci->i_rdcache_gen);
+                dout("invalidate_pages %p gen %d raced, now %d revoking %d\n",
+                     inode, orig_gen, ci->i_rdcache_gen,
+                     ci->i_rdcache_revoking);
         }
         spin_unlock(&inode->i_lock);
 
@@ -1447,7 +1477,7 @@ void ceph_queue_vmtruncate(struct inode *inode)
         if (queue_work(ceph_sb_to_client(inode->i_sb)->trunc_wq,
                        &ci->i_vmtruncate_work)) {
                 dout("ceph_queue_vmtruncate %p\n", inode);
-                igrab(inode);
+                ihold(inode);
         } else {
                 dout("ceph_queue_vmtruncate %p failed, pending=%d\n",
                      inode, ci->i_truncate_pending);
@@ -1533,11 +1563,12 @@ int ceph_setattr(struct dentry *dentry, struct iattr *attr)
         struct inode *parent_inode = dentry->d_parent->d_inode;
         const unsigned int ia_valid = attr->ia_valid;
         struct ceph_mds_request *req;
-        struct ceph_mds_client *mdsc = &ceph_sb_to_client(dentry->d_sb)->mdsc;
+        struct ceph_mds_client *mdsc = ceph_sb_to_client(dentry->d_sb)->mdsc;
         int issued;
         int release = 0, dirtied = 0;
         int mask = 0;
         int err = 0;
+        int inode_dirty_flags = 0;
 
         if (ceph_snap(inode) != CEPH_NOSNAP)
                 return -EROFS;
@@ -1696,15 +1727,19 @@ int ceph_setattr(struct dentry *dentry, struct iattr *attr)
                 dout("setattr %p ATTR_FILE ... hrm!\n", inode);
 
         if (dirtied) {
-                __ceph_mark_dirty_caps(ci, dirtied);
+                inode_dirty_flags = __ceph_mark_dirty_caps(ci, dirtied);
                 inode->i_ctime = CURRENT_TIME;
         }
 
         release &= issued;
         spin_unlock(&inode->i_lock);
 
+        if (inode_dirty_flags)
+                __mark_inode_dirty(inode, inode_dirty_flags);
+
         if (mask) {
-                req->r_inode = igrab(inode);
+                req->r_inode = inode;
+                ihold(inode);
                 req->r_inode_drop = release;
                 req->r_args.setattr.mask = cpu_to_le32(mask);
                 req->r_num_caps = 1;
@@ -1728,8 +1763,8 @@ out:
  */
 int ceph_do_getattr(struct inode *inode, int mask)
 {
-        struct ceph_client *client = ceph_sb_to_client(inode->i_sb);
-        struct ceph_mds_client *mdsc = &client->mdsc;
+        struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
+        struct ceph_mds_client *mdsc = fsc->mdsc;
         struct ceph_mds_request *req;
         int err;
 
@@ -1738,14 +1773,15 @@ int ceph_do_getattr(struct inode *inode, int mask)
                 return 0;
         }
 
-        dout("do_getattr inode %p mask %s\n", inode, ceph_cap_string(mask));
+        dout("do_getattr inode %p mask %s mode 0%o\n", inode, ceph_cap_string(mask), inode->i_mode);
         if (ceph_caps_issued_mask(ceph_inode(inode), mask, 1))
                 return 0;
 
         req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, USE_ANY_MDS);
         if (IS_ERR(req))
                 return PTR_ERR(req);
-        req->r_inode = igrab(inode);
+        req->r_inode = inode;
+        ihold(inode);
         req->r_num_caps = 1;
         req->r_args.getattr.mask = cpu_to_le32(mask);
         err = ceph_mdsc_do_request(mdsc, NULL, req);
@@ -1759,12 +1795,17 @@ int ceph_do_getattr(struct inode *inode, int mask)
  * Check inode permissions.  We verify we have a valid value for
  * the AUTH cap, then call the generic handler.
  */
-int ceph_permission(struct inode *inode, int mask)
+int ceph_permission(struct inode *inode, int mask, unsigned int flags)
 {
-        int err = ceph_do_getattr(inode, CEPH_CAP_AUTH_SHARED);
+        int err;
+
+        if (flags & IPERM_FLAG_RCU)
+                return -ECHILD;
+
+        err = ceph_do_getattr(inode, CEPH_CAP_AUTH_SHARED);
 
         if (!err)
-                err = generic_permission(inode, mask, NULL);
+                err = generic_permission(inode, mask, flags, NULL);
         return err;
 }
 
@@ -1782,13 +1823,17 @@ int ceph_getattr(struct vfsmount *mnt, struct dentry *dentry,
         err = ceph_do_getattr(inode, CEPH_STAT_CAP_INODE_ALL);
         if (!err) {
                 generic_fillattr(inode, stat);
-                stat->ino = inode->i_ino;
+                stat->ino = ceph_translate_ino(inode->i_sb, inode->i_ino);
                 if (ceph_snap(inode) != CEPH_NOSNAP)
                         stat->dev = ceph_snap(inode);
                 else
                         stat->dev = 0;
                 if (S_ISDIR(inode->i_mode)) {
-                        stat->size = ci->i_rbytes;
+                        if (ceph_test_mount_opt(ceph_sb_to_client(inode->i_sb),
+                                                RBYTES))
+                                stat->size = ci->i_rbytes;
+                        else
+                                stat->size = ci->i_files + ci->i_subdirs;
                         stat->blocks = 0;
                         stat->blksize = 65536;
                 }