Move IPC negotiation out of connect/accept.

This prevents a slow partner from blocking new connections from being
established on the server.  Before this a single partner could cause
the server to block waiting to complete the negotiation.

1 files changed, 34 insertions, 52 deletions

diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c
index 7a486b4d..6ad9da75 100644
--- a/src/transport/ipc/ipc.c
+++ b/src/transport/ipc/ipc.c
@@ -140,7 +140,7 @@ fail:
 
 
 static void
-nni_ipc_cancel_nego(nni_aio *aio)
+nni_ipc_cancel_start(nni_aio *aio)
 {
 	nni_ipc_pipe *pipe = aio->a_prov_data;
 
@@ -411,6 +411,38 @@ nni_ipc_pipe_recv(void *arg, nni_aio *aio)
 }
 
 
+static void
+nni_ipc_pipe_start(void *arg, nni_aio *aio)
+{
+	nni_ipc_pipe *pipe = arg;
+	int rv;
+
+	nni_mtx_lock(&pipe->mtx);
+	pipe->txhead[0] = 0;
+	pipe->txhead[1] = 'S';
+	pipe->txhead[2] = 'P';
+	pipe->txhead[3] = 0;
+	NNI_PUT16(&pipe->txhead[4], pipe->proto);
+	NNI_PUT16(&pipe->txhead[6], 0);
+
+	pipe->user_negaio = aio;
+	pipe->gotrxhead = 0;
+	pipe->gottxhead = 0;
+	pipe->wantrxhead = 8;
+	pipe->wanttxhead = 8;
+	pipe->negaio.a_niov = 1;
+	pipe->negaio.a_iov[0].iov_len = 8;
+	pipe->negaio.a_iov[0].iov_buf = &pipe->txhead[0];
+	rv = nni_aio_start(aio, nni_ipc_cancel_start, pipe);
+	if (rv != 0) {
+		nni_mtx_unlock(&pipe->mtx);
+		return;
+	}
+	nni_plat_ipc_send(pipe->isp, &pipe->negaio);
+	nni_mtx_unlock(&pipe->mtx);
+}
+
+
 static uint16_t
 nni_ipc_pipe_peer(void *arg)
 {
@@ -491,49 +523,6 @@ nni_ipc_ep_close(void *arg)
 
 
 static int
-nni_ipc_negotiate(nni_ipc_pipe *pipe)
-{
-	int rv;
-	nni_iov iov;
-	uint8_t buf[8];
-	nni_aio aio;
-
-	pipe->txhead[0] = 0;
-	pipe->txhead[1] = 'S';
-	pipe->txhead[2] = 'P';
-	pipe->txhead[3] = 0;
-	NNI_PUT16(&pipe->txhead[4], pipe->proto);
-	NNI_PUT16(&pipe->txhead[6], 0);
-
-	nni_aio_init(&aio, NULL, NULL);
-
-	nni_mtx_lock(&pipe->mtx);
-	pipe->user_negaio = &aio;
-	pipe->gotrxhead = 0;
-	pipe->gottxhead = 0;
-	pipe->wantrxhead = 8;
-	pipe->wanttxhead = 8;
-	pipe->negaio.a_niov = 1;
-	pipe->negaio.a_iov[0].iov_len = 8;
-	pipe->negaio.a_iov[0].iov_buf = &pipe->txhead[0];
-	rv = nni_aio_start(&aio, nni_ipc_cancel_nego, pipe);
-	if (rv != 0) {
-		nni_mtx_unlock(&pipe->mtx);
-		return (NNG_ECLOSED);
-	}
-	nni_plat_ipc_send(pipe->isp, &pipe->negaio);
-	nni_mtx_unlock(&pipe->mtx);
-
-	nni_aio_wait(&aio);
-	rv = nni_aio_result(&aio);
-	nni_aio_fini(&aio);
-	NNI_ASSERT(pipe->user_negaio == NULL);
-
-	return (rv);
-}
-
-
-static int
 nni_ipc_ep_connect_sync(void *arg, void **pipep)
 {
 	nni_ipc_ep *ep = arg;
@@ -557,10 +546,6 @@ nni_ipc_ep_connect_sync(void *arg, void **pipep)
 		return (rv);
 	}
 
-	if ((rv = nni_ipc_negotiate(pipe)) != 0) {
-		nni_ipc_pipe_fini(pipe);
-		return (rv);
-	}
 	*pipep = pipe;
 	return (0);
 }
@@ -600,10 +585,6 @@ nni_ipc_ep_accept_sync(void *arg, void **pipep)
 		nni_ipc_pipe_fini(pipe);
 		return (rv);
 	}
-	if ((rv = nni_ipc_negotiate(pipe)) != 0) {
-		nni_ipc_pipe_fini(pipe);
-		return (rv);
-	}
 	*pipep = pipe;
 	return (0);
 }
@@ -611,6 +592,7 @@ nni_ipc_ep_accept_sync(void *arg, void **pipep)
 
 static nni_tran_pipe nni_ipc_pipe_ops = {
 	.p_fini		= nni_ipc_pipe_fini,
+	.p_start	= nni_ipc_pipe_start,
 	.p_send		= nni_ipc_pipe_send,
 	.p_recv		= nni_ipc_pipe_recv,
 	.p_close	= nni_ipc_pipe_close,