sockfd: convert to use nni_aio_start

src/core/sockfd.c

@@ -1,5 +1,5 @@
 //
-// Copyright 2023 Staysail Systems, Inc. <[email protected]>
+// Copyright 2024 Staysail Systems, Inc. <[email protected]>
 //
 // This software is supplied under the terms of the MIT License, a
 // copy of which should be located in the distribution where this
@@ -116,10 +116,11 @@ sfd_listener_accept(void *arg, nng_aio *aio)
 {
 	sfd_listener *l = arg;
 
-	if (nni_aio_begin(aio) != 0) {
+	nni_mtx_lock(&l->mtx);
+	if (!nni_aio_start(aio, sfd_cancel_accept, l)) {
+		nni_mtx_unlock(&l->mtx);
 		return;
 	}
-	nni_mtx_lock(&l->mtx);
 	if (l->closed) {
 		nni_mtx_unlock(&l->mtx);
 		nni_aio_finish_error(aio, NNG_ECLOSED);
@@ -128,7 +129,7 @@ sfd_listener_accept(void *arg, nng_aio *aio)
 
 	if (l->listen_cnt) {
 		sfd_start_conn(l, aio);
-	} else if (nni_aio_defer(aio, sfd_cancel_accept, l)) {
+	} else {
 		nni_aio_list_append(&l->accept_q, aio);
 	}
 	nni_mtx_unlock(&l->mtx);

src/platform/posix/posix_sockfd.c

@@ -275,16 +275,11 @@ static void
 sfd_send(void *arg, nni_aio *aio)
 {
 	nni_sfd_conn *c = arg;
-	int           rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
 	nni_mtx_lock(&c->mtx);
 
-	if ((rv = nni_aio_schedule(aio, sfd_cancel, c)) != 0) {
+	if (!nni_aio_start(aio, sfd_cancel, c)) {
 		nni_mtx_unlock(&c->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_aio_list_append(&c->writeq, aio);
@@ -305,16 +300,11 @@ static void
 sfd_recv(void *arg, nni_aio *aio)
 {
 	nni_sfd_conn *c = arg;
-	int           rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
 	nni_mtx_lock(&c->mtx);
 
-	if ((rv = nni_aio_schedule(aio, sfd_cancel, c)) != 0) {
+	if (!nni_aio_start(aio, sfd_cancel, c)) {
 		nni_mtx_unlock(&c->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_aio_list_append(&c->readq, aio);

src/sp/transport/socket/sockfd.c

@@ -453,19 +453,10 @@ static void
 sfd_tran_pipe_send(void *arg, nni_aio *aio)
 {
 	sfd_tran_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_mtx_lock(&p->mtx);
-	if ((rv = nni_aio_schedule(aio, sfd_tran_pipe_send_cancel, p)) != 0) {
+	if (!nni_aio_start(aio, sfd_tran_pipe_send_cancel, p)) {
 		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	nni_list_append(&p->sendq, aio);
@@ -530,15 +521,10 @@ static void
 sfd_tran_pipe_recv(void *arg, nni_aio *aio)
 {
 	sfd_tran_pipe *p = arg;
-	int            rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
 	nni_mtx_lock(&p->mtx);
-	if ((rv = nni_aio_schedule(aio, sfd_tran_pipe_recv_cancel, p)) != 0) {
+	if (!nni_aio_start(aio, sfd_tran_pipe_recv_cancel, p)) {
 		nni_mtx_unlock(&p->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 
@@ -687,7 +673,7 @@ sfd_tran_accept_cb(void *arg)
 		ep->useraio = NULL;
 		nni_aio_finish_error(aio, rv);
 	}
-	if (!ep->closed) {
+	if (!ep->closed && rv != NNG_ESTOPPED) {
 		nng_stream_listener_accept(ep->listener, &ep->connaio);
 	}
 	nni_mtx_unlock(&ep->mtx);
@@ -791,11 +777,7 @@ static void
 sfd_tran_ep_accept(void *arg, nni_aio *aio)
 {
 	sfd_tran_ep *ep = arg;
-	int          rv;
 
-	if (nni_aio_begin(aio) != 0) {
-		return;
-	}
 	nni_mtx_lock(&ep->mtx);
 	if (ep->closed) {
 		nni_mtx_unlock(&ep->mtx);
@@ -807,9 +789,8 @@ sfd_tran_ep_accept(void *arg, nni_aio *aio)
 		nni_aio_finish_error(aio, NNG_EBUSY);
 		return;
 	}
-	if ((rv = nni_aio_schedule(aio, sfd_tran_ep_cancel, ep)) != 0) {
+	if (!nni_aio_start(aio, sfd_tran_ep_cancel, ep)) {
 		nni_mtx_unlock(&ep->mtx);
-		nni_aio_finish_error(aio, rv);
 		return;
 	}
 	ep->useraio = aio;