fixes #431 hang in taskq_wait

fixes #429 async websocket reap leads to crash

This tightens up the code for shutdown, ensuring that transport
callbacks are completely stopped before advancing to the next step
of teardown of transport pipes or endpoints.

It also fixes a problem where task_wait would sometimes get "stuck"
as tasks transitioned between asynch and synchronous completions.

Finally, it saves a few cycles by only calling a cancellation callback
once during cancellation of an aio.

4 files changed, 32 insertions, 4 deletions

diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c
index 1740bfcb..016e47ec 100644
--- a/src/transport/ipc/ipc.c
+++ b/src/transport/ipc/ipc.c
@@ -89,13 +89,19 @@ nni_ipc_pipe_close(void *arg)
 }
 
 static void
-nni_ipc_pipe_fini(void *arg)
+nni_ipc_pipe_stop(void *arg)
 {
 	nni_ipc_pipe *pipe = arg;
 
 	nni_aio_stop(pipe->rxaio);
 	nni_aio_stop(pipe->txaio);
 	nni_aio_stop(pipe->negaio);
+}
+
+static void
+nni_ipc_pipe_fini(void *arg)
+{
+	nni_ipc_pipe *pipe = arg;
 
 	nni_aio_fini(pipe->rxaio);
 	nni_aio_fini(pipe->txaio);
@@ -890,6 +896,7 @@ static nni_tran_pipe_option nni_ipc_pipe_options[] = {
 static nni_tran_pipe_ops nni_ipc_pipe_ops = {
 	.p_fini    = nni_ipc_pipe_fini,
 	.p_start   = nni_ipc_pipe_start,
+	.p_stop    = nni_ipc_pipe_stop,
 	.p_send    = nni_ipc_pipe_send,
 	.p_recv    = nni_ipc_pipe_recv,
 	.p_close   = nni_ipc_pipe_close,
diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c
index 22217699..171cc8ee 100644
--- a/src/transport/tcp/tcp.c
+++ b/src/transport/tcp/tcp.c
@@ -91,13 +91,19 @@ nni_tcp_pipe_close(void *arg)
 }
 
 static void
-nni_tcp_pipe_fini(void *arg)
+nni_tcp_pipe_stop(void *arg)
 {
 	nni_tcp_pipe *p = arg;
 
 	nni_aio_stop(p->rxaio);
 	nni_aio_stop(p->txaio);
 	nni_aio_stop(p->negaio);
+}
+
+static void
+nni_tcp_pipe_fini(void *arg)
+{
+	nni_tcp_pipe *p = arg;
 
 	nni_aio_fini(p->rxaio);
 	nni_aio_fini(p->txaio);
@@ -917,6 +923,7 @@ static nni_tran_pipe_option nni_tcp_pipe_options[] = {
 static nni_tran_pipe_ops nni_tcp_pipe_ops = {
 	.p_fini    = nni_tcp_pipe_fini,
 	.p_start   = nni_tcp_pipe_start,
+	.p_stop    = nni_tcp_pipe_stop,
 	.p_send    = nni_tcp_pipe_send,
 	.p_recv    = nni_tcp_pipe_recv,
 	.p_close   = nni_tcp_pipe_close,
diff --git a/src/transport/tls/tls.c b/src/transport/tls/tls.c
index 21557270..f7e90303 100644
--- a/src/transport/tls/tls.c
+++ b/src/transport/tls/tls.c
@@ -99,13 +99,19 @@ nni_tls_pipe_close(void *arg)
 }
 
 static void
-nni_tls_pipe_fini(void *arg)
+nni_tls_pipe_stop(void *arg)
 {
 	nni_tls_pipe *p = arg;
 
 	nni_aio_stop(p->rxaio);
 	nni_aio_stop(p->txaio);
 	nni_aio_stop(p->negaio);
+}
+
+static void
+nni_tls_pipe_fini(void *arg)
+{
+	nni_tls_pipe *p = arg;
 
 	nni_aio_fini(p->rxaio);
 	nni_aio_fini(p->txaio);
@@ -1043,6 +1049,7 @@ static nni_tran_pipe_option nni_tls_pipe_options[] = {
 static nni_tran_pipe_ops nni_tls_pipe_ops = {
 	.p_fini    = nni_tls_pipe_fini,
 	.p_start   = nni_tls_pipe_start,
+	.p_stop    = nni_tls_pipe_stop,
 	.p_send    = nni_tls_pipe_send,
 	.p_recv    = nni_tls_pipe_recv,
 	.p_close   = nni_tls_pipe_close,
diff --git a/src/transport/ws/websocket.c b/src/transport/ws/websocket.c
index a4081b25..36deddac 100644
--- a/src/transport/ws/websocket.c
+++ b/src/transport/ws/websocket.c
@@ -183,12 +183,18 @@ ws_pipe_send(void *arg, nni_aio *aio)
 }
 
 static void
-ws_pipe_fini(void *arg)
+ws_pipe_stop(void *arg)
 {
 	ws_pipe *p = arg;
 
 	nni_aio_stop(p->rxaio);
 	nni_aio_stop(p->txaio);
+}
+
+static void
+ws_pipe_fini(void *arg)
+{
+	ws_pipe *p = arg;
 
 	nni_aio_fini(p->rxaio);
 	nni_aio_fini(p->txaio);
@@ -593,6 +599,7 @@ static nni_tran_pipe_option ws_pipe_options[] = {
 
 static nni_tran_pipe_ops ws_pipe_ops = {
 	.p_fini    = ws_pipe_fini,
+	.p_stop    = ws_pipe_stop,
 	.p_send    = ws_pipe_send,
 	.p_recv    = ws_pipe_recv,
 	.p_close   = ws_pipe_close,