Surveyor pattern callback-driven.

6 files changed, 451 insertions, 324 deletions

diff --git a/src/core/msgqueue.c b/src/core/msgqueue.c
index 47b98629..60744ba1 100644
--- a/src/core/msgqueue.c
+++ b/src/core/msgqueue.c
@@ -217,27 +217,43 @@ nni_msgq_notify(nni_msgq *mq, nni_msgq_notify_fn fn, void *arg)
 
 
 void
-nni_msgq_set_put_error(nni_msgq *mq, int error)
+nni_msgq_set_get_error(nni_msgq *mq, int error)
 {
+	nni_aio *naio;
+	nni_aio *aio;
+
+	// Let all pending blockers know we are closing the queue.
 	nni_mtx_lock(&mq->mq_lock);
-	mq->mq_puterr = error;
-	if (error) {
-		mq->mq_wwait = 0;
-		nni_cv_wake(&mq->mq_writeable);
+	if (error != 0) {
+		naio = nni_list_first(&mq->mq_aio_getq);
+		while ((aio = naio) != NULL) {
+			naio = nni_list_next(&mq->mq_aio_getq, aio);
+			nni_list_remove(&mq->mq_aio_getq, aio);
+			nni_aio_finish(aio, error, 0);
+		}
 	}
+	mq->mq_geterr = error;
 	nni_mtx_unlock(&mq->mq_lock);
 }
 
 
 void
-nni_msgq_set_get_error(nni_msgq *mq, int error)
+nni_msgq_set_put_error(nni_msgq *mq, int error)
 {
+	nni_aio *naio;
+	nni_aio *aio;
+
+	// Let all pending blockers know we are closing the queue.
 	nni_mtx_lock(&mq->mq_lock);
-	mq->mq_geterr = error;
-	if (error) {
-		mq->mq_rwait = 0;
-		nni_cv_wake(&mq->mq_readable);
+	if (error != 0) {
+		naio = nni_list_first(&mq->mq_aio_putq);
+		while ((aio = naio) != NULL) {
+			naio = nni_list_next(&mq->mq_aio_getq, aio);
+			nni_list_remove(&mq->mq_aio_getq, aio);
+			nni_aio_finish(aio, error, 0);
+		}
 	}
+	mq->mq_puterr = error;
 	nni_mtx_unlock(&mq->mq_lock);
 }
 
@@ -245,34 +261,27 @@ nni_msgq_set_get_error(nni_msgq *mq, int error)
 void
 nni_msgq_set_error(nni_msgq *mq, int error)
 {
-	nni_mtx_lock(&mq->mq_lock);
-	mq->mq_geterr = error;
-	mq->mq_puterr = error;
-	if (error) {
-		mq->mq_rwait = 0;
-		mq->mq_wwait = 0;
-		nni_cv_wake(&mq->mq_readable);
-		nni_cv_wake(&mq->mq_writeable);
-	}
-	nni_mtx_unlock(&mq->mq_lock);
-}
-
+	nni_aio *naio;
+	nni_aio *aio;
 
-// nni_msgq_signal raises a signal on the signal object. This allows a
-// waiter to be signaled, so that it can be woken e.g. due to a pipe closing.
-// Note that the signal object must be *zero* if no signal is raised.
-void
-nni_msgq_signal(nni_msgq *mq, int *signal)
-{
+	// Let all pending blockers know we are closing the queue.
 	nni_mtx_lock(&mq->mq_lock);
-	*signal = 1;
-
-	// We have to wake everyone.
-	mq->mq_rwait = 0;
-	mq->mq_wwait = 0;
-	nni_cv_wake(&mq->mq_readable);
-	nni_cv_wake(&mq->mq_writeable);
-	nni_cv_wake(&mq->mq_notify_cv);
+	if (error != 0) {
+		naio = nni_list_first(&mq->mq_aio_getq);
+		while ((aio = naio) != NULL) {
+			naio = nni_list_next(&mq->mq_aio_getq, aio);
+			nni_list_remove(&mq->mq_aio_getq, aio);
+			nni_aio_finish(aio, error, 0);
+		}
+		naio = nni_list_first(&mq->mq_aio_putq);
+		while ((aio = naio) != NULL) {
+			naio = nni_list_next(&mq->mq_aio_getq, aio);
+			nni_list_remove(&mq->mq_aio_getq, aio);
+			nni_aio_finish(aio, error, 0);
+		}
+	}
+	mq->mq_puterr = error;
+	mq->mq_geterr = error;
 	nni_mtx_unlock(&mq->mq_lock);
 }
 
@@ -393,6 +402,11 @@ nni_msgq_aio_put(nni_msgq *mq, nni_aio *aio)
 		nni_mtx_unlock(&mq->mq_lock);
 		return;
 	}
+	if (mq->mq_puterr) {
+		nni_aio_finish(aio, mq->mq_puterr, 0);
+		nni_mtx_unlock(&mq->mq_lock);
+		return;
+	}
 	nni_list_append(&mq->mq_aio_putq, aio);
 	nni_msgq_run_putq(mq);
 	nni_msgq_run_notify(mq);
@@ -416,6 +430,11 @@ nni_msgq_aio_get(nni_msgq *mq, nni_aio *aio)
 		nni_mtx_unlock(&mq->mq_lock);
 		return;
 	}
+	if (mq->mq_geterr) {
+		nni_aio_finish(aio, mq->mq_geterr, 0);
+		nni_mtx_unlock(&mq->mq_lock);
+		return;
+	}
 	nni_list_append(&mq->mq_aio_getq, aio);
 	nni_msgq_run_getq(mq);
 	nni_msgq_run_notify(mq);
@@ -755,14 +774,6 @@ nni_msgq_get(nni_msgq *mq, nni_msg **msgp)
 }
 
 
-// XXX Remove this later.
-int
-nni_msgq_get_sig(nni_msgq *mq, nni_msg **msgp, nni_signal *signal)
-{
-	return (nni_msgq_get_(mq, msgp, NNI_TIME_NEVER, signal));
-}
-
-
 int
 nni_msgq_put_until(nni_msgq *mq, nni_msg *msg, nni_time expire)
 {
@@ -789,13 +800,6 @@ nni_msgq_put(nni_msgq *mq, nni_msg *msg)
 }
 
 
-int
-nni_msgq_put_sig(nni_msgq *mq, nni_msg *msg, nni_signal *signal)
-{
-	return (nni_msgq_put_(mq, msg, NNI_TIME_NEVER, signal));
-}
-
-
 void
 nni_msgq_drain(nni_msgq *mq, nni_time expire)
 {
diff --git a/src/core/msgqueue.h b/src/core/msgqueue.h
index de23ebac..51143706 100644
--- a/src/core/msgqueue.h
+++ b/src/core/msgqueue.h
@@ -72,23 +72,6 @@ extern int nni_msgq_put_until(nni_msgq *, nni_msg *, nni_time);
 // a message from the queue, it will return NNG_ETIMEDOUT.
 extern int nni_msgq_get_until(nni_msgq *, nni_msg **, nni_time);
 
-// nni_msgq_put_sig is an enhanced version of nni_msgq_put, but it
-// can be interrupted by nni_msgqueue_signal using the same final pointer,
-// which can be thought of as a turnstile.  If interrupted it returns EINTR.
-// The turnstile should be initialized to zero.
-extern int nni_msgq_put_sig(nni_msgq *, nni_msg *, nni_signal *);
-
-// nni_msgq_get_sig is an enhanced version of nni_msgq_get_t, but it
-// can be interrupted by nni_msgq_signal using the same final pointer,
-// which can be thought of as a turnstile.  If interrupted it returns EINTR.
-// The turnstile should be initialized to zero.
-extern int nni_msgq_get_sig(nni_msgq *, nni_msg **, nni_signal *);
-
-// nni_msgq_signal delivers a signal / interrupt to waiters blocked in
-// the msgq, if they have registered an interest in the same turnstile.
-// It modifies the turnstile's value under the lock to a non-zero value.
-extern void nni_msgq_signal(nni_msgq *, nni_signal *);
-
 // nni_msgq_set_error sets an error condition on the message queue,
 // which causes all current and future readers/writes to return the
 // given error condition (if non-zero).  Threads waiting to put or get
diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index 751a851b..736e4099 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -159,14 +159,12 @@ nni_rep_pipe_fini(void *arg)
 {
 	nni_rep_pipe *rp = arg;
 
-	if (rp != NULL) {
-		nni_msgq_fini(rp->sendq);
-		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_FREE_STRUCT(rp);
-	}
+	nni_msgq_fini(rp->sendq);
+	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_FREE_STRUCT(rp);
 }
 
 
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index 553ef0bf..f37913e6 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -110,12 +110,10 @@ nni_req_sock_fini(void *arg)
 {
 	nni_req_sock *req = arg;
 
-	if (req != NULL) {
-		if (req->reqmsg != NULL) {
-			nni_msg_free(req->reqmsg);
-		}
-		NNI_FREE_STRUCT(req);
+	if (req->reqmsg != NULL) {
+		nni_msg_free(req->reqmsg);
 	}
+	NNI_FREE_STRUCT(req);
 }
 
 
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index 8a7239b3..a70b08e6 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -19,14 +19,24 @@
 typedef struct nni_resp_pipe	nni_resp_pipe;
 typedef struct nni_resp_sock	nni_resp_sock;
 
+static void nni_resp_recv_cb(void *);
+static void nni_resp_putq_cb(void *);
+static void nni_resp_getq_cb(void *);
+static void nni_resp_send_cb(void *);
+static void nni_resp_sock_getq_cb(void *);
+static void nni_resp_pipe_fini(void *);
+
 // An nni_resp_sock is our per-socket protocol private structure.
 struct nni_resp_sock {
 	nni_sock *	nsock;
+	nni_msgq *	urq;
+	nni_msgq *	uwq;
 	int		raw;
 	int		ttl;
 	nni_idhash	pipes;
 	char *		btrace;
 	size_t		btrace_len;
+	nni_aio		aio_getq;
 };
 
 // An nni_resp_pipe is our per-pipe protocol private structure.
@@ -34,7 +44,10 @@ struct nni_resp_pipe {
 	nni_pipe *	npipe;
 	nni_resp_sock * psock;
 	nni_msgq *	sendq;
-	int		sigclose;
+	nni_aio		aio_getq;
+	nni_aio		aio_putq;
+	nni_aio		aio_send;
+	nni_aio		aio_recv;
 };
 
 static int
@@ -51,10 +64,18 @@ nni_resp_sock_init(void **pp, nni_sock *nsock)
 	psock->raw = 0;
 	psock->btrace = NULL;
 	psock->btrace_len = 0;
+	psock->urq = nni_sock_recvq(nsock);
+	psock->uwq = nni_sock_sendq(nsock);
 	if ((rv = nni_idhash_init(&psock->pipes)) != 0) {
 		NNI_FREE_STRUCT(psock);
 		return (rv);
 	}
+	rv = nni_aio_init(&psock->aio_getq, nni_resp_sock_getq_cb, psock);
+	if (rv != 0) {
+		nni_idhash_fini(&psock->pipes);
+		NNI_FREE_STRUCT(psock);
+		return (rv);
+	}
 
 	*pp = psock;
 	nni_sock_senderr(nsock, NNG_ESTATE);
@@ -68,6 +89,7 @@ nni_resp_sock_fini(void *arg)
 	nni_resp_sock *psock = arg;
 
 	if (psock != NULL) {
+		nni_aio_fini(&psock->aio_getq);
 		nni_idhash_fini(&psock->pipes);
 		if (psock->btrace != NULL) {
 			nni_free(psock->btrace, psock->btrace_len);
@@ -77,6 +99,24 @@ nni_resp_sock_fini(void *arg)
 }
 
 
+static void
+nni_resp_sock_open(void *arg)
+{
+	nni_resp_sock *psock = arg;
+
+	nni_msgq_aio_get(psock->uwq, &psock->aio_getq);
+}
+
+
+static void
+nni_resp_sock_close(void *arg)
+{
+	nni_resp_sock *psock = arg;
+
+	nni_msgq_aio_cancel(psock->uwq, &psock->aio_getq);
+}
+
+
 static int
 nni_resp_pipe_init(void **pp, nni_pipe *npipe, void *psock)
 {
@@ -87,12 +127,32 @@ nni_resp_pipe_init(void **pp, nni_pipe *npipe, void *psock)
 		return (NNG_ENOMEM);
 	}
 	if ((rv = nni_msgq_init(&ppipe->sendq, 2)) != 0) {
-		NNI_FREE_STRUCT(ppipe);
+		nni_resp_pipe_fini(ppipe);
+		return (rv);
+	}
+	rv = nni_aio_init(&ppipe->aio_putq, nni_resp_putq_cb, ppipe);
+	if (rv != 0) {
+		nni_resp_pipe_fini(ppipe);
+		return (rv);
+	}
+	rv = nni_aio_init(&ppipe->aio_recv, nni_resp_recv_cb, ppipe);
+	if (rv != 0) {
+		nni_resp_pipe_fini(ppipe);
+		return (rv);
+	}
+	rv = nni_aio_init(&ppipe->aio_getq, nni_resp_getq_cb, ppipe);
+	if (rv != 0) {
+		nni_resp_pipe_fini(ppipe);
+		return (rv);
+	}
+	rv = nni_aio_init(&ppipe->aio_send, nni_resp_send_cb, ppipe);
+	if (rv != 0) {
+		nni_resp_pipe_fini(ppipe);
 		return (rv);
 	}
+
 	ppipe->npipe = npipe;
 	ppipe->psock = psock;
-	ppipe->sigclose = 0;
 	*pp = ppipe;
 	return (0);
 }
@@ -103,10 +163,12 @@ nni_resp_pipe_fini(void *arg)
 {
 	nni_resp_pipe *ppipe = arg;
 
-	if (ppipe != NULL) {
-		nni_msgq_fini(ppipe->sendq);
-		NNI_FREE_STRUCT(ppipe);
-	}
+	nni_msgq_fini(ppipe->sendq);
+	nni_aio_fini(&ppipe->aio_putq);
+	nni_aio_fini(&ppipe->aio_getq);
+	nni_aio_fini(&ppipe->aio_send);
+	nni_aio_fini(&ppipe->aio_recv);
+	NNI_FREE_STRUCT(ppipe);
 }
 
 
@@ -118,6 +180,12 @@ nni_resp_pipe_add(void *arg)
 	int rv;
 
 	rv = nni_idhash_insert(&psock->pipes, nni_pipe_id(ppipe->npipe), ppipe);
+
+	nni_pipe_incref(ppipe->npipe);
+	nni_pipe_aio_recv(ppipe->npipe, &ppipe->aio_recv);
+
+	nni_pipe_incref(ppipe->npipe);
+	nni_msgq_aio_get(ppipe->sendq, &ppipe->aio_getq);
 	return (rv);
 }
 
@@ -129,150 +197,175 @@ nni_resp_pipe_rem(void *arg)
 	nni_resp_sock *psock = ppipe->psock;
 
 	nni_idhash_remove(&psock->pipes, nni_pipe_id(ppipe->npipe));
+
+	nni_msgq_close(ppipe->sendq);
+	nni_msgq_aio_cancel(psock->urq, &ppipe->aio_putq);
 }
 
 
 // nni_resp_sock_send 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.
-static void
-nni_resp_sock_send(void *arg)
+// or slow pipe from gumming up the works for the entire socket.s
+
+void
+nni_resp_sock_getq_cb(void *arg)
 {
 	nni_resp_sock *psock = arg;
-	nni_msgq *uwq = nni_sock_sendq(psock->nsock);
-	nni_mtx *mx = nni_sock_mtx(psock->nsock);
 	nni_msg *msg;
+	uint8_t *header;
+	uint32_t id;
+	nni_resp_pipe *ppipe;
+	int rv;
 
-	for (;;) {
-		uint8_t *header;
-		uint32_t id;
-		nni_resp_pipe *ppipe;
-		int rv;
+	if (nni_aio_result(&psock->aio_getq) != 0) {
+		return;
+	}
+	msg = psock->aio_getq.a_msg;
+	psock->aio_getq.a_msg = NULL;
 
-		if ((rv = nni_msgq_get(uwq, &msg)) != 0) {
-			break;
-		}
-		// We yank the outgoing pipe id from the header
-		if (nni_msg_header_len(msg) < 4) {
-			nni_msg_free(msg);
-			continue;
-		}
-		header = nni_msg_header(msg);
-		NNI_GET32(header, id);
-		nni_msg_trim_header(msg, 4);
+	// We yank the outgoing pipe id from the header
+	if (nni_msg_header_len(msg) < 4) {
+		nni_msg_free(msg);
+		// We can't really close down the socket, so just
+		// keep going.
+		nni_msgq_aio_get(psock->uwq, &psock->aio_getq);
+		return;
+	}
+	header = nni_msg_header(msg);
+	NNI_GET32(header, id);
+	nni_msg_trim_header(msg, 4);
 
-		nni_mtx_lock(mx);
-		if (nni_idhash_find(&psock->pipes, id, (void **) &ppipe) != 0) {
-			nni_mtx_unlock(mx);
-			nni_msg_free(msg);
-			continue;
-		}
-		// Try a non-blocking put to the lower writer.
-		rv = nni_msgq_put_until(ppipe->sendq, msg, NNI_TIME_ZERO);
-		if (rv != 0) {
-			// message queue is full, we have no choice but
-			// to drop it.  This should not happen under normal
-			// circumstances.
-			nni_msg_free(msg);
-		}
-		nni_mtx_unlock(mx);
+	nni_sock_lock(psock->nsock);
+	if (nni_idhash_find(&psock->pipes, id, (void **) &ppipe) != 0) {
+		nni_sock_unlock(psock->nsock);
+		nni_msg_free(msg);
+		nni_msgq_aio_get(psock->uwq, &psock->aio_getq);
+		return;
+	}
+
+	// Non-blocking put.
+	if (nni_msgq_tryput(ppipe->sendq, msg) != 0) {
+		nni_msg_free(msg);
 	}
+	nni_sock_unlock(psock->nsock);
 }
 
 
-static void
-nni_resp_pipe_send(void *arg)
+void
+nni_resp_getq_cb(void *arg)
 {
 	nni_resp_pipe *ppipe = arg;
-	nni_resp_sock *psock = ppipe->psock;
-	nni_pipe *npipe = ppipe->npipe;
-	nni_msgq *sendq = ppipe->sendq;
-	nni_msg *msg;
-	int rv;
 
-	for (;;) {
-		rv = nni_msgq_get_sig(sendq, &msg, &ppipe->sigclose);
-		if (rv != 0) {
-			break;
-		}
+	if (nni_aio_result(&ppipe->aio_getq) != 0) {
+		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
+		return;
+	}
 
-		rv = nni_pipe_send(npipe, msg);
-		if (rv != 0) {
-			nni_msg_free(msg);
-			break;
-		}
+	ppipe->aio_send.a_msg = ppipe->aio_getq.a_msg;
+	ppipe->aio_getq.a_msg = NULL;
+
+	nni_pipe_aio_send(ppipe->npipe, &ppipe->aio_send);
+}
+
+
+void
+nni_resp_send_cb(void *arg)
+{
+	nni_resp_pipe *ppipe = arg;
+
+	if (nni_aio_result(&ppipe->aio_send) != 0) {
+		nni_msg_free(ppipe->aio_send.a_msg);
+		ppipe->aio_send.a_msg = NULL;
+		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
+		return;
 	}
-	nni_msgq_signal(nni_sock_recvq(psock->nsock), &ppipe->sigclose);
-	nni_pipe_close(npipe);
+
+	nni_msgq_aio_get(ppipe->sendq, &ppipe->aio_getq);
 }
 
 
 static void
-nni_resp_pipe_recv(void *arg)
+nni_resp_recv_cb(void *arg)
 {
 	nni_resp_pipe *ppipe = arg;
 	nni_resp_sock *psock = ppipe->psock;
 	nni_msgq *urq = nni_sock_recvq(psock->nsock);
-	nni_pipe *npipe = ppipe->npipe;
 	nni_msg *msg;
-	int rv;
 	uint8_t idbuf[4];
-	uint32_t id = nni_pipe_id(npipe);
+	uint32_t id;
+	int hops;
+	int rv;
 
+	if (nni_aio_result(&ppipe->aio_recv) != 0) {
+		goto error;
+	}
+
+	id = nni_pipe_id(ppipe->npipe);
 	NNI_PUT32(idbuf, id);
 
+	msg = ppipe->aio_recv.a_msg;
+	ppipe->aio_recv.a_msg = NULL;
+
+	// Store the pipe id in the header, first thing.
+	if (nni_msg_append_header(msg, idbuf, 4) != 0) {
+		nni_msg_free(msg);
+		goto error;
+	}
+
+	// Move backtrace from body to header
+	hops = 0;
 	for (;;) {
+		int end = 0;
 		uint8_t *body;
-		int hops;
 
-again:
-		rv = nni_pipe_recv(npipe, &msg);
-		if (rv != 0) {
-			break;
-		}
-		// Store the pipe id in the header, first thing.
-		rv = nni_msg_append_header(msg, idbuf, 4);
-		if (rv != 0) {
+		if (hops >= psock->ttl) {
 			nni_msg_free(msg);
-			continue;
+			goto error;
 		}
-
-		// Move backtrace from body to header
-		hops = 0;
-		for (;;) {
-			int end = 0;
-			if (hops >= psock->ttl) {
-				nni_msg_free(msg);
-				goto again;
-			}
-			if (nni_msg_len(msg) < 4) {
-				nni_msg_free(msg);
-				goto again;
-			}
-			body = nni_msg_body(msg);
-			end = (body[0] & 0x80) ? 1 : 0;
-			rv = nni_msg_append_header(msg, body, 4);
-			if (rv != 0) {
-				nni_msg_free(msg);
-				goto again;
-			}
-			nni_msg_trim(msg, 4);
-			if (end) {
-				break;
-			}
+		if (nni_msg_len(msg) < 4) {
+			nni_msg_free(msg);
+			goto error;
 		}
-
-		// Now send it up.
-		rv = nni_msgq_put_sig(urq, msg, &ppipe->sigclose);
+		body = nni_msg_body(msg);
+		end = (body[0] & 0x80) ? 1 : 0;
+		rv = nni_msg_append_header(msg, body, 4);
 		if (rv != 0) {
 			nni_msg_free(msg);
+			goto error;
+		}
+		nni_msg_trim(msg, 4);
+		if (end) {
 			break;
 		}
 	}
-	nni_msgq_signal(nni_sock_sendq(psock->nsock), &ppipe->sigclose);
-	nni_msgq_signal(ppipe->sendq, &ppipe->sigclose);
-	nni_pipe_close(npipe);
+
+	// Now send it up.
+	ppipe->aio_putq.a_msg = msg;
+	nni_msgq_aio_put(urq, &ppipe->aio_putq);
+	return;
+
+error:
+	nni_pipe_close(ppipe->npipe);
+	nni_pipe_decref(ppipe->npipe);
+}
+
+
+static void
+nni_resp_putq_cb(void *arg)
+{
+	nni_resp_pipe *ppipe = arg;
+
+	if (nni_aio_result(&ppipe->aio_putq) != 0) {
+		nni_msg_free(ppipe->aio_putq.a_msg);
+		ppipe->aio_putq.a_msg = NULL;
+		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
+	}
+
+	nni_pipe_aio_recv(ppipe->npipe, &ppipe->aio_recv);
 }
 
 
@@ -397,19 +490,17 @@ static nni_proto_pipe_ops nni_resp_pipe_ops = {
 	.pipe_fini	= nni_resp_pipe_fini,
 	.pipe_add	= nni_resp_pipe_add,
 	.pipe_rem	= nni_resp_pipe_rem,
-	.pipe_worker	= { nni_resp_pipe_send,
-			    nni_resp_pipe_recv },
 };
 
 static nni_proto_sock_ops nni_resp_sock_ops = {
 	.sock_init	= nni_resp_sock_init,
 	.sock_fini	= nni_resp_sock_fini,
-	.sock_close	= NULL,
+	.sock_open	= nni_resp_sock_open,
+	.sock_close	= nni_resp_sock_close,
 	.sock_setopt	= nni_resp_sock_setopt,
 	.sock_getopt	= nni_resp_sock_getopt,
 	.sock_rfilter	= nni_resp_sock_rfilter,
 	.sock_sfilter	= nni_resp_sock_sfilter,
-	.sock_worker	= { nni_resp_sock_send },
 };
 
 nni_proto nni_respondent_proto = {
diff --git a/src/protocol/survey/survey.c b/src/protocol/survey/survey.c
index c16bad3b..323a262b 100644
--- a/src/protocol/survey/survey.c
+++ b/src/protocol/survey/survey.c
@@ -18,10 +18,16 @@
 typedef struct nni_surv_pipe	nni_surv_pipe;
 typedef struct nni_surv_sock	nni_surv_sock;
 
+static void nni_surv_sock_getq_cb(void *);
+static void nni_surv_getq_cb(void *);
+static void nni_surv_putq_cb(void *);
+static void nni_surv_send_cb(void *);
+static void nni_surv_recv_cb(void *);
+static void nni_surv_timeout(void *);
+
 // An nni_surv_sock is our per-socket protocol private structure.
 struct nni_surv_sock {
 	nni_sock *	nsock;
-	nni_cv		cv;
 	nni_duration	survtime;
 	nni_time	expire;
 	int		raw;
@@ -29,6 +35,10 @@ struct nni_surv_sock {
 	uint32_t	nextid;         // next id
 	uint8_t		survid[4];      // outstanding request ID (big endian)
 	nni_list	pipes;
+	nni_aio		aio_getq;
+	nni_timer_node	timer;
+	nni_msgq *	uwq;
+	nni_msgq *	urq;
 };
 
 // An nni_surv_pipe is our per-pipe protocol private structure.
@@ -37,7 +47,10 @@ struct nni_surv_pipe {
 	nni_surv_sock * psock;
 	nni_msgq *	sendq;
 	nni_list_node	node;
-	int		sigclose;
+	nni_aio		aio_getq;
+	nni_aio		aio_putq;
+	nni_aio		aio_send;
+	nni_aio		aio_recv;
 };
 
 static int
@@ -49,16 +62,21 @@ nni_surv_sock_init(void **sp, nni_sock *nsock)
 	if ((psock = NNI_ALLOC_STRUCT(psock)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	if ((rv = nni_cv_init(&psock->cv, nni_sock_mtx(nsock))) != 0) {
+	rv = nni_aio_init(&psock->aio_getq, nni_surv_sock_getq_cb, psock);
+	if (rv != 0) {
 		NNI_FREE_STRUCT(psock);
 		return (rv);
 	}
 	NNI_LIST_INIT(&psock->pipes, nni_surv_pipe, node);
+	nni_timer_init(&psock->timer, nni_surv_timeout, psock);
+
 	psock->nextid = nni_random();
 	psock->nsock = nsock;
 	psock->raw = 0;
 	psock->survtime = NNI_SECOND * 60;
 	psock->expire = NNI_TIME_ZERO;
+	psock->uwq = nni_sock_sendq(nsock);
+	psock->urq = nni_sock_recvq(nsock);
 
 	*sp = psock;
 	nni_sock_recverr(nsock, NNG_ESTATE);
@@ -67,13 +85,21 @@ nni_surv_sock_init(void **sp, nni_sock *nsock)
 
 
 static void
+nni_surv_sock_open(void *arg)
+{
+	nni_surv_sock *psock = arg;
+
+	nni_msgq_aio_get(psock->uwq, &psock->aio_getq);
+}
+
+
+static void
 nni_surv_sock_close(void *arg)
 {
 	nni_surv_sock *psock = arg;
 
-	// Shut down the resender.
-	psock->closing = 1;
-	nni_cv_wake(&psock->cv);
+	nni_timer_cancel(&psock->timer);
+	nni_msgq_aio_cancel(psock->uwq, &psock->aio_getq);
 }
 
 
@@ -82,10 +108,21 @@ nni_surv_sock_fini(void *arg)
 {
 	nni_surv_sock *psock = arg;
 
-	if (psock != NULL) {
-		nni_cv_fini(&psock->cv);
-		NNI_FREE_STRUCT(psock);
-	}
+	NNI_FREE_STRUCT(psock);
+}
+
+
+static void
+nni_surv_pipe_fini(void *arg)
+{
+	nni_surv_pipe *ppipe = arg;
+
+	nni_aio_fini(&ppipe->aio_getq);
+	nni_aio_fini(&ppipe->aio_send);
+	nni_aio_fini(&ppipe->aio_recv);
+	nni_aio_fini(&ppipe->aio_putq);
+	nni_msgq_fini(ppipe->sendq);
+	NNI_FREE_STRUCT(ppipe);
 }
 
 
@@ -100,26 +137,32 @@ nni_surv_pipe_init(void **pp, nni_pipe *npipe, void *psock)
 	}
 	// This depth could be tunable.
 	if ((rv = nni_msgq_init(&ppipe->sendq, 16)) != 0) {
-		NNI_FREE_STRUCT(ppipe);
-		return (rv);
+		goto failed;
+	}
+	rv = nni_aio_init(&ppipe->aio_getq, nni_surv_getq_cb, ppipe);
+	if (rv != 0) {
+		goto failed;
+	}
+	rv = nni_aio_init(&ppipe->aio_putq, nni_surv_putq_cb, ppipe);
+	if (rv != 0) {
+		goto failed;
+	}
+	rv = nni_aio_init(&ppipe->aio_send, nni_surv_send_cb, ppipe);
+	if (rv != 0) {
+		goto failed;
+	}
+	rv = nni_aio_init(&ppipe->aio_recv, nni_surv_recv_cb, ppipe);
+	if (rv != 0) {
+		goto failed;
 	}
 	ppipe->npipe = npipe;
 	ppipe->psock = psock;
-	ppipe->sigclose = 0;
 	*pp = ppipe;
 	return (0);
-}
-
 
-static void
-nni_surv_pipe_fini(void *arg)
-{
-	nni_surv_pipe *ppipe = arg;
-
-	if (ppipe != NULL) {
-		nni_msgq_fini(ppipe->sendq);
-		NNI_FREE_STRUCT(ppipe);
-	}
+failed:
+	nni_surv_pipe_fini(ppipe);
+	return (rv);
 }
 
 
@@ -130,6 +173,12 @@ nni_surv_pipe_add(void *arg)
 	nni_surv_sock *psock = ppipe->psock;
 
 	nni_list_append(&psock->pipes, ppipe);
+
+	nni_pipe_incref(ppipe->npipe);
+	nni_msgq_aio_get(ppipe->sendq, &ppipe->aio_getq);
+
+	nni_pipe_incref(ppipe->npipe);
+	nni_pipe_aio_recv(ppipe->npipe, &ppipe->aio_recv);
 	return (0);
 }
 
@@ -141,75 +190,100 @@ nni_surv_pipe_rem(void *arg)
 	nni_surv_sock *psock = ppipe->psock;
 
 	nni_list_remove(&psock->pipes, ppipe);
+	nni_msgq_close(ppipe->sendq);
+	nni_msgq_aio_cancel(psock->urq, &ppipe->aio_putq);
 }
 
 
 static void
-nni_surv_pipe_sender(void *arg)
+nni_surv_getq_cb(void *arg)
 {
 	nni_surv_pipe *ppipe = arg;
-	nni_pipe *npipe = ppipe->npipe;
-	nni_msgq *uwq = ppipe->sendq;
-	nni_msgq *urq = nni_sock_recvq(ppipe->psock->nsock);
-	nni_msg *msg;
-	int rv;
 
-	for (;;) {
-		rv = nni_msgq_get_sig(uwq, &msg, &ppipe->sigclose);
-		if (rv != 0) {
-			break;
-		}
-		rv = nni_pipe_send(npipe, msg);
-		if (rv != 0) {
-			nni_msg_free(msg);
-			break;
-		}
+	if (nni_aio_result(&ppipe->aio_getq) != 0) {
+		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
+		return;
 	}
-	nni_msgq_signal(urq, &ppipe->sigclose);
-	nni_pipe_close(npipe);
+
+	ppipe->aio_send.a_msg = ppipe->aio_getq.a_msg;
+	ppipe->aio_getq.a_msg = NULL;
+
+	nni_pipe_aio_send(ppipe->npipe, &ppipe->aio_send);
 }
 
 
 static void
-nni_surv_pipe_receiver(void *arg)
+nni_surv_send_cb(void *arg)
+{
+	nni_surv_pipe *ppipe = arg;
+
+	if (nni_aio_result(&ppipe->aio_send) != 0) {
+		nni_msg_free(ppipe->aio_send.a_msg);
+		ppipe->aio_send.a_msg = NULL;
+		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
+		return;
+	}
+
+	nni_msgq_aio_get(ppipe->psock->uwq, &ppipe->aio_getq);
+}
+
+
+static void
+nni_surv_putq_cb(void *arg)
+{
+	nni_surv_pipe *ppipe = arg;
+
+	if (nni_aio_result(&ppipe->aio_putq) != 0) {
+		nni_msg_free(ppipe->aio_putq.a_msg);
+		ppipe->aio_putq.a_msg = NULL;
+		nni_pipe_close(ppipe->npipe);
+		nni_pipe_decref(ppipe->npipe);
+		return;
+	}
+
+	nni_pipe_aio_recv(ppipe->npipe, &ppipe->aio_recv);
+}
+
+
+static void
+nni_surv_recv_cb(void *arg)
 {
 	nni_surv_pipe *ppipe = arg;
-	nni_surv_sock *psock = ppipe->psock;
-	nni_msgq *urq = nni_sock_recvq(psock->nsock);
-	nni_msgq *uwq = nni_sock_sendq(psock->nsock);
-	nni_pipe *npipe = ppipe->npipe;
 	nni_msg *msg;
 	int rv;
 
-	for (;;) {
-		rv = nni_pipe_recv(npipe, &msg);
-		if (rv != 0) {
-			break;
-		}
-		// We yank 4 bytes of body, and move them to the header.
-		if (nni_msg_len(msg) < 4) {
-			// Not enough data, just toss it.
-			nni_msg_free(msg);
-			continue;
-		}
-		if (nni_msg_append_header(msg, nni_msg_body(msg), 4) != 0) {
-			// Should be NNG_ENOMEM
-			nni_msg_free(msg);
-			continue;
-		}
-		if (nni_msg_trim(msg, 4) != 0) {
-			// This should never happen - could be an assert.
-			nni_panic("Failed to trim SURV header from body");
-		}
-		rv = nni_msgq_put_sig(urq, msg, &ppipe->sigclose);
-		if (rv != 0) {
-			nni_msg_free(msg);
-			break;
-		}
+	if (nni_aio_result(&ppipe->aio_recv) != 0) {
+		goto failed;
 	}
-	nni_msgq_signal(uwq, &ppipe->sigclose);
-	nni_msgq_signal(ppipe->sendq, &ppipe->sigclose);
-	nni_pipe_close(npipe);
+
+	msg = ppipe->aio_recv.a_msg;
+	ppipe->aio_recv.a_msg = NULL;
+
+	// We yank 4 bytes of body, and move them to the header.
+	if (nni_msg_len(msg) < 4) {
+		// Not enough data, just toss it.
+		nni_msg_free(msg);
+		goto failed;
+	}
+	if (nni_msg_append_header(msg, nni_msg_body(msg), 4) != 0) {
+		// Should be NNG_ENOMEM
+		nni_msg_free(msg);
+		goto failed;
+	}
+	if (nni_msg_trim(msg, 4) != 0) {
+		// This should never happen - could be an assert.
+		nni_msg_free(msg);
+		goto failed;
+	}
+	ppipe->aio_putq.a_msg = msg;
+	nni_msgq_aio_put(ppipe->psock->urq, &ppipe->aio_putq);
+	return;
+
+failed:
+	nni_pipe_close(ppipe->npipe);
+	nni_pipe_decref(ppipe->npipe);
 }
 
 
@@ -234,8 +308,7 @@ nni_surv_sock_setopt(void *arg, int opt, const void *buf, size_t sz)
 				nni_sock_recverr(psock->nsock, NNG_ESTATE);
 			}
 			memset(psock->survid, 0, sizeof (psock->survid));
-			psock->expire = NNI_TIME_NEVER;
-			nni_cv_wake(&psock->cv);
+			nni_timer_cancel(&psock->timer);
 		}
 		break;
 	default:
@@ -266,70 +339,53 @@ nni_surv_sock_getopt(void *arg, int opt, void *buf, size_t *szp)
 
 
 static void
-nni_surv_sock_sender(void *arg)
+nni_surv_sock_getq_cb(void *arg)
 {
 	nni_surv_sock *psock = arg;
-	nni_msgq *uwq = nni_sock_sendq(psock->nsock);
-	nni_mtx *mx = nni_sock_mtx(psock->nsock);
+	nni_surv_pipe *ppipe;
+	nni_surv_pipe *last;
 	nni_msg *msg, *dup;
 
-	for (;;) {
-		nni_surv_pipe *ppipe;
-		nni_surv_pipe *last;
-		int rv;
-
-		if ((rv = nni_msgq_get(uwq, &msg)) != 0) {
-			break;
-		}
-
-		nni_mtx_lock(mx);
-		last = nni_list_last(&psock->pipes);
-		NNI_LIST_FOREACH (&psock->pipes, ppipe) {
-			if (ppipe != last) {
-				rv = nni_msg_dup(&dup, msg);
-				if (rv != 0) {
-					continue;
-				}
-			} else {
-				dup = msg;
-			}
-			if ((rv = nni_msgq_tryput(ppipe->sendq, dup)) != 0) {
-				nni_msg_free(dup);
+	if (nni_aio_result(&psock->aio_getq) != 0) {
+		// Should be NNG_ECLOSED.
+		return;
+	}
+	msg = psock->aio_getq.a_msg;
+	psock->aio_getq.a_msg = NULL;
+
+	nni_sock_lock(psock->nsock);
+	last = nni_list_last(&psock->pipes);
+	NNI_LIST_FOREACH (&psock->pipes, ppipe) {
+		if (ppipe != last) {
+			if (nni_msg_dup(&dup, msg) != 0) {
+				continue;
 			}
+		} else {
+			dup = msg;
 		}
-		nni_mtx_unlock(mx);
-
-		if (last == NULL) {
-			nni_msg_free(msg);
+		if (nni_msgq_tryput(ppipe->sendq, dup) != 0) {
+			nni_msg_free(dup);
 		}
 	}
+	nni_sock_unlock(psock->nsock);
+
+	if (last == NULL) {
+		// If there were no pipes to send on, just toss the message.
+		nni_msg_free(msg);
+	}
 }
 
 
 static void
-nni_surv_sock_timeout(void *arg)
+nni_surv_timeout(void *arg)
 {
 	nni_surv_sock *psock = arg;
-	nni_mtx *mx = nni_sock_mtx(psock->nsock);
-	nni_msgq *urq = nni_sock_recvq(psock->nsock);
-
-	nni_mtx_lock(mx);
-	for (;;) {
-		if (psock->closing) {
-			nni_mtx_unlock(mx);
-			return;
-		}
-		if (nni_clock() > psock->expire) {
-			// Set the expiration ~forever
-			psock->expire = NNI_TIME_NEVER;
-			// Survey IDs *always* have the high order bit set,
-			// so zeroing means that nothing can match.
-			memset(psock->survid, 0, sizeof (psock->survid));
-			nni_sock_recverr(psock->nsock, NNG_ESTATE);
-			nni_msgq_set_get_error(urq, NNG_ETIMEDOUT);
-		}
-		nni_cv_until(&psock->cv, psock->expire);
-	}
+
+	nni_sock_lock(psock->nsock);
+	memset(psock->survid, 0, sizeof (psock->survid));
+	nni_sock_recverr(psock->nsock, NNG_ESTATE);
+	nni_msgq_set_get_error(psock->urq, NNG_ETIMEDOUT);
+	nni_sock_unlock(psock->nsock);
 }
 
 
@@ -363,11 +419,11 @@ nni_surv_sock_sfilter(void *arg, nni_msg *msg)
 	// survey expiration out.  The timeout thread will wake up in
 	// the wake below, and reschedule itself appropriately.
 	psock->expire = nni_clock() + psock->survtime;
-	nni_cv_wake(&psock->cv);
+	nni_timer_schedule(&psock->timer, psock->expire);
 
 	// Clear the error condition.
 	nni_sock_recverr(psock->nsock, 0);
-	nni_msgq_set_get_error(nni_sock_recvq(psock->nsock), 0);
+	//nni_msgq_set_get_error(nni_sock_recvq(psock->nsock), 0);
 
 	return (msg);
 }
@@ -405,20 +461,17 @@ static nni_proto_pipe_ops nni_surv_pipe_ops = {
 	.pipe_fini	= nni_surv_pipe_fini,
 	.pipe_add	= nni_surv_pipe_add,
 	.pipe_rem	= nni_surv_pipe_rem,
-	.pipe_worker	= { nni_surv_pipe_sender,
-			    nni_surv_pipe_receiver }
 };
 
 static nni_proto_sock_ops nni_surv_sock_ops = {
 	.sock_init	= nni_surv_sock_init,
 	.sock_fini	= nni_surv_sock_fini,
+	.sock_open	= nni_surv_sock_open,
 	.sock_close	= nni_surv_sock_close,
 	.sock_setopt	= nni_surv_sock_setopt,
 	.sock_getopt	= nni_surv_sock_getopt,
 	.sock_rfilter	= nni_surv_sock_rfilter,
 	.sock_sfilter	= nni_surv_sock_sfilter,
-	.sock_worker	= { nni_surv_sock_sender,
-			    nni_surv_sock_timeout }
 };
 
 // This is the global protocol structure -- our linkage to the core.