From: Sage Weil <sage@newdream.net>
Date: Tue, 29 Apr 2008 23:44:14 +0000 (-0700)
Subject: kclient: deal with osdc abort vs kick races; revoke pages on msg dup
X-Git-Tag: v0.2~78
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=e6e489b912c221c4ae05b165c2faa2b921436763;p=ceph.git

kclient: deal with osdc abort vs kick races; revoke pages on msg dup
---

diff --git a/src/kernel/messenger.c b/src/kernel/messenger.c
index 25b75f24c0145..1306249b66b54 100644
--- a/src/kernel/messenger.c
+++ b/src/kernel/messenger.c
@@ -886,6 +886,10 @@ out:
  */
 static void reset_connection(struct ceph_connection *con)
 {
+	derr(1, "%s%d %u.%u.%u.%u:%u connection reset\n", 
+	     ENTITY_NAME(con->peer_name),
+	     IPQUADPORT(con->peer_addr.ipaddr));
+
 	/* reset connection, out_queue, msg_ and connect_seq */
 	/* discard existing out_queue and msg_seq */
 	while (!list_empty(&con->out_queue)) {
@@ -1365,9 +1369,12 @@ struct ceph_msg *ceph_msg_maybe_dup(struct ceph_msg *old)
 	BUG_ON(!dup);
 	memcpy(dup->front.iov_base, old->front.iov_base,
 	       le32_to_cpu(old->hdr.front_len));
-	if (old->pages)
-		derr(0, "WARNING: unsafely referenced old pages for %p\n",
-		     old);
+	
+	/* revoke old message's pages */
+	mutex_lock(&old->page_mutex);
+	old->pages = 0;
+	mutex_unlock(&old->page_mutex);
+
 	ceph_msg_put(old);
 	return dup;
 }
diff --git a/src/kernel/osd_client.c b/src/kernel/osd_client.c
index 9a4848fe94d4e..fd7ad1d974819 100644
--- a/src/kernel/osd_client.c
+++ b/src/kernel/osd_client.c
@@ -72,6 +72,7 @@ static struct ceph_osd_request *alloc_request(int nr_pages,
 	req = kmalloc(sizeof(*req) + nr_pages*sizeof(void *), GFP_NOFS);
 	if (req == NULL)
 		return ERR_PTR(-ENOMEM);
+	req->r_aborted = 0;
 	req->r_request = msg;
 	req->r_nr_pages = nr_pages;
 	atomic_set(&req->r_ref, 1);
@@ -253,8 +254,8 @@ static int kick_requests(struct ceph_osd_client *osdc)
 	int osd;
 	int ret = 0;
 	
-more:
 	spin_lock(&osdc->request_lock);
+more:
 	got = radix_tree_gang_lookup(&osdc->request_tree, (void **)&req, 
 				     next_tid, 1);
 	if (got == 0)
@@ -270,9 +271,11 @@ more:
 		dout(20, "kicking tid %llu osd%d\n", req->r_tid, osd);
 		get_request(req);
 		spin_unlock(&osdc->request_lock);
-		req->r_request = ceph_msg_maybe_dup(req->r_request);  
-		req->r_flags |= CEPH_OSD_OP_RETRY;
-		send_request(osdc, req, osd);
+		req->r_request = ceph_msg_maybe_dup(req->r_request);
+		if (req->r_aborted) {
+			req->r_flags |= CEPH_OSD_OP_RETRY;
+			send_request(osdc, req, osd);
+		}
 		put_request(req);
 		goto more;
 	}
@@ -451,9 +454,17 @@ int do_request(struct ceph_osd_client *osdc, struct ceph_osd_request *req)
 
 	unregister_request(osdc, req);
 	if (rc < 0) {
-		struct ceph_msg *msg = req->r_request;
+		struct ceph_msg *msg;
 		dout(0, "tid %llu err %d, revoking %p pages\n", req->r_tid, 
-		     rc, msg);
+		     rc, req->r_request);
+		/* 
+		 * mark req aborted _before_ revoking pages, so that
+		 * if a racing kick_request _does_ dup the page vec
+		 * pointer, it will definitely then see the aborted
+		 * flag and not send the request.
+		 */
+		req->r_aborted = 1;
+		msg = req->r_request;
 		mutex_lock(&msg->page_mutex);
 		msg->pages = 0;
 		mutex_unlock(&msg->page_mutex);
diff --git a/src/kernel/osd_client.h b/src/kernel/osd_client.h
index c120cf7fa87d0..fc106e66f723b 100644
--- a/src/kernel/osd_client.h
+++ b/src/kernel/osd_client.h
@@ -22,6 +22,7 @@ enum {
 
 struct ceph_osd_request {
 	__u64             r_tid;
+	int               r_aborted;
 	int               r_flags;
 	struct ceph_msg  *r_request;
 	struct ceph_entity_addr r_last_osd;  /* last osd we sent request to */