ipc: use nni_aio_start

1 files changed, 9 insertions, 28 deletions

diff --git a/src/sp/transport/ipc/ipc.c b/src/sp/transport/ipc/ipc.c
index 68f148f8..997e1712 100644
--- a/src/sp/transport/ipc/ipc.c
+++ b/src/sp/transport/ipc/ipc.c
@@ -481,19 +481,11 @@ static void
 ipc_pipe_send(void *arg, nni_aio *aio)
 {
 	ipc_pipe *p = arg;
-	int       rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		// No way to give the message back to the protocol, so
-		// we just discard it silently to prevent it from leaking.
-		nni_msg_free(nni_aio_get_msg(aio));
-		nni_aio_set_msg(aio, NULL);
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&p->mtx);
-	if ((rv = nni_aio_schedule(aio, ipc_pipe_send_cancel, p)) != 0) {
+	if (!nni_aio_start(aio, ipc_pipe_send_cancel, p)) {
 		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_list_append(&p->send_q, aio);
@@ -556,20 +548,16 @@ static void
 ipc_pipe_recv(void *arg, nni_aio *aio)
 {
 	ipc_pipe *p = arg;
-	int       rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&p->mtx);
 	if (p->closed) {
 		nni_mtx_unlock(&p->mtx);
 		nni_aio_finish_error(aio, NNG_ECLOSED);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ipc_pipe_recv_cancel, p)) != 0) {
+	if (!nni_aio_start(aio, ipc_pipe_recv_cancel, p)) {
 		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
@@ -725,6 +713,7 @@ error:
 
 	switch (rv) {
 	case NNG_ECLOSED:
+	case NNG_ESTOPPED:
 		break;
 	case NNG_ENOMEM:
 	case NNG_ENOFILES:
@@ -857,11 +846,8 @@ static void
 ipc_ep_connect(void *arg, nni_aio *aio)
 {
 	ipc_ep *ep = arg;
-	int     rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&ep->mtx);
 	if (ep->closed) {
 		nni_mtx_unlock(&ep->mtx);
@@ -874,9 +860,8 @@ ipc_ep_connect(void *arg, nni_aio *aio)
 		return;
 	}
 
-	if ((rv = nni_aio_schedule(aio, ipc_ep_cancel, ep)) != 0) {
+	if (!nni_aio_start(aio, ipc_ep_cancel, ep)) {
 		nni_mtx_unlock(&ep->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	ep->user_aio = aio;
@@ -930,11 +915,8 @@ static void
 ipc_ep_accept(void *arg, nni_aio *aio)
 {
 	ipc_ep *ep = arg;
-	int     rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&ep->mtx);
 	if (ep->closed) {
 		nni_aio_finish_error(aio, NNG_ECLOSED);
@@ -946,9 +928,8 @@ ipc_ep_accept(void *arg, nni_aio *aio)
 		nni_mtx_unlock(&ep->mtx);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ipc_ep_cancel, ep)) != 0) {
+	if (!nni_aio_start(aio, ipc_ep_cancel, ep)) {
 		nni_mtx_unlock(&ep->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	ep->user_aio = aio;