Bind the pipe to the ep properly, and wake any closers needed.

6 files changed, 67 insertions, 27 deletions

diff --git a/etc/format-check.sh b/etc/format-check.sh
index 0438d8de..b3e0da3f 100755
--- a/etc/format-check.sh
+++ b/etc/format-check.sh
@@ -47,7 +47,6 @@ if [ "${maj}" -eq 3 -a "${min}" -lt 6 ]; then
 	echo "clang-format is too old.  Skipping checks."
 	exit 0
 fi
-echo "clang-format looks like ${maj} . ${min}"
 
 
 mytmpdir=`mktemp -d`
diff --git a/src/core/endpt.c b/src/core/endpt.c
index da0e3318..96e8e42c 100644
--- a/src/core/endpt.c
+++ b/src/core/endpt.c
@@ -94,7 +94,6 @@ nni_ep_create(nni_ep **epp, nni_sock *sock, const char *addr, int mode)
 	}
 	ep->ep_closed = 0;
 	ep->ep_bound  = 0;
-	ep->ep_pipe   = NULL;
 	ep->ep_id     = id;
 	ep->ep_data   = NULL;
 	ep->ep_refcnt = 0;
@@ -431,6 +430,45 @@ nni_ep_listen(nni_ep *ep, int flags)
 	return (0);
 }
 
+int
+nni_ep_pipe_add(nni_ep *ep, nni_pipe *p)
+{
+	nni_mtx_lock(&ep->ep_mtx);
+	if (ep->ep_closed) {
+		nni_mtx_unlock(&ep->ep_mtx);
+		return (NNG_ECLOSED);
+	}
+	nni_list_append(&ep->ep_pipes, p);
+	p->p_ep = ep;
+	nni_mtx_unlock(&ep->ep_mtx);
+	return (0);
+}
+
+void
+nni_ep_pipe_remove(nni_ep *ep, nni_pipe *pipe)
+{
+	// Break up the relationship between the EP and the pipe.
+	nni_mtx_lock(&ep->ep_mtx);
+	// During early init, the pipe might not have this set.
+	if (nni_list_active(&ep->ep_pipes, pipe)) {
+		nni_list_remove(&ep->ep_pipes, pipe);
+	}
+	pipe->p_ep = NULL;
+	// Wake up the close thread if it is waiting.
+	if (ep->ep_closed && nni_list_empty(&ep->ep_pipes)) {
+		nni_cv_wake(&ep->ep_cv);
+	}
+
+	// If this pipe closed, then lets restart the dial operation.
+	// Since the remote side seems to have closed, lets start with
+	// a backoff.  This keeps us from pounding the crap out of the
+	// thing if a remote server accepts but then disconnects immediately.
+	if ((!ep->ep_closed) && (ep->ep_mode == NNI_EP_MODE_DIAL)) {
+		nni_ep_backoff_start(ep);
+	}
+	nni_mtx_unlock(&ep->ep_mtx);
+}
+
 void
 nni_ep_list_init(nni_list *list)
 {
diff --git a/src/core/endpt.h b/src/core/endpt.h
index 2c14605c..0d2a4570 100644
--- a/src/core/endpt.h
+++ b/src/core/endpt.h
@@ -1,5 +1,6 @@
 //
 // Copyright 2016 Garrett D'Amore <garrett@damore.org>
+// Copyright 2017 Capitar IT Group BV <info@capitar.com>
 //
 // This software is supplied under the terms of the MIT License, a
 // copy of which should be located in the distribution where this
@@ -34,7 +35,6 @@ struct nni_ep {
 	int           ep_refcnt;
 	nni_mtx       ep_mtx;
 	nni_cv        ep_cv;
-	nni_pipe *    ep_pipe; // Connected pipe (dialers only)
 	nni_list      ep_pipes;
 	nni_aio       ep_acc_aio;
 	nni_aio       ep_con_aio;
@@ -62,6 +62,7 @@ extern void     nni_ep_close(nni_ep *);
 extern int      nni_ep_dial(nni_ep *, int);
 extern int      nni_ep_listen(nni_ep *, int);
 extern void     nni_ep_list_init(nni_list *);
-extern void     nni_ep_pipe_remove(nni_ep *, nni_pipe *);
+extern int nni_ep_pipe_add(nni_ep *ep, nni_pipe *);
+extern void nni_ep_pipe_remove(nni_ep *, nni_pipe *);
 
 #endif // CORE_ENDPT_H
diff --git a/src/core/pipe.c b/src/core/pipe.c
index e458a89d..c98a3243 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -120,10 +120,16 @@ nni_pipe_reap(nni_pipe *p)
 	// Transport close...
 	nni_pipe_close(p);
 
-	// Tell the protocol to stop.
-	nni_sock_pipe_stop(p->p_sock, p);
+	// Remove the pipe from the socket and the endpoint.  Note
+	// that it is in theory possible for either of these to be null
+	// if the pipe is being torn down before it is fully initialized.
+	if (p->p_ep != NULL) {
+		nni_ep_pipe_remove(p->p_ep, p);
+	}
+	if (p->p_sock != NULL) {
+		nni_sock_pipe_remove(p->p_sock, p);
+	}
 
-	// XXX: would be simpler to just do a destroy here
 	nni_pipe_destroy(p);
 }
 
@@ -208,8 +214,13 @@ nni_pipe_create(nni_ep *ep, void *tdata)
 	p->p_tran_ops  = *tran->tran_pipe;
 	p->p_tran_data = tdata;
 
-	// Attempt to initialize protocol data.
-	if ((rv = nni_sock_pipe_init(sock, p)) != 0) {
+	// Attempt to initialize sock protocol & endpoint.
+	if ((rv = nni_ep_pipe_add(ep, p)) != 0) {
+		nni_pipe_destroy(p);
+		return (rv);
+	}
+	if ((rv = nni_sock_pipe_add(sock, p)) != 0) {
+		nni_ep_pipe_remove(ep, p);
 		nni_pipe_destroy(p);
 		return (rv);
 	}
diff --git a/src/core/socket.c b/src/core/socket.c
index ecddf5dd..52732cde 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -78,12 +78,16 @@ nni_sock_rele(nni_sock *sock)
 }
 
 int
-nni_sock_pipe_init(nni_sock *sock, nni_pipe *pipe)
+nni_sock_pipe_add(nni_sock *sock, nni_pipe *pipe)
 {
 	int rv;
 
 	// Initialize protocol pipe data.
 	nni_mtx_lock(&sock->s_mx);
+	if (sock->s_closing) {
+		nni_mtx_unlock(&sock->s_mx);
+		return (NNG_ECLOSED);
+	}
 	rv = sock->s_pipe_ops.pipe_init(
 	    &pipe->p_proto_data, pipe, sock->s_data);
 	if (rv != 0) {
@@ -126,7 +130,7 @@ nni_sock_pipe_ready(nni_sock *sock, nni_pipe *pipe)
 }
 
 void
-nni_sock_pipe_stop(nni_sock *sock, nni_pipe *pipe)
+nni_sock_pipe_remove(nni_sock *sock, nni_pipe *pipe)
 {
 	void *  pdata;
 	nni_ep *ep;
@@ -140,19 +144,6 @@ nni_sock_pipe_stop(nni_sock *sock, nni_pipe *pipe)
 		return;
 	}
 
-	// Break up the relationship between the EP and the pipe.
-	if ((ep = pipe->p_ep) != NULL) {
-		nni_mtx_lock(&ep->ep_mtx);
-		// During early init, the pipe might not have this set.
-		if (nni_list_active(&ep->ep_pipes, pipe)) {
-			nni_list_remove(&ep->ep_pipes, pipe);
-		}
-		pipe->p_ep  = NULL;
-		ep->ep_pipe = NULL; // XXX: remove this soon
-		nni_cv_wake(&ep->ep_cv);
-		nni_mtx_unlock(&ep->ep_mtx);
-	}
-
 	sock->s_pipe_ops.pipe_stop(pdata);
 	if (nni_list_active(&sock->s_pipes, pipe)) {
 		nni_list_remove(&sock->s_pipes, pipe);
diff --git a/src/core/socket.h b/src/core/socket.h
index 6516de7e..d4c053c8 100644
--- a/src/core/socket.h
+++ b/src/core/socket.h
@@ -87,11 +87,11 @@ extern void        nni_sock_unnotify(nni_sock *, nni_notify *);
 
 extern void nni_sock_ep_remove(nni_sock *, nni_ep *);
 
-// nni_sock_pipe_init adds the pipe to the socket. It is called by
+// nni_sock_pipe_add adds the pipe to the socket. It is called by
 // the generic pipe creation code.
-extern int nni_sock_pipe_init(nni_sock *, nni_pipe *);
+extern int nni_sock_pipe_add(nni_sock *, nni_pipe *);
 
-extern void nni_sock_pipe_stop(nni_sock *, nni_pipe *);
+extern void nni_sock_pipe_remove(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,