tcp: use nni_aio_start

3 files changed, 17 insertions, 30 deletions

diff --git a/src/platform/posix/posix_tcpconn.c b/src/platform/posix/posix_tcpconn.c
index c23e1840..9837a839 100644
--- a/src/platform/posix/posix_tcpconn.c
+++ b/src/platform/posix/posix_tcpconn.c
@@ -283,16 +283,11 @@ static void
 tcp_send(void *arg, nni_aio *aio)
 {
 	nni_tcp_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, tcp_cancel, c)) != 0) {
+	if (!nni_aio_start(aio, tcp_cancel, c)) {
 		nni_mtx_unlock(&c->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_aio_list_append(&c->writeq, aio);
@@ -313,16 +308,11 @@ static void
 tcp_recv(void *arg, nni_aio *aio)
 {
 	nni_tcp_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, tcp_cancel, c)) != 0) {
+	if (!nni_aio_start(aio, tcp_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_tcpdial.c b/src/platform/posix/posix_tcpdial.c
index 41036366..e827b666 100644
--- a/src/platform/posix/posix_tcpdial.c
+++ b/src/platform/posix/posix_tcpdial.c
@@ -193,9 +193,7 @@ nni_tcp_dial(nni_tcp_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
 	int                     ka;
 	int                     nd;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
+	nni_aio_reset(aio);
 
 	if (((sslen = nni_posix_nn2sockaddr(&ss, sa)) == 0) ||
 	    ((ss.ss_family != AF_INET) && (ss.ss_family != AF_INET6))) {
@@ -208,15 +206,22 @@ nni_tcp_dial(nni_tcp_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
 		return;
 	}
 
-	nni_refcnt_hold(&d->ref);
-
 	if ((rv = nni_posix_tcp_alloc(&c, d, fd)) != 0) {
+		(void) close(fd);
 		nni_aio_finish_error(aio, rv);
-		nni_posix_tcp_dialer_rele(d);
 		return;
 	}
 
+	// hold for the stream
+	nni_refcnt_hold(&d->ref);
+
 	nni_mtx_lock(&d->mtx);
+	if (!nni_aio_start(aio, tcp_dialer_cancel, d)) {
+		nni_mtx_unlock(&d->mtx);
+		nng_stream_free(&c->stream);
+		return;
+	}
+
 	if (d->closed) {
 		rv = NNG_ECLOSED;
 		goto error;
@@ -227,9 +232,6 @@ nni_tcp_dial(nni_tcp_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
 			goto error;
 		}
 	}
-	if ((rv = nni_aio_schedule(aio, tcp_dialer_cancel, d)) != 0) {
-		goto error;
-	}
 	c->dial_aio = aio;
 	if (connect(fd, (void *) &ss, sslen) != 0) {
 		if (errno != EINPROGRESS) {
diff --git a/src/platform/posix/posix_tcplisten.c b/src/platform/posix/posix_tcplisten.c
index 78e9b253..b14114b7 100644
--- a/src/platform/posix/posix_tcplisten.c
+++ b/src/platform/posix/posix_tcplisten.c
@@ -283,15 +283,11 @@ nni_tcp_listener_fini(nni_tcp_listener *l)
 void
 nni_tcp_listener_accept(nni_tcp_listener *l, nni_aio *aio)
 {
-	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) {
@@ -304,9 +300,8 @@ nni_tcp_listener_accept(nni_tcp_listener *l, nni_aio *aio)
 		nni_aio_finish_error(aio, NNG_ECLOSED);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, tcp_listener_cancel, l)) != 0) {
+	if (!nni_aio_start(aio, tcp_listener_cancel, l)) {
 		nni_mtx_unlock(&l->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_aio_list_append(&l->acceptq, aio);