Substantial fixes for listen & dialers.

At this point listening and dialing operations appear to function properly.
As part of this I had to break the close logic up since otherwise we had a
loop trying to reap a thread from itself.  So there is now a separate reaper
thread for pipes per-socket.  I also changed lists to be a bit more rigid,
and allocations now zero memory initially.  (We had bugs due to uninitialized
memory, and rather than hunt them all down, lets just init them to sane zero
values.)

1 files changed, 28 insertions, 24 deletions

diff --git a/src/protocol/pair/pair.c b/src/protocol/pair/pair.c
index 692f4b0e..98995186 100644
--- a/src/protocol/pair/pair.c
+++ b/src/protocol/pair/pair.c
@@ -16,27 +16,30 @@
 // While a peer is connected to the server, all other peer connection
 // attempts are discarded.
 
+typedef struct nni_pair_pipe	nni_pair_pipe;
+typedef struct nni_pair_sock	nni_pair_sock;
+
 // An nni_pair_sock is our per-socket protocol private structure.
-typedef struct nni_pair_sock {
+struct nni_pair_sock {
 	nni_socket *	sock;
-	nni_pipe *	pipe;
+	nni_pair_pipe * pipe;
 	nni_mutex	mx;
 	nni_msgqueue *	uwq;
 	nni_msgqueue *	urq;
-} nni_pair_sock;
+};
 
 // An nni_pair_pipe is our per-pipe protocol private structure.  We keep
 // one of these even though in theory we'd only have a single underlying
 // pipe.  The separate data structure is more like other protocols that do
 // manage multiple pipes.
-typedef struct nni_pair_pipe {
+struct nni_pair_pipe {
 	nni_pipe *	pipe;
 	nni_pair_sock * pair;
 	int		good;
 	nni_thread *	sthr;
 	nni_thread *	rthr;
 	int		sigclose;
-} nni_pair_pipe;
+};
 
 static void nni_pair_receiver(void *);
 static void nni_pair_sender(void *);
@@ -55,6 +58,7 @@ nni_pair_create(void **pairp, nni_socket *sock)
 		return (rv);
 	}
 	pair->sock = sock;
+	pair->pipe = NULL;
 	pair->uwq = nni_socket_sendq(sock);
 	pair->urq = nni_socket_recvq(sock);
 	*pairp = pair;
@@ -77,7 +81,7 @@ nni_pair_destroy(void *arg)
 
 
 static int
-nni_pair_add_pipe(void *arg, nni_pipe *pipe)
+nni_pair_add_pipe(void *arg, nni_pipe *pipe, void **datap)
 {
 	nni_pair_sock *pair = arg;
 	nni_pair_pipe *pp;
@@ -89,6 +93,7 @@ nni_pair_add_pipe(void *arg, nni_pipe *pipe)
 	pp->sigclose = 0;
 	pp->sthr = NULL;
 	pp->rthr = NULL;
+	pp->pair = pair;
 
 	nni_mutex_enter(&pair->mx);
 	if (pair->pipe != NULL) {
@@ -106,31 +111,33 @@ nni_pair_add_pipe(void *arg, nni_pipe *pipe)
 		return (rv);
 	}
 	pp->good = 1;
-	pair->pipe = pipe;
+	pair->pipe = pp;
+	*datap = pp;
 	nni_mutex_exit(&pair->mx);
-	return (NNG_EINVAL);
+	return (0);
 }
 
 
-static int
-nni_pair_rem_pipe(void *arg, nni_pipe *pipe)
+static void
+nni_pair_rem_pipe(void *arg, void *data)
 {
-	nni_pair_pipe *pp = arg;
-	nni_pair_sock *pair = pp->pair;
+	nni_pair_sock *pair = arg;
+	nni_pair_pipe *pp = data;
 
-	if (pp->sthr) {
-		(void) nni_thread_reap(pp->sthr);
-	}
-	if (pp->rthr) {
-		(void) nni_thread_reap(pp->rthr);
-	}
 	nni_mutex_enter(&pair->mx);
-	if (pair->pipe != pipe) {
+	if (pair->pipe != pp) {
 		nni_mutex_exit(&pair->mx);
-		return (NNG_EINVAL);
+		return;
 	}
+	pair->pipe = NULL;
 	nni_mutex_exit(&pair->mx);
-	return (NNG_EINVAL);
+
+	if (pp->sthr != NULL) {
+		(void) nni_thread_reap(pp->sthr);
+	}
+	if (pp->rthr != NULL) {
+		(void) nni_thread_reap(pp->rthr);
+	}
 }
 
 
@@ -166,7 +173,6 @@ nni_pair_sender(void *arg)
 	}
 	nni_msgqueue_signal(urq, &pp->sigclose);
 	nni_pipe_close(pipe);
-	nni_socket_rem_pipe(pair->sock, pipe);
 }
 
 
@@ -187,7 +193,6 @@ nni_pair_receiver(void *arg)
 		return;
 	}
 	nni_mutex_exit(&pair->mx);
-
 	for (;;) {
 		rv = nni_pipe_recv(pipe, &msg);
 		if (rv != 0) {
@@ -201,7 +206,6 @@ nni_pair_receiver(void *arg)
 	}
 	nni_msgqueue_signal(uwq, &pp->sigclose);
 	nni_pipe_close(pipe);
-	nni_socket_rem_pipe(pair->sock, pipe);
 }