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.

3 files changed, 15 insertions, 23 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;
diff --git a/tests/scalability.c b/tests/scalability.c
index 16f4ba8c..1d8aeda3 100644
--- a/tests/scalability.c
+++ b/tests/scalability.c
@@ -39,7 +39,7 @@ client(void *arg)
 		return;
 	}
 
-	timeo = 40000; // 4 seconds
+	timeo = 4000000; // 4 seconds
 	if (((rv = nng_setopt(s, NNG_OPT_RCVTIMEO, &timeo, sizeof (timeo))) != 0) ||
 	    ((rv = nng_setopt(s, NNG_OPT_SNDTIMEO, &timeo, sizeof (timeo))) != 0)) {
 		*result = rv;
@@ -49,18 +49,7 @@ client(void *arg)
 
 	// Sleep for up to a second before issuing requests to avoid saturating
 	// the CPU with bazillions of requests at the same time.
-
-	if ((rv = nng_msg_alloc(&msg, 0)) != 0) {
-		*result = rv;
-		nng_close(s);
-		return;
-	}
-	if ((rv = nng_msg_append(msg, "abc", strlen("abc"))) != 0) {
-		*result = rv;
-		nng_msg_free(msg);
-		nng_close(s);
-		return;
-	}
+	nng_usleep(1000000);
 
 	for (i = 0; i < count; i++) {
 		// Sleep for up to a 1ms before issuing requests to