fixes #346 nng_recv() sometimes acts on null `msg` pointer

This closes a fundamental flaw in the way aio structures were
handled.  In paticular, aio expiration could race ahead, and
fire before the aio was properly registered by the provider.
This ultimately led to the possibility of duplicate completions
on the same aio.

The solution involved breaking up nni_aio_start into two functions.
nni_aio_begin (which can be run outside of external locks) simply
validates that nni_aio_fini() has not been called, and clears certain
fields in the aio to make it ready for use by the provider.

nni_aio_schedule does the work to register the aio with the expiration
thread, and should only be called when the aio is actually scheduled
for asynchronous completion.  nni_aio_schedule_verify does the same thing,
but returns NNG_ETIMEDOUT if the aio has a zero length timeout.

This change has a small negative performance impact.  We have plans to
rectify that by converting nni_aio_begin to use a locklesss flag for
the aio->a_fini bit.

While we were here, we fixed some error paths in the POSIX subsystem,
which would have returned incorrect error codes, and we made some
optmizations in the message queues to reduce conditionals while holding
locks in the hot code path.

4 files changed, 23 insertions, 18 deletions

diff --git a/src/supplemental/http/http_client.c b/src/supplemental/http/http_client.c
index c62a3c56..2a9be4bb 100644
--- a/src/supplemental/http/http_client.c
+++ b/src/supplemental/http/http_client.c
@@ -241,10 +241,11 @@ http_connect_cancel(nni_aio *aio, int rv)
 void
 nni_http_client_connect(nni_http_client *c, nni_aio *aio)
 {
-	if (nni_aio_start(aio, http_connect_cancel, aio) != 0) {
+	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
 	nni_mtx_lock(&c->mtx);
+	nni_aio_schedule(aio, http_connect_cancel, aio);
 	nni_list_append(&c->aios, aio);
 	if (nni_list_first(&c->aios) == aio) {
 		http_conn_start(c);
diff --git a/src/supplemental/http/http_conn.c b/src/supplemental/http/http_conn.c
index d37c9a2f..a3039d0d 100644
--- a/src/supplemental/http/http_conn.c
+++ b/src/supplemental/http/http_conn.c
@@ -360,13 +360,14 @@ http_rd_cancel(nni_aio *aio, int rv)
 static void
 http_rd_submit(nni_http_conn *conn, nni_aio *aio)
 {
-	if (nni_aio_start(aio, http_rd_cancel, conn) != 0) {
+	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
 	if (conn->closed) {
 		nni_aio_finish_error(aio, NNG_ECLOSED);
 		return;
 	}
+	nni_aio_schedule(aio, http_rd_cancel, conn);
 	nni_list_append(&conn->rdq, aio);
 	if (conn->rd_uaio == NULL) {
 		http_rd_start(conn);
@@ -473,7 +474,7 @@ http_wr_cancel(nni_aio *aio, int rv)
 static void
 http_wr_submit(nni_http_conn *conn, nni_aio *aio)
 {
-	if (nni_aio_start(aio, http_wr_cancel, conn) != 0) {
+	if (nni_aio_begin(aio) != 0) {
 		return;
 	}
 	if (conn->closed) {
@@ -481,6 +482,7 @@ http_wr_submit(nni_http_conn *conn, nni_aio *aio)
 		return;
 	}
 	nni_list_append(&conn->wrq, aio);
+	nni_aio_schedule(aio, http_wr_cancel, conn);
 	if (conn->wr_uaio == NULL) {
 		http_wr_start(conn);
 	}
diff --git a/src/supplemental/http/http_public.c b/src/supplemental/http/http_public.c
index 7107c029..7aae45b2 100644
--- a/src/supplemental/http/http_public.c
+++ b/src/supplemental/http/http_public.c
@@ -9,9 +9,9 @@
 //
 
 #include "core/nng_impl.h"
-#include "supplemental/tls/tls.h"
 #include "http.h"
 #include "http_api.h"
+#include "supplemental/tls/tls.h"
 
 // Symbols in this file are "public" versions of the HTTP API.
 // These are suitable for exposure to applications.
@@ -382,7 +382,7 @@ nng_http_conn_read(nng_http_conn *conn, nng_aio *aio)
 	nni_http_read(conn, aio);
 #else
 	NNI_ARG_UNUSED(conn);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
@@ -395,7 +395,7 @@ nng_http_conn_read_all(nng_http_conn *conn, nng_aio *aio)
 	nni_http_read_full(conn, aio);
 #else
 	NNI_ARG_UNUSED(conn);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
@@ -408,7 +408,7 @@ nng_http_conn_write(nng_http_conn *conn, nng_aio *aio)
 	nni_http_write(conn, aio);
 #else
 	NNI_ARG_UNUSED(conn);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
@@ -433,7 +433,7 @@ nng_http_conn_write_req(nng_http_conn *conn, nng_http_req *req, nng_aio *aio)
 #else
 	NNI_ARG_UNUSED(conn);
 	NNI_ARG_UNUSED(req);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
@@ -447,7 +447,7 @@ nng_http_conn_write_res(nng_http_conn *conn, nng_http_res *res, nng_aio *aio)
 #else
 	NNI_ARG_UNUSED(conn);
 	NNI_ARG_UNUSED(res);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
@@ -461,7 +461,7 @@ nng_http_conn_read_req(nng_http_conn *conn, nng_http_req *req, nng_aio *aio)
 #else
 	NNI_ARG_UNUSED(conn);
 	NNI_ARG_UNUSED(req);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
@@ -475,7 +475,7 @@ nng_http_conn_read_res(nng_http_conn *conn, nng_http_res *res, nng_aio *aio)
 #else
 	NNI_ARG_UNUSED(conn);
 	NNI_ARG_UNUSED(res);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
@@ -574,7 +574,8 @@ nng_http_handler_set_host(nng_http_handler *h, const char *host)
 #endif
 }
 
-int nng_http_handler_set_tree(nng_http_handler *h)
+int
+nng_http_handler_set_tree(nng_http_handler *h)
 {
 #ifdef NNG_SUPP_HTTP
 	return (nni_http_handler_set_tree(h));
@@ -698,12 +699,13 @@ nng_http_server_get_tls(nng_http_server *srv, nng_tls_config **cfgp)
 #endif
 }
 
-int nng_http_hijack(nng_http_conn * conn)
+int
+nng_http_hijack(nng_http_conn *conn)
 {
 #ifdef NNG_SUPP_HTTP
-    return (nni_http_hijack(conn));
+	return (nni_http_hijack(conn));
 #else
-    NNI_ARG_UNUSED(conn);
+	NNI_ARG_UNUSED(conn);
 	return (NNG_ENOTSUP);
 #endif
 }
@@ -762,7 +764,7 @@ nng_http_client_connect(nng_http_client *cli, nng_aio *aio)
 	nni_http_client_connect(cli, aio);
 #else
 	NNI_ARG_UNUSED(cli);
-	if (nni_aio_start(aio, NULL, NULL)) {
+	if (nni_aio_begin(aio) == 0) {
 		nni_aio_finish_error(aio, NNG_ENOTSUP);
 	}
 #endif
diff --git a/src/supplemental/http/http_server.c b/src/supplemental/http/http_server.c
index 7feadc96..c92de586 100644
--- a/src/supplemental/http/http_server.c
+++ b/src/supplemental/http/http_server.c
@@ -590,8 +590,8 @@ http_sconn_rxdone(void *arg)
 
 	// Technically, probably callback should initialize this with
 	// start, but we do it instead.
-
-	if (nni_aio_start(sc->cbaio, NULL, NULL) != 0) {
+	// This operation cannot at present canceled or timed out.
+	if (nni_aio_begin(sc->cbaio) != 0) {
 		nni_mtx_unlock(&s->mtx);
 		return;
 	}