Fix close-related leak of pipes.

We have seen leaks of pipes causing test failures (e.g. the Windows
IPC test) due to EADDRINUSE.  This was caused by a case where we
failed to pass the pipe up because the AIO had already been canceled,
and we didn't realize that we had oprhaned the pipe.  The fix is to
add a return value to nni_aio_finish, and verify that we did finish
properly, or if we did not then we must free the pipe ourself.  (The
zero return from nni_aio_finish indicates that it accepts ownership
of resources passed via the aio.)

11 files changed, 50 insertions, 24 deletions

diff --git a/src/core/aio.c b/src/core/aio.c
index 73c1dd9b..c6512eb4 100644
--- a/src/core/aio.c
+++ b/src/core/aio.c
@@ -206,14 +206,14 @@ nni_aio_cancel(nni_aio *aio, int rv)
 
 // I/O provider related functions.
 
-void
+int
 nni_aio_finish(nni_aio *aio, int result, size_t count)
 {
 	nni_mtx_lock(&aio->a_lk);
 	if (aio->a_flags & (NNI_AIO_DONE | NNI_AIO_FINI)) {
 		// Operation already done (canceled or timed out?)
 		nni_mtx_unlock(&aio->a_lk);
-		return;
+		return (NNG_ESTATE);
 	}
 	aio->a_flags |= NNI_AIO_DONE;
 	aio->a_result      = result;
@@ -228,6 +228,7 @@ nni_aio_finish(nni_aio *aio, int result, size_t count)
 
 	nni_taskq_dispatch(NULL, &aio->a_tqe);
 	nni_mtx_unlock(&aio->a_lk);
+	return (0);
 }
 
 void
@@ -307,7 +308,7 @@ nni_aio_expire_loop(void *arg)
 	nni_list *aios = &nni_aio_expire_aios;
 	nni_aio * aio;
 	nni_time  now;
-	int       rv;
+
 	void (*cancelfn)(nni_aio *);
 
 	NNI_ARG_UNUSED(arg);
diff --git a/src/core/aio.h b/src/core/aio.h
index 09923d7f..4f190aa1 100644
--- a/src/core/aio.h
+++ b/src/core/aio.h
@@ -104,8 +104,12 @@ extern int  nni_aio_list_active(nni_aio *);
 
 // nni_aio_finish is called by the provider when an operation is complete.
 // The provider gives the result code (0 for success, an NNG errno otherwise),
-// and the amount of data transferred (if any).
-extern void nni_aio_finish(nni_aio *, int, size_t);
+// and the amount of data transferred (if any).  If the return code is
+// non-zero, it indicates that the operation failed (usually because the aio
+// was already canceled.)  This is important for providers that need to
+// prevent resources (new pipes for example) from accidentally leaking
+// during close operations.
+extern int nni_aio_finish(nni_aio *, int, size_t);
 
 // nni_aio_cancel is used to cancel an operation.  Any pending I/O or
 // timeouts are canceled if possible, and the callback will be returned
diff --git a/src/core/endpt.c b/src/core/endpt.c
index 4ed678d7..8048de2b 100644
--- a/src/core/endpt.c
+++ b/src/core/endpt.c
@@ -397,7 +397,7 @@ nni_ep_accept_start(nni_ep *ep)
 	if (ep->ep_closed) {
 		return;
 	}
-
+	aio->a_pipe  = NULL;
 	aio->a_endpt = ep->ep_data;
 	ep->ep_ops.ep_accept(ep->ep_data, aio);
 }
diff --git a/src/core/socket.c b/src/core/socket.c
index 7ec383d4..10bc1c80 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -124,8 +124,7 @@ nni_sock_pipe_ready(nni_sock *sock, nni_pipe *pipe)
 void
 nni_sock_pipe_remove(nni_sock *sock, nni_pipe *pipe)
 {
-	void *  pdata;
-	nni_ep *ep;
+	void *pdata;
 
 	pdata = nni_pipe_get_proto_data(pipe);
 
diff --git a/src/platform/posix/posix_epdesc.c b/src/platform/posix/posix_epdesc.c
index a6de29e1..8cae2565 100644
--- a/src/platform/posix/posix_epdesc.c
+++ b/src/platform/posix/posix_epdesc.c
@@ -1,5 +1,6 @@
 //
 // Copyright 2017 Garrett D'Amore <garrett@damore.org>
+// Copyright 2017 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
@@ -72,8 +73,9 @@ nni_posix_epdesc_finish(nni_aio *aio, int rv, int newfd)
 			aio->a_pipe = pd;
 		}
 	}
-	// Abuse the count to hold our new fd.  This is only for accept.
-	nni_aio_finish(aio, rv, 0);
+	if ((nni_aio_finish(aio, rv, 0) != 0) && (rv == 0)) {
+		nni_posix_pipedesc_fini(pd);
+	}
 }
 
 static void
diff --git a/src/platform/windows/win_iocp.c b/src/platform/windows/win_iocp.c
index d0a39142..df0357c8 100644
--- a/src/platform/windows/win_iocp.c
+++ b/src/platform/windows/win_iocp.c
@@ -1,5 +1,6 @@
 //
 // Copyright 2017 Garrett D'Amore <garrett@damore.org>
+// Copyright 2017 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
@@ -213,12 +214,12 @@ nni_win_event_fini(nni_win_event *evt)
 
 			// Use provider specific cancellation.
 			evt->ops.wev_cancel(evt);
-
-			// Wait for everything to stop referencing this.
-			while (evt->flags & NNI_WIN_EVENT_RUNNING) {
-				nni_cv_wait(&evt->cv);
-			}
 		}
+		// Wait for everything to stop referencing this.
+		while (evt->flags & NNI_WIN_EVENT_RUNNING) {
+			nni_cv_wait(&evt->cv);
+		}
+
 		nni_mtx_unlock(&evt->mtx);
 	}
 
diff --git a/src/platform/windows/win_ipc.c b/src/platform/windows/win_ipc.c
index 1a672c70..85e3dfca 100644
--- a/src/platform/windows/win_ipc.c
+++ b/src/platform/windows/win_ipc.c
@@ -1,5 +1,6 @@
 //
 // Copyright 2017 Garrett D'Amore <garrett@damore.org>
+// Copyright 2017 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
@@ -333,7 +334,10 @@ nni_win_ipc_acc_finish(nni_win_event *evt, nni_aio *aio)
 	}
 
 	aio->a_pipe = pipe;
-	nni_aio_finish(aio, 0, 0);
+	// What if the pipe is already finished?
+	if (nni_aio_finish(aio, 0, 0) != 0) {
+		nni_plat_ipc_pipe_fini(pipe);
+	}
 }
 
 static void
@@ -559,8 +563,8 @@ nni_plat_ipc_ep_close(nni_plat_ipc_ep *ep)
 		break;
 
 	case NNI_EP_MODE_LISTEN:
+		nni_win_event_close(&ep->acc_ev);
 		if (ep->p != INVALID_HANDLE_VALUE) {
-			nni_win_event_close(&ep->acc_ev);
 			CloseHandle(ep->p);
 			ep->p = INVALID_HANDLE_VALUE;
 		}
diff --git a/src/platform/windows/win_net.c b/src/platform/windows/win_net.c
index d9abc670..59fc0986 100644
--- a/src/platform/windows/win_net.c
+++ b/src/platform/windows/win_net.c
@@ -1,5 +1,6 @@
 //
 // Copyright 2017 Garrett D'Amore <garrett@damore.org>
+// Copyright 2017 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
@@ -544,7 +545,9 @@ nni_win_tcp_acc_finish(nni_win_event *evt, nni_aio *aio)
 	}
 
 	aio->a_pipe = pipe;
-	nni_aio_finish(aio, 0, 0);
+	if (nni_aio_finish(aio, 0, 0) != 0) {
+		nni_plat_tcp_pipe_fini(pipe);
+	}
 }
 
 static int
diff --git a/src/transport/inproc/inproc.c b/src/transport/inproc/inproc.c
index 68490747..226a31ce 100644
--- a/src/transport/inproc/inproc.c
+++ b/src/transport/inproc/inproc.c
@@ -257,7 +257,12 @@ nni_inproc_conn_finish(nni_aio *aio, int rv)
 			}
 		}
 	}
-	nni_aio_finish(aio, rv, 0);
+	if (nni_aio_finish(aio, rv, 0) != 0) {
+		if (aio->a_pipe != NULL) {
+			nni_inproc_pipe_fini(aio->a_pipe);
+			aio->a_pipe = NULL;
+		}
+	}
 }
 
 static void
diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c
index b799dba1..862cf955 100644
--- a/src/transport/ipc/ipc.c
+++ b/src/transport/ipc/ipc.c
@@ -551,7 +551,11 @@ nni_ipc_ep_finish(nni_ipc_ep *ep)
 
 done:
 	ep->aio.a_pipe = NULL;
-	nni_aio_finish(aio, rv, 0);
+	if (nni_aio_finish(aio, rv, 0) != 0) {
+		if (rv == 0) {
+			nni_ipc_pipe_fini(pipe);
+		}
+	}
 }
 
 static void
@@ -592,7 +596,6 @@ nni_ipc_ep_accept(void *arg, nni_aio *aio)
 	}
 	ep->user_aio = aio;
 
-	// If we can't start, then its dying and we can't report either,
 	if ((rv = nni_aio_start(aio, nni_ipc_cancel_ep, ep)) != 0) {
 		ep->user_aio = NULL;
 		nni_mtx_unlock(&ep->mtx);
@@ -620,7 +623,8 @@ nni_ipc_ep_connect(void *arg, nni_aio *aio)
 
 	ep->user_aio = aio;
 
-	// If we can't start, then its dying and we can't report either,
+	// If we can't start, then its dying and we can't report
+	// either,
 	if ((rv = nni_aio_start(aio, nni_ipc_cancel_ep, ep)) != 0) {
 		ep->user_aio = NULL;
 		nni_mtx_unlock(&ep->mtx);
diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c
index dad4c46e..23472883 100644
--- a/src/transport/tcp/tcp.c
+++ b/src/transport/tcp/tcp.c
@@ -618,7 +618,11 @@ nni_tcp_ep_finish(nni_tcp_ep *ep)
 
 done:
 	ep->aio.a_pipe = NULL;
-	nni_aio_finish(aio, rv, 0);
+	if (nni_aio_finish(aio, rv, 0) != 0) {
+		if (rv == 0) {
+			nni_tcp_pipe_fini(pipe);
+		}
+	}
 }
 
 static void
@@ -657,7 +661,6 @@ nni_tcp_ep_accept(void *arg, nni_aio *aio)
 	NNI_ASSERT(ep->user_aio == NULL);
 	ep->user_aio = aio;
 
-	// If we can't start, then its dying and we can't report either,
 	if ((rv = nni_aio_start(aio, nni_tcp_cancel_ep, ep)) != 0) {
 		ep->user_aio = NULL;
 		nni_mtx_unlock(&ep->mtx);