3 files changed, 219 insertions, 44 deletions

diff --git a/src/platform/posix/posix_aio.h b/src/platform/posix/posix_aio.h
index 26ef19b9..3bb59b95 100644
--- a/src/platform/posix/posix_aio.h
+++ b/src/platform/posix/posix_aio.h
@@ -30,4 +30,10 @@ extern void nni_posix_pipedesc_read(nni_posix_pipedesc *, nni_aio *);
 extern void nni_posix_pipedesc_write(nni_posix_pipedesc *, nni_aio *);
 extern void nni_posix_pipedesc_close(nni_posix_pipedesc *);
 
+extern int nni_posix_epdesc_init(nni_posix_epdesc **, int);
+extern void nni_posix_epdesc_fini(nni_posix_epdesc *);
+extern void nni_posix_epdesc_close(nni_posix_epdesc *);
+extern void nni_posix_epdesc_connect(nni_posix_epdesc *, nni_aio *);
+extern void nni_posix_epdesc_accept(nni_posix_epdesc *, nni_aio *);
+
 #endif // PLATFORM_POSIX_AIO_H
diff --git a/src/platform/posix/posix_poll.c b/src/platform/posix/posix_poll.c
index 751044f0..07ab1dc5 100644
--- a/src/platform/posix/posix_poll.c
+++ b/src/platform/posix/posix_poll.c
@@ -55,6 +55,8 @@ struct nni_posix_epdesc {
 	nni_posix_pollq *	pq;
 	struct sockaddr_storage locaddr;
 	struct sockaddr_storage remaddr;
+	socklen_t		loclen;
+	socklen_t		remlen;
 };
 
 
@@ -80,6 +82,41 @@ struct nni_posix_pollq {
 static nni_posix_pollq nni_posix_global_pollq;
 
 
+static int
+nni_posix_poll_grow(nni_posix_pollq *pq)
+{
+	int grow = pq->npds + pq->neds + 2; // one for us, one for waker
+	int noldfds;
+	struct pollfd *oldfds;
+	struct pollfd *newfds;
+
+	if ((grow < pq->nfds) || (grow < pq->nnewfds)) {
+		return (0);
+	}
+
+	grow = grow + 128;
+
+	// Maybe we are adding a *lot* of pipes at once, and have to grow
+	// multiple times before the poller gets scheduled.  In that case
+	// toss the old array before we finish.
+	oldfds = pq->newfds;
+	noldfds = pq->nnewfds;
+
+	if ((newfds = nni_alloc(grow * sizeof (struct pollfd))) == NULL) {
+		return (NNG_ENOMEM);
+	}
+
+
+	pq->newfds = newfds;
+	pq->nnewfds = grow;
+
+	if (noldfds != 0) {
+		nni_free(oldfds, noldfds * sizeof (struct pollfd));
+	}
+	return (0);
+}
+
+
 static void
 nni_posix_epdesc_cancel(nni_aio *aio)
 {
@@ -131,9 +168,7 @@ nni_posix_poll_connect(nni_posix_epdesc *ed)
 		rv = -1;
 		sz = sizeof (rv);
 		if (getsockopt(ed->fd, SOL_SOCKET, SO_ERROR, &rv, &sz) < 0) {
-			nni_posix_epdesc_finish(aio,
-			    nni_plat_errno(errno), 0);
-			continue;
+			rv = errno;
 		}
 		switch (rv) {
 		case 0:
@@ -178,6 +213,7 @@ nni_posix_poll_accept(nni_posix_epdesc *ed)
 
 		if (newfd >= 0) {
 			// successful connection request!
+			// We abuse the count to hold our new file descriptor.
 			nni_posix_epdesc_finish(aio, 0, newfd);
 			continue;
 		}
@@ -218,6 +254,165 @@ nni_posix_poll_epclose(nni_posix_epdesc *ed)
 }
 
 
+static int
+nni_posix_epdesc_add(nni_posix_pollq *pq, nni_posix_epdesc *ed)
+{
+	int rv;
+
+	// Add epdesc to the pollq if it isn't already there.
+	if (!nni_list_active(&pq->eds, ed)) {
+		if ((rv = nni_posix_poll_grow(pq)) != 0) {
+			return (rv);
+		}
+		nni_list_append(&pq->eds, ed);
+		pq->neds++;
+	}
+	return (0);
+}
+
+
+void
+nni_posix_epdesc_connect(nni_posix_epdesc *ed, nni_aio *aio)
+{
+	// NB: We assume that the FD is already set to nonblocking mode.
+	int rv;
+	nni_posix_pollq *pq = ed->pq;
+	int wake;
+
+	nni_mtx_lock(&pq->mtx);
+	// If we can't start, it means that the AIO was stopped.
+	if ((rv = nni_aio_start(aio, nni_posix_epdesc_cancel, ed)) != 0) {
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
+	if (ed->fd < 0) {
+		nni_posix_epdesc_finish(aio, NNG_ECLOSED, 0);
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
+	rv = connect(ed->fd, (void *) &ed->remaddr, ed->remlen);
+	if (rv == 0) {
+		// Immediate connect, cool!  This probably only happens on
+		// loopback, and probably not on every platform.
+		nni_posix_epdesc_finish(aio, 0, 0);
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
+	if (errno != EINPROGRESS) {
+		// Some immediate failure occurred.
+		nni_posix_epdesc_finish(aio, nni_plat_errno(errno), 0);
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
+
+	// We have to submit to the pollq, because the connection is pending.
+	if ((rv = nni_posix_epdesc_add(pq, ed)) != 0) {
+		nni_posix_epdesc_finish(aio, rv, 0);
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
+
+	NNI_ASSERT(!nni_list_active(&ed->connectq, aio));
+	wake = nni_list_first(&ed->connectq) == NULL ? 1 : 0;
+	nni_list_append(&ed->connectq, aio);
+	if (wake) {
+		nni_plat_pipe_raise(pq->wakewfd);
+	}
+	nni_mtx_unlock(&pq->mtx);
+}
+
+
+void
+nni_posix_epdesc_accept(nni_posix_epdesc *ed, nni_aio *aio)
+{
+	// NB: We assume that the FD is already set to nonblocking mode.
+	int rv;
+	int wake;
+	nni_posix_pollq *pq = ed->pq;
+
+	// 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.
+	nni_mtx_lock(&pq->mtx);
+	// If we can't start, it means that the AIO was stopped.
+	if ((rv = nni_aio_start(aio, nni_posix_epdesc_cancel, ed)) != 0) {
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
+
+	if (ed->fd < 0) {
+		nni_mtx_unlock(&pq->mtx);
+		nni_posix_epdesc_finish(aio, NNG_ECLOSED, 0);
+		return;
+	}
+
+	// We have to submit to the pollq, because the connection is pending.
+	if ((rv = nni_posix_epdesc_add(pq, ed)) != 0) {
+		nni_posix_epdesc_finish(aio, rv, 0);
+		nni_mtx_lock(&pq->mtx);
+	}
+	NNI_ASSERT(!nni_list_active(&ed->acceptq, aio));
+	wake = nni_list_first(&ed->acceptq) == NULL ? 1 : 0;
+	nni_list_append(&ed->acceptq, aio);
+	if (wake) {
+		nni_plat_pipe_raise(pq->wakewfd);
+	}
+	nni_mtx_unlock(&pq->mtx);
+}
+
+
+int
+nni_posix_epdesc_init(nni_posix_epdesc **edp, int fd)
+{
+	nni_posix_epdesc *ed;
+
+
+	if ((ed = NNI_ALLOC_STRUCT(ed)) == NULL) {
+		return (NNG_ENOMEM);
+	}
+
+	// We could randomly choose a different pollq, or for efficiencies
+	// sake we could take a modulo of the file desc number to choose
+	// one.  For now we just have a global pollq.  Note that by tying
+	// the ed to a single pollq we may get some kind of cache warmth.
+
+	ed->pq = &nni_posix_global_pollq;
+	ed->fd = fd;
+	ed->index = 0;
+
+	// Ensure we are in non-blocking mode.
+	(void) fcntl(fd, F_SETFL, O_NONBLOCK);
+
+	NNI_LIST_INIT(&ed->connectq, nni_aio, a_prov_node);
+	NNI_LIST_INIT(&ed->acceptq, nni_aio, a_prov_node);
+
+	*edp = ed;
+	return (0);
+}
+
+
+void
+nni_posix_epdesc_fini(nni_posix_epdesc *ed)
+{
+	nni_aio *aio;
+	nni_posix_pollq *pq = ed->pq;
+
+	nni_mtx_lock(&pq->mtx);
+
+	// This removes any aios from our list.
+	nni_posix_poll_epclose(ed);
+
+	if (nni_list_active(&pq->eds, ed)) {
+		nni_list_remove(&pq->eds, ed);
+		pq->neds--;
+	}
+	nni_mtx_unlock(&pq->mtx);
+
+	NNI_FREE_STRUCT(ed);
+}
+
+
 static void
 nni_posix_pipedesc_finish(nni_aio *aio, int rv)
 {
@@ -568,41 +763,6 @@ nni_posix_pipedesc_cancel(nni_aio *aio)
 }
 
 
-static int
-nni_posix_poll_grow(nni_posix_pollq *pq)
-{
-	int grow = pq->npds + pq->neds + 2; // one for us, one for waker
-	int noldfds;
-	struct pollfd *oldfds;
-	struct pollfd *newfds;
-
-	if ((grow < pq->nfds) || (grow < pq->nnewfds)) {
-		return (0);
-	}
-
-	grow = grow + 128;
-
-	// Maybe we are adding a *lot* of pipes at once, and have to grow
-	// multiple times before the poller gets scheduled.  In that case
-	// toss the old array before we finish.
-	oldfds = pq->newfds;
-	noldfds = pq->nnewfds;
-
-	if ((newfds = nni_alloc(grow * sizeof (struct pollfd))) == NULL) {
-		return (NNG_ENOMEM);
-	}
-
-
-	pq->newfds = newfds;
-	pq->nnewfds = grow;
-
-	if (noldfds != 0) {
-		nni_free(oldfds, noldfds * sizeof (struct pollfd));
-	}
-	return (0);
-}
-
-
 static void
 nni_posix_pipedesc_submit(nni_posix_pipedesc *pd, nni_list *l, nni_aio *aio)
 {
@@ -611,9 +771,13 @@ nni_posix_pipedesc_submit(nni_posix_pipedesc *pd, nni_list *l, nni_aio *aio)
 	nni_posix_pollq *pq = pd->pq;
 
 	nni_mtx_lock(&pq->mtx);
+	if ((rv = nni_aio_start(aio, nni_posix_pipedesc_cancel, pd)) != 0) {
+		nni_mtx_unlock(&pq->mtx);
+		return;
+	}
 	if (pd->fd < 0) {
+		nni_posix_pipedesc_finish(aio, NNG_ECLOSED);
 		nni_mtx_unlock(&pq->mtx);
-		nni_aio_finish(aio, NNG_ECLOSED, aio->a_count);
 		return;
 	}
 	// XXX: We really should just make all the FDs nonblocking, but we
@@ -622,13 +786,9 @@ nni_posix_pipedesc_submit(nni_posix_pipedesc *pd, nni_list *l, nni_aio *aio)
 		(void) fcntl(pd->fd, F_SETFL, O_NONBLOCK);
 		pd->nonblocking = 1;
 	}
-	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);
+			nni_posix_pipedesc_finish(aio, rv);
 			nni_mtx_unlock(&pq->mtx);
 			return;
 		}
diff --git a/src/platform/posix/posix_socket.c b/src/platform/posix/posix_socket.c
index ce887a78..6d2b6c29 100644
--- a/src/platform/posix/posix_socket.c
+++ b/src/platform/posix/posix_socket.c
@@ -46,6 +46,7 @@ struct nni_posix_sock {
 	int			devnull;        // for shutting down accept()
 	char *			unlink;         // path to unlink at unbind
 	nni_posix_pipedesc *	pd;
+	nni_posix_epdesc *	ed;
 	int			tcpnodelay;
 };
 
@@ -213,6 +214,9 @@ nni_posix_sock_fini(nni_posix_sock *s)
 	if (s->pd != NULL) {
 		nni_posix_pipedesc_fini(s->pd);
 	}
+	if (s->ed != NULL) {
+		nni_posix_epdesc_fini(s->ed);
+	}
 	if (s->unlink != NULL) {
 		(void) unlink(s->unlink);
 		nni_free(s->unlink, strlen(s->unlink) + 1);
@@ -498,6 +502,11 @@ nni_posix_sock_connect_sync(nni_posix_sock *s, const nni_sockaddr *addr,
 		(void) close(fd);
 		return (rv);
 	}
+	if (s->pd != NULL) {
+		// If we had a prior pipedesc hanging around, nuke it.
+		nni_posix_pipedesc_fini(s->pd);
+		s->pd = NULL;
+	}
 	if ((rv = nni_posix_pipedesc_init(&s->pd, fd)) != 0) {
 		(void) close(fd);
 		return (rv);