From cfa62eac0fc8c0c26008a6e97d80807b4361a23f Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@newdream.net>
Date: Thu, 10 Apr 2008 13:43:42 -0700
Subject: [PATCH] kclient: separate thread/workqueue to do inode writeback

---
 src/kernel/client.c |  7 ++++++-
 src/kernel/inode.c  | 19 +++++++++++++------
 src/kernel/super.c  |  2 ++
 src/kernel/super.h  | 14 +++++++++++++-
 4 files changed, 34 insertions(+), 8 deletions(-)

diff --git a/src/kernel/client.c b/src/kernel/client.c
index 6113c564aa2da..45d53573c1c04 100644
--- 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 90ee5385683d8..a3ef360e1e8de 100644
--- 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 d692c8d53ceb4..283d8106d0062 100644
--- 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 939aaf9dc32df..d8cbaad372c56 100644
--- 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,
-- 
2.39.5