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.

6 files changed, 39 insertions, 26 deletions

diff --git a/src/core/aio.c b/src/core/aio.c
index 54ba7000..f039cdc8 100644
--- a/src/core/aio.c
+++ b/src/core/aio.c
@@ -60,11 +60,10 @@ static int                nni_aio_expire_q_cnt;
 // when we want to know if the operation itself is complete.
 //
 // In order to guard against aio reuse during teardown, we set the a_stop
-// flag.  Any attempt to initialize for a new operation after that point
-// will fail and the caller will get NNG_ECANCELED indicating this.  The
-// provider that calls nni_aio_begin() MUST check the return value, and
-// if it comes back nonzero (NNG_ECANCELED) then it must simply discard the
-// request and return.
+// flag.  Any attempt to submit new operation after that point will fail with
+// the status NNG_ESTOPPED indicating this.  The provider that calls
+// nni_aio_begin() MUST check the return value, and if it comes back
+// NNG_ESTOPPED then it must simply discard the request and // return.
 //
 // Calling nni_aio_wait waits for the current outstanding operation to
 // complete, but does not block another one from being started on the
@@ -118,7 +117,7 @@ nni_aio_fini(nni_aio *aio)
 		nni_mtx_unlock(&eq->eq_mtx);
 
 		if (fn != NULL) {
-			fn(aio, arg, NNG_ECLOSED);
+			fn(aio, arg, NNG_ESTOPPED);
 		}
 
 		nni_task_fini(&aio->a_task);
@@ -201,7 +200,7 @@ nni_aio_stop(nni_aio *aio)
 		nni_mtx_unlock(&eq->eq_mtx);
 
 		if (fn != NULL) {
-			fn(aio, arg, NNG_ECANCELED);
+			fn(aio, arg, NNG_ESTOPPED);
 		}
 
 		nni_aio_wait(aio);
@@ -226,7 +225,7 @@ nni_aio_close(nni_aio *aio)
 		nni_mtx_unlock(&eq->eq_mtx);
 
 		if (fn != NULL) {
-			fn(aio, arg, NNG_ECLOSED);
+			fn(aio, arg, NNG_ESTOPPED);
 		}
 	}
 }
@@ -353,12 +352,13 @@ nni_aio_begin(nni_aio *aio)
 
 	// We should not reschedule anything at this point.
 	if (aio->a_stop || eq->eq_stop) {
-		aio->a_result    = NNG_ECANCELED;
+		aio->a_result    = NNG_ESTOPPED;
 		aio->a_cancel_fn = NULL;
 		aio->a_expire    = NNI_TIME_NEVER;
+		aio->a_stop      = true;
 		nni_mtx_unlock(&eq->eq_mtx);
 
-		return (NNG_ECANCELED);
+		return (NNG_ESTOPPED);
 	}
 	nni_task_prep(&aio->a_task);
 	nni_mtx_unlock(&eq->eq_mtx);
@@ -386,15 +386,17 @@ nni_aio_schedule(nni_aio *aio, nni_aio_cancel_fn cancel, void *data)
 	}
 
 	nni_mtx_lock(&eq->eq_mtx);
+	if (aio->a_stop || eq->eq_stop) {
+		aio->a_stop = true;
+		nni_mtx_unlock(&eq->eq_mtx);
+		return (NNG_ESTOPPED);
+	}
 	if (aio->a_abort) {
 		int rv = aio->a_result;
 		nni_mtx_unlock(&eq->eq_mtx);
+		NNI_ASSERT(rv != 0);
 		return (rv);
 	}
-	if (aio->a_stop || eq->eq_stop) {
-		nni_mtx_unlock(&eq->eq_mtx);
-		return (NNG_ECLOSED);
-	}
 
 	NNI_ASSERT(aio->a_cancel_fn == NULL);
 	aio->a_cancel_fn  = cancel;
@@ -434,22 +436,25 @@ nni_aio_defer(nni_aio *aio, nni_aio_cancel_fn cancel, void *data)
 	}
 
 	nni_mtx_lock(&eq->eq_mtx);
-	if (timeout) {
+	if (aio->a_stop || eq->eq_stop) {
+		aio->a_stop   = true;
 		aio->a_sleep  = false;
-		aio->a_result = aio->a_expire_ok ? 0 : NNG_ETIMEDOUT;
+		aio->a_result = NNG_ESTOPPED;
 		nni_mtx_unlock(&eq->eq_mtx);
 		nni_task_dispatch(&aio->a_task);
 		return (false);
 	}
 	if (aio->a_abort) {
 		aio->a_sleep = false;
+		aio->a_abort = false;
 		nni_mtx_unlock(&eq->eq_mtx);
 		nni_task_dispatch(&aio->a_task);
 		return (false);
 	}
-	if (aio->a_stop || eq->eq_stop) {
+	if (timeout) {
 		aio->a_sleep  = false;
-		aio->a_result = NNG_ECLOSED;
+		aio->a_result = aio->a_expire_ok ? 0 : NNG_ETIMEDOUT;
+		aio->a_abort  = false;
 		nni_mtx_unlock(&eq->eq_mtx);
 		nni_task_dispatch(&aio->a_task);
 		return (false);
@@ -499,22 +504,26 @@ nni_aio_start(nni_aio *aio, nni_aio_cancel_fn cancel, void *data)
 	nni_task_prep(&aio->a_task);
 
 	nni_mtx_lock(&eq->eq_mtx);
-	if (timeout) {
+	if (aio->a_stop || eq->eq_stop) {
+		aio->a_stop   = true;
 		aio->a_sleep  = false;
-		aio->a_result = aio->a_expire_ok ? 0 : NNG_ETIMEDOUT;
+		aio->a_result = NNG_ESTOPPED;
 		nni_mtx_unlock(&eq->eq_mtx);
 		nni_task_dispatch(&aio->a_task);
 		return (false);
 	}
 	if (aio->a_abort) {
 		aio->a_sleep = false;
+		aio->a_abort = false;
+		NNI_ASSERT(aio->a_result != 0);
 		nni_mtx_unlock(&eq->eq_mtx);
 		nni_task_dispatch(&aio->a_task);
 		return (false);
 	}
-	if (aio->a_stop || eq->eq_stop) {
+	if (timeout) {
 		aio->a_sleep  = false;
-		aio->a_result = NNG_ECLOSED;
+		aio->a_result = aio->a_expire_ok ? 0 : NNG_ETIMEDOUT;
+		aio->a_abort  = false;
 		nni_mtx_unlock(&eq->eq_mtx);
 		nni_task_dispatch(&aio->a_task);
 		return (false);
@@ -767,7 +776,8 @@ nni_aio_expire_loop(void *arg)
 		for (uint32_t i = 0; i < exp_idx; i++) {
 			aio = expires[i];
 			if (q->eq_stop) {
-				rv = NNG_ECANCELED;
+				rv          = NNG_ESTOPPED;
+				aio->a_stop = true;
 			} else if (aio->a_expire_ok) {
 				aio->a_expire_ok = false;
 				rv               = 0;
diff --git a/src/core/aio_test.c b/src/core/aio_test.c
index f18eb5ed..66682460 100644
--- a/src/core/aio_test.c
+++ b/src/core/aio_test.c
@@ -76,8 +76,8 @@ static void
 sleep_reap(void *arg)
 {
 	nng_aio *aio = *(nng_aio **) arg;
-	if (nng_aio_result(aio) != NNG_ECANCELED) {
-		NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+	if (nng_aio_result(aio) != NNG_ESTOPPED) {
+		NUTS_FAIL(nng_aio_result(aio), NNG_ESTOPPED);
 	}
 	nng_aio_reap(aio);
 }
diff --git a/src/core/dialer.c b/src/core/dialer.c
index d684fd47..5bf9915c 100644
--- a/src/core/dialer.c
+++ b/src/core/dialer.c
@@ -421,6 +421,7 @@ dialer_connect_cb(void *arg)
 		break;
 	case NNG_ECLOSED:   // No further action.
 	case NNG_ECANCELED: // No further action.
+	case NNG_ESTOPPED:  // No further action.
 		nni_dialer_bump_error(d, rv);
 		break;
 	case NNG_ECONNREFUSED:
diff --git a/src/core/errors_test.c b/src/core/errors_test.c
index 72a107f4..b02bd36b 100644
--- a/src/core/errors_test.c
+++ b/src/core/errors_test.c
@@ -18,6 +18,7 @@ test_known_errors(void)
 	NUTS_MATCH(nng_strerror(0), "Hunky dory");
 	NUTS_MATCH(nng_strerror(NNG_ECLOSED), "Object closed");
 	NUTS_MATCH(nng_strerror(NNG_ETIMEDOUT), "Timed out");
+	NUTS_MATCH(nng_strerror(NNG_ESTOPPED), "Operation stopped");
 }
 
 static void
diff --git a/src/core/listener.c b/src/core/listener.c
index c212ce91..0ba745c9 100644
--- a/src/core/listener.c
+++ b/src/core/listener.c
@@ -417,6 +417,7 @@ listener_accept_cb(void *arg)
 		nni_listener_bump_error(l, rv);
 		listener_accept_start(l);
 		break;
+	case NNG_ESTOPPED:  // no further action
 	case NNG_ECLOSED:   // no further action
 	case NNG_ECANCELED: // no further action
 		nni_listener_bump_error(l, rv);
diff --git a/src/core/protocol.h b/src/core/protocol.h
index 0d4d12dc..5231c8f4 100644
--- a/src/core/protocol.h
+++ b/src/core/protocol.h
@@ -46,7 +46,7 @@ struct nni_proto_pipe_ops {
 	// pipe_close is an idempotent, non-blocking, operation, called
 	// when the pipe is being closed.  Any operations pending on the
 	// pipe should be canceled with NNG_ECLOSED.  (Best option is to
-	// use nng_aio_close() on them)
+	// use nni_aio_close() on them)
 	void (*pipe_close)(void *);
 
 	// pipe_stop is called during finalization, to ensure that