fixes #972 Very slow pull/push performance compared to ZMQ

This refactors the pipeline protocol to use lightweight mq
instead of the more expensive message queue structure.  It
also provides nicer backpressure and buffering support.
The test suite was updated and converted to NUTS as well.

This won't completely close the gap, but it should help quite
a bit.

10 files changed, 1203 insertions, 318 deletions

diff --git a/docs/man/nng_push.7.adoc b/docs/man/nng_push.7.adoc
index 18c48579..b630c2bb 100644
--- a/docs/man/nng_push.7.adoc
+++ b/docs/man/nng_push.7.adoc
@@ -1,6 +1,6 @@
 = nng_push(7)
 //
-// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
 // Copyright 2018 Capitar IT Group BV <info@capitar.com>
 //
 // This document is supplied under the terms of the MIT License, a
@@ -56,7 +56,16 @@ Only version 0 of this protocol is supported.
 
 === Protocol Options
 
-The _push_ protocol has no protocol-specific options.
+xref:nng_options.5.adoc#NNG_OPT_SENDBUF[`NNG_OPT_SENDBUF`]::
+
+    (`int`, 0 - 8192)
+    Normally this is set to zero, indicating that send operations are unbuffered.
+    In unbuffered operation, send operations will wait until a suitable peer is available to receive the message.
+    If this is set to a positive value (up to 8192), then an intermediate buffer is
+    provided for the socket with the specified depth (in messages).
+
+NOTE: Transport layer buffering may occur in addition to any socket
+    buffer determined by this option.
 
 === Protocol Headers
 
diff --git a/perf/perf.c b/perf/perf.c
index f1995aba..accac621 100644
--- a/perf/perf.c
+++ b/perf/perf.c
@@ -419,6 +419,10 @@ do_inproc_thr(int argc, char **argv)
 			open_client = nng_pair1_open;
 			open_server = nng_pair1_open;
 			break;
+		case OPT_PIPELINE0:
+			open_client = nng_pull0_open;
+			open_server = nng_push0_open;
+			break;
 		case OPT_URL:
 			addr = arg;
 			break;
diff --git a/src/core/lmq.c b/src/core/lmq.c
index bbf42356..47058fc9 100644
--- a/src/core/lmq.c
+++ b/src/core/lmq.c
@@ -22,7 +22,7 @@ nni_lmq_init(nni_lmq *lmq, size_t cap)
 	// operations as a power of two, for efficiency.  It does possibly
 	// waste some space, but never more than 2x.  Consumers should try
 	// for powers of two if they are concerned about efficiency.
-	alloc = 1;
+	alloc = 2;
 	while (alloc < cap) {
 		alloc *= 2;
 	}
@@ -127,7 +127,7 @@ nni_lmq_resize(nni_lmq *lmq, size_t cap)
 	size_t    alloc;
 	size_t    len;
 
-	alloc = 1;
+	alloc = 2;
 	while (alloc < cap) {
 		alloc *= 2;
 	}
diff --git a/src/protocol/pipeline0/CMakeLists.txt b/src/protocol/pipeline0/CMakeLists.txt
index 4f591450..8a10eab7 100644
--- a/src/protocol/pipeline0/CMakeLists.txt
+++ b/src/protocol/pipeline0/CMakeLists.txt
@@ -17,4 +17,7 @@ nng_defines_if(NNG_PROTO_PUSH0 NNG_HAVE_PUSH0)
 
 nng_sources_if(NNG_PROTO_PULL0 pull.c)
 nng_headers_if(NNG_PROTO_PULL0 nng/protocol/pipeline0/pull.h)
-nng_defines_if(NNG_PROTO_PULL0 NNG_HAVE_PULL0)
\ No newline at end of file
+nng_defines_if(NNG_PROTO_PULL0 NNG_HAVE_PULL0)
+
+nng_test(pull_test)
+nng_test(push_test)
diff --git a/src/protocol/pipeline0/pull.c b/src/protocol/pipeline0/pull.c
index 8feb08b8..94403431 100644
--- a/src/protocol/pipeline0/pull.c
+++ b/src/protocol/pipeline0/pull.c
@@ -26,37 +26,46 @@
 typedef struct pull0_pipe pull0_pipe;
 typedef struct pull0_sock pull0_sock;
 
-static void pull0_putq_cb(void *);
 static void pull0_recv_cb(void *);
-static void pull0_putq(pull0_pipe *, nni_msg *);
 
 // pull0_sock is our per-socket protocol private structure.
 struct pull0_sock {
-	nni_msgq *urq;
-	bool      raw;
+	bool         raw;
+	nni_list     pl; // pipe list (pipes with data ready)
+	nni_list     rq; // recv queue (aio list)
+	nni_mtx      m;
+	nni_pollable readable;
 };
 
 // pull0_pipe is our per-pipe protocol private structure.
 struct pull0_pipe {
-	nni_pipe *  pipe;
-	pull0_sock *pull;
-	nni_aio *   putq_aio;
-	nni_aio *   recv_aio;
+	nni_pipe *    p;
+	pull0_sock *  s;
+	nni_msg *     m;
+	nni_aio       aio;
+	bool          closed;
+	nni_list_node node;
 };
 
 static int
 pull0_sock_init(void *arg, nni_sock *sock)
 {
 	pull0_sock *s = arg;
+	NNI_ARG_UNUSED(sock);
 
-	s->urq = nni_sock_recvq(sock);
+	nni_aio_list_init(&s->rq);
+	NNI_LIST_INIT(&s->pl, pull0_pipe, node);
+	nni_mtx_init(&s->m);
+	nni_pollable_init(&s->readable);
 	return (0);
 }
 
 static void
 pull0_sock_fini(void *arg)
 {
-	NNI_ARG_UNUSED(arg);
+	pull0_sock *s = arg;
+	nni_mtx_fini(&s->m);
+	nni_pollable_fini(&s->readable);
 }
 
 static void
@@ -64,8 +73,7 @@ pull0_pipe_stop(void *arg)
 {
 	pull0_pipe *p = arg;
 
-	nni_aio_stop(p->putq_aio);
-	nni_aio_stop(p->recv_aio);
+	nni_aio_stop(&p->aio);
 }
 
 static void
@@ -73,24 +81,20 @@ pull0_pipe_fini(void *arg)
 {
 	pull0_pipe *p = arg;
 
-	nni_aio_free(p->putq_aio);
-	nni_aio_free(p->recv_aio);
+	nni_aio_fini(&p->aio);
+	if (p->m) {
+		nni_msg_free(p->m);
+	}
 }
 
 static int
 pull0_pipe_init(void *arg, nni_pipe *pipe, void *s)
 {
 	pull0_pipe *p = arg;
-	int         rv;
 
-	if (((rv = nni_aio_alloc(&p->putq_aio, pull0_putq_cb, p)) != 0) ||
-	    ((rv = nni_aio_alloc(&p->recv_aio, pull0_recv_cb, p)) != 0)) {
-		pull0_pipe_fini(p);
-		return (rv);
-	}
-
-	p->pipe = pipe;
-	p->pull = s;
+	nni_aio_init(&p->aio, pull0_recv_cb, p);
+	p->p = pipe;
+	p->s = s;
 	return (0);
 }
 
@@ -99,13 +103,13 @@ pull0_pipe_start(void *arg)
 {
 	pull0_pipe *p = arg;
 
-	if (nni_pipe_peer(p->pipe) != NNI_PROTO_PUSH_V0) {
+	if (nni_pipe_peer(p->p) != NNI_PROTO_PUSH_V0) {
 		// Peer protocol mismatch.
 		return (NNG_EPROTO);
 	}
 
-	// Start the pending pull...
-	nni_pipe_recv(p->pipe, p->recv_aio);
+	// Start the pending receive...
+	nni_pipe_recv(p->p, &p->aio);
 
 	return (0);
 }
@@ -114,58 +118,62 @@ static void
 pull0_pipe_close(void *arg)
 {
 	pull0_pipe *p = arg;
+	pull0_sock *s = p->s;
+
+	nni_mtx_lock(&s->m);
+	p->closed = true;
+	if (nni_list_node_active(&p->node)) {
+		nni_list_node_remove(&p->node);
+		if (nni_list_empty(&s->pl)) {
+			nni_pollable_clear(&s->readable);
+		}
+	}
+	nni_mtx_unlock(&s->m);
 
-	nni_aio_close(p->putq_aio);
-	nni_aio_close(p->recv_aio);
+	nni_aio_close(&p->aio);
 }
 
 static void
 pull0_recv_cb(void *arg)
 {
-	pull0_pipe *p   = arg;
-	nni_aio *   aio = p->recv_aio;
-	nni_msg *   msg;
+	pull0_pipe *p  = arg;
+	pull0_sock *s  = p->s;
+	nni_aio *   ap = &p->aio;
+	nni_aio *   as;
+	nni_msg *   m;
 
-	if (nni_aio_result(aio) != 0) {
+	if (nni_aio_result(ap) != 0) {
 		// Failed to get a message, probably the pipe is closed.
-		nni_pipe_close(p->pipe);
+		nni_pipe_close(p->p);
 		return;
 	}
 
 	// Got a message... start the put to send it up to the application.
-	msg = nni_aio_get_msg(aio);
-	nni_msg_set_pipe(msg, nni_pipe_id(p->pipe));
-	nni_aio_set_msg(aio, NULL);
-	pull0_putq(p, msg);
-}
-
-static void
-pull0_putq_cb(void *arg)
-{
-	pull0_pipe *p   = arg;
-	nni_aio *   aio = p->putq_aio;
-
-	if (nni_aio_result(aio) != 0) {
-		// If we failed to put, probably NNG_ECLOSED, nothing else
-		// we can do.  Just close the pipe.
-		nni_msg_free(nni_aio_get_msg(aio));
-		nni_aio_set_msg(aio, NULL);
-		nni_pipe_close(p->pipe);
+	m = nni_aio_get_msg(ap);
+	nni_aio_set_msg(ap, NULL);
+	nni_msg_set_pipe(m, nni_pipe_id(p->p));
+
+	nni_mtx_lock(&s->m);
+	if (p->closed) {
+		nni_mtx_unlock(&s->m);
+		nni_msg_free(m);
 		return;
 	}
-
-	nni_pipe_recv(p->pipe, p->recv_aio);
-}
-
-// pull0_putq schedules a put operation to the user socket (sendup).
-static void
-pull0_putq(pull0_pipe *p, nni_msg *msg)
-{
-	pull0_sock *s = p->pull;
-
-	nni_aio_set_msg(p->putq_aio, msg);
-
-	nni_msgq_aio_put(s->urq, p->putq_aio);
+	if (nni_list_empty(&s->rq)) {
+		nni_list_append(&s->pl, p);
+		if (nni_list_first(&s->pl) == p) {
+			nni_pollable_raise(&s->readable);
+		}
+		p->m = m;
+		nni_mtx_unlock(&s->m);
+		return;
+	}
+	nni_pipe_recv(p->p, ap);
+	as = nni_list_first(&s->rq);
+	nni_aio_list_remove(as);
+	nni_mtx_unlock(&s->m);
+	nni_aio_set_msg(as, m);
+	nni_aio_finish_sync(as, 0, nni_msg_len(m));
 }
 
 static void
@@ -177,7 +185,15 @@ pull0_sock_open(void *arg)
 static void
 pull0_sock_close(void *arg)
 {
-	NNI_ARG_UNUSED(arg);
+	pull0_sock *s = arg;
+	nni_aio *a;
+	nni_mtx_lock(&s->m);
+	while ((a = nni_list_first(&s->rq)) != NULL) {
+		nni_aio_list_remove(a);
+		nni_aio_finish_error(a, NNG_ECLOSED);
+	}
+	// NB: The common socket framework closes pipes before this.
+	nni_mtx_unlock(&s->m);
 }
 
 static void
@@ -188,13 +204,76 @@ pull0_sock_send(void *arg, nni_aio *aio)
 }
 
 static void
+pull0_cancel(nni_aio *aio, void *arg, int rv)
+{
+	pull0_sock *s = arg;
+	nni_mtx_lock(&s->m);
+	if (nni_aio_list_active(aio)) {
+		nni_aio_list_remove(aio);
+		nni_aio_finish_error(aio, rv);
+	}
+	nni_mtx_unlock(&s->m);
+}
+
+static void
 pull0_sock_recv(void *arg, nni_aio *aio)
 {
 	pull0_sock *s = arg;
+	pull0_pipe *p;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+
+	nni_mtx_lock(&s->m);
+	if ((p = nni_list_first(&s->pl)) == NULL) {
+
+		int rv;
+		if ((rv = nni_aio_schedule(aio, pull0_cancel, s)) != 0) {
+			nni_mtx_unlock(&s->m);
+			nni_aio_finish_error(aio, rv);
+			return;
+		}
+
+		nni_aio_list_append(&s->rq, aio);
+		nni_mtx_unlock(&s->m);
+		return;
+	}
+
+	nni_list_remove(&s->pl, p);
+	if (nni_list_empty(&s->pl)) {
+		nni_pollable_clear(&s->readable);
+	}
+	nni_aio_finish_msg(aio, p->m);
+	p->m = NULL;
+	nni_pipe_recv(p->p, &p->aio);
+	nni_mtx_unlock(&s->m);
+}
+
+static int
+pull0_sock_get_recv_fd(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	pull0_sock *s = arg;
+	int         rv;
+	int         fd;
 
-	nni_msgq_aio_get(s->urq, aio);
+	if ((rv = nni_pollable_getfd(&s->readable, &fd)) != 0) {
+		return (rv);
+	}
+	return (nni_copyout_int(fd, buf, szp, t));
 }
 
+static nni_option pull0_sock_options[] = {
+	{
+	    .o_name = NNG_OPT_RECVFD,
+	    .o_get  = pull0_sock_get_recv_fd,
+	},
+	// terminate list
+	{
+	    .o_name = NULL,
+	},
+};
+
 static nni_proto_pipe_ops pull0_pipe_ops = {
 	.pipe_size  = sizeof(pull0_pipe),
 	.pipe_init  = pull0_pipe_init,
@@ -204,13 +283,6 @@ static nni_proto_pipe_ops pull0_pipe_ops = {
 	.pipe_stop  = pull0_pipe_stop,
 };
 
-static nni_option pull0_sock_options[] = {
-	// terminate list
-	{
-	    .o_name = NULL,
-	},
-};
-
 static nni_proto_sock_ops pull0_sock_ops = {
 	.sock_size    = sizeof(pull0_sock),
 	.sock_init    = pull0_sock_init,
@@ -226,28 +298,29 @@ static nni_proto pull0_proto = {
 	.proto_version  = NNI_PROTOCOL_VERSION,
 	.proto_self     = { NNI_PROTO_PULL_V0, "pull" },
 	.proto_peer     = { NNI_PROTO_PUSH_V0, "push" },
-	.proto_flags    = NNI_PROTO_FLAG_RCV,
+	.proto_flags    = NNI_PROTO_FLAG_RCV | NNI_PROTO_FLAG_NOMSGQ,
 	.proto_pipe_ops = &pull0_pipe_ops,
 	.proto_sock_ops = &pull0_sock_ops,
 };
 
 static nni_proto pull0_proto_raw = {
-	.proto_version  = NNI_PROTOCOL_VERSION,
-	.proto_self     = { NNI_PROTO_PULL_V0, "pull" },
-	.proto_peer     = { NNI_PROTO_PUSH_V0, "push" },
-	.proto_flags    = NNI_PROTO_FLAG_RCV | NNI_PROTO_FLAG_RAW,
+	.proto_version = NNI_PROTOCOL_VERSION,
+	.proto_self    = { NNI_PROTO_PULL_V0, "pull" },
+	.proto_peer    = { NNI_PROTO_PUSH_V0, "push" },
+	.proto_flags =
+	    NNI_PROTO_FLAG_RCV | NNI_PROTO_FLAG_RAW | NNI_PROTO_FLAG_NOMSGQ,
 	.proto_pipe_ops = &pull0_pipe_ops,
 	.proto_sock_ops = &pull0_sock_ops,
 };
 
 int
-nng_pull0_open(nng_socket *sidp)
+nng_pull0_open(nng_socket *s)
 {
-	return (nni_proto_open(sidp, &pull0_proto));
+	return (nni_proto_open(s, &pull0_proto));
 }
 
 int
-nng_pull0_open_raw(nng_socket *sidp)
+nng_pull0_open_raw(nng_socket *s)
 {
-	return (nni_proto_open(sidp, &pull0_proto_raw));
+	return (nni_proto_open(s, &pull0_proto_raw));
 }
diff --git a/src/protocol/pipeline0/pull_test.c b/src/protocol/pipeline0/pull_test.c
new file mode 100644
index 00000000..25066093
--- /dev/null
+++ b/src/protocol/pipeline0/pull_test.c
@@ -0,0 +1,264 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include <nuts.h>
+
+static void
+test_pull_identity(void)
+{
+	nng_socket s;
+	int        p;
+	char *     n;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PROTO, &p));
+	NUTS_TRUE(p == NUTS_PROTO(5u, 1u)); // 81
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PEER, &p));
+	NUTS_TRUE(p == NUTS_PROTO(5u, 0u)); // 80
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PROTONAME, &n));
+	NUTS_MATCH(n, "pull");
+	nng_strfree(n);
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PEERNAME, &n));
+	NUTS_MATCH(n, "push");
+	nng_strfree(n);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_pull_cannot_send(void)
+{
+	nng_socket s;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_FAIL(nng_send(s, "", 0, 0), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_pull_no_context(void)
+{
+	nng_socket s;
+	nng_ctx    ctx;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_FAIL(nng_ctx_open(&ctx, s), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_pull_not_writeable(void)
+{
+	int        fd;
+	nng_socket s;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_FAIL(nng_socket_get_int(s, NNG_OPT_SENDFD, &fd), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_pull_poll_readable(void)
+{
+	int        fd;
+	nng_socket pull;
+	nng_socket push;
+
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_PASS(nng_push0_open(&push));
+	NUTS_PASS(nng_socket_set_ms(pull, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(push, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_get_int(pull, NNG_OPT_RECVFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// Not readable if not connected!
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// Even after connect (no message yet)
+	NUTS_MARRY(pull, push);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// But once we send messages, it is.
+	// We have to send a request, in order to send a reply.
+	NUTS_SEND(push, "abc");
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	// and receiving makes it no longer ready
+	NUTS_RECV(pull, "abc");
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	NUTS_CLOSE(pull);
+	NUTS_CLOSE(push);
+}
+
+static void
+test_pull_close_pending(void)
+{
+	int        fd;
+	nng_socket pull;
+	nng_socket push;
+	nng_pipe   p1, p2;
+	char *     addr;
+
+	NUTS_ADDR(addr, "inproc");
+
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_PASS(nng_push0_open(&push));
+	NUTS_PASS(nng_socket_get_int(pull, NNG_OPT_RECVFD, &fd));
+	NUTS_TRUE(fd >= 0);
+	NUTS_MARRY_EX(pull, push, addr, &p1, &p2);
+
+	// Send a message -- it's ready for reading.
+	NUTS_SEND(push, "abc");
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	// NB: We have to close the pipe instead of the socket.
+	// This is because the socket won't notice the remote pipe
+	// disconnect until we collect the message and start another
+	// receive operation.
+	nng_pipe_close(p1);
+	nng_pipe_close(p2);
+
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	NUTS_CLOSE(pull);
+	NUTS_CLOSE(push);
+}
+
+void
+test_pull_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char *     addr;
+
+	NUTS_ADDR(addr, "inproc");
+
+	NUTS_PASS(nng_pull0_open(&s1));
+	NUTS_PASS(nng_pull0_open(&s2));
+
+	NUTS_PASS(nng_listen(s1, addr, NULL, 0));
+	NUTS_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	NUTS_SLEEP(100);
+	NUTS_PASS(nng_stats_get(&stats));
+
+	NUTS_TRUE(stats != NULL);
+	NUTS_TRUE((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	NUTS_TRUE((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	NUTS_TRUE(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	NUTS_TRUE(nng_stat_value(reject) > 0);
+
+	NUTS_CLOSE(s1);
+	NUTS_CLOSE(s2);
+	nng_stats_free(stats);
+}
+
+static void
+test_pull_recv_aio_stopped(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_stop(aio);
+	nng_recv_aio(s, aio);
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+}
+
+static void
+test_pull_close_recv(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	nng_aio_set_timeout(aio, 1000);
+	nng_recv_aio(s, aio);
+	NUTS_PASS(nng_close(s));
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECLOSED);
+
+	nng_aio_free(aio);
+}
+
+static void
+test_pull_recv_nonblock(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_recv_aio(s, aio);
+
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+}
+
+static void
+test_pull_recv_cancel(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_set_timeout(aio, 1000);
+	nng_recv_aio(s, aio);
+	nng_aio_abort(aio, NNG_ECANCELED);
+
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+}
+
+static void
+test_pull_cooked(void)
+{
+	nng_socket s;
+	bool       b;
+
+	NUTS_PASS(nng_pull0_open(&s));
+	NUTS_PASS(nng_socket_get_bool(s, NNG_OPT_RAW, &b));
+	NUTS_TRUE(!b);
+	NUTS_CLOSE(s);
+}
+
+TEST_LIST = {
+	{ "pull identity", test_pull_identity },
+	{ "pull cannot send", test_pull_cannot_send },
+	{ "pull no context", test_pull_no_context },
+	{ "pull not writeable", test_pull_not_writeable },
+	{ "pull poll readable", test_pull_poll_readable },
+	{ "pull close pending", test_pull_close_pending },
+	{ "pull validate peer", test_pull_validate_peer },
+	{ "pull recv aio stopped", test_pull_recv_aio_stopped },
+	{ "pull close recv", test_pull_close_recv },
+	{ "pull recv nonblock", test_pull_recv_nonblock },
+	{ "pull recv cancel", test_pull_recv_cancel },
+	{ "pull cooked", test_pull_cooked },
+	{ NULL, NULL },
+};
diff --git a/src/protocol/pipeline0/push.c b/src/protocol/pipeline0/push.c
index 90c94af9..8aa5311c 100644
--- a/src/protocol/pipeline0/push.c
+++ b/src/protocol/pipeline0/push.c
@@ -30,11 +30,15 @@ typedef struct push0_sock push0_sock;
 
 static void push0_send_cb(void *);
 static void push0_recv_cb(void *);
-static void push0_getq_cb(void *);
+static void push0_pipe_ready(push0_pipe *);
 
 // push0_sock is our per-socket protocol private structure.
 struct push0_sock {
-	nni_msgq *uwq;
+	nni_lmq      wq; // list of messages queued
+	nni_list     aq; // list of aio senders waiting
+	nni_list     pl; // list of pipes ready to send
+	nni_pollable writable;
+	nni_mtx      m;
 };
 
 // push0_pipe is our per-pipe protocol private structure.
@@ -43,23 +47,32 @@ struct push0_pipe {
 	push0_sock *  push;
 	nni_list_node node;
 
-	nni_aio *aio_recv;
-	nni_aio *aio_send;
-	nni_aio *aio_getq;
+	nni_aio aio_recv;
+	nni_aio aio_send;
 };
 
 static int
 push0_sock_init(void *arg, nni_sock *sock)
 {
 	push0_sock *s = arg;
-	s->uwq        = nni_sock_sendq(sock);
+	NNI_ARG_UNUSED(sock);
+
+	nni_mtx_init(&s->m);
+	nni_aio_list_init(&s->aq);
+	NNI_LIST_INIT(&s->pl, push0_pipe, node);
+	nni_lmq_init(&s->wq, 0); // initially we start unbuffered.
+	nni_pollable_init(&s->writable);
+
 	return (0);
 }
 
 static void
 push0_sock_fini(void *arg)
 {
-	NNI_ARG_UNUSED(arg);
+	push0_sock *s = arg;
+	nni_pollable_fini(&s->writable);
+	nni_lmq_fini(&s->wq);
+	nni_mtx_fini(&s->m);
 }
 
 static void
@@ -71,7 +84,14 @@ push0_sock_open(void *arg)
 static void
 push0_sock_close(void *arg)
 {
-	NNI_ARG_UNUSED(arg);
+	push0_sock *s = arg;
+	nni_aio *   a;
+	nni_mtx_lock(&s->m);
+	while ((a = nni_list_first(&s->aq)) != NULL) {
+		nni_aio_list_remove(a);
+		nni_aio_finish_error(a, NNG_ECLOSED);
+	}
+	nni_mtx_unlock(&s->m);
 }
 
 static void
@@ -79,9 +99,8 @@ push0_pipe_stop(void *arg)
 {
 	push0_pipe *p = arg;
 
-	nni_aio_stop(p->aio_recv);
-	nni_aio_stop(p->aio_send);
-	nni_aio_stop(p->aio_getq);
+	nni_aio_stop(&p->aio_recv);
+	nni_aio_stop(&p->aio_send);
 }
 
 static void
@@ -89,23 +108,17 @@ push0_pipe_fini(void *arg)
 {
 	push0_pipe *p = arg;
 
-	nni_aio_free(p->aio_recv);
-	nni_aio_free(p->aio_send);
-	nni_aio_free(p->aio_getq);
+	nni_aio_fini(&p->aio_recv);
+	nni_aio_fini(&p->aio_send);
 }
 
 static int
 push0_pipe_init(void *arg, nni_pipe *pipe, void *s)
 {
 	push0_pipe *p = arg;
-	int         rv;
 
-	if (((rv = nni_aio_alloc(&p->aio_recv, push0_recv_cb, p)) != 0) ||
-	    ((rv = nni_aio_alloc(&p->aio_send, push0_send_cb, p)) != 0) ||
-	    ((rv = nni_aio_alloc(&p->aio_getq, push0_getq_cb, p)) != 0)) {
-		push0_pipe_fini(p);
-		return (rv);
-	}
+	nni_aio_init(&p->aio_recv, push0_recv_cb, p);
+	nni_aio_init(&p->aio_send, push0_send_cb, p);
 	NNI_LIST_NODE_INIT(&p->node);
 	p->pipe = pipe;
 	p->push = s;
@@ -116,7 +129,6 @@ static int
 push0_pipe_start(void *arg)
 {
 	push0_pipe *p = arg;
-	push0_sock *s = p->push;
 
 	if (nni_pipe_peer(p->pipe) != NNI_PROTO_PULL_V0) {
 		return (NNG_EPROTO);
@@ -124,10 +136,8 @@ push0_pipe_start(void *arg)
 
 	// Schedule a receiver.  This is mostly so that we can detect
 	// a closed transport pipe.
-	nni_pipe_recv(p->pipe, p->aio_recv);
-
-	// Schedule a sender.
-	nni_msgq_aio_get(s->uwq, p->aio_getq);
+	nni_pipe_recv(p->pipe, &p->aio_recv);
+	push0_pipe_ready(p);
 
 	return (0);
 }
@@ -136,10 +146,20 @@ static void
 push0_pipe_close(void *arg)
 {
 	push0_pipe *p = arg;
+	push0_sock *s = p->push;
+
+	nni_aio_close(&p->aio_recv);
+	nni_aio_close(&p->aio_send);
+
+	nni_mtx_lock(&s->m);
+	if (nni_list_node_active(&p->node)) {
+		nni_list_node_remove(&p->node);
 
-	nni_aio_close(p->aio_recv);
-	nni_aio_close(p->aio_send);
-	nni_aio_close(p->aio_getq);
+		if (nni_list_empty(&s->pl) && nni_lmq_full(&s->wq)) {
+			nni_pollable_clear(&s->writable);
+		}
+	}
+	nni_mtx_unlock(&s->m);
 }
 
 static void
@@ -149,55 +169,155 @@ push0_recv_cb(void *arg)
 
 	// We normally expect to receive an error.  If a pipe actually
 	// sends us data, we just discard it.
-	if (nni_aio_result(p->aio_recv) != 0) {
+	if (nni_aio_result(&p->aio_recv) != 0) {
 		nni_pipe_close(p->pipe);
 		return;
 	}
-	nni_msg_free(nni_aio_get_msg(p->aio_recv));
-	nni_aio_set_msg(p->aio_recv, NULL);
-	nni_pipe_recv(p->pipe, p->aio_recv);
+	nni_msg_free(nni_aio_get_msg(&p->aio_recv));
+	nni_aio_set_msg(&p->aio_recv, NULL);
+	nni_pipe_recv(p->pipe, &p->aio_recv);
 }
 
 static void
-push0_send_cb(void *arg)
+push0_pipe_ready(push0_pipe *p)
 {
-	push0_pipe *p = arg;
 	push0_sock *s = p->push;
+	nni_msg *   m;
+	nni_aio *   a = NULL;
+	size_t      l;
+	bool        blocked;
+
+	nni_mtx_lock(&s->m);
+
+	blocked = nni_lmq_full(&s->wq) && nni_list_empty(&s->pl);
+
+	// if  message is waiting in the buffered queue
+	// then we prefer that.
+	if (nni_lmq_getq(&s->wq, &m) == 0) {
+		nni_aio_set_msg(&p->aio_send, m);
+		nni_pipe_send(p->pipe, &p->aio_send);
+
+		if ((a = nni_list_first(&s->aq)) != NULL) {
+			nni_aio_list_remove(a);
+			m = nni_aio_get_msg(a);
+			l = nni_msg_len(m);
+			nni_lmq_putq(&s->wq, m);
+		}
+
+	} else if ((a = nni_list_first(&s->aq)) != NULL) {
+		// Looks like we had the unbuffered case, but
+		// someone was waiting.
+		nni_aio_list_remove(a);
+		m = nni_aio_get_msg(a);
+		l = nni_msg_len(m);
+
+		nni_aio_set_msg(&p->aio_send, m);
+		nni_pipe_send(p->pipe, &p->aio_send);
+	} else {
+		// We had nothing to send.  Just put us in the ready list.
+		nni_list_append(&s->pl, p);
+	}
 
-	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;
+	if (blocked) {
+		// if we blocked, then toggle the status.
+		if ((!nni_lmq_full(&s->wq)) || (!nni_list_empty(&s->pl))) {
+			nni_pollable_raise(&s->writable);
+		}
 	}
 
-	nni_msgq_aio_get(s->uwq, p->aio_getq);
+	nni_mtx_unlock(&s->m);
+
+	if (a != NULL) {
+		nni_aio_set_msg(a, NULL);
+		nni_aio_finish_sync(a, 0, l);
+	}
 }
 
 static void
-push0_getq_cb(void *arg)
+push0_send_cb(void *arg)
 {
-	push0_pipe *p   = arg;
-	nni_aio *   aio = p->aio_getq;
+	push0_pipe *p = arg;
 
-	if (nni_aio_result(aio) != 0) {
-		// If the socket is closing, nothing else we can do.
+	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;
 	}
 
-	nni_aio_set_msg(p->aio_send, nni_aio_get_msg(aio));
-	nni_aio_set_msg(aio, NULL);
+	push0_pipe_ready(p);
+}
 
-	nni_pipe_send(p->pipe, p->aio_send);
+static void
+push0_cancel(nni_aio *aio, void *arg, int rv)
+{
+	push0_sock *s = arg;
+
+	nni_mtx_lock(&s->m);
+	if (nni_aio_list_active(aio)) {
+		nni_aio_list_remove(aio);
+		nni_aio_finish_error(aio, rv);
+	}
+	nni_mtx_unlock(&s->m);
 }
 
 static void
 push0_sock_send(void *arg, nni_aio *aio)
 {
 	push0_sock *s = arg;
+	push0_pipe *p;
+	nni_msg *   m;
+	size_t      l;
+	int         rv;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+
+	m = nni_aio_get_msg(aio);
+	l = nni_msg_len(m);
+
+	nni_mtx_lock(&s->m);
+
+	// First we want to see if we can send it right now.
+	// Note that we don't block the sender until the read is complete,
+	// only until we have committed to send it.
+	if ((p = nni_list_first(&s->pl)) != NULL) {
+		nni_list_remove(&s->pl, p);
+		// NB: We won't have had any waiters in the message queue
+		// or the aio queue, because we would not put the pipe
+		// in the ready list in that case.  Note though that the
+		// wq may be "full" if we are unbuffered.
+		if (nni_list_empty(&s->pl) && (nni_lmq_full(&s->wq))) {
+			nni_pollable_clear(&s->writable);
+		}
+		nni_aio_set_msg(aio, NULL);
+		nni_aio_finish(aio, 0, l);
+		nni_aio_set_msg(&p->aio_send, m);
+		nni_pipe_send(p->pipe, &p->aio_send);
+		nni_mtx_unlock(&s->m);
+		return;
+	}
 
-	nni_msgq_aio_put(s->uwq, aio);
+	// Can we maybe queue it.
+	if (nni_lmq_putq(&s->wq, m) == 0) {
+		// Yay, we can.  So we're done.
+		nni_aio_set_msg(aio, NULL);
+		nni_aio_finish(aio, 0, l);
+		if (nni_lmq_full(&s->wq)) {
+			nni_pollable_clear(&s->writable);
+		}
+		nni_mtx_unlock(&s->m);
+		return;
+	}
+
+	if ((rv = nni_aio_schedule(aio, push0_cancel, s)) != 0) {
+		nni_aio_finish_error(aio, rv);
+		nni_mtx_unlock(&s->m);
+		return;
+	}
+	nni_aio_list_append(&s->aq, aio);
+	nni_mtx_unlock(&s->m);
 }
 
 static void
@@ -207,6 +327,54 @@ push0_sock_recv(void *arg, nni_aio *aio)
 	nni_aio_finish_error(aio, NNG_ENOTSUP);
 }
 
+static int
+push0_set_send_buf_len(void *arg, const void *buf, size_t sz, nni_type t)
+{
+	push0_sock *s = arg;
+	int         val;
+	int         rv;
+
+	if ((rv = nni_copyin_int(&val, buf, sz, 0, 8192, t)) != 0) {
+		return (rv);
+	}
+	nni_mtx_lock(&s->m);
+	rv = nni_lmq_resize(&s->wq, (size_t) val);
+	// Changing the size of the queue can affect our readiness.
+	if (!nni_lmq_full(&s->wq)) {
+		nni_pollable_raise(&s->writable);
+	} else if (nni_list_empty(&s->pl)) {
+		nni_pollable_clear(&s->writable);
+	}
+	nni_mtx_unlock(&s->m);
+	return (rv);
+}
+
+static int
+push0_get_send_buf_len(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	push0_sock *s = arg;
+	int val;
+
+	nni_mtx_lock(&s->m);
+	val = nni_lmq_cap(&s->wq);
+	nni_mtx_unlock(&s->m);
+
+	return (nni_copyout_int(val, buf, szp, t));
+}
+
+static int
+push0_sock_get_send_fd(void *arg, void *buf, size_t *szp, nni_opt_type t)
+{
+	push0_sock *s = arg;
+	int         rv;
+	int         fd;
+
+	if ((rv = nni_pollable_getfd(&s->writable, &fd)) != 0) {
+		return (rv);
+	}
+	return (nni_copyout_int(fd, buf, szp, t));
+}
+
 static nni_proto_pipe_ops push0_pipe_ops = {
 	.pipe_size  = sizeof(push0_pipe),
 	.pipe_init  = push0_pipe_init,
@@ -217,7 +385,16 @@ static nni_proto_pipe_ops push0_pipe_ops = {
 };
 
 static nni_option push0_sock_options[] = {
-	// terminate list
+	{
+	    .o_name = NNG_OPT_SENDFD,
+	    .o_get  = push0_sock_get_send_fd,
+	},
+	{
+	    .o_name = NNG_OPT_SENDBUF,
+	    .o_get  = push0_get_send_buf_len,
+	    .o_set  = push0_set_send_buf_len,
+	},
+    // terminate list
 	{
 	    .o_name = NULL,
 	},
@@ -238,28 +415,29 @@ static nni_proto push0_proto = {
 	.proto_version  = NNI_PROTOCOL_VERSION,
 	.proto_self     = { NNI_PROTO_PUSH_V0, "push" },
 	.proto_peer     = { NNI_PROTO_PULL_V0, "pull" },
-	.proto_flags    = NNI_PROTO_FLAG_SND,
+	.proto_flags    = NNI_PROTO_FLAG_SND | NNI_PROTO_FLAG_NOMSGQ,
 	.proto_pipe_ops = &push0_pipe_ops,
 	.proto_sock_ops = &push0_sock_ops,
 };
 
 static nni_proto push0_proto_raw = {
-	.proto_version  = NNI_PROTOCOL_VERSION,
-	.proto_self     = { NNI_PROTO_PUSH_V0, "push" },
-	.proto_peer     = { NNI_PROTO_PULL_V0, "pull" },
-	.proto_flags    = NNI_PROTO_FLAG_SND | NNI_PROTO_FLAG_RAW,
+	.proto_version = NNI_PROTOCOL_VERSION,
+	.proto_self    = { NNI_PROTO_PUSH_V0, "push" },
+	.proto_peer    = { NNI_PROTO_PULL_V0, "pull" },
+	.proto_flags =
+	    NNI_PROTO_FLAG_SND | NNI_PROTO_FLAG_RAW | NNI_PROTO_FLAG_NOMSGQ,
 	.proto_pipe_ops = &push0_pipe_ops,
 	.proto_sock_ops = &push0_sock_ops,
 };
 
 int
-nng_push0_open(nng_socket *sidp)
+nng_push0_open(nng_socket *s)
 {
-	return (nni_proto_open(sidp, &push0_proto));
+	return (nni_proto_open(s, &push0_proto));
 }
 
 int
-nng_push0_open_raw(nng_socket *sidp)
+nng_push0_open_raw(nng_socket *s)
 {
-	return (nni_proto_open(sidp, &push0_proto_raw));
+	return (nni_proto_open(s, &push0_proto_raw));
 }
diff --git a/src/protocol/pipeline0/push_test.c b/src/protocol/pipeline0/push_test.c
new file mode 100644
index 00000000..d22ccaa4
--- /dev/null
+++ b/src/protocol/pipeline0/push_test.c
@@ -0,0 +1,525 @@
+//
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include <nuts.h>
+
+static void
+test_push_identity(void)
+{
+	nng_socket s;
+	int        p;
+	char *     n;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PROTO, &p));
+	NUTS_TRUE(p == NUTS_PROTO(5u, 0u)); // 80
+	NUTS_PASS(nng_socket_get_int(s, NNG_OPT_PEER, &p));
+	NUTS_TRUE(p == NUTS_PROTO(5u, 1u)); // 81
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PROTONAME, &n));
+	NUTS_MATCH(n, "push");
+	nng_strfree(n);
+	NUTS_PASS(nng_socket_get_string(s, NNG_OPT_PEERNAME, &n));
+	NUTS_MATCH(n, "pull");
+	nng_strfree(n);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_push_cannot_recv(void)
+{
+	nng_socket s;
+	nng_msg *  m = NULL;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_FAIL(nng_recvmsg(s, &m, 0), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_push_no_context(void)
+{
+	nng_socket s;
+	nng_ctx    ctx;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_FAIL(nng_ctx_open(&ctx, s), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_push_not_readable(void)
+{
+	int        fd;
+	nng_socket s;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_FAIL(nng_socket_get_int(s, NNG_OPT_RECVFD, &fd), NNG_ENOTSUP);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_push_poll_writable(void)
+{
+	int        fd;
+	nng_socket pull;
+	nng_socket push;
+
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_PASS(nng_push0_open(&push));
+	NUTS_PASS(nng_socket_set_ms(pull, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(push, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_get_int(push, NNG_OPT_SENDFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// This tests unbuffered sockets for now.
+	// Note that for this we are using unbuffered inproc.
+	// If using TCP or similar, then transport buffering will
+	// break assumptions in this test.
+
+	// Not writable if not connected!
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// After connect we can write.
+	NUTS_MARRY(pull, push);
+	NUTS_TRUE(nuts_poll_fd(fd) == true);
+
+	// But once we send a message, it is not anymore.
+	NUTS_SEND(push, "abc");
+	// Have to send a second message, because the remote socket
+	// will have consumed the first one.
+	NUTS_SEND(push, "def");
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// and receiving receiving the message makes it possible again.
+	NUTS_RECV(pull, "abc");
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	NUTS_CLOSE(pull);
+	NUTS_CLOSE(push);
+}
+
+static void
+test_push_poll_buffered(void)
+{
+	int        fd;
+	nng_socket pull;
+	nng_socket push;
+
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_PASS(nng_push0_open(&push));
+	NUTS_PASS(nng_socket_set_ms(pull, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(push, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_int(push, NNG_OPT_SENDBUF, 2));
+	NUTS_PASS(nng_socket_get_int(push, NNG_OPT_SENDFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// We can write two message while unbuffered.
+	NUTS_TRUE(nuts_poll_fd(fd));
+	NUTS_SEND(push, "abc");
+	NUTS_TRUE(nuts_poll_fd(fd));
+	NUTS_SEND(push, "def");
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// After connect we remote end will pick up one of them.
+	// Also, the local pipe itself will pick up one.  So we
+	// have two.
+	NUTS_MARRY(pull, push);
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd));
+	NUTS_SEND(push, "ghi");
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd));
+	NUTS_SEND(push, "jkl");
+	// Now it should be full.
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	// and receiving receiving the message makes it possible again.
+	NUTS_RECV(pull, "abc");
+	NUTS_SLEEP(100);
+	NUTS_TRUE(nuts_poll_fd(fd));
+	NUTS_RECV(pull, "def");
+	NUTS_RECV(pull, "ghi");
+	NUTS_RECV(pull, "jkl");
+
+	NUTS_CLOSE(pull);
+	NUTS_CLOSE(push);
+}
+
+static void
+test_push_poll_truncate(void)
+{
+	int        fd;
+	nng_socket pull;
+	nng_socket push;
+
+	// This test starts with a buffer and then truncates it to see
+	// that shortening the buffer has an impact.
+
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_PASS(nng_push0_open(&push));
+	NUTS_PASS(nng_socket_set_ms(pull, NNG_OPT_RECVTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_ms(push, NNG_OPT_SENDTIMEO, 1000));
+	NUTS_PASS(nng_socket_set_int(push, NNG_OPT_SENDBUF, 3));
+	NUTS_PASS(nng_socket_get_int(push, NNG_OPT_SENDFD, &fd));
+	NUTS_TRUE(fd >= 0);
+
+	// We can write two message while unbuffered.
+	NUTS_TRUE(nuts_poll_fd(fd));
+	NUTS_SEND(push, "abc");
+	NUTS_TRUE(nuts_poll_fd(fd));
+	NUTS_SEND(push, "def");
+	NUTS_TRUE(nuts_poll_fd(fd));
+
+	NUTS_PASS(nng_socket_set_int(push, NNG_OPT_SENDBUF, 1));
+	NUTS_TRUE(nuts_poll_fd(fd) == false);
+
+	NUTS_MARRY(pull, push);
+	NUTS_RECV(pull, "abc");
+	// def got dropped
+	NUTS_SEND(push, "ghi");
+	NUTS_RECV(pull, "ghi");
+
+	NUTS_CLOSE(pull);
+	NUTS_SLEEP(100);
+
+	// We have a buffer of one.
+	NUTS_SEND(push, "jkl");
+	// Resize to 0 (unbuffered)
+	NUTS_PASS(nng_socket_set_int(push, NNG_OPT_SENDBUF, 0));
+
+	// reopen the pull socket and connect it
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_MARRY(push, pull);
+
+	// jkl got dropped.
+	NUTS_SEND(push, "mno");
+	NUTS_RECV(pull, "mno");
+
+	NUTS_CLOSE(pull);
+	NUTS_CLOSE(push);
+}
+
+void
+test_push_validate_peer(void)
+{
+	nng_socket s1, s2;
+	nng_stat * stats;
+	nng_stat * reject;
+	char *     addr;
+
+	NUTS_ADDR(addr, "inproc");
+
+	NUTS_PASS(nng_push0_open(&s1));
+	NUTS_PASS(nng_push0_open(&s2));
+
+	NUTS_PASS(nng_listen(s1, addr, NULL, 0));
+	NUTS_PASS(nng_dial(s2, addr, NULL, NNG_FLAG_NONBLOCK));
+
+	NUTS_SLEEP(100);
+	NUTS_PASS(nng_stats_get(&stats));
+
+	NUTS_TRUE(stats != NULL);
+	NUTS_TRUE((reject = nng_stat_find_socket(stats, s1)) != NULL);
+	NUTS_TRUE((reject = nng_stat_find(reject, "reject")) != NULL);
+
+	NUTS_TRUE(nng_stat_type(reject) == NNG_STAT_COUNTER);
+	NUTS_TRUE(nng_stat_value(reject) > 0);
+
+	NUTS_CLOSE(s1);
+	NUTS_CLOSE(s2);
+	nng_stats_free(stats);
+}
+
+static void
+test_push_send_aio_stopped(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+
+	nng_aio_set_msg(aio, m);
+	nng_aio_stop(aio);
+	nng_send_aio(s, aio);
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+	nng_msg_free(m);
+}
+
+static void
+test_push_close_send(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	nng_aio_set_timeout(aio, 1000);
+	nng_aio_set_msg(aio, m);
+	nng_send_aio(s, aio);
+	NUTS_PASS(nng_close(s));
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECLOSED);
+
+	nng_aio_free(aio);
+	nng_msg_free(m);
+}
+
+static void
+test_push_send_nonblock(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+
+	nng_aio_set_timeout(aio, 0); // Instant timeout
+	nng_aio_set_msg(aio, m);
+	nng_send_aio(s, aio);
+
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+	nng_msg_free(m);
+}
+
+static void
+test_push_send_timeout(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+
+	nng_aio_set_timeout(aio, 10);
+	nng_aio_set_msg(aio, m);
+	nng_send_aio(s, aio);
+
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT);
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+	nng_msg_free(m);
+}
+
+static void
+test_push_send_cancel(void)
+{
+	nng_socket s;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+
+	nng_aio_set_timeout(aio, 1000);
+	nng_aio_set_msg(aio, m);
+	nng_send_aio(s, aio);
+	nng_aio_abort(aio, NNG_ECANCELED);
+
+	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+	nng_msg_free(m);
+}
+
+static void
+test_push_send_late_unbuffered(void)
+{
+	nng_socket s;
+	nng_socket pull;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append(m, "123\0", 4));
+
+	nng_aio_set_timeout(aio, 1000);
+	nng_aio_set_msg(aio, m);
+	nng_send_aio(s, aio);
+
+	NUTS_MARRY(s, pull);
+
+	NUTS_RECV(pull, "123");
+
+	nng_aio_wait(aio);
+	NUTS_PASS(nng_aio_result(aio));
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+}
+
+
+static void
+test_push_send_late_buffered(void)
+{
+	nng_socket s;
+	nng_socket pull;
+	nng_aio *  aio;
+	nng_msg *  m;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_pull0_open(&pull));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_setopt_int(s, NNG_OPT_SENDBUF, 2));
+	NUTS_PASS(nng_msg_alloc(&m, 0));
+	NUTS_PASS(nng_msg_append(m, "123\0", 4));
+
+	nng_aio_set_timeout(aio, 1000);
+	nng_aio_set_msg(aio, m);
+	nng_send_aio(s, aio);
+
+	NUTS_MARRY(s, pull);
+
+	NUTS_RECV(pull, "123");
+
+	nng_aio_wait(aio);
+	NUTS_PASS(nng_aio_result(aio));
+	NUTS_CLOSE(s);
+	nng_aio_free(aio);
+}
+
+static void
+test_push_cooked(void)
+{
+	nng_socket s;
+	bool       b;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_socket_get_bool(s, NNG_OPT_RAW, &b));
+	NUTS_TRUE(!b);
+	NUTS_CLOSE(s);
+}
+
+static void
+test_push_load_balance_buffered(void)
+{
+	nng_socket s;
+	nng_socket pull1;
+	nng_socket pull2;
+	nng_socket pull3;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_pull0_open(&pull1));
+	NUTS_PASS(nng_pull0_open(&pull2));
+	NUTS_PASS(nng_pull0_open(&pull3));
+	NUTS_PASS(nng_setopt_int(s, NNG_OPT_SENDBUF, 4));
+	NUTS_MARRY(s, pull1);
+	NUTS_MARRY(s, pull2);
+	NUTS_MARRY(s, pull3);
+	NUTS_SLEEP(100);
+	NUTS_SEND(s, "one");
+	NUTS_SEND(s, "two");
+	NUTS_SEND(s, "three");
+	NUTS_RECV(pull1, "one");
+	NUTS_RECV(pull2, "two");
+	NUTS_RECV(pull3, "three");
+	NUTS_CLOSE(s);
+	NUTS_CLOSE(pull1);
+	NUTS_CLOSE(pull2);
+	NUTS_CLOSE(pull3);
+}
+
+static void
+test_push_load_balance_unbuffered(void)
+{
+	nng_socket s;
+	nng_socket pull1;
+	nng_socket pull2;
+	nng_socket pull3;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_pull0_open(&pull1));
+	NUTS_PASS(nng_pull0_open(&pull2));
+	NUTS_PASS(nng_pull0_open(&pull3));
+	NUTS_MARRY(s, pull1);
+	NUTS_MARRY(s, pull2);
+	NUTS_MARRY(s, pull3);
+	NUTS_SLEEP(100);
+	NUTS_SEND(s, "one");
+	NUTS_SEND(s, "two");
+	NUTS_SEND(s, "three");
+	NUTS_RECV(pull1, "one");
+	NUTS_RECV(pull2, "two");
+	NUTS_RECV(pull3, "three");
+	// Loop around is unpredictable somewhat, because the the
+	// pull sockets can take different periods of time to return
+	// back to readiness.
+	NUTS_CLOSE(s);
+	NUTS_CLOSE(pull1);
+	NUTS_CLOSE(pull2);
+	NUTS_CLOSE(pull3);
+}
+
+static void
+test_push_send_buffer(void)
+{
+	nng_socket s;
+	int        v;
+	bool       b;
+	size_t     sz;
+
+	NUTS_PASS(nng_push0_open(&s));
+	NUTS_PASS(nng_getopt_int(s, NNG_OPT_SENDBUF, &v));
+	NUTS_TRUE(v == 0);
+	NUTS_FAIL(nng_getopt_bool(s, NNG_OPT_SENDBUF, &b), NNG_EBADTYPE);
+	sz = 1;
+	NUTS_FAIL(nng_getopt(s, NNG_OPT_SENDBUF, &b, &sz), NNG_EINVAL);
+	NUTS_FAIL(nng_setopt_int(s, NNG_OPT_SENDBUF, -1), NNG_EINVAL);
+	NUTS_FAIL(nng_setopt_int(s, NNG_OPT_SENDBUF, 100000), NNG_EINVAL);
+	NUTS_FAIL(nng_setopt_bool(s, NNG_OPT_SENDBUF, false), NNG_EBADTYPE);
+	NUTS_FAIL(nng_setopt(s, NNG_OPT_SENDBUF, &b, 1), NNG_EINVAL);
+	NUTS_PASS(nng_setopt_int(s, NNG_OPT_SENDBUF, 100));
+	NUTS_PASS(nng_getopt_int(s, NNG_OPT_SENDBUF, &v));
+	NUTS_TRUE(v == 100);
+	NUTS_CLOSE(s);
+}
+
+TEST_LIST = {
+	{ "push identity", test_push_identity },
+	{ "push cannot recv", test_push_cannot_recv },
+	{ "push no context", test_push_no_context },
+	{ "push not readable", test_push_not_readable },
+	{ "push poll writable", test_push_poll_writable },
+	{ "push poll buffered", test_push_poll_buffered },
+	{ "push poll truncate", test_push_poll_truncate },
+	{ "push validate peer", test_push_validate_peer },
+	{ "push send aio stopped", test_push_send_aio_stopped },
+	{ "push close send", test_push_close_send },
+	{ "push send nonblock", test_push_send_nonblock },
+	{ "push send timeout", test_push_send_timeout },
+	{ "push send cancel", test_push_send_cancel },
+	{ "push send late buffered", test_push_send_late_buffered },
+	{ "push send late unbuffered", test_push_send_late_unbuffered },
+	{ "push cooked", test_push_cooked },
+	{ "push load balance buffered", test_push_load_balance_buffered },
+	{ "push load balance unbuffered", test_push_load_balance_unbuffered },
+	{ "push send buffer", test_push_send_buffer },
+	{ NULL, NULL },
+};
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index e3cbc5ad..5fcd4323 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -152,7 +152,6 @@ add_nng_test(wss 30)
 add_nng_test1(zt 60 NNG_TRANSPORT_ZEROTIER)
 
 add_nng_test(bus 5)
-add_nng_test(pipeline 5)
 add_nng_test(reqctx 5)
 add_nng_test(reqstress 60)
 
diff --git a/tests/pipeline.c b/tests/pipeline.c
deleted file mode 100644
index 034cb314..00000000
--- a/tests/pipeline.c
+++ /dev/null
@@ -1,170 +0,0 @@
-//
-// Copyright 2018 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
-// copy of which should be located in the distribution where this
-// file was obtained (LICENSE.txt).  A copy of the license may also be
-// found online at https://opensource.org/licenses/MIT.
-//
-
-#include <string.h>
-
-#include <nng/nng.h>
-#include <nng/protocol/pipeline0/pull.h>
-#include <nng/protocol/pipeline0/push.h>
-#include <nng/supplemental/util/platform.h>
-
-#include "convey.h"
-#include "stubs.h"
-
-#define APPENDSTR(m, s) nng_msg_append(m, s, strlen(s))
-#define CHECKSTR(m, s)                   \
-	So(nng_msg_len(m) == strlen(s)); \
-	So(memcmp(nng_msg_body(m), s, strlen(s)) == 0)
-#define MILLISECOND(x) (x)
-
-TestMain("PIPELINE (PUSH/PULL) pattern", {
-	atexit(nng_fini);
-	const char *addr = "inproc://test";
-	Convey("We can create a PUSH socket", {
-		nng_socket push;
-
-		So(nng_push_open(&push) == 0);
-
-		Reset({ nng_close(push); });
-
-		Convey("Recv fails", {
-			nng_msg *msg;
-			So(nng_recvmsg(push, &msg, 0) == NNG_ENOTSUP);
-		});
-	});
-
-	Convey("We can create a PULL socket", {
-		nng_socket pull;
-		So(nng_pull_open(&pull) == 0);
-
-		Reset({ nng_close(pull); });
-
-		Convey("Send fails", {
-			nng_msg *msg;
-			So(nng_msg_alloc(&msg, 0) == 0);
-			So(nng_sendmsg(pull, msg, 0) == NNG_ENOTSUP);
-			nng_msg_free(msg);
-		});
-	});
-
-	Convey("We can create a linked PUSH/PULL pair", {
-		nng_socket push;
-		nng_socket pull;
-		nng_socket what;
-
-		So(nng_push_open(&push) == 0);
-		So(nng_pull_open(&pull) == 0);
-		So(nng_push_open(&what) == 0);
-
-		Reset({
-			nng_close(push);
-			nng_close(pull);
-			nng_close(what);
-		});
-
-		// Its important to avoid a startup race that the
-		// sender be the dialer.  Otherwise you need a delay
-		// since the server accept is really asynchronous.
-		So(nng_listen(pull, addr, NULL, 0) == 0);
-		So(nng_dial(push, addr, NULL, 0) == 0);
-		So(nng_dial(what, addr, NULL, 0) == 0);
-		So(nng_close(what) == 0);
-
-		nng_msleep(20);
-
-		Convey("Push can send messages, and pull can recv", {
-			nng_msg *msg;
-
-			So(nng_msg_alloc(&msg, 0) == 0);
-			APPENDSTR(msg, "hello");
-			So(nng_sendmsg(push, msg, 0) == 0);
-			msg = NULL;
-			So(nng_recvmsg(pull, &msg, 0) == 0);
-			So(msg != NULL);
-			CHECKSTR(msg, "hello");
-			nng_msg_free(msg);
-		});
-	});
-
-	Convey("Load balancing", {
-		nng_msg *    abc;
-		nng_msg *    def;
-		nng_duration msecs;
-		nng_socket   push;
-		nng_socket   pull1;
-		nng_socket   pull2;
-		nng_socket   pull3;
-
-		So(nng_push_open(&push) == 0);
-		So(nng_pull_open(&pull1) == 0);
-		So(nng_pull_open(&pull2) == 0);
-		So(nng_pull_open(&pull3) == 0);
-
-		Reset({
-			nng_close(push);
-			nng_close(pull1);
-			nng_close(pull2);
-			nng_close(pull3);
-		});
-
-		// We need to increase the buffer from zero, because
-		// there is no guarantee that the various listeners
-		// will be present, which means that they will push
-		// back during load balancing.  Adding a small buffer
-		// ensures that we can write to each stream, even if
-		// the listeners are not running yet.
-		So(nng_setopt_int(push, NNG_OPT_RECVBUF, 4) == 0);
-		So(nng_setopt_int(push, NNG_OPT_SENDBUF, 4) == 0);
-		So(nng_setopt_int(pull1, NNG_OPT_RECVBUF, 4) == 0);
-		So(nng_setopt_int(pull1, NNG_OPT_SENDBUF, 4) == 0);
-		So(nng_setopt_int(pull2, NNG_OPT_RECVBUF, 4) == 0);
-		So(nng_setopt_int(pull2, NNG_OPT_SENDBUF, 4) == 0);
-		So(nng_setopt_int(pull3, NNG_OPT_RECVBUF, 4) == 0);
-		So(nng_setopt_int(pull3, NNG_OPT_SENDBUF, 4) == 0);
-
-		So(nng_msg_alloc(&abc, 0) == 0);
-		APPENDSTR(abc, "abc");
-		So(nng_msg_alloc(&def, 0) == 0);
-		APPENDSTR(def, "def");
-
-		msecs = MILLISECOND(100);
-		So(nng_setopt_ms(pull1, NNG_OPT_RECVTIMEO, msecs) == 0);
-		So(nng_setopt_ms(pull2, NNG_OPT_RECVTIMEO, msecs) == 0);
-		So(nng_setopt_ms(pull3, NNG_OPT_RECVTIMEO, msecs) == 0);
-		So(nng_listen(push, addr, NULL, 0) == 0);
-		So(nng_dial(pull1, addr, NULL, 0) == 0);
-		So(nng_dial(pull2, addr, NULL, 0) == 0);
-		So(nng_dial(pull3, addr, NULL, 0) == 0);
-		So(nng_close(pull3) == 0);
-
-		// So pull3 might not be done accepting yet, but pull1
-		// and pull2 definitely are, because otherwise the
-		// server couldn't have gotten to the accept.  (The
-		// accept logic is single threaded.)  Let's wait a bit
-		// though, to ensure that stuff has settled.
-		nng_msleep(100);
-
-		So(nng_sendmsg(push, abc, 0) == 0);
-		So(nng_sendmsg(push, def, 0) == 0);
-
-		abc = NULL;
-		def = NULL;
-
-		So(nng_recvmsg(pull1, &abc, 0) == 0);
-		CHECKSTR(abc, "abc");
-		So(nng_recvmsg(pull2, &def, 0) == 0);
-		CHECKSTR(def, "def");
-		nng_msg_free(abc);
-		nng_msg_free(def);
-
-		So(nng_recvmsg(pull1, &abc, 0) == NNG_ETIMEDOUT);
-		So(nng_recvmsg(pull2, &abc, 0) == NNG_ETIMEDOUT);
-	});
-})