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.

2 files changed, 31 insertions, 1 deletions

diff --git a/src/core/pipe.c b/src/core/pipe.c
index 08d3dee7..eaf4fa1c 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -53,6 +53,7 @@ nni_pipe_dtor(void *ptr)
 		p->p_tran_ops.p_fini(p->p_tran_data);
 	}
 
+	nni_aio_fini(&p->p_start_aio);
 	nni_mtx_fini(&p->p_mtx);
 	NNI_FREE_STRUCT(p);
 }
@@ -170,6 +171,28 @@ nni_pipe_peer(nni_pipe *p)
 }
 
 
+static void
+nni_pipe_start_cb(void *arg)
+{
+	nni_pipe *p = arg;
+	nni_aio *aio = &p->p_start_aio;
+	int rv;
+
+	nni_mtx_lock(&p->p_mtx);
+	if ((rv = nni_aio_result(aio)) != 0) {
+		nni_mtx_unlock(&p->p_mtx);
+		nni_pipe_stop(p);
+		return;
+	}
+
+	nni_mtx_unlock(&p->p_mtx);
+
+	if ((rv = nni_sock_pipe_ready(p->p_sock, p)) != 0) {
+		nni_pipe_stop(p);
+	}
+}
+
+
 int
 nni_pipe_create(nni_pipe **pp, nni_ep *ep, nni_sock *sock, nni_tran *tran)
 {
@@ -182,6 +205,10 @@ nni_pipe_create(nni_pipe **pp, nni_ep *ep, nni_sock *sock, nni_tran *tran)
 	if (rv != 0) {
 		return (rv);
 	}
+	if ((rv = nni_aio_init(&p->p_start_aio, nni_pipe_start_cb, p)) != 0) {
+		nni_objhash_unref(nni_pipes, p->p_id);
+		return (rv);
+	}
 	p->p_sock = sock;
 	p->p_ep = ep;
 
@@ -225,10 +252,12 @@ nni_pipe_start(nni_pipe *p)
 {
 	int rv;
 
-	if ((rv = nni_sock_pipe_ready(p->p_sock, p)) != 0) {
+	if (p->p_tran_ops.p_start == NULL) {
+		rv = nni_sock_pipe_ready(p->p_sock, p);
 		return (rv);
 	}
 
+	p->p_tran_ops.p_start(p->p_tran_data, &p->p_start_aio);
 	// XXX: Publish event
 
 	return (0);
diff --git a/src/core/pipe.h b/src/core/pipe.h
index e8e6ecad..f7df8232 100644
--- a/src/core/pipe.h
+++ b/src/core/pipe.h
@@ -32,6 +32,7 @@ struct nni_pipe {
 	int		p_stop;
 	nni_mtx		p_mtx;
 	nni_taskq_ent	p_reap_tqe;
+	nni_aio		p_start_aio;
 };
 
 extern int nni_pipe_sys_init(void);