Refactor stop again, closing numerous races (thanks valgrind!)

4 files changed, 33 insertions, 4 deletions

diff --git a/src/platform/posix/posix_aio.h b/src/platform/posix/posix_aio.h
index 9ab322a0..de662ff2 100644
--- a/src/platform/posix/posix_aio.h
+++ b/src/platform/posix/posix_aio.h
@@ -26,5 +26,6 @@ extern int nni_posix_pipedesc_init(nni_posix_pipedesc **, int);
 extern void nni_posix_pipedesc_fini(nni_posix_pipedesc *);
 extern int nni_posix_pipedesc_read(nni_posix_pipedesc *, nni_aio *);
 extern int nni_posix_pipedesc_write(nni_posix_pipedesc *, nni_aio *);
+extern void nni_posix_pipedesc_close(nni_posix_pipedesc *);
 
 #endif // PLATFORM_POSIX_AIO_H
diff --git a/src/platform/posix/posix_ipc.c b/src/platform/posix/posix_ipc.c
index ccf19fed..ba46b41e 100644
--- a/src/platform/posix/posix_ipc.c
+++ b/src/platform/posix/posix_ipc.c
@@ -243,6 +243,9 @@ nni_plat_ipc_shutdown(nni_plat_ipcsock *isp)
 		// (macOS does not see the shtudown).
 		(void) dup2(nni_plat_devnull, isp->fd);
 	}
+	if (isp->pd != NULL) {
+		nni_posix_pipedesc_close(isp->pd);
+	}
 }
 
 
diff --git a/src/platform/posix/posix_net.c b/src/platform/posix/posix_net.c
index c8b7766e..5ae9904a 100644
--- a/src/platform/posix/posix_net.c
+++ b/src/platform/posix/posix_net.c
@@ -299,6 +299,9 @@ nni_plat_tcp_shutdown(nni_plat_tcpsock *tsp)
 		// (macOS does not see the shtudown).
 		(void) dup2(nni_plat_devnull, tsp->fd);
 	}
+	if (tsp->pd != NULL) {
+		nni_posix_pipedesc_close(tsp->pd);
+	}
 }
 
 
diff --git a/src/platform/posix/posix_poll.c b/src/platform/posix/posix_poll.c
index a6024db0..7fb07917 100644
--- a/src/platform/posix/posix_poll.c
+++ b/src/platform/posix/posix_poll.c
@@ -224,6 +224,22 @@ nni_posix_poll_close(nni_posix_pipedesc *pd)
 }
 
 
+void
+nni_posix_pipedesc_close(nni_posix_pipedesc *pd)
+{
+	nni_posix_pollq *pq;
+
+	pq = pd->pq;
+	nni_mtx_lock(&pq->mtx);
+	pd->fd = -1;
+	nni_posix_poll_close(pd);
+	if (nni_list_active(&pq->pds, pd)) {
+		nni_list_remove(&pq->pds, pd);
+	}
+	nni_mtx_unlock(&pq->mtx);
+}
+
+
 static void
 nni_posix_poll_thr(void *arg)
 {
@@ -355,8 +371,6 @@ nni_posix_pipedesc_cancel(nni_aio *aio)
 	if (nni_list_active(&pd->readq, aio)) {
 		nni_list_remove(&pd->readq, aio);
 	}
-	aio->a_prov_cancel = NULL;
-	aio->a_prov_data = NULL;
 	nni_mtx_unlock(&pq->mtx);
 }
 
@@ -403,9 +417,19 @@ nni_posix_pipedesc_submit(nni_posix_pipedesc *pd, nni_list *l, nni_aio *aio)
 	int rv;
 	nni_posix_pollq *pq = pd->pq;
 
+	// XXX: this should be done only once, after tcp negot. is done
+	// or at init if we can get tcp negot. to be async.
 	(void) fcntl(pd->fd, F_SETFL, O_NONBLOCK);
 
 	nni_mtx_lock(&pq->mtx);
+	if (pd->fd < 0) {
+		nni_mtx_unlock(&pq->mtx);
+		nni_aio_finish(aio, NNG_ECLOSED, aio->a_count);
+	}
+	if ((rv = nni_aio_start(aio, nni_posix_pipedesc_cancel, pd)) != 0) {
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
 	if (!nni_list_active(&pq->pds, pd)) {
 		if ((rv = nni_posix_poll_grow(pq)) != 0) {
 			nni_aio_finish(aio, rv, aio->a_count);
@@ -416,8 +440,6 @@ nni_posix_pipedesc_submit(nni_posix_pipedesc *pd, nni_list *l, nni_aio *aio)
 		nni_list_append(&pq->pds, pd);
 	}
 	NNI_ASSERT(!nni_list_active(l, aio));
-	aio->a_prov_data = pd;
-	aio->a_prov_cancel = nni_posix_pipedesc_cancel;
 	// Only wake if we aren't already waiting for this type of I/O on
 	// this descriptor.
 	wake = nni_list_first(l) == NULL ? 1 : 0;