Start of close related race fixes. Scalability test.

6 files changed, 262 insertions, 103 deletions

diff --git a/src/core/aio.c b/src/core/aio.c
index ffc5ac06..2f871f73 100644
--- a/src/core/aio.c
+++ b/src/core/aio.c
@@ -41,6 +41,7 @@ nni_aio_init(nni_aio *aio, nni_cb cb, void *arg)
 void
 nni_aio_fini(nni_aio *aio)
 {
+	nni_taskq_cancel(&aio->a_tqe);
 	nni_cv_fini(&aio->a_cv);
 	nni_mtx_fini(&aio->a_lk);
 }
diff --git a/src/core/msgqueue.c b/src/core/msgqueue.c
index 9607f562..47b98629 100644
--- a/src/core/msgqueue.c
+++ b/src/core/msgqueue.c
@@ -388,6 +388,11 @@ nni_msgq_aio_put(nni_msgq *mq, nni_aio *aio)
 	nni_time expire = aio->a_expire;
 
 	nni_mtx_lock(&mq->mq_lock);
+	if (mq->mq_closed) {
+		nni_aio_finish(aio, NNG_ECLOSED, 0);
+		nni_mtx_unlock(&mq->mq_lock);
+		return;
+	}
 	nni_list_append(&mq->mq_aio_putq, aio);
 	nni_msgq_run_putq(mq);
 	nni_msgq_run_notify(mq);
@@ -406,6 +411,11 @@ nni_msgq_aio_get(nni_msgq *mq, nni_aio *aio)
 	nni_time expire = aio->a_expire;
 
 	nni_mtx_lock(&mq->mq_lock);
+	if (mq->mq_closed) {
+		nni_aio_finish(aio, NNG_ECLOSED, 0);
+		nni_mtx_unlock(&mq->mq_lock);
+		return;
+	}
 	nni_list_append(&mq->mq_aio_getq, aio);
 	nni_msgq_run_getq(mq);
 	nni_msgq_run_notify(mq);
@@ -428,6 +438,7 @@ nni_msgq_aio_cancel(nni_msgq *mq, nni_aio *aio)
 	// the node from either the getq or the putq list.
 	if (nni_list_active(&mq->mq_aio_getq, aio)) {
 		nni_list_remove(&mq->mq_aio_getq, aio);
+		nni_aio_finish(aio, NNG_ECANCELED, 0);
 	}
 	nni_mtx_unlock(&mq->mq_lock);
 }
@@ -437,6 +448,10 @@ int
 nni_msgq_canput(nni_msgq *mq)
 {
 	nni_mtx_lock(&mq->mq_lock);
+	if (mq->mq_closed) {
+		nni_mtx_unlock(&mq->mq_lock);
+		return (0);
+	}
 	if ((mq->mq_len < mq->mq_cap) ||
 	    (mq->mq_rwait != 0) ||      // XXX: REMOVE ME
 	    (nni_list_first(&mq->mq_aio_getq) != NULL)) {
@@ -452,6 +467,10 @@ int
 nni_msgq_canget(nni_msgq *mq)
 {
 	nni_mtx_lock(&mq->mq_lock);
+	if (mq->mq_closed) {
+		nni_mtx_unlock(&mq->mq_lock);
+		return (0);
+	}
 	if ((mq->mq_len != 0) ||
 	    (mq->mq_wwait != 0) ||
 	    (nni_list_first(&mq->mq_aio_putq) != NULL)) {
@@ -470,6 +489,10 @@ nni_msgq_tryput(nni_msgq *mq, nni_msg *msg)
 	size_t len = nni_msg_len(msg);
 
 	nni_mtx_lock(&mq->mq_lock);
+	if (mq->mq_closed) {
+		nni_mtx_unlock(&mq->mq_lock);
+		return (NNG_ECLOSED);
+	}
 
 	// The presence of any blocked reader indicates that
 	// the queue is empty, otherwise it would have just taken
@@ -804,6 +827,9 @@ nni_msgq_drain(nni_msgq *mq, nni_time expire)
 void
 nni_msgq_close(nni_msgq *mq)
 {
+	nni_aio *aio;
+	nni_aio *naio;
+
 	nni_mtx_lock(&mq->mq_lock);
 	mq->mq_closed = 1;
 	mq->mq_wwait = 0;
@@ -821,6 +847,22 @@ nni_msgq_close(nni_msgq *mq)
 		mq->mq_len--;
 		nni_msg_free(msg);
 	}
+
+	// Let all pending blockers know we are closing the queue.
+	naio = nni_list_first(&mq->mq_aio_getq);
+	while ((aio = naio) != NULL) {
+		naio = nni_list_next(&mq->mq_aio_getq, aio);
+		nni_list_remove(&mq->mq_aio_getq, aio);
+		nni_aio_finish(aio, NNG_ECLOSED, 0);
+	}
+
+	naio = nni_list_first(&mq->mq_aio_putq);
+	while ((aio = naio) != NULL) {
+		naio = nni_list_next(&mq->mq_aio_putq, aio);
+		nni_list_remove(&mq->mq_aio_putq, aio);
+		nni_aio_finish(aio, NNG_ECLOSED, 0);
+	}
+
 	nni_mtx_unlock(&mq->mq_lock);
 }
 
diff --git a/src/core/pipe.c b/src/core/pipe.c
index a401e4e3..18c47c60 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -50,6 +50,30 @@ nni_pipe_aio_send(nni_pipe *p, nni_aio *aio)
 }
 
 
+void
+nni_pipe_incref(nni_pipe *p)
+{
+	nni_mtx_lock(&p->p_mtx);
+	p->p_refcnt++;
+	nni_mtx_unlock(&p->p_mtx);
+}
+
+
+void
+nni_pipe_decref(nni_pipe *p)
+{
+	nni_mtx_lock(&p->p_mtx);
+	p->p_refcnt--;
+	if (p->p_refcnt == 0) {
+		nni_mtx_unlock(&p->p_mtx);
+
+		nni_pipe_destroy(p);
+		return;
+	}
+	nni_mtx_unlock(&p->p_mtx);
+}
+
+
 // nni_pipe_close closes the underlying connection.  It is expected that
 // subsequent attempts receive or send (including any waiting receive) will
 // simply return NNG_ECLOSED.
@@ -58,37 +82,34 @@ nni_pipe_close(nni_pipe *p)
 {
 	nni_sock *sock = p->p_sock;
 
+	nni_mtx_lock(&p->p_mtx);
+	if (p->p_reap == 1) {
+		// We already did a close.
+		nni_mtx_unlock(&p->p_mtx);
+		return;
+	}
+	p->p_reap = 1;
+
+	// Close the underlying transport.
 	if (p->p_tran_data != NULL) {
 		p->p_tran_ops.pipe_close(p->p_tran_data);
 	}
 
-	nni_mtx_lock(&sock->s_mx);
-	if (!p->p_reap) {
-		// schedule deferred reap/close
-		p->p_reap = 1;
-		nni_list_remove(&sock->s_pipes, p);
-		nni_list_append(&sock->s_reaps, p);
-		nni_cv_wake(&sock->s_cv);
+	// Unregister our ID so nobody else can find it.
+	if (p->p_id != 0) {
+		nni_mtx_lock(nni_idlock);
+		nni_idhash_remove(nni_pipes, p->p_id);
+		nni_mtx_unlock(nni_idlock);
+		p->p_id = 0;
 	}
-	nni_mtx_unlock(&sock->s_mx);
-}
 
+	nni_mtx_unlock(&p->p_mtx);
 
-// nni_pipe_bail is a special version of close, that is used to abort
-// from nni_pipe_start, when it fails.  It requires the lock to be held,
-// and this prevents us from dropping the lock, possibly leading to race
-// conditions.  It's critical that this not be called after the pipe is
-// started, or deadlock will occur.
-static void
-nni_pipe_bail(nni_pipe *p)
-{
-	nni_sock *sock = p->p_sock;
-
-	if (p->p_tran_data != NULL) {
-		p->p_tran_ops.pipe_close(p->p_tran_data);
-	}
+	// Let the socket (and endpoint) know we have closed.
+	nni_sock_pipe_closed(sock, p);
 
-	nni_pipe_destroy(p);
+	// Drop a reference count, possibly doing deferred destroy.
+	nni_pipe_decref(p);
 }
 
 
@@ -99,25 +120,6 @@ nni_pipe_peer(nni_pipe *p)
 }
 
 
-void
-nni_pipe_destroy(nni_pipe *p)
-{
-	int i;
-
-	for (i = 0; i < NNI_MAXWORKERS; i++) {
-		nni_thr_fini(&p->p_worker_thr[i]);
-	}
-
-	if (p->p_tran_data != NULL) {
-		p->p_tran_ops.pipe_destroy(p->p_tran_data);
-	}
-	if (p->p_proto_data != NULL) {
-		p->p_sock->s_pipe_ops.pipe_fini(p->p_proto_data);
-	}
-	NNI_FREE_STRUCT(p);
-}
-
-
 int
 nni_pipe_create(nni_pipe **pp, nni_ep *ep)
 {
@@ -126,15 +128,17 @@ nni_pipe_create(nni_pipe **pp, nni_ep *ep)
 	const nni_proto_pipe_ops *ops = &sock->s_pipe_ops;
 	void *pdata;
 	int rv;
-	int i;
 
 	if ((p = NNI_ALLOC_STRUCT(p)) == NULL) {
 		return (NNG_ENOMEM);
 	}
+	if ((rv = nni_mtx_init(&p->p_mtx)) != 0) {
+		NNI_FREE_STRUCT(p);
+		return (rv);
+	}
 	p->p_sock = sock;
 	p->p_tran_data = NULL;
 	p->p_proto_data = NULL;
-	p->p_active = 0;
 	p->p_id = 0;
 	NNI_LIST_NODE_INIT(&p->p_node);
 
@@ -143,30 +147,37 @@ nni_pipe_create(nni_pipe **pp, nni_ep *ep)
 	p->p_tran_ops = *ep->ep_tran->tran_pipe;
 
 	if ((rv = ops->pipe_init(&pdata, p, sock->s_data)) != 0) {
+		nni_mtx_fini(&p->p_mtx);
 		NNI_FREE_STRUCT(p);
 		return (rv);
 	}
 	p->p_proto_data = pdata;
-
-	for (i = 0; i < NNI_MAXWORKERS; i++) {
-		nni_worker fn = ops->pipe_worker[i];
-		rv = nni_thr_init(&p->p_worker_thr[i], fn, pdata);
-		if (rv != 0) {
-			while (i > 0) {
-				i--;
-				nni_thr_fini(&p->p_worker_thr[i]);
-			}
-			ops->pipe_fini(pdata);
-			NNI_FREE_STRUCT(p);
-			return (rv);
-		}
-	}
+	nni_sock_pipe_add(sock, p);
 
 	*pp = p;
 	return (0);
 }
 
 
+void
+nni_pipe_destroy(nni_pipe *p)
+{
+	NNI_ASSERT(p->p_refcnt == 0);
+
+	// The caller is responsible for ensuring that the pipe
+	// is not in use by any other consumers.  It must not be started
+	if (p->p_tran_data != NULL) {
+		p->p_tran_ops.pipe_destroy(p->p_tran_data);
+	}
+	if (p->p_proto_data != NULL) {
+		p->p_sock->s_pipe_ops.pipe_fini(p->p_proto_data);
+	}
+	nni_sock_pipe_rem(p->p_sock, p);
+	nni_mtx_fini(&p->p_mtx);
+	NNI_FREE_STRUCT(p);
+}
+
+
 int
 nni_pipe_getopt(nni_pipe *p, int opt, void *val, size_t *szp)
 {
@@ -179,55 +190,27 @@ nni_pipe_getopt(nni_pipe *p, int opt, void *val, size_t *szp)
 
 
 int
-nni_pipe_start(nni_pipe *pipe)
+nni_pipe_start(nni_pipe *p)
 {
 	int rv;
-	int i;
-	nni_sock *sock = pipe->p_sock;
-
-	nni_mtx_lock(&sock->s_mx);
-	if (sock->s_closing) {
-		nni_pipe_bail(pipe);
-		nni_mtx_unlock(&sock->s_mx);
-		return (NNG_ECLOSED);
-	}
 
-	if (nni_pipe_peer(pipe) != sock->s_peer) {
-		nni_pipe_bail(pipe);
-		nni_mtx_unlock(&sock->s_mx);
-		return (NNG_EPROTO);
-	}
+	nni_pipe_incref(p);
 
 	nni_mtx_lock(nni_idlock);
-	rv = nni_idhash_alloc(nni_pipes, &pipe->p_id, pipe);
+	rv = nni_idhash_alloc(nni_pipes, &p->p_id, p);
 	nni_mtx_unlock(nni_idlock);
 
 	if (rv != 0) {
-		nni_pipe_bail(pipe);
-		nni_mtx_unlock(&sock->s_mx);
+		nni_pipe_close(p);
 		return (rv);
 	}
 
-	if ((rv = sock->s_pipe_ops.pipe_add(pipe->p_proto_data)) != 0) {
-		nni_mtx_lock(nni_idlock);
-		nni_idhash_remove(nni_pipes, pipe->p_id);
-		pipe->p_id = 0;
-		nni_mtx_unlock(nni_idlock);
-
-		nni_pipe_bail(pipe);
-		nni_mtx_unlock(&sock->s_mx);
+	if ((rv = nni_sock_pipe_ready(p->p_sock, p)) != 0) {
+		nni_pipe_close(p);
 		return (rv);
 	}
 
-	pipe->p_active = 1;
-	nni_list_append(&sock->s_pipes, pipe);
-
-	for (i = 0; i < NNI_MAXWORKERS; i++) {
-		nni_thr_run(&pipe->p_worker_thr[i]);
-	}
-
 	// XXX: Publish event
 
-	nni_mtx_unlock(&sock->s_mx);
 	return (0);
 }
diff --git a/src/core/pipe.h b/src/core/pipe.h
index 3ec4a7a3..6cabf4e7 100644
--- a/src/core/pipe.h
+++ b/src/core/pipe.h
@@ -28,7 +28,8 @@ struct nni_pipe {
 	nni_ep *	p_ep;
 	int		p_reap;
 	int		p_active;
-	nni_thr		p_worker_thr[NNI_MAXWORKERS];
+	nni_mtx		p_mtx;
+	int		p_refcnt;
 };
 
 // AIO
@@ -40,6 +41,8 @@ extern int nni_pipe_recv(nni_pipe *, nng_msg **);
 extern int nni_pipe_send(nni_pipe *, nng_msg *);
 extern uint32_t nni_pipe_id(nni_pipe *);
 extern void nni_pipe_close(nni_pipe *);
+extern void nni_pipe_incref(nni_pipe *);
+extern void nni_pipe_decref(nni_pipe *);
 
 // Used only by the socket core - as we don't wish to expose the details
 // of the pipe structure outside of pipe.c.
diff --git a/src/core/socket.c b/src/core/socket.c
index d58e64ba..40fb42bc 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -128,12 +128,113 @@ nni_sock_held_close(nni_sock *sock)
 }
 
 
+void
+nni_sock_pipe_add(nni_sock *sock, nni_pipe *pipe)
+{
+	nni_mtx_lock(&sock->s_mx);
+	nni_list_append(&sock->s_pipes, pipe);
+	nni_mtx_unlock(&sock->s_mx);
+}
+
+
+int
+nni_sock_pipe_ready(nni_sock *sock, nni_pipe *pipe)
+{
+	int rv;
+
+	nni_mtx_lock(&sock->s_mx);
+
+	if (sock->s_closing) {
+		nni_mtx_unlock(&sock->s_mx);
+		return (NNG_ECLOSED);
+	}
+	if (nni_pipe_peer(pipe) != sock->s_peer) {
+		nni_mtx_unlock(&sock->s_mx);
+		return (NNG_EPROTO);
+	}
+
+	if ((rv = sock->s_pipe_ops.pipe_add(pipe->p_proto_data)) != 0) {
+		nni_mtx_unlock(&sock->s_mx);
+		return (rv);
+	}
+
+	pipe->p_active = 1;
+
+	nni_list_remove(&sock->s_idles, pipe);
+	nni_list_append(&sock->s_pipes, pipe);
+
+	nni_mtx_unlock(&sock->s_mx);
+
+	return (0);
+}
+
+
+void
+nni_sock_pipe_closed(nni_sock *sock, nni_pipe *pipe)
+{
+	nni_ep *ep;
+
+	nni_mtx_lock(&sock->s_mx);
+
+	// NB: nni_list_remove doesn't really care *which* list the pipe
+	// is on, and so if the pipe is already on the idle list these
+	// two statements are effectively a no-op.
+	nni_list_remove(&sock->s_pipes, pipe);
+	nni_list_append(&sock->s_idles, pipe);
+
+	if (pipe->p_active) {
+		pipe->p_active = 0;
+		sock->s_pipe_ops.pipe_rem(pipe->p_proto_data);
+	}
+
+	// Notify the endpoint that the pipe has closed.
+	if (((ep = pipe->p_ep) != NULL) && ((ep->ep_pipe == pipe))) {
+		ep->ep_pipe = NULL;
+		nni_cv_wake(&ep->ep_cv);
+	}
+	nni_mtx_unlock(&sock->s_mx);
+}
+
+
+void
+nni_sock_pipe_rem(nni_sock *sock, nni_pipe *pipe)
+{
+	nni_ep *ep;
+
+	nni_mtx_lock(&sock->s_mx);
+	nni_list_remove(&sock->s_idles, pipe);
+
+	// Notify the endpoint that the pipe has closed - if not already done.
+	if (((ep = pipe->p_ep) != NULL) && ((ep->ep_pipe == pipe))) {
+		ep->ep_pipe = NULL;
+		nni_cv_wake(&ep->ep_cv);
+	}
+	nni_cv_wake(&sock->s_cv);
+	nni_mtx_unlock(&sock->s_mx);
+}
+
+
+void
+nni_sock_lock(nni_sock *sock)
+{
+	nni_mtx_lock(&sock->s_mx);
+}
+
+
+void
+nni_sock_unlock(nni_sock *sock)
+{
+	nni_mtx_unlock(&sock->s_mx);
+}
+
+
 // Because we have to call back into the socket, and possibly also the proto,
 // and wait for threads to terminate, we do this in a special thread.  The
 // assumption is that closing is always a "fast" operation.
 static void
 nni_reaper(void *arg)
 {
+#if 0
 	nni_sock *sock = arg;
 
 	for (;;) {
@@ -183,6 +284,7 @@ nni_reaper(void *arg)
 		nni_cv_wait(&sock->s_cv);
 		nni_mtx_unlock(&sock->s_mx);
 	}
+#endif
 }
 
 
@@ -301,7 +403,7 @@ nni_sock_open(nni_sock **sockp, uint16_t pnum)
 	sock->s_reapexit = 0;
 	sock->s_rcvmaxsz = 1024 * 1024; // 1 MB by default
 	NNI_LIST_INIT(&sock->s_pipes, nni_pipe, p_node);
-	NNI_LIST_INIT(&sock->s_reaps, nni_pipe, p_node);
+	NNI_LIST_INIT(&sock->s_idles, nni_pipe, p_node);
 	NNI_LIST_INIT(&sock->s_eps, nni_ep, ep_node);
 	NNI_LIST_INIT(&sock->s_notify, nni_notify, n_node);
 	NNI_LIST_INIT(&sock->s_events, nni_event, e_node);
@@ -512,15 +614,14 @@ nni_sock_shutdown(nni_sock *sock)
 	nni_msgq_close(sock->s_urq);
 	nni_msgq_close(sock->s_uwq);
 
-	// For each pipe, close the underlying transport, and move it to
-	// deathrow (the reaplist).
+	// For each pipe, close the underlying transport.  Also move it
+	// to the idle list so we won't keep looping.
 	while ((pipe = nni_list_first(&sock->s_pipes)) != NULL) {
-		if (pipe->p_tran_data != NULL) {
-			pipe->p_tran_ops.pipe_close(pipe->p_tran_data);
-		}
-		pipe->p_reap = 1;
-		nni_list_remove(&sock->s_pipes, pipe);
-		nni_list_append(&sock->s_reaps, pipe);
+		nni_pipe_incref(pipe);
+		nni_mtx_unlock(&sock->s_mx);
+		nni_pipe_close(pipe);
+		nni_pipe_decref(pipe);
+		nni_mtx_lock(&sock->s_mx);
 	}
 
 	sock->s_sock_ops.sock_close(sock->s_data);
@@ -528,6 +629,11 @@ nni_sock_shutdown(nni_sock *sock)
 	sock->s_reapexit = 1;
 	nni_cv_wake(&sock->s_notify_cv);
 	nni_cv_wake(&sock->s_cv);
+
+	while ((nni_list_first(&sock->s_idles) != NULL) ||
+	    (nni_list_first(&sock->s_pipes) != NULL)) {
+		nni_cv_wait(&sock->s_cv);
+	}
 	nni_mtx_unlock(&sock->s_mx);
 
 	// Wait for the threads to exit.
diff --git a/src/core/socket.h b/src/core/socket.h
index d7a7eb5e..22873c3c 100644
--- a/src/core/socket.h
+++ b/src/core/socket.h
@@ -42,7 +42,8 @@ struct nni_socket {
 	nni_duration		s_reconnmax;    // max reconnect time
 
 	nni_list		s_eps;          // active endpoints
-	nni_list		s_pipes;        // pipes for this socket
+	nni_list		s_pipes;        // ready pipes (started)
+	nni_list		s_idles;        // idle pipes (not ready)
 	nni_list		s_events;       // pending events
 	nni_list		s_notify;       // event watchers
 	nni_cv			s_notify_cv;    // wakes notify thread
@@ -89,6 +90,29 @@ extern int nni_sock_dial(nni_sock *, const char *, nni_ep **, int);
 extern int nni_sock_listen(nni_sock *, const char *, nni_ep **, int);
 extern uint32_t nni_sock_id(nni_sock *);
 
+extern void nni_sock_lock(nni_sock *);
+extern void nni_sock_unlock(nni_sock *);
+
+// nni_sock_pipe_add is called by the pipe to register the pipe with
+// with the socket.  The pipe is added to the idle list.
+extern void nni_sock_pipe_add(nni_sock *, nni_pipe *);
+
+// nni_sock_pipe_rem deregisters the pipe from the socket.  The socket
+// will block during close if there are registered pipes outstanding.
+extern void nni_sock_pipe_rem(nni_sock *, nni_pipe *);
+
+// nni_sock_pipe_ready lets the socket know the pipe is ready for
+// business.  This also calls the socket/protocol specific add function,
+// and it may return an error.  A reference count on the pipe is incremented
+// on success.  The reference count should be dropped by nni_sock_pipe_closed.
+extern int nni_sock_pipe_ready(nni_sock *, nni_pipe *);
+
+// nni_sock_pipe_closed lets the socket know that the pipe is closed.
+// This keeps the socket from trying to schedule traffic to it.  It
+// also lets the endpoint know about it, to possibly restart a dial
+// operation.
+extern void nni_sock_pipe_closed(nni_sock *, nni_pipe *);
+
 // Set error codes for applications.  These are only ever
 // called from the filter functions in protocols, and thus
 // already have the socket lock held.