tcp: flatten the listener implementation

The endpoints both use a nesting level for some common code and some
platform dependent code.  But the common code isn't that much and we
have similar patterns for e.g. IPC.

This avoids a layer of indirection in the structs, and extra allocations.
The payoff will be even larger for the dialers, but that is next.
(Dialers are more complicated because of DNS.)

2 files changed, 180 insertions, 101 deletions

diff --git a/src/platform/windows/win_tcp.h b/src/platform/windows/win_tcp.h
index d151b9a3..655a969c 100644
--- a/src/platform/windows/win_tcp.h
+++ b/src/platform/windows/win_tcp.h
@@ -17,27 +17,27 @@
 #include "core/nng_impl.h"
 
 struct nni_tcp_conn {
-	nng_stream        ops;
-	SOCKET            s;
-	nni_win_io        recv_io;
-	nni_win_io        send_io;
-	nni_win_io        conn_io;
-	nni_list          recv_aios;
-	nni_list          send_aios;
-	nni_aio          *conn_aio;
-	SOCKADDR_STORAGE  sockname;
-	SOCKADDR_STORAGE  peername;
-	nni_tcp_dialer   *dialer;
-	nni_tcp_listener *listener;
-	int               recv_rv;
-	int               send_rv;
-	int               conn_rv;
-	bool              closed;
-	char              buf[512]; // to hold acceptex results
-	bool              sending;
-	bool              recving;
-	nni_mtx           mtx;
-	nni_cv            cv;
+	nng_stream       ops;
+	SOCKET           s;
+	nni_win_io       recv_io;
+	nni_win_io       send_io;
+	nni_win_io       conn_io;
+	nni_list         recv_aios;
+	nni_list         send_aios;
+	nni_aio         *conn_aio;
+	SOCKADDR_STORAGE sockname;
+	SOCKADDR_STORAGE peername;
+	nni_tcp_dialer  *dialer;
+	void            *listener;
+	int              recv_rv;
+	int              send_rv;
+	int              conn_rv;
+	bool             closed;
+	char             buf[512]; // to hold acceptex results
+	bool             sending;
+	bool             recving;
+	nni_mtx          mtx;
+	nni_cv           cv;
 };
 
 extern int nni_win_tcp_init(nni_tcp_conn **, SOCKET);
diff --git a/src/platform/windows/win_tcplisten.c b/src/platform/windows/win_tcplisten.c
index e7c7a966..d4ff7cbc 100644
--- a/src/platform/windows/win_tcplisten.c
+++ b/src/platform/windows/win_tcplisten.c
@@ -17,7 +17,9 @@
 
 #include "win_tcp.h"
 
-struct nni_tcp_listener {
+typedef struct tcp_listener {
+	nng_stream_listener       ops;
+	nng_sockaddr              sa;
 	SOCKET                    s;
 	nni_list                  aios;
 	bool                      closed;
@@ -33,12 +35,12 @@ struct nni_tcp_listener {
 	nni_win_io                accept_io;
 	int                       accept_rv;
 	nni_tcp_conn             *pend_conn;
-};
+} tcp_listener;
 
 // tcp_listener_funcs looks up function pointers we need for advanced accept
 // functionality on Windows.  Windows is weird.
 static int
-tcp_listener_funcs(nni_tcp_listener *l)
+tcp_listener_funcs(tcp_listener *l)
 {
 	static SRWLOCK                   lock = SRWLOCK_INIT;
 	static LPFN_ACCEPTEX             acceptex;
@@ -82,15 +84,16 @@ tcp_listener_funcs(nni_tcp_listener *l)
 	return (0);
 }
 
-static void tcp_listener_accepted(nni_tcp_listener *l);
-static void tcp_listener_doaccept(nni_tcp_listener *l);
+static void tcp_listener_accepted(tcp_listener *l);
+static void tcp_listener_doaccept(tcp_listener *l);
+static void tcp_listener_free(void *arg);
 
 static void
 tcp_accept_cb(nni_win_io *io, int rv, size_t cnt)
 {
-	nni_tcp_listener *l = io->ptr;
-	nni_aio          *aio;
-	nni_tcp_conn     *c;
+	tcp_listener *l = io->ptr;
+	nni_aio      *aio;
+	nni_tcp_conn *c;
 
 	NNI_ARG_UNUSED(cnt);
 
@@ -120,39 +123,13 @@ tcp_accept_cb(nni_win_io *io, int rv, size_t cnt)
 	nni_mtx_unlock(&l->mtx);
 }
 
-int
-nni_tcp_listener_init(nni_tcp_listener **lp)
-{
-	nni_tcp_listener *l;
-	int               rv;
-
-	if ((l = NNI_ALLOC_STRUCT(l)) == NULL) {
-		return (NNG_ENOMEM);
-	}
-	ZeroMemory(l, sizeof(*l));
-	nni_mtx_init(&l->mtx);
-	nni_aio_list_init(&l->aios);
-	nni_win_io_init(&l->accept_io, tcp_accept_cb, l);
-	l->accept_rv = 0;
-	if ((rv = tcp_listener_funcs(l)) != 0) {
-		nni_tcp_listener_fini(l);
-		return (rv);
-	}
-
-	// We assume these defaults -- not everyone will agree, but anyone
-	// can change them.
-	l->keepalive = false;
-	l->nodelay   = true;
-
-	*lp = l;
-	return (0);
-}
-
-void
-nni_tcp_listener_close(nni_tcp_listener *l)
+static void
+tcp_listener_close(void *arg)
 {
 	nni_aio      *aio;
 	nni_tcp_conn *conn;
+	tcp_listener *l = arg;
+
 	nni_mtx_lock(&l->mtx);
 	if (!l->closed) {
 		l->closed = true;
@@ -173,21 +150,24 @@ nni_tcp_listener_close(nni_tcp_listener *l)
 }
 
 static nni_reap_list tcp_listener_reap_list = {
-	.rl_offset = offsetof(nni_tcp_listener, reap),
-	.rl_func   = (nni_cb) nni_tcp_listener_fini,
+	.rl_offset = offsetof(tcp_listener, reap),
+	.rl_func   = (nni_cb) tcp_listener_free,
 };
 
-void
-nni_tcp_listener_stop(nni_tcp_listener *l)
+static void
+tcp_listener_stop(void *arg)
 {
-	nni_tcp_listener_close(l);
+	tcp_listener *l = arg;
+	tcp_listener_close(l);
 	// TODO: maybe wait for l->l_accept_io.olpd to finish?
 }
 
-void
-nni_tcp_listener_fini(nni_tcp_listener *l)
+static void
+tcp_listener_free(void *arg)
 {
-	nni_tcp_listener_close(l);
+	tcp_listener *l = arg;
+
+	tcp_listener_close(l);
 	nni_mtx_lock(&l->mtx);
 	if (l->running) {
 		nni_mtx_unlock(&l->mtx);
@@ -199,13 +179,14 @@ nni_tcp_listener_fini(nni_tcp_listener *l)
 	NNI_FREE_STRUCT(l);
 }
 
-int
-nni_tcp_listener_listen(nni_tcp_listener *l, const nni_sockaddr *sa)
+static int
+tcp_listener_listen(void *arg)
 {
-	int   rv;
-	BOOL  yes;
-	DWORD no;
-	int   len;
+	int           rv;
+	BOOL          yes;
+	DWORD         no;
+	int           len;
+	tcp_listener *l = arg;
 
 	nni_mtx_lock(&l->mtx);
 	if (l->closed) {
@@ -216,7 +197,7 @@ nni_tcp_listener_listen(nni_tcp_listener *l, const nni_sockaddr *sa)
 		nni_mtx_unlock(&l->mtx);
 		return (NNG_EBUSY);
 	}
-	if ((len = nni_win_nn2sockaddr(&l->ss, sa)) <= 0) {
+	if ((len = nni_win_nn2sockaddr(&l->ss, &l->sa)) <= 0) {
 		nni_mtx_unlock(&l->mtx);
 		return (NNG_EADDRINVAL);
 	}
@@ -266,7 +247,7 @@ nni_tcp_listener_listen(nni_tcp_listener *l, const nni_sockaddr *sa)
 static void
 tcp_accept_cancel(nni_aio *aio, void *arg, int rv)
 {
-	nni_tcp_listener *l = arg;
+	tcp_listener *l = arg;
 
 	nni_mtx_lock(&l->mtx);
 	if (aio == nni_list_first(&l->aios)) {
@@ -286,7 +267,7 @@ tcp_accept_cancel(nni_aio *aio, void *arg, int rv)
 }
 
 static void
-tcp_listener_accepted(nni_tcp_listener *l)
+tcp_listener_accepted(tcp_listener *l)
 {
 	int           len1;
 	int           len2;
@@ -324,7 +305,7 @@ tcp_listener_accepted(nni_tcp_listener *l)
 }
 
 static void
-tcp_listener_doaccept(nni_tcp_listener *l)
+tcp_listener_doaccept(tcp_listener *l)
 {
 	nni_aio      *aio;
 	SOCKET        s;
@@ -377,10 +358,10 @@ tcp_listener_doaccept(nni_tcp_listener *l)
 	l->running = false;
 }
 
-void
-nni_tcp_listener_accept(nni_tcp_listener *l, nni_aio *aio)
+static void
+tcp_listener_accept(void *arg, nni_aio *aio)
 {
-	int rv;
+	tcp_listener *l = arg;
 
 	nni_aio_reset(aio);
 
@@ -390,14 +371,17 @@ nni_tcp_listener_accept(nni_tcp_listener *l, nni_aio *aio)
 		nni_aio_finish_error(aio, NNG_ESTATE);
 		return;
 	}
-
+	if (l->closed) {
+		nni_mtx_unlock(&l->mtx);
+		nni_aio_finish_error(aio, NNG_ECLOSED);
+		return;
+	}
 	if (!nni_aio_start(aio, tcp_accept_cancel, l)) {
 		nni_mtx_unlock(&l->mtx);
 		return;
 	}
 
 	nni_aio_list_append(&l->aios, aio);
-
 	if (aio == nni_list_first(&l->aios)) {
 		tcp_listener_doaccept(l);
 	}
@@ -407,8 +391,8 @@ nni_tcp_listener_accept(nni_tcp_listener *l, nni_aio *aio)
 static int
 tcp_listener_get_locaddr(void *arg, void *buf, size_t *szp, nni_type t)
 {
-	nni_tcp_listener *l = arg;
-	nng_sockaddr      sa;
+	tcp_listener *l = arg;
+	nng_sockaddr  sa;
 	nni_mtx_lock(&l->mtx);
 	if (l->started) {
 		nni_win_sockaddr2nn(&sa, &l->ss);
@@ -422,9 +406,9 @@ tcp_listener_get_locaddr(void *arg, void *buf, size_t *szp, nni_type t)
 static int
 tcp_listener_set_nodelay(void *arg, const void *buf, size_t sz, nni_type t)
 {
-	nni_tcp_listener *l = arg;
-	int               rv;
-	bool              b;
+	tcp_listener *l = arg;
+	int           rv;
+	bool          b;
 
 	if (((rv = nni_copyin_bool(&b, buf, sz, t)) != 0) || (l == NULL)) {
 		return (rv);
@@ -438,8 +422,8 @@ tcp_listener_set_nodelay(void *arg, const void *buf, size_t sz, nni_type t)
 static int
 tcp_listener_get_nodelay(void *arg, void *buf, size_t *szp, nni_type t)
 {
-	bool              b;
-	nni_tcp_listener *l = arg;
+	bool          b;
+	tcp_listener *l = arg;
 	nni_mtx_lock(&l->mtx);
 	b = l->nodelay;
 	nni_mtx_unlock(&l->mtx);
@@ -449,9 +433,9 @@ tcp_listener_get_nodelay(void *arg, void *buf, size_t *szp, nni_type t)
 static int
 tcp_listener_set_keepalive(void *arg, const void *buf, size_t sz, nni_type t)
 {
-	nni_tcp_listener *l = arg;
-	int               rv;
-	bool              b;
+	tcp_listener *l = arg;
+	int           rv;
+	bool          b;
 
 	if (((rv = nni_copyin_bool(&b, buf, sz, t)) != 0) || (l == NULL)) {
 		return (rv);
@@ -465,14 +449,52 @@ tcp_listener_set_keepalive(void *arg, const void *buf, size_t sz, nni_type t)
 static int
 tcp_listener_get_keepalive(void *arg, void *buf, size_t *szp, nni_type t)
 {
-	bool              b;
-	nni_tcp_listener *l = arg;
+	bool          b;
+	tcp_listener *l = arg;
 	nni_mtx_lock(&l->mtx);
 	b = l->keepalive;
 	nni_mtx_unlock(&l->mtx);
 	return (nni_copyout_bool(b, buf, szp, t));
 }
 
+static int
+tcp_listener_get_port(void *arg, void *buf, size_t *szp, nni_type t)
+{
+	tcp_listener *l = arg;
+	int           rv;
+	nng_sockaddr  sa;
+	size_t        sz;
+	int           port;
+	uint8_t      *paddr;
+
+	sz = sizeof(sa);
+	rv = tcp_listener_get_locaddr(l, &sa, &sz, NNI_TYPE_SOCKADDR);
+	if (rv != 0) {
+		return (rv);
+	}
+
+	switch (sa.s_family) {
+	case NNG_AF_INET:
+		paddr = (void *) &sa.s_in.sa_port;
+		break;
+
+	case NNG_AF_INET6:
+		paddr = (void *) &sa.s_in6.sa_port;
+		break;
+
+	default:
+		paddr = NULL;
+		break;
+	}
+
+	if (paddr == NULL) {
+		return (NNG_ESTATE);
+	}
+
+	NNI_GET16(paddr, port);
+	return (nni_copyout_int(port, buf, szp, t));
+}
+
 static const nni_option tcp_listener_options[] = {
 	{
 	    .o_name = NNG_OPT_LOCADDR,
@@ -489,20 +511,77 @@ static const nni_option tcp_listener_options[] = {
 	    .o_get  = tcp_listener_get_keepalive,
 	},
 	{
+	    .o_name = NNG_OPT_TCP_BOUND_PORT,
+	    .o_get  = tcp_listener_get_port,
+	},
+	{
 	    .o_name = NULL,
 	},
 };
 
-int
-nni_tcp_listener_get(
-    nni_tcp_listener *l, const char *name, void *buf, size_t *szp, nni_type t)
+static int
+tcp_listener_get(
+    void *arg, const char *name, void *buf, size_t *szp, nni_type t)
+{
+	return (nni_getopt(tcp_listener_options, name, arg, buf, szp, t));
+}
+
+static int
+tcp_listener_set(
+    void *arg, const char *name, const void *buf, size_t sz, nni_type t)
 {
-	return (nni_getopt(tcp_listener_options, name, l, buf, szp, t));
+	return (nni_setopt(tcp_listener_options, name, arg, buf, sz, t));
+}
+
+static int
+tcp_listener_alloc_addr(nng_stream_listener **lp, const nng_sockaddr *sa)
+{
+	tcp_listener *l;
+	int           rv;
+
+	if ((l = NNI_ALLOC_STRUCT(l)) == NULL) {
+		return (NNG_ENOMEM);
+	}
+
+	nni_mtx_init(&l->mtx);
+	nni_aio_list_init(&l->aios);
+	nni_win_io_init(&l->accept_io, tcp_accept_cb, l);
+	if ((rv = tcp_listener_funcs(l)) != 0) {
+		NNI_FREE_STRUCT(l);
+		return (rv);
+	}
+
+	// We assume these defaults -- not everyone will agree, but anyone
+	// can change them.
+	l->keepalive = false;
+	l->nodelay   = true;
+	l->closed    = false;
+	l->started   = false;
+	l->nodelay   = true;
+	l->sa        = *sa;
+	l->accept_rv = 0;
+
+	l->ops.sl_free   = tcp_listener_free;
+	l->ops.sl_close  = tcp_listener_close;
+	l->ops.sl_stop   = tcp_listener_stop;
+	l->ops.sl_listen = tcp_listener_listen;
+	l->ops.sl_accept = tcp_listener_accept;
+	l->ops.sl_get    = tcp_listener_get;
+	l->ops.sl_set    = tcp_listener_set;
+
+	*lp = (void *) l;
+	return (0);
 }
 
 int
-nni_tcp_listener_set(nni_tcp_listener *l, const char *name, const void *buf,
-    size_t sz, nni_type t)
+nni_tcp_listener_alloc(nng_stream_listener **lp, const nng_url *url)
 {
-	return (nni_setopt(tcp_listener_options, name, l, buf, sz, t));
+	int          rv;
+	nng_sockaddr sa;
+
+	if ((rv = nni_url_to_address(&sa, url)) != 0) {
+		return (rv);
+	}
+
+	return (tcp_listener_alloc_addr(lp, &sa));
 }