fixes #808 Very slow PAIR performance compared to nanomsg

This is only the pair v1 protocol.  Pair v0 and polyamorous mode
still have work to do.  We probably won't "fix" the performance for
poly mode since that's deprecated anyway.

3 files changed, 37 insertions, 2 deletions

diff --git a/src/core/message.c b/src/core/message.c
index 79031400..824fc079 100644
--- a/src/core/message.c
+++ b/src/core/message.c
@@ -1,5 +1,5 @@
 //
-// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2021 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
@@ -326,6 +326,12 @@ nni_msg_unique(nni_msg *m)
 	return (m2);
 }
 
+bool
+nni_msg_shared(nni_msg *m)
+{
+	return (nni_atomic_get(&m->m_refcnt) > 1);
+}
+
 // nni_msg_pull_up ensures that the message is unique, and that any header
 // is merged with the message.  The main purpose of doing this is to break
 // up the inproc binding -- protocols send messages to inproc with a
@@ -575,6 +581,24 @@ nni_msg_header_append_u32(nni_msg *m, uint32_t val)
 	m->m_header_len += sizeof(val);
 }
 
+uint32_t
+nni_msg_header_peek_u32(nni_msg *m)
+{
+	uint32_t val;
+	uint8_t *dst;
+	dst = (void *) m->m_header_buf;
+	NNI_GET32(dst, val);
+	return (val);
+}
+
+void
+nni_msg_header_poke_u32(nni_msg *m, uint32_t val)
+{
+	uint8_t *dst;
+	dst = (void *) m->m_header_buf;
+	NNI_PUT32(dst, val);
+}
+
 void
 nni_msg_clear(nni_msg *m)
 {
diff --git a/src/core/message.h b/src/core/message.h
index 03991166..6400fd55 100644
--- a/src/core/message.h
+++ b/src/core/message.h
@@ -1,5 +1,5 @@
 //
-// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2021 Staysail Systems, Inc. <info@staysail.tech>
 // Copyright 2017 Capitar IT Group BV <info@capitar.com>
 //
 // This software is supplied under the terms of the MIT License, a
@@ -36,6 +36,11 @@ extern void     nni_msg_dump(const char *, const nni_msg *);
 extern void     nni_msg_header_append_u32(nni_msg *, uint32_t);
 extern uint32_t nni_msg_header_trim_u32(nni_msg *);
 extern uint32_t nni_msg_trim_u32(nni_msg *);
+// Peek and poke variants just access the first uint32 in the
+// header.  This is useful when incrementing reference counts, etc.
+// It's faster than trim and append, but logically equivalent.
+extern uint32_t nni_msg_header_peek_u32(nni_msg *);
+extern void     nni_msg_header_poke_u32(nni_msg *, uint32_t);
 extern void     nni_msg_set_pipe(nni_msg *, uint32_t);
 extern uint32_t nni_msg_get_pipe(const nni_msg *);
 
@@ -46,6 +51,8 @@ extern uint32_t nni_msg_get_pipe(const nni_msg *);
 // Failure to do so will likely result in corruption.
 extern void     nni_msg_clone(nni_msg *);
 extern nni_msg *nni_msg_unique(nni_msg *);
+extern bool     nni_msg_shared(nni_msg *);
+
 // nni_msg_pull_up ensures that the message is unique, and that any
 // header present is "pulled up" into the message body.  If the function
 // cannot do this for any reason (out of space in the body), then NULL
diff --git a/src/core/reconnect_test.c b/src/core/reconnect_test.c
index 308a3f78..120f0517 100644
--- a/src/core/reconnect_test.c
+++ b/src/core/reconnect_test.c
@@ -64,6 +64,10 @@ test_reconnect(void)
 	// Close the listener
 	NUTS_PASS(nng_listener_close(l));
 
+	// We need to wait 100 ms, or so, to allow the receiver to
+	// the disconnect.
+	NUTS_SLEEP(100);
+
 	NUTS_PASS(nng_listen(s1, addr, &l, 0));
 	NUTS_SEND(s1, "again");
 	NUTS_RECV(s2, "again");