From db89d371964dd4946946195bd378c2e7586ac4b8 Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@newdream.net>
Date: Wed, 2 Apr 2008 16:06:14 -0700
Subject: [PATCH] kclient: osdmap decoding bugfix; request osdmaps when pg has
 no primary

---
 src/kernel/addr.c       |  3 ++-
 src/kernel/client.c     | 16 +++++++++++-----
 src/kernel/mon_client.c | 35 +++++++++++++++++++++++++++++++++++
 src/kernel/mon_client.h |  5 ++++-
 src/kernel/osd_client.c | 15 +++++++++++----
 src/kernel/osdmap.c     | 18 +++++++++++++++---
 6 files changed, 78 insertions(+), 14 deletions(-)

diff --git a/src/kernel/addr.c b/src/kernel/addr.c
index 5bc3ec282558d..ef35e941d155a 100644
--- a/src/kernel/addr.c
+++ b/src/kernel/addr.c
@@ -208,7 +208,6 @@ retry:
 		
 		for (i = 0; i < nr_pages; i++) {
 			page = pvec.pages[i];
-			dout(30, "trying page %p\n", page);
 			if (first < 0)
 				lock_page(page);
 			else if (TestSetPageLocked(page))
@@ -243,8 +242,10 @@ retry:
 				end_page_writeback(page);
 				break;
 			}
+			/*
 			dout(20, "writepages locked page %p index %lu\n",
 			     page, page->index);
+			*/
 			kmap(page);
 			if (first < 0)
 				first = i;
diff --git a/src/kernel/client.c b/src/kernel/client.c
index a549a4350952a..84cffe331d145 100644
--- a/src/kernel/client.c
+++ b/src/kernel/client.c
@@ -92,21 +92,25 @@ int ceph_mount(struct ceph_client *client, struct ceph_mount_args *args, struct
 		mount_msg = ceph_msg_new(CEPH_MSG_CLIENT_MOUNT, 0, 0, 0, 0);
 		if (IS_ERR(mount_msg))
 			return PTR_ERR(mount_msg);
-		mount_msg->hdr.dst.name.type = cpu_to_le32(CEPH_ENTITY_TYPE_MON);
+		mount_msg->hdr.dst.name.type = 
+			cpu_to_le32(CEPH_ENTITY_TYPE_MON);
 		mount_msg->hdr.dst.name.num = cpu_to_le32(which);
 		mount_msg->hdr.dst.addr = args->mon_addr[which];
 		
 		ceph_msg_send(client->msgr, mount_msg, 0);
-		dout(10, "mount from mon%d, %d attempts left\n", which, attempts);
+		dout(10, "mount from mon%d, %d attempts left\n", 
+		     which, attempts);
 		
 		/* wait */
 		dout(10, "mount sent mount request, waiting for maps\n");
-		err = wait_for_completion_timeout(&client->mount_completion, 6*HZ);
+		err = wait_for_completion_timeout(&client->mount_completion, 
+						  6*HZ);
 		if (err == -EINTR)
 			return err; 
 		if (client->mounting == 7) 
 			break;  /* success */
-		dout(10, "mount still waiting for mount, attempts=%d\n", attempts);
+		dout(10, "mount still waiting for mount, attempts=%d\n", 
+		     attempts);
 		if (--attempts == 0)
 			return -EIO;
 	}
@@ -146,7 +150,9 @@ static void handle_monmap(struct ceph_client *client, struct ceph_msg *msg)
 	if (first) {
 		client->whoami = le32_to_cpu(msg->hdr.dst.name.num);
 		client->msgr->inst.name = msg->hdr.dst.name;
-		dout(1, "i am client%d\n", client->whoami);
+		dout(1, "i am client%d, fsid is %llx.%llx\n", client->whoami,
+		     le64_to_cpu(client->monc.monmap->fsid.major),
+		     le64_to_cpu(client->monc.monmap->fsid.minor));
 	}
 }
 
diff --git a/src/kernel/mon_client.c b/src/kernel/mon_client.c
index b13df90c1b559..dced9d7715c0f 100644
--- a/src/kernel/mon_client.c
+++ b/src/kernel/mon_client.c
@@ -112,6 +112,41 @@ int ceph_monc_got_mdsmap(struct ceph_mon_client *monc, __u32 have)
 	}
 }
 
+int ceph_monc_request_osdmap(struct ceph_mon_client *monc,
+			     __u32 have, __u32 want)
+{
+	struct ceph_msg *msg;
+	int mon = pick_mon(monc, -1);
+	
+	dout(5, "ceph_monc_request_osdmap from mon%d have %u want %u\n", 
+	     mon, have, want);
+	monc->want_mdsmap = have;
+	msg = ceph_msg_new(CEPH_MSG_OSD_GETMAP, 2*sizeof(__u32), 0, 0, 0);
+	if (IS_ERR(msg))
+		return PTR_ERR(msg);
+	*(__le32*)msg->front.iov_base = cpu_to_le32(have);
+	*(((__le32*)msg->front.iov_base)+1) = cpu_to_le32(want);
+	msg->hdr.dst = monc->monmap->mon_inst[mon];
+	ceph_msg_send(monc->client->msgr, msg, 0);
+	return 0;
+
+}
+
+int ceph_monc_got_osdmap(struct ceph_mon_client *monc, __u32 have)
+{
+	if (have > monc->want_osdmap) {
+		monc->want_osdmap = 0;
+		dout(5, "ceph_monc_got_osdmap have %u > wanted %u\n", 
+		     have, monc->want_osdmap);
+		return 0;
+	} else {
+		dout(5, "ceph_monc_got_osdmap have %u <= wanted %u *****\n", 
+		     have, monc->want_osdmap);
+		return -EAGAIN;
+	}
+}
+
+
 
 /*
  * statfs
diff --git a/src/kernel/mon_client.h b/src/kernel/mon_client.h
index e272621a71ffe..19274b1438c16 100644
--- a/src/kernel/mon_client.h
+++ b/src/kernel/mon_client.h
@@ -32,6 +32,7 @@ struct ceph_mon_client {
 	u64 last_tid;
 
 	u32 want_mdsmap;  /* protected by caller's lock */
+	u32 want_osdmap;  /* protected by caller's lock */
 };
 
 extern struct ceph_monmap *ceph_monmap_decode(void *p, void *end);
@@ -42,8 +43,10 @@ extern int ceph_monc_init(struct ceph_mon_client *monc, struct ceph_client *cl);
 extern int ceph_monc_request_mdsmap(struct ceph_mon_client *monc, __u32 have);
 extern int ceph_monc_got_mdsmap(struct ceph_mon_client *monc, __u32 have);
 
+extern int ceph_monc_request_osdmap(struct ceph_mon_client *monc,
+				    __u32 have, __u32 want);
+extern int ceph_monc_got_osdmap(struct ceph_mon_client *monc, __u32 have);
 
-extern void ceph_monc_request_osdmap(struct ceph_mon_client *monc, __u64 have);
 extern void ceph_monc_request_umount(struct ceph_mon_client *monc);
 extern void ceph_monc_report_failure(struct ceph_mon_client *monc, struct ceph_entity_inst *who);
 
diff --git a/src/kernel/osd_client.c b/src/kernel/osd_client.c
index 604099bf427c7..46ed855ed9d9d 100644
--- a/src/kernel/osd_client.c
+++ b/src/kernel/osd_client.c
@@ -42,14 +42,14 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
 	struct ceph_osdmap *newmap = 0;
 	int err;
 
-	dout(1, "handle_map\n");
+	dout(1, "handle_map, have %u\n", osdc->osdmap ? osdc->osdmap->epoch:0);
 	p = msg->front.iov_base;
 	end = p + msg->front.iov_len;
 
 	/* incremental maps */
 	ceph_decode_32_safe(&p, end, nr_maps, bad);
 	dout(10, " %d inc maps\n", nr_maps);
-	while (nr_maps--) {
+	while (nr_maps > 0) {
 		ceph_decode_need(&p, end, 2*sizeof(__u32), bad);
 		ceph_decode_32(&p, epoch);
 		ceph_decode_32(&p, maplen);
@@ -68,9 +68,11 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
 				osdc->osdmap = newmap;
 			}
 		} else {
-			dout(10, "ignoring incremental map %u len %d\n", epoch, maplen);
+			dout(10, "ignoring incremental map %u len %d\n",
+			     epoch, maplen);
 		}
 		p = next;
+		nr_maps--;
 	}
 	if (newmap) 
 		goto out;
@@ -86,6 +88,7 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
 		dout(5, "skipping non-latest full map %u len %d\n", 
 		     epoch, maplen);
 		p += maplen;
+		nr_maps--;
 	}
 	if (nr_maps) {
 		ceph_decode_need(&p, end, 2*sizeof(__u32), bad);
@@ -111,6 +114,8 @@ void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
 	}
 	dout(1, "handle_map done\n");
 	
+	ceph_monc_got_osdmap(&osdc->client->monc, osdc->osdmap->epoch);
+
 	/* kick any pending requests that need kicking */
 	/* WRITE ME */
 
@@ -229,7 +234,9 @@ static void send_request(struct ceph_osd_client *osdc,
 		ceph_msg_get(req->r_request); /* send consumes a ref */
 		ceph_msg_send(osdc->client->msgr, req->r_request, 0);
 	} else {
-		dout(10, "send_request no osds in pg are up\n");
+		dout(10, "send_request no osds in this pg are up\n");
+		ceph_monc_request_osdmap(&osdc->client->monc, 
+					 osdc->osdmap->epoch, 0);
 	}
 }
 
diff --git a/src/kernel/osdmap.c b/src/kernel/osdmap.c
index 7a1d6d4815848..0bd042e90653f 100644
--- a/src/kernel/osdmap.c
+++ b/src/kernel/osdmap.c
@@ -391,7 +391,7 @@ struct ceph_osdmap *apply_incremental(void **p, void *end,
 	__u32 epoch;
 	struct ceph_timespec ctime;
 	__u32 len;
-	__u32 max;
+	__s32 max;
 	int err = -EINVAL;
 
 	ceph_decode_need(p, end, 2*sizeof(__u64)+4*sizeof(__u32), bad);
@@ -405,6 +405,8 @@ struct ceph_osdmap *apply_incremental(void **p, void *end,
 	/* full map? */
 	ceph_decode_32(p, len);
 	if (len > 0) {
+		dout(20, "apply_incremental full map len %d, %p to %p\n",
+		     len, *p, end);
 		newmap = osdmap_decode(p, min(*p+len, end));
 		return newmap;  /* error or not */
 	}
@@ -412,6 +414,8 @@ struct ceph_osdmap *apply_incremental(void **p, void *end,
 	/* new crush? */
 	ceph_decode_32_safe(p, end, len, bad);
 	if (len > 0) {
+		dout(20, "apply_incremental new crush map len %d, %p to %p\n",
+		     len, *p, end);
 		newcrush = crush_decode(p, min(*p+len, end));
 		if (IS_ERR(newcrush))
 			return ERR_PTR(PTR_ERR(newcrush));
@@ -426,7 +430,7 @@ struct ceph_osdmap *apply_incremental(void **p, void *end,
 	ceph_decode_need(p, end, 3*sizeof(__u32), bad);
 	ceph_decode_32(p, max);
 	*p += 4*sizeof(__u32);  /* skip new_pg_num et al for now.  FIXME. */
-	if (max > 0) {
+	if (max >= 0) {
 		if ((err = osdmap_set_max_osd(map, max)) < 0)
 			goto bad;
 	}
@@ -474,7 +478,15 @@ struct ceph_osdmap *apply_incremental(void **p, void *end,
 		if (osd < map->max_osd) 
 			map->crush->device_offload[osd] = off;
 	}
-	
+
+	/* skip old/new pg_swap stuff */
+	ceph_decode_32_safe(p, end, len, bad);
+	*p += len * (sizeof(__u64) + sizeof(__u32));
+	ceph_decode_32_safe(p, end, len, bad);
+	*p += len * sizeof(__u64);
+
+	if (*p != end)
+		goto bad;
 	return map;
 
 bad:
-- 
2.39.5