2 files changed, 230 insertions, 185 deletions

diff --git a/src/protocol/pair1/pair.c b/src/protocol/pair1/pair.c
index b3b64a79..a99a0d17 100644
--- a/src/protocol/pair1/pair.c
+++ b/src/protocol/pair1/pair.c
@@ -17,50 +17,50 @@
 // usually, but it can support a polyamorous mode where a single server can
 // communicate with multiple partners.
 
-#ifndef NNI_PROTO_PAIR_V1
-#define NNI_PROTO_PAIR_V1 NNI_PROTO(1, 1)
-#endif
-
 #define BUMP_STAT(x) nni_stat_inc_atomic(x, 1)
 
 typedef struct pair1_pipe pair1_pipe;
 typedef struct pair1_sock pair1_sock;
 
-static void pair1_sock_getq_cb(void *);
+static void pair1_sock_get_cb(void *);
 static void pair1_pipe_send_cb(void *);
 static void pair1_pipe_recv_cb(void *);
-static void pair1_pipe_getq_cb(void *);
-static void pair1_pipe_putq_cb(void *);
+static void pair1_pipe_get_cb(void *);
+static void pair1_pipe_put_cb(void *);
 static void pair1_pipe_fini(void *);
 
 // pair1_sock is our per-socket protocol private structure.
 struct pair1_sock {
-	nni_msgq *    uwq;
-	nni_msgq *    urq;
-	nni_sock *    nsock;
-	bool          raw;
-	int           ttl;
-	nni_mtx       mtx;
-	nni_idhash *  pipes;
-	nni_list      plist;
-	bool          started;
-	bool          poly;
-	nni_aio *     aio_getq;
-	nni_stat_item stat_poly;
-	nni_stat_item stat_raw;
-	nni_stat_item stat_rejmismatch;
-	nni_stat_item stat_rejinuse;
+	nni_msgq *     uwq;
+	nni_msgq *     urq;
+	nni_sock *     sock;
+	bool           raw;
+	nni_atomic_int ttl;
+	nni_mtx        mtx;
+	nni_idhash *   pipes;
+	nni_list       plist;
+	bool           started;
+	bool           poly;
+	nni_aio      aio_get;
+	nni_stat_item  stat_poly;
+	nni_stat_item  stat_raw;
+	nni_stat_item  stat_reject_mismatch;
+	nni_stat_item  stat_reject_already;
+	nni_stat_item  stat_ttl_drop;
+	nni_stat_item  stat_rx_malformed;
+	nni_stat_item  stat_tx_malformed;
+	nni_stat_item stat_tx_drop;
 };
 
 // pair1_pipe is our per-pipe protocol private structure.
 struct pair1_pipe {
-	nni_pipe *    npipe;
-	pair1_sock *  psock;
-	nni_msgq *    sendq;
-	nni_aio *     aio_send;
-	nni_aio *     aio_recv;
-	nni_aio *     aio_getq;
-	nni_aio *     aio_putq;
+	nni_pipe *    pipe;
+	pair1_sock *  pair;
+	nni_msgq *    send_queue;
+	nni_aio       aio_send;
+	nni_aio       aio_recv;
+	nni_aio       aio_get;
+	nni_aio       aio_put;
 	nni_list_node node;
 };
 
@@ -69,16 +69,15 @@ pair1_sock_fini(void *arg)
 {
 	pair1_sock *s = arg;
 
-	nni_aio_free(s->aio_getq);
+	nni_aio_fini(&s->aio_get);
 	nni_idhash_fini(s->pipes);
 	nni_mtx_fini(&s->mtx);
 }
 
 static int
-pair1_sock_init_impl(void *arg, nni_sock *nsock, bool raw)
+pair1_sock_init_impl(void *arg, nni_sock *sock, bool raw)
 {
 	pair1_sock *s = arg;
-	int         rv;
 
 	if (nni_idhash_init(&s->pipes) != 0) {
 		return (NNG_ENOMEM);
@@ -88,47 +87,69 @@ pair1_sock_init_impl(void *arg, nni_sock *nsock, bool raw)
 	// Raw mode uses this.
 	nni_mtx_init(&s->mtx);
 
-	if ((rv = nni_aio_alloc(&s->aio_getq, pair1_sock_getq_cb, s)) != 0) {
-		pair1_sock_fini(s);
-		return (rv);
-	}
+	nni_aio_init(&s->aio_get, pair1_sock_get_cb, s);
 
 	nni_stat_init_bool(
 	    &s->stat_poly, "polyamorous", "polyamorous mode?", false);
 	nni_stat_set_lock(&s->stat_poly, &s->mtx);
-	nni_sock_add_stat(nsock, &s->stat_poly);
+	nni_sock_add_stat(sock, &s->stat_poly);
 
 	nni_stat_init_bool(&s->stat_raw, "raw", "raw mode?", raw);
-	nni_sock_add_stat(nsock, &s->stat_raw);
+	nni_sock_add_stat(sock, &s->stat_raw);
 
-	nni_stat_init_atomic(&s->stat_rejmismatch, "mismatch",
+	nni_stat_init_atomic(&s->stat_reject_mismatch, "mismatch",
 	    "pipes rejected (protocol mismatch)");
-	nni_sock_add_stat(nsock, &s->stat_rejmismatch);
+	nni_sock_add_stat(sock, &s->stat_reject_mismatch);
 
-	nni_stat_init_atomic(&s->stat_rejinuse, "already",
+	nni_stat_init_atomic(&s->stat_reject_already, "already",
 	    "pipes rejected (already connected)");
-	nni_sock_add_stat(nsock, &s->stat_rejinuse);
+	nni_sock_add_stat(sock, &s->stat_reject_already);
+
+	nni_stat_init_atomic(&s->stat_ttl_drop, "ttl_drop",
+	    "messages dropped due to too many hops");
+	nni_stat_set_unit(&s->stat_ttl_drop, NNG_UNIT_MESSAGES);
+	nni_sock_add_stat(sock, &s->stat_ttl_drop);
+
+	// This can only increment in polyamorous mode.
+	nni_stat_init_atomic(&s->stat_tx_drop, "tx_drop",
+	    "messages dropped undeliverable");
+	nni_stat_set_unit(&s->stat_tx_drop, NNG_UNIT_MESSAGES);
+	nni_sock_add_stat(sock, &s->stat_tx_drop);
+
+	nni_stat_init_atomic(&s->stat_rx_malformed, "rx_malformed",
+	    "malformed messages received");
+	nni_stat_set_unit(&s->stat_rx_malformed, NNG_UNIT_MESSAGES);
+	nni_sock_add_stat(sock, &s->stat_rx_malformed);
+
+	nni_stat_init_atomic(&s->stat_tx_malformed, "tx_malformed",
+	    "malformed messages not sent");
+	nni_stat_set_unit(&s->stat_tx_malformed, NNG_UNIT_MESSAGES);
+	if (raw) {
+		// This stat only makes sense in raw mode.
+		nni_sock_add_stat(sock, &s->stat_tx_malformed);
+	}
 
-	s->nsock = nsock;
+	s->sock  = sock;
 	s->raw   = raw;
 	s->poly  = false;
-	s->uwq   = nni_sock_sendq(nsock);
-	s->urq   = nni_sock_recvq(nsock);
-	s->ttl   = 8;
+	s->uwq   = nni_sock_sendq(sock);
+	s->urq   = nni_sock_recvq(sock);
+	nni_atomic_init(&s->ttl);
+	nni_atomic_set(&s->ttl, 8);
 
 	return (0);
 }
 
 static int
-pair1_sock_init(void *arg, nni_sock *nsock)
+pair1_sock_init(void *arg, nni_sock *sock)
 {
-	return (pair1_sock_init_impl(arg, nsock, false));
+	return (pair1_sock_init_impl(arg, sock, false));
 }
 
 static int
-pair1_sock_init_raw(void *arg, nni_sock *nsock)
+pair1_sock_init_raw(void *arg, nni_sock *sock)
 {
-	return (pair1_sock_init_impl(arg, nsock, true));
+	return (pair1_sock_init_impl(arg, sock, true));
 }
 
 static void
@@ -136,10 +157,10 @@ pair1_pipe_stop(void *arg)
 {
 	pair1_pipe *p = arg;
 
-	nni_aio_stop(p->aio_send);
-	nni_aio_stop(p->aio_recv);
-	nni_aio_stop(p->aio_putq);
-	nni_aio_stop(p->aio_getq);
+	nni_aio_stop(&p->aio_send);
+	nni_aio_stop(&p->aio_recv);
+	nni_aio_stop(&p->aio_put);
+	nni_aio_stop(&p->aio_get);
 }
 
 static void
@@ -147,51 +168,52 @@ pair1_pipe_fini(void *arg)
 {
 	pair1_pipe *p = arg;
 
-	nni_aio_free(p->aio_send);
-	nni_aio_free(p->aio_recv);
-	nni_aio_free(p->aio_putq);
-	nni_aio_free(p->aio_getq);
-	nni_msgq_fini(p->sendq);
+	nni_aio_fini(&p->aio_send);
+	nni_aio_fini(&p->aio_recv);
+	nni_aio_fini(&p->aio_put);
+	nni_aio_fini(&p->aio_get);
+	nni_msgq_fini(p->send_queue);
 }
 
 static int
-pair1_pipe_init(void *arg, nni_pipe *npipe, void *psock)
+pair1_pipe_init(void *arg, nni_pipe *pipe, void *pair)
 {
 	pair1_pipe *p = arg;
 	int         rv;
 
-	if (((rv = nni_msgq_init(&p->sendq, 2)) != 0) ||
-	    ((rv = nni_aio_alloc(&p->aio_send, pair1_pipe_send_cb, p)) != 0) ||
-	    ((rv = nni_aio_alloc(&p->aio_recv, pair1_pipe_recv_cb, p)) != 0) ||
-	    ((rv = nni_aio_alloc(&p->aio_getq, pair1_pipe_getq_cb, p)) != 0) ||
-	    ((rv = nni_aio_alloc(&p->aio_putq, pair1_pipe_putq_cb, p)) != 0)) {
+	nni_aio_init(&p->aio_send, pair1_pipe_send_cb, p);
+	nni_aio_init(&p->aio_recv, pair1_pipe_recv_cb, p);
+	nni_aio_init(&p->aio_get, pair1_pipe_get_cb, p);
+	nni_aio_init(&p->aio_put, pair1_pipe_put_cb, p);
+
+	if ((rv = nni_msgq_init(&p->send_queue, 2)) != 0) {
 		pair1_pipe_fini(p);
-		return (NNG_ENOMEM);
+		return (rv);
 	}
 
-	p->npipe = npipe;
-	p->psock = psock;
+	p->pipe  = pipe;
+	p->pair  = pair;
 
-	return (rv);
+	return (0);
 }
 
 static int
 pair1_pipe_start(void *arg)
 {
 	pair1_pipe *p = arg;
-	pair1_sock *s = p->psock;
+	pair1_sock *s = p->pair;
 	uint32_t    id;
 	int         rv;
 
 	nni_mtx_lock(&s->mtx);
-	if (nni_pipe_peer(p->npipe) != NNI_PROTO_PAIR_V1) {
+	if (nni_pipe_peer(p->pipe) != NNG_PAIR1_PEER) {
 		nni_mtx_unlock(&s->mtx);
-		BUMP_STAT(&s->stat_rejmismatch);
+		BUMP_STAT(&s->stat_reject_mismatch);
 		// Peer protocol mismatch.
 		return (NNG_EPROTO);
 	}
 
-	id = nni_pipe_id(p->npipe);
+	id = nni_pipe_id(p->pipe);
 	if ((rv = nni_idhash_insert(s->pipes, id, p)) != 0) {
 		nni_mtx_unlock(&s->mtx);
 		return (rv);
@@ -200,29 +222,29 @@ pair1_pipe_start(void *arg)
 		if (!nni_list_empty(&s->plist)) {
 			nni_idhash_remove(s->pipes, id);
 			nni_mtx_unlock(&s->mtx);
-			BUMP_STAT(&s->stat_rejinuse);
+			BUMP_STAT(&s->stat_reject_already);
 			return (NNG_EBUSY);
 		}
 	} else {
 		if (!s->started) {
-			nni_msgq_aio_get(s->uwq, s->aio_getq);
+			nni_msgq_aio_get(s->uwq, &s->aio_get);
 		}
 	}
 	nni_list_append(&s->plist, p);
 	s->started = true;
 	nni_mtx_unlock(&s->mtx);
 
-	// Schedule a getq.  In polyamorous mode we get on the per pipe
+	// Schedule a get.  In polyamorous mode we get on the per pipe
 	// send_queue, as the socket distributes to us. In monogamous mode
-	// we bypass and get from the upper writeq directly (saving a
+	// we bypass and get from the upper write queue directly (saving a
 	// set of context switches).
 	if (s->poly) {
-		nni_msgq_aio_get(p->sendq, p->aio_getq);
+		nni_msgq_aio_get(p->send_queue, &p->aio_get);
 	} else {
-		nni_msgq_aio_get(s->uwq, p->aio_getq);
+		nni_msgq_aio_get(s->uwq, &p->aio_get);
 	}
 	// And the pipe read of course.
-	nni_pipe_recv(p->npipe, p->aio_recv);
+	nni_pipe_recv(p->pipe, &p->aio_recv);
 
 	return (0);
 }
@@ -231,100 +253,91 @@ static void
 pair1_pipe_close(void *arg)
 {
 	pair1_pipe *p = arg;
-	pair1_sock *s = p->psock;
+	pair1_sock *s = p->pair;
 
-	nni_aio_close(p->aio_send);
-	nni_aio_close(p->aio_recv);
-	nni_aio_close(p->aio_putq);
-	nni_aio_close(p->aio_getq);
+	nni_aio_close(&p->aio_send);
+	nni_aio_close(&p->aio_recv);
+	nni_aio_close(&p->aio_put);
+	nni_aio_close(&p->aio_get);
 
 	nni_mtx_lock(&s->mtx);
-	nni_idhash_remove(s->pipes, nni_pipe_id(p->npipe));
+	nni_idhash_remove(s->pipes, nni_pipe_id(p->pipe));
 	nni_list_node_remove(&p->node);
 	nni_mtx_unlock(&s->mtx);
 
-	nni_msgq_close(p->sendq);
+	nni_msgq_close(p->send_queue);
 }
 
 static void
 pair1_pipe_recv_cb(void *arg)
 {
 	pair1_pipe *p = arg;
-	pair1_sock *s = p->psock;
+	pair1_sock *s = p->pair;
 	nni_msg *   msg;
 	uint32_t    hdr;
-	nni_pipe *  npipe = p->npipe;
+	nni_pipe *  pipe = p->pipe;
 	size_t      len;
 
-	if (nni_aio_result(p->aio_recv) != 0) {
-		nni_pipe_close(p->npipe);
+	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);
 
 	// Store the pipe ID.
-	nni_msg_set_pipe(msg, nni_pipe_id(p->npipe));
+	nni_msg_set_pipe(msg, nni_pipe_id(p->pipe));
 
 	// If the message is missing the hop count header, scrap it.
-	if (nni_msg_len(msg) < sizeof(uint32_t)) {
-		nni_msg_free(msg);
-		nni_pipe_close(npipe);
-		return;
-	}
-	hdr = nni_msg_trim_u32(msg);
-	if (hdr & 0xffffff00) {
+	if ((nni_msg_len(msg) < sizeof(uint32_t)) ||
+	    ((hdr = nni_msg_trim_u32(msg)) > 0xff)) {
+		BUMP_STAT(&s->stat_rx_malformed);
 		nni_msg_free(msg);
-		nni_pipe_close(npipe);
+		nni_pipe_close(pipe);
 		return;
 	}
+
 	len = nni_msg_len(msg);
 
 	// If we bounced too many times, discard the message, but
 	// keep getting more.
-	if (hdr > (unsigned) s->ttl) {
-		// STAT: bump TTLdrop
+	if ((int) hdr > nni_atomic_get(&s->ttl)) {
+		BUMP_STAT(&s->stat_ttl_drop);
 		nni_msg_free(msg);
-		nni_pipe_recv(npipe, p->aio_recv);
+		nni_pipe_recv(pipe, &p->aio_recv);
 		return;
 	}
 
 	// Store the hop count in the header.
-	if (nni_msg_header_append_u32(msg, hdr) != 0) {
-		// STAT: bump allocfail
-		nni_msg_free(msg);
-		nni_pipe_recv(npipe, p->aio_recv);
-		return;
-	}
+	nni_msg_header_must_append_u32(msg, hdr);
 
 	// Send the message up.
-	nni_aio_set_msg(p->aio_putq, msg);
-	nni_sock_bump_rx(s->nsock, len);
-	nni_msgq_aio_put(s->urq, p->aio_putq);
+	nni_aio_set_msg(&p->aio_put, msg);
+	nni_sock_bump_rx(s->sock, len);
+	nni_msgq_aio_put(s->urq, &p->aio_put);
 }
 
 static void
-pair1_sock_getq_cb(void *arg)
+pair1_sock_get_cb(void *arg)
 {
 	pair1_pipe *p;
 	pair1_sock *s = arg;
 	nni_msg *   msg;
 	uint32_t    id;
 
-	if (nni_aio_result(s->aio_getq) != 0) {
+	if (nni_aio_result(&s->aio_get) != 0) {
 		// Socket closing...
 		return;
 	}
 
-	msg = nni_aio_get_msg(s->aio_getq);
-	nni_aio_set_msg(s->aio_getq, NULL);
-
-	// By definition we are in polyamorous mode.
-	NNI_ASSERT(s->poly);
+	msg = nni_aio_get_msg(&s->aio_get);
+	nni_aio_set_msg(&s->aio_get, NULL);
 
 	p = NULL;
 	nni_mtx_lock(&s->mtx);
+	// By definition we are in polyamorous mode.
+	NNI_ASSERT(s->poly);
 	// If no pipe was requested, we look for any connected peer.
 	if (((id = nni_msg_get_pipe(msg)) == 0) &&
 	    (!nni_list_empty(&s->plist))) {
@@ -332,64 +345,61 @@ pair1_sock_getq_cb(void *arg)
 	} else {
 		nni_idhash_find(s->pipes, id, (void **) &p);
 	}
-	if (p == NULL) {
-		// Pipe not present!
-		nni_mtx_unlock(&s->mtx);
-		nni_msg_free(msg);
-		nni_msgq_aio_get(s->uwq, s->aio_getq);
-		return;
-	}
 
 	// Try a non-blocking send.  If this fails we just discard the
 	// message.  We have to do this to avoid head-of-line blocking
 	// for messages sent to other pipes.  Note that there is some
 	// buffering in the send_queue.
-	if (nni_msgq_tryput(p->sendq, msg) != 0) {
+	if ((p == NULL) || nni_msgq_tryput(p->send_queue, msg) != 0) {
+		BUMP_STAT(&s->stat_tx_drop);
 		nni_msg_free(msg);
 	}
 
 	nni_mtx_unlock(&s->mtx);
-	nni_msgq_aio_get(s->uwq, s->aio_getq);
+	nni_msgq_aio_get(s->uwq, &s->aio_get);
 }
 
 static void
-pair1_pipe_putq_cb(void *arg)
+pair1_pipe_put_cb(void *arg)
 {
 	pair1_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);
-		nni_pipe_close(p->npipe);
+	if (nni_aio_result(&p->aio_put) != 0) {
+		nni_msg_free(nni_aio_get_msg(&p->aio_put));
+		nni_aio_set_msg(&p->aio_put, NULL);
+		nni_pipe_close(p->pipe);
 		return;
 	}
-	nni_pipe_recv(p->npipe, p->aio_recv);
+	nni_pipe_recv(p->pipe, &p->aio_recv);
 }
 
 static void
-pair1_pipe_getq_cb(void *arg)
+pair1_pipe_get_cb(void *arg)
 {
 	pair1_pipe *p = arg;
-	pair1_sock *s = p->psock;
+	pair1_sock *s = p->pair;
 	nni_msg *   msg;
 	uint32_t    hops;
 
-	if (nni_aio_result(p->aio_getq) != 0) {
-		nni_pipe_close(p->npipe);
+	if (nni_aio_result(&p->aio_get) != 0) {
+		nni_pipe_close(p->pipe);
 		return;
 	}
 
-	msg = nni_aio_get_msg(p->aio_getq);
-	nni_aio_set_msg(p->aio_getq, NULL);
+	msg = nni_aio_get_msg(&p->aio_get);
+	nni_aio_set_msg(&p->aio_get, NULL);
 
-	// Raw mode messages have the header already formed, with
-	// a hop count.  Cooked mode messages have no
-	// header so we have to add one.
+	// Raw mode messages have the header already formed, with a hop count.
+	// Cooked mode messages have no header so we have to add one.
 	if (s->raw) {
-		if (nni_msg_header_len(msg) != sizeof(uint32_t)) {
-			goto badmsg;
+		if ((nni_msg_header_len(msg) != sizeof(uint32_t)) ||
+		    ((hops = nni_msg_header_trim_u32(msg)) > 254)) {
+			BUMP_STAT(&s->stat_tx_malformed);
+			nni_msg_free(msg);
+			nni_msgq_aio_get(
+			    s->poly ? p->send_queue : s->uwq, &p->aio_get);
+			return;
 		}
-		hops = nni_msg_header_trim_u32(msg);
 	} else {
 		// Strip off any previously existing header, such as when
 		// replying to messages.
@@ -400,35 +410,28 @@ pair1_pipe_getq_cb(void *arg)
 	hops++;
 
 	// Insert the hops header.
-	if (nni_msg_header_append_u32(msg, hops) != 0) {
-		goto badmsg;
-	}
-
-	nni_aio_set_msg(p->aio_send, msg);
-	nni_pipe_send(p->npipe, p->aio_send);
-	return;
+	nni_msg_header_must_append_u32(msg, hops);
 
-badmsg:
-	nni_msg_free(msg);
-	nni_msgq_aio_get(s->poly ? p->sendq : s->uwq, p->aio_getq);
+	nni_aio_set_msg(&p->aio_send, msg);
+	nni_pipe_send(p->pipe, &p->aio_send);
 }
 
 static void
 pair1_pipe_send_cb(void *arg)
 {
 	pair1_pipe *p = arg;
-	pair1_sock *s = p->psock;
+	pair1_sock *s = p->pair;
 
-	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->npipe);
+	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;
 	}
 
 	// In polyamorous mode, we want to get from the send_queue; in
-	// monogamous we get from upper writeq.
-	nni_msgq_aio_get(s->poly ? p->sendq : s->uwq, p->aio_getq);
+	// monogamous we get from upper write queue.
+	nni_msgq_aio_get(s->poly ? p->send_queue : s->uwq, &p->aio_get);
 }
 
 static void
@@ -441,25 +444,28 @@ static void
 pair1_sock_close(void *arg)
 {
 	pair1_sock *s = arg;
-	nni_aio_close(s->aio_getq);
+	nni_aio_close(&s->aio_get);
 }
 
 static int
-pair1_sock_set_maxttl(void *arg, const void *buf, size_t sz, nni_opt_type t)
+pair1_sock_set_max_ttl(void *arg, const void *buf, size_t sz, nni_opt_type t)
 {
 	pair1_sock *s = arg;
 	int         rv;
-	nni_mtx_lock(&s->mtx); // Have to be locked against recv cb.
-	rv = nni_copyin_int(&s->ttl, buf, sz, 1, 255, t);
-	nni_mtx_unlock(&s->mtx);
+	int         ttl;
+
+	if ((rv = nni_copyin_int(&ttl, buf, sz, 1, 255, t)) == 0) {
+		nni_atomic_set(&s->ttl, ttl);
+	}
+
 	return (rv);
 }
 
 static int
-pair1_sock_get_maxttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
+pair1_sock_get_max_ttl(void *arg, void *buf, size_t *szp, nni_opt_type t)
 {
 	pair1_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
@@ -486,7 +492,7 @@ pair1_sock_send(void *arg, nni_aio *aio)
 {
 	pair1_sock *s = arg;
 
-	nni_sock_bump_tx(s->nsock, nni_msg_len(nni_aio_get_msg(aio)));
+	nni_sock_bump_tx(s->sock, nni_msg_len(nni_aio_get_msg(aio)));
 	nni_msgq_aio_put(s->uwq, aio);
 }
 
@@ -510,8 +516,8 @@ static nni_proto_pipe_ops pair1_pipe_ops = {
 static nni_option pair1_sock_options[] = {
 	{
 	    .o_name = NNG_OPT_MAXTTL,
-	    .o_get  = pair1_sock_get_maxttl,
-	    .o_set  = pair1_sock_set_maxttl,
+	    .o_get  = pair1_sock_get_max_ttl,
+	    .o_set  = pair1_sock_set_max_ttl,
 	},
 	{
 	    .o_name = NNG_OPT_PAIR1_POLY,
@@ -537,17 +543,17 @@ static nni_proto_sock_ops pair1_sock_ops = {
 
 static nni_proto pair1_proto = {
 	.proto_version  = NNI_PROTOCOL_VERSION,
-	.proto_self     = { NNI_PROTO_PAIR_V1, "pair1" },
-	.proto_peer     = { NNI_PROTO_PAIR_V1, "pair1" },
+	.proto_self     = { NNG_PAIR1_SELF, NNG_PAIR1_SELF_NAME },
+	.proto_peer     = { NNG_PAIR1_PEER, NNG_PAIR1_PEER_NAME },
 	.proto_flags    = NNI_PROTO_FLAG_SNDRCV,
 	.proto_sock_ops = &pair1_sock_ops,
 	.proto_pipe_ops = &pair1_pipe_ops,
 };
 
 int
-nng_pair1_open(nng_socket *sidp)
+nng_pair1_open(nng_socket *sock)
 {
-	return (nni_proto_open(sidp, &pair1_proto));
+	return (nni_proto_open(sock, &pair1_proto));
 }
 
 static nni_proto_sock_ops pair1_sock_ops_raw = {
@@ -563,15 +569,15 @@ static nni_proto_sock_ops pair1_sock_ops_raw = {
 
 static nni_proto pair1_proto_raw = {
 	.proto_version  = NNI_PROTOCOL_VERSION,
-	.proto_self     = { NNI_PROTO_PAIR_V1, "pair1" },
-	.proto_peer     = { NNI_PROTO_PAIR_V1, "pair1" },
+	.proto_self     = { NNG_PAIR1_SELF, NNG_PAIR1_SELF_NAME },
+	.proto_peer     = { NNG_PAIR1_PEER, NNG_PAIR1_PEER_NAME },
 	.proto_flags    = NNI_PROTO_FLAG_SNDRCV | NNI_PROTO_FLAG_RAW,
 	.proto_sock_ops = &pair1_sock_ops_raw,
 	.proto_pipe_ops = &pair1_pipe_ops,
 };
 
 int
-nng_pair1_open_raw(nng_socket *sidp)
+nng_pair1_open_raw(nng_socket *sock)
 {
-	return (nni_proto_open(sidp, &pair1_proto_raw));
+	return (nni_proto_open(sock, &pair1_proto_raw));
 }
diff --git a/src/protocol/pair1/pair1_test.c b/src/protocol/pair1/pair1_test.c
index 998d408a..08f808a5 100644
--- a/src/protocol/pair1/pair1_test.c
+++ b/src/protocol/pair1/pair1_test.c
@@ -1,5 +1,5 @@
 //
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 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
@@ -12,6 +12,7 @@
 
 #include <nng/nng.h>
 #include <nng/protocol/pair1/pair.h>
+#include <nng/protocol/pair0/pair.h>
 
 #include <testutil.h>
 
@@ -146,7 +147,7 @@ test_mono_raw_exchange(void)
 	TEST_NNG_PASS(testutil_marry(s1, c1));
 
 	nng_pipe p = NNG_PIPE_INITIALIZER;
-	TEST_NNG_PASS(nng_msg_alloc(&msg, 0) );
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
 	APPEND_STR(msg, "GAMMA");
 	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 1));
 	TEST_CHECK(nng_msg_header_len(msg) == sizeof(uint32_t));
@@ -215,6 +216,12 @@ test_mono_raw_header(void)
 	TEST_CHECK(nng_sendmsg(c1, msg, 0) == 0);
 	TEST_CHECK(nng_recvmsg(s1, &msg, 0) == NNG_ETIMEDOUT);
 
+	// Header with no chance to add another hop gets dropped
+	TEST_NNG_PASS(nng_msg_alloc(&msg, 0));
+	TEST_NNG_PASS(nng_msg_header_append_u32(msg, 0xff));
+	TEST_NNG_PASS(nng_sendmsg(c1, msg, 0));
+	TEST_NNG_FAIL(nng_recvmsg(s1, &msg, 0), NNG_ETIMEDOUT);
+
 	// With the same bits clear it works
 	TEST_CHECK(nng_msg_alloc(&msg, 0) == 0);
 	TEST_CHECK(nng_msg_append_u32(msg, 0xFEEDFACE) == 0);
@@ -591,6 +598,37 @@ test_ttl(void)
 	TEST_CHECK(nng_close(c1) == 0);
 }
 
+void
+test_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_pair1_open(&s1));
+	TEST_NNG_PASS(nng_pair0_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);
+}
+
 TEST_LIST = {
 	{ "pair1 monogamous cooked", test_mono_cooked },
 	{ "pair1 monogamous faithful", test_mono_faithful },
@@ -604,6 +642,7 @@ TEST_LIST = {
 	{ "pair1 polyamorous raw", test_poly_raw },
 	{ "pair1 raw", test_raw },
 	{ "pair1 ttl", test_ttl },
+	{ "pair1 validate peer", test_validate_peer },
 
 	{ NULL, NULL },
 };