11 files changed, 109 insertions, 66 deletions

diff --git a/fs/ceph/addr.c b/fs/ceph/addr.c
index e159c529fd2b..33da49dc3cc6 100644
--- a/fs/ceph/addr.c
+++ b/fs/ceph/addr.c
@@ -775,6 +775,13 @@ get_more_pages:
                                             ci->i_truncate_seq,
                                             ci->i_truncate_size,
                                             &inode->i_mtime, true, 1, 0);
+
+                                if (!req) {
+                                        rc = -ENOMEM;
+                                        unlock_page(page);
+                                        break;
+                                }
+
                                 max_pages = req->r_num_pages;
 
                                 alloc_page_vec(fsc, req);
@@ -841,7 +848,8 @@ get_more_pages:
                 op->payload_len = cpu_to_le32(len);
                 req->r_request->hdr.data_len = cpu_to_le32(len);
 
-                ceph_osdc_start_request(&fsc->client->osdc, req, true);
+                rc = ceph_osdc_start_request(&fsc->client->osdc, req, true);
+                BUG_ON(rc);
                 req = NULL;
 
                 /* continue? */
@@ -873,8 +881,6 @@ release_pvec_pages:
 out:
         if (req)
                 ceph_osdc_put_request(req);
-        if (rc > 0)
-                rc = 0;  /* vfs expects us to return 0 */
         ceph_put_snap_context(snapc);
         dout("writepages done, rc = %d\n", rc);
         return rc;
diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index 5323c330bbf3..1f72b00447c4 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -569,7 +569,8 @@ retry:
                 list_add_tail(&cap->session_caps, &session->s_caps);
                 session->s_nr_caps++;
                 spin_unlock(&session->s_cap_lock);
-        }
+        } else if (new_cap)
+                ceph_put_cap(mdsc, new_cap);
 
         if (!ci->i_snap_realm) {
                 /*
@@ -819,7 +820,7 @@ int __ceph_caps_used(struct ceph_inode_info *ci)
                 used |= CEPH_CAP_FILE_CACHE;
         if (ci->i_wr_ref)
                 used |= CEPH_CAP_FILE_WR;
-        if (ci->i_wrbuffer_ref)
+        if (ci->i_wb_ref || ci->i_wrbuffer_ref)
                 used |= CEPH_CAP_FILE_BUFFER;
         return used;
 }
@@ -1331,10 +1332,11 @@ static void ceph_flush_snaps(struct ceph_inode_info *ci)
 }
 
 /*
- * Mark caps dirty.  If inode is newly dirty, add to the global dirty
- * list.
+ * Mark caps dirty.  If inode is newly dirty, return the dirty flags.
+ * Caller is then responsible for calling __mark_inode_dirty with the
+ * returned flags value.
  */
-void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
+int __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
 {
         struct ceph_mds_client *mdsc =
                 ceph_sb_to_client(ci->vfs_inode.i_sb)->mdsc;
@@ -1357,7 +1359,7 @@ void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
                 list_add(&ci->i_dirty_item, &mdsc->cap_dirty);
                 spin_unlock(&mdsc->cap_dirty_lock);
                 if (ci->i_flushing_caps == 0) {
-                        igrab(inode);
+                        ihold(inode);
                         dirty |= I_DIRTY_SYNC;
                 }
         }
@@ -1365,9 +1367,8 @@ void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask)
         if (((was | ci->i_flushing_caps) & CEPH_CAP_FILE_BUFFER) &&
             (mask & CEPH_CAP_FILE_BUFFER))
                 dirty |= I_DIRTY_DATASYNC;
-        if (dirty)
-                __mark_inode_dirty(inode, dirty);
         __cap_delay_requeue(mdsc, ci);
+        return dirty;
 }
 
 /*
@@ -1990,11 +1991,11 @@ static void __take_cap_refs(struct ceph_inode_info *ci, int got)
         if (got & CEPH_CAP_FILE_WR)
                 ci->i_wr_ref++;
         if (got & CEPH_CAP_FILE_BUFFER) {
-                if (ci->i_wrbuffer_ref == 0)
-                        igrab(&ci->vfs_inode);
-                ci->i_wrbuffer_ref++;
-                dout("__take_cap_refs %p wrbuffer %d -> %d (?)\n",
-                     &ci->vfs_inode, ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref);
+                if (ci->i_wb_ref == 0)
+                        ihold(&ci->vfs_inode);
+                ci->i_wb_ref++;
+                dout("__take_cap_refs %p wb %d -> %d (?)\n",
+                     &ci->vfs_inode, ci->i_wb_ref-1, ci->i_wb_ref);
         }
 }
 
@@ -2169,12 +2170,12 @@ void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
                 if (--ci->i_rdcache_ref == 0)
                         last++;
         if (had & CEPH_CAP_FILE_BUFFER) {
-                if (--ci->i_wrbuffer_ref == 0) {
+                if (--ci->i_wb_ref == 0) {
                         last++;
                         put++;
                 }
-                dout("put_cap_refs %p wrbuffer %d -> %d (?)\n",
-                     inode, ci->i_wrbuffer_ref+1, ci->i_wrbuffer_ref);
+                dout("put_cap_refs %p wb %d -> %d (?)\n",
+                     inode, ci->i_wb_ref+1, ci->i_wb_ref);
         }
         if (had & CEPH_CAP_FILE_WR)
                 if (--ci->i_wr_ref == 0) {
@@ -2634,6 +2635,7 @@ static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex,
                               struct ceph_mds_session *session,
                               int *open_target_sessions)
 {
+        struct ceph_mds_client *mdsc = ceph_inode_to_client(inode)->mdsc;
         struct ceph_inode_info *ci = ceph_inode(inode);
         int mds = session->s_mds;
         unsigned mseq = le32_to_cpu(ex->migrate_seq);
@@ -2670,6 +2672,19 @@ static void handle_cap_export(struct inode *inode, struct ceph_mds_caps *ex,
                          * export targets, so that we get the matching IMPORT
                          */
                         *open_target_sessions = 1;
+
+                        /*
+                         * we can't flush dirty caps that we've seen the
+                         * EXPORT but no IMPORT for
+                         */
+                        spin_lock(&mdsc->cap_dirty_lock);
+                        if (!list_empty(&ci->i_dirty_item)) {
+                                dout(" moving %p to cap_dirty_migrating\n",
+                                     inode);
+                                list_move(&ci->i_dirty_item,
+                                          &mdsc->cap_dirty_migrating);
+                        }
+                        spin_unlock(&mdsc->cap_dirty_lock);
                 }
                 __ceph_remove_cap(cap);
         }
@@ -2707,6 +2722,13 @@ static void handle_cap_import(struct ceph_mds_client *mdsc,
                 ci->i_cap_exporting_issued = 0;
                 ci->i_cap_exporting_mseq = 0;
                 ci->i_cap_exporting_mds = -1;
+
+                spin_lock(&mdsc->cap_dirty_lock);
+                if (!list_empty(&ci->i_dirty_item)) {
+                        dout(" moving %p back to cap_dirty\n", inode);
+                        list_move(&ci->i_dirty_item, &mdsc->cap_dirty);
+                }
+                spin_unlock(&mdsc->cap_dirty_lock);
         } else {
                 dout("handle_cap_import inode %p ci %p mds%d mseq %d\n",
                      inode, ci, mds, mseq);
@@ -2910,38 +2932,16 @@ void ceph_check_delayed_caps(struct ceph_mds_client *mdsc)
  */
 void ceph_flush_dirty_caps(struct ceph_mds_client *mdsc)
 {
-        struct ceph_inode_info *ci, *nci = NULL;
-        struct inode *inode, *ninode = NULL;
-        struct list_head *p, *n;
+        struct ceph_inode_info *ci;
+        struct inode *inode;
 
         dout("flush_dirty_caps\n");
         spin_lock(&mdsc->cap_dirty_lock);
-        list_for_each_safe(p, n, &mdsc->cap_dirty) {
-                if (nci) {
-                        ci = nci;
-                        inode = ninode;
-                        ci->i_ceph_flags &= ~CEPH_I_NOFLUSH;
-                        dout("flush_dirty_caps inode %p (was next inode)\n",
-                             inode);
-                } else {
-                        ci = list_entry(p, struct ceph_inode_info,
-                                        i_dirty_item);
-                        inode = igrab(&ci->vfs_inode);
-                        BUG_ON(!inode);
-                        dout("flush_dirty_caps inode %p\n", inode);
-                }
-                if (n != &mdsc->cap_dirty) {
-                        nci = list_entry(n, struct ceph_inode_info,
-                                         i_dirty_item);
-                        ninode = igrab(&nci->vfs_inode);
-                        BUG_ON(!ninode);
-                        nci->i_ceph_flags |= CEPH_I_NOFLUSH;
-                        dout("flush_dirty_caps next inode %p, noflush\n",
-                             ninode);
-                } else {
-                        nci = NULL;
-                        ninode = NULL;
-                }
+        while (!list_empty(&mdsc->cap_dirty)) {
+                ci = list_first_entry(&mdsc->cap_dirty, struct ceph_inode_info,
+                                      i_dirty_item);
+                inode = igrab(&ci->vfs_inode);
+                dout("flush_dirty_caps %p\n", inode);
                 spin_unlock(&mdsc->cap_dirty_lock);
                 if (inode) {
                         ceph_check_caps(ci, CHECK_CAPS_NODELAY|CHECK_CAPS_FLUSH,
@@ -2951,6 +2951,7 @@ void ceph_flush_dirty_caps(struct ceph_mds_client *mdsc)
                 spin_lock(&mdsc->cap_dirty_lock);
         }
         spin_unlock(&mdsc->cap_dirty_lock);
+        dout("flush_dirty_caps done\n");
 }
 
 /*
diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index 1a867a3601ae..33729e822bb9 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -360,7 +360,7 @@ more:
         rinfo = &fi->last_readdir->r_reply_info;
         dout("readdir frag %x num %d off %d chunkoff %d\n", frag,
              rinfo->dir_nr, off, fi->offset);
-        while (off - fi->offset >= 0 && off - fi->offset < rinfo->dir_nr) {
+        while (off >= fi->offset && off - fi->offset < rinfo->dir_nr) {
                 u64 pos = ceph_make_fpos(frag, off);
                 struct ceph_mds_reply_inode *in =
                         rinfo->dir_in[off - fi->offset].in;
@@ -1066,16 +1066,17 @@ static ssize_t ceph_read_dir(struct file *file, char __user *buf, size_t size,
         struct inode *inode = file->f_dentry->d_inode;
         struct ceph_inode_info *ci = ceph_inode(inode);
         int left;
+        const int bufsize = 1024;
 
         if (!ceph_test_mount_opt(ceph_sb_to_client(inode->i_sb), DIRSTAT))
                 return -EISDIR;
 
         if (!cf->dir_info) {
-                cf->dir_info = kmalloc(1024, GFP_NOFS);
+                cf->dir_info = kmalloc(bufsize, GFP_NOFS);
                 if (!cf->dir_info)
                         return -ENOMEM;
                 cf->dir_info_len =
-                        sprintf(cf->dir_info,
+                        snprintf(cf->dir_info, bufsize,
                                 "entries:   %20lld\n"
                                 " files:    %20lld\n"
                                 " subdirs:  %20lld\n"
diff --git a/fs/ceph/export.c b/fs/ceph/export.c
index e41056174bf8..a610d3d67488 100644
--- a/fs/ceph/export.c
+++ b/fs/ceph/export.c
@@ -86,6 +86,7 @@ static int ceph_encode_fh(struct dentry *dentry, u32 *rawfh, int *max_len,
 static struct dentry *__fh_to_dentry(struct super_block *sb,
                                      struct ceph_nfs_fh *fh)
 {
+        struct ceph_mds_client *mdsc = ceph_sb_to_client(sb)->mdsc;
         struct inode *inode;
         struct dentry *dentry;
         struct ceph_vino vino;
@@ -95,8 +96,24 @@ static struct dentry *__fh_to_dentry(struct super_block *sb,
         vino.ino = fh->ino;
         vino.snap = CEPH_NOSNAP;
         inode = ceph_find_inode(sb, vino);
-        if (!inode)
-                return ERR_PTR(-ESTALE);
+        if (!inode) {
+                struct ceph_mds_request *req;
+
+                req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LOOKUPINO,
+                                               USE_ANY_MDS);
+                if (IS_ERR(req))
+                        return ERR_CAST(req);
+
+                req->r_ino1 = vino;
+                req->r_num_caps = 1;
+                err = ceph_mdsc_do_request(mdsc, NULL, req);
+                inode = req->r_target_inode;
+                if (inode)
+                        igrab(inode);
+                ceph_mdsc_put_request(req);
+                if (!inode)
+                        return ERR_PTR(-ESTALE);
+        }
 
         dentry = d_obtain_alias(inode);
         if (IS_ERR(dentry)) {
@@ -148,8 +165,10 @@ static struct dentry *__cfh_to_dentry(struct super_block *sb,
                 snprintf(req->r_path2, 16, "%d", cfh->parent_name_hash);
                 req->r_num_caps = 1;
                 err = ceph_mdsc_do_request(mdsc, NULL, req);
+                inode = req->r_target_inode;
+                if (inode)
+                        igrab(inode);
                 ceph_mdsc_put_request(req);
-                inode = ceph_find_inode(sb, vino);
                 if (!inode)
                         return ERR_PTR(err ? err : -ESTALE);
         }
diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index 159b512d5a27..203252d88d9f 100644
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -734,9 +734,12 @@ retry_snap:
                 }
         }
         if (ret >= 0) {
+                int dirty;
                 spin_lock(&inode->i_lock);
-                __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR);
+                dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR);
                 spin_unlock(&inode->i_lock);
+                if (dirty)
+                        __mark_inode_dirty(inode, dirty);
         }
 
 out:
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index b54c97da1c43..70b6a4839c38 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -355,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;
@@ -1567,6 +1568,7 @@ int ceph_setattr(struct dentry *dentry, struct iattr *attr)
         int release = 0, dirtied = 0;
         int mask = 0;
         int err = 0;
+        int inode_dirty_flags = 0;
 
         if (ceph_snap(inode) != CEPH_NOSNAP)
                 return -EROFS;
@@ -1725,13 +1727,16 @@ 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_drop = release;
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index f60b07b0feb0..79743d146be6 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -578,6 +578,7 @@ static void __register_request(struct ceph_mds_client *mdsc,
         if (dir) {
                 struct ceph_inode_info *ci = ceph_inode(dir);
 
+                ihold(dir);
                 spin_lock(&ci->i_unsafe_lock);
                 req->r_unsafe_dir = dir;
                 list_add_tail(&req->r_unsafe_dir_item, &ci->i_unsafe_dirops);
@@ -598,6 +599,9 @@ static void __unregister_request(struct ceph_mds_client *mdsc,
                 spin_lock(&ci->i_unsafe_lock);
                 list_del_init(&req->r_unsafe_dir_item);
                 spin_unlock(&ci->i_unsafe_lock);
+
+                iput(req->r_unsafe_dir);
+                req->r_unsafe_dir = NULL;
         }
 
         ceph_mdsc_put_request(req);
@@ -2691,7 +2695,6 @@ static void handle_lease(struct ceph_mds_client *mdsc,
 {
         struct super_block *sb = mdsc->fsc->sb;
         struct inode *inode;
-        struct ceph_inode_info *ci;
         struct dentry *parent, *dentry;
         struct ceph_dentry_info *di;
         int mds = session->s_mds;
@@ -2728,7 +2731,6 @@ static void handle_lease(struct ceph_mds_client *mdsc,
                 dout("handle_lease no inode %llx\n", vino.ino);
                 goto release;
         }
-        ci = ceph_inode(inode);
 
         /* dentry */
         parent = d_find_alias(inode);
@@ -3002,6 +3004,7 @@ int ceph_mdsc_init(struct ceph_fs_client *fsc)
         spin_lock_init(&mdsc->snap_flush_lock);
         mdsc->cap_flush_seq = 0;
         INIT_LIST_HEAD(&mdsc->cap_dirty);
+        INIT_LIST_HEAD(&mdsc->cap_dirty_migrating);
         mdsc->num_cap_flushing = 0;
         spin_lock_init(&mdsc->cap_dirty_lock);
         init_waitqueue_head(&mdsc->cap_flushing_wq);
@@ -3304,8 +3307,8 @@ static void con_put(struct ceph_connection *con)
 {
         struct ceph_mds_session *s = con->private;
 
+        dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref) - 1);
         ceph_put_mds_session(s);
-        dout("mdsc con_put %p (%d)\n", s, atomic_read(&s->s_ref));
 }
 
 /*
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 4e3a9cc0bba6..7d8a0d662d56 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -278,6 +278,7 @@ struct ceph_mds_client {
 
         u64               cap_flush_seq;
         struct list_head  cap_dirty;        /* inodes with dirty caps */
+        struct list_head  cap_dirty_migrating; /* ...that are migration... */
         int               num_cap_flushing; /* # caps we are flushing */
         spinlock_t        cap_dirty_lock;   /* protects above items */
         wait_queue_head_t cap_flushing_wq;
diff --git a/fs/ceph/snap.c b/fs/ceph/snap.c
index e86ec1155f8f..24067d68a554 100644
--- a/fs/ceph/snap.c
+++ b/fs/ceph/snap.c
@@ -206,7 +206,7 @@ void ceph_put_snap_realm(struct ceph_mds_client *mdsc,
                 up_write(&mdsc->snap_rwsem);
         } else {
                 spin_lock(&mdsc->snap_empty_lock);
-                list_add(&mdsc->snap_empty, &realm->empty_item);
+                list_add(&realm->empty_item, &mdsc->snap_empty);
                 spin_unlock(&mdsc->snap_empty_lock);
         }
 }
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 619fe719968f..f5cabefa98dc 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -293,7 +293,7 @@ struct ceph_inode_info {
 
         /* held references to caps */
         int i_pin_ref;
-        int i_rd_ref, i_rdcache_ref, i_wr_ref;
+        int i_rd_ref, i_rdcache_ref, i_wr_ref, i_wb_ref;
         int i_wrbuffer_ref, i_wrbuffer_ref_head;
         u32 i_shared_gen;       /* increment each time we get FILE_SHARED */
         u32 i_rdcache_gen;      /* incremented each time we get FILE_CACHE. */
@@ -506,7 +506,7 @@ static inline int __ceph_caps_dirty(struct ceph_inode_info *ci)
 {
         return ci->i_dirty_caps | ci->i_flushing_caps;
 }
-extern void __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask);
+extern int __ceph_mark_dirty_caps(struct ceph_inode_info *ci, int mask);
 
 extern int ceph_caps_revoking(struct ceph_inode_info *ci, int mask);
 extern int __ceph_caps_used(struct ceph_inode_info *ci);
diff --git a/fs/ceph/xattr.c b/fs/ceph/xattr.c
index 8c9eba6ef9df..f2b628696180 100644
--- a/fs/ceph/xattr.c
+++ b/fs/ceph/xattr.c
@@ -703,6 +703,7 @@ int ceph_setxattr(struct dentry *dentry, const char *name,
         struct ceph_inode_xattr *xattr = NULL;
         int issued;
         int required_blob_size;
+        int dirty;
 
         if (ceph_snap(inode) != CEPH_NOSNAP)
                 return -EROFS;
@@ -763,11 +764,12 @@ retry:
         dout("setxattr %p issued %s\n", inode, ceph_cap_string(issued));
         err = __set_xattr(ci, newname, name_len, newval,
                           val_len, 1, 1, 1, &xattr);
-        __ceph_mark_dirty_caps(ci, CEPH_CAP_XATTR_EXCL);
+        dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_XATTR_EXCL);
         ci->i_xattrs.dirty = true;
         inode->i_ctime = CURRENT_TIME;
         spin_unlock(&inode->i_lock);
-
+        if (dirty)
+                __mark_inode_dirty(inode, dirty);
         return err;
 
 do_sync:
@@ -810,6 +812,7 @@ int ceph_removexattr(struct dentry *dentry, const char *name)
         struct ceph_vxattr_cb *vxattrs = ceph_inode_vxattrs(inode);
         int issued;
         int err;
+        int dirty;
 
         if (ceph_snap(inode) != CEPH_NOSNAP)
                 return -EROFS;
@@ -833,12 +836,13 @@ int ceph_removexattr(struct dentry *dentry, const char *name)
                 goto do_sync;
 
         err = __remove_xattr_by_name(ceph_inode(inode), name);
-        __ceph_mark_dirty_caps(ci, CEPH_CAP_XATTR_EXCL);
+        dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_XATTR_EXCL);
         ci->i_xattrs.dirty = true;
         inode->i_ctime = CURRENT_TIME;
 
         spin_unlock(&inode->i_lock);
-
+        if (dirty)
+                __mark_inode_dirty(inode, dirty);
         return err;
 do_sync:
         spin_unlock(&inode->i_lock);