Eliminate p_active, better names for pipe start and stop.

2 files changed, 23 insertions, 14 deletions

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