Yet more race condition fixes.

We need to remember that protocol stops can run synchronously, and
therefore we need to wait for the aio to complete.  Further, we need
to break apart shutting down aio activity from deallocation, as we need
to shut down *all* async activity before deallocating *anything*.

Noticed that we had a pipe race in the surveyor pattern too.

2 files changed, 9 insertions, 12 deletions

diff --git a/src/protocol/pipeline/pull.c b/src/protocol/pipeline/pull.c
index 39e809e6..cde79824 100644
--- a/src/protocol/pipeline/pull.c
+++ b/src/protocol/pipeline/pull.c
@@ -90,11 +90,9 @@ nni_pull_pipe_fini(void *arg)
 {
 	nni_pull_pipe *pp = arg;
 
-	if (pp != NULL) {
-		nni_aio_fini(&pp->putq_aio);
-		nni_aio_fini(&pp->recv_aio);
-		NNI_FREE_STRUCT(pp);
-	}
+	nni_aio_fini(&pp->putq_aio);
+	nni_aio_fini(&pp->recv_aio);
+	NNI_FREE_STRUCT(pp);
 }
 
 static int
@@ -113,8 +111,8 @@ nni_pull_pipe_stop(void *arg)
 {
 	nni_pull_pipe *pp = arg;
 
-	nni_aio_cancel(&pp->putq_aio, NNG_ECANCELED);
-	nni_aio_cancel(&pp->recv_aio, NNG_ECANCELED);
+	nni_aio_stop(&pp->putq_aio);
+	nni_aio_stop(&pp->recv_aio);
 }
 
 static void
diff --git a/src/protocol/pipeline/push.c b/src/protocol/pipeline/push.c
index 43c0feaf..b7d4322c 100644
--- a/src/protocol/pipeline/push.c
+++ b/src/protocol/pipeline/push.c
@@ -132,12 +132,11 @@ nni_push_pipe_start(void *arg)
 static void
 nni_push_pipe_stop(void *arg)
 {
-	nni_push_pipe *pp   = arg;
-	nni_push_sock *push = pp->push;
+	nni_push_pipe *pp = arg;
 
-	nni_aio_cancel(&pp->aio_recv, NNG_ECANCELED);
-	nni_aio_cancel(&pp->aio_send, NNG_ECANCELED);
-	nni_aio_cancel(&pp->aio_getq, NNG_ECANCELED);
+	nni_aio_stop(&pp->aio_recv);
+	nni_aio_stop(&pp->aio_send);
+	nni_aio_stop(&pp->aio_getq);
 }
 
 static void