Eliminate p_active, better names for pipe start and stop.

2 files changed, 23 insertions, 11 deletions

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 = {