Revert "fixes #599 nng_dial sync should not return until added to socket"

This changeset needs work.  We are seeing errors described by
This reverts commit d7f7c896c0ede24249ef63b1e45b1878bf4bd473.

4 files changed, 1101 insertions, 907 deletions

diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c
index 3d798d4c..ee72d6d8 100644
--- a/src/transport/ipc/ipc.c
+++ b/src/transport/ipc/ipc.c
@@ -23,20 +23,14 @@
 typedef struct ipctran_pipe     ipctran_pipe;
 typedef struct ipctran_dialer   ipctran_dialer;
 typedef struct ipctran_listener ipctran_listener;
-typedef struct ipctran_ep       ipctran_ep;
 
 // ipc_pipe is one end of an IPC connection.
 struct ipctran_pipe {
-	nni_ipc_conn *  conn;
-	uint16_t        peer;
-	uint16_t        proto;
-	size_t          rcvmax;
-	nni_atomic_flag reaped;
-	nni_sockaddr    sa;
-	nni_list_node   node;
-	ipctran_ep *    ep;
-	nni_reap_item   reap;
-	bool            closed;
+	nni_ipc_conn *conn;
+	uint16_t      peer;
+	uint16_t      proto;
+	size_t        rcvmax;
+	nni_sockaddr  sa;
 
 	uint8_t txhead[1 + sizeof(uint64_t)];
 	uint8_t rxhead[1 + sizeof(uint64_t)];
@@ -47,34 +41,32 @@ struct ipctran_pipe {
 
 	nni_list recvq;
 	nni_list sendq;
-	nni_aio *useraio;
+	nni_aio *user_negaio;
 	nni_aio *txaio;
 	nni_aio *rxaio;
-	nni_aio *negoaio;
-	nni_aio *connaio;
+	nni_aio *negaio;
 	nni_msg *rxmsg;
 	nni_mtx  mtx;
 };
 
-struct ipctran_ep {
-	nni_mtx       mtx;
-	nni_list      pipes;
-	bool          fini;
-	size_t        rcvmax;
-	uint16_t      proto;
-	nni_cb        dtor;
-	nni_sockaddr  sa;
-	nni_reap_item reap;
-};
-
 struct ipctran_dialer {
-	ipctran_ep      ep;
+	nni_sockaddr    sa;
 	nni_ipc_dialer *dialer;
+	uint16_t        proto;
+	size_t          rcvmax;
+	nni_aio *       aio;
+	nni_aio *       user_aio;
+	nni_mtx         mtx;
 };
 
 struct ipctran_listener {
-	ipctran_ep        ep;
+	nni_sockaddr      sa;
 	nni_ipc_listener *listener;
+	uint16_t          proto;
+	size_t            rcvmax;
+	nni_aio *         aio;
+	nni_aio *         user_aio;
+	nni_mtx           mtx;
 };
 
 static void ipctran_pipe_send_start(ipctran_pipe *);
@@ -82,8 +74,8 @@ static void ipctran_pipe_recv_start(ipctran_pipe *);
 static void ipctran_pipe_send_cb(void *);
 static void ipctran_pipe_recv_cb(void *);
 static void ipctran_pipe_nego_cb(void *);
-static void ipctran_pipe_conn_cb(void *);
-static void ipctran_pipe_reap(ipctran_pipe *);
+static void ipctran_dialer_cb(void *);
+static void ipctran_listener_cb(void *);
 
 static int
 ipctran_init(void)
@@ -103,8 +95,7 @@ ipctran_pipe_close(void *arg)
 
 	nni_aio_close(p->rxaio);
 	nni_aio_close(p->txaio);
-	nni_aio_close(p->negoaio);
-	nni_aio_close(p->connaio);
+	nni_aio_close(p->negaio);
 
 	nni_ipc_conn_close(p->conn);
 }
@@ -114,39 +105,19 @@ ipctran_pipe_stop(void *arg)
 {
 	ipctran_pipe *p = arg;
 
-	nni_mtx_lock(&p->mtx);
-	p->closed = true;
-	nni_mtx_unlock(&p->mtx);
-
 	nni_aio_stop(p->rxaio);
 	nni_aio_stop(p->txaio);
-	nni_aio_stop(p->negoaio);
-	nni_aio_stop(p->connaio);
+	nni_aio_stop(p->negaio);
 }
 
 static void
 ipctran_pipe_fini(void *arg)
 {
 	ipctran_pipe *p = arg;
-	ipctran_ep *  ep;
-
-	if (p == NULL) {
-		return;
-	}
-	ipctran_pipe_stop(p);
-	if ((ep = p->ep) != NULL) {
-		nni_mtx_lock(&ep->mtx);
-		nni_list_remove(&ep->pipes, p);
-		if (ep->fini && nni_list_empty(&ep->pipes)) {
-			nni_reap(&ep->reap, ep->dtor, ep);
-		}
-		nni_mtx_unlock(&ep->mtx);
-	}
 
 	nni_aio_fini(p->rxaio);
 	nni_aio_fini(p->txaio);
-	nni_aio_fini(p->negoaio);
-	nni_aio_fini(p->connaio);
+	nni_aio_fini(p->negaio);
 	if (p->conn != NULL) {
 		nni_ipc_conn_fini(p->conn);
 	}
@@ -157,16 +128,8 @@ ipctran_pipe_fini(void *arg)
 	NNI_FREE_STRUCT(p);
 }
 
-static void
-ipctran_pipe_reap(ipctran_pipe *p)
-{
-	if (!nni_atomic_flag_test_and_set(&p->reaped)) {
-		nni_reap(&p->reap, ipctran_pipe_fini, p);
-	}
-}
-
 static int
-ipctran_pipe_init(ipctran_pipe **pipep, ipctran_ep *ep)
+ipctran_pipe_init(ipctran_pipe **pipep, void *conn)
 {
 	ipctran_pipe *p;
 	int           rv;
@@ -175,75 +138,51 @@ ipctran_pipe_init(ipctran_pipe **pipep, ipctran_ep *ep)
 		return (NNG_ENOMEM);
 	}
 	nni_mtx_init(&p->mtx);
-	NNI_LIST_NODE_INIT(&p->node);
 	if (((rv = nni_aio_init(&p->txaio, ipctran_pipe_send_cb, p)) != 0) ||
 	    ((rv = nni_aio_init(&p->rxaio, ipctran_pipe_recv_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->negoaio, ipctran_pipe_nego_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->connaio, ipctran_pipe_conn_cb, p)) != 0)) {
-		ipctran_pipe_reap(p);
+	    ((rv = nni_aio_init(&p->negaio, ipctran_pipe_nego_cb, p)) != 0)) {
+		ipctran_pipe_fini(p);
 		return (rv);
 	}
 	nni_aio_list_init(&p->sendq);
 	nni_aio_list_init(&p->recvq);
 
-	nni_list_append(&ep->pipes, p);
-	p->ep     = ep;
-	p->proto  = ep->proto;
-	p->rcvmax = ep->rcvmax;
-	p->sa     = ep->sa;
-
+	p->conn = conn;
+#if 0
+	p->proto              = ep->proto;
+	p->rcvmax             = ep->rcvmax;
+	p->sa.s_ipc.sa_family = NNG_AF_IPC;
+	p->sa                 = ep->sa;
+#endif
 	*pipep = p;
 	return (0);
 }
 
 static void
-ipctran_pipe_conn_cb(void *arg)
+ipctran_pipe_nego_cancel(nni_aio *aio, int rv)
 {
-	ipctran_pipe *p   = arg;
-	nni_aio *     aio = p->connaio;
-	nni_iov       iov;
-	int           rv;
+	ipctran_pipe *p = nni_aio_get_prov_data(aio);
 
-	nni_mtx_lock(&p->ep->mtx);
-	if ((rv = nni_aio_result(aio)) != 0) {
-		nni_aio *uaio;
-		if ((uaio = p->useraio) != NULL) {
-			p->useraio = NULL;
-			nni_aio_finish_error(uaio, rv);
-		}
-		nni_mtx_unlock(&p->ep->mtx);
-		ipctran_pipe_reap(p);
+	nni_mtx_lock(&p->mtx);
+	if (p->user_negaio != aio) {
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
+	p->user_negaio = NULL;
+	nni_mtx_unlock(&p->mtx);
 
-	p->conn      = nni_aio_get_output(aio, 0);
-	p->txhead[0] = 0;
-	p->txhead[1] = 'S';
-	p->txhead[2] = 'P';
-	p->txhead[3] = 0;
-	NNI_PUT16(&p->txhead[4], p->proto);
-	NNI_PUT16(&p->txhead[6], 0);
-
-	p->gotrxhead  = 0;
-	p->gottxhead  = 0;
-	p->wantrxhead = 8;
-	p->wanttxhead = 8;
-	iov.iov_len   = 8;
-	iov.iov_buf   = &p->txhead[0];
-	nni_aio_set_iov(p->negoaio, 1, &iov);
-	nni_ipc_conn_send(p->conn, p->negoaio);
-	nni_mtx_unlock(&p->ep->mtx);
+	nni_aio_abort(p->negaio, rv);
+	nni_aio_finish_error(aio, rv);
 }
 
 static void
 ipctran_pipe_nego_cb(void *arg)
 {
 	ipctran_pipe *p   = arg;
-	nni_aio *     aio = p->negoaio;
-	nni_aio *     uaio;
+	nni_aio *     aio = p->negaio;
 	int           rv;
 
-	nni_mtx_lock(&p->ep->mtx);
+	nni_mtx_lock(&p->mtx);
 	if ((rv = nni_aio_result(aio)) != 0) {
 		goto done;
 	}
@@ -262,7 +201,7 @@ ipctran_pipe_nego_cb(void *arg)
 		nni_aio_set_iov(aio, 1, &iov);
 		// send it down...
 		nni_ipc_conn_send(p->conn, aio);
-		nni_mtx_unlock(&p->ep->mtx);
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
 	if (p->gotrxhead < p->wantrxhead) {
@@ -271,7 +210,7 @@ ipctran_pipe_nego_cb(void *arg)
 		iov.iov_buf = &p->rxhead[p->gotrxhead];
 		nni_aio_set_iov(aio, 1, &iov);
 		nni_ipc_conn_recv(p->conn, aio);
-		nni_mtx_unlock(&p->ep->mtx);
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
 	// We have both sent and received the headers.  Lets check the
@@ -286,19 +225,11 @@ ipctran_pipe_nego_cb(void *arg)
 	NNI_GET16(&p->rxhead[4], p->peer);
 
 done:
-
-	if ((uaio = p->useraio) != NULL) {
-		p->useraio = NULL;
-		if (rv == 0) {
-			nni_aio_set_output(uaio, 0, p);
-			nni_aio_finish(uaio, 0, 0);
-			nni_mtx_unlock(&p->ep->mtx);
-			return;
-		}
-		nni_aio_finish_error(uaio, rv);
+	if ((aio = p->user_negaio) != NULL) {
+		p->user_negaio = NULL;
+		nni_aio_finish(aio, rv, 0);
 	}
-	nni_mtx_unlock(&p->ep->mtx);
-	ipctran_pipe_reap(p);
+	nni_mtx_unlock(&p->mtx);
 }
 
 static void
@@ -312,20 +243,17 @@ ipctran_pipe_send_cb(void *arg)
 	nni_aio *     txaio = p->txaio;
 
 	nni_mtx_lock(&p->mtx);
+	aio = nni_list_first(&p->sendq);
 
-	rv = p->closed ? NNG_ECLOSED : nni_aio_result(txaio);
-
-	if (rv != 0) {
+	if ((rv = nni_aio_result(txaio)) != 0) {
 		// Intentionally we do not queue up another transfer.
 		// There's an excellent chance that the pipe is no longer
 		// usable, with a partial transfer.
 		// The protocol should see this error, and close the
 		// pipe itself, we hope.
-		while ((aio = nni_list_first(&p->sendq)) != NULL) {
-			nni_aio_list_remove(aio);
-			nni_aio_finish_error(aio, rv);
-		}
+		nni_aio_list_remove(aio);
 		nni_mtx_unlock(&p->mtx);
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
@@ -337,7 +265,6 @@ ipctran_pipe_send_cb(void *arg)
 		return;
 	}
 
-	aio = nni_list_first(&p->sendq);
 	nni_aio_list_remove(aio);
 	ipctran_pipe_send_start(p);
 
@@ -361,6 +288,7 @@ ipctran_pipe_recv_cb(void *arg)
 	nni_aio *     rxaio = p->rxaio;
 
 	nni_mtx_lock(&p->mtx);
+	aio = nni_list_first(&p->recvq);
 
 	if ((rv = nni_aio_result(rxaio)) != 0) {
 		// Error on receive.  This has to cause an error back
@@ -368,10 +296,6 @@ ipctran_pipe_recv_cb(void *arg)
 		// toss it.
 		goto recv_error;
 	}
-	if (p->closed) {
-		rv = NNG_ECLOSED;
-		goto recv_error;
-	}
 
 	n = nni_aio_count(rxaio);
 	nni_aio_iov_advance(rxaio, n);
@@ -430,7 +354,6 @@ ipctran_pipe_recv_cb(void *arg)
 	// Otherwise we got a message read completely.  Let the user know the
 	// good news.
 
-	aio = nni_list_first(&p->recvq);
 	nni_aio_list_remove(aio);
 	msg      = p->rxmsg;
 	p->rxmsg = NULL;
@@ -444,10 +367,7 @@ ipctran_pipe_recv_cb(void *arg)
 	return;
 
 recv_error:
-	while ((aio = nni_list_first(&p->recvq)) != NULL) {
-		nni_aio_list_remove(aio);
-		nni_aio_finish_error(aio, rv);
-	}
+	nni_aio_list_remove(aio);
 	msg      = p->rxmsg;
 	p->rxmsg = NULL;
 	// Intentionally, we do not queue up another receive.
@@ -455,6 +375,7 @@ recv_error:
 	nni_mtx_unlock(&p->mtx);
 
 	nni_msg_free(msg);
+	nni_aio_finish_error(aio, rv);
 }
 
 static void
@@ -531,11 +452,6 @@ ipctran_pipe_send(void *arg, nni_aio *aio)
 		return;
 	}
 	nni_mtx_lock(&p->mtx);
-	if (p->closed) {
-		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, NNG_ECLOSED);
-		return;
-	}
 	if ((rv = nni_aio_schedule(aio, ipctran_pipe_send_cancel, p)) != 0) {
 		nni_mtx_unlock(&p->mtx);
 		nni_aio_finish_error(aio, rv);
@@ -598,11 +514,6 @@ ipctran_pipe_recv(void *arg, nni_aio *aio)
 	}
 	nni_mtx_lock(&p->mtx);
 
-	if (p->closed) {
-		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, NNG_ECLOSED);
-		return;
-	}
 	if ((rv = nni_aio_schedule(aio, ipctran_pipe_recv_cancel, p)) != 0) {
 		nni_mtx_unlock(&p->mtx);
 		nni_aio_finish_error(aio, rv);
@@ -616,6 +527,43 @@ ipctran_pipe_recv(void *arg, nni_aio *aio)
 	nni_mtx_unlock(&p->mtx);
 }
 
+static void
+ipctran_pipe_start(void *arg, nni_aio *aio)
+{
+	ipctran_pipe *p = arg;
+	nni_aio *     negaio;
+	nni_iov       iov;
+	int           rv;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+	nni_mtx_lock(&p->mtx);
+	if ((rv = nni_aio_schedule(aio, ipctran_pipe_nego_cancel, p)) != 0) {
+		nni_mtx_unlock(&p->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	p->txhead[0] = 0;
+	p->txhead[1] = 'S';
+	p->txhead[2] = 'P';
+	p->txhead[3] = 0;
+	NNI_PUT16(&p->txhead[4], p->proto);
+	NNI_PUT16(&p->txhead[6], 0);
+
+	p->user_negaio = aio;
+	p->gotrxhead   = 0;
+	p->gottxhead   = 0;
+	p->wantrxhead  = 8;
+	p->wanttxhead  = 8;
+	negaio         = p->negaio;
+	iov.iov_len    = 8;
+	iov.iov_buf    = &p->txhead[0];
+	nni_aio_set_iov(negaio, 1, &iov);
+	nni_ipc_conn_send(p->conn, negaio);
+	nni_mtx_unlock(&p->mtx);
+}
+
 static uint16_t
 ipctran_pipe_peer(void *arg)
 {
@@ -684,42 +632,21 @@ ipctran_dialer_fini(void *arg)
 {
 	ipctran_dialer *d = arg;
 
-	nni_mtx_lock(&d->ep.mtx);
+	nni_aio_stop(d->aio);
 	if (d->dialer != NULL) {
 		nni_ipc_dialer_fini(d->dialer);
-		d->dialer = NULL;
-	}
-	d->ep.fini = true;
-	if (!nni_list_empty(&d->ep.pipes)) {
-		nni_mtx_unlock(&d->ep.mtx);
-		return;
 	}
-	nni_mtx_unlock(&d->ep.mtx);
-
-	nni_mtx_fini(&d->ep.mtx);
+	nni_aio_fini(d->aio);
+	nni_mtx_fini(&d->mtx);
 	NNI_FREE_STRUCT(d);
 }
 
 static void
-ipctran_ep_close(void *arg)
-{
-	ipctran_ep *  ep = arg;
-	ipctran_pipe *p;
-
-	nni_mtx_lock(&ep->mtx);
-	NNI_LIST_FOREACH (&ep->pipes, p) {
-		nni_aio_close(p->negoaio);
-		nni_aio_close(p->connaio);
-	}
-	nni_mtx_unlock(&ep->mtx);
-}
-
-static void
 ipctran_dialer_close(void *arg)
 {
 	ipctran_dialer *d = arg;
 
-	ipctran_ep_close(&d->ep);
+	nni_aio_close(d->aio);
 	nni_ipc_dialer_close(d->dialer);
 }
 
@@ -733,46 +660,90 @@ ipctran_dialer_init(void **dp, nni_url *url, nni_sock *sock)
 	if ((d = NNI_ALLOC_STRUCT(d)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	nni_mtx_init(&d->ep.mtx);
-	NNI_LIST_INIT(&d->ep.pipes, ipctran_pipe, node);
+	nni_mtx_init(&d->mtx);
 
-	sz                       = sizeof(d->ep.sa.s_ipc.sa_path);
-	d->ep.sa.s_ipc.sa_family = NNG_AF_IPC;
-	d->ep.proto              = nni_sock_proto_id(sock);
-	d->ep.dtor               = ipctran_dialer_fini;
+	sz                    = sizeof(d->sa.s_ipc.sa_path);
+	d->sa.s_ipc.sa_family = NNG_AF_IPC;
 
-	if (nni_strlcpy(d->ep.sa.s_ipc.sa_path, url->u_path, sz) >= sz) {
+	if (nni_strlcpy(d->sa.s_ipc.sa_path, url->u_path, sz) >= sz) {
 		ipctran_dialer_fini(d);
 		return (NNG_EADDRINVAL);
 	}
 
-	if ((rv = nni_ipc_dialer_init(&d->dialer)) != 0) {
+	if (((rv = nni_ipc_dialer_init(&d->dialer)) != 0) ||
+	    ((rv = nni_aio_init(&d->aio, ipctran_dialer_cb, d)) != 0)) {
 		ipctran_dialer_fini(d);
 		return (rv);
 	}
 
+	d->proto = nni_sock_proto_id(sock);
+
 	*dp = d;
 	return (0);
 }
 
 static void
-ipctran_pipe_conn_cancel(nni_aio *aio, int rv)
+ipctran_dialer_cb(void *arg)
 {
-	ipctran_pipe *p = nni_aio_get_prov_data(aio);
+	ipctran_dialer *d = arg;
+	ipctran_pipe *  p;
+	nni_ipc_conn *  conn;
+	nni_aio *       aio;
+	int             rv;
+
+	nni_mtx_lock(&d->mtx);
+	aio = d->user_aio;
+	rv  = nni_aio_result(d->aio);
+
+	if (aio == NULL) {
+		nni_mtx_unlock(&d->mtx);
+		if (rv == 0) {
+			conn = nni_aio_get_output(d->aio, 0);
+			nni_ipc_conn_fini(conn);
+		}
+		return;
+	}
+
+	if (rv != 0) {
+		d->user_aio = NULL;
+		nni_mtx_unlock(&d->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
 
-	nni_mtx_lock(&p->ep->mtx);
-	if (p->useraio != aio) {
-		nni_mtx_unlock(&p->ep->mtx);
+	d->user_aio = NULL;
+	conn        = nni_aio_get_output(d->aio, 0);
+	NNI_ASSERT(conn != NULL);
+	if ((rv = ipctran_pipe_init(&p, conn)) != 0) {
+		nni_mtx_unlock(&d->mtx);
+		nni_ipc_conn_fini(conn);
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
-	// Close the underlying AIOs.  This will abort the operation.
-	// The pipe is removed from pending list at completion callback.
-	p->useraio = NULL;
-	nni_aio_close(p->negoaio);
-	nni_aio_close(p->connaio);
-	nni_mtx_unlock(&p->ep->mtx);
+	p->proto  = d->proto;
+	p->rcvmax = d->rcvmax;
+	p->sa     = d->sa;
+	nni_mtx_unlock(&d->mtx);
+
+	nni_aio_set_output(aio, 0, p);
+	nni_aio_finish(aio, 0, 0);
+}
+
+static void
+ipctran_dialer_cancel(nni_aio *aio, int rv)
+{
+	ipctran_dialer *d = nni_aio_get_prov_data(aio);
+
+	nni_mtx_lock(&d->mtx);
+	if (d->user_aio != aio) {
+		nni_mtx_unlock(&d->mtx);
+		return;
+	}
+	d->user_aio = NULL;
+	nni_mtx_unlock(&d->mtx);
 
+	nni_aio_abort(d->aio, rv);
 	nni_aio_finish_error(aio, rv);
 }
 
@@ -780,48 +751,47 @@ static void
 ipctran_dialer_connect(void *arg, nni_aio *aio)
 {
 	ipctran_dialer *d = arg;
-	ipctran_pipe *  p = NULL;
 	int             rv;
 
 	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
+	nni_mtx_lock(&d->mtx);
+	NNI_ASSERT(d->user_aio == NULL);
 
-	nni_mtx_lock(&d->ep.mtx);
-	if (((rv = ipctran_pipe_init(&p, &d->ep)) != 0) ||
-	    ((rv = nni_aio_schedule(aio, ipctran_pipe_conn_cancel, p)) != 0)) {
-		nni_mtx_unlock(&d->ep.mtx);
+	if ((rv = nni_aio_schedule(aio, ipctran_dialer_cancel, d)) != 0) {
+		nni_mtx_unlock(&d->mtx);
 		nni_aio_finish_error(aio, rv);
-		ipctran_pipe_reap(p);
 		return;
 	}
-	p->useraio = aio;
+	d->user_aio = aio;
 
-	nni_ipc_dialer_dial(d->dialer, &p->sa, p->connaio);
-	nni_mtx_unlock(&d->ep.mtx);
+	nni_ipc_dialer_dial(d->dialer, &d->sa, d->aio);
+	nni_mtx_unlock(&d->mtx);
 }
 
 static int
-ipctran_ep_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t)
+ipctran_dialer_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t)
 {
-	ipctran_ep *ep = arg;
-	int         rv;
-	nni_mtx_lock(&ep->mtx);
-	rv = nni_copyout_size(ep->rcvmax, v, szp, t);
-	nni_mtx_unlock(&ep->mtx);
+	ipctran_dialer *d = arg;
+	int             rv;
+	nni_mtx_lock(&d->mtx);
+	rv = nni_copyout_size(d->rcvmax, v, szp, t);
+	nni_mtx_unlock(&d->mtx);
 	return (rv);
 }
 
 static int
-ipctran_ep_set_recvmaxsz(void *arg, const void *v, size_t sz, nni_opt_type t)
+ipctran_dialer_set_recvmaxsz(
+    void *arg, const void *v, size_t sz, nni_opt_type t)
 {
-	ipctran_ep *ep = arg;
-	size_t      val;
-	int         rv;
+	ipctran_dialer *d = arg;
+	size_t          val;
+	int             rv;
 	if ((rv = nni_copyin_size(&val, v, sz, 0, NNI_MAXSZ, t)) == 0) {
-		nni_mtx_lock(&ep->mtx);
-		ep->rcvmax = val;
-		nni_mtx_unlock(&ep->mtx);
+		nni_mtx_lock(&d->mtx);
+		d->rcvmax = val;
+		nni_mtx_unlock(&d->mtx);
 	}
 	return (rv);
 }
@@ -831,19 +801,12 @@ ipctran_listener_fini(void *arg)
 {
 	ipctran_listener *l = arg;
 
-	nni_mtx_lock(&l->ep.mtx);
+	nni_aio_stop(l->aio);
 	if (l->listener != NULL) {
 		nni_ipc_listener_fini(l->listener);
-		l->listener = NULL;
-	}
-	l->ep.fini = true;
-	if (!nni_list_empty(&l->ep.pipes)) {
-		nni_mtx_unlock(&l->ep.mtx);
-		return;
 	}
-	nni_mtx_unlock(&l->ep.mtx);
-
-	nni_mtx_fini(&l->ep.mtx);
+	nni_aio_fini(l->aio);
+	nni_mtx_fini(&l->mtx);
 	NNI_FREE_STRUCT(l);
 }
 
@@ -857,24 +820,24 @@ ipctran_listener_init(void **lp, nni_url *url, nni_sock *sock)
 	if ((l = NNI_ALLOC_STRUCT(l)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	nni_mtx_init(&l->ep.mtx);
-	NNI_LIST_INIT(&l->ep.pipes, ipctran_pipe, node);
+	nni_mtx_init(&l->mtx);
 
-	sz                       = sizeof(l->ep.sa.s_ipc.sa_path);
-	l->ep.sa.s_ipc.sa_family = NNG_AF_IPC;
-	l->ep.proto              = nni_sock_proto_id(sock);
-	l->ep.dtor               = ipctran_listener_fini;
+	sz                    = sizeof(l->sa.s_ipc.sa_path);
+	l->sa.s_ipc.sa_family = NNG_AF_IPC;
 
-	if (nni_strlcpy(l->ep.sa.s_ipc.sa_path, url->u_path, sz) >= sz) {
+	if (nni_strlcpy(l->sa.s_ipc.sa_path, url->u_path, sz) >= sz) {
 		ipctran_listener_fini(l);
 		return (NNG_EADDRINVAL);
 	}
 
-	if ((rv = nni_ipc_listener_init(&l->listener)) != 0) {
+	if (((rv = nni_ipc_listener_init(&l->listener)) != 0) ||
+	    ((rv = nni_aio_init(&l->aio, ipctran_listener_cb, l)) != 0)) {
 		ipctran_listener_fini(l);
 		return (rv);
 	}
 
+	l->proto = nni_sock_proto_id(sock);
+
 	*lp = l;
 	return (0);
 }
@@ -884,7 +847,7 @@ ipctran_listener_close(void *arg)
 {
 	ipctran_listener *l = arg;
 
-	ipctran_ep_close(&l->ep);
+	nni_aio_close(l->aio);
 	nni_ipc_listener_close(l->listener);
 }
 
@@ -894,35 +857,128 @@ ipctran_listener_bind(void *arg)
 	ipctran_listener *l = arg;
 	int               rv;
 
-	nni_mtx_lock(&l->ep.mtx);
-	rv = nni_ipc_listener_listen(l->listener, &l->ep.sa);
-	nni_mtx_unlock(&l->ep.mtx);
+	nni_mtx_lock(&l->mtx);
+	rv = nni_ipc_listener_listen(l->listener, &l->sa);
+	nni_mtx_unlock(&l->mtx);
 	return (rv);
 }
 
 static void
+ipctran_listener_cb(void *arg)
+{
+	ipctran_listener *l = arg;
+	nni_aio *         aio;
+	int               rv;
+	ipctran_pipe *    p = NULL;
+	nni_ipc_conn *    conn;
+
+	nni_mtx_lock(&l->mtx);
+	rv          = nni_aio_result(l->aio);
+	aio         = l->user_aio;
+	l->user_aio = NULL;
+
+	if (aio == NULL) {
+		nni_mtx_unlock(&l->mtx);
+		if (rv == 0) {
+			conn = nni_aio_get_output(l->aio, 0);
+			nni_ipc_conn_fini(conn);
+		}
+		return;
+	}
+
+	if (rv != 0) {
+		nni_mtx_unlock(&l->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	conn = nni_aio_get_output(l->aio, 0);
+	NNI_ASSERT(conn != NULL);
+
+	// Attempt to allocate the parent pipe.  If this fails we'll
+	// drop the connection (ENOMEM probably).
+	if ((rv = ipctran_pipe_init(&p, conn)) != 0) {
+		nni_mtx_unlock(&l->mtx);
+		nni_ipc_conn_fini(conn);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	p->proto  = l->proto;
+	p->rcvmax = l->rcvmax;
+	p->sa     = l->sa;
+	nni_mtx_unlock(&l->mtx);
+
+	nni_aio_set_output(aio, 0, p);
+	nni_aio_finish(aio, 0, 0);
+}
+
+static void
+ipctran_listener_cancel(nni_aio *aio, int rv)
+{
+	ipctran_listener *l = nni_aio_get_prov_data(aio);
+
+	NNI_ASSERT(rv != 0);
+	nni_mtx_lock(&l->mtx);
+	if (l->user_aio != aio) {
+		nni_mtx_unlock(&l->mtx);
+		return;
+	}
+	l->user_aio = NULL;
+	nni_mtx_unlock(&l->mtx);
+
+	nni_aio_abort(l->aio, rv);
+	nni_aio_finish_error(aio, rv);
+}
+
+static void
 ipctran_listener_accept(void *arg, nni_aio *aio)
 {
 	ipctran_listener *l = arg;
-	ipctran_pipe *    p;
 	int               rv;
 
 	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
+	nni_mtx_lock(&l->mtx);
+	NNI_ASSERT(l->user_aio == NULL);
 
-	nni_mtx_lock(&l->ep.mtx);
-	if (((rv = ipctran_pipe_init(&p, &l->ep)) != 0) ||
-	    ((rv = nni_aio_schedule(aio, ipctran_pipe_conn_cancel, p)) != 0)) {
-		nni_mtx_unlock(&l->ep.mtx);
+	if ((rv = nni_aio_schedule(aio, ipctran_listener_cancel, l)) != 0) {
+		nni_mtx_unlock(&l->mtx);
 		nni_aio_finish_error(aio, rv);
-		ipctran_pipe_reap(p);
 		return;
 	}
-	p->useraio = aio;
+	l->user_aio = aio;
+
+	nni_ipc_listener_accept(l->listener, l->aio);
+	nni_mtx_unlock(&l->mtx);
+}
+
+static int
+ipctran_listener_set_recvmaxsz(
+    void *arg, const void *data, size_t sz, nni_opt_type t)
+{
+	ipctran_listener *l = arg;
+	size_t            val;
+	int               rv;
+
+	if ((rv = nni_copyin_size(&val, data, sz, 0, NNI_MAXSZ, t)) == 0) {
+		nni_mtx_lock(&l->mtx);
+		l->rcvmax = val;
+		nni_mtx_unlock(&l->mtx);
+	}
+	return (rv);
+}
 
-	nni_ipc_listener_accept(l->listener, p->connaio);
-	nni_mtx_unlock(&l->ep.mtx);
+static int
+ipctran_listener_get_recvmaxsz(
+    void *arg, void *data, size_t *szp, nni_opt_type t)
+{
+	ipctran_listener *l = arg;
+	int               rv;
+	nni_mtx_lock(&l->mtx);
+	rv = nni_copyout_size(l->rcvmax, data, szp, t);
+	nni_mtx_unlock(&l->mtx);
+	return (rv);
 }
 
 static int
@@ -931,9 +987,9 @@ ipctran_listener_get_locaddr(void *arg, void *buf, size_t *szp, nni_opt_type t)
 	ipctran_listener *l = arg;
 	int               rv;
 
-	nni_mtx_lock(&l->ep.mtx);
-	rv = nni_copyout_sockaddr(&l->ep.sa, buf, szp, t);
-	nni_mtx_unlock(&l->ep.mtx);
+	nni_mtx_lock(&l->mtx);
+	rv = nni_copyout_sockaddr(&l->sa, buf, szp, t);
+	nni_mtx_unlock(&l->mtx);
 	return (rv);
 }
 
@@ -954,9 +1010,9 @@ ipctran_listener_set_perms(
 	// Probably we could further limit this -- most systems don't have
 	// meaningful chmod beyond the lower 9 bits.
 	if ((rv = nni_copyin_int(&val, data, sz, 0, 0x7FFFFFFF, t)) == 0) {
-		nni_mtx_lock(&l->ep.mtx);
+		nni_mtx_lock(&l->mtx);
 		rv = nni_ipc_listener_set_permissions(l->listener, val);
-		nni_mtx_unlock(&l->ep.mtx);
+		nni_mtx_unlock(&l->mtx);
 	}
 	return (rv);
 }
@@ -976,10 +1032,10 @@ ipctran_listener_set_sec_desc(
 	int               rv;
 
 	if ((rv = nni_copyin_ptr(&ptr, data, sz, t)) == 0) {
-		nni_mtx_lock(&l->ep.mtx);
+		nni_mtx_lock(&l->mtx);
 		rv =
 		    nni_ipc_listener_set_security_descriptor(l->listener, ptr);
-		nni_mtx_unlock(&l->ep.mtx);
+		nni_mtx_unlock(&l->mtx);
 	}
 	return (rv);
 }
@@ -1029,6 +1085,7 @@ static nni_tran_option ipctran_pipe_options[] = {
 
 static nni_tran_pipe_ops ipctran_pipe_ops = {
 	.p_fini    = ipctran_pipe_fini,
+	.p_start   = ipctran_pipe_start,
 	.p_stop    = ipctran_pipe_stop,
 	.p_send    = ipctran_pipe_send,
 	.p_recv    = ipctran_pipe_recv,
@@ -1041,8 +1098,8 @@ static nni_tran_option ipctran_dialer_options[] = {
 	{
 	    .o_name = NNG_OPT_RECVMAXSZ,
 	    .o_type = NNI_TYPE_SIZE,
-	    .o_get  = ipctran_ep_get_recvmaxsz,
-	    .o_set  = ipctran_ep_set_recvmaxsz,
+	    .o_get  = ipctran_dialer_get_recvmaxsz,
+	    .o_set  = ipctran_dialer_set_recvmaxsz,
 	    .o_chk  = ipctran_check_recvmaxsz,
 	},
 	// terminate list
@@ -1055,8 +1112,8 @@ static nni_tran_option ipctran_listener_options[] = {
 	{
 	    .o_name = NNG_OPT_RECVMAXSZ,
 	    .o_type = NNI_TYPE_SIZE,
-	    .o_get  = ipctran_ep_get_recvmaxsz,
-	    .o_set  = ipctran_ep_set_recvmaxsz,
+	    .o_get  = ipctran_listener_get_recvmaxsz,
+	    .o_set  = ipctran_listener_set_recvmaxsz,
 	    .o_chk  = ipctran_check_recvmaxsz,
 	},
 	{
diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c
index 6cf69972..0d47529e 100644
--- a/src/transport/tcp/tcp.c
+++ b/src/transport/tcp/tcp.c
@@ -20,26 +20,19 @@
 typedef struct tcptran_pipe     tcptran_pipe;
 typedef struct tcptran_dialer   tcptran_dialer;
 typedef struct tcptran_listener tcptran_listener;
-typedef struct tcptran_ep       tcptran_ep;
 
 // tcp_pipe is one end of a TCP connection.
 struct tcptran_pipe {
-	nni_tcp_conn *  conn;
-	uint16_t        peer;
-	uint16_t        proto;
-	size_t          rcvmax;
-	nni_sockaddr    sa;
-	bool            nodelay;
-	bool            keepalive;
-	bool            closed;
-	nni_atomic_flag reaped;
-	nni_list_node   node;
-	nni_reap_item   reap;
-	tcptran_ep *    ep;
+	nni_tcp_conn *conn;
+	uint16_t      peer;
+	uint16_t      proto;
+	size_t        rcvmax;
+	bool          nodelay;
+	bool          keepalive;
 
 	nni_list recvq;
 	nni_list sendq;
-	nni_aio *useraio;
+	nni_aio *user_negaio;
 
 	uint8_t  txlen[sizeof(uint64_t)];
 	uint8_t  rxlen[sizeof(uint64_t)];
@@ -49,48 +42,47 @@ struct tcptran_pipe {
 	size_t   wantrxhead;
 	nni_aio *txaio;
 	nni_aio *rxaio;
-	nni_aio *negoaio;
-	nni_aio *connaio;
-	nni_aio *rslvaio;
+	nni_aio *negaio;
 	nni_msg *rxmsg;
 	nni_mtx  mtx;
 };
 
-struct tcptran_ep {
-	nni_mtx       mtx;
-	nni_list      pipes;
-	nni_url *     url;
-	size_t        rcvmax;
-	bool          nodelay;
-	bool          keepalive;
-	bool          fini;
-	uint16_t      proto;
-	nni_cb        dtor;
-	nni_reap_item reap;
-};
-
 struct tcptran_dialer {
-	tcptran_ep      ep;
 	nni_tcp_dialer *dialer;
-	nng_sockaddr    sa;
+	uint16_t        proto;
 	uint16_t        af;
+	size_t          rcvmax;
+	bool            nodelay;
+	bool            keepalive;
+	bool            resolving;
+	nng_sockaddr    sa;
+	nni_aio *       aio;
+	nni_aio *       user_aio;
+	nni_url *       url;
+	nni_mtx         mtx;
 };
 
 struct tcptran_listener {
-	tcptran_ep        ep;
 	nni_tcp_listener *listener;
+	uint16_t          proto;
+	size_t            rcvmax;
+	bool              nodelay;
+	bool              keepalive;
+	nni_aio *         aio;
+	nni_aio *         user_aio;
+	nni_url *         url;
 	nng_sockaddr      sa;
 	nng_sockaddr      bsa; // bound addr
+	nni_mtx           mtx;
 };
 
 static void tcptran_pipe_send_start(tcptran_pipe *);
 static void tcptran_pipe_recv_start(tcptran_pipe *);
 static void tcptran_pipe_send_cb(void *);
 static void tcptran_pipe_recv_cb(void *);
-static void tcptran_pipe_rslv_cb(void *);
 static void tcptran_pipe_nego_cb(void *);
-static void tcptran_pipe_conn_cb(void *);
-static void tcptran_pipe_reap(tcptran_pipe *);
+static void tcptran_dialer_cb(void *arg);
+static void tcptran_listener_cb(void *arg);
 
 static int
 tcptran_init(void)
@@ -108,15 +100,9 @@ tcptran_pipe_close(void *arg)
 {
 	tcptran_pipe *p = arg;
 
-	nni_mtx_lock(&p->mtx);
-	p->closed = true;
-	nni_mtx_unlock(&p->mtx);
-
 	nni_aio_close(p->rxaio);
 	nni_aio_close(p->txaio);
-	nni_aio_close(p->negoaio);
-	nni_aio_close(p->connaio);
-	nni_aio_close(p->rslvaio);
+	nni_aio_close(p->negaio);
 
 	nni_tcp_conn_close(p->conn);
 }
@@ -128,34 +114,17 @@ tcptran_pipe_stop(void *arg)
 
 	nni_aio_stop(p->rxaio);
 	nni_aio_stop(p->txaio);
-	nni_aio_stop(p->negoaio);
-	nni_aio_stop(p->connaio);
-	nni_aio_stop(p->rslvaio);
+	nni_aio_stop(p->negaio);
 }
 
 static void
 tcptran_pipe_fini(void *arg)
 {
 	tcptran_pipe *p = arg;
-	tcptran_ep *  ep;
 
-	if (p == NULL) {
-		return;
-	}
-	tcptran_pipe_stop(p);
-	if ((ep = p->ep) != NULL) {
-		nni_mtx_lock(&ep->mtx);
-		nni_list_remove(&ep->pipes, p);
-		if (ep->fini && nni_list_empty(&ep->pipes)) {
-			nni_reap(&ep->reap, ep->dtor, ep);
-		}
-		nni_mtx_unlock(&ep->mtx);
-	}
 	nni_aio_fini(p->rxaio);
 	nni_aio_fini(p->txaio);
-	nni_aio_fini(p->negoaio);
-	nni_aio_fini(p->connaio);
-	nni_aio_fini(p->rslvaio);
+	nni_aio_fini(p->negaio);
 	if (p->conn != NULL) {
 		nni_tcp_conn_fini(p->conn);
 	}
@@ -164,16 +133,8 @@ tcptran_pipe_fini(void *arg)
 	NNI_FREE_STRUCT(p);
 }
 
-static void
-tcptran_pipe_reap(tcptran_pipe *p)
-{
-	if (!nni_atomic_flag_test_and_set(&p->reaped)) {
-		nni_reap(&p->reap, tcptran_pipe_fini, p);
-	}
-}
-
 static int
-tcptran_pipe_init(tcptran_pipe **pipep, tcptran_ep *ep)
+tcptran_pipe_init(tcptran_pipe **pipep, void *conn)
 {
 	tcptran_pipe *p;
 	int           rv;
@@ -182,108 +143,45 @@ tcptran_pipe_init(tcptran_pipe **pipep, tcptran_ep *ep)
 		return (NNG_ENOMEM);
 	}
 	nni_mtx_init(&p->mtx);
-	NNI_LIST_NODE_INIT(&p->node);
-	nni_aio_list_init(&p->recvq);
-	nni_aio_list_init(&p->sendq);
-	nni_atomic_flag_reset(&p->reaped);
-
 	if (((rv = nni_aio_init(&p->txaio, tcptran_pipe_send_cb, p)) != 0) ||
 	    ((rv = nni_aio_init(&p->rxaio, tcptran_pipe_recv_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->rslvaio, tcptran_pipe_rslv_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->connaio, tcptran_pipe_conn_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->negoaio, tcptran_pipe_nego_cb, p)) != 0)) {
-		tcptran_pipe_reap(p);
+	    ((rv = nni_aio_init(&p->negaio, tcptran_pipe_nego_cb, p)) != 0)) {
+		tcptran_pipe_fini(p);
 		return (rv);
 	}
-	nni_list_append(&ep->pipes, p);
-	p->ep        = ep;
-	p->proto     = ep->proto;
-	p->rcvmax    = ep->rcvmax;
-	p->keepalive = ep->keepalive;
-	p->nodelay   = ep->nodelay;
+	nni_aio_list_init(&p->recvq);
+	nni_aio_list_init(&p->sendq);
 
-	*pipep = p;
+	p->conn = conn;
+	*pipep  = p;
 	return (0);
 }
 
 static void
-tcptran_pipe_rslv_cb(void *arg)
+tcptran_pipe_nego_cancel(nni_aio *aio, int rv)
 {
-	tcptran_pipe *  p = arg;
-	tcptran_dialer *d;
-	nni_aio *       aio = p->rslvaio;
-	int             rv;
-
-	nni_mtx_lock(&p->ep->mtx);
-	if ((rv = nni_aio_result(aio)) != 0) {
-		nni_aio *uaio;
-		if ((uaio = p->useraio) != NULL) {
-			p->useraio = NULL;
-			nni_aio_finish_error(uaio, rv);
-		}
-		nni_mtx_unlock(&p->ep->mtx);
-		tcptran_pipe_reap(p);
-		return;
-	}
-	d = (void *) p->ep;
-	if (d->dialer != NULL) {
-		nni_tcp_dialer_dial(d->dialer, &p->sa, p->connaio);
-	}
-	nni_mtx_unlock(&p->ep->mtx);
-}
-
-static void
-tcptran_pipe_conn_cb(void *arg)
-{
-	tcptran_pipe *p   = arg;
-	nni_aio *     aio = p->connaio;
-	nni_iov       iov;
-	int           rv;
+	tcptran_pipe *p = nni_aio_get_prov_data(aio);
 
-	nni_mtx_lock(&p->ep->mtx);
-	if ((rv = nni_aio_result(aio)) != 0) {
-		nni_aio *uaio;
-		if ((uaio = p->useraio) != NULL) {
-			p->useraio = NULL;
-			nni_aio_finish_error(uaio, rv);
-		}
-		nni_mtx_unlock(&p->ep->mtx);
-		tcptran_pipe_reap(p);
+	nni_mtx_lock(&p->mtx);
+	if (p->user_negaio != aio) {
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
+	p->user_negaio = NULL;
+	nni_mtx_unlock(&p->mtx);
 
-	p->conn = nni_aio_get_output(aio, 0);
-
-	(void) nni_tcp_conn_set_nodelay(p->conn, p->nodelay);
-	(void) nni_tcp_conn_set_keepalive(p->conn, p->keepalive);
-
-	p->txlen[0] = 0;
-	p->txlen[1] = 'S';
-	p->txlen[2] = 'P';
-	p->txlen[3] = 0;
-	NNI_PUT16(&p->txlen[4], p->proto);
-	NNI_PUT16(&p->txlen[6], 0);
-
-	p->gotrxhead  = 0;
-	p->gottxhead  = 0;
-	p->wantrxhead = 8;
-	p->wanttxhead = 8;
-	iov.iov_len   = 8;
-	iov.iov_buf   = &p->txlen[0];
-	nni_aio_set_iov(p->negoaio, 1, &iov);
-	nni_tcp_conn_send(p->conn, p->negoaio);
-	nni_mtx_unlock(&p->ep->mtx);
+	nni_aio_abort(p->negaio, rv);
+	nni_aio_finish_error(aio, rv);
 }
 
 static void
 tcptran_pipe_nego_cb(void *arg)
 {
 	tcptran_pipe *p   = arg;
-	nni_aio *     aio = p->negoaio;
-	nni_aio *     uaio;
+	nni_aio *     aio = p->negaio;
 	int           rv;
 
-	nni_mtx_lock(&p->ep->mtx);
+	nni_mtx_lock(&p->mtx);
 	if ((rv = nni_aio_result(aio)) != 0) {
 		goto done;
 	}
@@ -302,7 +200,7 @@ tcptran_pipe_nego_cb(void *arg)
 		// send it down...
 		nni_aio_set_iov(aio, 1, &iov);
 		nni_tcp_conn_send(p->conn, aio);
-		nni_mtx_unlock(&p->ep->mtx);
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
 	if (p->gotrxhead < p->wantrxhead) {
@@ -311,7 +209,7 @@ tcptran_pipe_nego_cb(void *arg)
 		iov.iov_buf = &p->rxlen[p->gotrxhead];
 		nni_aio_set_iov(aio, 1, &iov);
 		nni_tcp_conn_recv(p->conn, aio);
-		nni_mtx_unlock(&p->ep->mtx);
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
 	// We have both sent and received the headers.  Lets check the
@@ -326,18 +224,11 @@ tcptran_pipe_nego_cb(void *arg)
 	NNI_GET16(&p->rxlen[4], p->peer);
 
 done:
-	if ((uaio = p->useraio) != NULL) {
-		p->useraio = NULL;
-		if (rv == 0) {
-			nni_aio_set_output(uaio, 0, p);
-			nni_aio_finish(uaio, 0, 0);
-			nni_mtx_unlock(&p->ep->mtx);
-			return;
-		}
-		nni_aio_finish_error(uaio, rv);
+	if ((aio = p->user_negaio) != NULL) {
+		p->user_negaio = NULL;
+		nni_aio_finish(aio, rv, 0);
 	}
-	nni_mtx_unlock(&p->ep->mtx);
-	tcptran_pipe_reap(p);
+	nni_mtx_unlock(&p->mtx);
 }
 
 static void
@@ -351,19 +242,17 @@ tcptran_pipe_send_cb(void *arg)
 	nni_aio *     txaio = p->txaio;
 
 	nni_mtx_lock(&p->mtx);
-	rv = p->closed ? NNG_ECLOSED : nni_aio_result(txaio);
+	aio = nni_list_first(&p->sendq);
 
-	if (rv != 0) {
+	if ((rv = nni_aio_result(txaio)) != 0) {
 		// Intentionally we do not queue up another transfer.
 		// There's an excellent chance that the pipe is no longer
 		// usable, with a partial transfer.
 		// The protocol should see this error, and close the
 		// pipe itself, we hope.
-		while ((aio = nni_list_first(&p->sendq)) != NULL) {
-			nni_aio_list_remove(aio);
-			nni_aio_finish_error(aio, rv);
-		}
+		nni_aio_list_remove(aio);
 		nni_mtx_unlock(&p->mtx);
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
@@ -375,7 +264,6 @@ tcptran_pipe_send_cb(void *arg)
 		return;
 	}
 
-	aio = nni_list_first(&p->sendq);
 	nni_aio_list_remove(aio);
 	tcptran_pipe_send_start(p);
 
@@ -399,14 +287,11 @@ tcptran_pipe_recv_cb(void *arg)
 	nni_aio *     rxaio = p->rxaio;
 
 	nni_mtx_lock(&p->mtx);
+	aio = nni_list_first(&p->recvq);
 
 	if ((rv = nni_aio_result(rxaio)) != 0) {
 		goto recv_error;
 	}
-	if (p->closed) {
-		rv = NNG_ECLOSED;
-		goto recv_error;
-	}
 
 	n = nni_aio_count(rxaio);
 	nni_aio_iov_advance(rxaio, n);
@@ -450,7 +335,6 @@ tcptran_pipe_recv_cb(void *arg)
 	}
 
 	// We read a message completely.  Let the user know the good news.
-	aio = nni_list_first(&p->recvq);
 	nni_aio_list_remove(aio);
 	msg      = p->rxmsg;
 	p->rxmsg = NULL;
@@ -464,10 +348,7 @@ tcptran_pipe_recv_cb(void *arg)
 	return;
 
 recv_error:
-	while ((aio = nni_list_first(&p->recvq)) != NULL) {
-		nni_aio_list_remove(aio);
-		nni_aio_finish_error(aio, rv);
-	}
+	nni_aio_list_remove(aio);
 	msg      = p->rxmsg;
 	p->rxmsg = NULL;
 	// Intentionally, we do not queue up another receive.
@@ -475,6 +356,7 @@ recv_error:
 	nni_mtx_unlock(&p->mtx);
 
 	nni_msg_free(msg);
+	nni_aio_finish_error(aio, rv);
 }
 
 static void
@@ -550,11 +432,6 @@ tcptran_pipe_send(void *arg, nni_aio *aio)
 		return;
 	}
 	nni_mtx_lock(&p->mtx);
-	if (p->closed) {
-		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, NNG_ECLOSED);
-		return;
-	}
 	if ((rv = nni_aio_schedule(aio, tcptran_pipe_send_cancel, p)) != 0) {
 		nni_mtx_unlock(&p->mtx);
 		nni_aio_finish_error(aio, rv);
@@ -597,7 +474,7 @@ tcptran_pipe_recv_start(tcptran_pipe *p)
 	nni_iov  iov;
 	NNI_ASSERT(p->rxmsg == NULL);
 
-	// Schedule a read of the header.
+	// Schedule a read of the IPC header.
 	rxaio       = p->rxaio;
 	iov.iov_buf = p->rxlen;
 	iov.iov_len = sizeof(p->rxlen);
@@ -616,11 +493,6 @@ tcptran_pipe_recv(void *arg, nni_aio *aio)
 		return;
 	}
 	nni_mtx_lock(&p->mtx);
-	if (p->closed) {
-		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, NNG_ECLOSED);
-		return;
-	}
 	if ((rv = nni_aio_schedule(aio, tcptran_pipe_recv_cancel, p)) != 0) {
 		nni_mtx_unlock(&p->mtx);
 		nni_aio_finish_error(aio, rv);
@@ -685,36 +557,54 @@ tcptran_pipe_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t)
 }
 
 static void
-tcptran_dialer_fini(void *arg)
+tcptran_pipe_start(void *arg, nni_aio *aio)
 {
-	tcptran_dialer *d = arg;
+	tcptran_pipe *p = arg;
+	nni_aio *     negaio;
+	nni_iov       iov;
+	int           rv;
 
-	nni_mtx_lock(&d->ep.mtx);
-	if (d->dialer != NULL) {
-		nni_tcp_dialer_fini(d->dialer);
-		d->dialer = NULL;
+	if (nni_aio_begin(aio) != 0) {
+		return;
 	}
-	d->ep.fini = true;
-	if (!nni_list_empty(&d->ep.pipes)) {
-		nni_mtx_unlock(&d->ep.mtx);
+	nni_mtx_lock(&p->mtx);
+	if ((rv = nni_aio_schedule(aio, tcptran_pipe_nego_cancel, p)) != 0) {
+		nni_mtx_unlock(&p->mtx);
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
-	nni_mtx_unlock(&d->ep.mtx);
-	nni_mtx_fini(&d->ep.mtx);
-	NNI_FREE_STRUCT(d);
+	p->txlen[0] = 0;
+	p->txlen[1] = 'S';
+	p->txlen[2] = 'P';
+	p->txlen[3] = 0;
+	NNI_PUT16(&p->txlen[4], p->proto);
+	NNI_PUT16(&p->txlen[6], 0);
+
+	p->user_negaio = aio;
+	p->gotrxhead   = 0;
+	p->gottxhead   = 0;
+	p->wantrxhead  = 8;
+	p->wanttxhead  = 8;
+	negaio         = p->negaio;
+	iov.iov_len    = 8;
+	iov.iov_buf    = &p->txlen[0];
+	nni_aio_set_iov(negaio, 1, &iov);
+	nni_tcp_conn_send(p->conn, negaio);
+	nni_mtx_unlock(&p->mtx);
 }
 
 static void
-tcptran_ep_close(tcptran_ep *ep)
+tcptran_dialer_fini(void *arg)
 {
-	tcptran_pipe *p;
-	nni_mtx_lock(&ep->mtx);
-	NNI_LIST_FOREACH (&ep->pipes, p) {
-		nni_aio_close(p->negoaio);
-		nni_aio_close(p->connaio);
-		nni_aio_close(p->rslvaio);
+	tcptran_dialer *d = arg;
+
+	nni_aio_stop(d->aio);
+	if (d->dialer != NULL) {
+		nni_tcp_dialer_fini(d->dialer);
 	}
-	nni_mtx_unlock(&ep->mtx);
+	nni_aio_fini(d->aio);
+	nni_mtx_fini(&d->mtx);
+	NNI_FREE_STRUCT(d);
 }
 
 static void
@@ -722,7 +612,7 @@ tcptran_dialer_close(void *arg)
 {
 	tcptran_dialer *d = arg;
 
-	tcptran_ep_close(&d->ep);
+	nni_aio_close(d->aio);
 	nni_tcp_dialer_close(d->dialer);
 }
 
@@ -756,72 +646,127 @@ tcptran_dialer_init(void **dp, nni_url *url, nni_sock *sock)
 	if ((d = NNI_ALLOC_STRUCT(d)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	nni_mtx_init(&d->ep.mtx);
-	NNI_LIST_INIT(&d->ep.pipes, tcptran_pipe, node);
-	d->ep.dtor      = tcptran_dialer_fini;
-	d->ep.nodelay   = true;
-	d->ep.keepalive = false;
-	d->ep.proto     = nni_sock_proto_id(sock);
-	d->ep.url       = url;
-	d->af           = af;
+	nni_mtx_init(&d->mtx);
 
-	if ((rv = nni_tcp_dialer_init(&d->dialer)) != 0) {
+	if (((rv = nni_tcp_dialer_init(&d->dialer)) != 0) ||
+	    ((rv = nni_aio_init(&d->aio, tcptran_dialer_cb, d)) != 0)) {
 		tcptran_dialer_fini(d);
 		return (rv);
 	}
 
+	d->url       = url;
+	d->proto     = nni_sock_proto_id(sock);
+	d->nodelay   = true;
+	d->keepalive = false;
+	d->af        = af;
+
 	*dp = d;
 	return (0);
 }
 
 static void
-tcptran_pipe_conn_cancel(nni_aio *aio, int rv)
+tcptran_dialer_cb(void *arg)
 {
-	tcptran_pipe *p = nni_aio_get_prov_data(aio);
+	tcptran_dialer *d = arg;
+	tcptran_pipe *  p;
+	nni_tcp_conn *  conn;
+	nni_aio *       aio;
+	int             rv;
+
+	nni_mtx_lock(&d->mtx);
+	aio = d->user_aio;
+	rv  = nni_aio_result(d->aio);
 
-	nni_mtx_lock(&p->ep->mtx);
-	if (p->useraio != aio) {
-		nni_mtx_unlock(&p->ep->mtx);
+	if (aio == NULL) {
+		nni_mtx_unlock(&d->mtx);
+		if ((rv == 0) && !d->resolving) {
+			conn = nni_aio_get_output(d->aio, 0);
+			nni_tcp_conn_fini(conn);
+		}
 		return;
 	}
 
-	// Close the underlying AIOs.  This will abort the operation.
-	// The pipe is removed from pending list at completion callback.
-	p->useraio = NULL;
-	nni_aio_close(p->negoaio);
-	nni_aio_close(p->connaio);
-	nni_aio_close(p->rslvaio);
-	nni_mtx_unlock(&p->ep->mtx);
+	if (rv != 0) {
+		d->user_aio = NULL;
+		nni_mtx_unlock(&d->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
 
+	if (d->resolving) {
+		// Name resolution complete.  Now go to next step.
+		d->resolving = false;
+		nni_tcp_dialer_dial(d->dialer, &d->sa, d->aio);
+		nni_mtx_unlock(&d->mtx);
+		return;
+	}
+
+	d->user_aio = NULL;
+	conn        = nni_aio_get_output(d->aio, 0);
+	NNI_ASSERT(conn != NULL);
+	if ((rv = tcptran_pipe_init(&p, conn)) != 0) {
+		nni_mtx_unlock(&d->mtx);
+		nni_tcp_conn_fini(conn);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	p->proto     = d->proto;
+	p->rcvmax    = d->rcvmax;
+	p->nodelay   = d->nodelay;
+	p->keepalive = d->keepalive;
+	nni_mtx_unlock(&d->mtx);
+
+	(void) nni_tcp_conn_set_nodelay(conn, p->nodelay);
+	(void) nni_tcp_conn_set_keepalive(conn, p->keepalive);
+
+	nni_aio_set_output(aio, 0, p);
+	nni_aio_finish(aio, 0, 0);
+}
+
+static void
+tcptran_dialer_cancel(nni_aio *aio, int rv)
+{
+	tcptran_dialer *d = nni_aio_get_prov_data(aio);
+
+	nni_mtx_lock(&d->mtx);
+	if (d->user_aio != aio) {
+		nni_mtx_unlock(&d->mtx);
+		return;
+	}
+	d->user_aio = NULL;
+	nni_mtx_unlock(&d->mtx);
+
+	nni_aio_abort(d->aio, rv);
 	nni_aio_finish_error(aio, rv);
 }
 
 static void
 tcptran_dialer_connect(void *arg, nni_aio *aio)
 {
-	tcptran_dialer *d  = arg;
-	tcptran_ep *    ep = &d->ep;
-	tcptran_pipe *  p  = NULL;
+	tcptran_dialer *d = arg;
 	int             rv;
 
 	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
-	nni_mtx_lock(&ep->mtx);
-	if (((rv = tcptran_pipe_init(&p, ep)) != 0) ||
-	    ((rv = nni_aio_schedule(aio, tcptran_pipe_conn_cancel, p)) != 0)) {
-		nni_mtx_unlock(&ep->mtx);
+	nni_mtx_lock(&d->mtx);
+	NNI_ASSERT(d->user_aio == NULL);
+
+	if ((rv = nni_aio_schedule(aio, tcptran_dialer_cancel, d)) != 0) {
+		nni_mtx_unlock(&d->mtx);
 		nni_aio_finish_error(aio, rv);
-		tcptran_pipe_reap(p);
 		return;
 	}
-	p->useraio = aio;
+	d->user_aio = aio;
+
+	d->resolving = true;
 
-	// Start the name resolution as first step.
-	nni_aio_set_input(p->rslvaio, 0, &p->sa);
-	nni_tcp_resolv(
-	    ep->url->u_hostname, ep->url->u_port, d->af, 0, p->rslvaio);
-	nni_mtx_unlock(&ep->mtx);
+	// Start the name resolution.  Callback will see resolving, and then
+	// switch to doing actual connect.
+	nni_aio_set_input(d->aio, 0, &d->sa);
+	nni_tcp_resolv(d->url->u_hostname, d->url->u_port, d->af, 0, d->aio);
+	nni_mtx_unlock(&d->mtx);
 }
 
 static int
@@ -829,81 +774,79 @@ tcptran_dialer_get_url(void *arg, void *v, size_t *szp, nni_opt_type t)
 {
 	tcptran_dialer *d = arg;
 
-	return (nni_copyout_str(d->ep.url->u_rawurl, v, szp, t));
+	return (nni_copyout_str(d->url->u_rawurl, v, szp, t));
 }
 
 static int
-tcptran_ep_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t)
+tcptran_dialer_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t)
 {
-	tcptran_ep *ep = arg;
-	int         rv;
+	tcptran_dialer *d = arg;
+	int             rv;
 
-	nni_mtx_lock(&ep->mtx);
-	rv = nni_copyout_size(ep->rcvmax, v, szp, t);
-	nni_mtx_unlock(&ep->mtx);
+	nni_mtx_lock(&d->mtx);
+	rv = nni_copyout_size(d->rcvmax, v, szp, t);
+	nni_mtx_unlock(&d->mtx);
 	return (rv);
 }
 
 static int
-tcptran_ep_set_recvmaxsz(void *arg, const void *v, size_t sz, nni_opt_type t)
+tcptran_dialer_set_recvmaxsz(
+    void *arg, const void *v, size_t sz, nni_opt_type t)
 {
-	tcptran_ep *ep = arg;
-	size_t      val;
-	int         rv;
+	tcptran_dialer *d = arg;
+	size_t          val;
+	int             rv;
 	if ((rv = nni_copyin_size(&val, v, sz, 0, NNI_MAXSZ, t)) == 0) {
-		nni_mtx_lock(&ep->mtx);
-		ep->rcvmax = val;
-		nni_mtx_unlock(&ep->mtx);
+		nni_mtx_lock(&d->mtx);
+		d->rcvmax = val;
+		nni_mtx_unlock(&d->mtx);
 	}
 	return (rv);
 }
 
 static int
-tcptran_ep_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t)
+tcptran_dialer_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t)
 {
-	tcptran_ep *ep = arg;
-	int         rv;
-	nni_mtx_lock(&ep->mtx);
-	rv = nni_copyout_bool(ep->nodelay, v, szp, t);
-	nni_mtx_unlock(&ep->mtx);
+	tcptran_dialer *d = arg;
+	int             rv;
+	nni_mtx_lock(&d->mtx);
+	rv = nni_copyout_bool(d->nodelay, v, szp, t);
+	nni_mtx_unlock(&d->mtx);
 	return (rv);
 }
 
 static int
-tcptran_ep_set_nodelay(void *arg, const void *v, size_t sz, nni_opt_type t)
+tcptran_dialer_set_nodelay(void *arg, const void *v, size_t sz, nni_opt_type t)
 {
-	tcptran_ep *ep = arg;
-	bool        val;
-	int         rv;
+	tcptran_dialer *d = arg;
+	bool            val;
+	int             rv;
 	if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) {
-		nni_mtx_lock(&ep->mtx);
-		ep->nodelay = val;
-		nni_mtx_unlock(&ep->mtx);
+		nni_mtx_lock(&d->mtx);
+		d->nodelay = val;
+		nni_mtx_unlock(&d->mtx);
 	}
 	return (rv);
 }
 
 static int
-tcptran_ep_get_keepalive(void *arg, void *v, size_t *szp, nni_opt_type t)
+tcptran_dialer_get_keepalive(void *arg, void *v, size_t *szp, nni_opt_type t)
 {
-	tcptran_ep *ep = arg;
-	int         rv;
-	nni_mtx_lock(&ep->mtx);
-	rv = nni_copyout_bool(ep->keepalive, v, szp, t);
-	nni_mtx_unlock(&ep->mtx);
-	return (rv);
+	tcptran_dialer *d = arg;
+	return (nni_copyout_bool(d->keepalive, v, szp, t));
 }
 
 static int
-tcptran_ep_set_keepalive(void *arg, const void *v, size_t sz, nni_opt_type t)
+tcptran_dialer_set_keepalive(
+    void *arg, const void *v, size_t sz, nni_opt_type t)
 {
-	tcptran_ep *ep = arg;
-	bool        val;
-	int         rv;
+	tcptran_dialer *d = arg;
+	bool            val;
+	int             rv;
 	if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) {
-		nni_mtx_lock(&ep->mtx);
-		ep->keepalive = val;
-		nni_mtx_unlock(&ep->mtx);
+		nni_mtx_lock(&d->mtx);
+		d->keepalive = val;
+		nni_mtx_unlock(&d->mtx);
 	}
 	return (rv);
 }
@@ -913,18 +856,12 @@ tcptran_listener_fini(void *arg)
 {
 	tcptran_listener *l = arg;
 
-	nni_mtx_lock(&l->ep.mtx);
+	nni_aio_stop(l->aio);
 	if (l->listener != NULL) {
 		nni_tcp_listener_fini(l->listener);
-		l->listener = NULL;
 	}
-	l->ep.fini = true;
-	if (!nni_list_empty(&l->ep.pipes)) {
-		nni_mtx_unlock(&l->ep.mtx);
-		return;
-	}
-	nni_mtx_unlock(&l->ep.mtx);
-	nni_mtx_fini(&l->ep.mtx);
+	nni_aio_fini(l->aio);
+	nni_mtx_fini(&l->mtx);
 	NNI_FREE_STRUCT(l);
 }
 
@@ -959,13 +896,8 @@ tcptran_listener_init(void **lp, nni_url *url, nni_sock *sock)
 	if ((l = NNI_ALLOC_STRUCT(l)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	nni_mtx_init(&l->ep.mtx);
-	NNI_LIST_INIT(&l->ep.pipes, tcptran_pipe, node);
-	l->ep.dtor      = tcptran_listener_fini;
-	l->ep.url       = url;
-	l->ep.proto     = nni_sock_proto_id(sock);
-	l->ep.nodelay   = true;
-	l->ep.keepalive = false;
+	nni_mtx_init(&l->mtx);
+	l->url = url;
 
 	if (strlen(url->u_hostname) == 0) {
 		host = NULL;
@@ -991,12 +923,21 @@ tcptran_listener_init(void **lp, nni_url *url, nni_sock *sock)
 	rv = nni_aio_result(aio);
 	nni_aio_fini(aio);
 
-	if ((rv != 0) || ((rv = nni_tcp_listener_init(&l->listener)) != 0)) {
+	if (rv != 0) {
 		tcptran_listener_fini(l);
 		return (rv);
 	}
 
-	l->bsa = l->sa;
+	if (((rv = nni_tcp_listener_init(&l->listener)) != 0) ||
+	    ((rv = nni_aio_init(&l->aio, tcptran_listener_cb, l)) != 0)) {
+		tcptran_listener_fini(l);
+		return (rv);
+	}
+
+	l->proto     = nni_sock_proto_id(sock);
+	l->nodelay   = true;
+	l->keepalive = false;
+	l->bsa       = l->sa;
 
 	*lp = l;
 	return (0);
@@ -1007,7 +948,7 @@ tcptran_listener_close(void *arg)
 {
 	tcptran_listener *l = arg;
 
-	tcptran_ep_close(&l->ep);
+	nni_aio_close(l->aio);
 	nni_tcp_listener_close(l->listener);
 }
 
@@ -1017,36 +958,171 @@ tcptran_listener_bind(void *arg)
 	tcptran_listener *l = arg;
 	int               rv;
 
-	nni_mtx_lock(&l->ep.mtx);
+	nni_mtx_lock(&l->mtx);
 	l->bsa = l->sa;
 	rv     = nni_tcp_listener_listen(l->listener, &l->bsa);
-	nni_mtx_unlock(&l->ep.mtx);
+	nni_mtx_unlock(&l->mtx);
 
 	return (rv);
 }
 
 static void
-tcptran_listener_accept(void *arg, nni_aio *aio)
+tcptran_listener_cb(void *arg)
 {
 	tcptran_listener *l = arg;
+	nni_aio *         aio;
+	int               rv;
 	tcptran_pipe *    p = NULL;
+	nni_tcp_conn *    conn;
+
+	nni_mtx_lock(&l->mtx);
+	rv          = nni_aio_result(l->aio);
+	aio         = l->user_aio;
+	l->user_aio = NULL;
+
+	if (aio == NULL) {
+		nni_mtx_unlock(&l->mtx);
+		if (rv == 0) {
+			conn = nni_aio_get_output(l->aio, 0);
+			nni_tcp_conn_fini(conn);
+		}
+		return;
+	}
+
+	if (rv != 0) {
+		nni_mtx_unlock(&l->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	conn = nni_aio_get_output(l->aio, 0);
+
+	NNI_ASSERT(conn != NULL);
+	if ((rv = tcptran_pipe_init(&p, conn)) != 0) {
+		nni_mtx_unlock(&l->mtx);
+		nni_tcp_conn_fini(conn);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	p->proto     = l->proto;
+	p->rcvmax    = l->rcvmax;
+	p->nodelay   = l->nodelay;
+	p->keepalive = l->keepalive;
+	nni_mtx_unlock(&l->mtx);
+
+	(void) nni_tcp_conn_set_nodelay(conn, p->nodelay);
+	(void) nni_tcp_conn_set_keepalive(conn, p->keepalive);
+
+	nni_aio_set_output(aio, 0, p);
+	nni_aio_finish(aio, 0, 0);
+}
+
+static void
+tcptran_listener_cancel(nni_aio *aio, int rv)
+{
+	tcptran_listener *l = nni_aio_get_prov_data(aio);
+
+	nni_mtx_lock(&l->mtx);
+	if (l->user_aio != aio) {
+		nni_mtx_unlock(&l->mtx);
+		return;
+	}
+	l->user_aio = NULL;
+	nni_mtx_unlock(&l->mtx);
+
+	nni_aio_abort(l->aio, rv);
+	nni_aio_finish_error(aio, rv);
+}
+
+static void
+tcptran_listener_accept(void *arg, nni_aio *aio)
+{
+	tcptran_listener *l = arg;
 	int               rv;
 
 	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
-	nni_mtx_lock(&l->ep.mtx);
-	if (((rv = tcptran_pipe_init(&p, &l->ep)) != 0) ||
-	    ((rv = nni_aio_schedule(aio, tcptran_pipe_conn_cancel, p)) != 0)) {
-		nni_mtx_unlock(&l->ep.mtx);
+	nni_mtx_lock(&l->mtx);
+	NNI_ASSERT(l->user_aio == NULL);
+
+	if ((rv = nni_aio_schedule(aio, tcptran_listener_cancel, l)) != 0) {
+		nni_mtx_unlock(&l->mtx);
 		nni_aio_finish_error(aio, rv);
-		tcptran_pipe_reap(p);
 		return;
 	}
-	p->useraio = aio;
+	l->user_aio = aio;
+
+	nni_tcp_listener_accept(l->listener, l->aio);
+	nni_mtx_unlock(&l->mtx);
+}
+
+static int
+tcptran_listener_set_nodelay(
+    void *arg, const void *v, size_t sz, nni_opt_type t)
+{
+	tcptran_listener *l = arg;
+	bool              val;
+	int               rv;
+	if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) {
+		nni_mtx_lock(&l->mtx);
+		l->nodelay = val;
+		nni_mtx_unlock(&l->mtx);
+	}
+	return (rv);
+}
+
+static int
+tcptran_listener_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t)
+{
+	tcptran_listener *l = arg;
+	int               rv;
+	nni_mtx_lock(&l->mtx);
+	rv = nni_copyout_bool(l->nodelay, v, szp, t);
+	nni_mtx_unlock(&l->mtx);
+	return (rv);
+}
+
+static int
+tcptran_listener_set_recvmaxsz(
+    void *arg, const void *v, size_t sz, nni_opt_type t)
+{
+	tcptran_listener *l = arg;
+	size_t            val;
+	int               rv;
+	if ((rv = nni_copyin_size(&val, v, sz, 0, NNI_MAXSZ, t)) == 0) {
+		nni_mtx_lock(&l->mtx);
+		l->rcvmax = val;
+		nni_mtx_unlock(&l->mtx);
+	}
+	return (rv);
+}
 
-	nni_tcp_listener_accept(l->listener, p->connaio);
-	nni_mtx_unlock(&l->ep.mtx);
+static int
+tcptran_listener_set_keepalive(
+    void *arg, const void *v, size_t sz, nni_opt_type t)
+{
+	tcptran_listener *l = arg;
+	bool              val;
+	int               rv;
+	if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) {
+		nni_mtx_lock(&l->mtx);
+		l->keepalive = val;
+		nni_mtx_unlock(&l->mtx);
+	}
+	return (rv);
+}
+
+static int
+tcptran_listener_get_keepalive(void *arg, void *v, size_t *szp, nni_opt_type t)
+{
+	tcptran_listener *l = arg;
+	int               rv;
+	nni_mtx_lock(&l->mtx);
+	rv = nni_copyout_bool(l->keepalive, v, szp, t);
+	nni_mtx_unlock(&l->mtx);
+	return (rv);
 }
 
 static int
@@ -1063,14 +1139,26 @@ tcptran_listener_get_url(void *arg, void *v, size_t *szp, nni_opt_type t)
 }
 
 static int
+tcptran_listener_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t)
+{
+	tcptran_listener *l = arg;
+	int               rv;
+
+	nni_mtx_lock(&l->mtx);
+	rv = nni_copyout_size(l->rcvmax, v, szp, t);
+	nni_mtx_unlock(&l->mtx);
+	return (rv);
+}
+
+static int
 tcptran_listener_get_locaddr(void *arg, void *buf, size_t *szp, nni_opt_type t)
 {
 	tcptran_listener *l = arg;
 	int               rv;
 
-	nni_mtx_lock(&l->ep.mtx);
+	nni_mtx_lock(&l->mtx);
 	rv = nni_copyout_sockaddr(&l->bsa, buf, szp, t);
-	nni_mtx_unlock(&l->ep.mtx);
+	nni_mtx_unlock(&l->mtx);
 	return (rv);
 }
 
@@ -1115,6 +1203,7 @@ static nni_tran_option tcptran_pipe_options[] = {
 
 static nni_tran_pipe_ops tcptran_pipe_ops = {
 	.p_fini    = tcptran_pipe_fini,
+	.p_start   = tcptran_pipe_start,
 	.p_stop    = tcptran_pipe_stop,
 	.p_send    = tcptran_pipe_send,
 	.p_recv    = tcptran_pipe_recv,
@@ -1127,8 +1216,8 @@ static nni_tran_option tcptran_dialer_options[] = {
 	{
 	    .o_name = NNG_OPT_RECVMAXSZ,
 	    .o_type = NNI_TYPE_SIZE,
-	    .o_get  = tcptran_ep_get_recvmaxsz,
-	    .o_set  = tcptran_ep_set_recvmaxsz,
+	    .o_get  = tcptran_dialer_get_recvmaxsz,
+	    .o_set  = tcptran_dialer_set_recvmaxsz,
 	    .o_chk  = tcptran_check_recvmaxsz,
 	},
 	{
@@ -1139,15 +1228,15 @@ static nni_tran_option tcptran_dialer_options[] = {
 	{
 	    .o_name = NNG_OPT_TCP_NODELAY,
 	    .o_type = NNI_TYPE_BOOL,
-	    .o_get  = tcptran_ep_get_nodelay,
-	    .o_set  = tcptran_ep_set_nodelay,
+	    .o_get  = tcptran_dialer_get_nodelay,
+	    .o_set  = tcptran_dialer_set_nodelay,
 	    .o_chk  = tcptran_check_bool,
 	},
 	{
 	    .o_name = NNG_OPT_TCP_KEEPALIVE,
 	    .o_type = NNI_TYPE_BOOL,
-	    .o_get  = tcptran_ep_get_keepalive,
-	    .o_set  = tcptran_ep_set_keepalive,
+	    .o_get  = tcptran_dialer_get_keepalive,
+	    .o_set  = tcptran_dialer_set_keepalive,
 	    .o_chk  = tcptran_check_bool,
 	},
 	// terminate list
@@ -1160,8 +1249,8 @@ static nni_tran_option tcptran_listener_options[] = {
 	{
 	    .o_name = NNG_OPT_RECVMAXSZ,
 	    .o_type = NNI_TYPE_SIZE,
-	    .o_get  = tcptran_ep_get_recvmaxsz,
-	    .o_set  = tcptran_ep_set_recvmaxsz,
+	    .o_get  = tcptran_listener_get_recvmaxsz,
+	    .o_set  = tcptran_listener_set_recvmaxsz,
 	    .o_chk  = tcptran_check_recvmaxsz,
 	},
 	{
@@ -1177,15 +1266,15 @@ static nni_tran_option tcptran_listener_options[] = {
 	{
 	    .o_name = NNG_OPT_TCP_NODELAY,
 	    .o_type = NNI_TYPE_BOOL,
-	    .o_get  = tcptran_ep_get_nodelay,
-	    .o_set  = tcptran_ep_set_nodelay,
+	    .o_get  = tcptran_listener_get_nodelay,
+	    .o_set  = tcptran_listener_set_nodelay,
 	    .o_chk  = tcptran_check_bool,
 	},
 	{
 	    .o_name = NNG_OPT_TCP_KEEPALIVE,
 	    .o_type = NNI_TYPE_BOOL,
-	    .o_get  = tcptran_ep_get_keepalive,
-	    .o_set  = tcptran_ep_set_keepalive,
+	    .o_get  = tcptran_listener_get_keepalive,
+	    .o_set  = tcptran_listener_set_keepalive,
 	    .o_chk  = tcptran_check_bool,
 	},
 	// terminate list
diff --git a/src/transport/tls/tls.c b/src/transport/tls/tls.c
index 88e201b9..e6701f5f 100644
--- a/src/transport/tls/tls.c
+++ b/src/transport/tls/tls.c
@@ -29,23 +29,16 @@ typedef struct tlstran_pipe     tlstran_pipe;
 
 // tlstran_pipe is one end of a TLS connection.
 struct tlstran_pipe {
-	nni_tls *       tls;
-	uint16_t        peer;
-	uint16_t        proto;
-	size_t          rcvmax;
-	bool            nodelay;
-	bool            keepalive;
-	nni_atomic_flag reaped;
-	nni_list_node   node; // When pending, we are on ep list.
-	tlstran_ep *    ep;
-	nni_tcp_dialer *dialer; // Client side.
-	nni_sockaddr    sa;
-	nni_reap_item   reap;
-	bool            closed;
+	nni_tls *tls;
+	uint16_t peer;
+	uint16_t proto;
+	size_t   rcvmax;
+	bool     nodelay;
+	bool     keepalive;
 
 	nni_list sendq;
 	nni_list recvq;
-	nni_aio *useraio;
+	nni_aio *user_negaio;
 
 	uint8_t  txlen[sizeof(uint64_t)];
 	uint8_t  rxlen[sizeof(uint64_t)];
@@ -55,9 +48,7 @@ struct tlstran_pipe {
 	size_t   wantrxhead;
 	nni_aio *txaio;
 	nni_aio *rxaio;
-	nni_aio *negoaio;
-	nni_aio *connaio;
-	nni_aio *rslvaio;
+	nni_aio *negaio;
 	nni_msg *rxmsg;
 	nni_mtx  mtx;
 };
@@ -68,25 +59,27 @@ struct tlstran_ep {
 	size_t          rcvmax;
 	bool            nodelay;
 	bool            keepalive;
-	bool            fini;
-	nni_reap_item   reap;
-	nni_cb          dtor;
 	int             authmode;
 	nng_tls_config *cfg;
 	nni_url *       url;
 	nni_mtx         mtx;
-	nni_list        pipes;
 };
 
 struct tlstran_dialer {
 	tlstran_ep      ep; // must be first
 	nni_tcp_dialer *dialer;
 	uint16_t        af;
+	nni_aio *       aio;
+	nni_aio *       user_aio;
+	bool            resolving;
+	nng_sockaddr    sa;
 };
 
 struct tlstran_listener {
 	tlstran_ep        ep; // must be first
 	nni_tcp_listener *listener;
+	nni_aio *         aio;
+	nni_aio *         user_aio;
 	nng_sockaddr      sa;
 	nng_sockaddr      bsa; // bound addr
 };
@@ -95,10 +88,9 @@ static void tlstran_pipe_send_start(tlstran_pipe *);
 static void tlstran_pipe_recv_start(tlstran_pipe *);
 static void tlstran_pipe_send_cb(void *);
 static void tlstran_pipe_recv_cb(void *);
-static void tlstran_pipe_rslv_cb(void *);
-static void tlstran_pipe_conn_cb(void *);
 static void tlstran_pipe_nego_cb(void *);
-static void tlstran_pipe_reap(tlstran_pipe *);
+static void tlstran_dialer_cb(void *);
+static void tlstran_listener_cb(void *);
 
 static int
 tlstran_init(void)
@@ -116,15 +108,9 @@ tlstran_pipe_close(void *arg)
 {
 	tlstran_pipe *p = arg;
 
-	nni_mtx_lock(&p->mtx);
-	p->closed = true;
-	nni_mtx_unlock(&p->mtx);
-
 	nni_aio_close(p->rxaio);
 	nni_aio_close(p->txaio);
-	nni_aio_close(p->negoaio);
-	nni_aio_close(p->connaio);
-	nni_aio_close(p->rslvaio);
+	nni_aio_close(p->negaio);
 
 	nni_tls_close(p->tls);
 }
@@ -136,53 +122,26 @@ tlstran_pipe_stop(void *arg)
 
 	nni_aio_stop(p->rxaio);
 	nni_aio_stop(p->txaio);
-	nni_aio_stop(p->negoaio);
-	nni_aio_stop(p->connaio);
-	nni_aio_stop(p->rslvaio);
+	nni_aio_stop(p->negaio);
 }
 
 static void
 tlstran_pipe_fini(void *arg)
 {
 	tlstran_pipe *p = arg;
-	tlstran_ep *  ep;
-
-	if (p == NULL) {
-		return;
-	}
-	tlstran_pipe_stop(p);
-	if ((ep = p->ep) != NULL) {
-		nni_mtx_lock(&ep->mtx);
-		nni_list_remove(&ep->pipes, p);
-		if (ep->fini && nni_list_empty(&ep->pipes)) {
-			nni_reap(&ep->reap, ep->dtor, ep);
-		}
-		nni_mtx_unlock(&ep->mtx);
-	}
 
 	nni_aio_fini(p->rxaio);
 	nni_aio_fini(p->txaio);
-	nni_aio_fini(p->negoaio);
-	nni_aio_fini(p->connaio);
-	nni_aio_fini(p->rslvaio);
+	nni_aio_fini(p->negaio);
 	if (p->tls != NULL) {
 		nni_tls_fini(p->tls);
 	}
 	nni_msg_free(p->rxmsg);
-	nni_mtx_fini(&p->mtx);
 	NNI_FREE_STRUCT(p);
 }
 
-static void
-tlstran_pipe_reap(tlstran_pipe *p)
-{
-	if (!nni_atomic_flag_test_and_set(&p->reaped)) {
-		nni_reap(&p->reap, tlstran_pipe_fini, p);
-	}
-}
-
 static int
-tlstran_pipe_init(tlstran_pipe **pipep, tlstran_ep *ep)
+tlstran_pipe_init(tlstran_pipe **pipep, nni_tls *tls)
 {
 	tlstran_pipe *p;
 	int           rv;
@@ -191,110 +150,46 @@ tlstran_pipe_init(tlstran_pipe **pipep, tlstran_ep *ep)
 		return (NNG_ENOMEM);
 	}
 	nni_mtx_init(&p->mtx);
-	nni_aio_list_init(&p->recvq);
-	nni_aio_list_init(&p->sendq);
-	nni_atomic_flag_reset(&p->reaped);
-	nni_list_append(&ep->pipes, p);
-
-	p->proto     = ep->proto;
-	p->rcvmax    = ep->rcvmax;
-	p->nodelay   = ep->nodelay;
-	p->keepalive = ep->keepalive;
-	p->ep        = ep;
 
 	if (((rv = nni_aio_init(&p->txaio, tlstran_pipe_send_cb, p)) != 0) ||
 	    ((rv = nni_aio_init(&p->rxaio, tlstran_pipe_recv_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->rslvaio, tlstran_pipe_rslv_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->connaio, tlstran_pipe_conn_cb, p)) != 0) ||
-	    ((rv = nni_aio_init(&p->negoaio, tlstran_pipe_nego_cb, p)) != 0)) {
-		tlstran_pipe_reap(p);
+	    ((rv = nni_aio_init(&p->negaio, tlstran_pipe_nego_cb, p)) != 0)) {
+		tlstran_pipe_fini(p);
 		return (rv);
 	}
+	nni_aio_list_init(&p->recvq);
+	nni_aio_list_init(&p->sendq);
 
+	p->tls = tls;
 	*pipep = p;
 	return (0);
 }
 
 static void
-tlstran_pipe_rslv_cb(void *arg)
+tlstran_pipe_cancel_nego(nni_aio *aio, int rv)
 {
-	tlstran_pipe *  p   = arg;
-	tlstran_dialer *d   = (void *) p->ep;
-	nni_aio *       aio = p->rslvaio;
-	int             rv;
-
-	nni_mtx_lock(&p->ep->mtx);
-	if ((rv = nni_aio_result(aio)) != 0) {
-		nni_aio *uaio;
-		if ((uaio = p->useraio) != NULL) {
-			p->useraio = NULL;
-			nni_aio_finish_error(uaio, rv);
-		}
-		nni_mtx_unlock(&p->ep->mtx);
-		tlstran_pipe_reap(p);
-		return;
-	}
-	if (d->dialer != NULL) {
-		nni_tcp_dialer_dial(d->dialer, &p->sa, p->connaio);
-	}
-	nni_mtx_unlock(&p->ep->mtx);
-}
-
-static void
-tlstran_pipe_conn_cb(void *arg)
-{
-	tlstran_pipe *p   = arg;
-	nni_aio *     aio = p->connaio;
-	nni_iov       iov;
-	int           rv;
-
-	nni_mtx_lock(&p->ep->mtx);
-
-	if ((rv = nni_aio_result(aio)) == 0) {
-		nni_tcp_conn *tcp = nni_aio_get_output(aio, 0);
-		if ((rv = nni_tls_init(&p->tls, p->ep->cfg, tcp)) != 0) {
-			nni_tcp_conn_fini(tcp);
-		}
-	}
+	tlstran_pipe *p = nni_aio_get_prov_data(aio);
 
-	if (rv != 0) {
-		nni_aio *uaio;
-		if ((uaio = p->useraio) != NULL) {
-			p->useraio = NULL;
-			nni_aio_finish_error(uaio, rv);
-		}
-		nni_mtx_unlock(&p->ep->mtx);
-		tlstran_pipe_reap(p);
+	nni_mtx_lock(&p->mtx);
+	if (p->user_negaio != aio) {
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
+	p->user_negaio = NULL;
+	nni_mtx_unlock(&p->mtx);
 
-	p->txlen[0] = 0;
-	p->txlen[1] = 'S';
-	p->txlen[2] = 'P';
-	p->txlen[3] = 0;
-	NNI_PUT16(&p->txlen[4], p->proto);
-	NNI_PUT16(&p->txlen[6], 0);
-
-	p->gotrxhead  = 0;
-	p->gottxhead  = 0;
-	p->wantrxhead = 8;
-	p->wanttxhead = 8;
-	iov.iov_len   = 8;
-	iov.iov_buf   = &p->txlen[0];
-	nni_aio_set_iov(p->negoaio, 1, &iov);
-	nni_tls_send(p->tls, p->negoaio);
-	nni_mtx_unlock(&p->ep->mtx);
+	nni_aio_abort(p->negaio, rv);
+	nni_aio_finish_error(aio, rv);
 }
 
 static void
 tlstran_pipe_nego_cb(void *arg)
 {
 	tlstran_pipe *p   = arg;
-	nni_aio *     aio = p->negoaio;
-	nni_aio *     uaio;
+	nni_aio *     aio = p->negaio;
 	int           rv;
 
-	nni_mtx_lock(&p->ep->mtx);
+	nni_mtx_lock(&p->mtx);
 	if ((rv = nni_aio_result(aio)) != 0) {
 		goto done;
 	}
@@ -313,7 +208,7 @@ tlstran_pipe_nego_cb(void *arg)
 		nni_aio_set_iov(aio, 1, &iov);
 		// send it down...
 		nni_tls_send(p->tls, aio);
-		nni_mtx_unlock(&p->ep->mtx);
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
 	if (p->gotrxhead < p->wantrxhead) {
@@ -322,7 +217,7 @@ tlstran_pipe_nego_cb(void *arg)
 		iov.iov_buf = &p->rxlen[p->gotrxhead];
 		nni_aio_set_iov(aio, 1, &iov);
 		nni_tls_recv(p->tls, aio);
-		nni_mtx_unlock(&p->ep->mtx);
+		nni_mtx_unlock(&p->mtx);
 		return;
 	}
 	// We have both sent and received the headers.  Lets check the
@@ -337,23 +232,17 @@ tlstran_pipe_nego_cb(void *arg)
 	NNI_GET16(&p->rxlen[4], p->peer);
 
 done:
-	if ((uaio = p->useraio) != NULL) {
-		p->useraio = NULL;
-		if (rv == 0) {
-			// These can fail. Note that the TLS stack
-			// automatically starts out in NODELAY to make the
-			// handshake performant.
-			(void) nni_tls_set_nodelay(p->tls, p->nodelay);
-			(void) nni_tls_set_keepalive(p->tls, p->keepalive);
-			nni_aio_set_output(uaio, 0, p);
-			nni_aio_finish(uaio, 0, 0);
-			nni_mtx_unlock(&p->ep->mtx);
-			return;
-		}
-		nni_aio_finish_error(aio, rv);
+	if (rv == 0) {
+		// These can fail. Note that the TLS stack automatically
+		// starts out in NODELAY to make the handshake performant.
+		(void) nni_tls_set_nodelay(p->tls, p->nodelay);
+		(void) nni_tls_set_keepalive(p->tls, p->keepalive);
+	}
+	if ((aio = p->user_negaio) != NULL) {
+		p->user_negaio = NULL;
+		nni_aio_finish(aio, rv, 0);
 	}
-	nni_mtx_unlock(&p->ep->mtx);
-	tlstran_pipe_reap(p);
+	nni_mtx_unlock(&p->mtx);
 }
 
 static void
@@ -367,18 +256,17 @@ tlstran_pipe_send_cb(void *arg)
 	nni_aio *     txaio = p->txaio;
 
 	nni_mtx_lock(&p->mtx);
-	rv = p->closed ? NNG_ECLOSED : nni_aio_result(txaio);
-	if (rv != 0) {
+	aio = nni_list_first(&p->sendq);
+
+	if ((rv = nni_aio_result(txaio)) != 0) {
 		// Intentionally we do not queue up another transfer.
 		// There's an excellent chance that the pipe is no longer
 		// usable, with a partial transfer.
 		// The protocol should see this error, and close the
 		// pipe itself, we hope.
-		while ((aio = nni_list_first(&p->sendq)) != NULL) {
-			nni_aio_list_remove(aio);
-			nni_aio_finish_error(aio, rv);
-		}
+		nni_aio_list_remove(aio);
 		nni_mtx_unlock(&p->mtx);
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
@@ -389,7 +277,6 @@ tlstran_pipe_send_cb(void *arg)
 		nni_mtx_unlock(&p->mtx);
 		return;
 	}
-	aio = nni_list_first(&p->sendq);
 	nni_aio_list_remove(aio);
 	tlstran_pipe_send_start(p);
 	nni_mtx_unlock(&p->mtx);
@@ -412,14 +299,11 @@ tlstran_pipe_recv_cb(void *arg)
 	nni_aio *     rxaio = p->rxaio;
 
 	nni_mtx_lock(&p->mtx);
+	aio = nni_list_first(&p->recvq);
 
 	if ((rv = nni_aio_result(p->rxaio)) != 0) {
 		goto recv_error;
 	}
-	if (p->closed) {
-		rv = NNG_ECLOSED;
-		goto recv_error;
-	}
 
 	n = nni_aio_count(rxaio);
 	nni_aio_iov_advance(rxaio, n);
@@ -464,7 +348,6 @@ tlstran_pipe_recv_cb(void *arg)
 	}
 
 	// We read a message completely.  Let the user know the good news.
-	aio = nni_list_first(&p->recvq);
 	nni_aio_list_remove(aio);
 	msg      = p->rxmsg;
 	p->rxmsg = NULL;
@@ -478,16 +361,14 @@ tlstran_pipe_recv_cb(void *arg)
 	return;
 
 recv_error:
-	while ((aio = nni_list_first(&p->recvq)) != NULL) {
-		nni_aio_list_remove(aio);
-		nni_aio_finish_error(aio, rv);
-	}
+	nni_aio_list_remove(aio);
 	msg      = p->rxmsg;
 	p->rxmsg = NULL;
 	// Intentionally, we do not queue up another receive.
 	// The protocol should notice this error and close the pipe.
 	nni_mtx_unlock(&p->mtx);
 	nni_msg_free(msg);
+	nni_aio_finish_error(aio, rv);
 }
 
 static void
@@ -563,11 +444,6 @@ tlstran_pipe_send(void *arg, nni_aio *aio)
 		return;
 	}
 	nni_mtx_lock(&p->mtx);
-	if (p->closed) {
-		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, NNG_ECLOSED);
-		return;
-	}
 	if ((rv = nni_aio_schedule(aio, tlstran_pipe_send_cancel, p)) != 0) {
 		nni_mtx_unlock(&p->mtx);
 		nni_aio_finish_error(aio, rv);
@@ -610,7 +486,7 @@ tlstran_pipe_recv_start(tlstran_pipe *p)
 	nni_iov  iov;
 	NNI_ASSERT(p->rxmsg == NULL);
 
-	// Schedule a read of the header.
+	// Schedule a read of the IPC header.
 	rxaio       = p->rxaio;
 	iov.iov_buf = p->rxlen;
 	iov.iov_len = sizeof(p->rxlen);
@@ -629,11 +505,6 @@ tlstran_pipe_recv(void *arg, nni_aio *aio)
 		return;
 	}
 	nni_mtx_lock(&p->mtx);
-	if (p->closed) {
-		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, NNG_ECLOSED);
-		return;
-	}
 	if ((rv = nni_aio_schedule(aio, tlstran_pipe_recv_cancel, p)) != 0) {
 		nni_mtx_unlock(&p->mtx);
 		nni_aio_finish_error(aio, rv);
@@ -698,40 +569,56 @@ tlstran_pipe_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t)
 }
 
 static void
-tlstran_ep_close(tlstran_ep *ep)
+tlstran_pipe_start(void *arg, nni_aio *aio)
 {
-	tlstran_pipe *p;
-	nni_mtx_lock(&ep->mtx);
-	NNI_LIST_FOREACH (&ep->pipes, p) {
-		nni_aio_close(p->negoaio);
-		nni_aio_close(p->connaio);
-		nni_aio_close(p->rslvaio);
+	tlstran_pipe *p = arg;
+	nni_aio *     negaio;
+	nni_iov       iov;
+	int           rv;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
 	}
-	nni_mtx_unlock(&ep->mtx);
+	nni_mtx_lock(&p->mtx);
+	if ((rv = nni_aio_schedule(aio, tlstran_pipe_cancel_nego, p)) != 0) {
+		nni_mtx_unlock(&p->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	p->txlen[0] = 0;
+	p->txlen[1] = 'S';
+	p->txlen[2] = 'P';
+	p->txlen[3] = 0;
+	NNI_PUT16(&p->txlen[4], p->proto);
+	NNI_PUT16(&p->txlen[6], 0);
+
+	p->user_negaio = aio;
+	p->gotrxhead   = 0;
+	p->gottxhead   = 0;
+	p->wantrxhead  = 8;
+	p->wanttxhead  = 8;
+	negaio         = p->negaio;
+	iov.iov_len    = 8;
+	iov.iov_buf    = &p->txlen[0];
+	nni_aio_set_iov(negaio, 1, &iov);
+	nni_tls_send(p->tls, negaio);
+	nni_mtx_unlock(&p->mtx);
 }
 
 static void
 tlstran_dialer_fini(void *arg)
 {
-	tlstran_dialer *d  = arg;
-	tlstran_ep *    ep = &d->ep;
+	tlstran_dialer *d = arg;
 
-	nni_mtx_lock(&ep->mtx);
-	d->ep.fini = true;
+	nni_aio_stop(d->aio);
 	if (d->dialer != NULL) {
 		nni_tcp_dialer_fini(d->dialer);
-		d->dialer = NULL;
-	}
-	if (!nni_list_empty(&ep->pipes)) {
-		nni_mtx_unlock(&ep->mtx);
-		return;
 	}
-	nni_mtx_unlock(&ep->mtx);
-	if (ep->cfg != NULL) {
-		nni_tls_config_fini(ep->cfg);
+	nni_aio_fini(d->aio);
+	if (d->ep.cfg != NULL) {
+		nni_tls_config_fini(d->ep.cfg);
 	}
-	nni_mtx_fini(&ep->mtx);
-
+	nni_mtx_fini(&d->ep.mtx);
 	NNI_FREE_STRUCT(d);
 }
 
@@ -740,8 +627,8 @@ tlstran_dialer_close(void *arg)
 {
 	tlstran_dialer *d = arg;
 
+	nni_aio_close(d->aio);
 	nni_tcp_dialer_close(d->dialer);
-	tlstran_ep_close(&d->ep);
 }
 
 static int
@@ -777,48 +664,110 @@ tlstran_dialer_init(void **dp, nni_url *url, nni_sock *sock)
 	}
 
 	nni_mtx_init(&d->ep.mtx);
-	NNI_LIST_INIT(&d->ep.pipes, tlstran_pipe, node);
 	d->ep.authmode  = NNG_TLS_AUTH_MODE_REQUIRED;
 	d->ep.url       = url;
 	d->ep.proto     = nni_sock_proto_id(sock);
 	d->ep.nodelay   = true;
 	d->ep.keepalive = false;
-	d->ep.dtor      = tlstran_dialer_fini;
-	d->af           = af;
 
 	if (((rv = nni_tcp_dialer_init(&d->dialer)) != 0) ||
 	    ((rv = nni_tls_config_init(&d->ep.cfg, NNG_TLS_MODE_CLIENT)) !=
 	        0) ||
 	    ((rv = nng_tls_config_auth_mode(d->ep.cfg, d->ep.authmode)) !=
 	        0) ||
-	    ((rv = nng_tls_config_server_name(d->ep.cfg, host)) != 0)) {
+	    ((rv = nng_tls_config_server_name(d->ep.cfg, host)) != 0) ||
+	    ((rv = nni_aio_init(&d->aio, tlstran_dialer_cb, d)) != 0)) {
 		tlstran_dialer_fini(d);
 		return (rv);
 	}
+	d->af = af;
 
 	*dp = d;
 	return (0);
 }
 
 static void
-tlstran_pipe_conn_cancel(nni_aio *aio, int rv)
+tlstran_dialer_cb(void *arg)
 {
-	tlstran_pipe *p = nni_aio_get_prov_data(aio);
+	tlstran_dialer *d = arg;
+	tlstran_pipe *  p;
+	nni_tcp_conn *  conn;
+	nni_tls *       tls;
+	nni_aio *       aio;
+	int             rv;
 
-	nni_mtx_lock(&p->ep->mtx);
-	if (p->useraio != aio) {
-		nni_mtx_unlock(&p->ep->mtx);
+	nni_mtx_lock(&d->ep.mtx);
+	aio = d->user_aio;
+	rv  = nni_aio_result(d->aio);
+
+	if (aio == NULL) {
+		nni_mtx_unlock(&d->ep.mtx);
+		if ((rv == 0) && !d->resolving) {
+			conn = nni_aio_get_output(d->aio, 0);
+			nni_tcp_conn_fini(conn);
+		}
+		return;
+	}
+
+	if (rv != 0) {
+		d->user_aio = NULL;
+		nni_mtx_unlock(&d->ep.mtx);
+		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
-	// Close the underlying AIOs.  This will abort the operation.
-	// The pipe is removed from pending list at completion callback.
-	p->useraio = NULL;
-	nni_aio_close(p->negoaio);
-	nni_aio_close(p->connaio);
-	nni_aio_close(p->rslvaio);
-	nni_mtx_unlock(&p->ep->mtx);
+	if (d->resolving) {
+		// Name resolution complete.  Now go to next step.
+		d->resolving = false;
+		nni_tcp_dialer_dial(d->dialer, &d->sa, d->aio);
+		nni_mtx_unlock(&d->ep.mtx);
+		return;
+	}
+	d->user_aio = NULL;
+	conn        = nni_aio_get_output(d->aio, 0);
+	NNI_ASSERT(conn != NULL);
 
+	if ((rv = nni_tls_init(&tls, d->ep.cfg, conn)) != 0) {
+		nni_mtx_unlock(&d->ep.mtx);
+		nni_tcp_conn_fini(conn);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	if ((rv = tlstran_pipe_init(&p, tls)) != 0) {
+		nni_mtx_unlock(&d->ep.mtx);
+		nni_tls_fini(tls);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	p->proto     = d->ep.proto;
+	p->rcvmax    = d->ep.rcvmax;
+	p->nodelay   = d->ep.nodelay;
+	p->keepalive = d->ep.keepalive;
+	nni_mtx_unlock(&d->ep.mtx);
+
+	(void) nni_tls_set_nodelay(tls, p->nodelay);
+	(void) nni_tls_set_keepalive(tls, p->keepalive);
+
+	nni_aio_set_output(aio, 0, p);
+	nni_aio_finish(aio, 0, 0);
+}
+
+static void
+tlstran_dialer_cancel(nni_aio *aio, int rv)
+{
+	tlstran_dialer *d = nni_aio_get_prov_data(aio);
+
+	nni_mtx_lock(&d->ep.mtx);
+	if (d->user_aio != aio) {
+		nni_mtx_unlock(&d->ep.mtx);
+		return;
+	}
+	d->user_aio = NULL;
+	nni_mtx_unlock(&d->ep.mtx);
+
+	nni_aio_abort(d->aio, rv);
 	nni_aio_finish_error(aio, rv);
 }
 
@@ -826,52 +775,45 @@ static void
 tlstran_dialer_connect(void *arg, nni_aio *aio)
 {
 	tlstran_dialer *d = arg;
-	tlstran_pipe *  p = NULL;
 	int             rv;
 
 	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
 	nni_mtx_lock(&d->ep.mtx);
-	if (((rv = tlstran_pipe_init(&p, &d->ep)) != 0) ||
-	    ((rv = nni_aio_schedule(aio, tlstran_pipe_conn_cancel, p)) != 0)) {
+	NNI_ASSERT(d->user_aio == NULL);
+
+	if ((rv = nni_aio_schedule(aio, tlstran_dialer_cancel, d)) != 0) {
 		nni_mtx_unlock(&d->ep.mtx);
 		nni_aio_finish_error(aio, rv);
-		tlstran_pipe_reap(p);
 		return;
 	}
-	p->useraio = aio;
-	p->dialer  = d->dialer;
+	d->user_aio = aio;
 
-	// Start the name resolution.
-	nni_aio_set_input(p->rslvaio, 0, &p->sa);
+	d->resolving = true;
+
+	// Start the name resolution.  Callback will see resolving, and then
+	// switch to doing actual connect.
+	nni_aio_set_input(d->aio, 0, &d->sa);
 	nni_tcp_resolv(
-	    d->ep.url->u_hostname, d->ep.url->u_port, d->af, 0, p->rslvaio);
+	    d->ep.url->u_hostname, d->ep.url->u_port, d->af, 0, d->aio);
 	nni_mtx_unlock(&d->ep.mtx);
 }
 
 static void
 tlstran_listener_fini(void *arg)
 {
-	tlstran_listener *l  = arg;
-	tlstran_ep *      ep = &l->ep;
+	tlstran_listener *l = arg;
 
-	nni_mtx_lock(&ep->mtx);
-	l->ep.fini = true;
+	nni_aio_stop(l->aio);
 	if (l->listener != NULL) {
 		nni_tcp_listener_fini(l->listener);
-		l->listener = NULL;
 	}
-	if (!nni_list_empty(&ep->pipes)) {
-		nni_mtx_unlock(&ep->mtx);
-		return;
-	}
-	nni_mtx_unlock(&ep->mtx);
-	if (ep->cfg != NULL) {
-		nni_tls_config_fini(ep->cfg);
+	nni_aio_fini(l->aio);
+	if (l->ep.cfg != NULL) {
+		nni_tls_config_fini(l->ep.cfg);
 	}
-	nni_mtx_fini(&ep->mtx);
-
+	nni_mtx_fini(&l->ep.mtx);
 	NNI_FREE_STRUCT(l);
 }
 
@@ -880,7 +822,7 @@ tlstran_listener_close(void *arg)
 {
 	tlstran_listener *l = arg;
 
-	tlstran_ep_close(&l->ep);
+	nni_aio_close(l->aio);
 	nni_tcp_listener_close(l->listener);
 }
 
@@ -917,13 +859,11 @@ tlstran_listener_init(void **lp, nni_url *url, nni_sock *sock)
 		return (NNG_ENOMEM);
 	}
 	nni_mtx_init(&l->ep.mtx);
-	NNI_LIST_INIT(&l->ep.pipes, tlstran_pipe, node);
 	l->ep.url       = url;
 	l->ep.authmode  = NNG_TLS_AUTH_MODE_NONE;
 	l->ep.keepalive = false;
 	l->ep.nodelay   = true;
 	l->ep.proto     = nni_sock_proto_id(sock);
-	l->ep.dtor      = tlstran_listener_fini;
 
 	if (strlen(host) == 0) {
 		host = NULL;
@@ -956,7 +896,8 @@ tlstran_listener_init(void **lp, nni_url *url, nni_sock *sock)
 	    ((rv = nni_tls_config_init(&l->ep.cfg, NNG_TLS_MODE_SERVER)) !=
 	        0) ||
 	    ((rv = nng_tls_config_auth_mode(l->ep.cfg, l->ep.authmode)) !=
-	        0)) {
+	        0) ||
+	    ((rv = nni_aio_init(&l->aio, tlstran_listener_cb, l)) != 0)) {
 		tlstran_listener_fini(l);
 		return (rv);
 	}
@@ -981,26 +922,98 @@ tlstran_listener_bind(void *arg)
 }
 
 static void
-tlstran_listener_accept(void *arg, nni_aio *aio)
+tlstran_listener_cb(void *arg)
 {
 	tlstran_listener *l = arg;
+	nni_aio *         aio;
+	int               rv;
 	tlstran_pipe *    p = NULL;
+	nni_tcp_conn *    conn;
+	nni_tls *         tls;
+
+	nni_mtx_lock(&l->ep.mtx);
+	rv          = nni_aio_result(l->aio);
+	aio         = l->user_aio;
+	l->user_aio = NULL;
+
+	if (aio == NULL) {
+		nni_mtx_unlock(&l->ep.mtx);
+		if (rv == 0) {
+			conn = nni_aio_get_output(l->aio, 0);
+			nni_tcp_conn_fini(conn);
+		}
+		return;
+	}
+	if (rv != 0) {
+		nni_mtx_unlock(&l->ep.mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	conn = nni_aio_get_output(l->aio, 0);
+	if ((rv = nni_tls_init(&tls, l->ep.cfg, conn)) != 0) {
+		nni_mtx_unlock(&l->ep.mtx);
+		nni_tcp_conn_fini(conn);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	if ((rv = tlstran_pipe_init(&p, tls)) != 0) {
+		nni_mtx_unlock(&l->ep.mtx);
+		nni_tls_fini(tls);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	p->proto     = l->ep.proto;
+	p->rcvmax    = l->ep.rcvmax;
+	p->nodelay   = l->ep.nodelay;
+	p->keepalive = l->ep.keepalive;
+
+	(void) nni_tls_set_nodelay(tls, p->nodelay);
+	(void) nni_tls_set_keepalive(tls, p->keepalive);
+
+	nni_mtx_unlock(&l->ep.mtx);
+
+	nni_aio_set_output(aio, 0, p);
+	nni_aio_finish(aio, 0, 0);
+}
+
+static void
+tlstran_listener_cancel(nni_aio *aio, int rv)
+{
+	tlstran_listener *l = nni_aio_get_prov_data(aio);
+
+	nni_mtx_lock(&l->ep.mtx);
+	if (l->user_aio != aio) {
+		nni_mtx_unlock(&l->ep.mtx);
+		return;
+	}
+	l->user_aio = NULL;
+	nni_mtx_unlock(&l->ep.mtx);
+
+	nni_aio_abort(l->aio, rv);
+	nni_aio_finish_error(aio, rv);
+}
+
+static void
+tlstran_listener_accept(void *arg, nni_aio *aio)
+{
+	tlstran_listener *l = arg;
 	int               rv;
 
 	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
 	nni_mtx_lock(&l->ep.mtx);
-	if (((rv = tlstran_pipe_init(&p, &l->ep)) != 0) ||
-	    ((rv = nni_aio_schedule(aio, tlstran_pipe_conn_cancel, p)) != 0)) {
+	NNI_ASSERT(l->user_aio == NULL);
+
+	if ((rv = nni_aio_schedule(aio, tlstran_listener_cancel, l)) != 0) {
 		nni_mtx_unlock(&l->ep.mtx);
 		nni_aio_finish_error(aio, rv);
-		tlstran_pipe_reap(p);
 		return;
 	}
-	p->useraio = aio;
+	l->user_aio = aio;
 
-	nni_tcp_listener_accept(l->listener, p->connaio);
+	nni_tcp_listener_accept(l->listener, l->aio);
 	nni_mtx_unlock(&l->ep.mtx);
 }
 
@@ -1293,6 +1306,7 @@ static nni_tran_option tlstran_pipe_options[] = {
 
 static nni_tran_pipe_ops tlstran_pipe_ops = {
 	.p_fini    = tlstran_pipe_fini,
+	.p_start   = tlstran_pipe_start,
 	.p_stop    = tlstran_pipe_stop,
 	.p_send    = tlstran_pipe_send,
 	.p_recv    = tlstran_pipe_recv,
diff --git a/src/transport/zerotier/zerotier.c b/src/transport/zerotier/zerotier.c
index d1072931..a2163e3f 100644
--- a/src/transport/zerotier/zerotier.c
+++ b/src/transport/zerotier/zerotier.c
@@ -183,28 +183,27 @@ struct zt_fraglist {
 };
 
 struct zt_pipe {
-	nni_list_node   zp_link;
-	const char *    zp_addr;
-	zt_node *       zp_ztn;
-	uint64_t        zp_nwid;
-	uint64_t        zp_laddr;
-	uint64_t        zp_raddr;
-	uint16_t        zp_peer;
-	uint16_t        zp_proto;
-	uint16_t        zp_next_msgid;
-	size_t          zp_rcvmax;
-	size_t          zp_mtu;
-	nni_aio *       zp_user_rxaio;
-	nni_time        zp_last_recv;
-	zt_fraglist     zp_recvq[zt_recvq];
-	int             zp_ping_try;
-	int             zp_ping_tries;
-	bool            zp_closed;
-	nni_duration    zp_ping_time;
-	nni_aio *       zp_ping_aio;
-	uint8_t *       zp_send_buf;
-	nni_atomic_flag zp_reaped;
-	nni_reap_item   zp_reap;
+	nni_list_node zp_link;
+	const char *  zp_addr;
+	zt_node *     zp_ztn;
+	uint64_t      zp_nwid;
+	uint64_t      zp_laddr;
+	uint64_t      zp_raddr;
+	uint16_t      zp_peer;
+	uint16_t      zp_proto;
+	uint16_t      zp_next_msgid;
+	size_t        zp_rcvmax;
+	size_t        zp_mtu;
+	int           zp_closed;
+	nni_aio *     zp_user_rxaio;
+	nni_time      zp_last_recv;
+	zt_fraglist   zp_recvq[zt_recvq];
+	int           zp_ping_try;
+	int           zp_ping_tries;
+	int           zp_ping_active;
+	nni_duration  zp_ping_time;
+	nni_aio *     zp_ping_aio;
+	uint8_t *     zp_send_buf;
 };
 
 typedef struct zt_creq zt_creq;
@@ -281,7 +280,6 @@ static void zt_fraglist_free(zt_fraglist *);
 static void zt_virtual_recv(ZT_Node *, void *, void *, uint64_t, void **,
     uint64_t, uint64_t, unsigned int, unsigned int, const void *,
     unsigned int);
-static void zt_pipe_start_ping(zt_pipe *);
 
 static int64_t
 zt_now(void)
@@ -807,8 +805,10 @@ zt_pipe_close_err(zt_pipe *p, int err, uint8_t code, const char *msg)
 		p->zp_user_rxaio = NULL;
 		nni_aio_finish_error(aio, err);
 	}
-	nni_aio_close(p->zp_ping_aio);
-	p->zp_closed = true;
+	if ((aio = p->zp_ping_aio) != NULL) {
+		nni_aio_finish_error(aio, NNG_ECLOSED);
+	}
+	p->zp_closed = 1;
 	if (msg != NULL) {
 		zt_pipe_send_err(p, code, msg);
 	}
@@ -974,7 +974,7 @@ zt_pipe_recv_disc_req(zt_pipe *p, const uint8_t *data, size_t len)
 	// Don't bother to check the length, going to disconnect anyway.
 	if ((aio = p->zp_user_rxaio) != NULL) {
 		p->zp_user_rxaio = NULL;
-		p->zp_closed     = true;
+		p->zp_closed     = 1;
 		nni_aio_finish_error(aio, NNG_ECLOSED);
 	}
 }
@@ -990,7 +990,7 @@ zt_pipe_recv_error(zt_pipe *p, const uint8_t *data, size_t len)
 	// the day, the details are just not that interesting.
 	if ((aio = p->zp_user_rxaio) != NULL) {
 		p->zp_user_rxaio = NULL;
-		p->zp_closed     = true;
+		p->zp_closed     = 1;
 		nni_aio_finish_error(aio, NNG_ETRANERR);
 	}
 }
@@ -1627,8 +1627,7 @@ zt_pipe_close(void *arg)
 	nni_aio *aio;
 
 	nni_mtx_lock(&zt_lk);
-	p->zp_closed = true;
-	nni_aio_close(p->zp_ping_aio);
+	p->zp_closed = 1;
 	if ((aio = p->zp_user_rxaio) != NULL) {
 		p->zp_user_rxaio = NULL;
 		nni_aio_finish_error(aio, NNG_ECLOSED);
@@ -1660,14 +1659,6 @@ zt_pipe_fini(void *arg)
 	NNI_FREE_STRUCT(p);
 }
 
-static void
-zt_pipe_reap(zt_pipe *p)
-{
-	if (!nni_atomic_flag_test_and_set(&p->zp_reaped)) {
-		nni_reap(&p->zp_reap, zt_pipe_fini, p);
-	}
-}
-
 static int
 zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr)
 {
@@ -1696,7 +1687,6 @@ zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr)
 	p->zp_ping_time  = ep->ze_ping_time;
 	p->zp_next_msgid = (uint16_t) nni_random();
 	p->zp_ping_try   = 0;
-	nni_atomic_flag_reset(&p->zp_reaped);
 
 	if (ep->ze_mode == NNI_EP_MODE_DIAL) {
 		rv = nni_idhash_insert(ztn->zn_lpipes, laddr, p);
@@ -1706,8 +1696,7 @@ zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr)
 	if ((rv != 0) ||
 	    ((rv = nni_idhash_insert(ztn->zn_peers, p->zp_raddr, p)) != 0) ||
 	    ((rv = nni_aio_init(&p->zp_ping_aio, zt_pipe_ping_cb, p)) != 0)) {
-		zt_pipe_reap(p);
-		return (rv);
+		zt_pipe_fini(p);
 	}
 
 	// The largest fragment we can accept on this pipe. The MTU is
@@ -1728,7 +1717,7 @@ zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr)
 		fl->fl_missingsz = (maxfrags + 7) / 8;
 		fl->fl_missing   = nni_alloc(fl->fl_missingsz);
 		if (fl->fl_missing == NULL) {
-			zt_pipe_reap(p);
+			zt_pipe_fini(p);
 			return (NNG_ENOMEM);
 		}
 	}
@@ -2027,50 +2016,96 @@ zt_pipe_get_node(void *arg, void *buf, size_t *szp, nni_opt_type t)
 }
 
 static void
+zt_pipe_cancel_ping(nni_aio *aio, int rv)
+{
+	zt_pipe *p = nni_aio_get_prov_data(aio);
+
+	nni_mtx_lock(&zt_lk);
+	if (p->zp_ping_active) {
+		p->zp_ping_active = 0;
+		nni_aio_finish_error(aio, rv);
+	}
+	nni_mtx_unlock(&zt_lk);
+}
+
+static void
 zt_pipe_ping_cb(void *arg)
 {
 	zt_pipe *p   = arg;
 	nni_aio *aio = p->zp_ping_aio;
-	int      rv;
 
-	if ((rv = nni_aio_result(aio)) != 0) {
-		// We were canceled.  That means we're done.
-		return;
-	}
 	nni_mtx_lock(&zt_lk);
+
+	p->zp_ping_active = 0;
 	if (p->zp_closed || aio == NULL || (p->zp_ping_tries == 0) ||
 	    (p->zp_ping_time == NNG_DURATION_INFINITE) ||
 	    (p->zp_ping_time == NNG_DURATION_ZERO)) {
 		nni_mtx_unlock(&zt_lk);
 		return;
 	}
-	if (p->zp_ping_try >= p->zp_ping_tries) {
-		// Ping count exceeded; the other side is AFK.
-		// Close the pipe, but no need to send a reason to the peer.
-		zt_pipe_close_err(p, NNG_ECLOSED, 0, NULL);
+	if (nni_aio_result(aio) != NNG_ETIMEDOUT) {
 		nni_mtx_unlock(&zt_lk);
 		return;
 	}
-
-	if (nni_clock() > (p->zp_last_recv + p->zp_ping_time)) {
-		p->zp_ping_try++;
-		zt_pipe_send_ping(p);
+	if (p->zp_ping_try < p->zp_ping_tries) {
+		nni_time now = nni_clock();
+		nni_aio_set_timeout(aio, p->zp_ping_time);
+		// We want pings.  We only send one if needed, but we
+		// use the the timer to wake us up even if we aren't
+		// going to send a ping.  (We don't increment the try count
+		// unless we actually do send one though.)
+		if (nni_aio_begin(aio) == 0) {
+			int rv;
+			rv = nni_aio_schedule(aio, zt_pipe_cancel_ping, p);
+			if (rv != 0) {
+				nni_mtx_unlock(&zt_lk);
+				nni_aio_finish_error(aio, rv);
+				return;
+			}
+			p->zp_ping_active = 1;
+			if (now > (p->zp_last_recv + p->zp_ping_time)) {
+				p->zp_ping_try++;
+				zt_pipe_send_ping(p);
+			}
+		}
+	} else {
+		// Ping count exceeded; the other side is AFK.
+		// Close the pipe, but no need to send a reason to the peer.
+		zt_pipe_close_err(p, NNG_ECLOSED, 0, NULL);
 	}
-
-	nni_sleep_aio(p->zp_ping_time, aio); // Schedule a recheck.
 	nni_mtx_unlock(&zt_lk);
 }
 
 static void
-zt_pipe_start_ping(zt_pipe *p)
+zt_pipe_start(void *arg, nni_aio *aio)
 {
+	zt_pipe *p = arg;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+	nni_mtx_lock(&zt_lk);
+	p->zp_ping_active = 0;
 	// send a gratuitous ping, and start the ping interval timer.
 	if ((p->zp_ping_tries > 0) && (p->zp_ping_time != NNG_DURATION_ZERO) &&
-	    (p->zp_ping_time != NNG_DURATION_INFINITE)) {
+	    (p->zp_ping_time != NNG_DURATION_INFINITE) &&
+	    (p->zp_ping_aio != NULL)) {
 		p->zp_ping_try = 0;
-		zt_pipe_send_ping(p);
-		nni_sleep_aio(p->zp_ping_time, p->zp_ping_aio);
+		nni_aio_set_timeout(aio, p->zp_ping_time);
+		if (nni_aio_begin(p->zp_ping_aio) == 0) {
+			int rv;
+			rv = nni_aio_schedule(
+			    p->zp_ping_aio, zt_pipe_cancel_ping, p);
+			if (rv != 0) {
+				nni_aio_finish_error(p->zp_ping_aio, rv);
+			} else {
+				p->zp_ping_active = 1;
+				zt_pipe_send_ping(p);
+			}
+		}
 	}
+	nni_aio_finish(aio, 0, 0);
+	nni_mtx_unlock(&zt_lk);
 }
 
 static void
@@ -2390,7 +2425,6 @@ zt_ep_doaccept(zt_ep *ep)
 		}
 		p->zp_peer = creq.cr_proto;
 		zt_pipe_send_conn_ack(p);
-		zt_pipe_start_ping(p);
 		nni_aio_set_output(aio, 0, p);
 		nni_aio_finish(aio, 0, 0);
 	}
@@ -2450,7 +2484,6 @@ zt_ep_conn_req_cb(void *arg)
 		// Already canceled, or already handled?
 		if ((uaio = nni_list_first(&ep->ze_aios)) != NULL) {
 			nni_aio_list_remove(uaio);
-			zt_pipe_start_ping(p);
 			nni_aio_set_output(uaio, 0, p);
 			nni_aio_finish(uaio, 0, 0);
 		} else {
@@ -2958,6 +2991,7 @@ static nni_tran_option zt_pipe_options[] = {
 
 static nni_tran_pipe_ops zt_pipe_ops = {
 	.p_fini    = zt_pipe_fini,
+	.p_start   = zt_pipe_start,
 	.p_send    = zt_pipe_send,
 	.p_recv    = zt_pipe_recv,
 	.p_close   = zt_pipe_close,