Bus protocol close races fixed.

1 files changed, 13 insertions, 2 deletions

diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c
index 1e2a72b3..b88337d1 100644
--- a/src/protocol/bus/bus.c
+++ b/src/protocol/bus/bus.c
@@ -172,7 +172,9 @@ nni_bus_pipe_add(void *arg)
 
 	nni_list_append(&psock->pipes, ppipe);
 
+	nni_pipe_incref(ppipe->npipe);
 	nni_bus_pipe_recv(ppipe);
+	nni_pipe_incref(ppipe->npipe);
 	nni_bus_pipe_getq(ppipe);
 
 	return (0);
@@ -187,7 +189,7 @@ nni_bus_pipe_rem(void *arg)
 
 	nni_list_remove(&psock->pipes, ppipe);
 
-	nni_msgq_aio_cancel(ppipe->sendq, &ppipe->aio_getq);
+	nni_msgq_close(ppipe->sendq);
 	nni_msgq_aio_cancel(nni_sock_recvq(psock->nsock), &ppipe->aio_putq);
 }
 
@@ -200,6 +202,7 @@ nni_bus_pipe_getq_cb(void *arg)
 	if (nni_aio_result(&ppipe->aio_getq) != 0) {
 		// closed?
 		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
 		return;
 	}
 	ppipe->aio_send.a_msg = ppipe->aio_getq.a_msg;
@@ -216,7 +219,10 @@ nni_bus_pipe_send_cb(void *arg)
 
 	if (nni_aio_result(&ppipe->aio_send) != 0) {
 		// closed?
+		nni_msg_free(ppipe->aio_send.a_msg);
+		ppipe->aio_send.a_msg = NULL;
 		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
 		return;
 	}
 
@@ -234,6 +240,7 @@ nni_bus_pipe_recv_cb(void *arg)
 
 	if (nni_aio_result(&ppipe->aio_recv) != 0) {
 		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
 		return;
 	}
 	msg = ppipe->aio_recv.a_msg;
@@ -242,7 +249,8 @@ 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_recv(ppipe);
+		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
 		return;
 	}
 
@@ -257,7 +265,10 @@ nni_bus_pipe_putq_cb(void *arg)
 	nni_bus_pipe *ppipe = arg;
 
 	if (nni_aio_result(&ppipe->aio_putq) != 0) {
+		nni_msg_free(ppipe->aio_putq.a_msg);
+		ppipe->aio_putq.a_msg = NULL;
 		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
 		return;
 	}