aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/core/defs.h2
-rw-r--r--src/core/message.c23
-rw-r--r--src/core/message.h7
-rw-r--r--src/core/platform.h24
-rw-r--r--src/platform/posix/posix_atomic.c151
-rw-r--r--src/platform/posix/posix_impl.h8
-rw-r--r--src/platform/windows/win_impl.h4
-rw-r--r--src/platform/windows/win_thread.c66
-rw-r--r--src/protocol/reqrep0/CMakeLists.txt1
-rw-r--r--src/protocol/reqrep0/rep.c39
-rw-r--r--src/protocol/reqrep0/rep_test.c6
-rw-r--r--src/protocol/reqrep0/req.c92
-rw-r--r--src/protocol/reqrep0/xrep.c17
-rw-r--r--src/protocol/reqrep0/xrep_test.c32
-rw-r--r--src/protocol/reqrep0/xreq.c130
-rw-r--r--src/protocol/reqrep0/xreq_test.c333
-rw-r--r--tests/testutil.c19
-rw-r--r--tests/testutil.h6
18 files changed, 792 insertions, 168 deletions
diff --git a/src/core/defs.h b/src/core/defs.h
index 0e0deac2..9e320e1d 100644
--- a/src/core/defs.h
+++ b/src/core/defs.h
@@ -23,7 +23,7 @@
if (!(x)) \
nni_panic("%s: %d: assert err: %s", __FILE__, __LINE__, #x)
#else
-#define NNI_ASSERT(x)
+#define NNI_ASSERT(x) (0)
#endif
// Returns the size of an array in elements. (Convenience.)
diff --git a/src/core/message.c b/src/core/message.c
index 58baa4b7..fd14c171 100644
--- a/src/core/message.c
+++ b/src/core/message.c
@@ -650,6 +650,23 @@ nni_msg_header_chop(nni_msg *m, size_t len)
return (nni_chunk_##z##_u##x(&m->m_header, v)); \
}
+#define DEF_MSG_MUST_ADD_N(z, x) \
+ void nni_msg_must_##z##_u##x(nni_msg *m, uint##x##_t v) \
+ { \
+ int rv; \
+ if ((rv = nni_msg_##z##_u##x(m, v)) != 0) { \
+ nni_panic("nni_msg_%s_u%s failed: %d", #z, #x, rv); \
+ } \
+ } \
+ void nni_msg_header_must_##z##_u##x(nni_msg *m, uint##x##_t v) \
+ { \
+ int rv; \
+ if ((rv = nni_msg_header_##z##_u##x(m, v)) != 0) { \
+ nni_panic( \
+ "nni_msg_header_%s_u%s failed: %d", #z, #x, rv); \
+ } \
+ }
+
#define DEF_MSG_REM_N(z, x) \
uint##x##_t nni_msg_##z##_u##x(nni_msg *m) \
{ \
@@ -662,17 +679,23 @@ nni_msg_header_chop(nni_msg *m, size_t len)
#define DEF_MSG_ADD(op) \
DEF_MSG_ADD_N(op, 16) DEF_MSG_ADD_N(op, 32) DEF_MSG_ADD_N(op, 64)
+
+#define DEF_MSG_MUST_ADD(op) DEF_MSG_MUST_ADD_N(op, 32)
+
#define DEF_MSG_REM(op) \
DEF_MSG_REM_N(op, 16) DEF_MSG_REM_N(op, 32) DEF_MSG_REM_N(op, 64)
DEF_MSG_ADD(append)
+DEF_MSG_MUST_ADD(append)
DEF_MSG_ADD(insert)
DEF_MSG_REM(chop)
DEF_MSG_REM(trim)
#undef DEF_MSG_ADD_N
+#undef DEF_MUST_ADD_N
#undef DEF_MSG_REM_N
#undef DEF_MSG_ADD
+#undef DEF_MSG_MUST_ADD
#undef DEF_MSG_REM
void
diff --git a/src/core/message.h b/src/core/message.h
index 5ec1d9a1..c96bae99 100644
--- a/src/core/message.h
+++ b/src/core/message.h
@@ -62,4 +62,11 @@ extern uint64_t nni_msg_header_chop_u64(nni_msg *);
extern void nni_msg_set_pipe(nni_msg *, uint32_t);
extern uint32_t nni_msg_get_pipe(const nni_msg *);
+// These should only be used when the transport or protocol is absolutely
+// certain that there is adequate room. There is about 32 bytes of
+// header and trailer space for a newly allocated message, and transports
+// should generally not be burning more than half that.
+extern void nni_msg_must_append_u32(nni_msg *, uint32_t);
+extern void nni_msg_header_must_append_u32(nni_msg *, uint32_t);
+
#endif // CORE_SOCKET_H
diff --git a/src/core/platform.h b/src/core/platform.h
index 53d25137..70420061 100644
--- a/src/core/platform.h
+++ b/src/core/platform.h
@@ -192,7 +192,29 @@ extern void nni_atomic_inc64(nni_atomic_u64 *);
// nni_atomic_cas64 is a compare and swap. The second argument is the
// value to compare against, and the third is the new value. Returns
// true if the value was set.
-extern bool nni_atomic_cas64(nni_atomic_u64 *, uint64_t, uint64_t);
+extern bool nni_atomic_cas64(nni_atomic_u64 *, uint64_t, uint64_t);
+
+// In a lot of circumstances, we want a simple atomic reference count,
+// or atomic tunable values for integers like queue lengths or TTLs.
+// These native integer forms should be preferred over the 64 bit versions
+// unless larger bit sizes are truly needed. They will be more efficient
+// on many platforms.
+typedef struct nni_atomic_int nni_atomic_int;
+
+extern void nni_atomic_init(nni_atomic_int *);
+extern void nni_atomic_add(nni_atomic_int *, int);
+extern void nni_atomic_sub(nni_atomic_int *, int);
+extern int nni_atomic_get(nni_atomic_int *);
+extern void nni_atomic_set(nni_atomic_int *, int);
+extern int nni_atomic_swap(nni_atomic_int *, int);
+extern int nni_atomic_dec_nv(nni_atomic_int *);
+extern void nni_atomic_dec(nni_atomic_int *);
+extern void nni_atomic_inc(nni_atomic_int *);
+
+// nni_atomic_cas is a compare and swap. The second argument is the
+// value to compare against, and the third is the new value. Returns
+// true if the value was set.
+extern bool nni_atomic_cas(nni_atomic_int *, int, int);
//
// Clock Support
diff --git a/src/platform/posix/posix_atomic.c b/src/platform/posix/posix_atomic.c
index 0d866071..6b110712 100644
--- a/src/platform/posix/posix_atomic.c
+++ b/src/platform/posix/posix_atomic.c
@@ -55,6 +55,66 @@ nni_atomic_init_bool(nni_atomic_bool *v)
}
void
+nni_atomic_init(nni_atomic_int *v)
+{
+ atomic_init(&v->v, 0);
+}
+
+void
+nni_atomic_add(nni_atomic_int *v, int bump)
+{
+ (void) atomic_fetch_add_explicit(&v->v, bump, memory_order_relaxed);
+}
+
+void
+nni_atomic_sub(nni_atomic_int *v, int bump)
+{
+ (void) atomic_fetch_sub_explicit(&v->v, bump, memory_order_relaxed);
+}
+
+int
+nni_atomic_get(nni_atomic_int *v)
+{
+ return (atomic_load(&v->v));
+}
+
+void
+nni_atomic_set(nni_atomic_int *v, int i)
+{
+ return (atomic_store(&v->v, i));
+}
+
+int
+nni_atomic_swap(nni_atomic_int *v, int i)
+{
+ return (atomic_exchange(&v->v, i));
+}
+
+void
+nni_atomic_inc(nni_atomic_int *v)
+{
+ atomic_fetch_add(&v->v, 1);
+}
+
+void
+nni_atomic_dec(nni_atomic_int *v)
+{
+ atomic_fetch_sub(&v->v, 1);
+}
+
+int
+nni_atomic_dec_nv(nni_atomic_int *v)
+{
+ return (atomic_fetch_sub(&v->v, 1) - 1);
+}
+
+bool
+nni_atomic_cas(nni_atomic_int *v, int comp, int new)
+{
+ return (atomic_compare_exchange_strong(&v->v, &comp, new));
+}
+
+void
nni_atomic_add64(nni_atomic_u64 *v, uint64_t bump)
{
(void) atomic_fetch_add_explicit(
@@ -261,6 +321,97 @@ nni_atomic_cas64(nni_atomic_u64 *v, uint64_t comp, uint64_t new)
return (result);
}
+void
+nni_atomic_init(nni_atomic_int *v)
+{
+ atomic_init(&v->v, 0);
+}
+
+void
+nni_atomic_add(nni_atomic_int *v, int bump)
+{
+ pthread_mutex_lock(&plat_atomic_lock);
+ v->v += bump;
+ pthread_mutex_unlock(&plat_atomic_lock);
+}
+
+void
+nni_atomic_sub(nni_atomic_int *v, int bump)
+{
+ pthread_mutex_lock(&plat_atomic_lock);
+ v->v -= bump;
+ pthread_mutex_unlock(&plat_atomic_lock);
+}
+
+int
+nni_atomic_get(nni_atomic_int *v)
+{
+ int rv;
+ pthread_mutex_lock(&plat_atomic_lock);
+ rv = v->v;
+ pthread_mutex_unlock(&plat_atomic_lock);
+ return (rv);
+}
+
+void
+nni_atomic_set(nni_atomic_int *v, int i)
+{
+ pthread_mutex_lock(&plat_atomic_lock);
+ v->v = i;
+ pthread_mutex_unlock(&plat_atomic_lock);
+}
+
+int
+nni_atomic_swap(nni_atomic_int *v, int i)
+{
+ int rv;
+ pthread_mutex_lock(&plat_atomic_lock);
+ rv = v->v;
+ v->v = i;
+ pthread_mutex_unlock(&plat_atomic_lock);
+ return (rv);
+}
+
+void
+nni_atomic_inc(nni_atomic_int *v)
+{
+ pthread_mutex_lock(&plat_atomic_lock);
+ v->v++;
+ pthread_mutex_unlock(&plat_atomic_lock);
+}
+
+void
+nni_atomic_dec(nni_atomic_int *v)
+{
+ pthread_mutex_lock(&plat_atomic_lock);
+ v->v--;
+ pthread_mutex_unlock(&plat_atomic_lock);
+}
+
+int
+nni_atomic_dec_nv(nni_atomic_int *v)
+{
+ int nv;
+ pthread_mutex_lock(&plat_atomic_lock);
+ v->v--;
+ nv = v->v;
+ pthread_mutex_unlock(&plat_atomic_lock);
+ return (nv);
+}
+
+bool
+nni_atomic_cas(nni_atomic_int *v, int comp, int new)
+{
+ bool result = false;
+ pthread_mutex_lock(&plat_atomic_lock);
+ if (v->v == comp) {
+ v->v = new;
+ result = true;
+ }
+ pthread_mutex_unlock(&plat_atomic_lock);
+ return (result);
+}
+
#endif
#endif // NNG_PLATFORM_POSIX
diff --git a/src/platform/posix/posix_impl.h b/src/platform/posix/posix_impl.h
index a1cb62c5..3ec66b6b 100644
--- a/src/platform/posix/posix_impl.h
+++ b/src/platform/posix/posix_impl.h
@@ -83,6 +83,10 @@ struct nni_atomic_flag {
atomic_flag f;
};
+struct nni_atomic_int {
+ atomic_int v;
+};
+
struct nni_atomic_u64 {
atomic_uint_fast64_t v;
};
@@ -100,6 +104,10 @@ struct nni_atomic_bol {
bool b;
};
+struct nni_atomic_int {
+ int v;
+};
+
struct nni_atomic_u64 {
uint64_t v;
};
diff --git a/src/platform/windows/win_impl.h b/src/platform/windows/win_impl.h
index befcd185..17260676 100644
--- a/src/platform/windows/win_impl.h
+++ b/src/platform/windows/win_impl.h
@@ -54,6 +54,10 @@ struct nni_atomic_bool {
LONG v;
};
+struct nni_atomic_int {
+ LONG v;
+};
+
struct nni_atomic_u64 {
LONGLONG v;
};
diff --git a/src/platform/windows/win_thread.c b/src/platform/windows/win_thread.c
index 0b77e078..5d98ddde 100644
--- a/src/platform/windows/win_thread.c
+++ b/src/platform/windows/win_thread.c
@@ -1,5 +1,5 @@
//
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
//
// This software is supplied under the terms of the MIT License, a
@@ -8,13 +8,13 @@
// found online at https://opensource.org/licenses/MIT.
//
-// POSIX threads.
+// Windows threads.
#include "core/nng_impl.h"
#ifdef NNG_PLATFORM_WINDOWS
-// mingw does not define InterlockedAddNoFence64, use the mingw equivelent
+// mingw does not define InterlockedAddNoFence64, use the mingw equivalent
#if defined(__MINGW32__) || defined(__MINGW64__)
#define InterlockedAddNoFence64(a, b) \
__atomic_add_fetch(a, b, __ATOMIC_RELAXED)
@@ -221,6 +221,66 @@ nni_atomic_cas64(nni_atomic_u64 *v, uint64_t comp, uint64_t new)
return (old == comp);
}
+
+void
+nni_atomic_add(nni_atomic_int *v, int bump)
+{
+ InterlockedAddNoFence(&v->v, (LONG) bump);
+}
+
+void
+nni_atomic_sub(nni_atomic_int *v, int bump)
+{
+ // Windows lacks a sub, so we add the negative.
+ InterlockedAddNoFence(&v->v, (LONG) -bump);
+}
+
+int
+nni_atomic_get(nni_atomic_int *v)
+{
+
+ return (InterlockedExchangeAdd(&v->v, 0));
+}
+
+void
+nni_atomic_set(nni_atomic_int *v, int i)
+{
+ (void) InterlockedExchange(&v->v, (LONG) i);
+}
+
+int
+nni_atomic_swap(nni_atomic_int *v, int i)
+{
+ return (InterlockedExchange(&v->v, (LONG) i));
+}
+
+void
+nni_atomic_init(nni_atomic_int *v)
+{
+ InterlockedExchange(&v->v, 0);
+}
+
+void
+nni_atomic_inc(nni_atomic_int *v)
+{
+ (void) InterlockedIncrementAcquire(&v->v);
+}
+
+int
+nni_atomic_dec_nv(nni_atomic_int *v)
+{
+ return (InterlockedDecrementRelease(&v->v));
+}
+
+bool
+nni_atomic_cas(nni_atomic_int *v, int comp, int new)
+{
+ int old;
+ old = InterlockedCompareExchange(&v->v, (LONG)new, (LONG)comp);
+ return (old == comp);
+}
+
+
static unsigned int __stdcall nni_plat_thr_main(void *arg)
{
nni_plat_thr *thr = arg;
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 <stdio.h>
-#include <stdlib.h>
#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 <stdio.h>
#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. <info@staysail.tech>
+//
+// 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 <string.h>
+
+#include <nng/nng.h>
+#include <nng/protocol/reqrep0/rep.h>
+#include <nng/protocol/reqrep0/req.h>
+
+#include <acutest.h>
+#include <testutil.h>
+
+#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 },
+};
diff --git a/tests/testutil.c b/tests/testutil.c
index 36999448..356eb333 100644
--- a/tests/testutil.c
+++ b/tests/testutil.c
@@ -1,5 +1,5 @@
//
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
//
// This software is supplied under the terms of the MIT License, a
@@ -303,20 +303,23 @@ struct marriage_notice {
int s2;
int cnt1;
int cnt2;
+ nng_pipe p1;
+ nng_pipe p2;
};
static void
married(nng_pipe p, nng_pipe_ev ev, void *arg)
{
struct marriage_notice *notice = arg;
- (void) p;
(void) ev;
nng_mtx_lock(notice->mx);
if (nng_socket_id(nng_pipe_socket(p)) == notice->s1) {
notice->cnt1++;
+ notice->p1 = p;
} else if (nng_socket_id(nng_pipe_socket(p)) == notice->s2) {
notice->cnt2++;
+ notice->p2 = p;
}
nng_cv_wake(notice->cv);
nng_mtx_unlock(notice->mx);
@@ -325,6 +328,12 @@ married(nng_pipe p, nng_pipe_ev ev, void *arg)
int
testutil_marry(nng_socket s1, nng_socket s2)
{
+ return (testutil_marry_ex(s1, s2, NULL, NULL));
+}
+
+int
+testutil_marry_ex(nng_socket s1, nng_socket s2, nng_pipe *p1, nng_pipe *p2)
+{
struct marriage_notice note;
nng_time timeout;
int rv;
@@ -357,6 +366,12 @@ testutil_marry(nng_socket s1, nng_socket s2)
}
}
nng_mtx_unlock(note.mx);
+ if (p1 != NULL) {
+ *p1 = note.p1;
+ }
+ if (p2 != NULL) {
+ *p2 = note.p2;
+ }
done:
nng_pipe_notify(s1, NNG_PIPE_EV_ADD_POST, NULL, NULL);
diff --git a/tests/testutil.h b/tests/testutil.h
index 008c8061..16e665f2 100644
--- a/tests/testutil.h
+++ b/tests/testutil.h
@@ -1,5 +1,5 @@
//
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
//
// This software is supplied under the terms of the MIT License, a
// copy of which should be located in the distribution where this
@@ -50,6 +50,10 @@ extern void testutil_scratch_addr(const char *, size_t, char *);
// are fully connected.
extern int testutil_marry(nng_socket, nng_socket);
+// testutil_marry_ex is like testutil_marry, but returns the pipes that
+// were connected. The pipe pointers may be NULL if not needed.
+extern int testutil_marry_ex(nng_socket, nng_socket, nng_pipe *, nng_pipe *);
+
// TEST_NNG_PASS tests for NNG success. It reports the failure if it
// did not.
#define TEST_NNG_PASS(cond) \
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 18:01:37 +0000