kclient: separate thread/workqueue to do inode writeback

diff --git a/src/kernel/client.c b/src/kernel/client.c
index 6113c564aa2da43c6f0e13f76dae6e1c12f12e7d..45d53573c1c04f15dba6a2385ae3d662995fc43b 100644 (file)
--- a/src/kernel/client.c
+++ b/src/kernel/client.c
@@ -177,7 +177,7 @@ struct ceph_client *ceph_create_client(struct ceph_mount_args *args, struct supe
 {
        struct ceph_client *cl;
        struct ceph_entity_addr *myaddr = 0;
-       int err;
+       int err = -ENOMEM;
 
        cl = kzalloc(sizeof(*cl), GFP_KERNEL);
        if (cl == NULL)
@@ -187,6 +187,10 @@ struct ceph_client *ceph_create_client(struct ceph_mount_args *args, struct supe
        spin_lock_init(&cl->sb_lock);
        get_client_counter();
 
+       cl->wb_wq = create_workqueue("ceph-writeback");
+       if (cl->wb_wq == 0)
+               goto fail;
+
        /* messenger */
        if (args->flags & CEPH_MOUNT_MYIP)
                myaddr = &args->my_addr;
@@ -231,6 +235,7 @@ void ceph_destroy_client(struct ceph_client *cl)
 
        ceph_messenger_destroy(cl->msgr);
        put_client_counter();
+       destroy_workqueue(cl->wb_wq);
        kfree(cl);
        dout(10, "destroy_client %p done\n", cl);
 }

diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index 90ee5385683d8496161de4234204695c63622ced..a3ef360e1e8ded434ccf8a534b83350fed2bb27c 100644 (file)
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -1024,20 +1024,27 @@ out:
                wake_up(&ci->i_cap_wq);
        if (writeback_now) {
                /*
-                * _queue_ inode for writeback, but don't actually
-                * call writepages from this context!!
+                * queue inode for writeback; we can't actually call
+                * write_inode_now, writepages, etc. from this
+                * context.
                 */
                dout(10, "queueing %p for writeback\n", inode);
-               spin_lock(&inode_lock);
-               list_move(&inode->i_list, &inode->i_sb->s_more_io);
-               spin_unlock(&inode_lock);
-               wakeup_pdflush(0);  /* this is overkill? */
+               ceph_queue_writeback(ceph_client(inode->i_sb), ci);
        }
        if (invalidate)
                invalidate_mapping_pages(&inode->i_data, 0, -1);
        return reply;
 }
 
+void ceph_inode_writeback(struct work_struct *work)
+{
+       struct ceph_inode_info *ci = container_of(work, struct ceph_inode_info,
+                                                 i_wb_work);
+       dout(10, "writeback %p\n", &ci->vfs_inode);
+       write_inode_now(&ci->vfs_inode, 0);
+}
+
+
 void apply_truncate(struct inode *inode, loff_t size)
 {
        struct ceph_inode_info *ci = ceph_inode(inode);

diff --git a/src/kernel/super.c b/src/kernel/super.c
index d692c8d53ceb4295685f1f9186e11af2ec169da9..283d8106d006202f89a32e3f45eff973ebb99cba 100644 (file)
--- a/src/kernel/super.c
+++ b/src/kernel/super.c
@@ -152,6 +152,8 @@ static struct inode *ceph_alloc_inode(struct super_block *sb)
 
        ci->i_hashval = 0;
 
+       INIT_WORK(&ci->i_wb_work, ceph_inode_writeback);
+
        return &ci->vfs_inode;
 }
 

diff --git a/src/kernel/super.h b/src/kernel/super.h
index 939aaf9dc32df158b31091ba7cf665dfd99e90d0..d8cbaad372c5687417ed83b64cb2c9cff305082c 100644 (file)
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -91,13 +91,15 @@ struct ceph_client {
        struct ceph_mds_client mdsc;
        struct ceph_osd_client osdc;
 
+       /* writeback */
+       struct workqueue_struct *wb_wq;
+
        /* lets ignore all this until later */
        spinlock_t sb_lock;
        int num_sb;      /* ref count (for each sb_info that points to me) */
        struct list_head sb_list;
 };
 
-
 /*
  * CEPH per-mount superblock info
  */
@@ -167,6 +169,8 @@ struct ceph_inode_info {
 
        unsigned long i_hashval;
 
+       struct work_struct i_wb_work;  /* writeback work */
+
        struct inode vfs_inode; /* at end */
 };
 
@@ -189,6 +193,13 @@ static inline struct ceph_dentry_info *ceph_dentry(struct dentry *dentry)
 extern void ceph_revoke_inode_lease(struct ceph_inode_info *ci, int mask);
 extern void ceph_revoke_dentry_lease(struct dentry *dentry);
 
+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
  */
@@ -382,6 +393,7 @@ extern void ceph_check_caps(struct ceph_inode_info *ci);
 extern void ceph_get_mode(struct ceph_inode_info *ci, int mode);
 extern void ceph_put_mode(struct ceph_inode_info *ci, int mode);
 extern void ceph_inode_set_size(struct inode *inode, loff_t size);
+extern void ceph_inode_writeback(struct work_struct *work);
 
 extern int ceph_setattr(struct dentry *dentry, struct iattr *attr);
 extern int ceph_inode_getattr(struct vfsmount *mnt, struct dentry *dentry,