aio: introduce NNG_ESTOPPED

This error code results when an AIO is stopped permanently, as a result
of nni_aio_close or nni_aio_stop.  The associated AIO object cannot be
used again.  This discrimantes against a file being closed, or a temporary
cancellation which might allow the aio to be reused.

Consumers must check for this error status in their callbacks, and not
resubmit an operation that failed with this error.  Doing so, will result
in an infinite loop of submit / errors.

4 files changed, 63 insertions, 1 deletions

diff --git a/src/sp/protocol/reqrep0/rep_test.c b/src/sp/protocol/reqrep0/rep_test.c
index 579f795c..2a07ecbc 100644
--- a/src/sp/protocol/reqrep0/rep_test.c
+++ b/src/sp/protocol/reqrep0/rep_test.c
@@ -369,6 +369,26 @@ test_rep_ctx_recv_aio_stopped(void)
 	nng_aio_stop(aio);
 	nng_ctx_recv(ctx, aio);
 	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ESTOPPED);
+	NUTS_PASS(nng_ctx_close(ctx));
+	NUTS_CLOSE(rep);
+	nng_aio_free(aio);
+}
+
+void
+test_rep_ctx_recv_aio_canceled(void)
+{
+	nng_socket rep;
+	nng_ctx    ctx;
+	nng_aio   *aio;
+
+	NUTS_PASS(nng_rep0_open(&rep));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+	NUTS_PASS(nng_ctx_open(&ctx, rep));
+
+	nng_ctx_recv(ctx, aio);
+	nng_aio_cancel(aio);
+	nng_aio_wait(aio);
 	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
 	NUTS_PASS(nng_ctx_close(ctx));
 	NUTS_CLOSE(rep);
@@ -755,6 +775,7 @@ NUTS_TESTS = {
 	{ "rep close pipe before send", test_rep_close_pipe_before_send },
 	{ "rep close pipe during send", test_rep_close_pipe_during_send },
 	{ "rep recv aio ctx stopped", test_rep_ctx_recv_aio_stopped },
+	{ "rep recv aio ctx canceled", test_rep_ctx_recv_aio_canceled },
 	{ "rep close pipe context send", test_rep_close_pipe_context_send },
 	{ "rep close context send", test_rep_close_context_send },
 	{ "rep close recv", test_rep_close_recv },
diff --git a/src/sp/protocol/reqrep0/reqstress_test.c b/src/sp/protocol/reqrep0/reqstress_test.c
index cd0004a0..fd964003 100644
--- a/src/sp/protocol/reqrep0/reqstress_test.c
+++ b/src/sp/protocol/reqrep0/reqstress_test.c
@@ -79,7 +79,8 @@ fatal(const char *msg, int rv)
 void
 error(test_case *c, const char *msg, int rv)
 {
-	if ((rv == NNG_ECLOSED) || (rv == NNG_ECANCELED)) {
+	if ((rv == NNG_ECLOSED) || (rv == NNG_ECANCELED) ||
+	    (rv == NNG_ESTOPPED)) {
 		return;
 	}
 	fprintf(
diff --git a/src/sp/protocol/reqrep0/xrep_test.c b/src/sp/protocol/reqrep0/xrep_test.c
index d5110469..068d64a7 100644
--- a/src/sp/protocol/reqrep0/xrep_test.c
+++ b/src/sp/protocol/reqrep0/xrep_test.c
@@ -262,6 +262,23 @@ test_xrep_recv_aio_stopped(void)
 	nng_aio_stop(aio);
 	nng_recv_aio(rep, aio);
 	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ESTOPPED);
+	NUTS_CLOSE(rep);
+	nng_aio_free(aio);
+}
+
+static void
+test_xrep_recv_aio_canceled(void)
+{
+	nng_socket rep;
+	nng_aio   *aio;
+
+	NUTS_PASS(nng_rep0_open_raw(&rep));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_recv_aio(rep, aio);
+	nng_aio_cancel(aio);
+	nng_aio_wait(aio);
 	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
 	NUTS_CLOSE(rep);
 	nng_aio_free(aio);
@@ -415,6 +432,7 @@ NUTS_TESTS = {
 	{ "xrep close pipe during send", test_xrep_close_pipe_during_send },
 	{ "xrep close during recv", test_xrep_close_during_recv },
 	{ "xrep recv aio stopped", test_xrep_recv_aio_stopped },
+	{ "xrep recv aio canceled", test_xrep_recv_aio_canceled },
 	{ "xrep send no header", test_xrep_send_no_header },
 	{ "xrep recv garbage", test_xrep_recv_garbage },
 	{ "xrep ttl option", test_xrep_ttl_option },
diff --git a/src/sp/protocol/reqrep0/xreq_test.c b/src/sp/protocol/reqrep0/xreq_test.c
index 28f381fe..1f06eb17 100644
--- a/src/sp/protocol/reqrep0/xreq_test.c
+++ b/src/sp/protocol/reqrep0/xreq_test.c
@@ -168,7 +168,28 @@ test_xreq_recv_aio_stopped(void)
 	nng_aio_stop(aio);
 	nng_recv_aio(req, aio);
 	nng_aio_wait(aio);
+	NUTS_FAIL(nng_aio_result(aio), NNG_ESTOPPED);
+	NUTS_CLOSE(req);
+	nng_aio_free(aio);
+}
+
+static void
+test_xreq_send_aio_canceled(void)
+{
+	nng_socket req;
+	nng_aio   *aio;
+	nng_msg   *msg;
+
+	NUTS_PASS(nng_msg_alloc(&msg, 64));
+	NUTS_PASS(nng_req0_open_raw(&req));
+	NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL));
+
+	nng_aio_set_msg(aio, msg);
+	nng_send_aio(req, aio);
+	nng_aio_cancel(aio);
+	nng_aio_wait(aio);
 	NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	nng_msg_free(msg);
 	NUTS_CLOSE(req);
 	nng_aio_free(aio);
 }
@@ -348,6 +369,7 @@ NUTS_TESTS = {
 	{ "xreq poll writable", test_xreq_poll_writeable },
 	{ "xreq validate peer", test_xreq_validate_peer },
 	{ "xreq recv aio stopped", test_xreq_recv_aio_stopped },
+	{ "xreq send aio canceled", test_xreq_send_aio_canceled },
 	{ "xreq recv garbage", test_xreq_recv_garbage },
 	{ "xreq recv header", test_xreq_recv_header },
 	{ "xreq close during recv", test_xreq_close_during_recv },