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.

6 files changed, 62 insertions, 19 deletions

diff --git a/src/platform/windows/win_ipcconn.c b/src/platform/windows/win_ipcconn.c
index 5f540a6c..56a688a4 100644
--- a/src/platform/windows/win_ipcconn.c
+++ b/src/platform/windows/win_ipcconn.c
@@ -352,7 +352,7 @@ ipc_close(void *arg)
 }
 
 static void
-ipc_free(void *arg)
+ipc_stop(void *arg)
 {
 	ipc_conn *c = arg;
 	nni_aio  *aio;
@@ -381,6 +381,14 @@ ipc_free(void *arg)
 		DisconnectNamedPipe(f);
 		CloseHandle(f);
 	}
+}
+
+static void
+ipc_free(void *arg)
+{
+	ipc_conn *c = arg;
+
+	ipc_stop(c);
 
 	nni_cv_fini(&c->cv);
 	nni_mtx_fini(&c->mtx);
@@ -460,6 +468,7 @@ nni_win_ipc_init(
 	c->sa             = *sa;
 	c->stream.s_free  = ipc_free;
 	c->stream.s_close = ipc_close;
+	c->stream.s_stop  = ipc_stop;
 	c->stream.s_send  = ipc_send;
 	c->stream.s_recv  = ipc_recv;
 	c->stream.s_get   = ipc_get;
diff --git a/src/platform/windows/win_ipcdial.c b/src/platform/windows/win_ipcdial.c
index cb826308..347ee39b 100644
--- a/src/platform/windows/win_ipcdial.c
+++ b/src/platform/windows/win_ipcdial.c
@@ -192,10 +192,17 @@ ipc_dialer_close(void *arg)
 }
 
 static void
-ipc_dialer_free(void *arg)
+ipc_dialer_stop(void *arg)
 {
 	ipc_dialer *d = arg;
 	ipc_dialer_close(d);
+}
+
+static void
+ipc_dialer_free(void *arg)
+{
+	ipc_dialer *d = arg;
+	ipc_dialer_stop(d);
 	if (d->path) {
 		nni_strfree(d->path);
 	}
@@ -260,6 +267,7 @@ nni_ipc_dialer_alloc(nng_stream_dialer **dp, const nng_url *url)
 	d->closed             = false;
 	d->sd.sd_free         = ipc_dialer_free;
 	d->sd.sd_close        = ipc_dialer_close;
+	d->sd.sd_stop         = ipc_dialer_stop;
 	d->sd.sd_dial         = ipc_dialer_dial;
 	d->sd.sd_get          = ipc_dialer_get;
 	d->sd.sd_set          = ipc_dialer_set;
diff --git a/src/platform/windows/win_ipclisten.c b/src/platform/windows/win_ipclisten.c
index 98cb8273..d09c98c7 100644
--- a/src/platform/windows/win_ipclisten.c
+++ b/src/platform/windows/win_ipclisten.c
@@ -300,8 +300,18 @@ ipc_listener_close(void *arg)
 		nni_aio_list_remove(aio);
 		nni_aio_finish_error(aio, NNG_ECLOSED);
 	}
-	bool accepting = l->accepting;
 	nni_mtx_unlock(&l->mtx);
+}
+
+static void
+ipc_listener_stop(void *arg)
+{
+	ipc_listener *l = arg;
+
+	ipc_listener_close(l);
+
+	nni_mtx_lock(&l->mtx);
+	bool accepting = l->accepting;
 
 	// This craziness because CancelIoEx on ConnectNamedPipe
 	// seems to be incredibly unreliable. It does work, sometimes,
@@ -309,6 +319,7 @@ ipc_listener_close(void *arg)
 	// to be retired in favor of UNIX domain sockets anyway.
 
 	while (accepting) {
+		nni_mtx_unlock(&l->mtx);
 		if (!CancelIoEx(l->f, &l->io.olpd)) {
 			// operation not found probably
 			// We just inject a safety sleep to
@@ -323,8 +334,8 @@ ipc_listener_close(void *arg)
 		nng_msleep(100);
 		nni_mtx_lock(&l->mtx);
 		accepting = l->accepting;
-		nni_mtx_unlock(&l->mtx);
 	}
+	nni_mtx_unlock(&l->mtx);
 	DisconnectNamedPipe(l->f);
 	CloseHandle(l->f);
 }
@@ -360,6 +371,7 @@ nni_ipc_listener_alloc(nng_stream_listener **lp, const nng_url *url)
 	l->sec_attr.bInheritHandle       = FALSE;
 	l->sa.s_ipc.sa_family            = NNG_AF_IPC;
 	l->sl.sl_free                    = ipc_listener_free;
+	l->sl.sl_stop                    = ipc_listener_stop;
 	l->sl.sl_close                   = ipc_listener_close;
 	l->sl.sl_listen                  = ipc_listener_listen;
 	l->sl.sl_accept                  = ipc_listener_accept;
diff --git a/src/platform/windows/win_tcpconn.c b/src/platform/windows/win_tcpconn.c
index 3f3acd17..2f2fd378 100644
--- a/src/platform/windows/win_tcpconn.c
+++ b/src/platform/windows/win_tcpconn.c
@@ -180,6 +180,7 @@ tcp_send_start(nni_tcp_conn *c)
 			return;
 		}
 	}
+	nni_cv_wake(&c->cv);
 }
 
 static void
@@ -266,17 +267,6 @@ tcp_close(void *arg)
 			closesocket(s);
 		}
 	}
-	now = nni_clock();
-	// wait up to a maximum of 10 seconds before assuming something is
-	// badly amiss. from what we can tell, this doesn't happen, and we do
-	// see the timer expire properly, but this safeguard can prevent a
-	// hang.
-	while ((c->recving || c->sending) &&
-	    ((nni_clock() - now) < (NNI_SECOND * 10))) {
-		nni_mtx_unlock(&c->mtx);
-		nni_msleep(1);
-		nni_mtx_lock(&c->mtx);
-	}
 	nni_mtx_unlock(&c->mtx);
 }
 
@@ -369,21 +359,28 @@ tcp_set(void *arg, const char *name, const void *buf, size_t sz, nni_type t)
 }
 
 static void
-tcp_free(void *arg)
+tcp_stop(void *arg)
 {
 	nni_tcp_conn *c = arg;
 	tcp_close(c);
 
 	nni_mtx_lock(&c->mtx);
-	while ((!nni_list_empty(&c->recv_aios)) ||
+	while (c->recving || c->sending || (!nni_list_empty(&c->recv_aios)) ||
 	    (!nni_list_empty(&c->send_aios))) {
 		nni_cv_wait(&c->cv);
 	}
 	nni_mtx_unlock(&c->mtx);
-
 	if (c->s != INVALID_SOCKET) {
 		closesocket(c->s);
 	}
+}
+
+static void
+tcp_free(void *arg)
+{
+	nni_tcp_conn *c = arg;
+	tcp_stop(c);
+
 	nni_cv_fini(&c->cv);
 	nni_mtx_fini(&c->mtx);
 	NNI_FREE_STRUCT(c);
@@ -410,6 +407,7 @@ nni_win_tcp_init(nni_tcp_conn **connp, SOCKET s)
 	nni_aio_list_init(&c->send_aios);
 	c->conn_aio    = NULL;
 	c->ops.s_close = tcp_close;
+	c->ops.s_stop  = tcp_stop;
 	c->ops.s_free  = tcp_free;
 	c->ops.s_send  = tcp_send;
 	c->ops.s_recv  = tcp_recv;
diff --git a/src/platform/windows/win_tcpdial.c b/src/platform/windows/win_tcpdial.c
index 5492d265..c0385648 100644
--- a/src/platform/windows/win_tcpdial.c
+++ b/src/platform/windows/win_tcpdial.c
@@ -95,9 +95,16 @@ static nni_reap_list tcp_dialer_reap_list = {
 };
 
 void
-nni_tcp_dialer_fini(nni_tcp_dialer *d)
+nni_tcp_dialer_stop(nni_tcp_dialer *d)
 {
 	nni_tcp_dialer_close(d);
+	// TODO: wait for conn_io.olpd?
+}
+
+void
+nni_tcp_dialer_fini(nni_tcp_dialer *d)
+{
+	nni_tcp_dialer_stop(d);
 	nni_mtx_lock(&d->mtx);
 	if (!nni_list_empty(&d->aios)) {
 		nni_mtx_unlock(&d->mtx);
@@ -155,6 +162,8 @@ tcp_dial_cb(nni_win_io *io, int rv, size_t cnt)
 	nni_mtx_unlock(&d->mtx);
 
 	if (rv != 0) {
+		nng_stream_close(&c->ops);
+		nng_stream_stop(&c->ops);
 		nng_stream_free(&c->ops);
 		nni_aio_finish_error(aio, rv);
 	} else {
diff --git a/src/platform/windows/win_tcplisten.c b/src/platform/windows/win_tcplisten.c
index 021483d4..4e4fb090 100644
--- a/src/platform/windows/win_tcplisten.c
+++ b/src/platform/windows/win_tcplisten.c
@@ -178,6 +178,13 @@ static nni_reap_list tcp_listener_reap_list = {
 };
 
 void
+nni_tcp_listener_stop(nni_tcp_listener *l)
+{
+	nni_tcp_listener_close(l);
+	// TODO: maybe wait for l->l_accept_io.olpd to finish?
+}
+
+void
 nni_tcp_listener_fini(nni_tcp_listener *l)
 {
 	nni_tcp_listener_close(l);