kclient: do/wait for writeback in write_begin if page isn't clean or current snapc

diff --git a/src/kernel/addr.c b/src/kernel/addr.c
index b303507237c385f92d8110686b4392df472fc584..e07b04d1a5287df8ef3eebe420706e1a2203373b 100644 (file)
--- a/src/kernel/addr.c
+++ b/src/kernel/addr.c
@@ -61,12 +61,12 @@ static int ceph_set_page_dirty(struct page *page,
                list_for_each(p, &ci->i_cap_snaps) {
                        capsnap = list_entry(p, struct ceph_cap_snap,
                                             ci_item);
-                       if (capsnap->context == snapc) {
-                               capsnap->dirty++;
+                       if (capsnap->context == snapc)
                                break;
-                       }
                }
                BUG_ON(!capsnap);
+               BUG_ON(capsnap->context != snapc);
+               capsnap->dirty++;
                dout(20, "%p set_page_dirty %p snap %lld %d/%d -> %d/%d"
                     " snapc %p seq %lld (%d snaps)\n",
                     mapping->host, page, capsnap->follows,
@@ -294,7 +294,6 @@ static struct ceph_snap_context *__get_oldest_context(struct inode *inode)
        struct list_head *p;
        struct ceph_cap_snap *capsnap = 0;
 
-       spin_lock(&inode->i_lock);
        list_for_each(p, &ci->i_cap_snaps) {
                capsnap = list_entry(p, struct ceph_cap_snap, ci_item);
                dout(20, " cap_snap %p has %d dirty pages\n", capsnap,
@@ -308,12 +307,26 @@ static struct ceph_snap_context *__get_oldest_context(struct inode *inode)
                snapc = ceph_get_snap_context(ci->i_snap_realm->cached_context);
                dout(20, " head has %d dirty pages\n", ci->i_wrbuffer_ref_head);
        }
-       spin_unlock(&inode->i_lock);
 
        return snapc;
 }
 
 
+static struct ceph_snap_context *get_oldest_context(struct inode *inode)
+{
+       struct ceph_snap_context *snapc = 0;
+       spin_lock(&inode->i_lock);
+       snapc = __get_oldest_context(inode);
+       spin_unlock(&inode->i_lock);
+       return snapc;
+}
+
+static int context_is_writeable(struct inode *inode,
+                               struct ceph_snap_context *snapc)
+{
+       return snapc == get_oldest_context(inode);
+}
+
 /*
  * ceph_writepage:
  *  clear dirty page, and set the writeback flag in the radix tree.
@@ -348,20 +361,15 @@ static int writepage_nounlock(struct page *page, struct writeback_control *wbc)
        dout(10, "writepage %p page %p index %lu on %llu~%u\n",
             inode, page, page->index, page_off, len);
 
-       /* verify this is current snapc */
-       snapc = __get_oldest_context(inode);
-       if (!snapc) {
-               dout(20, "writepage %p page %p no snapc with dirty data?\n",
-                    inode, page);
-               goto out;
-       }
-       if ((void *)page->private == 0) {
+       /* verify this is a writeable snap context */
+       snapc = (void *)page->private;
+       if (snapc == 0) {
                dout(20, "writepage %p page %p not dirty?\n", inode, page);
                goto out;
        }
-       if ((void *)page->private != snapc) {
-               dout(10, "writepage %p page %p snapc %p newer than %p - noop\n",
-                    inode, page, (void *)page->private, snapc);
+       if (!context_is_writeable(inode, snapc)) {
+               dout(10, "writepage %p page %p snapc %p not writeable - noop\n",
+                    inode, page, (void *)page->private);
                goto out;
        }
 
@@ -476,8 +484,7 @@ static int ceph_writepages(struct address_space *mapping,
 retry:
        /* find oldest snap context with dirty data */
        ceph_put_snap_context(snapc);
-
-       snapc = __get_oldest_context(inode);
+       snapc = get_oldest_context(inode);
        if (!snapc) {
                /* hmm, why does writepages get called when there
                   is no dirty data? */
@@ -715,9 +722,11 @@ static int ceph_write_begin(struct file *file, struct address_space *mapping,
        int pos_in_page = pos & ~PAGE_MASK;
        int end_in_page = pos_in_page + len;
        loff_t i_size;
+       struct ceph_snap_context *snapc;
        int r;
 
        /* get a page*/
+retry:
        page = __grab_cache_page(mapping, index);
        if (!page)
                return -ENOMEM;
@@ -735,10 +744,28 @@ static int ceph_write_begin(struct file *file, struct address_space *mapping,
        BUG_ON(!ci->i_snap_realm->cached_context);
        if (page->private &&
            (void *)page->private != ci->i_snap_realm->cached_context) {
-               /* force early writeback of snapped page */
-               r = writepage_nounlock(page, 0);
-               if (r < 0)
-                       goto fail;
+               snapc = get_oldest_context(inode);
+               if (snapc == (void *)page->private) {
+                       /* yay, writeable, do it now */
+                       dout(10, " page %p snapc %p not current, but oldest\n",
+                            page, snapc);
+                       r = writepage_nounlock(page, 0);
+                       if (r < 0)
+                               goto fail;
+               } else {
+                       dout(10, " page %p snapc %p not current or oldest\n",
+                            page, (void *)page->private);
+                       /* queue for writeback, and wait */
+                       snapc = ceph_get_snap_context((void *)page->private);
+                       unlock_page(page);
+                       ceph_queue_writeback(inode);
+                       r = wait_event_interruptible(ci->i_cap_wq,
+                                      context_is_writeable(inode, snapc));
+                       ceph_put_snap_context(snapc);
+                       if (r < 0)
+                               return r; /* FIXME? */
+                       goto retry;
+               }
        }
 
        if (PageUptodate(page))

diff --git a/src/kernel/caps.c b/src/kernel/caps.c
index b992f671616a09ec3886c7eb06db732b3aa0aa28..f646b8fc15fcf119fe8b33c81a1388a3e0495e76 100644 (file)
--- a/src/kernel/caps.c
+++ b/src/kernel/caps.c
@@ -667,6 +667,7 @@ void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
                spin_lock(&inode->i_lock);
                __ceph_flush_snaps(ci);
                spin_unlock(&inode->i_lock);
+               wake_up(&ci->i_cap_wq);
        }
 }
 
@@ -800,7 +801,7 @@ out:
                 * context.
                 */
                dout(10, "queueing %p for writeback\n", inode);
-               ceph_queue_writeback(ceph_client(inode->i_sb), ci);
+               ceph_queue_writeback(inode);
        }
        if (invalidate)
                invalidate_mapping_pages(&inode->i_data, 0, -1);

diff --git a/src/kernel/super.h b/src/kernel/super.h
index 0465bfe3affb3640f9c75ffca041603f47706de9..f98d6f7171d449ada0d05a6325a541f7a2f26337 100644 (file)
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -319,12 +319,6 @@ static inline struct ceph_dentry_info *ceph_dentry(struct dentry *dentry)
        return (struct ceph_dentry_info *)dentry->d_fsdata;
 }
 
-static inline void ceph_queue_writeback(struct ceph_client *cl,
-                                       struct ceph_inode_info *ci)
-{
-       queue_work(cl->wb_wq, &ci->i_wb_work);
-}
-
 
 /*
  * ino_t is <64 bits on many architectures, blech.
@@ -438,6 +432,15 @@ static inline struct ceph_client *ceph_sb_to_client(struct super_block *sb)
 {
        return (struct ceph_client *)sb->s_fs_info;
 }
+
+
+static inline void ceph_queue_writeback(struct inode *inode)
+{
+       queue_work(ceph_inode_to_client(inode)->wb_wq,
+                  &ceph_inode(inode)->i_wb_work);
+}
+
+
 /*
  * keep readdir buffers attached to file->private_data
  */