kclient: fail con->ops->get() if ref count is zero, and avoid requeueing work

This fixes a bug where a socket status change races with con_destroy:

ref 1 -> 0
	con_destroy()
					sock state change, requeue
		destroy socket
		kfree(con)
					...do work and crash...

diff --git a/src/TODO b/src/TODO
index e9f70cc5314544ff0b2f4765773016504ba92ccc..fc888c9bf0ea80f7e64864c2f5298ac0e891e749 100644 (file)
--- a/src/TODO
+++ b/src/TODO
@@ -31,9 +31,8 @@ v0.14
 - msgr: unidirectional option
   - what about mon -> mds/osd messages?
 - kclient msgr fixups
-  - private field in ceph_connection
-  - kill radix_tree; allocate connections in callers (osdc, monc, mdsc)
-  - simplify ceph_ping (should be a single byte)
+  - allow msg to be revoked from connection  (per-msg queue_seq, mutex)
+  - kill maybe_dup_msg
 
 
 bugs

diff --git a/src/kernel/mds_client.c b/src/kernel/mds_client.c
index df7da75a49f4cfe710e8af15d9dd5abcfdc28b1a..896f6c37828254a98abaddcc421cda3748b9defe 100644 (file)
--- a/src/kernel/mds_client.c
+++ b/src/kernel/mds_client.c
@@ -2856,11 +2856,13 @@ bad:
        return;
 }
 
-static void con_get(struct ceph_connection *con)
+static struct ceph_connection *con_get(struct ceph_connection *con)
 {
        struct ceph_mds_session *s = con->private;
 
-       ceph_get_mds_session(s);
+       if (ceph_get_mds_session(s))
+               return con;
+       return NULL;
 }
 
 static void con_put(struct ceph_connection *con)

diff --git a/src/kernel/messenger.c b/src/kernel/messenger.c
index 1ad72c096d1873a56cb4082e7cb07cc7a27c7e9d..ddcd8b7b2022a062c0bf09fcfe0157766beba555 100644 (file)
--- a/src/kernel/messenger.c
+++ b/src/kernel/messenger.c
@@ -247,11 +247,13 @@ void ceph_con_destroy(struct ceph_connection *con)
 /*
  * generic get/put
  */
-void ceph_con_get(struct ceph_connection *con)
+struct ceph_connection *ceph_con_get(struct ceph_connection *con)
 {
        dout("con_get %p nref = %d -> %d\n", con,
             atomic_read(&con->nref), atomic_read(&con->nref) + 1);
-       atomic_inc(&con->nref);
+       if (atomic_inc_not_zero(&con->nref))
+               return con;
+       return NULL;
 }
 
 void ceph_con_put(struct ceph_connection *con)
@@ -1390,7 +1392,10 @@ static void ceph_queue_con(struct ceph_connection *con)
                return;
        }
 
-       con->ops->get(con);
+       if (!con->ops->get(con)) {
+               dout("ceph_queue_con %p ref count 0\n", con);
+               return;
+       }
 
        set_bit(QUEUED, &con->state);
        if (test_bit(BUSY, &con->state) ||

diff --git a/src/kernel/messenger.h b/src/kernel/messenger.h
index 9547213b78e459253d9479e3b30d699aa977908d..e8e7e3cce488c06dc01e7513ad67ab354f57fb2a 100644 (file)
--- a/src/kernel/messenger.h
+++ b/src/kernel/messenger.h
@@ -28,7 +28,7 @@ extern struct workqueue_struct *ceph_msgr_wq;       /* receive work queue */
  * Ceph defines these callbacks for handling connection events.
  */
 struct ceph_connection_operations {
-       void (*get)(struct ceph_connection *);
+       struct ceph_connection *(*get)(struct ceph_connection *);
        void (*put)(struct ceph_connection *);
 
        /* handle an incoming message. */
@@ -222,7 +222,7 @@ extern void ceph_con_destroy(struct ceph_connection *con);
 extern void ceph_con_send(struct ceph_connection *con, struct ceph_msg *msg);
 extern void ceph_con_keepalive(struct ceph_connection *con);
 extern void ceph_con_close(struct ceph_connection *con);
-extern void ceph_con_get(struct ceph_connection *con);
+extern struct ceph_connection *ceph_con_get(struct ceph_connection *con);
 extern void ceph_con_put(struct ceph_connection *con);
 
 extern struct ceph_msg *ceph_msg_new(int type, int front_len,