From 6d92c73e5cdf93fe70b0646e78a250e01a8d2f65 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Sat, 11 Jan 2020 13:15:40 -0800 Subject: XREQ and others race on TTL. The TTL in these cases should have been atomic. To facilitate things we actually introduce an atomic int for convenience. We also introduce a convenience nni_msg_must_append_u32() and nni_msg_header_must_append_u32(), so that we can eliminate some failure tests that cannot ever happen. Combined with a new test for xreq, we have 100% coverage for xreq and more coverage for the other REQ/REP protocols. --- src/protocol/reqrep0/CMakeLists.txt | 1 + src/protocol/reqrep0/rep.c | 39 +++-- src/protocol/reqrep0/rep_test.c | 6 +- src/protocol/reqrep0/req.c | 92 +++++----- src/protocol/reqrep0/xrep.c | 17 +- src/protocol/reqrep0/xrep_test.c | 32 ++-- src/protocol/reqrep0/xreq.c | 130 +++++++------- src/protocol/reqrep0/xreq_test.c | 333 ++++++++++++++++++++++++++++++++++++ 8 files changed, 490 insertions(+), 160 deletions(-) create mode 100644 src/protocol/reqrep0/xreq_test.c (limited to 'src/protocol') diff --git a/src/protocol/reqrep0/CMakeLists.txt b/src/protocol/reqrep0/CMakeLists.txt index 3cd8d366..4e1b7a8f 100644 --- a/src/protocol/reqrep0/CMakeLists.txt +++ b/src/protocol/reqrep0/CMakeLists.txt @@ -26,3 +26,4 @@ nng_defines_if(NNG_PROTO_REP0 NNG_HAVE_REP0) nng_test(reqrep_test) nng_test(rep_test) nng_test(xrep_test) +nng_test(xreq_test) diff --git a/src/protocol/reqrep0/rep.c b/src/protocol/reqrep0/rep.c index a29c3120..3e1a34a9 100644 --- a/src/protocol/reqrep0/rep.c +++ b/src/protocol/reqrep0/rep.c @@ -48,14 +48,14 @@ struct rep0_ctx { // rep0_sock is our per-socket protocol private structure. struct rep0_sock { - nni_mtx lk; - int ttl; - nni_idhash * pipes; - nni_list recvpipes; // list of pipes with data to receive - nni_list recvq; - rep0_ctx ctx; - nni_pollable readable; - nni_pollable writable; + nni_mtx lk; + nni_atomic_int ttl; + nni_idhash * pipes; + nni_list recvpipes; // list of pipes with data to receive + nni_list recvq; + rep0_ctx ctx; + nni_pollable readable; + nni_pollable writable; }; // rep0_pipe is our per-pipe protocol private structure. @@ -241,8 +241,8 @@ rep0_sock_init(void *arg, nni_sock *sock) NNI_LIST_INIT(&s->recvq, rep0_ctx, rqnode); NNI_LIST_INIT(&s->recvpipes, rep0_pipe, rnode); - - s->ttl = 8; + nni_atomic_init(&s->ttl); + nni_atomic_set(&s->ttl, 8); (void) rep0_ctx_init(&s->ctx, s); @@ -506,7 +506,7 @@ rep0_pipe_recv_cb(void *arg) for (;;) { bool end; - if (hops > s->ttl) { + if (hops > nni_atomic_get(&s->ttl)) { // This isn't malformed, but it has gone // through too many hops. Do not disconnect, // because we can legitimately receive messages @@ -574,19 +574,24 @@ drop: } static int -rep0_sock_set_maxttl(void *arg, const void *buf, size_t sz, nni_opt_type t) +rep0_sock_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t) { rep0_sock *s = arg; + int ttl; + int rv; - return (nni_copyin_int(&s->ttl, buf, sz, 1, 255, t)); + if ((rv = nni_copyin_int(&ttl, buf, sz, 1, 255, t)) == 0) { + nni_atomic_set(&s->ttl, ttl); + } + return (rv); } static int -rep0_sock_get_maxttl(void *arg, void *buf, size_t *szp, nni_opt_type t) +rep0_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t) { rep0_sock *s = arg; - return (nni_copyout_int(s->ttl, buf, szp, t)); + return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t)); } static int @@ -654,8 +659,8 @@ static nni_proto_ctx_ops rep0_ctx_ops = { static nni_option rep0_sock_options[] = { { .o_name = NNG_OPT_MAXTTL, - .o_get = rep0_sock_get_maxttl, - .o_set = rep0_sock_set_maxttl, + .o_get = rep0_sock_get_max_ttl, + .o_set = rep0_sock_set_max_ttl, }, { .o_name = NNG_OPT_RECVFD, diff --git a/src/protocol/reqrep0/rep_test.c b/src/protocol/reqrep0/rep_test.c index f339e68d..7c6d6ba0 100644 --- a/src/protocol/reqrep0/rep_test.c +++ b/src/protocol/reqrep0/rep_test.c @@ -117,7 +117,7 @@ test_rep_poll_readable(void) TEST_CHECK(testutil_pollfd(fd) == true); - // and receiving makes it no longer pollable + // and receiving makes it no longer ready TEST_NNG_PASS(nng_recvmsg(rep, &msg, 0)); nng_msg_free(msg); TEST_CHECK(testutil_pollfd(fd) == false); @@ -129,7 +129,7 @@ test_rep_poll_readable(void) } void -test_rep_context_not_pollable(void) +test_rep_context_no_poll(void) { int fd; nng_socket req; @@ -439,7 +439,7 @@ TEST_LIST = { { "rep send bad state", test_rep_send_bad_state }, { "rep poll readable", test_rep_poll_readable }, { "rep poll writable", test_rep_poll_writeable }, - { "rep context not pollable", test_rep_context_not_pollable }, + { "rep context does not poll", test_rep_context_no_poll }, { "rep validate peer", test_rep_validate_peer }, { "rep double recv", test_rep_double_recv }, { "rep close pipe before send", test_rep_close_pipe_before_send }, diff --git a/src/protocol/reqrep0/req.c b/src/protocol/reqrep0/req.c index 14da7143..796bd71e 100644 --- a/src/protocol/reqrep0/req.c +++ b/src/protocol/reqrep0/req.c @@ -9,7 +9,6 @@ // #include -#include #include "core/nng_impl.h" #include "nng/protocol/reqrep0/req.h" @@ -56,19 +55,19 @@ struct req0_ctx { // A req0_sock is our per-socket protocol private structure. struct req0_sock { - nni_duration retry; - bool closed; - int ttl; - req0_ctx master; // base socket master - nni_list ready_pipes; - nni_list busy_pipes; - nni_list stop_pipes; - nni_list contexts; - nni_list send_queue; // contexts waiting to send. - nni_idhash * requests; // contexts by request ID - nni_pollable readable; - nni_pollable writable; - nni_mtx mtx; + nni_duration retry; + bool closed; + nni_atomic_int ttl; + req0_ctx master; // base socket master + nni_list ready_pipes; + nni_list busy_pipes; + nni_list stop_pipes; + nni_list contexts; + nni_list send_queue; // contexts waiting to send. + nni_idhash * requests; // contexts by request ID + nni_pollable readable; + nni_pollable writable; + nni_mtx mtx; }; // A req0_pipe is our per-pipe protocol private structure. @@ -78,8 +77,8 @@ struct req0_pipe { nni_list_node node; nni_list contexts; // contexts with pending traffic bool closed; - nni_aio * aio_send; - nni_aio * aio_recv; + nni_aio aio_send; + nni_aio aio_recv; }; static void req0_sock_fini(void *); @@ -120,7 +119,8 @@ req0_sock_init(void *arg, nni_sock *sock) nni_pollable_init(&s->writable); nni_pollable_init(&s->readable); - s->ttl = 8; + nni_atomic_init(&s->ttl); + nni_atomic_set(&s->ttl, 8); return (0); } @@ -172,8 +172,8 @@ req0_pipe_stop(void *arg) req0_pipe *p = arg; req0_sock *s = p->req; - nni_aio_stop(p->aio_recv); - nni_aio_stop(p->aio_send); + nni_aio_stop(&p->aio_recv); + nni_aio_stop(&p->aio_send); nni_mtx_lock(&s->mtx); nni_list_node_remove(&p->node); nni_mtx_unlock(&s->mtx); @@ -184,22 +184,17 @@ req0_pipe_fini(void *arg) { req0_pipe *p = arg; - nni_aio_free(p->aio_recv); - nni_aio_free(p->aio_send); + nni_aio_fini(&p->aio_recv); + nni_aio_fini(&p->aio_send); } static int req0_pipe_init(void *arg, nni_pipe *pipe, void *s) { req0_pipe *p = arg; - int rv; - - if (((rv = nni_aio_alloc(&p->aio_recv, req0_recv_cb, p)) != 0) || - ((rv = nni_aio_alloc(&p->aio_send, req0_send_cb, p)) != 0)) { - req0_pipe_fini(p); - return (rv); - } + nni_aio_init(&p->aio_recv, req0_recv_cb, p); + nni_aio_init(&p->aio_send, req0_send_cb, p); NNI_LIST_NODE_INIT(&p->node); NNI_LIST_INIT(&p->contexts, req0_ctx, pipe_node); p->pipe = pipe; @@ -227,7 +222,7 @@ req0_pipe_start(void *arg) req0_run_send_queue(s, NULL); nni_mtx_unlock(&s->mtx); - nni_pipe_recv(p->pipe, p->aio_recv); + nni_pipe_recv(p->pipe, &p->aio_recv); return (0); } @@ -238,8 +233,8 @@ req0_pipe_close(void *arg) req0_sock *s = p->req; req0_ctx * ctx; - nni_aio_close(p->aio_recv); - nni_aio_close(p->aio_send); + nni_aio_close(&p->aio_recv); + nni_aio_close(&p->aio_send); nni_mtx_lock(&s->mtx); // This removes the node from either busy_pipes or ready_pipes. @@ -277,10 +272,10 @@ req0_send_cb(void *arg) nni_list send_list; nni_aio_list_init(&send_list); - if (nni_aio_result(p->aio_send) != 0) { + if (nni_aio_result(&p->aio_send) != 0) { // We failed to send... clean up and deal with it. - nni_msg_free(nni_aio_get_msg(p->aio_send)); - nni_aio_set_msg(p->aio_send, NULL); + nni_msg_free(nni_aio_get_msg(&p->aio_send)); + nni_aio_set_msg(&p->aio_send, NULL); nni_pipe_close(p->pipe); return; } @@ -319,13 +314,13 @@ req0_recv_cb(void *arg) nni_aio * aio; uint32_t id; - if (nni_aio_result(p->aio_recv) != 0) { + if (nni_aio_result(&p->aio_recv) != 0) { nni_pipe_close(p->pipe); return; } - msg = nni_aio_get_msg(p->aio_recv); - nni_aio_set_msg(p->aio_recv, NULL); + msg = nni_aio_get_msg(&p->aio_recv); + nni_aio_set_msg(&p->aio_recv, NULL); nni_msg_set_pipe(msg, nni_pipe_id(p->pipe)); // We yank 4 bytes from front of body, and move them to the header. @@ -334,17 +329,11 @@ req0_recv_cb(void *arg) goto malformed; } id = nni_msg_trim_u32(msg); - if (nni_msg_header_append_u32(msg, id) != 0) { - // Arguably we could just discard and carry on. But - // dropping the connection is probably more helpful since - // it lets the other side see that a problem occurred. - // Plus it gives us a chance to reclaim some memory. - goto malformed; - } + nni_msg_header_must_append_u32(msg, id); // Schedule another receive while we are processing this. nni_mtx_lock(&s->mtx); - nni_pipe_recv(p->pipe, p->aio_recv); + nni_pipe_recv(p->pipe, &p->aio_recv); // Look for a context to receive it. if ((nni_idhash_find(s->requests, id, (void **) &ctx) != 0) || @@ -526,8 +515,8 @@ req0_run_send_queue(req0_sock *s, nni_list *send_list) } } - nni_aio_set_msg(p->aio_send, msg); - nni_pipe_send(p->pipe, p->aio_send); + nni_aio_set_msg(&p->aio_send, msg); + nni_pipe_send(p->pipe, &p->aio_send); } } @@ -763,14 +752,19 @@ static int req0_sock_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t) { req0_sock *s = arg; - return (nni_copyin_int(&s->ttl, buf, sz, 1, 255, t)); + int ttl; + int rv; + if ((rv = nni_copyin_int(&ttl, buf, sz, 1, 255, t)) == 0) { + nni_atomic_set(&s->ttl, ttl); + } + return (rv); } static int req0_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t) { req0_sock *s = arg; - return (nni_copyout_int(s->ttl, buf, szp, t)); + return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t)); } static int diff --git a/src/protocol/reqrep0/xrep.c b/src/protocol/reqrep0/xrep.c index aac6d6b9..e5d96f02 100644 --- a/src/protocol/reqrep0/xrep.c +++ b/src/protocol/reqrep0/xrep.c @@ -40,7 +40,7 @@ struct xrep0_sock { nni_msgq * uwq; nni_msgq * urq; nni_mtx lk; - nni_atomic_u64 ttl; + nni_atomic_int ttl; nni_idhash * pipes; nni_aio aio_getq; }; @@ -74,8 +74,8 @@ xrep0_sock_init(void *arg, nni_sock *sock) nni_mtx_init(&s->lk); nni_aio_init(&s->aio_getq, xrep0_sock_getq_cb, s); - nni_atomic_init64(&s->ttl); - nni_atomic_set64(&s->ttl, 8); // Per RFC + nni_atomic_init(&s->ttl); + nni_atomic_set(&s->ttl, 8); // Per RFC s->uwq = nni_sock_sendq(sock); s->urq = nni_sock_recvq(sock); @@ -295,17 +295,14 @@ xrep0_pipe_recv_cb(void *arg) nni_msg_set_pipe(msg, nni_pipe_id(p->pipe)); // Store the pipe id in the header, first thing. - if (nni_msg_header_append_u32(msg, nni_pipe_id(p->pipe)) != 0) { - // Failure here causes us to drop the message. - goto drop; - } + nni_msg_header_must_append_u32(msg, nni_pipe_id(p->pipe)); // Move backtrace from body to header hops = 1; for (;;) { bool end = 0; uint8_t *body; - if (hops > (int)nni_atomic_get64(&s->ttl)) { + if (hops > (int)nni_atomic_get(&s->ttl)) { // This isn't malformed, but it has gone through // too many hops. Do not disconnect, because we // can legitimately receive messages with too many @@ -364,7 +361,7 @@ xrep0_sock_set_maxttl(void *arg, const void *buf, size_t sz, nni_opt_type t) int ttl; int rv; if ((rv = nni_copyin_int(&ttl, buf, sz, 1, 255, t)) == 0) { - nni_atomic_set64(&s->ttl, (uint64_t) ttl); + nni_atomic_set(&s->ttl, ttl); } return (rv); } @@ -373,7 +370,7 @@ static int xrep0_sock_get_maxttl(void *arg, void *buf, size_t *szp, nni_opt_type t) { xrep0_sock *s = arg; - return (nni_copyout_int((int) nni_atomic_get64(&s->ttl), buf, szp, t)); + return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t)); } static void diff --git a/src/protocol/reqrep0/xrep_test.c b/src/protocol/reqrep0/xrep_test.c index 93573be1..958f3240 100644 --- a/src/protocol/reqrep0/xrep_test.c +++ b/src/protocol/reqrep0/xrep_test.c @@ -20,7 +20,7 @@ #define NNI_PROTO(x, y) (((x) << 4u) | (y)) #endif -void +static void test_xrep_identity(void) { nng_socket s; @@ -41,7 +41,7 @@ test_xrep_identity(void) TEST_CHECK(nng_close(s) == 0); } -void +static void test_xrep_raw(void) { nng_socket s; @@ -53,7 +53,7 @@ test_xrep_raw(void) TEST_NNG_PASS(nng_close(s)); } -void +static void test_xrep_no_context(void) { nng_socket s; @@ -64,7 +64,7 @@ test_xrep_no_context(void) TEST_NNG_PASS(nng_close(s)); } -void +static void test_xrep_poll_writeable(void) { int fd; @@ -91,7 +91,7 @@ test_xrep_poll_writeable(void) TEST_NNG_PASS(nng_close(rep)); } -void +static void test_xrep_poll_readable(void) { int fd; @@ -118,7 +118,7 @@ test_xrep_poll_readable(void) TEST_CHECK(testutil_pollfd(fd) == true); - // and receiving makes it no longer pollable + // and receiving makes it no longer ready TEST_NNG_PASS(nng_recvmsg(rep, &msg, 0)); nng_msg_free(msg); TEST_CHECK(testutil_pollfd(fd) == false); @@ -127,7 +127,7 @@ test_xrep_poll_readable(void) TEST_NNG_PASS(nng_close(rep)); } -void +static void test_xrep_validate_peer(void) { nng_socket s1, s2; @@ -158,7 +158,7 @@ test_xrep_validate_peer(void) nng_stats_free(stats); } -void +static void test_xrep_close_pipe_before_send(void) { nng_socket rep; @@ -190,7 +190,7 @@ test_xrep_close_pipe_before_send(void) nng_aio_free(aio1); } -void +static void test_xrep_close_pipe_during_send(void) { nng_socket rep; @@ -231,7 +231,7 @@ test_xrep_close_pipe_during_send(void) TEST_NNG_PASS(nng_close(rep)); } -void +static void test_xrep_close_during_recv(void) { nng_socket rep; @@ -261,7 +261,7 @@ test_xrep_close_during_recv(void) TEST_NNG_PASS(nng_close(rep)); } -void +static void test_xrep_recv_aio_stopped(void) { nng_socket rep; @@ -278,7 +278,7 @@ test_xrep_recv_aio_stopped(void) nng_aio_free(aio); } -void +static void test_xrep_send_no_header(void) { nng_socket rep; @@ -302,7 +302,7 @@ test_xrep_send_no_header(void) TEST_NNG_PASS(nng_close(rep)); } -void +static void test_xrep_recv_garbage(void) { nng_socket rep; @@ -326,13 +326,13 @@ test_xrep_recv_garbage(void) TEST_NNG_PASS(nng_close(rep)); } -void +static void test_xrep_ttl_option(void) { nng_socket rep; int v; bool b; - size_t sz = sizeof(v); + size_t sz; const char *opt = NNG_OPT_MAXTTL; TEST_NNG_PASS(nng_rep0_open_raw(&rep)); @@ -359,7 +359,7 @@ test_xrep_ttl_option(void) TEST_CHECK(nng_close(rep) == 0); } -void +static void test_xrep_ttl_drop(void) { nng_socket rep; diff --git a/src/protocol/reqrep0/xreq.c b/src/protocol/reqrep0/xreq.c index 15652f4f..ef24010e 100644 --- a/src/protocol/reqrep0/xreq.c +++ b/src/protocol/reqrep0/xreq.c @@ -11,7 +11,7 @@ #include #include "core/nng_impl.h" -#include "nng/protocol/reqrep0/req.h" +//#include "nng/protocol/reqrep0/req.h" // Request protocol. The REQ protocol is the "request" side of a // request-reply pair. This is useful for building RPC clients, for example. @@ -29,19 +29,19 @@ typedef struct xreq0_sock xreq0_sock; // An xreq0_sock is our per-socket protocol private structure. struct xreq0_sock { - nni_msgq *uwq; - nni_msgq *urq; - int ttl; + nni_msgq * uwq; + nni_msgq * urq; + nni_atomic_int ttl; }; // A req0_pipe is our per-pipe protocol private structure. struct xreq0_pipe { nni_pipe * pipe; xreq0_sock *req; - nni_aio * aio_getq; - nni_aio * aio_send; - nni_aio * aio_recv; - nni_aio * aio_putq; + nni_aio aio_getq; + nni_aio aio_send; + nni_aio aio_recv; + nni_aio aio_putq; }; static void xreq0_sock_fini(void *); @@ -55,7 +55,8 @@ xreq0_sock_init(void *arg, nni_sock *sock) { xreq0_sock *s = arg; - s->ttl = 8; + nni_atomic_init(&s->ttl); + nni_atomic_set(&s->ttl, 8); s->uwq = nni_sock_sendq(sock); s->urq = nni_sock_recvq(sock); @@ -85,10 +86,10 @@ xreq0_pipe_stop(void *arg) { xreq0_pipe *p = arg; - nni_aio_stop(p->aio_getq); - nni_aio_stop(p->aio_putq); - nni_aio_stop(p->aio_recv); - nni_aio_stop(p->aio_send); + nni_aio_stop(&p->aio_getq); + nni_aio_stop(&p->aio_putq); + nni_aio_stop(&p->aio_recv); + nni_aio_stop(&p->aio_send); } static void @@ -96,25 +97,21 @@ xreq0_pipe_fini(void *arg) { xreq0_pipe *p = arg; - nni_aio_free(p->aio_getq); - nni_aio_free(p->aio_putq); - nni_aio_free(p->aio_recv); - nni_aio_free(p->aio_send); + nni_aio_fini(&p->aio_getq); + nni_aio_fini(&p->aio_putq); + nni_aio_fini(&p->aio_recv); + nni_aio_fini(&p->aio_send); } static int xreq0_pipe_init(void *arg, nni_pipe *pipe, void *s) { xreq0_pipe *p = arg; - int rv; - if (((rv = nni_aio_alloc(&p->aio_getq, xreq0_getq_cb, p)) != 0) || - ((rv = nni_aio_alloc(&p->aio_putq, xreq0_putq_cb, p)) != 0) || - ((rv = nni_aio_alloc(&p->aio_recv, xreq0_recv_cb, p)) != 0) || - ((rv = nni_aio_alloc(&p->aio_send, xreq0_send_cb, p)) != 0)) { - xreq0_pipe_fini(p); - return (rv); - } + nni_aio_init(&p->aio_getq, xreq0_getq_cb, p); + nni_aio_init(&p->aio_putq, xreq0_putq_cb, p); + nni_aio_init(&p->aio_recv, xreq0_recv_cb, p); + nni_aio_init(&p->aio_send, xreq0_send_cb, p); p->pipe = pipe; p->req = s; @@ -131,8 +128,8 @@ xreq0_pipe_start(void *arg) return (NNG_EPROTO); } - nni_msgq_aio_get(s->uwq, p->aio_getq); - nni_pipe_recv(p->pipe, p->aio_recv); + nni_msgq_aio_get(s->uwq, &p->aio_getq); + nni_pipe_recv(p->pipe, &p->aio_recv); return (0); } @@ -141,14 +138,14 @@ xreq0_pipe_close(void *arg) { xreq0_pipe *p = arg; - nni_aio_close(p->aio_getq); - nni_aio_close(p->aio_putq); - nni_aio_close(p->aio_recv); - nni_aio_close(p->aio_send); + nni_aio_close(&p->aio_getq); + nni_aio_close(&p->aio_putq); + nni_aio_close(&p->aio_recv); + nni_aio_close(&p->aio_send); } -// For raw mode we can just let the pipes "contend" via getq to get a -// message from the upper write queue. The msgqueue implementation +// For raw mode we can just let the pipes "contend" via get queue to get a +// message from the upper write queue. The msg queue implementation // actually provides ordering, so load will be spread automatically. // (NB: We may have to revise this in the future if we want to provide some // kind of priority.) @@ -158,15 +155,15 @@ xreq0_getq_cb(void *arg) { xreq0_pipe *p = arg; - if (nni_aio_result(p->aio_getq) != 0) { + if (nni_aio_result(&p->aio_getq) != 0) { nni_pipe_close(p->pipe); return; } - nni_aio_set_msg(p->aio_send, nni_aio_get_msg(p->aio_getq)); - nni_aio_set_msg(p->aio_getq, NULL); + nni_aio_set_msg(&p->aio_send, nni_aio_get_msg(&p->aio_getq)); + nni_aio_set_msg(&p->aio_getq, NULL); - nni_pipe_send(p->pipe, p->aio_send); + nni_pipe_send(p->pipe, &p->aio_send); } static void @@ -174,15 +171,15 @@ xreq0_send_cb(void *arg) { xreq0_pipe *p = arg; - if (nni_aio_result(p->aio_send) != 0) { - nni_msg_free(nni_aio_get_msg(p->aio_send)); - nni_aio_set_msg(p->aio_send, NULL); + if (nni_aio_result(&p->aio_send) != 0) { + nni_msg_free(nni_aio_get_msg(&p->aio_send)); + nni_aio_set_msg(&p->aio_send, NULL); nni_pipe_close(p->pipe); return; } // Sent a message so we just need to look for another one. - nni_msgq_aio_get(p->req->uwq, p->aio_getq); + nni_msgq_aio_get(p->req->uwq, &p->aio_getq); } static void @@ -190,15 +187,15 @@ xreq0_putq_cb(void *arg) { xreq0_pipe *p = arg; - if (nni_aio_result(p->aio_putq) != 0) { - nni_msg_free(nni_aio_get_msg(p->aio_putq)); - nni_aio_set_msg(p->aio_putq, NULL); + if (nni_aio_result(&p->aio_putq) != 0) { + nni_msg_free(nni_aio_get_msg(&p->aio_putq)); + nni_aio_set_msg(&p->aio_putq, NULL); nni_pipe_close(p->pipe); return; } - nni_aio_set_msg(p->aio_putq, NULL); + nni_aio_set_msg(&p->aio_putq, NULL); - nni_pipe_recv(p->pipe, p->aio_recv); + nni_pipe_recv(p->pipe, &p->aio_recv); } static void @@ -209,13 +206,13 @@ xreq0_recv_cb(void *arg) nni_msg * msg; uint32_t id; - if (nni_aio_result(p->aio_recv) != 0) { + if (nni_aio_result(&p->aio_recv) != 0) { nni_pipe_close(p->pipe); return; } - msg = nni_aio_get_msg(p->aio_recv); - nni_aio_set_msg(p->aio_recv, NULL); + msg = nni_aio_get_msg(&p->aio_recv); + nni_aio_set_msg(&p->aio_recv, NULL); nni_msg_set_pipe(msg, nni_pipe_id(p->pipe)); // We yank 4 bytes from front of body, and move them to the header. @@ -226,15 +223,13 @@ xreq0_recv_cb(void *arg) return; } id = nni_msg_trim_u32(msg); - if (nni_msg_header_append_u32(msg, id) != 0) { - // Probably ENOMEM, discard and carry on. - nni_msg_free(msg); - nni_pipe_recv(p->pipe, p->aio_recv); - return; - } - nni_aio_set_msg(p->aio_putq, msg); - nni_msgq_aio_put(sock->urq, p->aio_putq); + // Since we got this from the transport, there had better be some + // room in the header for our stuff. + nni_msg_header_must_append_u32(msg, id); + + nni_aio_set_msg(&p->aio_putq, msg); + nni_msgq_aio_put(sock->urq, &p->aio_putq); } static void @@ -254,17 +249,22 @@ xreq0_sock_recv(void *arg, nni_aio *aio) } static int -xreq0_sock_set_maxttl(void *arg, const void *buf, size_t sz, nni_opt_type t) +xreq0_sock_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t) { xreq0_sock *s = arg; - return (nni_copyin_int(&s->ttl, buf, sz, 1, 255, t)); + int ttl; + int rv; + if ((rv = nni_copyin_int(&ttl, buf, sz, 1, 255, t)) == 0) { + nni_atomic_set(&s->ttl, ttl); + } + return (rv); } static int -xreq0_sock_get_maxttl(void *arg, void *buf, size_t *szp, nni_opt_type t) +xreq0_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t) { xreq0_sock *s = arg; - return (nni_copyout_int(s->ttl, buf, szp, t)); + return (nni_copyout_int(nni_atomic_get(&s->ttl), buf, szp, t)); } static nni_proto_pipe_ops xreq0_pipe_ops = { @@ -279,8 +279,8 @@ static nni_proto_pipe_ops xreq0_pipe_ops = { static nni_option xreq0_sock_options[] = { { .o_name = NNG_OPT_MAXTTL, - .o_get = xreq0_sock_get_maxttl, - .o_set = xreq0_sock_set_maxttl, + .o_get = xreq0_sock_get_max_ttl, + .o_set = xreq0_sock_set_max_ttl, }, // terminate list { @@ -310,7 +310,7 @@ static nni_proto xreq0_proto = { }; int -nng_req0_open_raw(nng_socket *sidp) +nng_req0_open_raw(nng_socket *sock) { - return (nni_proto_open(sidp, &xreq0_proto)); + return (nni_proto_open(sock, &xreq0_proto)); } diff --git a/src/protocol/reqrep0/xreq_test.c b/src/protocol/reqrep0/xreq_test.c new file mode 100644 index 00000000..7f14f0cf --- /dev/null +++ b/src/protocol/reqrep0/xreq_test.c @@ -0,0 +1,333 @@ +// +// Copyright 2020 Staysail Systems, Inc. +// +// 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 + +#ifndef NNI_PROTO +#define NNI_PROTO(x, y) (((x) << 4u) | (y)) +#endif + +static void +test_xreq_identity(void) +{ + nng_socket s; + int p; + char * n; + + TEST_CHECK(nng_req0_open_raw(&s) == 0); + TEST_CHECK(nng_getopt_int(s, NNG_OPT_PROTO, &p) == 0); + TEST_CHECK(p == NNI_PROTO(3u, 0u)); // 48 + TEST_CHECK(nng_getopt_int(s, NNG_OPT_PEER, &p) == 0); + TEST_CHECK(p == NNI_PROTO(3u, 1u)); // 49 + TEST_CHECK(nng_getopt_string(s, NNG_OPT_PROTONAME, &n) == 0); + TEST_CHECK(strcmp(n, "req") == 0); + nng_strfree(n); + TEST_CHECK(nng_getopt_string(s, NNG_OPT_PEERNAME, &n) == 0); + TEST_CHECK(strcmp(n, "rep") == 0); + nng_strfree(n); + TEST_CHECK(nng_close(s) == 0); +} + +static void +test_xreq_raw(void) +{ + nng_socket s; + bool b; + + TEST_NNG_PASS(nng_req0_open_raw(&s)); + TEST_NNG_PASS(nng_getopt_bool(s, NNG_OPT_RAW, &b)); + TEST_CHECK(b); + TEST_NNG_PASS(nng_close(s)); +} + +static void +test_xreq_no_context(void) +{ + nng_socket s; + nng_ctx ctx; + + TEST_NNG_PASS(nng_req0_open_raw(&s)); + TEST_NNG_FAIL(nng_ctx_open(&ctx, s), NNG_ENOTSUP); + TEST_NNG_PASS(nng_close(s)); +} + +static void +test_xreq_poll_writeable(void) +{ + int fd; + nng_socket req; + nng_socket rep; + + TEST_NNG_PASS(nng_req0_open_raw(&req)); + TEST_NNG_PASS(nng_rep0_open(&rep)); + TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_SENDFD, &fd)); + TEST_CHECK(fd >= 0); + + // We can't write until we have a connection. + TEST_CHECK(testutil_pollfd(fd) == false); + + TEST_NNG_PASS(testutil_marry(req, rep)); + + // Now it's writable. + TEST_CHECK(testutil_pollfd(fd) == true); + + TEST_NNG_PASS(nng_close(req)); + TEST_NNG_PASS(nng_close(rep)); +} + +static void +test_xreq_poll_readable(void) +{ + int fd; + nng_socket req; + nng_socket rep; + nng_msg * msg; + + TEST_NNG_PASS(nng_req0_open_raw(&req)); + TEST_NNG_PASS(nng_rep0_open(&rep)); + TEST_NNG_PASS(nng_getopt_int(req, NNG_OPT_RECVFD, &fd)); + TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000)); + TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, 1000)); + TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000)); + TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000)); + + TEST_CHECK(fd >= 0); + + // Not readable if not connected! + TEST_CHECK(testutil_pollfd(fd) == false); + + // Even after connect (no message yet) + TEST_NNG_PASS(testutil_marry(req, rep)); + TEST_CHECK(testutil_pollfd(fd) == false); + + // But once we send messages, it is. + // We have to send a request, in order to send a reply. + TEST_NNG_PASS(nng_msg_alloc(&msg, 0)); + // Request ID + TEST_NNG_PASS(nng_msg_append_u32(msg, 0x80000000)); + TEST_NNG_PASS(nng_sendmsg(req, msg, 0)); + + TEST_NNG_PASS(nng_recvmsg(rep, &msg, 0)); + TEST_NNG_PASS(nng_sendmsg(rep, msg, 0)); + + testutil_sleep(100); + + TEST_CHECK(testutil_pollfd(fd) == true); + + // and receiving makes it no longer ready + TEST_NNG_PASS(nng_recvmsg(req, &msg, 0)); + nng_msg_free(msg); + TEST_CHECK(testutil_pollfd(fd) == false); + + TEST_NNG_PASS(nng_close(req)); + TEST_NNG_PASS(nng_close(rep)); +} + +static void +test_xreq_validate_peer(void) +{ + nng_socket s1, s2; + nng_stat * stats; + nng_stat * reject; + char addr[64]; + + testutil_scratch_addr("inproc", sizeof(addr), addr); + + TEST_NNG_PASS(nng_req0_open_raw(&s1)); + TEST_NNG_PASS(nng_req0_open(&s2)); + + TEST_NNG_PASS(nng_listen(s1, addr, NULL, 0)); + TEST_NNG_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK)); + + testutil_sleep(100); + TEST_NNG_PASS(nng_stats_get(&stats)); + + TEST_CHECK(stats != NULL); + TEST_CHECK((reject = nng_stat_find_socket(stats, s1)) != NULL); + TEST_CHECK((reject = nng_stat_find(reject, "reject")) != NULL); + + TEST_CHECK(nng_stat_type(reject) == NNG_STAT_COUNTER); + TEST_CHECK(nng_stat_value(reject) > 0); + + TEST_NNG_PASS(nng_close(s1)); + TEST_NNG_PASS(nng_close(s2)); + nng_stats_free(stats); +} + +static void +test_xreq_recv_aio_stopped(void) +{ + nng_socket req; + nng_aio * aio; + + TEST_NNG_PASS(nng_req0_open_raw(&req)); + TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL)); + + nng_aio_stop(aio); + nng_recv_aio(req, aio); + nng_aio_wait(aio); + TEST_NNG_FAIL(nng_aio_result(aio), NNG_ECANCELED); + TEST_NNG_PASS(nng_close(req)); + nng_aio_free(aio); +} + +static void +test_xreq_recv_garbage(void) +{ + nng_socket rep; + nng_socket req; + nng_msg * m; + uint32_t req_id; + + TEST_NNG_PASS(nng_rep0_open_raw(&rep)); + TEST_NNG_PASS(nng_req0_open_raw(&req)); + TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_RECVTIMEO, 100)); + TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 1000)); + TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000)); + TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_SENDTIMEO, 1000)); + + TEST_NNG_PASS(testutil_marry(req, rep)); + + TEST_NNG_PASS(nng_msg_alloc(&m, 0)); + TEST_NNG_PASS(nng_msg_append_u32(m, 0x80000000)); + TEST_NNG_PASS(nng_sendmsg(req, m, 0)); + + TEST_NNG_PASS(nng_recvmsg(rep, &m, 0)); + + // The message will have a header that contains the 32-bit pipe ID, + // followed by the 32-bit request ID. We will discard the request + // ID before sending it out. + TEST_CHECK(nng_msg_header_len(m) == 8); + TEST_NNG_PASS(nng_msg_header_chop_u32(m, &req_id)); + TEST_CHECK(req_id == 0x80000000); + + TEST_NNG_PASS(nng_sendmsg(rep, m, 0)); + TEST_NNG_FAIL(nng_recvmsg(req, &m, 0), NNG_ETIMEDOUT); + + TEST_NNG_PASS(nng_close(req)); + TEST_NNG_PASS(nng_close(rep)); +} + +static void +test_xreq_close_during_recv(void) +{ + nng_socket rep; + nng_socket req; + nng_msg * m; + nng_pipe p1; + nng_pipe p2; + + TEST_NNG_PASS(nng_rep0_open_raw(&rep)); + TEST_NNG_PASS(nng_req0_open_raw(&req)); + TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000)); + TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 100)); + TEST_NNG_PASS(nng_setopt_int(req, NNG_OPT_RECVBUF, 5)); + TEST_NNG_PASS(nng_setopt_int(rep, NNG_OPT_SENDBUF, 20)); + + TEST_NNG_PASS(testutil_marry_ex(req, rep, &p1, &p2)); + TEST_CHECK(nng_pipe_id(p1) > 0); + TEST_CHECK(nng_pipe_id(p2) > 0); + + for (unsigned i = 0; i < 20; i++) { + TEST_NNG_PASS(nng_msg_alloc(&m, 4)); + TEST_NNG_PASS(nng_msg_header_append_u32(m, nng_pipe_id(p2))); + TEST_NNG_PASS(nng_msg_header_append_u32(m, i | 0x80000000u)); + testutil_sleep(10); + TEST_NNG_PASS(nng_sendmsg(rep, m, 0)); + } + TEST_NNG_PASS(nng_close(req)); + TEST_NNG_PASS(nng_close(rep)); +} + +static void +test_xreq_close_pipe_during_send(void) +{ + nng_socket rep; + nng_socket req; + nng_msg * m; + nng_pipe p1; + nng_pipe p2; + + TEST_NNG_PASS(nng_rep0_open_raw(&rep)); + TEST_NNG_PASS(nng_req0_open_raw(&req)); + TEST_NNG_PASS(nng_setopt_ms(rep, NNG_OPT_RECVTIMEO, 1000)); + TEST_NNG_PASS(nng_setopt_ms(req, NNG_OPT_SENDTIMEO, 100)); + TEST_NNG_PASS(nng_setopt_int(rep, NNG_OPT_RECVBUF, 5)); + TEST_NNG_PASS(nng_setopt_int(req, NNG_OPT_SENDBUF, 20)); + + TEST_NNG_PASS(testutil_marry_ex(req, rep, &p1, &p2)); + TEST_CHECK(nng_pipe_id(p1) > 0); + TEST_CHECK(nng_pipe_id(p2) > 0); + + for (unsigned i = 0; i < 20; i++) { + TEST_NNG_PASS(nng_msg_alloc(&m, 4)); + TEST_NNG_PASS(nng_msg_header_append_u32(m, i | 0x80000000u)); + testutil_sleep(10); + TEST_NNG_PASS(nng_sendmsg(req, m, 0)); + } + + TEST_NNG_PASS(nng_pipe_close(p1)); + TEST_NNG_PASS(nng_close(req)); + TEST_NNG_PASS(nng_close(rep)); +} + +static void +test_xreq_ttl_option(void) +{ + nng_socket rep; + int v; + bool b; + size_t sz = sizeof(v); + const char *opt = NNG_OPT_MAXTTL; + + TEST_NNG_PASS(nng_req0_open_raw(&rep)); + + TEST_NNG_PASS(nng_setopt_int(rep, opt, 1)); + TEST_NNG_FAIL(nng_setopt_int(rep, opt, 0), NNG_EINVAL); + TEST_NNG_FAIL(nng_setopt_int(rep, opt, -1), NNG_EINVAL); + TEST_NNG_FAIL(nng_setopt_int(rep, opt, 256), NNG_EINVAL); + TEST_NNG_PASS(nng_setopt_int(rep, opt, 3)); + TEST_NNG_PASS(nng_getopt_int(rep, opt, &v)); + TEST_CHECK(v == 3); + v = 0; + sz = sizeof(v); + TEST_NNG_PASS(nng_getopt(rep, opt, &v, &sz)); + TEST_CHECK(v == 3); + TEST_CHECK(sz == sizeof(v)); + + TEST_CHECK(nng_setopt(rep, opt, "", 1) == NNG_EINVAL); + sz = 1; + TEST_CHECK(nng_getopt(rep, opt, &v, &sz) == NNG_EINVAL); + TEST_CHECK(nng_setopt_bool(rep, opt, true) == NNG_EBADTYPE); + TEST_CHECK(nng_getopt_bool(rep, opt, &b) == NNG_EBADTYPE); + + TEST_CHECK(nng_close(rep) == 0); +} + +TEST_LIST = { + { "xreq identity", test_xreq_identity }, + { "xreq raw", test_xreq_raw }, + { "xreq no context", test_xreq_no_context }, + { "xreq poll readable", test_xreq_poll_readable }, + { "xreq poll writable", test_xreq_poll_writeable }, + { "xreq validate peer", test_xreq_validate_peer }, + { "xreq recv aio stopped", test_xreq_recv_aio_stopped }, + { "xreq recv garbage", test_xreq_recv_garbage }, + { "xreq close during recv", test_xreq_close_during_recv }, + { "xreq close pipe during send", test_xreq_close_pipe_during_send }, + { "xreq ttl option", test_xreq_ttl_option }, + { NULL, NULL }, +}; -- cgit v1.2.3-70-g09d2