fixes #419 want to nni_aio_stop without blocking (#428)

* fixes #419 want to nni_aio_stop without blocking

This actually introduces an nni_aio_close() API that causes
nni_aio_begin to return NNG_ECLOSED, while scheduling a callback
on the AIO to do an NNG_ECLOSED as well.  This should be called
in non-blocking close() contexts instead of nni_aio_stop(), and
the cases where we call nni_aio_fini() multiple times are updated
updated to add nni_aio_stop() calls on all "interlinked" aios before
finalizing them.

Furthermore, we call nni_aio_close() as soon as practical in the
close path.  This closes an annoying race condition where the
callback from a lower subsystem could wind up rescheduling an
operation that we wanted to abort.

6 files changed, 46 insertions, 28 deletions

diff --git a/src/transport/inproc/inproc.c b/src/transport/inproc/inproc.c
index 0f159d3a..bc51d971 100644
--- a/src/transport/inproc/inproc.c
+++ b/src/transport/inproc/inproc.c
@@ -198,7 +198,7 @@ nni_inproc_ep_init(void **epp, nni_url *url, nni_sock *sock, int mode)
 	}
 
 	ep->mode  = mode;
-	ep->proto = nni_sock_proto(sock);
+	ep->proto = nni_sock_proto_id(sock);
 	NNI_LIST_INIT(&ep->clients, nni_inproc_ep, node);
 	nni_aio_list_init(&ep->aios);
 
@@ -452,7 +452,7 @@ static nni_tran_pipe_option nni_inproc_pipe_options[] = {
 	},
 };
 
-static nni_tran_pipe nni_inproc_pipe_ops = {
+static nni_tran_pipe_ops nni_inproc_pipe_ops = {
 	.p_fini    = nni_inproc_pipe_fini,
 	.p_send    = nni_inproc_pipe_send,
 	.p_recv    = nni_inproc_pipe_recv,
@@ -468,7 +468,7 @@ static nni_tran_ep_option nni_inproc_ep_options[] = {
 	},
 };
 
-static nni_tran_ep nni_inproc_ep_ops = {
+static nni_tran_ep_ops nni_inproc_ep_ops = {
 	.ep_init    = nni_inproc_ep_init,
 	.ep_fini    = nni_inproc_ep_fini,
 	.ep_connect = nni_inproc_ep_connect,
diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c
index 2347e24c..1740bfcb 100644
--- a/src/transport/ipc/ipc.c
+++ b/src/transport/ipc/ipc.c
@@ -81,6 +81,10 @@ nni_ipc_pipe_close(void *arg)
 {
 	nni_ipc_pipe *pipe = arg;
 
+	nni_aio_close(pipe->rxaio);
+	nni_aio_close(pipe->txaio);
+	nni_aio_close(pipe->negaio);
+
 	nni_plat_ipc_pipe_close(pipe->ipp);
 }
 
@@ -644,7 +648,7 @@ nni_ipc_ep_init(void **epp, nni_url *url, nni_sock *sock, int mode)
 		nni_ipc_ep_fini(ep);
 		return (rv);
 	}
-	ep->proto = nni_sock_proto(sock);
+	ep->proto = nni_sock_proto_id(sock);
 
 	*epp = ep;
 	return (0);
@@ -655,11 +659,11 @@ nni_ipc_ep_close(void *arg)
 {
 	nni_ipc_ep *ep = arg;
 
+	nni_aio_close(ep->aio);
+
 	nni_mtx_lock(&ep->mtx);
 	nni_plat_ipc_ep_close(ep->iep);
 	nni_mtx_unlock(&ep->mtx);
-
-	nni_aio_stop(ep->aio);
 }
 
 static int
@@ -883,7 +887,7 @@ static nni_tran_pipe_option nni_ipc_pipe_options[] = {
 	},
 };
 
-static nni_tran_pipe nni_ipc_pipe_ops = {
+static nni_tran_pipe_ops nni_ipc_pipe_ops = {
 	.p_fini    = nni_ipc_pipe_fini,
 	.p_start   = nni_ipc_pipe_start,
 	.p_send    = nni_ipc_pipe_send,
@@ -924,7 +928,7 @@ static nni_tran_ep_option nni_ipc_ep_options[] = {
 	},
 };
 
-static nni_tran_ep nni_ipc_ep_ops = {
+static nni_tran_ep_ops nni_ipc_ep_ops = {
 	.ep_init    = nni_ipc_ep_init,
 	.ep_fini    = nni_ipc_ep_fini,
 	.ep_connect = nni_ipc_ep_connect,
diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c
index f2cdf8ac..22217699 100644
--- a/src/transport/tcp/tcp.c
+++ b/src/transport/tcp/tcp.c
@@ -81,9 +81,13 @@ nni_tcp_tran_fini(void)
 static void
 nni_tcp_pipe_close(void *arg)
 {
-	nni_tcp_pipe *pipe = arg;
+	nni_tcp_pipe *p = arg;
+
+	nni_aio_close(p->rxaio);
+	nni_aio_close(p->txaio);
+	nni_aio_close(p->negaio);
 
-	nni_plat_tcp_pipe_close(pipe->tpp);
+	nni_plat_tcp_pipe_close(p->tpp);
 }
 
 static void
@@ -666,7 +670,7 @@ nni_tcp_ep_init(void **epp, nni_url *url, nni_sock *sock, int mode)
 		nni_tcp_ep_fini(ep);
 		return (rv);
 	}
-	ep->proto     = nni_sock_proto(sock);
+	ep->proto     = nni_sock_proto_id(sock);
 	ep->mode      = mode;
 	ep->nodelay   = true;
 	ep->keepalive = false;
@@ -680,11 +684,11 @@ nni_tcp_ep_close(void *arg)
 {
 	nni_tcp_ep *ep = arg;
 
+	nni_aio_close(ep->aio);
+
 	nni_mtx_lock(&ep->mtx);
 	nni_plat_tcp_ep_close(ep->tep);
 	nni_mtx_unlock(&ep->mtx);
-
-	nni_aio_stop(ep->aio);
 }
 
 static int
@@ -910,7 +914,7 @@ static nni_tran_pipe_option nni_tcp_pipe_options[] = {
 	},
 };
 
-static nni_tran_pipe nni_tcp_pipe_ops = {
+static nni_tran_pipe_ops nni_tcp_pipe_ops = {
 	.p_fini    = nni_tcp_pipe_fini,
 	.p_start   = nni_tcp_pipe_start,
 	.p_send    = nni_tcp_pipe_send,
@@ -951,7 +955,7 @@ static nni_tran_ep_option nni_tcp_ep_options[] = {
 	},
 };
 
-static nni_tran_ep nni_tcp_ep_ops = {
+static nni_tran_ep_ops nni_tcp_ep_ops = {
 	.ep_init    = nni_tcp_ep_init,
 	.ep_fini    = nni_tcp_ep_fini,
 	.ep_connect = nni_tcp_ep_connect,
diff --git a/src/transport/tls/tls.c b/src/transport/tls/tls.c
index c863a85e..21557270 100644
--- a/src/transport/tls/tls.c
+++ b/src/transport/tls/tls.c
@@ -91,6 +91,10 @@ nni_tls_pipe_close(void *arg)
 {
 	nni_tls_pipe *p = arg;
 
+	nni_aio_close(p->rxaio);
+	nni_aio_close(p->txaio);
+	nni_aio_close(p->negaio);
+
 	nni_tls_close(p->tls);
 }
 
@@ -687,7 +691,7 @@ nni_tls_ep_init(void **epp, nni_url *url, nni_sock *sock, int mode)
 			return (rv);
 		}
 	}
-	ep->proto    = nni_sock_proto(sock);
+	ep->proto    = nni_sock_proto_id(sock);
 	ep->authmode = authmode;
 
 	*epp = ep;
@@ -699,11 +703,11 @@ nni_tls_ep_close(void *arg)
 {
 	nni_tls_ep *ep = arg;
 
+	nni_aio_close(ep->aio);
+
 	nni_mtx_lock(&ep->mtx);
 	nni_plat_tcp_ep_close(ep->tep);
 	nni_mtx_unlock(&ep->mtx);
-
-	nni_aio_stop(ep->aio);
 }
 
 static int
@@ -1036,7 +1040,7 @@ static nni_tran_pipe_option nni_tls_pipe_options[] = {
 	},
 };
 
-static nni_tran_pipe nni_tls_pipe_ops = {
+static nni_tran_pipe_ops nni_tls_pipe_ops = {
 	.p_fini    = nni_tls_pipe_fini,
 	.p_start   = nni_tls_pipe_start,
 	.p_send    = nni_tls_pipe_send,
@@ -1107,7 +1111,7 @@ static nni_tran_ep_option nni_tls_ep_options[] = {
 	},
 };
 
-static nni_tran_ep nni_tls_ep_ops = {
+static nni_tran_ep_ops nni_tls_ep_ops = {
 	.ep_init    = nni_tls_ep_init,
 	.ep_fini    = nni_tls_ep_fini,
 	.ep_connect = nni_tls_ep_connect,
diff --git a/src/transport/ws/websocket.c b/src/transport/ws/websocket.c
index 12f3aeb5..a4081b25 100644
--- a/src/transport/ws/websocket.c
+++ b/src/transport/ws/websocket.c
@@ -205,6 +205,9 @@ ws_pipe_close(void *arg)
 {
 	ws_pipe *p = arg;
 
+	nni_aio_close(p->rxaio);
+	nni_aio_close(p->txaio);
+
 	nni_mtx_lock(&p->mtx);
 	nni_ws_close(p->ws);
 	nni_mtx_unlock(&p->mtx);
@@ -588,7 +591,7 @@ static nni_tran_pipe_option ws_pipe_options[] = {
 	}
 };
 
-static nni_tran_pipe ws_pipe_ops = {
+static nni_tran_pipe_ops ws_pipe_ops = {
 	.p_fini    = ws_pipe_fini,
 	.p_send    = ws_pipe_send,
 	.p_recv    = ws_pipe_recv,
@@ -690,6 +693,9 @@ ws_ep_close(void *arg)
 {
 	ws_ep *ep = arg;
 
+	nni_aio_close(ep->accaio);
+	nni_aio_close(ep->connaio);
+
 	if (ep->mode == NNI_EP_MODE_LISTEN) {
 		nni_ws_listener_close(ep->listener);
 	} else {
@@ -750,8 +756,8 @@ ws_ep_init(void **epp, nni_url *url, nni_sock *sock, int mode)
 	nni_aio_list_init(&ep->aios);
 
 	ep->mode   = mode;
-	ep->lproto = nni_sock_proto(sock);
-	ep->rproto = nni_sock_peer(sock);
+	ep->lproto = nni_sock_proto_id(sock);
+	ep->rproto = nni_sock_peer_id(sock);
 
 	if (mode == NNI_EP_MODE_DIAL) {
 		pname = nni_sock_peer_name(sock);
@@ -795,7 +801,7 @@ ws_tran_fini(void)
 {
 }
 
-static nni_tran_ep ws_ep_ops = {
+static nni_tran_ep_ops ws_ep_ops = {
 	.ep_init    = ws_ep_init,
 	.ep_fini    = ws_ep_fini,
 	.ep_connect = ws_ep_connect,
@@ -1015,7 +1021,7 @@ static nni_tran_ep_option wss_ep_options[] = {
 	},
 };
 
-static nni_tran_ep wss_ep_ops = {
+static nni_tran_ep_ops wss_ep_ops = {
 	.ep_init    = ws_ep_init,
 	.ep_fini    = ws_ep_fini,
 	.ep_connect = ws_ep_connect,
diff --git a/src/transport/zerotier/zerotier.c b/src/transport/zerotier/zerotier.c
index f7139b94..46fe476e 100644
--- a/src/transport/zerotier/zerotier.c
+++ b/src/transport/zerotier/zerotier.c
@@ -2182,7 +2182,7 @@ zt_ep_init(void **epp, nni_url *url, nni_sock *sock, int mode)
 	ep->ze_ping_time  = zt_ping_time;
 	ep->ze_conn_time  = zt_conn_time;
 	ep->ze_conn_tries = zt_conn_tries;
-	ep->ze_proto      = nni_sock_proto(sock);
+	ep->ze_proto      = nni_sock_proto_id(sock);
 
 	nni_aio_list_init(&ep->ze_aios);
 
@@ -2888,7 +2888,7 @@ static nni_tran_pipe_option zt_pipe_options[] = {
 	},
 };
 
-static nni_tran_pipe zt_pipe_ops = {
+static nni_tran_pipe_ops zt_pipe_ops = {
 	.p_fini    = zt_pipe_fini,
 	.p_start   = zt_pipe_start,
 	.p_send    = zt_pipe_send,
@@ -2983,7 +2983,7 @@ static nni_tran_ep_option zt_ep_options[] = {
 	},
 };
 
-static nni_tran_ep zt_ep_ops = {
+static nni_tran_ep_ops zt_ep_ops = {
 	.ep_init    = zt_ep_init,
 	.ep_fini    = zt_ep_fini,
 	.ep_connect = zt_ep_connect,