Eliminate p_active, better names for pipe start and stop.

13 files changed, 91 insertions, 70 deletions

diff --git a/src/core/pipe.h b/src/core/pipe.h
index e3539e1f..55246388 100644
--- a/src/core/pipe.h
+++ b/src/core/pipe.h
@@ -28,7 +28,6 @@ struct nni_pipe {
 	nni_sock *	p_sock;
 	nni_ep *	p_ep;
 	int		p_reap;
-	int		p_active;
 	nni_mtx		p_mtx;
 	int		p_refcnt;
 };
diff --git a/src/core/protocol.h b/src/core/protocol.h
index 5f7fc4c8..745f5986 100644
--- a/src/core/protocol.h
+++ b/src/core/protocol.h
@@ -31,18 +31,19 @@ struct nni_proto_pipe_ops {
 	// pipe threads have been stopped.
 	void	(*pipe_fini)(void *);
 
-	// pipe_add is called to register a pipe with the protocol.  The
+	// pipe_start is called to register a pipe with the protocol.  The
 	// protocol can reject this, for example if another pipe is already
 	// active on a 1:1 protocol.  The protocol may not block during this,
 	// as the socket lock is held.
-	int	(*pipe_add)(void *);
+	int	(*pipe_start)(void *);
 
-	// pipe_rem is called to unregister a pipe from the protocol.
+	// pipe_stop is called to unregister a pipe from the protocol.
 	// Threads may still acccess data structures, so the protocol
 	// should not free anything yet.  This is called with the socket
 	// lock held, so the protocol may not call back into the socket, and
-	// must not block.
-	void	(*pipe_rem)(void *);
+	// must not block.  This operation must be idempotent, and may
+	// be called even if pipe_start was not.
+	void	(*pipe_stop)(void *);
 };
 
 struct nni_proto_sock_ops {
diff --git a/src/core/socket.c b/src/core/socket.c
index 42de8fee..76843174 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -163,13 +163,11 @@ nni_sock_pipe_ready(nni_sock *sock, nni_pipe *pipe)
 		return (NNG_EPROTO);
 	}
 
-	if ((rv = sock->s_pipe_ops.pipe_add(pdata)) != 0) {
+	if ((rv = sock->s_pipe_ops.pipe_start(pdata)) != 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);
 
@@ -193,10 +191,7 @@ nni_sock_pipe_closed(nni_sock *sock, nni_pipe *pipe)
 	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(pdata);
-	}
+	sock->s_pipe_ops.pipe_stop(pdata);
 
 	// Notify the endpoint that the pipe has closed.
 	if (((ep = pipe->p_ep) != NULL) && ((ep->ep_pipe == pipe))) {
@@ -380,7 +375,7 @@ nni_sock_nullop(void *arg)
 
 
 static int
-nni_sock_nulladdpipe(void *arg)
+nni_sock_nullstartpipe(void *arg)
 {
 	NNI_ARG_UNUSED(arg);
 
@@ -450,11 +445,11 @@ nni_sock_open(nni_sock **sockp, uint16_t pnum)
 	}
 	sock->s_pipe_ops = *proto->proto_pipe_ops;
 	pops = &sock->s_pipe_ops;
-	if (pops->pipe_add == NULL) {
-		pops->pipe_add = nni_sock_nulladdpipe;
+	if (pops->pipe_start == NULL) {
+		pops->pipe_start = nni_sock_nullstartpipe;
 	}
-	if (pops->pipe_rem == NULL) {
-		pops->pipe_rem = nni_sock_nullop;
+	if (pops->pipe_stop == NULL) {
+		pops->pipe_stop = nni_sock_nullop;
 	}
 
 	if (((rv = nni_mtx_init(&sock->s_mx)) != 0) ||
diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c
index b88337d1..e8428e42 100644
--- a/src/protocol/bus/bus.c
+++ b/src/protocol/bus/bus.c
@@ -165,7 +165,7 @@ nni_bus_pipe_fini(void *arg)
 
 
 static int
-nni_bus_pipe_add(void *arg)
+nni_bus_pipe_start(void *arg)
 {
 	nni_bus_pipe *ppipe = arg;
 	nni_bus_sock *psock = ppipe->psock;
@@ -182,15 +182,18 @@ nni_bus_pipe_add(void *arg)
 
 
 static void
-nni_bus_pipe_rem(void *arg)
+nni_bus_pipe_stop(void *arg)
 {
 	nni_bus_pipe *ppipe = arg;
 	nni_bus_sock *psock = ppipe->psock;
+	nni_sock *nsock = psock->nsock;
 
-	nni_list_remove(&psock->pipes, ppipe);
+	if (nni_list_active(&psock->pipes, ppipe)) {
+		nni_list_remove(&psock->pipes, ppipe);
 
-	nni_msgq_close(ppipe->sendq);
-	nni_msgq_aio_cancel(nni_sock_recvq(psock->nsock), &ppipe->aio_putq);
+		nni_msgq_close(ppipe->sendq);
+		nni_msgq_aio_cancel(nni_sock_recvq(nsock), &ppipe->aio_putq);
+	}
 }
 
 
@@ -390,8 +393,8 @@ nni_bus_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 static nni_proto_pipe_ops nni_bus_pipe_ops = {
 	.pipe_init	= nni_bus_pipe_init,
 	.pipe_fini	= nni_bus_pipe_fini,
-	.pipe_add	= nni_bus_pipe_add,
-	.pipe_rem	= nni_bus_pipe_rem,
+	.pipe_start	= nni_bus_pipe_start,
+	.pipe_stop	= nni_bus_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_bus_sock_ops = {
diff --git a/src/protocol/pair/pair.c b/src/protocol/pair/pair.c
index 65eabd87..ed5cee9f 100644
--- a/src/protocol/pair/pair.c
+++ b/src/protocol/pair/pair.c
@@ -130,7 +130,7 @@ nni_pair_pipe_fini(void *arg)
 
 
 static int
-nni_pair_pipe_add(void *arg)
+nni_pair_pipe_start(void *arg)
 {
 	nni_pair_pipe *ppipe = arg;
 	nni_pair_sock *psock = ppipe->psock;
@@ -152,7 +152,7 @@ nni_pair_pipe_add(void *arg)
 
 
 static void
-nni_pair_pipe_rem(void *arg)
+nni_pair_pipe_stop(void *arg)
 {
 	nni_pair_pipe *ppipe = arg;
 	nni_pair_sock *psock = ppipe->psock;
@@ -277,8 +277,8 @@ nni_pair_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 static nni_proto_pipe_ops nni_pair_pipe_ops = {
 	.pipe_init	= nni_pair_pipe_init,
 	.pipe_fini	= nni_pair_pipe_fini,
-	.pipe_add	= nni_pair_pipe_add,
-	.pipe_rem	= nni_pair_pipe_rem,
+	.pipe_start	= nni_pair_pipe_start,
+	.pipe_stop	= nni_pair_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_pair_sock_ops = {
diff --git a/src/protocol/pipeline/pull.c b/src/protocol/pipeline/pull.c
index 622bd958..b569253d 100644
--- a/src/protocol/pipeline/pull.c
+++ b/src/protocol/pipeline/pull.c
@@ -229,8 +229,8 @@ nni_pull_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 static nni_proto_pipe_ops nni_pull_pipe_ops = {
 	.pipe_init	= nni_pull_pipe_init,
 	.pipe_fini	= nni_pull_pipe_fini,
-	.pipe_add	= nni_pull_pipe_start,
-	.pipe_rem	= nni_pull_pipe_stop,
+	.pipe_start	= nni_pull_pipe_start,
+	.pipe_stop	= nni_pull_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_pull_sock_ops = {
diff --git a/src/protocol/pipeline/push.c b/src/protocol/pipeline/push.c
index d2d8363a..9691d8d8 100644
--- a/src/protocol/pipeline/push.c
+++ b/src/protocol/pipeline/push.c
@@ -116,7 +116,7 @@ nni_push_pipe_fini(void *arg)
 
 
 static int
-nni_push_pipe_add(void *arg)
+nni_push_pipe_start(void *arg)
 {
 	nni_push_pipe *pp = arg;
 	nni_push_sock *push = pp->push;
@@ -139,7 +139,7 @@ nni_push_pipe_add(void *arg)
 
 
 static void
-nni_push_pipe_rem(void *arg)
+nni_push_pipe_stop(void *arg)
 {
 	nni_push_pipe *pp = arg;
 	nni_push_sock *push = pp->push;
@@ -243,8 +243,8 @@ nni_push_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 static nni_proto_pipe_ops nni_push_pipe_ops = {
 	.pipe_init	= nni_push_pipe_init,
 	.pipe_fini	= nni_push_pipe_fini,
-	.pipe_add	= nni_push_pipe_add,
-	.pipe_rem	= nni_push_pipe_rem,
+	.pipe_start	= nni_push_pipe_start,
+	.pipe_stop	= nni_push_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_push_sock_ops = {
diff --git a/src/protocol/pubsub/pub.c b/src/protocol/pubsub/pub.c
index ad0b9580..a37fe991 100644
--- a/src/protocol/pubsub/pub.c
+++ b/src/protocol/pubsub/pub.c
@@ -147,7 +147,7 @@ nni_pub_pipe_fini(void *arg)
 
 
 static int
-nni_pub_pipe_add(void *arg)
+nni_pub_pipe_start(void *arg)
 {
 	nni_pub_pipe *pp = arg;
 	nni_pub_sock *pub = pp->pub;
@@ -168,13 +168,15 @@ nni_pub_pipe_add(void *arg)
 
 
 static void
-nni_pub_pipe_rem(void *arg)
+nni_pub_pipe_stop(void *arg)
 {
 	nni_pub_pipe *pp = arg;
 	nni_pub_sock *pub = pp->pub;
 
-	nni_list_remove(&pub->pipes, pp);
-	nni_msgq_close(pp->sendq);
+	if (nni_list_active(&pub->pipes, pp)) {
+		nni_list_remove(&pub->pipes, pp);
+		nni_msgq_close(pp->sendq);
+	}
 }
 
 
@@ -314,8 +316,8 @@ nni_pub_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 static nni_proto_pipe_ops nni_pub_pipe_ops = {
 	.pipe_init	= nni_pub_pipe_init,
 	.pipe_fini	= nni_pub_pipe_fini,
-	.pipe_add	= nni_pub_pipe_add,
-	.pipe_rem	= nni_pub_pipe_rem,
+	.pipe_start	= nni_pub_pipe_start,
+	.pipe_stop	= nni_pub_pipe_stop,
 };
 
 nni_proto_sock_ops nni_pub_sock_ops = {
diff --git a/src/protocol/pubsub/sub.c b/src/protocol/pubsub/sub.c
index 6fae1962..ecfbf540 100644
--- a/src/protocol/pubsub/sub.c
+++ b/src/protocol/pubsub/sub.c
@@ -115,7 +115,7 @@ nni_sub_pipe_fini(void *arg)
 
 
 static int
-nni_sub_pipe_add(void *arg)
+nni_sub_pipe_start(void *arg)
 {
 	nni_sub_pipe *sp = arg;
 
@@ -126,7 +126,7 @@ nni_sub_pipe_add(void *arg)
 
 
 static void
-nni_sub_pipe_close(void *arg)
+nni_sub_pipe_stop(void *arg)
 {
 	nni_sub_pipe *sp = arg;
 
@@ -340,8 +340,8 @@ nni_sub_sock_rfilter(void *arg, nni_msg *msg)
 static nni_proto_pipe_ops nni_sub_pipe_ops = {
 	.pipe_init	= nni_sub_pipe_init,
 	.pipe_fini	= nni_sub_pipe_fini,
-	.pipe_add	= nni_sub_pipe_add,
-	.pipe_rem	= nni_sub_pipe_close,
+	.pipe_start	= nni_sub_pipe_start,
+	.pipe_stop	= nni_sub_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_sub_sock_ops = {
diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index 55a6855c..ec5b77cf 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -49,6 +49,7 @@ struct nni_rep_pipe {
 	nni_aio		aio_send;
 	nni_aio		aio_recv;
 	nni_aio		aio_putq;
+	int		running;
 };
 
 static int
@@ -168,7 +169,7 @@ nni_rep_pipe_fini(void *arg)
 
 
 static int
-nni_rep_pipe_add(void *arg)
+nni_rep_pipe_start(void *arg)
 {
 	nni_rep_pipe *rp = arg;
 	nni_rep_sock *rep = rp->rep;
@@ -183,19 +184,23 @@ nni_rep_pipe_add(void *arg)
 	nni_msgq_aio_get(rp->sendq, &rp->aio_getq);
 	nni_pipe_incref(rp->pipe);
 	nni_pipe_aio_recv(rp->pipe, &rp->aio_recv);
+	rp->running = 1;
 	return (0);
 }
 
 
 static void
-nni_rep_pipe_rem(void *arg)
+nni_rep_pipe_stop(void *arg)
 {
 	nni_rep_pipe *rp = arg;
 	nni_rep_sock *rep = rp->rep;
 
-	nni_msgq_close(rp->sendq);
-	nni_msgq_aio_cancel(rep->urq, &rp->aio_putq);
-	nni_idhash_remove(&rep->pipes, nni_pipe_id(rp->pipe));
+	if (rp->running) {
+		rp->running = 0;
+		nni_msgq_close(rp->sendq);
+		nni_msgq_aio_cancel(rep->urq, &rp->aio_putq);
+		nni_idhash_remove(&rep->pipes, nni_pipe_id(rp->pipe));
+	}
 }
 
 
@@ -488,8 +493,8 @@ nni_rep_sock_rfilter(void *arg, nni_msg *msg)
 static nni_proto_pipe_ops nni_rep_pipe_ops = {
 	.pipe_init	= nni_rep_pipe_init,
 	.pipe_fini	= nni_rep_pipe_fini,
-	.pipe_add	= nni_rep_pipe_add,
-	.pipe_rem	= nni_rep_pipe_rem,
+	.pipe_start	= nni_rep_pipe_start,
+	.pipe_stop	= nni_rep_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_rep_sock_ops = {
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index f37913e6..af4dbd42 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -56,6 +56,7 @@ struct nni_req_pipe {
 	nni_aio		aio_sendcooked;         // cooked mode only
 	nni_aio		aio_recv;
 	nni_aio		aio_putq;
+	int		running;
 };
 
 static void nni_req_resender(void *);
@@ -173,7 +174,7 @@ nni_req_pipe_fini(void *arg)
 
 
 static int
-nni_req_pipe_add(void *arg)
+nni_req_pipe_start(void *arg)
 {
 	nni_req_pipe *rp = arg;
 	nni_req_sock *req = rp->req;
@@ -190,16 +191,22 @@ nni_req_pipe_add(void *arg)
 	nni_msgq_aio_get(req->uwq, &rp->aio_getq);
 	nni_pipe_incref(rp->pipe);
 	nni_pipe_aio_recv(rp->pipe, &rp->aio_recv);
+	rp->running = 1;
 	return (0);
 }
 
 
 static void
-nni_req_pipe_rem(void *arg)
+nni_req_pipe_stop(void *arg)
 {
 	nni_req_pipe *rp = arg;
 	nni_req_sock *req = rp->req;
 
+	if (!rp->running) {
+		return;
+	}
+	rp->running = 0;
+
 	// This removes the node from either busypipes or readypipes.
 	// It doesn't much matter which.
 	nni_list_remove(&req->readypipes, rp);
@@ -565,8 +572,8 @@ nni_req_sock_rfilter(void *arg, nni_msg *msg)
 static nni_proto_pipe_ops nni_req_pipe_ops = {
 	.pipe_init	= nni_req_pipe_init,
 	.pipe_fini	= nni_req_pipe_fini,
-	.pipe_add	= nni_req_pipe_add,
-	.pipe_rem	= nni_req_pipe_rem,
+	.pipe_start	= nni_req_pipe_start,
+	.pipe_stop	= nni_req_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_req_sock_ops = {
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index a70b08e6..8fe12e19 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -48,6 +48,7 @@ struct nni_resp_pipe {
 	nni_aio		aio_putq;
 	nni_aio		aio_send;
 	nni_aio		aio_recv;
+	int		running;
 };
 
 static int
@@ -173,7 +174,7 @@ nni_resp_pipe_fini(void *arg)
 
 
 static int
-nni_resp_pipe_add(void *arg)
+nni_resp_pipe_start(void *arg)
 {
 	nni_resp_pipe *ppipe = arg;
 	nni_resp_sock *psock = ppipe->psock;
@@ -186,20 +187,24 @@ nni_resp_pipe_add(void *arg)
 
 	nni_pipe_incref(ppipe->npipe);
 	nni_msgq_aio_get(ppipe->sendq, &ppipe->aio_getq);
+	ppipe->running = 1;
 	return (rv);
 }
 
 
 static void
-nni_resp_pipe_rem(void *arg)
+nni_resp_pipe_stop(void *arg)
 {
 	nni_resp_pipe *ppipe = arg;
 	nni_resp_sock *psock = ppipe->psock;
 
-	nni_idhash_remove(&psock->pipes, nni_pipe_id(ppipe->npipe));
+	if (ppipe->running) {
+		ppipe->running = 0;
+		nni_idhash_remove(&psock->pipes, nni_pipe_id(ppipe->npipe));
 
-	nni_msgq_close(ppipe->sendq);
-	nni_msgq_aio_cancel(psock->urq, &ppipe->aio_putq);
+		nni_msgq_close(ppipe->sendq);
+		nni_msgq_aio_cancel(psock->urq, &ppipe->aio_putq);
+	}
 }
 
 
@@ -488,8 +493,8 @@ nni_resp_sock_rfilter(void *arg, nni_msg *msg)
 static nni_proto_pipe_ops nni_resp_pipe_ops = {
 	.pipe_init	= nni_resp_pipe_init,
 	.pipe_fini	= nni_resp_pipe_fini,
-	.pipe_add	= nni_resp_pipe_add,
-	.pipe_rem	= nni_resp_pipe_rem,
+	.pipe_start	= nni_resp_pipe_start,
+	.pipe_stop	= nni_resp_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_resp_sock_ops = {
diff --git a/src/protocol/survey/survey.c b/src/protocol/survey/survey.c
index 4fa89af3..0edcfef2 100644
--- a/src/protocol/survey/survey.c
+++ b/src/protocol/survey/survey.c
@@ -51,6 +51,7 @@ struct nni_surv_pipe {
 	nni_aio		aio_putq;
 	nni_aio		aio_send;
 	nni_aio		aio_recv;
+	int		running;
 };
 
 static int
@@ -168,7 +169,7 @@ failed:
 
 
 static int
-nni_surv_pipe_add(void *arg)
+nni_surv_pipe_start(void *arg)
 {
 	nni_surv_pipe *ppipe = arg;
 	nni_surv_sock *psock = ppipe->psock;
@@ -180,19 +181,22 @@ nni_surv_pipe_add(void *arg)
 
 	nni_pipe_incref(ppipe->npipe);
 	nni_pipe_aio_recv(ppipe->npipe, &ppipe->aio_recv);
+	ppipe->running = 1;
 	return (0);
 }
 
 
 static void
-nni_surv_pipe_rem(void *arg)
+nni_surv_pipe_stop(void *arg)
 {
 	nni_surv_pipe *ppipe = arg;
 	nni_surv_sock *psock = ppipe->psock;
 
-	nni_list_remove(&psock->pipes, ppipe);
-	nni_msgq_close(ppipe->sendq);
-	nni_msgq_aio_cancel(psock->urq, &ppipe->aio_putq);
+	if (ppipe->running) {
+		nni_list_remove(&psock->pipes, ppipe);
+		nni_msgq_close(ppipe->sendq);
+		nni_msgq_aio_cancel(psock->urq, &ppipe->aio_putq);
+	}
 }
 
 
@@ -460,8 +464,8 @@ nni_surv_sock_rfilter(void *arg, nni_msg *msg)
 static nni_proto_pipe_ops nni_surv_pipe_ops = {
 	.pipe_init	= nni_surv_pipe_init,
 	.pipe_fini	= nni_surv_pipe_fini,
-	.pipe_add	= nni_surv_pipe_add,
-	.pipe_rem	= nni_surv_pipe_rem,
+	.pipe_start	= nni_surv_pipe_start,
+	.pipe_stop	= nni_surv_pipe_stop,
 };
 
 static nni_proto_sock_ops nni_surv_sock_ops = {