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

1 files changed, 14 insertions, 37 deletions

diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c
index cb624a10..1c32fec6 100644
--- a/src/protocol/bus/bus.c
+++ b/src/protocol/bus/bus.c
@@ -51,7 +51,6 @@ struct nni_bus_pipe {
 	nni_aio		aio_send;
 	nni_aio		aio_putq;
 	nni_mtx		mtx;
-	int		refcnt;
 };
 
 
@@ -133,7 +132,6 @@ nni_bus_pipe_init(void **pp, nni_pipe *npipe, void *psock)
 		return (NNG_ENOMEM);
 	}
 	NNI_LIST_NODE_INIT(&ppipe->node);
-	ppipe->refcnt = 0;
 	if (((rv = nni_mtx_init(&ppipe->mtx)) != 0) ||
 	    ((rv = nni_msgq_init(&ppipe->sendq, 16)) != 0)) {
 		goto fail;
@@ -176,12 +174,6 @@ nni_bus_pipe_start(void *arg)
 	nni_list_append(&psock->pipes, ppipe);
 	nni_mtx_unlock(&psock->mtx);
 
-	// Mark the ppipe busy twice -- once for each of the oustanding
-	// asynchronous "threads" of operation.
-	nni_mtx_lock(&ppipe->mtx);
-	ppipe->refcnt = 2;
-	nni_mtx_unlock(&ppipe->mtx);
-
 	nni_bus_pipe_recv(ppipe);
 	nni_bus_pipe_getq(ppipe);
 
@@ -189,40 +181,24 @@ nni_bus_pipe_start(void *arg)
 }
 
 
-// nni_bus_pipe_stop is called only internally when one of our handlers notices
-// that the transport layer has closed.  This allows us to stop all further
-// actions.
 static void
-nni_bus_pipe_stop(nni_bus_pipe *ppipe)
+nni_bus_pipe_stop(void *arg)
 {
-	int refcnt;
+	nni_bus_pipe *ppipe = arg;
 	nni_bus_sock *psock = ppipe->psock;
 
-	// As we are called only on error paths, shut down the underlying
-	// pipe transport.  This should cause any other consumer to also get
-	// a suitable error (NNG_ECLOSED), so that we can shut down completely.
-	nni_pipe_close(ppipe->npipe);
+	nni_msgq_close(ppipe->sendq);
+
+	nni_aio_stop(&ppipe->aio_getq);
+	nni_aio_stop(&ppipe->aio_send);
+	nni_aio_stop(&ppipe->aio_recv);
+	nni_aio_stop(&ppipe->aio_putq);
 
 	nni_mtx_lock(&ppipe->psock->mtx);
 	if (nni_list_active(&psock->pipes, ppipe)) {
 		nni_list_remove(&psock->pipes, ppipe);
-
-		nni_msgq_close(ppipe->sendq);
-		nni_msgq_aio_cancel(nni_sock_recvq(psock->nsock),
-		    &ppipe->aio_putq);
 	}
 	nni_mtx_unlock(&ppipe->psock->mtx);
-
-	nni_mtx_lock(&ppipe->mtx);
-	NNI_ASSERT(ppipe->refcnt > 0);
-	refcnt = --ppipe->refcnt;
-	nni_mtx_unlock(&ppipe->mtx);
-
-	// If we are done with the pipe, let the system know so it can
-	// deregister it.
-	if (refcnt == 0) {
-		nni_pipe_remove(ppipe->npipe);
-	}
 }
 
 
@@ -233,7 +209,7 @@ nni_bus_pipe_getq_cb(void *arg)
 
 	if (nni_aio_result(&ppipe->aio_getq) != 0) {
 		// closed?
-		nni_bus_pipe_stop(ppipe);
+		nni_pipe_stop(ppipe->npipe);
 		return;
 	}
 	ppipe->aio_send.a_msg = ppipe->aio_getq.a_msg;
@@ -252,7 +228,7 @@ nni_bus_pipe_send_cb(void *arg)
 		// closed?
 		nni_msg_free(ppipe->aio_send.a_msg);
 		ppipe->aio_send.a_msg = NULL;
-		nni_bus_pipe_stop(ppipe);
+		nni_pipe_stop(ppipe->npipe);
 		return;
 	}
 
@@ -269,7 +245,7 @@ nni_bus_pipe_recv_cb(void *arg)
 	uint32_t id;
 
 	if (nni_aio_result(&ppipe->aio_recv) != 0) {
-		nni_bus_pipe_stop(ppipe);
+		nni_pipe_stop(ppipe->npipe);
 		return;
 	}
 	msg = ppipe->aio_recv.a_msg;
@@ -278,7 +254,7 @@ nni_bus_pipe_recv_cb(void *arg)
 	if (nni_msg_prepend_header(msg, &id, 4) != 0) {
 		// XXX: bump a nomemory stat
 		nni_msg_free(msg);
-		nni_bus_pipe_stop(ppipe);
+		nni_pipe_stop(ppipe->npipe);
 		return;
 	}
 
@@ -295,7 +271,7 @@ nni_bus_pipe_putq_cb(void *arg)
 	if (nni_aio_result(&ppipe->aio_putq) != 0) {
 		nni_msg_free(ppipe->aio_putq.a_msg);
 		ppipe->aio_putq.a_msg = NULL;
-		nni_bus_pipe_stop(ppipe);
+		nni_pipe_stop(ppipe->npipe);
 		return;
 	}
 
@@ -417,6 +393,7 @@ static nni_proto_pipe_ops nni_bus_pipe_ops = {
 	.pipe_init	= nni_bus_pipe_init,
 	.pipe_fini	= nni_bus_pipe_fini,
 	.pipe_start	= nni_bus_pipe_start,
+	.pipe_stop	= nni_bus_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_bus_sock_ops = {