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, 54 insertions, 2 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