Pipe ID race on close (pipe IDs are zero at close).

This should address some of the errors we've seen.  Additionally,
the scalability test was a bit brittle due to too-short timeouts.

2 files changed, 13 insertions, 10 deletions

diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index a7295eec..5a5f6bff 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -45,6 +45,7 @@ struct nni_rep_pipe {
 	nni_pipe *	pipe;
 	nni_rep_sock *	rep;
 	nni_msgq *	sendq;
+	uint32_t	id;             // we have to save it
 	nni_aio		aio_getq;
 	nni_aio		aio_send;
 	nni_aio		aio_recv;
@@ -175,7 +176,8 @@ nni_rep_pipe_start(void *arg)
 	nni_rep_sock *rep = rp->rep;
 	int rv;
 
-	rv = nni_idhash_insert(&rep->pipes, nni_pipe_id(rp->pipe), rp);
+	rp->id = nni_pipe_id(rp->pipe);
+	rv = nni_idhash_insert(&rep->pipes, rp->id, rp);
 	if (rv != 0) {
 		return (rv);
 	}
@@ -199,7 +201,7 @@ nni_rep_pipe_stop(void *arg)
 		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));
+		nni_idhash_remove(&rep->pipes, rp->id);
 	}
 }
 
@@ -302,7 +304,6 @@ nni_rep_pipe_recv_cb(void *arg)
 	nni_msg *msg;
 	int rv;
 	uint8_t idbuf[4];
-	uint32_t id;
 	uint8_t *body;
 	int hops;
 
@@ -312,8 +313,7 @@ nni_rep_pipe_recv_cb(void *arg)
 		return;
 	}
 
-	id = nni_pipe_id(rp->pipe);
-	NNI_PUT32(idbuf, id);
+	NNI_PUT32(idbuf, rp->id);
 
 	msg = rp->aio_recv.a_msg;
 	rp->aio_recv.a_msg = NULL;
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index 8fe12e19..8e36cbb0 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -43,6 +43,7 @@ struct nni_resp_sock {
 struct nni_resp_pipe {
 	nni_pipe *	npipe;
 	nni_resp_sock * psock;
+	uint32_t	id;
 	nni_msgq *	sendq;
 	nni_aio		aio_getq;
 	nni_aio		aio_putq;
@@ -180,7 +181,11 @@ nni_resp_pipe_start(void *arg)
 	nni_resp_sock *psock = ppipe->psock;
 	int rv;
 
-	rv = nni_idhash_insert(&psock->pipes, nni_pipe_id(ppipe->npipe), ppipe);
+	ppipe->id = nni_pipe_id(ppipe->npipe);
+	rv = nni_idhash_insert(&psock->pipes, ppipe->id, ppipe);
+	if (rv != 0) {
+		return (rv);
+	}
 
 	nni_pipe_incref(ppipe->npipe);
 	nni_pipe_aio_recv(ppipe->npipe, &ppipe->aio_recv);
@@ -200,7 +205,7 @@ nni_resp_pipe_stop(void *arg)
 
 	if (ppipe->running) {
 		ppipe->running = 0;
-		nni_idhash_remove(&psock->pipes, nni_pipe_id(ppipe->npipe));
+		nni_idhash_remove(&psock->pipes, ppipe->id);
 
 		nni_msgq_close(ppipe->sendq);
 		nni_msgq_aio_cancel(psock->urq, &ppipe->aio_putq);
@@ -300,7 +305,6 @@ nni_resp_recv_cb(void *arg)
 	nni_msgq *urq = nni_sock_recvq(psock->nsock);
 	nni_msg *msg;
 	uint8_t idbuf[4];
-	uint32_t id;
 	int hops;
 	int rv;
 
@@ -308,8 +312,7 @@ nni_resp_recv_cb(void *arg)
 		goto error;
 	}
 
-	id = nni_pipe_id(ppipe->npipe);
-	NNI_PUT32(idbuf, id);
+	NNI_PUT32(idbuf, ppipe->id);
 
 	msg = ppipe->aio_recv.a_msg;
 	ppipe->aio_recv.a_msg = NULL;