fixes #112 Need to move some stuff from socket to message queues

11 files changed, 371 insertions, 130 deletions

diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c
index cad21989..046aa19a 100644
--- a/src/protocol/bus/bus.c
+++ b/src/protocol/bus/bus.c
@@ -22,6 +22,9 @@ typedef struct bus_pipe bus_pipe;
 typedef struct bus_sock bus_sock;
 
 static void bus_sock_getq(bus_sock *);
+static void bus_sock_send(void *, nni_aio *);
+static void bus_sock_recv(void *, nni_aio *);
+
 static void bus_pipe_getq(bus_pipe *);
 static void bus_pipe_send(bus_pipe *);
 static void bus_pipe_recv(bus_pipe *);
@@ -39,6 +42,8 @@ struct bus_sock {
 	nni_aio * aio_getq;
 	nni_list  pipes;
 	nni_mtx   mtx;
+	nni_msgq *uwq;
+	nni_msgq *urq;
 };
 
 // A bus_pipe is our per-pipe protocol private structure.
@@ -82,6 +87,8 @@ bus_sock_init(void **sp, nni_sock *nsock)
 	}
 	s->nsock = nsock;
 	s->raw   = 0;
+	s->uwq   = nni_sock_sendq(nsock);
+	s->urq   = nni_sock_recvq(nsock);
 
 	*sp = s;
 	return (0);
@@ -259,7 +266,6 @@ bus_sock_getq_cb(void *arg)
 	bus_sock *s = arg;
 	bus_pipe *p;
 	bus_pipe *lastp;
-	nni_msgq *uwq = nni_sock_sendq(s->nsock);
 	nni_msg * msg;
 	nni_msg * dup;
 	uint32_t  sender;
@@ -338,6 +344,24 @@ bus_sock_getopt_raw(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->raw, buf, szp));
 }
 
+static void
+bus_sock_send(void *arg, nni_aio *aio)
+{
+	bus_sock *s = arg;
+
+	nni_sock_send_pending(s->nsock);
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+bus_sock_recv(void *arg, nni_aio *aio)
+{
+	bus_sock *s = arg;
+
+	nni_sock_recv_pending(s->nsock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 static nni_proto_pipe_ops bus_pipe_ops = {
 	.pipe_init  = bus_pipe_init,
 	.pipe_fini  = bus_pipe_fini,
@@ -360,6 +384,8 @@ static nni_proto_sock_ops bus_sock_ops = {
 	.sock_fini    = bus_sock_fini,
 	.sock_open    = bus_sock_open,
 	.sock_close   = bus_sock_close,
+	.sock_send    = bus_sock_send,
+	.sock_recv    = bus_sock_recv,
 	.sock_options = bus_sock_options,
 };
 
diff --git a/src/protocol/pair/pair_v0.c b/src/protocol/pair/pair_v0.c
index a87af593..cca03cc8 100644
--- a/src/protocol/pair/pair_v0.c
+++ b/src/protocol/pair/pair_v0.c
@@ -243,6 +243,24 @@ pair0_sock_getopt_raw(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->raw, buf, szp));
 }
 
+static void
+pair0_sock_send(void *arg, nni_aio *aio)
+{
+	pair0_sock *s = arg;
+
+	nni_sock_send_pending(s->nsock);
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+pair0_sock_recv(void *arg, nni_aio *aio)
+{
+	pair0_sock *s = arg;
+
+	nni_sock_recv_pending(s->nsock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 static nni_proto_pipe_ops pair0_pipe_ops = {
 	.pipe_init  = pair0_pipe_init,
 	.pipe_fini  = pair0_pipe_fini,
@@ -265,6 +283,8 @@ static nni_proto_sock_ops pair0_sock_ops = {
 	.sock_fini    = pair0_sock_fini,
 	.sock_open    = pair0_sock_open,
 	.sock_close   = pair0_sock_close,
+	.sock_send    = pair0_sock_send,
+	.sock_recv    = pair0_sock_recv,
 	.sock_options = pair0_sock_options,
 };
 
diff --git a/src/protocol/pair/pair_v1.c b/src/protocol/pair/pair_v1.c
index 6c9d4bd3..f43a4785 100644
--- a/src/protocol/pair/pair_v1.c
+++ b/src/protocol/pair/pair_v1.c
@@ -446,6 +446,24 @@ pair1_sock_getopt_poly(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->poly, buf, szp));
 }
 
+static void
+pair1_sock_send(void *arg, nni_aio *aio)
+{
+	pair1_sock *s = arg;
+
+	nni_sock_send_pending(s->nsock);
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+pair1_sock_recv(void *arg, nni_aio *aio)
+{
+	pair1_sock *s = arg;
+
+	nni_sock_recv_pending(s->nsock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 static nni_proto_pipe_ops pair1_pipe_ops = {
 	.pipe_init  = pair1_pipe_init,
 	.pipe_fini  = pair1_pipe_fini,
@@ -478,6 +496,8 @@ static nni_proto_sock_ops pair1_sock_ops = {
 	.sock_fini    = pair1_sock_fini,
 	.sock_open    = pair1_sock_open,
 	.sock_close   = pair1_sock_close,
+	.sock_recv    = pair1_sock_recv,
+	.sock_send    = pair1_sock_send,
 	.sock_options = pair1_sock_options,
 };
 
diff --git a/src/protocol/pipeline/pull.c b/src/protocol/pipeline/pull.c
index 80db2425..7dd0c8ed 100644
--- a/src/protocol/pipeline/pull.c
+++ b/src/protocol/pipeline/pull.c
@@ -26,6 +26,7 @@ static void pull_putq(pull_pipe *, nni_msg *);
 struct pull_sock {
 	nni_msgq *urq;
 	int       raw;
+	nni_sock *sock;
 };
 
 // A pull_pipe is our per-pipe protocol private structure.
@@ -44,10 +45,11 @@ pull_sock_init(void **sp, nni_sock *sock)
 	if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	s->raw = 0;
-	s->urq = nni_sock_recvq(sock);
-	*sp    = s;
-	nni_sock_senderr(sock, NNG_ENOTSUP);
+	s->raw  = 0;
+	s->urq  = nni_sock_recvq(sock);
+	s->sock = sock;
+
+	*sp = s;
 	return (0);
 }
 
@@ -185,6 +187,21 @@ pull_sock_getopt_raw(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->raw, buf, szp));
 }
 
+static void
+pull_sock_send(void *arg, nni_aio *aio)
+{
+	nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static void
+pull_sock_recv(void *arg, nni_aio *aio)
+{
+	pull_sock *s = arg;
+
+	nni_sock_recv_pending(s->sock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 static nni_proto_pipe_ops pull_pipe_ops = {
 	.pipe_init  = pull_pipe_init,
 	.pipe_fini  = pull_pipe_fini,
@@ -207,6 +224,8 @@ static nni_proto_sock_ops pull_sock_ops = {
 	.sock_fini    = pull_sock_fini,
 	.sock_open    = pull_sock_open,
 	.sock_close   = pull_sock_close,
+	.sock_send    = pull_sock_send,
+	.sock_recv    = pull_sock_recv,
 	.sock_options = pull_sock_options,
 };
 
diff --git a/src/protocol/pipeline/push.c b/src/protocol/pipeline/push.c
index 77db7fcb..af7b80ca 100644
--- a/src/protocol/pipeline/push.c
+++ b/src/protocol/pipeline/push.c
@@ -54,7 +54,6 @@ push_sock_init(void **sp, nni_sock *sock)
 	s->sock = sock;
 	s->uwq  = nni_sock_sendq(sock);
 	*sp     = s;
-	nni_sock_recverr(sock, NNG_ENOTSUP);
 	return (0);
 }
 
@@ -205,6 +204,21 @@ push_sock_getopt_raw(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->raw, buf, szp));
 }
 
+static void
+push_sock_send(void *arg, nni_aio *aio)
+{
+	push_sock *s = arg;
+
+	nni_sock_send_pending(s->sock);
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+push_sock_recv(void *arg, nni_aio *aio)
+{
+	nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
 static nni_proto_pipe_ops push_pipe_ops = {
 	.pipe_init  = push_pipe_init,
 	.pipe_fini  = push_pipe_fini,
@@ -228,6 +242,8 @@ static nni_proto_sock_ops push_sock_ops = {
 	.sock_open    = push_sock_open,
 	.sock_close   = push_sock_close,
 	.sock_options = push_sock_options,
+	.sock_send    = push_sock_send,
+	.sock_recv    = push_sock_recv,
 };
 
 static nni_proto push_proto = {
diff --git a/src/protocol/pubsub/pub.c b/src/protocol/pubsub/pub.c
index 03f4603a..4604d0ff 100644
--- a/src/protocol/pubsub/pub.c
+++ b/src/protocol/pubsub/pub.c
@@ -82,7 +82,6 @@ pub_sock_init(void **sp, nni_sock *sock)
 	s->uwq = nni_sock_sendq(sock);
 
 	*sp = s;
-	nni_sock_recverr(sock, NNG_ENOTSUP);
 	return (0);
 }
 
@@ -281,6 +280,21 @@ pub_sock_getopt_raw(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->raw, buf, szp));
 }
 
+static void
+pub_sock_recv(void *arg, nni_aio *aio)
+{
+	nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static void
+pub_sock_send(void *arg, nni_aio *aio)
+{
+	pub_sock *s = arg;
+
+	nni_sock_send_pending(s->sock);
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
 static nni_proto_pipe_ops pub_pipe_ops = {
 	.pipe_init  = pub_pipe_init,
 	.pipe_fini  = pub_pipe_fini,
@@ -303,6 +317,8 @@ static nni_proto_sock_ops pub_sock_ops = {
 	.sock_fini    = pub_sock_fini,
 	.sock_open    = pub_sock_open,
 	.sock_close   = pub_sock_close,
+	.sock_send    = pub_sock_send,
+	.sock_recv    = pub_sock_recv,
 	.sock_options = pub_sock_options,
 };
 
diff --git a/src/protocol/pubsub/sub.c b/src/protocol/pubsub/sub.c
index 8b2ed209..323bbb2f 100644
--- a/src/protocol/pubsub/sub.c
+++ b/src/protocol/pubsub/sub.c
@@ -40,6 +40,7 @@ struct sub_sock {
 	nni_list  topics;
 	nni_msgq *urq;
 	int       raw;
+	nni_mtx   lk;
 };
 
 // An nni_rep_pipe is our per-pipe protocol private structure.
@@ -58,13 +59,13 @@ sub_sock_init(void **sp, nni_sock *sock)
 	if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
 		return (NNG_ENOMEM);
 	}
+	nni_mtx_init(&s->lk);
 	NNI_LIST_INIT(&s->topics, sub_topic, node);
 	s->sock = sock;
 	s->raw  = 0;
 
 	s->urq = nni_sock_recvq(sock);
-	nni_sock_senderr(sock, NNG_ENOTSUP);
-	*sp = s;
+	*sp    = s;
 	return (0);
 }
 
@@ -79,6 +80,7 @@ sub_sock_fini(void *arg)
 		nni_free(topic->buf, topic->len);
 		NNI_FREE_STRUCT(topic);
 	}
+	nni_mtx_fini(&s->lk);
 	NNI_FREE_STRUCT(s);
 }
 
@@ -190,6 +192,7 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
 	sub_topic *topic;
 	sub_topic *newtopic;
 
+	nni_mtx_lock(&s->lk);
 	NNI_LIST_FOREACH (&s->topics, topic) {
 		int rv;
 
@@ -201,6 +204,7 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
 		if (rv == 0) {
 			if (topic->len == sz) {
 				// Already inserted.
+				nni_mtx_unlock(&s->lk);
 				return (0);
 			}
 			if (topic->len > sz) {
@@ -212,9 +216,11 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
 	}
 
 	if ((newtopic = NNI_ALLOC_STRUCT(newtopic)) == NULL) {
+		nni_mtx_unlock(&s->lk);
 		return (NNG_ENOMEM);
 	}
 	if ((newtopic->buf = nni_alloc(sz)) == NULL) {
+		nni_mtx_unlock(&s->lk);
 		return (NNG_ENOMEM);
 	}
 	NNI_LIST_NODE_INIT(&newtopic->node);
@@ -225,6 +231,7 @@ sub_subscribe(void *arg, const void *buf, size_t sz)
 	} else {
 		nni_list_append(&s->topics, newtopic);
 	}
+	nni_mtx_unlock(&s->lk);
 	return (0);
 }
 
@@ -235,6 +242,7 @@ sub_unsubscribe(void *arg, const void *buf, size_t sz)
 	sub_topic *topic;
 	int        rv;
 
+	nni_mtx_lock(&s->lk);
 	NNI_LIST_FOREACH (&s->topics, topic) {
 		if (topic->len >= sz) {
 			rv = memcmp(topic->buf, buf, sz);
@@ -244,18 +252,22 @@ sub_unsubscribe(void *arg, const void *buf, size_t sz)
 		if (rv == 0) {
 			if (topic->len == sz) {
 				nni_list_remove(&s->topics, topic);
+				nni_mtx_unlock(&s->lk);
 				nni_free(topic->buf, topic->len);
 				NNI_FREE_STRUCT(topic);
 				return (0);
 			}
 			if (topic->len > sz) {
+				nni_mtx_unlock(&s->lk);
 				return (NNG_ENOENT);
 			}
 		}
 		if (rv > 0) {
+			nni_mtx_unlock(&s->lk);
 			return (NNG_ENOENT);
 		}
 	}
+	nni_mtx_unlock(&s->lk);
 	return (NNG_ENOENT);
 }
 
@@ -273,8 +285,23 @@ sub_sock_getopt_raw(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->raw, buf, szp));
 }
 
+static void
+sub_sock_send(void *arg, nni_aio *aio)
+{
+	nni_aio_finish_error(aio, NNG_ENOTSUP);
+}
+
+static void
+sub_sock_recv(void *arg, nni_aio *aio)
+{
+	sub_sock *s = arg;
+
+	nni_sock_recv_pending(s->sock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 static nni_msg *
-sub_sock_rfilter(void *arg, nni_msg *msg)
+sub_sock_filter(void *arg, nni_msg *msg)
 {
 	sub_sock * s = arg;
 	sub_topic *topic;
@@ -282,7 +309,9 @@ sub_sock_rfilter(void *arg, nni_msg *msg)
 	size_t     len;
 	int        match;
 
+	nni_mtx_lock(&s->lk);
 	if (s->raw) {
+		nni_mtx_unlock(&s->lk);
 		return (msg);
 	}
 
@@ -308,6 +337,7 @@ sub_sock_rfilter(void *arg, nni_msg *msg)
 			break;
 		}
 	}
+	nni_mtx_unlock(&s->lk);
 	if (!match) {
 		nni_msg_free(msg);
 		return (NULL);
@@ -349,8 +379,10 @@ static nni_proto_sock_ops sub_sock_ops = {
 	.sock_fini    = sub_sock_fini,
 	.sock_open    = sub_sock_open,
 	.sock_close   = sub_sock_close,
+	.sock_send    = sub_sock_send,
+	.sock_recv    = sub_sock_recv,
+	.sock_filter  = sub_sock_filter,
 	.sock_options = sub_sock_options,
-	.sock_rfilter = sub_sock_rfilter,
 };
 
 static nni_proto sub_proto = {
diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index cd9411d9..510beab3 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -32,6 +32,7 @@ struct rep_sock {
 	nni_sock *  sock;
 	nni_msgq *  uwq;
 	nni_msgq *  urq;
+	nni_mtx     lk;
 	int         raw;
 	int         ttl;
 	nni_idhash *pipes;
@@ -62,6 +63,7 @@ rep_sock_fini(void *arg)
 	if (s->btrace != NULL) {
 		nni_free(s->btrace, s->btrace_len);
 	}
+	nni_mtx_fini(&s->lk);
 	NNI_FREE_STRUCT(s);
 }
 
@@ -74,6 +76,7 @@ rep_sock_init(void **sp, nni_sock *sock)
 	if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
 		return (NNG_ENOMEM);
 	}
+	nni_mtx_init(&s->lk);
 	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);
@@ -89,7 +92,6 @@ rep_sock_init(void **sp, nni_sock *sock)
 	s->urq        = nni_sock_recvq(sock);
 
 	*sp = s;
-	nni_sock_senderr(sock, NNG_ESTATE);
 
 	return (0);
 }
@@ -215,6 +217,7 @@ 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.
+	// XXX: LOCKING?!?!
 	if ((rv = nni_idhash_find(s->pipes, id, (void **) &p)) == 0) {
 		rv = nni_msgq_tryput(p->sendq, msg);
 	}
@@ -347,10 +350,10 @@ rep_sock_setopt_raw(void *arg, const void *buf, size_t sz)
 {
 	rep_sock *s = arg;
 	int       rv;
+
+	nni_mtx_lock(&s->lk);
 	rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
-	if (rv == 0) {
-		nni_sock_senderr(s->sock, s->raw ? 0 : NNG_ESTATE);
-	}
+	nni_mtx_unlock(&s->lk);
 	return (rv);
 }
 
@@ -376,53 +379,18 @@ rep_sock_getopt_maxttl(void *arg, void *buf, size_t *szp)
 }
 
 static nni_msg *
-rep_sock_sfilter(void *arg, nni_msg *msg)
-{
-	rep_sock *s = arg;
-
-	if (s->raw) {
-		return (msg);
-	}
-
-	// Cannot send again until a receive is done...
-	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 (s->btrace == NULL) {
-		nni_msg_free(msg);
-		return (NULL);
-	}
-
-	// drop anything else in the header...
-	nni_msg_header_clear(msg);
-
-	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(s->btrace, s->btrace_len);
-	s->btrace     = NULL;
-	s->btrace_len = 0;
-	return (msg);
-}
-
-static nni_msg *
-rep_sock_rfilter(void *arg, nni_msg *msg)
+rep_sock_filter(void *arg, nni_msg *msg)
 {
 	rep_sock *s = arg;
 	char *    header;
 	size_t    len;
 
+	nni_mtx_lock(&s->lk);
 	if (s->raw) {
+		nni_mtx_unlock(&s->lk);
 		return (msg);
 	}
 
-	nni_sock_senderr(s->sock, 0);
 	len    = nni_msg_header_len(msg);
 	header = nni_msg_header(msg);
 	if (s->btrace != NULL) {
@@ -437,9 +405,61 @@ rep_sock_rfilter(void *arg, nni_msg *msg)
 	s->btrace_len = len;
 	memcpy(s->btrace, header, len);
 	nni_msg_header_clear(msg);
+	nni_mtx_unlock(&s->lk);
 	return (msg);
 }
 
+static void
+rep_sock_send(void *arg, nni_aio *aio)
+{
+	rep_sock *s = arg;
+	int       rv;
+	nni_msg * msg;
+
+	nni_mtx_lock(&s->lk);
+	if (s->raw) {
+		// Pass thru
+		nni_mtx_unlock(&s->lk);
+		nni_sock_send_pending(s->sock);
+		nni_msgq_aio_put(s->uwq, aio);
+		return;
+	}
+	if (s->btrace == NULL) {
+		nni_mtx_unlock(&s->lk);
+		nni_aio_finish_error(aio, NNG_ESTATE);
+		return;
+	}
+
+	msg = nni_aio_get_msg(aio);
+
+	// drop anything else in the header...  (it should already be
+	// empty, but there can be stale backtrace info there.)
+	nni_msg_header_clear(msg);
+
+	if ((rv = nni_msg_header_append(msg, s->btrace, s->btrace_len)) != 0) {
+		nni_mtx_unlock(&s->lk);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+
+	nni_free(s->btrace, s->btrace_len);
+	s->btrace     = NULL;
+	s->btrace_len = 0;
+
+	nni_mtx_unlock(&s->lk);
+	nni_sock_send_pending(s->sock);
+	nni_msgq_aio_put(s->uwq, aio);
+}
+
+static void
+rep_sock_recv(void *arg, nni_aio *aio)
+{
+	rep_sock *s = arg;
+
+	nni_sock_recv_pending(s->sock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 // 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 rep_pipe_ops = {
@@ -470,8 +490,9 @@ static nni_proto_sock_ops rep_sock_ops = {
 	.sock_open    = rep_sock_open,
 	.sock_close   = rep_sock_close,
 	.sock_options = rep_sock_options,
-	.sock_rfilter = rep_sock_rfilter,
-	.sock_sfilter = rep_sock_sfilter,
+	.sock_filter  = rep_sock_filter,
+	.sock_send    = rep_sock_send,
+	.sock_recv    = rep_sock_recv,
 };
 
 static nni_proto nni_rep_proto = {
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index 24d01df2..81abd306 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -99,7 +99,7 @@ req_sock_init(void **sp, nni_sock *sock)
 	s->uwq    = nni_sock_sendq(sock);
 	s->urq    = nni_sock_recvq(sock);
 	*sp       = s;
-	nni_sock_recverr(sock, NNG_ESTATE);
+
 	return (0);
 }
 
@@ -249,12 +249,7 @@ static int
 req_sock_setopt_raw(void *arg, const void *buf, size_t sz)
 {
 	req_sock *s = arg;
-	int       rv;
-	rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
-	if (rv == 0) {
-		nni_sock_recverr(s->sock, s->raw ? 0 : NNG_ESTATE);
-	}
-	return (rv);
+	return (nni_setopt_int(&s->raw, buf, sz, 0, 1));
 }
 
 static int
@@ -505,35 +500,46 @@ req_resend(req_sock *s)
 	}
 }
 
-static nni_msg *
-req_sock_sfilter(void *arg, nni_msg *msg)
+static void
+req_sock_send(void *arg, nni_aio *aio)
 {
 	req_sock *s = arg;
 	uint32_t  id;
+	size_t    len;
+	nni_msg * msg;
+	int       rv;
 
+	nni_mtx_lock(&s->mtx);
 	if (s->raw) {
-		// No automatic retry, and the request ID must
-		// be in the header coming down.
-		return (msg);
+		nni_mtx_unlock(&s->mtx);
+		nni_sock_send_pending(s->sock);
+		nni_msgq_aio_put(s->uwq, aio);
+		return;
 	}
 
+	msg = nni_aio_get_msg(aio);
+	len = nni_msg_len(msg);
+
+	// In cooked mode, because we need to manage our own resend logic,
+	// we bypass the upper writeq entirely.
+
 	// 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 = (s->nextid++) | 0x80000000u;
-
 	// Request ID is in big endian format.
 	NNI_PUT32(s->reqid, id);
 
-	if (nni_msg_header_append(msg, s->reqid, 4) != 0) {
-		// Should be ENOMEM.
-		nni_msg_free(msg);
-		return (NULL);
+	if ((rv = nni_msg_header_append(msg, s->reqid, 4)) != 0) {
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
 	}
 
-	// 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(&s->mtx);
+	// XXX: I think we should just not do this... and leave the
+	// socket "permanently writeable".  This does screw up all the
+	// backpressure.
+	// nni_sock_send_pending(s->sock);
 
 	// If another message is there, this cancels it.
 	if (s->reqmsg != NULL) {
@@ -541,6 +547,8 @@ req_sock_sfilter(void *arg, nni_msg *msg)
 		s->reqmsg = NULL;
 	}
 
+	nni_aio_set_msg(aio, NULL);
+
 	// Make a duplicate message... for retries.
 	s->reqmsg = msg;
 	// Schedule for immediate send
@@ -548,40 +556,41 @@ req_sock_sfilter(void *arg, nni_msg *msg)
 	s->wantw  = 1;
 
 	req_resend(s);
-	nni_mtx_unlock(&s->mtx);
 
-	// Clear the error condition.
-	nni_sock_recverr(s->sock, 0);
+	nni_mtx_unlock(&s->mtx);
 
-	return (NULL);
+	nni_aio_finish(aio, 0, len);
 }
 
 static nni_msg *
-req_sock_rfilter(void *arg, nni_msg *msg)
+req_sock_filter(void *arg, nni_msg *msg)
 {
 	req_sock *s = arg;
 	nni_msg * rmsg;
 
+	nni_mtx_lock(&s->mtx);
 	if (s->raw) {
 		// Pass it unmolested
+		nni_mtx_unlock(&s->mtx);
 		return (msg);
 	}
 
 	if (nni_msg_header_len(msg) < 4) {
+		nni_mtx_unlock(&s->mtx);
 		nni_msg_free(msg);
 		return (NULL);
 	}
 
-	nni_mtx_lock(&s->mtx);
-
 	if ((rmsg = s->reqmsg) == NULL) {
-		// We had no outstanding request.
+		// We had no outstanding request.  (Perhaps canceled,
+		// or duplicate response.)
 		nni_mtx_unlock(&s->mtx);
 		nni_msg_free(msg);
 		return (NULL);
 	}
+
 	if (memcmp(nni_msg_header(msg), s->reqid, 4) != 0) {
-		// Wrong request id
+		// Wrong request id.
 		nni_mtx_unlock(&s->mtx);
 		nni_msg_free(msg);
 		return (NULL);
@@ -591,12 +600,29 @@ req_sock_rfilter(void *arg, nni_msg *msg)
 	s->pendpipe = NULL;
 	nni_mtx_unlock(&s->mtx);
 
-	nni_sock_recverr(s->sock, NNG_ESTATE);
 	nni_msg_free(rmsg);
 
 	return (msg);
 }
 
+static void
+req_sock_recv(void *arg, nni_aio *aio)
+{
+	req_sock *s = arg;
+
+	nni_mtx_lock(&s->mtx);
+	if (!s->raw) {
+		if (s->reqmsg == NULL) {
+			nni_mtx_unlock(&s->mtx);
+			nni_aio_finish_error(aio, NNG_ESTATE);
+			return;
+		}
+	}
+	nni_mtx_unlock(&s->mtx);
+	nni_sock_recv_pending(s->sock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 static nni_proto_pipe_ops req_pipe_ops = {
 	.pipe_init  = req_pipe_init,
 	.pipe_fini  = req_pipe_fini,
@@ -630,8 +656,9 @@ static nni_proto_sock_ops req_sock_ops = {
 	.sock_open    = req_sock_open,
 	.sock_close   = req_sock_close,
 	.sock_options = req_sock_options,
-	.sock_rfilter = req_sock_rfilter,
-	.sock_sfilter = req_sock_sfilter,
+	.sock_filter  = req_sock_filter,
+	.sock_send    = req_sock_send,
+	.sock_recv    = req_sock_recv,
 };
 
 static nni_proto req_proto = {
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index fcb067b0..4c3ea8e3 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -77,6 +77,7 @@ resp_sock_init(void **sp, nni_sock *nsock)
 	if ((s = NNI_ALLOC_STRUCT(s)) == NULL) {
 		return (NNG_ENOMEM);
 	}
+	nni_mtx_init(&s->mtx);
 	if (((rv = nni_idhash_init(&s->pipes)) != 0) ||
 	    ((rv = nni_aio_init(&s->aio_getq, resp_sock_getq_cb, s)) != 0)) {
 		resp_sock_fini(s);
@@ -91,10 +92,7 @@ resp_sock_init(void **sp, nni_sock *nsock)
 	s->urq        = nni_sock_recvq(nsock);
 	s->uwq        = nni_sock_sendq(nsock);
 
-	nni_mtx_init(&s->mtx);
-
 	*sp = s;
-	nni_sock_senderr(nsock, NNG_ESTATE);
 	return (0);
 }
 
@@ -349,9 +347,9 @@ resp_sock_setopt_raw(void *arg, const void *buf, size_t sz)
 	resp_sock *s = arg;
 	int        rv;
 
-	if ((rv = nni_setopt_int(&s->raw, buf, sz, 0, 1)) == 0) {
-		nni_sock_senderr(s->nsock, s->raw ? 0 : NNG_ESTATE);
-	}
+	nni_mtx_lock(&s->mtx);
+	rv = nni_setopt_int(&s->raw, buf, sz, 0, 1);
+	nni_mtx_unlock(&s->mtx);
 	return (rv);
 }
 
@@ -376,54 +374,62 @@ resp_sock_getopt_maxttl(void *arg, void *buf, size_t *szp)
 	return (nni_getopt_int(s->ttl, buf, szp));
 }
 
-static nni_msg *
-resp_sock_sfilter(void *arg, nni_msg *msg)
+static void
+resp_sock_send(void *arg, nni_aio *aio)
 {
 	resp_sock *s = arg;
+	nni_msg *  msg;
+	int        rv;
 
+	nni_mtx_lock(&s->mtx);
 	if (s->raw) {
-		return (msg);
+		nni_mtx_unlock(&s->mtx);
+		nni_sock_send_pending(s->nsock);
+		nni_msgq_aio_put(s->uwq, aio);
+		return;
 	}
 
-	// Cannot send again until a receive is done...
-	nni_sock_senderr(s->nsock, NNG_ESTATE);
+	msg = nni_aio_get_msg(aio);
 
 	// If we have a stored backtrace, append it to the header...
 	// if we don't have a backtrace, discard the message.
 	if (s->btrace == NULL) {
-		nni_msg_free(msg);
-		return (NULL);
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, NNG_ESTATE);
+		return;
 	}
 
 	// drop anything else in the header...
 	nni_msg_header_clear(msg);
 
-	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);
+	if ((rv = nni_msg_header_append(msg, s->btrace, s->btrace_len)) != 0) {
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
 	}
 
 	nni_free(s->btrace, s->btrace_len);
 	s->btrace     = NULL;
 	s->btrace_len = 0;
-	return (msg);
+
+	nni_mtx_unlock(&s->mtx);
+	nni_sock_send_pending(s->nsock);
+	nni_msgq_aio_put(s->uwq, aio);
 }
 
 static nni_msg *
-resp_sock_rfilter(void *arg, nni_msg *msg)
+resp_sock_filter(void *arg, nni_msg *msg)
 {
 	resp_sock *s = arg;
 	char *     header;
 	size_t     len;
 
+	nni_mtx_lock(&s->mtx);
 	if (s->raw) {
+		nni_mtx_unlock(&s->mtx);
 		return (msg);
 	}
 
-	nni_sock_senderr(s->nsock, 0);
 	len    = nni_msg_header_len(msg);
 	header = nni_msg_header(msg);
 	if (s->btrace != NULL) {
@@ -432,15 +438,26 @@ resp_sock_rfilter(void *arg, nni_msg *msg)
 		s->btrace_len = 0;
 	}
 	if ((s->btrace = nni_alloc(len)) == NULL) {
+		nni_mtx_unlock(&s->mtx);
 		nni_msg_free(msg);
 		return (NULL);
 	}
 	s->btrace_len = len;
 	memcpy(s->btrace, header, len);
 	nni_msg_header_clear(msg);
+	nni_mtx_unlock(&s->mtx);
 	return (msg);
 }
 
+static void
+resp_sock_recv(void *arg, nni_aio *aio)
+{
+	resp_sock *s = arg;
+
+	nni_sock_recv_pending(s->nsock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
 static nni_proto_pipe_ops resp_pipe_ops = {
 	.pipe_init  = resp_pipe_init,
 	.pipe_fini  = resp_pipe_fini,
@@ -468,9 +485,10 @@ static nni_proto_sock_ops resp_sock_ops = {
 	.sock_fini    = resp_sock_fini,
 	.sock_open    = resp_sock_open,
 	.sock_close   = resp_sock_close,
+	.sock_filter  = resp_sock_filter,
+	.sock_send    = resp_sock_send,
+	.sock_recv    = resp_sock_recv,
 	.sock_options = resp_sock_options,
-	.sock_rfilter = resp_sock_rfilter,
-	.sock_sfilter = resp_sock_sfilter,
 };
 
 static nni_proto resp_proto = {
diff --git a/src/protocol/survey/survey.c b/src/protocol/survey/survey.c
index 1205402e..1c15054f 100644
--- a/src/protocol/survey/survey.c
+++ b/src/protocol/survey/survey.c
@@ -92,7 +92,6 @@ surv_sock_init(void **sp, nni_sock *nsock)
 	s->urq      = nni_sock_recvq(nsock);
 
 	*sp = s;
-	nni_sock_recverr(nsock, NNG_ESTATE);
 	return (0);
 }
 
@@ -273,11 +272,12 @@ surv_sock_setopt_raw(void *arg, const void *buf, size_t sz)
 	surv_sock *s = arg;
 	int        rv;
 
+	nni_mtx_lock(&s->mtx);
 	if ((rv = nni_setopt_int(&s->raw, buf, sz, 0, 1)) == 0) {
-		nni_sock_recverr(s->nsock, s->raw ? 0 : NNG_ESTATE);
 		s->survid = 0;
 		nni_timer_cancel(&s->timer);
 	}
+	nni_mtx_unlock(&s->mtx);
 	return (rv);
 }
 
@@ -344,22 +344,43 @@ surv_timeout(void *arg)
 {
 	surv_sock *s = arg;
 
-	nni_sock_lock(s->nsock);
+	nni_mtx_lock(&s->mtx);
 	s->survid = 0;
-	nni_sock_recverr(s->nsock, NNG_ESTATE);
+	nni_mtx_unlock(&s->mtx);
 	nni_msgq_set_get_error(s->urq, NNG_ETIMEDOUT);
-	nni_sock_unlock(s->nsock);
 }
 
-static nni_msg *
-surv_sock_sfilter(void *arg, nni_msg *msg)
+static void
+surv_sock_recv(void *arg, nni_aio *aio)
+{
+	surv_sock *s = arg;
+
+	nni_mtx_lock(&s->mtx);
+	if (s->survid == 0) {
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, NNG_ESTATE);
+		return;
+	}
+	nni_mtx_unlock(&s->mtx);
+	nni_sock_recv_pending(s->nsock);
+	nni_msgq_aio_get(s->urq, aio);
+}
+
+static void
+surv_sock_send(void *arg, nni_aio *aio)
 {
 	surv_sock *s = arg;
+	nni_msg *  msg;
+	int        rv;
 
+	nni_mtx_lock(&s->mtx);
 	if (s->raw) {
 		// No automatic retry, and the request ID must
 		// be in the header coming down.
-		return (msg);
+		nni_mtx_unlock(&s->mtx);
+		nni_sock_send_pending(s->nsock);
+		nni_msgq_aio_put(s->uwq, aio);
+		return;
 	}
 
 	// Generate a new request ID.  We always set the high
@@ -367,10 +388,12 @@ surv_sock_sfilter(void *arg, nni_msg *msg)
 	// backtrace.  (Pipe IDs have the high order bit clear.)
 	s->survid = (s->nextid++) | 0x80000000u;
 
-	if (nni_msg_header_append_u32(msg, s->survid) != 0) {
-		// Should be ENOMEM.
-		nni_msg_free(msg);
-		return (NULL);
+	msg = nni_aio_get_msg(aio);
+	nni_msg_header_clear(msg);
+	if ((rv = nni_msg_header_append_u32(msg, s->survid)) != 0) {
+		nni_mtx_unlock(&s->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
 	}
 
 	// If another message is there, this cancels it.  We move the
@@ -379,20 +402,21 @@ surv_sock_sfilter(void *arg, nni_msg *msg)
 	s->expire = nni_clock() + s->survtime;
 	nni_timer_schedule(&s->timer, s->expire);
 
-	// Clear the error condition.
-	nni_sock_recverr(s->nsock, 0);
-	// nni_msgq_set_get_error(nni_sock_recvq(psock->nsock), 0);
+	nni_mtx_unlock(&s->mtx);
 
-	return (msg);
+	nni_sock_send_pending(s->nsock);
+	nni_msgq_aio_put(s->uwq, aio);
 }
 
 static nni_msg *
-surv_sock_rfilter(void *arg, nni_msg *msg)
+surv_sock_filter(void *arg, nni_msg *msg)
 {
 	surv_sock *s = arg;
 
+	nni_mtx_lock(&s->mtx);
 	if (s->raw) {
 		// Pass it unmolested
+		nni_mtx_unlock(&s->mtx);
 		return (msg);
 	}
 
@@ -402,6 +426,7 @@ surv_sock_rfilter(void *arg, nni_msg *msg)
 		nni_msg_free(msg);
 		return (NULL);
 	}
+	nni_mtx_unlock(&s->mtx);
 
 	return (msg);
 }
@@ -433,9 +458,10 @@ static nni_proto_sock_ops surv_sock_ops = {
 	.sock_fini    = surv_sock_fini,
 	.sock_open    = surv_sock_open,
 	.sock_close   = surv_sock_close,
+	.sock_send    = surv_sock_send,
+	.sock_recv    = surv_sock_recv,
+	.sock_filter  = surv_sock_filter,
 	.sock_options = surv_sock_options,
-	.sock_rfilter = surv_sock_rfilter,
-	.sock_sfilter = surv_sock_sfilter,
 };
 
 static nni_proto surv_proto = {