4 files changed, 73 insertions, 37 deletions

diff --git a/src/platform/posix/posix_aiothr.c b/src/platform/posix/posix_aiothr.c
index 2c11dcb2..a01fa194 100644
--- a/src/platform/posix/posix_aiothr.c
+++ b/src/platform/posix/posix_aiothr.c
@@ -239,14 +239,16 @@ nni_posix_aioq_start(nni_posix_aioq *q)
 static void
 nni_posix_aioq_fini(nni_posix_aioq *q)
 {
-	nni_mtx_lock(&q->aq_lk);
-	q->aq_fd = -1;
-	nni_cv_wake(&q->aq_cv);
-	nni_mtx_unlock(&q->aq_lk);
+	if (q->aq_fd > 0) {
+		nni_mtx_lock(&q->aq_lk);
+		q->aq_fd = -1;
+		nni_cv_wake(&q->aq_cv);
+		nni_mtx_unlock(&q->aq_lk);
 
-	nni_thr_fini(&q->aq_thr);
-	nni_cv_fini(&q->aq_cv);
-	nni_mtx_fini(&q->aq_lk);
+		nni_thr_fini(&q->aq_thr);
+		nni_cv_fini(&q->aq_cv);
+		nni_mtx_fini(&q->aq_lk);
+	}
 }
 
 
diff --git a/src/platform/posix/posix_impl.h b/src/platform/posix/posix_impl.h
index 5da18323..dea09fa1 100644
--- a/src/platform/posix/posix_impl.h
+++ b/src/platform/posix/posix_impl.h
@@ -35,15 +35,6 @@ extern int nni_plat_errno(int);
 #endif
 
 
-
-#ifdef PLATFORM_POSIX_IPC
-struct nni_plat_ipcsock {
-	int	fd;
-	int	devnull;        // used for shutting down blocking accept()
-	char *	unlink;         // path to unlink at termination
-};
-#endif
-
 // Define types that this platform uses.
 #ifdef PLATFORM_POSIX_THREAD
 
diff --git a/src/platform/posix/posix_ipc.c b/src/platform/posix/posix_ipc.c
index 7a9c6d52..e75edeca 100644
--- a/src/platform/posix/posix_ipc.c
+++ b/src/platform/posix/posix_ipc.c
@@ -10,6 +10,7 @@
 #include "core/nng_impl.h"
 
 #ifdef PLATFORM_POSIX_IPC
+#include "platform/posix/posix_aio.h"
 
 #include <errno.h>
 #include <stdlib.h>
@@ -29,6 +30,13 @@
 #undef sun
 #endif
 
+struct nni_plat_ipcsock {
+	int			fd;
+	int			devnull;        // for shutting down accept()
+	char *			unlink;         // path to unlink at fini
+	nni_posix_aio_pipe	aiop;
+};
+
 #ifdef  SOCK_CLOEXEC
 #define NNI_IPC_SOCKTYPE	(SOCK_STREAM | SOCK_CLOEXEC)
 #else
@@ -59,6 +67,20 @@ nni_plat_ipc_path_to_sockaddr(struct sockaddr_un *sun, const char *path)
 
 
 int
+nni_plat_ipc_aio_send(nni_plat_ipcsock *isp, nni_aio *aio)
+{
+	return (nni_posix_aio_write(&isp->aiop, aio));
+}
+
+
+int
+nni_plat_ipc_aio_recv(nni_plat_ipcsock *isp, nni_aio *aio)
+{
+	return (nni_posix_aio_read(&isp->aiop, aio));
+}
+
+
+int
 nni_plat_ipc_send(nni_plat_ipcsock *s, nni_iov *iovs, int cnt)
 {
 	struct iovec iov[4];    // We never have more than 3 at present
@@ -178,36 +200,46 @@ nni_plat_ipc_setopts(int fd)
 
 
 int
-nni_plat_ipc_init(nni_plat_ipcsock *s)
+nni_plat_ipc_init(nni_plat_ipcsock **ispp)
 {
-	s->fd = -1;
+	nni_plat_ipcsock *isp;
+
+	if ((isp = NNI_ALLOC_STRUCT(isp)) == NULL) {
+		return (NNG_ENOMEM);
+	}
+	isp->fd = -1;
+	*ispp = isp;
 	return (0);
 }
 
 
 void
-nni_plat_ipc_fini(nni_plat_ipcsock *s)
+nni_plat_ipc_fini(nni_plat_ipcsock *isp)
 {
-	if (s->fd != -1) {
-		(void) close(s->fd);
-		s->fd = -1;
+	if (isp->fd != -1) {
+		(void) close(isp->fd);
+		isp->fd = -1;
 	}
-	if (s->unlink != NULL) {
-		(void) unlink(s->unlink);
-		nni_free(s->unlink, strlen(s->unlink) + 1);
+	if (isp->unlink != NULL) {
+		(void) unlink(isp->unlink);
+		nni_free(isp->unlink, strlen(isp->unlink) + 1);
 	}
+
+	nni_posix_aio_pipe_fini(&isp->aiop);
+
+	NNI_FREE_STRUCT(isp);
 }
 
 
 void
-nni_plat_ipc_shutdown(nni_plat_ipcsock *s)
+nni_plat_ipc_shutdown(nni_plat_ipcsock *isp)
 {
-	if (s->fd != -1) {
-		(void) shutdown(s->fd, SHUT_RDWR);
+	if (isp->fd != -1) {
+		(void) shutdown(isp->fd, SHUT_RDWR);
 		// This causes the equivalent of a close.  Hopefully waking
 		// up anything that didn't get the hint with the shutdown.
 		// (macOS does not see the shtudown).
-		(void) dup2(nni_plat_devnull, s->fd);
+		(void) dup2(nni_plat_devnull, isp->fd);
 	}
 }
 
@@ -278,7 +310,7 @@ nni_plat_ipc_listen(nni_plat_ipcsock *s, const char *path)
 
 
 int
-nni_plat_ipc_connect(nni_plat_ipcsock *s, const char *path)
+nni_plat_ipc_connect(nni_plat_ipcsock *isp, const char *path)
 {
 	int fd;
 	int len;
@@ -305,15 +337,22 @@ nni_plat_ipc_connect(nni_plat_ipcsock *s, const char *path)
 		}
 		return (rv);
 	}
-	s->fd = fd;
+
+	if ((rv = nni_posix_aio_pipe_init(&isp->aiop, fd)) != 0) {
+		(void) close(fd);
+		return (rv);
+	}
+
+	isp->fd = fd;
 	return (0);
 }
 
 
 int
-nni_plat_ipc_accept(nni_plat_ipcsock *s, nni_plat_ipcsock *server)
+nni_plat_ipc_accept(nni_plat_ipcsock *isp, nni_plat_ipcsock *server)
 {
 	int fd;
+	int rv;
 
 	for (;;) {
 #ifdef NNG_USE_ACCEPT4
@@ -341,7 +380,12 @@ nni_plat_ipc_accept(nni_plat_ipcsock *s, nni_plat_ipcsock *server)
 
 	nni_plat_ipc_setopts(fd);
 
-	s->fd = fd;
+	if ((rv = nni_posix_aio_pipe_init(&isp->aiop, fd)) != 0) {
+		(void) close(fd);
+		return (rv);
+	}
+
+	isp->fd = fd;
 	return (0);
 }
 
diff --git a/src/platform/posix/posix_net.c b/src/platform/posix/posix_net.c
index c7651655..94dc2667 100644
--- a/src/platform/posix/posix_net.c
+++ b/src/platform/posix/posix_net.c
@@ -8,9 +8,9 @@
 //
 
 #include "core/nng_impl.h"
-#include "platform/posix/posix_aio.h"
 
 #ifdef PLATFORM_POSIX_NET
+#include "platform/posix/posix_aio.h"
 
 #include <errno.h>
 #include <stdlib.h>
@@ -31,13 +31,11 @@
 #define NNI_TCP_SOCKTYPE	SOCK_STREAM
 #endif
 
-#ifdef PLATFORM_POSIX_NET
 struct nni_plat_tcpsock {
 	int			fd;
-	int			devnull; // used for shutting down blocking accept()
+	int			devnull; // for shutting down accept()
 	nni_posix_aio_pipe	aiop;
 };
-#endif
 
 static int
 nni_plat_to_sockaddr(struct sockaddr_storage *ss, const nni_sockaddr *sa)
@@ -284,6 +282,7 @@ nni_plat_tcp_fini(nni_plat_tcpsock *tsp)
 		(void) close(tsp->fd);
 		tsp->fd = -1;
 	}
+	nni_posix_aio_pipe_fini(&tsp->aiop);
 	NNI_FREE_STRUCT(tsp);
 }