streams: add explicit stop functions

This allows us to explicitly stop streams, dialers, and listeners,
before we start tearing down things. This hopefully will be useful
in resolving use-after-free bugs in http, tls, and websockets.

The new functions are not yet documented, but they are
nng_stream_stop, nng_stream_dialer_stop, and nng_stream_listener_stop.
They should be called after close, and before free.  The close
functions now close without blocking, but the stop function is
allowed to block.

4 files changed, 13 insertions, 2 deletions

diff --git a/src/sp/transport/ipc/ipc.c b/src/sp/transport/ipc/ipc.c
index dadb1909..c659b8fd 100644
--- a/src/sp/transport/ipc/ipc.c
+++ b/src/sp/transport/ipc/ipc.c
@@ -111,6 +111,7 @@ ipc_pipe_stop(void *arg)
 	nni_aio_stop(&p->rx_aio);
 	nni_aio_stop(&p->tx_aio);
 	nni_aio_stop(&p->neg_aio);
+	nng_stream_stop(p->conn);
 	nni_mtx_lock(&ep->mtx);
 	nni_list_node_remove(&p->node);
 	nni_mtx_unlock(&ep->mtx);
@@ -655,6 +656,8 @@ ipc_ep_stop(void *arg)
 
 	nni_aio_stop(&ep->time_aio);
 	nni_aio_stop(&ep->conn_aio);
+	nng_stream_dialer_stop(ep->dialer);
+	nng_stream_listener_stop(ep->listener);
 }
 
 static void
diff --git a/src/sp/transport/tcp/tcp.c b/src/sp/transport/tcp/tcp.c
index 2497b899..51991cbd 100644
--- a/src/sp/transport/tcp/tcp.c
+++ b/src/sp/transport/tcp/tcp.c
@@ -124,6 +124,7 @@ tcptran_pipe_stop(void *arg)
 	nni_aio_stop(p->rxaio);
 	nni_aio_stop(p->txaio);
 	nni_aio_stop(p->negoaio);
+	nng_stream_stop(p->conn);
 }
 
 static int
@@ -674,6 +675,8 @@ tcptran_ep_stop(void *arg)
 
 	nni_aio_stop(ep->timeaio);
 	nni_aio_stop(ep->connaio);
+	nng_stream_dialer_stop(ep->dialer);
+	nng_stream_listener_stop(ep->listener);
 }
 
 static void
diff --git a/src/sp/transport/tls/tls.c b/src/sp/transport/tls/tls.c
index 7a907e8e..dcb100f5 100644
--- a/src/sp/transport/tls/tls.c
+++ b/src/sp/transport/tls/tls.c
@@ -657,6 +657,8 @@ tlstran_ep_stop(void *arg)
 
 	nni_aio_stop(ep->timeaio);
 	nni_aio_stop(ep->connaio);
+	nng_stream_dialer_stop(ep->dialer);
+	nng_stream_listener_stop(ep->listener);
 }
 
 static void
diff --git a/src/sp/transport/ws/websocket.c b/src/sp/transport/ws/websocket.c
index 27cc6434..16929ec9 100644
--- a/src/sp/transport/ws/websocket.c
+++ b/src/sp/transport/ws/websocket.c
@@ -183,6 +183,7 @@ wstran_pipe_stop(void *arg)
 
 	nni_aio_stop(&p->rxaio);
 	nni_aio_stop(&p->txaio);
+	nng_stream_stop(p->ws);
 }
 
 static int
@@ -214,9 +215,7 @@ wstran_pipe_close(void *arg)
 	nni_aio_close(&p->rxaio);
 	nni_aio_close(&p->txaio);
 
-	nni_mtx_lock(&p->mtx);
 	nng_stream_close(p->ws);
-	nni_mtx_unlock(&p->mtx);
 }
 
 static int
@@ -373,6 +372,7 @@ wstran_dialer_stop(void *arg)
 	ws_dialer *d = arg;
 
 	nni_aio_stop(&d->connaio);
+	nng_stream_dialer_stop(d->dialer);
 }
 
 static void
@@ -392,6 +392,7 @@ wstran_listener_stop(void *arg)
 	ws_listener *l = arg;
 
 	nni_aio_stop(&l->accaio);
+	nng_stream_listener_stop(l->listener);
 }
 
 static void
@@ -421,6 +422,7 @@ wstran_connect_cb(void *arg)
 	}
 	if ((uaio = nni_list_first(&d->aios)) == NULL) {
 		// The client stopped caring about this!
+		nng_stream_stop(ws);
 		nng_stream_free(ws);
 		nni_mtx_unlock(&d->mtx);
 		return;
@@ -430,6 +432,7 @@ wstran_connect_cb(void *arg)
 	if ((rv = nni_aio_result(caio)) != 0) {
 		nni_aio_finish_error(uaio, rv);
 	} else if ((rv = wstran_pipe_alloc(&p, ws)) != 0) {
+		nng_stream_stop(ws);
 		nng_stream_free(ws);
 		nni_aio_finish_error(uaio, rv);
 	} else {