Refactor stop again, closing numerous races (thanks valgrind!)

2 files changed, 22 insertions, 30 deletions

diff --git a/src/protocol/pubsub/pub.c b/src/protocol/pubsub/pub.c
index 8ad9bb6d..2767f6b6 100644
--- a/src/protocol/pubsub/pub.c
+++ b/src/protocol/pubsub/pub.c
@@ -46,8 +46,6 @@ struct nni_pub_pipe {
 	nni_aio		aio_send;
 	nni_aio		aio_recv;
 	nni_list_node	node;
-	int		refcnt;
-	nni_mtx		mtx;
 };
 
 static int
@@ -109,7 +107,6 @@ nni_pub_pipe_fini(void *arg)
 	nni_aio_fini(&pp->aio_getq);
 	nni_aio_fini(&pp->aio_send);
 	nni_aio_fini(&pp->aio_recv);
-	nni_mtx_fini(&pp->mtx);
 	NNI_FREE_STRUCT(pp);
 }
 
@@ -124,8 +121,7 @@ nni_pub_pipe_init(void **ppp, nni_pipe *pipe, void *psock)
 		return (NNG_ENOMEM);
 	}
 	// XXX: consider making this depth tunable
-	if (((rv = nni_msgq_init(&pp->sendq, 16)) != 0) ||
-	    ((rv = nni_mtx_init(&pp->mtx)) != 0)) {
+	if ((rv = nni_msgq_init(&pp->sendq, 16)) != 0) {
 		goto fail;
 	}
 
@@ -163,43 +159,35 @@ nni_pub_pipe_start(void *arg)
 	if (nni_pipe_peer(pp->pipe) != NNG_PROTO_SUB) {
 		return (NNG_EPROTO);
 	}
+	nni_mtx_lock(&pub->mtx);
 	nni_list_append(&pub->pipes, pp);
-
-	nni_mtx_lock(&pp->mtx);
-	pp->refcnt = 2;
-	nni_mtx_unlock(&pp->mtx);
+	nni_mtx_unlock(&pub->mtx);
 
 	// Start the receiver and the queue reader.
 	nni_pipe_aio_recv(pp->pipe, &pp->aio_recv);
 	nni_msgq_aio_get(pp->sendq, &pp->aio_getq);
 
-
 	return (0);
 }
 
 
 static void
-nni_pub_pipe_stop(nni_pub_pipe *pp)
+nni_pub_pipe_stop(void *arg)
 {
+	nni_pub_pipe *pp = arg;
 	nni_pub_sock *pub = pp->pub;
 	int refcnt;
 
+	nni_aio_stop(&pp->aio_getq);
+	nni_aio_stop(&pp->aio_send);
+	nni_aio_stop(&pp->aio_recv);
+	nni_msgq_close(pp->sendq);
+
 	nni_mtx_lock(&pub->mtx);
 	if (nni_list_active(&pub->pipes, pp)) {
 		nni_list_remove(&pub->pipes, pp);
-		nni_msgq_close(pp->sendq);
 	}
 	nni_mtx_unlock(&pub->mtx);
-
-	nni_mtx_lock(&pp->mtx);
-	NNI_ASSERT(pp->refcnt > 0);
-	pp->refcnt--;
-	refcnt = pp->refcnt;
-	nni_mtx_unlock(&pp->mtx);
-
-	if (refcnt == 0) {
-		nni_pipe_remove(pp->pipe);
-	}
 }
 
 
@@ -252,7 +240,7 @@ nni_pub_pipe_recv_cb(void *arg)
 	nni_pub_pipe *pp = arg;
 
 	if (nni_aio_result(&pp->aio_recv) != 0) {
-		nni_pub_pipe_stop(pp);
+		nni_pipe_stop(pp->pipe);
 		return;
 	}
 
@@ -268,7 +256,7 @@ nni_pub_pipe_getq_cb(void *arg)
 	nni_pub_pipe *pp = arg;
 
 	if (nni_aio_result(&pp->aio_getq) != 0) {
-		nni_pub_pipe_stop(pp);
+		nni_pipe_stop(pp->pipe);
 		return;
 	}
 
@@ -287,7 +275,7 @@ nni_pub_pipe_send_cb(void *arg)
 	if (nni_aio_result(&pp->aio_send) != 0) {
 		nni_msg_free(pp->aio_send.a_msg);
 		pp->aio_send.a_msg = NULL;
-		nni_pub_pipe_stop(pp);
+		nni_pipe_stop(pp->pipe);
 		return;
 	}
 
@@ -336,6 +324,7 @@ static nni_proto_pipe_ops nni_pub_pipe_ops = {
 	.pipe_init	= nni_pub_pipe_init,
 	.pipe_fini	= nni_pub_pipe_fini,
 	.pipe_start	= nni_pub_pipe_start,
+	.pipe_stop	= nni_pub_pipe_stop,
 };
 
 nni_proto_sock_ops nni_pub_sock_ops = {
diff --git a/src/protocol/pubsub/sub.c b/src/protocol/pubsub/sub.c
index cab745b5..5733b8f2 100644
--- a/src/protocol/pubsub/sub.c
+++ b/src/protocol/pubsub/sub.c
@@ -123,10 +123,12 @@ nni_sub_pipe_start(void *arg)
 
 
 static void
-nni_sub_pipe_stop(nni_sub_pipe *sp)
+nni_sub_pipe_stop(void *arg)
 {
-	nni_msgq_aio_cancel(sp->sub->urq, &sp->aio_putq);
-	nni_pipe_remove(sp->pipe);
+	nni_sub_pipe *sp = arg;
+
+	nni_aio_stop(&sp->aio_putq);
+	nni_aio_stop(&sp->aio_recv);
 }
 
 
@@ -139,7 +141,7 @@ nni_sub_recv_cb(void *arg)
 	nni_msg *msg;
 
 	if (nni_aio_result(&sp->aio_recv) != 0) {
-		nni_sub_pipe_stop(sp);
+		nni_pipe_stop(sp->pipe);
 		return;
 	}
 
@@ -157,7 +159,7 @@ nni_sub_putq_cb(void *arg)
 	if (nni_aio_result(&sp->aio_putq) != 0) {
 		nni_msg_free(sp->aio_putq.a_msg);
 		sp->aio_putq.a_msg = NULL;
-		nni_sub_pipe_stop(sp);
+		nni_pipe_stop(sp->pipe);
 		return;
 	}
 
@@ -335,6 +337,7 @@ static nni_proto_pipe_ops nni_sub_pipe_ops = {
 	.pipe_init	= nni_sub_pipe_init,
 	.pipe_fini	= nni_sub_pipe_fini,
 	.pipe_start	= nni_sub_pipe_start,
+	.pipe_stop	= nni_sub_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_sub_sock_ops = {