fixes #423 desire tunable stress time and pressure in stress tests

Use -p STRESSPRESSURE=<count> (default 32) and -p STRESSTIME=<sec>
to affect stress run.  Pressure should be a number [2,x] where x
is determined by the number of threads and file descriptors your
platform can handle.  Modern platforms should be able to handle
at least 100.

2 files changed, 61 insertions, 20 deletions

diff --git a/tests/multistress.c b/tests/multistress.c
index d88d0af5..62838a73 100644
--- a/tests/multistress.c
+++ b/tests/multistress.c
@@ -154,13 +154,17 @@ req_client(void *arg)
 
 		nng_msleep(rand() % 10);
 
-		if (((rv = nng_msg_alloc(&msg, 0)) != 0) ||
-		    ((rv = nng_msg_append(msg, buf, strlen(buf) + 1)) != 0)) {
+		if ((rv = nng_msg_alloc(&msg, 0)) != 0) {
 			error(c, "alloc fail", rv);
 			return;
 		}
-
+		if ((rv = nng_msg_append(msg, buf, strlen(buf) + 1)) != 0) {
+			nng_msg_free(msg);
+			error(c, "append fail", rv);
+			return;
+		}
 		if ((rv = nng_sendmsg(req, msg, 0)) != 0) {
+			nng_msg_free(msg);
 			error(c, "sendmsg", rv);
 			return;
 		}
@@ -247,11 +251,13 @@ pair0_bouncer(void *arg)
 
 		r = rand();
 		if ((rv = nng_msg_append(msg, &r, sizeof(r))) != 0) {
+			nng_msg_free(msg);
 			error(c, "msg_append", rv);
 			return;
 		}
 
 		if ((rv = nng_sendmsg(c->socket, msg, 0)) != 0) {
+			nng_msg_free(msg);
 			error(c, "sendmsg", rv);
 			return;
 		}
@@ -325,11 +331,13 @@ bus0_bouncer(void *arg)
 
 		r = rand();
 		if ((rv = nng_msg_append(msg, &r, sizeof(r))) != 0) {
+			nng_msg_free(msg);
 			error(c, "msg_append", rv);
 			return;
 		}
 
 		if ((rv = nng_sendmsg(c->socket, msg, 0)) != 0) {
+			nng_msg_free(msg);
 			error(c, "sendmsg", rv);
 			return;
 		}
@@ -402,11 +410,13 @@ pub0_sender(void *arg)
 		}
 
 		if ((rv = nng_msg_append(msg, "SUB", 4)) != 0) {
+			nng_msg_free(msg);
 			error(c, "msg_append", rv);
 			return;
 		}
 
 		if ((rv = nng_sendmsg(c->socket, msg, 0)) != 0) {
+			nng_msg_free(msg);
 			error(c, "sendmsg", rv);
 			return;
 		}
@@ -501,11 +511,13 @@ pipeline0_pusher(void *arg)
 		}
 
 		if ((rv = nng_msg_append(msg, "PUSH", 5)) != 0) {
+			nng_msg_free(msg);
 			error(c, "msg_append", rv);
 			return;
 		}
 
 		if ((rv = nng_sendmsg(c->socket, msg, 0)) != 0) {
+			nng_msg_free(msg);
 			error(c, "sendmsg", rv);
 			return;
 		}
@@ -539,6 +551,8 @@ pipeline0_test(int ntests)
 	char       addr[NNG_MAXADDRLEN];
 	int        rv;
 
+	atexit(nng_fini);
+
 	if (ntests < 2) {
 		// Need a client *and* a server.
 		return;
@@ -581,19 +595,30 @@ pipeline0_test(int ntests)
 }
 
 Main({
-	int i;
+	int   i;
+	char *str;
+	int   tmo;
+
+	atexit(nng_fini);
 
 	// Each run should truly be random.
 	srand((int) time(NULL));
 
+	if (((str = ConveyGetEnv("STRESSTIME")) == NULL) ||
+	    ((tmo = atoi(str)) < 1)) {
+		tmo = 30;
+	}
+	// We have to keep this relatively low by default because some
+	// platforms don't support large numbers of threads.
+	if (((str = ConveyGetEnv("STRESSPRESSURE")) == NULL) ||
+	    ((ncases = atoi(str)) < 1)) {
+		ncases = 32;
+	}
+
 	// Reduce the likelihood of address in use conflicts between
 	// subsequent runs.
 	next_port += (rand() % 100) * 100;
 
-	// We have to keep this relatively low because some platforms
-	// don't support large numbers of threads.
-	ncases = 32;
-
 	i = ncases;
 
 	cases = calloc(ncases, sizeof(test_case));
@@ -627,8 +652,8 @@ Main({
 		i -= x;
 	}
 
-	dprintf("WAITING for 30 sec...\n");
-	nng_msleep(30000); // sleep 30 sec
+	dprintf("WAITING for %d sec...\n", tmo);
+	nng_msleep(tmo * 1000); // sleep 30 sec
 	nng_closeall();
 
 	Test("MultiProtocol/Transport Stress", {
diff --git a/tests/reqstress.c b/tests/reqstress.c
index 81249946..f2ee502d 100644
--- a/tests/reqstress.c
+++ b/tests/reqstress.c
@@ -150,13 +150,17 @@ req_client(void *arg)
 
 		nng_msleep(rand() % 10);
 
-		if (((rv = nng_msg_alloc(&msg, 0)) != 0) ||
-		    ((rv = nng_msg_append(msg, buf, strlen(buf) + 1)) != 0)) {
+		if ((rv = nng_msg_alloc(&msg, 0)) != 0) {
 			error(c, "alloc fail", rv);
 			return;
 		}
-
+		if ((rv = nng_msg_append(msg, buf, strlen(buf) + 1)) != 0) {
+			nng_msg_free(msg);
+			error(c, "append fail", rv);
+			return;
+		}
 		if ((rv = nng_sendmsg(req, msg, 0)) != 0) {
+			nng_msg_free(msg);
 			error(c, "sendmsg", rv);
 			return;
 		}
@@ -169,6 +173,7 @@ req_client(void *arg)
 
 		if (strcmp(nng_msg_body(msg), buf) != 0) {
 			error(c, "mismatched message", NNG_EINTERNAL);
+			nng_msg_free(msg);
 			return;
 		}
 
@@ -226,7 +231,22 @@ reqrep_test(int ntests)
 }
 
 Main({
-	int i;
+	int   i;
+	int   tmo;
+	char *str;
+
+	atexit(nng_fini);
+
+	if (((str = ConveyGetEnv("STRESSTIME")) == NULL) ||
+	    ((tmo = atoi(str)) < 1)) {
+		tmo = 30;
+	}
+	// We have to keep this relatively low by default because some
+	// platforms don't support large numbers of threads.
+	if (((str = ConveyGetEnv("STRESSPRESSURE")) == NULL) ||
+	    ((ncases = atoi(str)) < 1)) {
+		ncases = 32;
+	}
 
 	// Each run should truly be random.
 	srand((int) time(NULL));
@@ -235,10 +255,6 @@ Main({
 	// subsequent runs.
 	next_port += (rand() % 100) * 100;
 
-	// We have to keep this relatively low because some platforms
-	// don't support large numbers of threads.
-	ncases = 32;
-
 	i = ncases;
 
 	cases = calloc(ncases, sizeof(test_case));
@@ -251,8 +267,8 @@ Main({
 		i -= x;
 	}
 
-	dprintf("WAITING for 30 sec...\n");
-	nng_msleep(30000); // sleep 30 sec
+	dprintf("WAITING for %d sec...\n", tmo);
+	nng_msleep(tmo * 1000); // sleep 30 sec
 	nng_closeall();
 
 	Test("Req/Rep Stress", {