fixes #523 dialers could support multiple outstanding dial requests

fixes #179 DNS resolution should be done at connect time
fixes #586 Windows IO completion port work could be better
fixes #339 Windows iocp could use synchronous completions
fixes #280 TCP abstraction improvements

This is a rather monstrous set of changes, which refactors TCP, and
the underlying Windows I/O completion path logic, in order to obtain
a cleaner, simpler API, with support for asynchronous DNS lookups performed
on connect rather than initialization time, the ability to have multiple
connects or accepts pending, as well as fewer extraneous function calls.

The Windows code also benefits from greatly reduced context switching,
fewer lock operations performed, and a reduced number of system calls
on the hot code path.  (We use automatic event resetting instead of manual.)

Some dead code was removed as well, and a few potential edge case leaks
on failure paths (in the websocket code) were plugged.

Note that all TCP based transports benefit from this work. The IPC code
on Windows still uses the legacy IOCP for now, as does the UDP code (used
for ZeroTier.)  We will be converting those soon too.

1 files changed, 410 insertions, 0 deletions

diff --git a/src/platform/posix/posix_tcpconn.c b/src/platform/posix/posix_tcpconn.c
new file mode 100644
index 00000000..30525648
--- /dev/null
+++ b/src/platform/posix/posix_tcpconn.c
@@ -0,0 +1,410 @@
+//
+// Copyright 2018 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2018 Capitar IT Group BV <info@capitar.com>
+//
+// This software is supplied under the terms of the MIT License, a
+// copy of which should be located in the distribution where this
+// file was obtained (LICENSE.txt).  A copy of the license may also be
+// found online at https://opensource.org/licenses/MIT.
+//
+
+#include "core/nng_impl.h"
+
+#ifdef NNG_PLATFORM_POSIX
+#include "platform/posix/posix_aio.h"
+
+#include <arpa/inet.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <poll.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/uio.h>
+#include <unistd.h>
+#ifdef NNG_HAVE_ALLOCA
+#include <alloca.h>
+#endif
+
+#ifndef MSG_NOSIGNAL
+#define MSG_NOSIGNAL 0
+#endif
+
+#include "posix_tcp.h"
+
+static void
+tcp_conn_dowrite(nni_tcp_conn *c)
+{
+	nni_aio *aio;
+	int      fd;
+
+	if (c->closed || ((fd = nni_posix_pfd_fd(c->pfd)) < 0)) {
+		return;
+	}
+
+	while ((aio = nni_list_first(&c->writeq)) != NULL) {
+		unsigned      i;
+		int           n;
+		int           niov;
+		unsigned      naiov;
+		nni_iov *     aiov;
+		struct msghdr hdr;
+#ifdef NNG_HAVE_ALLOCA
+		struct iovec *iovec;
+#else
+		struct iovec iovec[16];
+#endif
+
+		memset(&hdr, 0, sizeof(hdr));
+		nni_aio_get_iov(aio, &naiov, &aiov);
+
+#ifdef NNG_HAVE_ALLOCA
+		if (naiov > 64) {
+			nni_aio_list_remove(aio);
+			nni_aio_finish_error(aio, NNG_EINVAL);
+			continue;
+		}
+		iovec = alloca(naiov * sizeof(*iovec));
+#else
+		if (naiov > NNI_NUM_ELEMENTS(iovec)) {
+			nni_aio_list_remove(aio);
+			nni_aio_finish_error(aio, NNG_EINVAL);
+			continue;
+		}
+#endif
+
+		for (niov = 0, i = 0; i < naiov; i++) {
+			if (aiov[i].iov_len > 0) {
+				iovec[niov].iov_len  = aiov[i].iov_len;
+				iovec[niov].iov_base = aiov[i].iov_buf;
+				niov++;
+			}
+		}
+
+		hdr.msg_iovlen = niov;
+		hdr.msg_iov    = iovec;
+
+		if ((n = sendmsg(fd, &hdr, MSG_NOSIGNAL)) < 0) {
+			switch (errno) {
+			case EINTR:
+				continue;
+			case EAGAIN:
+#ifdef EWOULDBLOCK
+#if EWOULDBLOCK != EAGAIN
+			case EWOULDBLOCK:
+#endif
+#endif
+				return;
+			default:
+				nni_aio_list_remove(aio);
+				nni_aio_finish_error(
+				    aio, nni_plat_errno(errno));
+				return;
+			}
+		}
+
+		nni_aio_bump_count(aio, n);
+		// We completed the entire operation on this aio.
+		// (Sendmsg never returns a partial result.)
+		nni_aio_list_remove(aio);
+		nni_aio_finish(aio, 0, nni_aio_count(aio));
+
+		// Go back to start of loop to see if there is another
+		// aio ready for us to process.
+	}
+}
+
+static void
+tcp_conn_doread(nni_tcp_conn *c)
+{
+	nni_aio *aio;
+	int      fd;
+
+	if (c->closed || ((fd = nni_posix_pfd_fd(c->pfd)) < 0)) {
+		return;
+	}
+
+	while ((aio = nni_list_first(&c->readq)) != NULL) {
+		unsigned i;
+		int      n;
+		int      niov;
+		unsigned naiov;
+		nni_iov *aiov;
+#ifdef NNG_HAVE_ALLOCA
+		struct iovec *iovec;
+#else
+		struct iovec iovec[16];
+#endif
+
+		nni_aio_get_iov(aio, &naiov, &aiov);
+#ifdef NNG_HAVE_ALLOCA
+		if (naiov > 64) {
+			nni_aio_list_remove(aio);
+			nni_aio_finish_error(aio, NNG_EINVAL);
+			continue;
+		}
+		iovec = alloca(naiov * sizeof(*iovec));
+#else
+		if (naiov > NNI_NUM_ELEMENTS(iovec)) {
+			nni_aio_list_remove(aio);
+			nni_aio_finish_error(aio, NNG_EINVAL);
+			continue;
+		}
+#endif
+		for (niov = 0, i = 0; i < naiov; i++) {
+			if (aiov[i].iov_len != 0) {
+				iovec[niov].iov_len  = aiov[i].iov_len;
+				iovec[niov].iov_base = aiov[i].iov_buf;
+				niov++;
+			}
+		}
+
+		if ((n = readv(fd, iovec, niov)) < 0) {
+			switch (errno) {
+			case EINTR:
+				continue;
+			case EAGAIN:
+				return;
+			default:
+				nni_aio_list_remove(aio);
+				nni_aio_finish_error(
+				    aio, nni_plat_errno(errno));
+				return;
+			}
+		}
+
+		if (n == 0) {
+			// No bytes indicates a closed descriptor.
+			// This implicitly completes this (all!) aio.
+			nni_aio_list_remove(aio);
+			nni_aio_finish_error(aio, NNG_ECLOSED);
+			continue;
+		}
+
+		nni_aio_bump_count(aio, n);
+
+		// We completed the entire operation on this aio.
+		nni_aio_list_remove(aio);
+		nni_aio_finish(aio, 0, nni_aio_count(aio));
+
+		// Go back to start of loop to see if there is another
+		// aio ready for us to process.
+	}
+}
+
+void
+nni_tcp_conn_close(nni_tcp_conn *c)
+{
+	nni_mtx_lock(&c->mtx);
+	if (!c->closed) {
+		nni_aio *aio;
+		c->closed = true;
+		while (((aio = nni_list_first(&c->readq)) != NULL) ||
+		    ((aio = nni_list_first(&c->writeq)) != NULL)) {
+			nni_aio_list_remove(aio);
+			nni_aio_finish_error(aio, NNG_ECLOSED);
+		}
+		nni_posix_pfd_close(c->pfd);
+	}
+	nni_mtx_unlock(&c->mtx);
+}
+
+static void
+tcp_conn_cb(nni_posix_pfd *pfd, int events, void *arg)
+{
+	nni_tcp_conn *c = arg;
+
+	if (events & (POLLHUP | POLLERR | POLLNVAL)) {
+		nni_tcp_conn_close(c);
+		return;
+	}
+	nni_mtx_lock(&c->mtx);
+	if (events & POLLIN) {
+		tcp_conn_doread(c);
+	}
+	if (events & POLLOUT) {
+		tcp_conn_dowrite(c);
+	}
+	events = 0;
+	if (!nni_list_empty(&c->writeq)) {
+		events |= POLLOUT;
+	}
+	if (!nni_list_empty(&c->readq)) {
+		events |= POLLIN;
+	}
+	if ((!c->closed) && (events != 0)) {
+		nni_posix_pfd_arm(pfd, events);
+	}
+	nni_mtx_unlock(&c->mtx);
+}
+
+static void
+tcp_conn_cancel(nni_aio *aio, int rv)
+{
+	nni_tcp_conn *c = nni_aio_get_prov_data(aio);
+
+	nni_mtx_lock(&c->mtx);
+	if (nni_aio_list_active(aio)) {
+		nni_aio_list_remove(aio);
+		nni_aio_finish_error(aio, rv);
+	}
+	nni_mtx_unlock(&c->mtx);
+}
+
+void
+nni_tcp_conn_send(nni_tcp_conn *c, nni_aio *aio)
+{
+
+	int rv;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+	nni_mtx_lock(&c->mtx);
+
+	if ((rv = nni_aio_schedule(aio, tcp_conn_cancel, c)) != 0) {
+		nni_mtx_unlock(&c->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	nni_aio_list_append(&c->writeq, aio);
+
+	if (nni_list_first(&c->writeq) == aio) {
+		tcp_conn_dowrite(c);
+		// If we are still the first thing on the list, that
+		// means we didn't finish the job, so arm the poller to
+		// complete us.
+		if (nni_list_first(&c->writeq) == aio) {
+			nni_posix_pfd_arm(c->pfd, POLLOUT);
+		}
+	}
+	nni_mtx_unlock(&c->mtx);
+}
+
+void
+nni_tcp_conn_recv(nni_tcp_conn *c, nni_aio *aio)
+{
+	int rv;
+
+	if (nni_aio_begin(aio) != 0) {
+		return;
+	}
+	nni_mtx_lock(&c->mtx);
+
+	if ((rv = nni_aio_schedule(aio, tcp_conn_cancel, c)) != 0) {
+		nni_mtx_unlock(&c->mtx);
+		nni_aio_finish_error(aio, rv);
+		return;
+	}
+	nni_aio_list_append(&c->readq, aio);
+
+	// If we are only job on the list, go ahead and try to do an
+	// immediate transfer. This allows for faster completions in
+	// many cases.  We also need not arm a list if it was already
+	// armed.
+	if (nni_list_first(&c->readq) == aio) {
+		tcp_conn_doread(c);
+		// If we are still the first thing on the list, that
+		// means we didn't finish the job, so arm the poller to
+		// complete us.
+		if (nni_list_first(&c->readq) == aio) {
+			nni_posix_pfd_arm(c->pfd, POLLIN);
+		}
+	}
+	nni_mtx_unlock(&c->mtx);
+}
+
+int
+nni_tcp_conn_peername(nni_tcp_conn *c, nni_sockaddr *sa)
+{
+	struct sockaddr_storage ss;
+	socklen_t               sslen = sizeof(ss);
+	int                     fd    = nni_posix_pfd_fd(c->pfd);
+
+	if (getpeername(fd, (void *) &ss, &sslen) != 0) {
+		return (nni_plat_errno(errno));
+	}
+	return (nni_posix_sockaddr2nn(sa, &ss));
+}
+
+int
+nni_tcp_conn_sockname(nni_tcp_conn *c, nni_sockaddr *sa)
+{
+	struct sockaddr_storage ss;
+	socklen_t               sslen = sizeof(ss);
+	int                     fd    = nni_posix_pfd_fd(c->pfd);
+
+	if (getsockname(fd, (void *) &ss, &sslen) != 0) {
+		return (nni_plat_errno(errno));
+	}
+	return (nni_posix_sockaddr2nn(sa, &ss));
+}
+
+int
+nni_tcp_conn_set_keepalive(nni_tcp_conn *c, bool keep)
+{
+	int val = keep ? 1 : 0;
+	int fd  = nni_posix_pfd_fd(c->pfd);
+
+	if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val)) != 0) {
+		return (nni_plat_errno(errno));
+	}
+	return (0);
+}
+
+int
+nni_tcp_conn_set_nodelay(nni_tcp_conn *c, bool nodelay)
+{
+
+	int val = nodelay ? 1 : 0;
+	int fd  = nni_posix_pfd_fd(c->pfd);
+
+	if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &val, sizeof(val)) != 0) {
+		return (nni_plat_errno(errno));
+	}
+	return (0);
+}
+
+int
+nni_posix_tcp_conn_init(nni_tcp_conn **cp, nni_posix_pfd *pfd)
+{
+	nni_tcp_conn *c;
+
+	if ((c = NNI_ALLOC_STRUCT(c)) == NULL) {
+		return (NNG_ENOMEM);
+	}
+
+	c->closed = false;
+	c->pfd    = pfd;
+
+	nni_mtx_init(&c->mtx);
+	nni_aio_list_init(&c->readq);
+	nni_aio_list_init(&c->writeq);
+
+	*cp = c;
+	return (0);
+}
+
+void
+nni_posix_tcp_conn_start(nni_tcp_conn *c)
+{
+	nni_posix_pfd_set_cb(c->pfd, tcp_conn_cb, c);
+}
+
+void
+nni_tcp_conn_fini(nni_tcp_conn *c)
+{
+	nni_tcp_conn_close(c);
+	nni_posix_pfd_fini(c->pfd);
+	c->pfd = NULL;
+	nni_mtx_fini(&c->mtx);
+
+	NNI_FREE_STRUCT(c);
+}
+
+#endif // NNG_PLATFORM_POSIX