From 1e7c5b719cc56026c0f511eb49c88c3a5908d860 Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@newdream.net>
Date: Tue, 15 Apr 2008 14:50:28 -0700
Subject: [PATCH] kclient: fix up msg refcounting

---
 src/kernel/mds_client.c |  4 ++--
 src/kernel/messenger.c  | 13 +++++++++----
 src/kernel/messenger.h  |  2 ++
 3 files changed, 13 insertions(+), 6 deletions(-)

diff --git a/src/kernel/mds_client.c b/src/kernel/mds_client.c
index b8da21d5aad7a..ecdacd141cf1e 100644
--- a/src/kernel/mds_client.c
+++ b/src/kernel/mds_client.c
@@ -267,8 +267,8 @@ void put_session(struct ceph_mds_session *s)
 	dout(10, "put_session %p %d -> %d\n", s,
 	     atomic_read(&s->s_ref), atomic_read(&s->s_ref)-1);
 	if (atomic_dec_and_test(&s->s_ref)) {
+		dout(10, "KFREE %p\n", s);
 		kfree(s);
-		s = NULL;
 	}
 }
 
@@ -414,7 +414,6 @@ static struct ceph_mds_request *new_request(struct ceph_msg *msg)
 	req->r_resend_mds = -1;
 	atomic_set(&req->r_ref, 1);  /* one for request_tree, one for caller */
 	init_completion(&req->r_completion);
-	ceph_msg_get(msg);  /* grab reference */
 
 	return req;
 }
@@ -849,6 +848,7 @@ retry:
 	/* send and wait */
 	spin_unlock(&mdsc->lock);
 	dout(10, "do_request %p r_expects_cap=%d\n", req, req->r_expects_cap);
+	ceph_msg_get(req->r_request);
 	send_msg_mds(mdsc, req->r_request, mds);
 	wait_for_completion(&req->r_completion);
 	spin_lock(&mdsc->lock);
diff --git a/src/kernel/messenger.c b/src/kernel/messenger.c
index d79e6f0b50da3..6cae342131749 100644
--- a/src/kernel/messenger.c
+++ b/src/kernel/messenger.c
@@ -1243,9 +1243,10 @@ void ceph_messenger_mark_down(struct ceph_messenger *msgr,
 
 
 /*
- * queue up an outgoing message
+ * queue up an outgoing message.  
  *
- * will take+drop msgr, then connection locks.
+ * this consumes a msg reference.  that is, if the caller wants to
+ * keep @msg around, it had better call ceph_msg_get first.
  */
 int ceph_msg_send(struct ceph_messenger *msgr, struct ceph_msg *msg, 
 		  unsigned long timeout)
@@ -1304,7 +1305,7 @@ int ceph_msg_send(struct ceph_messenger *msgr, struct ceph_msg *msg,
 	     le64_to_cpu(msg->hdr.seq),
 	     ceph_name_type_str(le32_to_cpu(msg->hdr.dst.name.type)),
 	     le32_to_cpu(msg->hdr.dst.name.num), con);
-	ceph_msg_get(msg);
+	//ceph_msg_get(msg);
 	list_add_tail(&msg->list_head, &con->out_queue);
 	spin_unlock(&con->out_queue_lock);
 
@@ -1319,6 +1320,7 @@ int ceph_msg_send(struct ceph_messenger *msgr, struct ceph_msg *msg,
 
 /*
  * construct a new message with given type, size
+ * the new msg has a ref count of 1.
  */
 struct ceph_msg *ceph_msg_new(int type, int front_len,
 			      int page_len, int page_off, struct page **pages)
@@ -1349,6 +1351,7 @@ struct ceph_msg *ceph_msg_new(int type, int front_len,
 	m->pages = pages;
 
 	INIT_LIST_HEAD(&m->list_head);
+	dout(0, "ceph_msg_new %p\n", m);
 	return m;
 
 out2:
@@ -1361,8 +1364,10 @@ out:
 void ceph_msg_put(struct ceph_msg *m)
 {
 	BUG_ON(atomic_read(&m->nref) <= 0);
+	dout(0, "ceph_msg_put %p %d -> %d\n", m, atomic_read(&m->nref),
+	     atomic_read(&m->nref)-1);
 	if (atomic_dec_and_test(&m->nref)) {
-		dout(30, "ceph_msg_put last one on %p\n", m);
+		dout(0, "ceph_msg_put last one on %p\n", m);
 		BUG_ON(!list_empty(&m->list_head));
 		if (m->front.iov_base)
 			kfree(m->front.iov_base);
diff --git a/src/kernel/messenger.h b/src/kernel/messenger.h
index 6972f969c16d4..a836bef0de186 100644
--- a/src/kernel/messenger.h
+++ b/src/kernel/messenger.h
@@ -131,6 +131,8 @@ extern struct ceph_msg *ceph_msg_new(int type, int front_len,
 				     int page_len, int page_off,
 				     struct page **pages);
 static __inline__ void ceph_msg_get(struct ceph_msg *msg) {
+	printk("ceph_msg_get %p %d -> %d\n", msg, atomic_read(&msg->nref),
+	       atomic_read(&msg->nref)+1);
 	atomic_inc(&msg->nref);
 }
 extern void ceph_msg_put(struct ceph_msg *msg);
-- 
2.39.5