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.

4 files changed, 226 insertions, 3 deletions

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;