From 3ca4e96aacce13e12652393df6d4724bada84c25 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Fri, 29 Nov 2024 12:24:53 -0500 Subject: tests: req stress test converted to NUTS --- src/sp/protocol/reqrep0/CMakeLists.txt | 1 + src/sp/protocol/reqrep0/reqstress_test.c | 336 +++++++++++++++++++++++++++++++ tests/CMakeLists.txt | 2 - tests/reqstress.c | 336 ------------------------------- 4 files changed, 337 insertions(+), 338 deletions(-) create mode 100644 src/sp/protocol/reqrep0/reqstress_test.c delete mode 100644 tests/reqstress.c diff --git a/src/sp/protocol/reqrep0/CMakeLists.txt b/src/sp/protocol/reqrep0/CMakeLists.txt index a3cecfd0..b5d44759 100644 --- a/src/sp/protocol/reqrep0/CMakeLists.txt +++ b/src/sp/protocol/reqrep0/CMakeLists.txt @@ -23,3 +23,4 @@ nng_test(req_test) nng_test(rep_test) nng_test(xrep_test) nng_test(xreq_test) +nng_test(reqstress_test) diff --git a/src/sp/protocol/reqrep0/reqstress_test.c b/src/sp/protocol/reqrep0/reqstress_test.c new file mode 100644 index 00000000..cd0004a0 --- /dev/null +++ b/src/sp/protocol/reqrep0/reqstress_test.c @@ -0,0 +1,336 @@ +// +// Copyright 2024 Staysail Systems, Inc. +// Copyright 2018 Capitar IT Group BV +// +// This software is supplied under the terms of the MIT License, a +// copy of which should be located in the distribution where this +// file was obtained (LICENSE.txt). A copy of the license may also be +// found online at https://opensource.org/licenses/MIT. +// + +#include +#include +#include + +#include +#include +#include +#include + +#include + +#ifdef NDEBUG +#define dprintf(...) +#else +#define dprintf printf +#endif + +static int next_port = 20000; // port number kind of. + +char tcp4_template[] = "tcp://127.0.0.1:%d"; +char tcp6_template[] = "tcp://[::1]:%d"; +char inproc_template[] = "inproc://nng_reqstress_%d"; +char ipc_template[] = "ipc:///tmp/nng_reqstress_%d"; +char ws_template[] = "ws://127.0.0.1:%d/nng_reqstress"; + +char *templates[] = { + tcp4_template, +// It would be nice to test TCPv6, but CI doesn't support it. +// Outside of CI, it does seem to work though. +#ifdef NNG_TEST_TCPV6 + tcp6_template, +#endif + inproc_template, + ipc_template, +#ifdef NNG_TRANSPORT_WS + ws_template, +#endif +}; + +#define NTEMPLATES (sizeof(templates) / sizeof(templates[0])) + +char **addresses; +int naddresses; +int allocaddrs; + +typedef struct test_case { + nng_socket socket; + const char *name; + int nrecv; + int nsend; + int nfail; + nng_aio *recv_aio; + nng_aio *send_aio; + nng_aio *time_aio; + char buf[32]; +} test_case; + +static test_case *cases; +int ncases; +int curcase; + +void +fatal(const char *msg, int rv) +{ + fprintf(stderr, "%s: %s\n", msg, nng_strerror(rv)); + abort(); +} + +void +error(test_case *c, const char *msg, int rv) +{ + if ((rv == NNG_ECLOSED) || (rv == NNG_ECANCELED)) { + return; + } + fprintf( + stderr, "%s: %s: %s (%d)\n", c->name, msg, nng_strerror(rv), rv); + c->nfail++; +} + +// Get a random address -- TCP, IP, whatever. +char * +getaddr(char *buf) +{ + int i; + i = rand() % NTEMPLATES; + + snprintf(buf, NNG_MAXADDRLEN, templates[i], next_port++); + return (buf); +} + +// Request/Reply test. For this test, we open a server socket, +// and bind it to each protocol. Then we run a bunch of clients +// against it. + +// REP server implemented via callbacks. +static void +rep_recv_cb(void *arg) +{ + test_case *c = arg; + int rv; + + if ((rv = nng_aio_result(c->recv_aio)) != 0) { + error(c, "recv", rv); + return; + } + c->nrecv++; + nng_aio_set_msg(c->send_aio, nng_aio_get_msg(c->recv_aio)); + nng_aio_set_msg(c->recv_aio, NULL); + nng_send_aio(c->socket, c->send_aio); +} + +static void +rep_send_cb(void *arg) +{ + test_case *c = arg; + int rv; + + if ((rv = nng_aio_result(c->send_aio)) != 0) { + error(c, "send", rv); + nng_msg_free(nng_aio_get_msg(c->send_aio)); + nng_aio_set_msg(c->send_aio, NULL); + return; + } + c->nsend++; + nng_recv_aio(c->socket, c->recv_aio); +} + +static void +req_time_cb(void *arg) +{ + test_case *c = arg; + nng_msg *msg = NULL; + int rv; + + if ((rv = nng_aio_result(c->time_aio)) != 0) { + error(c, "sleep", rv); + return; + } + (void) snprintf( + c->buf, sizeof(c->buf), "%u-%d", c->socket.id, c->nsend); + if (((rv = nng_msg_alloc(&msg, 0)) != 0) || + ((rv = nng_msg_append(msg, c->buf, strlen(c->buf) + 1)) != 0)) { + error(c, "alloc", rv); + nng_msg_free(msg); + return; + } + nng_aio_set_msg(c->send_aio, msg); + nng_send_aio(c->socket, c->send_aio); +} + +static void +req_recv_cb(void *arg) +{ + test_case *c = arg; + nng_msg *msg = NULL; + int rv; + + if ((rv = nng_aio_result(c->recv_aio)) != 0) { + error(c, "recv", rv); + return; + } + + msg = nng_aio_get_msg(c->recv_aio); + if ((nng_msg_len(msg) != (strlen(c->buf) + 1)) || + (strcmp(c->buf, nng_msg_body(msg)) != 0)) { + error(c, "msg mismatch", rv); + nng_msg_free(msg); + return; + } + + nng_msg_free(msg); + memset(c->buf, 0, sizeof(c->buf)); + + c->nrecv++; + nng_sleep_aio(rand() % 10, c->time_aio); +} + +static void +req_send_cb(void *arg) +{ + test_case *c = arg; + int rv; + + if ((rv = nng_aio_result(c->send_aio)) != 0) { + error(c, "send", rv); + nng_msg_free(nng_aio_get_msg(c->send_aio)); + nng_aio_set_msg(c->send_aio, NULL); + return; + } + c->nsend++; + nng_recv_aio(c->socket, c->recv_aio); +} + +void +reqrep_test(int ntests) +{ + test_case *srv, *cli; + int i; + char addr[NNG_MAXADDRLEN]; + int rv; + + if (ntests < 2) { + // Need a client *and* a server. + return; + } + + srv = &cases[curcase++]; + srv->name = "rep"; + + if ((rv = nng_rep0_open(&srv->socket)) != 0) { + fatal("nng_rep0_open", rv); + } + + if (((rv = nng_aio_alloc(&srv->send_aio, rep_send_cb, srv)) != 0) || + ((rv = nng_aio_alloc(&srv->recv_aio, rep_recv_cb, srv)) != 0)) { + fatal("nng_aio_alloc", rv); + } + + nng_recv_aio(srv->socket, srv->recv_aio); + + for (i = 1; i < ntests; i++) { + cli = &cases[curcase++]; + + if (((rv = nng_aio_alloc(&cli->send_aio, req_send_cb, cli)) != + 0) || + ((rv = nng_aio_alloc(&cli->recv_aio, req_recv_cb, cli)) != + 0) || + ((rv = nng_aio_alloc(&cli->time_aio, req_time_cb, cli)) != + 0)) { + fatal("nng_aio_alloc", rv); + } + + if ((rv = nng_req0_open(&cli->socket)) != 0) { + fatal("nng_req0_open", rv); + } + + cli->name = "req"; + getaddr(addr); + dprintf("DOING reqrep0 (req %u rep %u) address: %s\n", + cli->socket.id, srv->socket.id, addr); + + if ((rv = nng_listen(srv->socket, addr, NULL, 0)) != 0) { + fatal("nng_listen", rv); + } + if ((rv = nng_dial(cli->socket, addr, NULL, 0)) != 0) { + fatal("nng_dial", rv); + } + + nng_sleep_aio(1, cli->time_aio); + } +} + +static void +test_req_stress(void) +{ + int i; + int tmo; + char *str; + + if (((str = getenv("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 sockets. (On macOS + // laptop I can run this with 500 though.) + if (((str = getenv("STRESSPRESSURE")) == NULL) || + ((ncases = atoi(str)) < 1)) { + ncases = 32; + } + + // Each run should truly be random. + srand((int) time(NULL)); + + // Reduce the likelihood of address in use conflicts between + // subsequent runs. + next_port += (rand() % 100) * 100; + + i = ncases; + + cases = calloc(ncases, sizeof(test_case)); + while (i > 1) { + int x = rand() % NTEMPLATES; + if (x > i) { + x = i; + } + reqrep_test(x); + i -= x; + } + + dprintf("WAITING for %d sec...\n", tmo); + nng_msleep(tmo * 1000); // sleep 30 sec + + // Close the timeouts first, before closing sockets. This tends to + // ensure that we complete exchanges. + for (i = 0; i < ncases; i++) { + nng_aio_stop(cases[i].time_aio); + } + nng_msleep(100); + + for (i = 0; i < ncases; i++) { + nng_aio_stop(cases[i].recv_aio); + nng_aio_stop(cases[i].send_aio); + nng_aio_stop(cases[i].time_aio); + nng_aio_free(cases[i].recv_aio); + nng_aio_free(cases[i].send_aio); + nng_aio_free(cases[i].time_aio); + if (cases[i].name != NULL) { + dprintf("RESULT socket %u (%s) sent %d " + "recd %d fail %d\n", + cases[i].socket.id, cases[i].name, cases[i].nsend, + cases[i].nrecv, cases[i].nfail); + NUTS_TRUE(cases[i].nfail == 0); + NUTS_TRUE(cases[i].nsend > 0 || cases[i].nrecv > 0); + NUTS_CLOSE(cases[i].socket); + } + } + + free(cases); + nng_fini(); +} + +NUTS_TESTS = { + { "req stress", test_req_stress }, + { NULL, NULL }, +}; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 86a76069..f41379e0 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -140,7 +140,5 @@ add_nng_test(ws 30) add_nng_test(wss 30) add_nng_test1(zt 60 NNG_TRANSPORT_ZEROTIER) -add_nng_test(reqstress 60) - # c++ tests add_nng_cpp_test(cplusplus_pair 5) diff --git a/tests/reqstress.c b/tests/reqstress.c deleted file mode 100644 index de31e987..00000000 --- a/tests/reqstress.c +++ /dev/null @@ -1,336 +0,0 @@ -// -// Copyright 2024 Staysail Systems, Inc. -// Copyright 2018 Capitar IT Group BV -// -// This software is supplied under the terms of the MIT License, a -// copy of which should be located in the distribution where this -// file was obtained (LICENSE.txt). A copy of the license may also be -// found online at https://opensource.org/licenses/MIT. -// - -#include -#include -#include - -#include -#include -#include -#include - -#include "convey.h" -#include "stubs.h" - -#ifdef NDEBUG -#define dprintf(...) -#else -#define dprintf printf -#endif - -static int next_port = 20000; // port number kind of. - -char tcp4_template[] = "tcp://127.0.0.1:%d"; -char tcp6_template[] = "tcp://[::1]:%d"; -char inproc_template[] = "inproc://nng_reqstress_%d"; -char ipc_template[] = "ipc:///tmp/nng_reqstress_%d"; -char ws_template[] = "ws://127.0.0.1:%d/nng_reqstress"; - -char *templates[] = { - tcp4_template, -// It would be nice to test TCPv6, but CI doesn't support it. -// Outside of CI, it does seem to work though. -#ifdef NNG_TEST_TCPV6 - tcp6_template, -#endif - inproc_template, - ipc_template, -#ifdef NNG_TRANSPORT_WS - ws_template, -#endif -}; - -#define NTEMPLATES (sizeof(templates) / sizeof(templates[0])) - -char **addresses; -int naddresses; -int allocaddrs; - -typedef struct test_case { - nng_socket socket; - const char *name; - int nrecv; - int nsend; - int nfail; - nng_aio *recv_aio; - nng_aio *send_aio; - nng_aio *time_aio; - char buf[32]; -} test_case; - -static test_case *cases; -int ncases; -int curcase; - -void -fatal(const char *msg, int rv) -{ - fprintf(stderr, "%s: %s\n", msg, nng_strerror(rv)); - abort(); -} - -void -error(test_case *c, const char *msg, int rv) -{ - if ((rv == NNG_ECLOSED) || (rv == NNG_ECANCELED)) { - return; - } - fprintf( - stderr, "%s: %s: %s (%d)\n", c->name, msg, nng_strerror(rv), rv); - c->nfail++; -} - -// Get a random address -- TCP, IP, whatever. -char * -getaddr(char *buf) -{ - int i; - i = rand() % NTEMPLATES; - - snprintf(buf, NNG_MAXADDRLEN, templates[i], next_port++); - return (buf); -} - -// Request/Reply test. For this test, we open a server socket, -// and bind it to each protocol. Then we run a bunch of clients -// against it. - -// REP server implemented via callbacks. -static void -rep_recv_cb(void *arg) -{ - test_case *c = arg; - int rv; - - if ((rv = nng_aio_result(c->recv_aio)) != 0) { - error(c, "recv", rv); - return; - } - c->nrecv++; - nng_aio_set_msg(c->send_aio, nng_aio_get_msg(c->recv_aio)); - nng_aio_set_msg(c->recv_aio, NULL); - nng_send_aio(c->socket, c->send_aio); -} - -static void -rep_send_cb(void *arg) -{ - test_case *c = arg; - int rv; - - if ((rv = nng_aio_result(c->send_aio)) != 0) { - error(c, "send", rv); - nng_msg_free(nng_aio_get_msg(c->send_aio)); - nng_aio_set_msg(c->send_aio, NULL); - return; - } - c->nsend++; - nng_recv_aio(c->socket, c->recv_aio); -} - -static void -req_time_cb(void *arg) -{ - test_case *c = arg; - nng_msg *msg = NULL; - int rv; - - if ((rv = nng_aio_result(c->time_aio)) != 0) { - error(c, "sleep", rv); - return; - } - (void) snprintf( - c->buf, sizeof(c->buf), "%u-%d", c->socket.id, c->nsend); - if (((rv = nng_msg_alloc(&msg, 0)) != 0) || - ((rv = nng_msg_append(msg, c->buf, strlen(c->buf) + 1)) != 0)) { - error(c, "alloc", rv); - nng_msg_free(msg); - return; - } - nng_aio_set_msg(c->send_aio, msg); - nng_send_aio(c->socket, c->send_aio); -} - -static void -req_recv_cb(void *arg) -{ - test_case *c = arg; - nng_msg *msg = NULL; - int rv; - - if ((rv = nng_aio_result(c->recv_aio)) != 0) { - error(c, "recv", rv); - return; - } - - msg = nng_aio_get_msg(c->recv_aio); - if ((nng_msg_len(msg) != (strlen(c->buf) + 1)) || - (strcmp(c->buf, nng_msg_body(msg)) != 0)) { - error(c, "msg mismatch", rv); - nng_msg_free(msg); - return; - } - - nng_msg_free(msg); - memset(c->buf, 0, sizeof(c->buf)); - - c->nrecv++; - nng_sleep_aio(rand() % 10, c->time_aio); -} - -static void -req_send_cb(void *arg) -{ - test_case *c = arg; - int rv; - - if ((rv = nng_aio_result(c->send_aio)) != 0) { - error(c, "send", rv); - nng_msg_free(nng_aio_get_msg(c->send_aio)); - nng_aio_set_msg(c->send_aio, NULL); - return; - } - c->nsend++; - nng_recv_aio(c->socket, c->recv_aio); -} - -void -reqrep_test(int ntests) -{ - test_case *srv, *cli; - int i; - char addr[NNG_MAXADDRLEN]; - int rv; - - if (ntests < 2) { - // Need a client *and* a server. - return; - } - - srv = &cases[curcase++]; - srv->name = "rep"; - - if ((rv = nng_rep0_open(&srv->socket)) != 0) { - fatal("nng_rep0_open", rv); - } - - if (((rv = nng_aio_alloc(&srv->send_aio, rep_send_cb, srv)) != 0) || - ((rv = nng_aio_alloc(&srv->recv_aio, rep_recv_cb, srv)) != 0)) { - fatal("nng_aio_alloc", rv); - } - - nng_recv_aio(srv->socket, srv->recv_aio); - - for (i = 1; i < ntests; i++) { - cli = &cases[curcase++]; - - if (((rv = nng_aio_alloc(&cli->send_aio, req_send_cb, cli)) != - 0) || - ((rv = nng_aio_alloc(&cli->recv_aio, req_recv_cb, cli)) != - 0) || - ((rv = nng_aio_alloc(&cli->time_aio, req_time_cb, cli)) != - 0)) { - fatal("nng_aio_alloc", rv); - } - - if ((rv = nng_req0_open(&cli->socket)) != 0) { - fatal("nng_req0_open", rv); - } - - cli->name = "req"; - getaddr(addr); - dprintf("DOING reqrep0 (req %u rep %u) address: %s\n", - cli->socket.id, srv->socket.id, addr); - - if ((rv = nng_listen(srv->socket, addr, NULL, 0)) != 0) { - fatal("nng_listen", rv); - } - if ((rv = nng_dial(cli->socket, addr, NULL, 0)) != 0) { - fatal("nng_dial", rv); - } - - nng_sleep_aio(1, cli->time_aio); - } -} - -Main({ - int i; - int tmo; - char *str; - - 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 sockets. (On macOS - // laptop I can run this with 500 though.) - if (((str = ConveyGetEnv("STRESSPRESSURE")) == NULL) || - ((ncases = atoi(str)) < 1)) { - ncases = 32; - } - - // Each run should truly be random. - srand((int) time(NULL)); - - // Reduce the likelihood of address in use conflicts between - // subsequent runs. - next_port += (rand() % 100) * 100; - - i = ncases; - - cases = calloc(ncases, sizeof(test_case)); - while (i > 1) { - int x = rand() % NTEMPLATES; - if (x > i) { - x = i; - } - reqrep_test(x); - i -= x; - } - - dprintf("WAITING for %d sec...\n", tmo); - nng_msleep(tmo * 1000); // sleep 30 sec - - // Close the timeouts first, before closing sockets. This tends to - // ensure that we complete exchanges. - for (i = 0; i < ncases; i++) { - nng_aio_stop(cases[i].time_aio); - } - nng_msleep(100); - - Test("Req/Rep Stress", { - Convey("All tests worked", { - for (i = 0; i < ncases; i++) { - nng_aio_stop(cases[i].recv_aio); - nng_aio_stop(cases[i].send_aio); - nng_aio_stop(cases[i].time_aio); - nng_aio_free(cases[i].recv_aio); - nng_aio_free(cases[i].send_aio); - nng_aio_free(cases[i].time_aio); - if (cases[i].name != NULL) { - dprintf( - "RESULT socket %u (%s) sent %d " - "recd %d fail %d\n", - cases[i].socket.id, cases[i].name, - cases[i].nsend, cases[i].nrecv, - cases[i].nfail); - So(cases[i].nfail == 0); - So(cases[i].nsend > 0 || - cases[i].nrecv > 0); - } - } - }); - }); - - free(cases); - nng_fini(); -}) -- cgit v1.2.3-70-g09d2