Allocate AIOs dynamically.

We allocate AIO structures dynamically, so that we can use them
abstractly in more places without inlining them.  This will be used
for the ZeroTier transport to allow us to create operations consisting
of just the AIO.  Furthermore, we provide accessors for some of the
aio members, in the hopes that we will be able to wrap these for
"safe" version of the AIO capability to export to applications, and
to protocol and transport implementors.

While here we cleaned up the protocol details to use consistently
shorter names (no nni_ prefix for static symbols needed), and we
also fixed a bug in the surveyor code.

2 files changed, 462 insertions, 464 deletions

diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index cd33d019..14a3e46b 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -17,18 +17,18 @@
 // request-reply pair.  This is useful for building RPC servers, for
 // example.
 
-typedef struct nni_rep_pipe nni_rep_pipe;
-typedef struct nni_rep_sock nni_rep_sock;
-
-static void nni_rep_sock_getq_cb(void *);
-static void nni_rep_pipe_getq_cb(void *);
-static void nni_rep_pipe_putq_cb(void *);
-static void nni_rep_pipe_send_cb(void *);
-static void nni_rep_pipe_recv_cb(void *);
-static void nni_rep_pipe_fini(void *);
-
-// An nni_rep_sock is our per-socket protocol private structure.
-struct nni_rep_sock {
+typedef struct rep_pipe rep_pipe;
+typedef struct rep_sock rep_sock;
+
+static void rep_sock_getq_cb(void *);
+static void rep_pipe_getq_cb(void *);
+static void rep_pipe_putq_cb(void *);
+static void rep_pipe_send_cb(void *);
+static void rep_pipe_recv_cb(void *);
+static void rep_pipe_fini(void *);
+
+// A rep_sock is our per-socket protocol private structure.
+struct rep_sock {
 	nni_sock *  sock;
 	nni_msgq *  uwq;
 	nni_msgq *  urq;
@@ -37,179 +37,176 @@ struct nni_rep_sock {
 	nni_idhash *pipes;
 	char *      btrace;
 	size_t      btrace_len;
-	nni_aio     aio_getq;
+	nni_aio *   aio_getq;
 };
 
-// An nni_rep_pipe is our per-pipe protocol private structure.
-struct nni_rep_pipe {
-	nni_pipe *    pipe;
-	nni_rep_sock *rep;
-	nni_msgq *    sendq;
-	nni_aio       aio_getq;
-	nni_aio       aio_send;
-	nni_aio       aio_recv;
-	nni_aio       aio_putq;
+// A rep_pipe is our per-pipe protocol private structure.
+struct rep_pipe {
+	nni_pipe *pipe;
+	rep_sock *rep;
+	nni_msgq *sendq;
+	nni_aio * aio_getq;
+	nni_aio * aio_send;
+	nni_aio * aio_recv;
+	nni_aio * aio_putq;
 };
 
 static void
-nni_rep_sock_fini(void *arg)
+rep_sock_fini(void *arg)
 {
-	nni_rep_sock *rep = arg;
+	rep_sock *s = arg;
 
-	nni_aio_stop(&rep->aio_getq);
-	nni_aio_fini(&rep->aio_getq);
-	nni_idhash_fini(rep->pipes);
-	if (rep->btrace != NULL) {
-		nni_free(rep->btrace, rep->btrace_len);
+	nni_aio_stop(s->aio_getq);
+	nni_aio_fini(s->aio_getq);
+	nni_idhash_fini(s->pipes);
+	if (s->btrace != NULL) {
+		nni_free(s->btrace, s->btrace_len);
 	}
-	NNI_FREE_STRUCT(rep);
+	NNI_FREE_STRUCT(s);
 }
 
 static int
-nni_rep_sock_init(void **repp, nni_sock *sock)
+rep_sock_init(void **sp, nni_sock *sock)
 {
-	nni_rep_sock *rep;
-	int           rv;
+	rep_sock *s;
+	int       rv;
 
-	if ((rep = NNI_ALLOC_STRUCT(rep)) == NULL) {
+	if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	if ((rv = nni_idhash_init(&rep->pipes)) != 0) {
-		NNI_FREE_STRUCT(rep);
+	if (((rv = nni_idhash_init(&s->pipes)) != 0) ||
+	    ((rv = nni_aio_init(&s->aio_getq, rep_sock_getq_cb, s)) != 0)) {
+		rep_sock_fini(s);
 		return (rv);
 	}
 
-	rep->ttl        = 8; // Per RFC
-	rep->sock       = sock;
-	rep->raw        = 0;
-	rep->btrace     = NULL;
-	rep->btrace_len = 0;
+	s->ttl        = 8; // Per RFC
+	s->sock       = sock;
+	s->raw        = 0;
+	s->btrace     = NULL;
+	s->btrace_len = 0;
+	s->uwq        = nni_sock_sendq(sock);
+	s->urq        = nni_sock_recvq(sock);
 
-	nni_aio_init(&rep->aio_getq, nni_rep_sock_getq_cb, rep);
-
-	rep->uwq = nni_sock_sendq(sock);
-	rep->urq = nni_sock_recvq(sock);
-
-	*repp = rep;
+	*sp = s;
 	nni_sock_senderr(sock, NNG_ESTATE);
 
 	return (0);
 }
 
 static void
-nni_rep_sock_open(void *arg)
+rep_sock_open(void *arg)
 {
-	nni_rep_sock *rep = arg;
+	rep_sock *s = arg;
 
-	nni_msgq_aio_get(rep->uwq, &rep->aio_getq);
+	nni_msgq_aio_get(s->uwq, s->aio_getq);
 }
 
 static void
-nni_rep_sock_close(void *arg)
+rep_sock_close(void *arg)
 {
-	nni_rep_sock *rep = arg;
+	rep_sock *s = arg;
+
+	nni_aio_cancel(s->aio_getq, NNG_ECLOSED);
+}
 
-	nni_aio_cancel(&rep->aio_getq, NNG_ECLOSED);
+static void
+rep_pipe_fini(void *arg)
+{
+	rep_pipe *p = arg;
+
+	nni_aio_fini(p->aio_getq);
+	nni_aio_fini(p->aio_send);
+	nni_aio_fini(p->aio_recv);
+	nni_aio_fini(p->aio_putq);
+	nni_msgq_fini(p->sendq);
+	NNI_FREE_STRUCT(p);
 }
 
 static int
-nni_rep_pipe_init(void **rpp, nni_pipe *pipe, void *rsock)
+rep_pipe_init(void **pp, nni_pipe *pipe, void *s)
 {
-	nni_rep_pipe *rp;
-	int           rv;
+	rep_pipe *p;
+	int       rv;
 
-	if ((rp = NNI_ALLOC_STRUCT(rp)) == NULL) {
+	if ((p = NNI_ALLOC_STRUCT(p)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	if ((rv = nni_msgq_init(&rp->sendq, 2)) != 0) {
-		NNI_FREE_STRUCT(rp);
+	if (((rv = nni_msgq_init(&p->sendq, 2)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_getq, rep_pipe_getq_cb, p)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_send, rep_pipe_send_cb, p)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_recv, rep_pipe_recv_cb, p)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_putq, rep_pipe_putq_cb, p)) != 0)) {
+		rep_pipe_fini(p);
 		return (rv);
 	}
-	nni_aio_init(&rp->aio_getq, nni_rep_pipe_getq_cb, rp);
-	nni_aio_init(&rp->aio_send, nni_rep_pipe_send_cb, rp);
-	nni_aio_init(&rp->aio_recv, nni_rep_pipe_recv_cb, rp);
-	nni_aio_init(&rp->aio_putq, nni_rep_pipe_putq_cb, rp);
-
-	rp->pipe = pipe;
-	rp->rep  = rsock;
-	*rpp     = rp;
-	return (0);
-}
 
-static void
-nni_rep_pipe_fini(void *arg)
-{
-	nni_rep_pipe *rp = arg;
-
-	nni_aio_fini(&rp->aio_getq);
-	nni_aio_fini(&rp->aio_send);
-	nni_aio_fini(&rp->aio_recv);
-	nni_aio_fini(&rp->aio_putq);
-	nni_msgq_fini(rp->sendq);
-	NNI_FREE_STRUCT(rp);
+	p->pipe = pipe;
+	p->rep  = s;
+	*pp     = p;
+	return (0);
 }
 
 static int
-nni_rep_pipe_start(void *arg)
+rep_pipe_start(void *arg)
 {
-	nni_rep_pipe *rp  = arg;
-	nni_rep_sock *rep = rp->rep;
-	int           rv;
+	rep_pipe *p = arg;
+	rep_sock *s = p->rep;
+	int       rv;
 
-	rv = nni_idhash_insert(rep->pipes, nni_pipe_id(rp->pipe), rp);
-	if (rv != 0) {
+	if ((rv = nni_idhash_insert(s->pipes, nni_pipe_id(p->pipe), p)) != 0) {
 		return (rv);
 	}
 
-	nni_msgq_aio_get(rp->sendq, &rp->aio_getq);
-	nni_pipe_recv(rp->pipe, &rp->aio_recv);
+	nni_msgq_aio_get(p->sendq, p->aio_getq);
+	nni_pipe_recv(p->pipe, p->aio_recv);
 	return (0);
 }
 
 static void
-nni_rep_pipe_stop(void *arg)
+rep_pipe_stop(void *arg)
 {
-	nni_rep_pipe *rp  = arg;
-	nni_rep_sock *rep = rp->rep;
+	rep_pipe *p = arg;
+	rep_sock *s = p->rep;
 
-	nni_msgq_close(rp->sendq);
-	nni_aio_stop(&rp->aio_getq);
-	nni_aio_stop(&rp->aio_send);
-	nni_aio_stop(&rp->aio_recv);
-	nni_aio_stop(&rp->aio_putq);
+	nni_msgq_close(p->sendq);
+	nni_aio_stop(p->aio_getq);
+	nni_aio_stop(p->aio_send);
+	nni_aio_stop(p->aio_recv);
+	nni_aio_stop(p->aio_putq);
 
-	nni_idhash_remove(rep->pipes, nni_pipe_id(rp->pipe));
+	nni_idhash_remove(s->pipes, nni_pipe_id(p->pipe));
 }
 
 static void
-nni_rep_sock_getq_cb(void *arg)
+rep_sock_getq_cb(void *arg)
 {
-	nni_rep_sock *rep = arg;
-	nni_msgq *    uwq = rep->uwq;
-	nni_msg *     msg;
-	uint32_t      id;
-	nni_rep_pipe *rp;
-	int           rv;
+	rep_sock *s   = arg;
+	nni_msgq *uwq = s->uwq;
+	nni_msg * msg;
+	uint32_t  id;
+	rep_pipe *p;
+	int       rv;
 
 	// This watches for messages from the upper write queue,
 	// extracts the destination pipe, and forwards it to the appropriate
 	// destination pipe via a separate queue.  This prevents a single bad
 	// or slow pipe from gumming up the works for the entire socket.
 
-	if (nni_aio_result(&rep->aio_getq) != 0) {
+	if (nni_aio_result(s->aio_getq) != 0) {
 		// Closed socket?
 		return;
 	}
 
-	msg                 = rep->aio_getq.a_msg;
-	rep->aio_getq.a_msg = NULL;
+	msg = nni_aio_get_msg(s->aio_getq);
+	nni_aio_set_msg(s->aio_getq, NULL);
 
 	// We yank the outgoing pipe id from the header
 	if (nni_msg_header_len(msg) < 4) {
 		nni_msg_free(msg);
 
 		// Look for another message on the upper write queue.
-		nni_msgq_aio_get(uwq, &rep->aio_getq);
+		nni_msgq_aio_get(uwq, s->aio_getq);
 		return;
 	}
 
@@ -218,69 +215,68 @@ nni_rep_sock_getq_cb(void *arg)
 	// Look for the pipe, and attempt to put the message there
 	// (nonblocking) if we can.  If we can't for any reason, then we
 	// free the message.
-	rv = nni_idhash_find(rep->pipes, id, (void **) &rp);
-	if (rv == 0) {
-		rv = nni_msgq_tryput(rp->sendq, msg);
+	if ((rv = nni_idhash_find(s->pipes, id, (void **) &p)) == 0) {
+		rv = nni_msgq_tryput(p->sendq, msg);
 	}
 	if (rv != 0) {
 		nni_msg_free(msg);
 	}
 
 	// Now look for another message on the upper write queue.
-	nni_msgq_aio_get(uwq, &rep->aio_getq);
+	nni_msgq_aio_get(uwq, s->aio_getq);
 }
 
 static void
-nni_rep_pipe_getq_cb(void *arg)
+rep_pipe_getq_cb(void *arg)
 {
-	nni_rep_pipe *rp = arg;
+	rep_pipe *p = arg;
 
-	if (nni_aio_result(&rp->aio_getq) != 0) {
-		nni_pipe_stop(rp->pipe);
+	if (nni_aio_result(p->aio_getq) != 0) {
+		nni_pipe_stop(p->pipe);
 		return;
 	}
 
-	rp->aio_send.a_msg = rp->aio_getq.a_msg;
-	rp->aio_getq.a_msg = 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(rp->pipe, &rp->aio_send);
+	nni_pipe_send(p->pipe, p->aio_send);
 }
 
 static void
-nni_rep_pipe_send_cb(void *arg)
+rep_pipe_send_cb(void *arg)
 {
-	nni_rep_pipe *rp = arg;
+	rep_pipe *p = arg;
 
-	if (nni_aio_result(&rp->aio_send) != 0) {
-		nni_msg_free(rp->aio_send.a_msg);
-		rp->aio_send.a_msg = NULL;
-		nni_pipe_stop(rp->pipe);
+	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_stop(p->pipe);
 		return;
 	}
 
-	nni_msgq_aio_get(rp->sendq, &rp->aio_getq);
+	nni_msgq_aio_get(p->sendq, p->aio_getq);
 }
 
 static void
-nni_rep_pipe_recv_cb(void *arg)
+rep_pipe_recv_cb(void *arg)
 {
-	nni_rep_pipe *rp  = arg;
-	nni_rep_sock *rep = rp->rep;
-	nni_msg *     msg;
-	int           rv;
-	uint8_t *     body;
-	int           hops;
-
-	if (nni_aio_result(&rp->aio_recv) != 0) {
-		nni_pipe_stop(rp->pipe);
+	rep_pipe *p = arg;
+	rep_sock *s = p->rep;
+	nni_msg * msg;
+	int       rv;
+	uint8_t * body;
+	int       hops;
+
+	if (nni_aio_result(p->aio_recv) != 0) {
+		nni_pipe_stop(p->pipe);
 		return;
 	}
 
-	msg                = rp->aio_recv.a_msg;
-	rp->aio_recv.a_msg = NULL;
+	msg = nni_aio_get_msg(p->aio_recv);
+	nni_aio_set_msg(p->aio_recv, NULL);
 
 	// Store the pipe id in the header, first thing.
-	rv = nni_msg_header_append_u32(msg, nni_pipe_id(rp->pipe));
+	rv = nni_msg_header_append_u32(msg, nni_pipe_id(p->pipe));
 	if (rv != 0) {
 		// Failure here causes us to drop the message.
 		goto drop;
@@ -290,7 +286,7 @@ nni_rep_pipe_recv_cb(void *arg)
 	hops = 1;
 	for (;;) {
 		int end = 0;
-		if (hops >= rep->ttl) {
+		if (hops >= 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
@@ -300,7 +296,7 @@ nni_rep_pipe_recv_cb(void *arg)
 		if (nni_msg_len(msg) < 4) {
 			// Peer is speaking garbage. Kick it.
 			nni_msg_free(msg);
-			nni_pipe_stop(rp->pipe);
+			nni_pipe_stop(p->pipe);
 			return;
 		}
 		body = nni_msg_body(msg);
@@ -320,74 +316,74 @@ nni_rep_pipe_recv_cb(void *arg)
 	}
 
 	// Go ahead and send it up.
-	rp->aio_putq.a_msg = msg;
-	nni_msgq_aio_put(rp->rep->urq, &rp->aio_putq);
+	nni_aio_set_msg(p->aio_putq, msg);
+	nni_msgq_aio_put(s->urq, p->aio_putq);
 	return;
 
 drop:
 	nni_msg_free(msg);
-	nni_pipe_recv(rp->pipe, &rp->aio_recv);
+	nni_pipe_recv(p->pipe, p->aio_recv);
 }
 
 static void
-nni_rep_pipe_putq_cb(void *arg)
+rep_pipe_putq_cb(void *arg)
 {
-	nni_rep_pipe *rp = arg;
+	rep_pipe *p = arg;
 
-	if (nni_aio_result(&rp->aio_putq) != 0) {
-		nni_msg_free(rp->aio_putq.a_msg);
-		rp->aio_putq.a_msg = NULL;
-		nni_pipe_stop(rp->pipe);
+	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_stop(p->pipe);
 		return;
 	}
 
-	nni_pipe_recv(rp->pipe, &rp->aio_recv);
+	nni_pipe_recv(p->pipe, p->aio_recv);
 }
 
 static int
-nni_rep_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
+rep_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
 {
-	nni_rep_sock *rep = arg;
-	int           rv  = NNG_ENOTSUP;
+	rep_sock *s  = arg;
+	int       rv = NNG_ENOTSUP;
 
 	if (opt == nng_optid_maxttl) {
-		rv = nni_setopt_int(&rep->ttl, buf, sz, 1, 255);
+		rv = nni_setopt_int(&s->ttl, buf, sz, 1, 255);
 	} else if (opt == nng_optid_raw) {
-		rv = nni_setopt_int(&rep->raw, buf, sz, 0, 1);
-		nni_sock_senderr(rep->sock, rep->raw ? 0 : NNG_ESTATE);
+		rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
+		nni_sock_senderr(s->sock, s->raw ? 0 : NNG_ESTATE);
 	}
 	return (rv);
 }
 
 static int
-nni_rep_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
+rep_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 {
-	nni_rep_sock *rep = arg;
-	int           rv  = NNG_ENOTSUP;
+	rep_sock *s  = arg;
+	int       rv = NNG_ENOTSUP;
 
 	if (opt == nng_optid_maxttl) {
-		rv = nni_getopt_int(&rep->ttl, buf, szp);
+		rv = nni_getopt_int(&s->ttl, buf, szp);
 	} else if (opt == nng_optid_raw) {
-		rv = nni_getopt_int(&rep->raw, buf, szp);
+		rv = nni_getopt_int(&s->raw, buf, szp);
 	}
 	return (rv);
 }
 
 static nni_msg *
-nni_rep_sock_sfilter(void *arg, nni_msg *msg)
+rep_sock_sfilter(void *arg, nni_msg *msg)
 {
-	nni_rep_sock *rep = arg;
+	rep_sock *s = arg;
 
-	if (rep->raw) {
+	if (s->raw) {
 		return (msg);
 	}
 
 	// Cannot send again until a receive is done...
-	nni_sock_senderr(rep->sock, NNG_ESTATE);
+	nni_sock_senderr(s->sock, NNG_ESTATE);
 
 	// If we have a stored backtrace, append it to the header...
 	// if we don't have a backtrace, discard the message.
-	if (rep->btrace == NULL) {
+	if (s->btrace == NULL) {
 		nni_msg_free(msg);
 		return (NULL);
 	}
@@ -395,76 +391,76 @@ nni_rep_sock_sfilter(void *arg, nni_msg *msg)
 	// drop anything else in the header...
 	nni_msg_header_clear(msg);
 
-	if (nni_msg_header_append(msg, rep->btrace, rep->btrace_len) != 0) {
-		nni_free(rep->btrace, rep->btrace_len);
-		rep->btrace     = NULL;
-		rep->btrace_len = 0;
+	if (nni_msg_header_append(msg, s->btrace, s->btrace_len) != 0) {
+		nni_free(s->btrace, s->btrace_len);
+		s->btrace     = NULL;
+		s->btrace_len = 0;
 		nni_msg_free(msg);
 		return (NULL);
 	}
 
-	nni_free(rep->btrace, rep->btrace_len);
-	rep->btrace     = NULL;
-	rep->btrace_len = 0;
+	nni_free(s->btrace, s->btrace_len);
+	s->btrace     = NULL;
+	s->btrace_len = 0;
 	return (msg);
 }
 
 static nni_msg *
-nni_rep_sock_rfilter(void *arg, nni_msg *msg)
+rep_sock_rfilter(void *arg, nni_msg *msg)
 {
-	nni_rep_sock *rep = arg;
-	char *        header;
-	size_t        len;
+	rep_sock *s = arg;
+	char *    header;
+	size_t    len;
 
-	if (rep->raw) {
+	if (s->raw) {
 		return (msg);
 	}
 
-	nni_sock_senderr(rep->sock, 0);
+	nni_sock_senderr(s->sock, 0);
 	len    = nni_msg_header_len(msg);
 	header = nni_msg_header(msg);
-	if (rep->btrace != NULL) {
-		nni_free(rep->btrace, rep->btrace_len);
-		rep->btrace     = NULL;
-		rep->btrace_len = 0;
+	if (s->btrace != NULL) {
+		nni_free(s->btrace, s->btrace_len);
+		s->btrace     = NULL;
+		s->btrace_len = 0;
 	}
-	if ((rep->btrace = nni_alloc(len)) == NULL) {
+	if ((s->btrace = nni_alloc(len)) == NULL) {
 		nni_msg_free(msg);
 		return (NULL);
 	}
-	rep->btrace_len = len;
-	memcpy(rep->btrace, header, len);
+	s->btrace_len = len;
+	memcpy(s->btrace, header, len);
 	nni_msg_header_clear(msg);
 	return (msg);
 }
 
 // This is the global protocol structure -- our linkage to the core.
 // This should be the only global non-static symbol in this file.
-static nni_proto_pipe_ops nni_rep_pipe_ops = {
-	.pipe_init  = nni_rep_pipe_init,
-	.pipe_fini  = nni_rep_pipe_fini,
-	.pipe_start = nni_rep_pipe_start,
-	.pipe_stop  = nni_rep_pipe_stop,
+static nni_proto_pipe_ops rep_pipe_ops = {
+	.pipe_init  = rep_pipe_init,
+	.pipe_fini  = rep_pipe_fini,
+	.pipe_start = rep_pipe_start,
+	.pipe_stop  = rep_pipe_stop,
 };
 
-static nni_proto_sock_ops nni_rep_sock_ops = {
-	.sock_init    = nni_rep_sock_init,
-	.sock_fini    = nni_rep_sock_fini,
-	.sock_open    = nni_rep_sock_open,
-	.sock_close   = nni_rep_sock_close,
-	.sock_setopt  = nni_rep_sock_setopt,
-	.sock_getopt  = nni_rep_sock_getopt,
-	.sock_rfilter = nni_rep_sock_rfilter,
-	.sock_sfilter = nni_rep_sock_sfilter,
+static nni_proto_sock_ops rep_sock_ops = {
+	.sock_init    = rep_sock_init,
+	.sock_fini    = rep_sock_fini,
+	.sock_open    = rep_sock_open,
+	.sock_close   = rep_sock_close,
+	.sock_setopt  = rep_sock_setopt,
+	.sock_getopt  = rep_sock_getopt,
+	.sock_rfilter = rep_sock_rfilter,
+	.sock_sfilter = rep_sock_sfilter,
 };
 
-nni_proto nni_rep_proto = {
+static nni_proto nni_rep_proto = {
 	.proto_version  = NNI_PROTOCOL_VERSION,
 	.proto_self     = { NNG_PROTO_REP_V0, "rep" },
 	.proto_peer     = { NNG_PROTO_REQ_V0, "req" },
 	.proto_flags    = NNI_PROTO_FLAG_SNDRCV,
-	.proto_sock_ops = &nni_rep_sock_ops,
-	.proto_pipe_ops = &nni_rep_pipe_ops,
+	.proto_sock_ops = &rep_sock_ops,
+	.proto_pipe_ops = &rep_pipe_ops,
 };
 
 int
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index 2579417a..5da5003f 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -18,15 +18,15 @@
 // request-reply pair.  This is useful for building RPC clients, for
 // example.
 
-typedef struct nni_req_pipe nni_req_pipe;
-typedef struct nni_req_sock nni_req_sock;
+typedef struct req_pipe req_pipe;
+typedef struct req_sock req_sock;
 
-static void nni_req_resend(nni_req_sock *);
-static void nni_req_timeout(void *);
-static void nni_req_pipe_fini(void *);
+static void req_resend(req_sock *);
+static void req_timeout(void *);
+static void req_pipe_fini(void *);
 
-// An nni_req_sock is our per-socket protocol private structure.
-struct nni_req_sock {
+// A req_sock is our per-socket protocol private structure.
+struct req_sock {
 	nni_sock *   sock;
 	nni_msgq *   uwq;
 	nni_msgq *   urq;
@@ -38,7 +38,7 @@ struct nni_req_sock {
 	int          ttl;
 	nni_msg *    reqmsg;
 
-	nni_req_pipe *pendpipe;
+	req_pipe *pendpipe;
 
 	nni_list readypipes;
 	nni_list busypipes;
@@ -51,230 +51,235 @@ struct nni_req_sock {
 	nni_cv   cv;
 };
 
-// An nni_req_pipe is our per-pipe protocol private structure.
-struct nni_req_pipe {
+// A req_pipe is our per-pipe protocol private structure.
+struct req_pipe {
 	nni_pipe *    pipe;
-	nni_req_sock *req;
+	req_sock *    req;
 	nni_list_node node;
-	nni_aio       aio_getq;       // raw mode only
-	nni_aio       aio_sendraw;    // raw mode only
-	nni_aio       aio_sendcooked; // cooked mode only
-	nni_aio       aio_recv;
-	nni_aio       aio_putq;
+	nni_aio *     aio_getq;       // raw mode only
+	nni_aio *     aio_sendraw;    // raw mode only
+	nni_aio *     aio_sendcooked; // cooked mode only
+	nni_aio *     aio_recv;
+	nni_aio *     aio_putq;
 	nni_mtx       mtx;
 };
 
-static void nni_req_resender(void *);
-static void nni_req_getq_cb(void *);
-static void nni_req_sendraw_cb(void *);
-static void nni_req_sendcooked_cb(void *);
-static void nni_req_recv_cb(void *);
-static void nni_req_putq_cb(void *);
+static void req_resender(void *);
+static void req_getq_cb(void *);
+static void req_sendraw_cb(void *);
+static void req_sendcooked_cb(void *);
+static void req_recv_cb(void *);
+static void req_putq_cb(void *);
 
 static int
-nni_req_sock_init(void **reqp, nni_sock *sock)
+req_sock_init(void **sp, nni_sock *sock)
 {
-	nni_req_sock *req;
+	req_sock *s;
 
-	if ((req = NNI_ALLOC_STRUCT(req)) == NULL) {
+	if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	nni_mtx_init(&req->mtx);
-	nni_cv_init(&req->cv, &req->mtx);
+	nni_mtx_init(&s->mtx);
+	nni_cv_init(&s->cv, &s->mtx);
 
-	NNI_LIST_INIT(&req->readypipes, nni_req_pipe, node);
-	NNI_LIST_INIT(&req->busypipes, nni_req_pipe, node);
-	nni_timer_init(&req->timer, nni_req_timeout, req);
+	NNI_LIST_INIT(&s->readypipes, req_pipe, node);
+	NNI_LIST_INIT(&s->busypipes, req_pipe, node);
+	nni_timer_init(&s->timer, req_timeout, s);
 
 	// this is "semi random" start for request IDs.
-	req->nextid = nni_random();
-
-	req->retry  = NNI_SECOND * 60;
-	req->sock   = sock;
-	req->reqmsg = NULL;
-	req->raw    = 0;
-	req->wantw  = 0;
-	req->resend = NNI_TIME_ZERO;
-	req->ttl    = 8;
-
-	req->uwq = nni_sock_sendq(sock);
-	req->urq = nni_sock_recvq(sock);
-	*reqp    = req;
+	s->nextid = nni_random();
+	s->retry  = NNI_SECOND * 60;
+	s->sock   = sock;
+	s->reqmsg = NULL;
+	s->raw    = 0;
+	s->wantw  = 0;
+	s->resend = NNI_TIME_ZERO;
+	s->ttl    = 8;
+	s->uwq    = nni_sock_sendq(sock);
+	s->urq    = nni_sock_recvq(sock);
+	*sp       = s;
 	nni_sock_recverr(sock, NNG_ESTATE);
 	return (0);
 }
 
 static void
-nni_req_sock_open(void *arg)
+req_sock_open(void *arg)
 {
 	NNI_ARG_UNUSED(arg);
 }
 
 static void
-nni_req_sock_close(void *arg)
+req_sock_close(void *arg)
 {
-	nni_req_sock *req = arg;
+	req_sock *s = arg;
 
-	nni_mtx_lock(&req->mtx);
-	req->closed = 1;
-	nni_mtx_unlock(&req->mtx);
+	nni_mtx_lock(&s->mtx);
+	s->closed = 1;
+	nni_mtx_unlock(&s->mtx);
 
-	nni_timer_cancel(&req->timer);
+	nni_timer_cancel(&s->timer);
 }
 
 static void
-nni_req_sock_fini(void *arg)
+req_sock_fini(void *arg)
 {
-	nni_req_sock *req = arg;
+	req_sock *s = arg;
 
-	nni_mtx_lock(&req->mtx);
-	while ((!nni_list_empty(&req->readypipes)) ||
-	    (!nni_list_empty(&req->busypipes))) {
-		nni_cv_wait(&req->cv);
+	nni_mtx_lock(&s->mtx);
+	while ((!nni_list_empty(&s->readypipes)) ||
+	    (!nni_list_empty(&s->busypipes))) {
+		nni_cv_wait(&s->cv);
 	}
-	if (req->reqmsg != NULL) {
-		nni_msg_free(req->reqmsg);
+	if (s->reqmsg != NULL) {
+		nni_msg_free(s->reqmsg);
 	}
-	nni_mtx_unlock(&req->mtx);
-	nni_cv_fini(&req->cv);
-	nni_mtx_fini(&req->mtx);
-	NNI_FREE_STRUCT(req);
+	nni_mtx_unlock(&s->mtx);
+	nni_cv_fini(&s->cv);
+	nni_mtx_fini(&s->mtx);
+	NNI_FREE_STRUCT(s);
+}
+
+static void
+req_pipe_fini(void *arg)
+{
+	req_pipe *p = arg;
+
+	nni_aio_fini(p->aio_getq);
+	nni_aio_fini(p->aio_putq);
+	nni_aio_fini(p->aio_recv);
+	nni_aio_fini(p->aio_sendcooked);
+	nni_aio_fini(p->aio_sendraw);
+	nni_mtx_fini(&p->mtx);
+	NNI_FREE_STRUCT(p);
 }
 
 static int
-nni_req_pipe_init(void **rpp, nni_pipe *pipe, void *rsock)
+req_pipe_init(void **pp, nni_pipe *pipe, void *s)
 {
-	nni_req_pipe *rp;
+	req_pipe *p;
+	int       rv;
 
-	if ((rp = NNI_ALLOC_STRUCT(rp)) == NULL) {
+	if ((p = NNI_ALLOC_STRUCT(p)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	nni_mtx_init(&rp->mtx);
-	nni_aio_init(&rp->aio_getq, nni_req_getq_cb, rp);
-	nni_aio_init(&rp->aio_putq, nni_req_putq_cb, rp);
-	nni_aio_init(&rp->aio_recv, nni_req_recv_cb, rp);
-	nni_aio_init(&rp->aio_sendraw, nni_req_sendraw_cb, rp);
-	nni_aio_init(&rp->aio_sendcooked, nni_req_sendcooked_cb, rp);
-
-	NNI_LIST_NODE_INIT(&rp->node);
-	rp->pipe = pipe;
-	rp->req  = rsock;
-	*rpp     = rp;
-	return (0);
-}
+	nni_mtx_init(&p->mtx);
+	if (((rv = nni_aio_init(&p->aio_getq, req_getq_cb, p)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_putq, req_putq_cb, p)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_recv, req_recv_cb, p)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_sendraw, req_sendraw_cb, p)) != 0) ||
+	    ((rv = nni_aio_init(&p->aio_sendcooked, req_sendcooked_cb, p)) !=
+	        0)) {
+		req_pipe_fini(p);
+		return (rv);
+	}
 
-static void
-nni_req_pipe_fini(void *arg)
-{
-	nni_req_pipe *rp = arg;
-
-	nni_aio_fini(&rp->aio_getq);
-	nni_aio_fini(&rp->aio_putq);
-	nni_aio_fini(&rp->aio_recv);
-	nni_aio_fini(&rp->aio_sendcooked);
-	nni_aio_fini(&rp->aio_sendraw);
-	nni_mtx_fini(&rp->mtx);
-	NNI_FREE_STRUCT(rp);
+	NNI_LIST_NODE_INIT(&p->node);
+	p->pipe = pipe;
+	p->req  = s;
+	*pp     = p;
+	return (0);
 }
 
 static int
-nni_req_pipe_start(void *arg)
+req_pipe_start(void *arg)
 {
-	nni_req_pipe *rp  = arg;
-	nni_req_sock *req = rp->req;
+	req_pipe *p = arg;
+	req_sock *s = p->req;
 
-	if (nni_pipe_peer(rp->pipe) != NNG_PROTO_REP) {
+	if (nni_pipe_peer(p->pipe) != NNG_PROTO_REP) {
 		return (NNG_EPROTO);
 	}
 
-	nni_mtx_lock(&req->mtx);
-	if (req->closed) {
-		nni_mtx_unlock(&req->mtx);
+	nni_mtx_lock(&s->mtx);
+	if (s->closed) {
+		nni_mtx_unlock(&s->mtx);
 		return (NNG_ECLOSED);
 	}
-	nni_list_append(&req->readypipes, rp);
-	if (req->wantw) {
-		nni_req_resend(req);
+	nni_list_append(&s->readypipes, p);
+	// If sock was waiting for somewhere to send data, go ahead and
+	// send it to this pipe.
+	if (s->wantw) {
+		req_resend(s);
 	}
-	nni_mtx_unlock(&req->mtx);
+	nni_mtx_unlock(&s->mtx);
 
-	nni_msgq_aio_get(req->uwq, &rp->aio_getq);
-	nni_pipe_recv(rp->pipe, &rp->aio_recv);
+	nni_msgq_aio_get(s->uwq, p->aio_getq);
+	nni_pipe_recv(p->pipe, p->aio_recv);
 	return (0);
 }
 
 static void
-nni_req_pipe_stop(void *arg)
+req_pipe_stop(void *arg)
 {
-	nni_req_pipe *rp  = arg;
-	nni_req_sock *req = rp->req;
+	req_pipe *p = arg;
+	req_sock *s = p->req;
 
-	nni_aio_stop(&rp->aio_getq);
-	nni_aio_stop(&rp->aio_putq);
-	nni_aio_stop(&rp->aio_recv);
-	nni_aio_stop(&rp->aio_sendcooked);
-	nni_aio_stop(&rp->aio_sendraw);
+	nni_aio_stop(p->aio_getq);
+	nni_aio_stop(p->aio_putq);
+	nni_aio_stop(p->aio_recv);
+	nni_aio_stop(p->aio_sendcooked);
+	nni_aio_stop(p->aio_sendraw);
 
 	// At this point there should not be any further AIOs running.
 	// Further, any completion tasks have completed.
 
-	nni_mtx_lock(&req->mtx);
+	nni_mtx_lock(&s->mtx);
 	// This removes the node from either busypipes or readypipes.
 	// It doesn't much matter which.
-	if (nni_list_node_active(&rp->node)) {
-		nni_list_node_remove(&rp->node);
-		if (req->closed) {
-			nni_cv_wake(&req->cv);
+	if (nni_list_node_active(&p->node)) {
+		nni_list_node_remove(&p->node);
+		if (s->closed) {
+			nni_cv_wake(&s->cv);
 		}
 	}
 
-	if ((rp == req->pendpipe) && (req->reqmsg != NULL)) {
+	if ((p == s->pendpipe) && (s->reqmsg != NULL)) {
 		// removing the pipe we sent the last request on...
 		// schedule immediate resend.
-		req->pendpipe = NULL;
-		req->resend   = NNI_TIME_ZERO;
-		req->wantw    = 1;
-		nni_req_resend(req);
+		s->pendpipe = NULL;
+		s->resend   = NNI_TIME_ZERO;
+		s->wantw    = 1;
+		req_resend(s);
 	}
-	nni_mtx_unlock(&req->mtx);
+	nni_mtx_unlock(&s->mtx);
 }
 
 static int
-nni_req_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
+req_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
 {
-	nni_req_sock *req = arg;
-	int           rv  = NNG_ENOTSUP;
+	req_sock *s  = arg;
+	int       rv = NNG_ENOTSUP;
 
 	if (opt == nng_optid_req_resendtime) {
-		rv = nni_setopt_usec(&req->retry, buf, sz);
+		rv = nni_setopt_usec(&s->retry, buf, sz);
 
 	} else if (opt == nng_optid_raw) {
-		rv = nni_setopt_int(&req->raw, buf, sz, 0, 1);
+		rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
 		if (rv == 0) {
-			nni_sock_recverr(req->sock, req->raw ? 0 : NNG_ESTATE);
+			nni_sock_recverr(s->sock, s->raw ? 0 : NNG_ESTATE);
 		}
 
 	} else if (opt == nng_optid_maxttl) {
-		rv = nni_setopt_int(&req->ttl, buf, sz, 1, 255);
+		rv = nni_setopt_int(&s->ttl, buf, sz, 1, 255);
 	}
 
 	return (rv);
 }
 
 static int
-nni_req_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
+req_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 {
-	nni_req_sock *req = arg;
-	int           rv  = NNG_ENOTSUP;
+	req_sock *s  = arg;
+	int       rv = NNG_ENOTSUP;
 
 	if (opt == nng_optid_req_resendtime) {
-		rv = nni_getopt_usec(&req->retry, buf, szp);
+		rv = nni_getopt_usec(&s->retry, buf, szp);
 
 	} else if (opt == nng_optid_raw) {
-		rv = nni_getopt_int(&req->raw, buf, szp);
+		rv = nni_getopt_int(&s->raw, buf, szp);
 
 	} else if (opt == nng_optid_maxttl) {
-		rv = nni_getopt_int(&req->ttl, buf, szp);
+		rv = nni_getopt_int(&s->ttl, buf, szp);
 	}
 
 	return (rv);
@@ -297,10 +302,10 @@ nni_req_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 // kind of priority.)
 
 static void
-nni_req_getq_cb(void *arg)
+req_getq_cb(void *arg)
 {
-	nni_req_pipe *rp  = arg;
-	nni_req_sock *req = rp->req;
+	req_pipe *p = arg;
+	req_sock *s = p->req;
 
 	// We should be in RAW mode.  Cooked mode traffic bypasses
 	// the upper write queue entirely, and should never end up here.
@@ -308,47 +313,47 @@ nni_req_getq_cb(void *arg)
 	// that's ok (there's an inherent race anyway).  (One minor
 	// exception: we wind up here in error state when the uwq is closed.)
 
-	if (nni_aio_result(&rp->aio_getq) != 0) {
-		nni_pipe_stop(rp->pipe);
+	if (nni_aio_result(p->aio_getq) != 0) {
+		nni_pipe_stop(p->pipe);
 		return;
 	}
 
-	rp->aio_sendraw.a_msg = rp->aio_getq.a_msg;
-	rp->aio_getq.a_msg    = NULL;
+	nni_aio_set_msg(p->aio_sendraw, nni_aio_get_msg(p->aio_getq));
+	nni_aio_set_msg(p->aio_getq, NULL);
 
 	// Send the message, but use the raw mode aio.
-	nni_pipe_send(rp->pipe, &rp->aio_sendraw);
+	nni_pipe_send(p->pipe, p->aio_sendraw);
 }
 
 static void
-nni_req_sendraw_cb(void *arg)
+req_sendraw_cb(void *arg)
 {
-	nni_req_pipe *rp = arg;
+	req_pipe *p = arg;
 
-	if (nni_aio_result(&rp->aio_sendraw) != 0) {
-		nni_msg_free(rp->aio_sendraw.a_msg);
-		rp->aio_sendraw.a_msg = NULL;
-		nni_pipe_stop(rp->pipe);
+	if (nni_aio_result(p->aio_sendraw) != 0) {
+		nni_msg_free(nni_aio_get_msg(p->aio_sendraw));
+		nni_aio_set_msg(p->aio_sendraw, NULL);
+		nni_pipe_stop(p->pipe);
 		return;
 	}
 
 	// Sent a message so we just need to look for another one.
-	nni_msgq_aio_get(rp->req->uwq, &rp->aio_getq);
+	nni_msgq_aio_get(p->req->uwq, p->aio_getq);
 }
 
 static void
-nni_req_sendcooked_cb(void *arg)
+req_sendcooked_cb(void *arg)
 {
-	nni_req_pipe *rp  = arg;
-	nni_req_sock *req = rp->req;
+	req_pipe *p = arg;
+	req_sock *s = p->req;
 
-	if (nni_aio_result(&rp->aio_sendcooked) != 0) {
+	if (nni_aio_result(p->aio_sendcooked) != 0) {
 		// We failed to send... clean up and deal with it.
 		// We leave ourselves on the busy list for now, which
 		// means no new asynchronous traffic can occur here.
-		nni_msg_free(rp->aio_sendcooked.a_msg);
-		rp->aio_sendcooked.a_msg = NULL;
-		nni_pipe_stop(rp->pipe);
+		nni_msg_free(nni_aio_get_msg(p->aio_sendcooked));
+		nni_aio_set_msg(p->aio_sendcooked, NULL);
+		nni_pipe_stop(p->pipe);
 		return;
 	}
 
@@ -356,49 +361,50 @@ nni_req_sendcooked_cb(void *arg)
 	// reinsert ourselves in the ready list, and possibly schedule
 	// a resend.
 
-	nni_mtx_lock(&req->mtx);
-	if (nni_list_active(&req->busypipes, rp)) {
-		nni_list_remove(&req->busypipes, rp);
-		nni_list_append(&req->readypipes, rp);
-		nni_req_resend(req);
+	nni_mtx_lock(&s->mtx);
+	if (nni_list_active(&s->busypipes, p)) {
+		nni_list_remove(&s->busypipes, p);
+		nni_list_append(&s->readypipes, p);
+		req_resend(s);
 	} else {
 		// We wind up here if stop was called from the reader
 		// side while we were waiting to be scheduled to run for the
 		// writer side.  In this case we can't complete the operation,
 		// and we have to abort.
-		nni_pipe_stop(rp->pipe);
+		nni_pipe_stop(p->pipe);
 	}
-	nni_mtx_unlock(&req->mtx);
+	nni_mtx_unlock(&s->mtx);
 }
 
 static void
-nni_req_putq_cb(void *arg)
+req_putq_cb(void *arg)
 {
-	nni_req_pipe *rp = arg;
+	req_pipe *p = arg;
 
-	if (nni_aio_result(&rp->aio_putq) != 0) {
-		nni_msg_free(rp->aio_putq.a_msg);
-		nni_pipe_stop(rp->pipe);
+	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_stop(p->pipe);
 		return;
 	}
-	rp->aio_putq.a_msg = NULL;
+	nni_aio_set_msg(p->aio_putq, NULL);
 
-	nni_pipe_recv(rp->pipe, &rp->aio_recv);
+	nni_pipe_recv(p->pipe, p->aio_recv);
 }
 
 static void
-nni_req_recv_cb(void *arg)
+req_recv_cb(void *arg)
 {
-	nni_req_pipe *rp = arg;
-	nni_msg *     msg;
+	req_pipe *p = arg;
+	nni_msg * msg;
 
-	if (nni_aio_result(&rp->aio_recv) != 0) {
-		nni_pipe_stop(rp->pipe);
+	if (nni_aio_result(p->aio_recv) != 0) {
+		nni_pipe_stop(p->pipe);
 		return;
 	}
 
-	msg                = rp->aio_recv.a_msg;
-	rp->aio_recv.a_msg = NULL;
+	msg = nni_aio_get_msg(p->aio_recv);
+	nni_aio_set_msg(p->aio_recv, NULL);
 
 	// We yank 4 bytes of body, and move them to the header.
 	if (nni_msg_len(msg) < 4) {
@@ -414,92 +420,90 @@ nni_req_recv_cb(void *arg)
 	}
 	(void) nni_msg_trim(msg, 4); // Cannot fail
 
-	rp->aio_putq.a_msg = msg;
-	nni_msgq_aio_put(rp->req->urq, &rp->aio_putq);
+	nni_aio_set_msg(p->aio_putq, msg);
+	nni_msgq_aio_put(p->req->urq, p->aio_putq);
 	return;
 
 malformed:
 	nni_msg_free(msg);
-	nni_pipe_stop(rp->pipe);
+	nni_pipe_stop(p->pipe);
 }
 
 static void
-nni_req_timeout(void *arg)
+req_timeout(void *arg)
 {
-	nni_req_sock *req = arg;
+	req_sock *s = arg;
 
-	nni_mtx_lock(&req->mtx);
-	if (req->reqmsg != NULL) {
-		req->wantw = 1;
-		nni_req_resend(req);
+	nni_mtx_lock(&s->mtx);
+	if (s->reqmsg != NULL) {
+		s->wantw = 1;
+		req_resend(s);
 	}
-	nni_mtx_unlock(&req->mtx);
+	nni_mtx_unlock(&s->mtx);
 }
 
 static void
-nni_req_resend(nni_req_sock *req)
+req_resend(req_sock *s)
 {
-	nni_req_pipe *rp;
-	nni_msg *     msg;
+	req_pipe *p;
+	nni_msg * msg;
 
 	// Note: This routine should be called with the socket lock held.
 	// Also, this should only be called while handling cooked mode
 	// requests.
-	if ((msg = req->reqmsg) == NULL) {
+	if ((msg = s->reqmsg) == NULL) {
 		return;
 	}
 
-	if (req->closed) {
-		req->reqmsg = NULL;
+	if (s->closed) {
+		s->reqmsg = NULL;
 		nni_msg_free(msg);
 	}
 
-	if (req->wantw) {
-		req->wantw = 0;
+	if (s->wantw) {
+		s->wantw = 0;
 
-		if (nni_msg_dup(&msg, req->reqmsg) != 0) {
+		if (nni_msg_dup(&msg, s->reqmsg) != 0) {
 			// Failed to alloc message, reschedule it. Also,
 			// mark that we have a message we want to resend,
 			// in case something comes available.
-			req->wantw = 1;
-			nni_timer_schedule(
-			    &req->timer, nni_clock() + req->retry);
+			s->wantw = 1;
+			nni_timer_schedule(&s->timer, nni_clock() + s->retry);
 			return;
 		}
 
 		// Now we iterate across all possible outpipes, until
 		// one accepts it.
-		rp = nni_list_first(&req->readypipes);
-		if (rp == NULL) {
+		if ((p = nni_list_first(&s->readypipes)) == NULL) {
 			// No pipes ready to process us.  Note that we have
 			// something to send, and schedule it.
 			nni_msg_free(msg);
-			req->wantw = 1;
+			s->wantw = 1;
 			return;
 		}
 
-		nni_list_remove(&req->readypipes, rp);
-		nni_list_append(&req->busypipes, rp);
+		nni_list_remove(&s->readypipes, p);
+		nni_list_append(&s->busypipes, p);
 
-		req->pendpipe            = rp;
-		req->resend              = nni_clock() + req->retry;
-		rp->aio_sendcooked.a_msg = msg;
+		s->pendpipe = p;
+		s->resend   = nni_clock() + s->retry;
+		nni_aio_set_msg(p->aio_sendcooked, msg);
 
 		// Note that because we were ready rather than busy, we
 		// should not have any I/O oustanding and hence the aio
 		// object will be available for our use.
-		nni_pipe_send(rp->pipe, &rp->aio_sendcooked);
-		nni_timer_schedule(&req->timer, req->resend);
+		nni_pipe_send(p->pipe, p->aio_sendcooked);
+		nni_timer_schedule(&s->timer, s->resend);
 	}
 }
 
 static nni_msg *
-nni_req_sock_sfilter(void *arg, nni_msg *msg)
+req_sock_sfilter(void *arg, nni_msg *msg)
 {
-	nni_req_sock *req = arg;
-	uint32_t      id;
+	req_sock *s = arg;
+	uint32_t  id;
 
-	if (req->raw) {
+	if (s->raw) {
 		// No automatic retry, and the request ID must
 		// be in the header coming down.
 		return (msg);
@@ -508,12 +512,12 @@ nni_req_sock_sfilter(void *arg, nni_msg *msg)
 	// Generate a new request ID.  We always set the high
 	// order bit so that the peer can locate the end of the
 	// backtrace.  (Pipe IDs have the high order bit clear.)
-	id = (req->nextid++) | 0x80000000u;
+	id = (s->nextid++) | 0x80000000u;
 
 	// Request ID is in big endian format.
-	NNI_PUT32(req->reqid, id);
+	NNI_PUT32(s->reqid, id);
 
-	if (nni_msg_header_append(msg, req->reqid, 4) != 0) {
+	if (nni_msg_header_append(msg, s->reqid, 4) != 0) {
 		// Should be ENOMEM.
 		nni_msg_free(msg);
 		return (NULL);
@@ -521,36 +525,36 @@ nni_req_sock_sfilter(void *arg, nni_msg *msg)
 
 	// NB: The socket lock is also held, so this is always self-serialized.
 	// But we have to serialize against other async callbacks.
-	nni_mtx_lock(&req->mtx);
+	nni_mtx_lock(&s->mtx);
 
 	// If another message is there, this cancels it.
-	if (req->reqmsg != NULL) {
-		nni_msg_free(req->reqmsg);
-		req->reqmsg = NULL;
+	if (s->reqmsg != NULL) {
+		nni_msg_free(s->reqmsg);
+		s->reqmsg = NULL;
 	}
 
 	// Make a duplicate message... for retries.
-	req->reqmsg = msg;
+	s->reqmsg = msg;
 	// Schedule for immediate send
-	req->resend = NNI_TIME_ZERO;
-	req->wantw  = 1;
+	s->resend = NNI_TIME_ZERO;
+	s->wantw  = 1;
 
-	nni_req_resend(req);
-	nni_mtx_unlock(&req->mtx);
+	req_resend(s);
+	nni_mtx_unlock(&s->mtx);
 
 	// Clear the error condition.
-	nni_sock_recverr(req->sock, 0);
+	nni_sock_recverr(s->sock, 0);
 
 	return (NULL);
 }
 
 static nni_msg *
-nni_req_sock_rfilter(void *arg, nni_msg *msg)
+req_sock_rfilter(void *arg, nni_msg *msg)
 {
-	nni_req_sock *req = arg;
-	nni_msg *     rmsg;
+	req_sock *s = arg;
+	nni_msg * rmsg;
 
-	if (req->raw) {
+	if (s->raw) {
 		// Pass it unmolested
 		return (msg);
 	}
@@ -560,62 +564,60 @@ nni_req_sock_rfilter(void *arg, nni_msg *msg)
 		return (NULL);
 	}
 
-	nni_mtx_lock(&req->mtx);
+	nni_mtx_lock(&s->mtx);
 
-	if ((rmsg = req->reqmsg) == NULL) {
+	if ((rmsg = s->reqmsg) == NULL) {
 		// We had no outstanding request.
-		nni_mtx_unlock(&req->mtx);
+		nni_mtx_unlock(&s->mtx);
 		nni_msg_free(msg);
 		return (NULL);
 	}
-	if (memcmp(nni_msg_header(msg), req->reqid, 4) != 0) {
+	if (memcmp(nni_msg_header(msg), s->reqid, 4) != 0) {
 		// Wrong request id
-		nni_mtx_unlock(&req->mtx);
+		nni_mtx_unlock(&s->mtx);
 		nni_msg_free(msg);
 		return (NULL);
 	}
 
-	req->reqmsg   = NULL;
-	req->pendpipe = NULL;
-	nni_mtx_unlock(&req->mtx);
+	s->reqmsg   = NULL;
+	s->pendpipe = NULL;
+	nni_mtx_unlock(&s->mtx);
 
-	nni_sock_recverr(req->sock, NNG_ESTATE);
+	nni_sock_recverr(s->sock, NNG_ESTATE);
 	nni_msg_free(rmsg);
 
 	return (msg);
 }
 
-// This is the global protocol structure -- our linkage to the core.
-// This should be the only global non-static symbol in this file.
-static nni_proto_pipe_ops nni_req_pipe_ops = {
-	.pipe_init  = nni_req_pipe_init,
-	.pipe_fini  = nni_req_pipe_fini,
-	.pipe_start = nni_req_pipe_start,
-	.pipe_stop  = nni_req_pipe_stop,
+static nni_proto_pipe_ops req_pipe_ops = {
+	.pipe_init  = req_pipe_init,
+	.pipe_fini  = req_pipe_fini,
+	.pipe_start = req_pipe_start,
+	.pipe_stop  = req_pipe_stop,
 };
 
-static nni_proto_sock_ops nni_req_sock_ops = {
-	.sock_init    = nni_req_sock_init,
-	.sock_fini    = nni_req_sock_fini,
-	.sock_open    = nni_req_sock_open,
-	.sock_close   = nni_req_sock_close,
-	.sock_setopt  = nni_req_sock_setopt,
-	.sock_getopt  = nni_req_sock_getopt,
-	.sock_rfilter = nni_req_sock_rfilter,
-	.sock_sfilter = nni_req_sock_sfilter,
+static nni_proto_sock_ops req_sock_ops = {
+	.sock_init    = req_sock_init,
+	.sock_fini    = req_sock_fini,
+	.sock_open    = req_sock_open,
+	.sock_close   = req_sock_close,
+	.sock_setopt  = req_sock_setopt,
+	.sock_getopt  = req_sock_getopt,
+	.sock_rfilter = req_sock_rfilter,
+	.sock_sfilter = req_sock_sfilter,
 };
 
-nni_proto nni_req_proto = {
+static nni_proto req_proto = {
 	.proto_version  = NNI_PROTOCOL_VERSION,
 	.proto_self     = { NNG_PROTO_REQ_V0, "req" },
 	.proto_peer     = { NNG_PROTO_REP_V0, "rep" },
 	.proto_flags    = NNI_PROTO_FLAG_SNDRCV,
-	.proto_sock_ops = &nni_req_sock_ops,
-	.proto_pipe_ops = &nni_req_pipe_ops,
+	.proto_sock_ops = &req_sock_ops,
+	.proto_pipe_ops = &req_pipe_ops,
 };
 
 int
 nng_req0_open(nng_socket *sidp)
 {
-	return (nni_proto_open(sidp, &nni_req_proto));
+	return (nni_proto_open(sidp, &req_proto));
 }