Skip to content

Commit

Permalink
Add REP huge reply (10MB) test
Browse files Browse the repository at this point in the history
  • Loading branch information
@gdamore
gdamore committed Feb 18, 2024
1 parent 02cc3d4 commit ee697a2
Showing 1 changed file with 80 additions and 24 deletions.
104 changes: 80 additions & 24 deletions src/sp/protocol/reqrep0/rep_test.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@ test_rep_identity(void)
{
nng_socket s;
int p1, p2;
char * n1;
char * n2;
char *n1;
char *n2;

NUTS_PASS(nng_rep0_open(&s));
NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PROTO, &p1));
Expand All @@ -35,7 +35,7 @@ void
test_rep_send_bad_state(void)
{
nng_socket rep;
nng_msg * msg = NULL;
nng_msg *msg = NULL;

NUTS_TRUE(nng_rep0_open(&rep) == 0);
NUTS_TRUE(nng_msg_alloc(&msg, 0) == 0);
Expand Down Expand Up @@ -86,7 +86,7 @@ test_rep_poll_readable(void)
int fd;
nng_socket req;
nng_socket rep;
nng_msg * msg;
nng_msg *msg;

NUTS_PASS(nng_req0_open(&req));
NUTS_PASS(nng_rep0_open(&rep));
Expand Down Expand Up @@ -137,9 +137,9 @@ void
test_rep_validate_peer(void)
{
nng_socket s1, s2;
nng_stat * stats;
nng_stat * reject;
char * addr;
nng_stat *stats;
nng_stat *reject;
char *addr;

NUTS_ADDR(addr, "inproc");
NUTS_PASS(nng_rep0_open(&s1));
Expand Down Expand Up @@ -167,8 +167,8 @@ void
test_rep_double_recv(void)
{
nng_socket s1;
nng_aio * aio1;
nng_aio * aio2;
nng_aio *aio1;
nng_aio *aio2;

NUTS_PASS(nng_rep0_open(&s1));
NUTS_PASS(nng_aio_alloc(&aio1, NULL, NULL));
Expand All @@ -185,14 +185,69 @@ test_rep_double_recv(void)
nng_aio_free(aio2);
}

void
test_rep_huge_send(void)
{
nng_socket rep;
nng_socket req;
nng_msg *m;
nng_msg *d;
nng_aio *aio;

NUTS_PASS(nng_rep_open(&rep));
NUTS_PASS(nng_req_open(&req));
NUTS_PASS(nng_socket_set_ms(rep, NNG_OPT_RECVTIMEO, 1000));
NUTS_PASS(nng_socket_set_ms(req, NNG_OPT_RECVTIMEO, 1000));
NUTS_PASS(nng_socket_set_ms(rep, NNG_OPT_SENDTIMEO, 1000));
NUTS_PASS(nng_socket_set_ms(req, NNG_OPT_SENDTIMEO, 1000));
NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
NUTS_PASS(nng_msg_alloc(&m, 10 << 20)); // 10 MB
NUTS_MARRY(req, rep);
char *body = nng_msg_body(m);

NUTS_ASSERT(nng_msg_len(m) == 10 << 20);
for (size_t i = 0; i < nng_msg_len(m); i++) {
body[i] = i % 16 + 'A';
}
NUTS_PASS(nng_msg_dup(&d, m));
NUTS_SEND(req, "R");
NUTS_RECV(rep, "R");
nng_aio_set_msg(aio, m);
nng_send_aio(rep, aio);
nng_aio_wait(aio);
NUTS_PASS(nng_aio_result(aio));
nng_aio_set_msg(aio, NULL);
m = NULL;
nng_recv_aio(req, aio);
nng_aio_wait(aio);
NUTS_PASS(nng_aio_result(aio));
m = nng_aio_get_msg(aio);
NUTS_ASSERT(m != NULL);
NUTS_ASSERT(nng_msg_len(m) == nng_msg_len(d));
NUTS_ASSERT(
memcmp(nng_msg_body(m), nng_msg_body(d), nng_msg_len(m)) == 0);

// make sure other messages still flow afterwards
NUTS_SEND(req, "E");
NUTS_RECV(rep, "E");
NUTS_SEND(rep, "E");
NUTS_RECV(req, "E");

nng_aio_free(aio);
nng_msg_free(m);
nng_msg_free(d);
NUTS_CLOSE(rep);
NUTS_CLOSE(req);
}

void
test_rep_close_pipe_before_send(void)
{
nng_socket rep;
nng_socket req;
nng_pipe p;
nng_aio * aio1;
nng_msg * m;
nng_aio *aio1;
nng_msg *m;

NUTS_PASS(nng_rep0_open(&rep));
NUTS_PASS(nng_req0_open(&req));
Expand Down Expand Up @@ -223,7 +278,7 @@ test_rep_close_pipe_during_send(void)
nng_socket rep;
nng_socket req;
nng_pipe p = NNG_PIPE_INITIALIZER;
nng_msg * m;
nng_msg *m;

NUTS_PASS(nng_rep0_open(&rep));
NUTS_PASS(nng_req0_open_raw(&req));
Expand Down Expand Up @@ -263,7 +318,7 @@ test_rep_ctx_recv_aio_stopped(void)
{
nng_socket rep;
nng_ctx ctx;
nng_aio * aio;
nng_aio *aio;

NUTS_PASS(nng_rep0_open(&rep));
NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
Expand All @@ -284,9 +339,9 @@ test_rep_close_pipe_context_send(void)
nng_socket rep;
nng_socket req;
nng_pipe p = NNG_PIPE_INITIALIZER;
nng_msg * m;
nng_msg *m;
nng_ctx ctx[100];
nng_aio * aio[100];
nng_aio *aio[100];
int i;

NUTS_PASS(nng_rep0_open(&rep));
Expand Down Expand Up @@ -343,9 +398,9 @@ test_rep_close_context_send(void)
{
nng_socket rep;
nng_socket req;
nng_msg * m;
nng_msg *m;
nng_ctx ctx[100];
nng_aio * aio[100];
nng_aio *aio[100];
int i;

NUTS_PASS(nng_rep0_open(&rep));
Expand Down Expand Up @@ -399,7 +454,7 @@ test_rep_close_recv(void)
{
nng_socket rep;
nng_socket req;
nng_aio * aio;
nng_aio *aio;

NUTS_PASS(nng_rep0_open(&rep));
NUTS_PASS(nng_req0_open_raw(&req));
Expand All @@ -420,7 +475,7 @@ test_rep_close_recv(void)
struct rep_close_recv_cb_state {
nng_aio *aio;
nng_mtx *mtx;
nng_cv * cv;
nng_cv *cv;
int done;
int result;
nng_msg *msg;
Expand Down Expand Up @@ -479,7 +534,7 @@ test_rep_ctx_recv_nonblock(void)
{
nng_socket rep;
nng_ctx ctx;
nng_aio * aio;
nng_aio *aio;

NUTS_PASS(nng_rep0_open(&rep));
NUTS_PASS(nng_ctx_open(&ctx, rep));
Expand All @@ -500,8 +555,8 @@ test_rep_ctx_send_nonblock(void)
nng_socket rep;
nng_socket req;
nng_ctx ctx;
nng_aio * aio;
nng_msg * msg;
nng_aio *aio;
nng_msg *msg;

NUTS_PASS(nng_req0_open(&req));
NUTS_PASS(nng_rep0_open(&rep));
Expand Down Expand Up @@ -535,7 +590,7 @@ test_rep_ctx_send_nonblock2(void)
nng_socket rep;
nng_socket req;
nng_ctx rep_ctx[10];
nng_aio * rep_aio[10];
nng_aio *rep_aio[10];
int num_good = 0;
int num_fail = 0;

Expand Down Expand Up @@ -626,7 +681,7 @@ test_rep_recv_garbage(void)
{
nng_socket rep;
nng_socket req;
nng_msg * m;
nng_msg *m;

NUTS_PASS(nng_rep0_open(&rep));
NUTS_PASS(nng_req0_open_raw(&req));
Expand All @@ -652,6 +707,7 @@ NUTS_TESTS = {
{ "rep poll writable", test_rep_poll_writeable },
{ "rep context does not poll", test_rep_context_no_poll },
{ "rep validate peer", test_rep_validate_peer },
{ "rep huge send", test_rep_huge_send },
{ "rep double recv", test_rep_double_recv },
{ "rep send nonblock", test_rep_send_nonblock },
{ "rep close pipe before send", test_rep_close_pipe_before_send },
Expand Down

0 comments on commit ee697a2

Please sign in to comment.