ipc: use nni_aio_start

7 files changed, 41 insertions, 83 deletions

diff --git a/src/platform/posix/posix_ipcconn.c b/src/platform/posix/posix_ipcconn.c
index ea4f5828..72fe660b 100644
--- a/src/platform/posix/posix_ipcconn.c
+++ b/src/platform/posix/posix_ipcconn.c
@@ -247,12 +247,10 @@ ipc_send(void *arg, nni_aio *aio)
 {
 	ipc_conn *c = arg;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&c->mtx);
 
-	if (!nni_aio_defer(aio, ipc_cancel, c)) {
+	if (!nni_aio_start(aio, ipc_cancel, c)) {
 		nni_mtx_unlock(&c->mtx);
 		return;
 	}
@@ -274,16 +272,12 @@ static void
 ipc_recv(void *arg, nni_aio *aio)
 {
 	ipc_conn *c = arg;
-	int       rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&c->mtx);
 
-	if ((rv = nni_aio_schedule(aio, ipc_cancel, c)) != 0) {
+	if (!nni_aio_start(aio, ipc_cancel, c)) {
 		nni_mtx_unlock(&c->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_aio_list_append(&c->readq, aio);
diff --git a/src/platform/posix/posix_ipcdial.c b/src/platform/posix/posix_ipcdial.c
index 8f54f588..2b6d547a 100644
--- a/src/platform/posix/posix_ipcdial.c
+++ b/src/platform/posix/posix_ipcdial.c
@@ -167,9 +167,7 @@ ipc_dialer_dial(void *arg, nni_aio *aio)
 	int                     fd;
 	int                     rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 
 	if (((len = nni_posix_nn2sockaddr(&ss, &d->sa)) == 0) ||
 	    (ss.ss_family != AF_UNIX)) {
@@ -182,23 +180,26 @@ ipc_dialer_dial(void *arg, nni_aio *aio)
 		return;
 	}
 
-	nni_refcnt_hold(&d->ref);
-
 	if ((rv = nni_posix_ipc_alloc(&c, &d->sa, d, fd)) != 0) {
 		(void) close(fd);
-		nni_posix_ipc_dialer_rele(d);
 		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
+	// hold for the conn
+	nni_refcnt_hold(&d->ref);
+
 	nni_mtx_lock(&d->mtx);
+	if (!nni_aio_start(aio, ipc_dialer_cancel, d)) {
+		nni_mtx_unlock(&d->mtx);
+		nng_stream_free(&c->stream);
+		return;
+	}
+
 	if (d->closed) {
 		rv = NNG_ECLOSED;
 		goto error;
 	}
-	if ((rv = nni_aio_schedule(aio, ipc_dialer_cancel, d)) != 0) {
-		goto error;
-	}
 	c->dial_aio = aio;
 	if (connect(fd, (void *) &ss, len) != 0) {
 		if (errno != EINPROGRESS && errno != EAGAIN) {
diff --git a/src/platform/posix/posix_ipclisten.c b/src/platform/posix/posix_ipclisten.c
index 7e0250f4..aa6ffda8 100644
--- a/src/platform/posix/posix_ipclisten.c
+++ b/src/platform/posix/posix_ipclisten.c
@@ -415,15 +415,12 @@ static void
 ipc_listener_accept(void *arg, nni_aio *aio)
 {
 	ipc_listener *l = arg;
-	int           rv;
 
 	// Accept is simpler than the connect case.  With accept we just
 	// need to wait for the socket to be readable to indicate an incoming
 	// connection is ready for us.  There isn't anything else for us to
 	// do really, as that will have been done in listen.
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&l->mtx);
 
 	if (!l->started) {
@@ -436,9 +433,8 @@ ipc_listener_accept(void *arg, nni_aio *aio)
 		nni_aio_finish_error(aio, NNG_ECLOSED);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ipc_listener_cancel, l)) != 0) {
+	if (!nni_aio_start(aio, ipc_listener_cancel, l)) {
 		nni_mtx_unlock(&l->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_aio_list_append(&l->acceptq, aio);
diff --git a/src/platform/windows/win_ipcconn.c b/src/platform/windows/win_ipcconn.c
index 56a688a4..f6c1b2a4 100644
--- a/src/platform/windows/win_ipcconn.c
+++ b/src/platform/windows/win_ipcconn.c
@@ -163,11 +163,8 @@ static void
 ipc_recv(void *arg, nni_aio *aio)
 {
 	ipc_conn *c = arg;
-	int       rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&c->mtx);
 	if (c->closed) {
 		nni_mtx_unlock(&c->mtx);
@@ -175,14 +172,13 @@ ipc_recv(void *arg, nni_aio *aio)
 		return;
 	}
 	if (c->recv_fail) {
-		rv = c->recv_rv;
+		int rv = c->recv_rv;
 		nni_mtx_unlock(&c->mtx);
 		nni_aio_finish_error(aio, rv);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ipc_recv_cancel, c)) != 0) {
+	if (!nni_aio_start(aio, ipc_recv_cancel, c)) {
 		nni_mtx_unlock(&c->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_list_append(&c->recv_aios, aio);
@@ -305,15 +301,11 @@ static void
 ipc_send(void *arg, nni_aio *aio)
 {
 	ipc_conn *c = arg;
-	int       rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 	nni_mtx_lock(&c->mtx);
-	if ((rv = nni_aio_schedule(aio, ipc_send_cancel, c)) != 0) {
+	if (!nni_aio_start(aio, ipc_send_cancel, c)) {
 		nni_mtx_unlock(&c->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_list_append(&c->send_aios, aio);
diff --git a/src/platform/windows/win_ipcdial.c b/src/platform/windows/win_ipcdial.c
index 347ee39b..d5e44750 100644
--- a/src/platform/windows/win_ipcdial.c
+++ b/src/platform/windows/win_ipcdial.c
@@ -145,16 +145,12 @@ ipc_dialer_dial(void *arg, nni_aio *aio)
 {
 	ipc_dialer    *d = arg;
 	ipc_dial_work *w = &ipc_connector;
-	int            rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 
 	nni_mtx_lock(&w->mtx);
-	if ((rv = nni_aio_schedule(aio, ipc_dial_cancel, d)) != 0) {
+	if (!nni_aio_start(aio, ipc_dial_cancel, d)) {
 		nni_mtx_unlock(&w->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
diff --git a/src/platform/windows/win_ipclisten.c b/src/platform/windows/win_ipclisten.c
index d09c98c7..72c68f94 100644
--- a/src/platform/windows/win_ipclisten.c
+++ b/src/platform/windows/win_ipclisten.c
@@ -261,23 +261,21 @@ static void
 ipc_listener_accept(void *arg, nni_aio *aio)
 {
 	ipc_listener *l = arg;
-	int           rv;
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+
+	nni_aio_reset(aio);
 	nni_mtx_lock(&l->mtx);
 	if (!l->started) {
 		nni_mtx_unlock(&l->mtx);
 		nni_aio_finish_error(aio, NNG_ESTATE);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, ipc_accept_cancel, l)) != 0) {
-		nni_aio_finish_error(aio, rv);
-	} else {
-		nni_list_append(&l->aios, aio);
-		if (nni_list_first(&l->aios) == aio) {
-			ipc_accept_start(l);
-		}
+	if (!nni_aio_start(aio, ipc_accept_cancel, l)) {
+		nni_mtx_unlock(&l->mtx);
+		return;
+	}
+	nni_list_append(&l->aios, aio);
+	if (nni_list_first(&l->aios) == aio) {
+		ipc_accept_start(l);
 	}
 	nni_mtx_unlock(&l->mtx);
 }
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;