Convert to POSIX polled I/O for async; start of cancelable aio.

This eliminates the two threads per pipe that were being used to provide
basic I/O handling, replacing them with a single global thread for now,
that uses poll and nonblocking I/O.  This should lead to great scalability.

The infrastructure is in place to easily expand to multiple polling worker
threads.  Some thought needs to be given about how to scale this to engage
multiple CPUs.  Horizontal scaling may also shorten the poll() lists easing
C10K problem.

We should look into better solutions than poll() for platforms that have
them (epoll on Linux, kqueue on BSD, and event ports on illumos).

Note that the file descriptors start out in blocking mode for now, but
then are placed into non-blocking mode.  This is because the negotiation
phase is not yet callback driven, and so needs to be synchronous.

1 files changed, 8 insertions, 6 deletions

diff --git a/src/platform/posix/posix_ipc.c b/src/platform/posix/posix_ipc.c
index e75edeca..ccf19fed 100644
--- a/src/platform/posix/posix_ipc.c
+++ b/src/platform/posix/posix_ipc.c
@@ -34,7 +34,7 @@ struct nni_plat_ipcsock {
 	int			fd;
 	int			devnull;        // for shutting down accept()
 	char *			unlink;         // path to unlink at fini
-	nni_posix_aio_pipe	aiop;
+	nni_posix_pipedesc *	pd;
 };
 
 #ifdef  SOCK_CLOEXEC
@@ -69,14 +69,14 @@ 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));
+	return (nni_posix_pipedesc_write(isp->pd, aio));
 }
 
 
 int
 nni_plat_ipc_aio_recv(nni_plat_ipcsock *isp, nni_aio *aio)
 {
-	return (nni_posix_aio_read(&isp->aiop, aio));
+	return (nni_posix_pipedesc_read(isp->pd, aio));
 }
 
 
@@ -225,7 +225,9 @@ nni_plat_ipc_fini(nni_plat_ipcsock *isp)
 		nni_free(isp->unlink, strlen(isp->unlink) + 1);
 	}
 
-	nni_posix_aio_pipe_fini(&isp->aiop);
+	if (isp->pd != NULL) {
+		nni_posix_pipedesc_fini(isp->pd);
+	}
 
 	NNI_FREE_STRUCT(isp);
 }
@@ -338,7 +340,7 @@ nni_plat_ipc_connect(nni_plat_ipcsock *isp, const char *path)
 		return (rv);
 	}
 
-	if ((rv = nni_posix_aio_pipe_init(&isp->aiop, fd)) != 0) {
+	if ((rv = nni_posix_pipedesc_init(&isp->pd, fd)) != 0) {
 		(void) close(fd);
 		return (rv);
 	}
@@ -380,7 +382,7 @@ nni_plat_ipc_accept(nni_plat_ipcsock *isp, nni_plat_ipcsock *server)
 
 	nni_plat_ipc_setopts(fd);
 
-	if ((rv = nni_posix_aio_pipe_init(&isp->aiop, fd)) != 0) {
+	if ((rv = nni_posix_pipedesc_init(&isp->pd, fd)) != 0) {
 		(void) close(fd);
 		return (rv);
 	}