From 413d5e6aa73b4a0e7eca08fd803057aefc2b2e0d Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@newdream.net>
Date: Sat, 17 Jan 2009 18:05:16 -0800
Subject: [PATCH] kclient: fix cap writeback vs revocation

We need to reply to a revocation via check_caps and __send_cap if we are
writing back dirty metadata.

We also need to take care to look at issued|implemented (not just issued)
when deciding what to write back (since it may be being revoked).
---
 src/kernel/caps.c       | 47 ++++++++++++++++++++++++++---------------
 src/kernel/file.c       |  2 +-
 src/kernel/inode.c      | 10 ++++-----
 src/kernel/mds_client.c |  2 +-
 src/kernel/super.h      |  5 +++--
 5 files changed, 40 insertions(+), 26 deletions(-)

diff --git a/src/kernel/caps.c b/src/kernel/caps.c
index b0bc9c7574116..8f9ab258e5145 100644
--- a/src/kernel/caps.c
+++ b/src/kernel/caps.c
@@ -499,6 +499,7 @@ static void __send_cap(struct ceph_mds_client *mdsc,
 {
 	struct ceph_inode_info *ci = cap->ci;
 	struct inode *inode = &ci->vfs_inode;
+	int held = cap->issued | cap->implemented;
 	int revoking = cap->implemented & ~cap->issued;
 	int dropping = cap->issued & ~retain;
 	int keep;
@@ -517,11 +518,13 @@ static void __send_cap(struct ceph_mds_client *mdsc,
 	if (retain == 0)
 		op = CEPH_CAP_OP_RELEASE;
 
-	dout(10, "__send_cap cap %p session %p %d -> %d\n", cap, cap->session,
-	     cap->issued, cap->issued & retain);
+	dout(10, "__send_cap cap %p session %p %s -> %s (revoking %s)\n",
+	     cap, cap->session,
+	     ceph_cap_string(held), ceph_cap_string(held & retain),
+	     ceph_cap_string(revoking));
 
 	dirty = __ceph_caps_dirty(ci);
-	cap->flushing |= dirty & cap->issued;
+	cap->flushing |= dirty & held;
 	if (cap->flushing) {
 		ci->i_dirty_caps &= ~cap->flushing;
 		dout(10, "__send_cap flushing %s, dirty_caps now %s\n",
@@ -705,14 +708,14 @@ static void ceph_flush_snaps(struct ceph_inode_info *ci)
  * @is_delayed indicates caller is delayed work and we should not
  * delay further.
  */
-void ceph_check_caps(struct ceph_inode_info *ci, int is_delayed, int drop)
+void ceph_check_caps(struct ceph_inode_info *ci, int is_delayed, int drop,
+		     struct ceph_mds_session *session)
 {
 	struct ceph_client *client = ceph_inode_to_client(&ci->vfs_inode);
 	struct ceph_mds_client *mdsc = &client->mdsc;
 	struct inode *inode = &ci->vfs_inode;
 	struct ceph_cap *cap;
 	int file_wanted, used;
-	struct ceph_mds_session *session = NULL;    /* if set, i hold s_mutex */
 	int took_snap_rwsem = 0;             /* true if mdsc->snap_rwsem held */
 	int want, retain, revoking;
 	int mds = -1;   /* keep track of how far we've gone through i_caps list
@@ -1025,7 +1028,7 @@ void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
 	     last ? "last" : "");
 
 	if (last && !flushsnaps)
-		ceph_check_caps(ci, 0, 0);
+		ceph_check_caps(ci, 0, 0, NULL);
 	else if (flushsnaps)
 		ceph_flush_snaps(ci);
 	if (wake)
@@ -1087,7 +1090,7 @@ void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
 	spin_unlock(&inode->i_lock);
 
 	if (last) {
-		ceph_check_caps(ci, 0, 0);
+		ceph_check_caps(ci, 0, 0, NULL);
 	} else if (last_snap) {
 		ceph_flush_snaps(ci);
 		wake_up(&ci->i_cap_wq);
@@ -1104,6 +1107,7 @@ void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
  * return value:
  *  0 - ok
  *  1 - send the msg back to mds
+ *  2 - check_caps
  */
 static int handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
 			    struct ceph_mds_session *session,
@@ -1116,13 +1120,11 @@ static int handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
 	int mds = session->s_mds;
 	int seq = le32_to_cpu(grant->seq);
 	int newcaps = le32_to_cpu(grant->caps);
-	int used;
-	int issued; /* to me, before */
-	int wanted;
-	int reply = 0;
+	int issued, used, wanted, dirty;
 	u64 size = le64_to_cpu(grant->size);
 	u64 max_size = le64_to_cpu(grant->max_size);
 	struct timespec mtime, atime, ctime;
+	int reply = 0;
 	int wake = 0;
 	int writeback = 0;
 	int revoked_rdcache = 0;
@@ -1246,8 +1248,11 @@ start:
 	/* check cap bits */
 	wanted = __ceph_caps_wanted(ci);
 	used = __ceph_caps_used(ci);
-	dout(10, " my wanted = %s, used = %s\n", ceph_cap_string(wanted),
-	     ceph_cap_string(used));
+	dirty = __ceph_caps_dirty(ci);
+	dout(10, " my wanted = %s, used = %s, dirty %s\n",
+	     ceph_cap_string(wanted),
+	     ceph_cap_string(used),
+	     ceph_cap_string(dirty));
 	if (wanted != le32_to_cpu(grant->wanted)) {
 		dout(10, "mds wanted %s -> %s\n",
 		     ceph_cap_string(le32_to_cpu(grant->wanted)),
@@ -1266,7 +1271,9 @@ start:
 		     ceph_cap_string(newcaps));
 		if ((used & ~newcaps) & CEPH_CAP_FILE_WRBUFFER) {
 			writeback = 1; /* will delay ack */
-		} else if (((used & ~newcaps) & CEPH_CAP_FILE_RDCACHE) == 0 ||
+		} else if (dirty & ~newcaps)
+			reply = 2;     /* initiate writeback in check_caps */
+		else if (((used & ~newcaps) & CEPH_CAP_FILE_RDCACHE) == 0 ||
 			   revoked_rdcache) {
 			/*
 			 * we're not using revoked caps.. ack now.
@@ -1573,6 +1580,7 @@ void ceph_handle_caps(struct ceph_mds_client *mdsc,
 	u64 size, max_size;
 	int check_caps = 0;
 	void *xattr_data = NULL;
+	int r;
 
 	dout(10, "handle_caps from mds%d\n", mds);
 
@@ -1652,10 +1660,14 @@ void ceph_handle_caps(struct ceph_mds_client *mdsc,
 	switch (op) {
 	case CEPH_CAP_OP_GRANT:
 		up_write(&mdsc->snap_rwsem);
-		if (handle_cap_grant(inode, h, session, cap,&xattr_data) == 1) {
+		r = handle_cap_grant(inode, h, session, cap,&xattr_data);
+		if (r == 1) {
 			dout(10, " sending reply back to mds%d\n", mds);
 			ceph_msg_get(msg);
 			ceph_send_msg_mds(mdsc, msg, mds);
+		} else if (r == 2) {
+			ceph_check_caps(ceph_inode(inode), 1, 0, session);
+			goto done_unlocked;
 		}
 		break;
 
@@ -1679,11 +1691,12 @@ void ceph_handle_caps(struct ceph_mds_client *mdsc,
 
 done:
 	mutex_unlock(&session->s_mutex);
+done_unlocked:
 	ceph_put_mds_session(session);
 
 	kfree(xattr_data);
 	if (check_caps)
-		ceph_check_caps(ceph_inode(inode), 1, 0);
+		ceph_check_caps(ceph_inode(inode), 1, 0, NULL);
 	if (inode)
 		iput(inode);
 	return;
@@ -1724,7 +1737,7 @@ void ceph_check_delayed_caps(struct ceph_mds_client *mdsc)
 		list_del_init(&ci->i_cap_delay_list);
 		spin_unlock(&mdsc->cap_delay_lock);
 		dout(10, "check_delayed_caps on %p\n", &ci->vfs_inode);
-		ceph_check_caps(ci, 1, 0);
+		ceph_check_caps(ci, 1, 0, NULL);
 		iput(&ci->vfs_inode);
 	}
 	spin_unlock(&mdsc->cap_delay_lock);
diff --git a/src/kernel/file.c b/src/kernel/file.c
index 4574e4c20f9f4..018618da8e4f6 100644
--- a/src/kernel/file.c
+++ b/src/kernel/file.c
@@ -346,7 +346,7 @@ static void check_max_size(struct inode *inode, loff_t endoff)
 	}
 	spin_unlock(&inode->i_lock);
 	if (check)
-		ceph_check_caps(ci, 0, 0);
+		ceph_check_caps(ci, 0, 0, NULL);
 }
 
 /*
diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index 2bf2bc6da400d..646d7ba5d9b1d 100644
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -1237,7 +1237,7 @@ void ceph_inode_set_size(struct inode *inode, loff_t size)
 	if ((size << 1) >= ci->i_max_size &&
 	    (ci->i_reported_size << 1) < ci->i_max_size) {
 		spin_unlock(&inode->i_lock);
-		ceph_check_caps(ci, 0, 0);
+		ceph_check_caps(ci, 0, 0, NULL);
 	} else {
 		spin_unlock(&inode->i_lock);
 	}
@@ -1261,7 +1261,7 @@ void ceph_put_fmode(struct ceph_inode_info *ci, int fmode)
 	spin_unlock(&ci->vfs_inode.i_lock);
 
 	if (last && ci->i_vino.snap == CEPH_NOSNAP)
-		ceph_check_caps(ci, 0, 0);
+		ceph_check_caps(ci, 0, 0, NULL);
 }
 
 
@@ -1322,7 +1322,7 @@ static void ceph_inode_invalidate_pages(struct work_struct *work)
 	spin_unlock(&inode->i_lock);
 
 	if (check)
-		ceph_check_caps(ci, 0, 0);
+		ceph_check_caps(ci, 0, 0, NULL);
 out:
 	iput(inode);
 }
@@ -1368,7 +1368,7 @@ void __ceph_do_pending_vmtruncate(struct inode *inode)
 		dout(10, "__do_pending_vmtruncate %p to %lld\n", inode, to);
 		truncate_inode_pages(inode->i_mapping, to);
 		if (wrbuffer_refs == 0)
-			ceph_check_caps(ci, 0, 0);
+			ceph_check_caps(ci, 0, 0, NULL);
 	} else {
 		dout(10, "__do_pending_vmtruncate %p nothing to do\n", inode);
 	}
@@ -1643,7 +1643,7 @@ int ceph_setattr(struct dentry *dentry, struct iattr *attr)
 		     inode->i_uid, attr->ia_uid);
 	if (ia_valid & ATTR_GID)
 		dout(10, "setattr: %p gid %d -> %d\n", inode,
-		     inode->i_uid, attr->ia_uid);
+		     inode->i_gid, attr->ia_gid);
 	if (ia_valid & ATTR_MODE)
 		dout(10, "setattr: %p mode 0%o -> 0%o\n", inode, inode->i_mode,
 		     attr->ia_mode);
diff --git a/src/kernel/mds_client.c b/src/kernel/mds_client.c
index 205aba072bd5b..cc39d30fa384d 100644
--- a/src/kernel/mds_client.c
+++ b/src/kernel/mds_client.c
@@ -881,7 +881,7 @@ static void check_all_caps(struct ceph_mds_client *mdsc,
 
 		igrab(inode);
 		mutex_unlock(&session->s_mutex);
-		ceph_check_caps(ceph_inode(inode), 1, 0);
+		ceph_check_caps(ceph_inode(inode), 1, 0, NULL);
 		mutex_lock(&session->s_mutex);
 		iput(inode);
 	}
diff --git a/src/kernel/super.h b/src/kernel/super.h
index e049f9b6d9069..fe7f39f90d29c 100644
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -713,13 +713,14 @@ extern void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
 				       struct ceph_snap_context *snapc);
 extern void __ceph_flush_snaps(struct ceph_inode_info *ci,
 			       struct ceph_mds_session **psession);
-extern void ceph_check_caps(struct ceph_inode_info *ci, int delayed, int drop);
+extern void ceph_check_caps(struct ceph_inode_info *ci, int delayed, int drop,
+			    struct ceph_mds_session *session);
 extern void ceph_check_delayed_caps(struct ceph_mds_client *mdsc);
 void ceph_trim_session_rdcaps(struct ceph_mds_session *session);
 
 static inline void ceph_release_caps(struct inode *inode, int mask)
 {
-	ceph_check_caps(ceph_inode(inode), 1, mask);
+	ceph_check_caps(ceph_inode(inode), 1, mask, NULL);
 }
 
 /* addr.c */
-- 
2.39.5