Fix close related races (POSIX close is a PITA).

6 files changed, 83 insertions, 17 deletions

diff --git a/src/core/platform.h b/src/core/platform.h
index 3292d6f9..afc4c06c 100644
--- a/src/core/platform.h
+++ b/src/core/platform.h
@@ -60,10 +60,10 @@ extern void *nni_alloc(size_t);
 // Most implementations can just call free() here.
 extern void nni_free(void *, size_t);
 
-typedef struct nni_plat_mtx	nni_plat_mtx;
-typedef struct nni_plat_cv	nni_plat_cv;
-typedef struct nni_plat_thr	nni_plat_thr;
-typedef struct nni_plat_tcpsock nni_plat_tcpsock;
+typedef struct nni_plat_mtx		nni_plat_mtx;
+typedef struct nni_plat_cv		nni_plat_cv;
+typedef struct nni_plat_thr		nni_plat_thr;
+typedef struct nni_plat_tcpsock		nni_plat_tcpsock;
 
 // Mutex handling.
 
@@ -174,8 +174,19 @@ extern const char *nni_plat_strerror(int);
 // returned on dual stack machines.
 extern int nni_plat_lookup_host(const char *, nni_sockaddr *, int);
 
-// nni_plat_tcp_close just closes a TCP socket.
-extern void nni_plat_tcp_close(nni_plat_tcpsock *);
+// nni_plat_tcp_init initializes the socket, for example it can
+// set underlying file descriptors to -1, etc.
+extern void nni_plat_tcp_init(nni_plat_tcpsock *);
+
+// nni_plat_tcp_fini just closes a TCP socket, and releases any related
+// resources.
+extern void nni_plat_tcp_fini(nni_plat_tcpsock *);
+
+// nni_plat_tcp_shutdown performs a shutdown of the socket.  For
+// BSD sockets, this closes both sides of the TCP connection gracefully,
+// but the underlying file descriptor is left open.  (This part is critical
+// to prevention of close() related races.)
+extern void nni_plat_tcp_shutdown(nni_plat_tcpsock *);
 
 // nni_plat_tcp_listen creates a TCP socket in listening mode, bound
 // to the specified address.  Note that nni_plat_tcpsock should be defined
diff --git a/src/platform/posix/posix_impl.h b/src/platform/posix/posix_impl.h
index 2b72f574..ed52b6fb 100644
--- a/src/platform/posix/posix_impl.h
+++ b/src/platform/posix/posix_impl.h
@@ -35,13 +35,16 @@ extern int nni_plat_errno(int);
 
 #ifdef PLATFORM_POSIX_NET
 struct nni_plat_tcpsock {
-	int fd;
+	int	fd;
+	int	devnull; // used for shutting down blocking accept()
 };
 #endif
 
 // Define types that this platform uses.
 #ifdef PLATFORM_POSIX_THREAD
 
+extern int nni_plat_devnull;    // open descriptor on /dev/null
+
 #include <pthread.h>
 
 // These types are provided for here, to permit them to be directly inlined
diff --git a/src/platform/posix/posix_net.c b/src/platform/posix/posix_net.c
index d0cfb49c..151e14f1 100644
--- a/src/platform/posix/posix_net.c
+++ b/src/platform/posix/posix_net.c
@@ -231,12 +231,35 @@ nni_plat_tcp_setopts(int fd)
 
 
 void
-nni_plat_tcp_close(nni_plat_tcpsock *s)
+nni_plat_tcp_init(nni_plat_tcpsock *s)
 {
-	(void) close(s->fd);
 	s->fd = -1;
 }
 
+
+void
+nni_plat_tcp_fini(nni_plat_tcpsock *s)
+{
+	if (s->fd != -1) {
+		(void) close(s->fd);
+		s->fd = -1;
+	}
+}
+
+
+void
+nni_plat_tcp_shutdown(nni_plat_tcpsock *s)
+{
+	if (s->fd != -1) {
+		(void) shutdown(s->fd, SHUT_RDWR);
+		// This causes the equivalent of a close.  Hopefully waking
+		// up anything that didn't get the hint with the shutdown.
+		// (macOS does not see the shtudown).
+		(void) dup2(nni_plat_devnull, s->fd);
+	}
+}
+
+
 // nni_plat_tcp_bind creates a file descriptor bound to the given address.
 // This basically does the equivalent of socket, bind, and listen.  We have
 // chosen a default value for the listen backlog of 128, which should be
@@ -257,7 +280,7 @@ nni_plat_tcp_listen(nni_plat_tcpsock *s, const nni_sockaddr *addr)
 	}
 
 #ifdef SOCK_CLOEXEC
-	fd = socket(ss.ss_family, SOCK_STREAM, SOCK_CLOEXEC);
+	fd = socket(ss.ss_family, SOCK_STREAM | SOCK_CLOEXEC, 0);
 #else
 	fd = socket(ss.ss_family, SOCK_STREAM, 0);
 #endif
@@ -305,7 +328,7 @@ nni_plat_tcp_connect(nni_plat_tcpsock *s, const nni_sockaddr *addr,
 	}
 
 #ifdef  SOCK_CLOEXEC
-	fd = socket(ss.ss_family, SOCK_STREAM, SOCK_CLOEXEC);
+	fd = socket(ss.ss_family, SOCK_STREAM | SOCK_CLOEXEC, 0);
 #else
 	fd = socket(ss.ss_family, SOCK_STREAM, 0);
 #endif
@@ -346,9 +369,9 @@ nni_plat_tcp_accept(nni_plat_tcpsock *s, nni_plat_tcpsock *server)
 
 	for (;;) {
 #ifdef NNG_USE_ACCEPT4
-		fd = accept4(server, NULL, NULL, SOCK_CLOEXEC);
+		fd = accept4(server->fd, NULL, NULL, SOCK_CLOEXEC);
 		if ((fd < 0) && ((errrno == ENOSYS) || (errno == ENOTSUP))) {
-			fd = accept(server, NULL, NULL);
+			fd = accept(server->fd, NULL, NULL);
 		}
 #else
 		fd = accept(server->fd, NULL, NULL);
@@ -359,6 +382,9 @@ nni_plat_tcp_accept(nni_plat_tcpsock *s, nni_plat_tcpsock *server)
 				// These are not fatal errors, keep trying
 				continue;
 			}
+			if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
+				continue;
+			}
 			return (nni_plat_errno(errno));
 		} else {
 			break;
diff --git a/src/platform/posix/posix_thread.c b/src/platform/posix/posix_thread.c
index 02f0b058..00419aac 100644
--- a/src/platform/posix/posix_thread.c
+++ b/src/platform/posix/posix_thread.c
@@ -19,6 +19,7 @@
 #include <stdlib.h>
 #include <sys/types.h>
 #include <unistd.h>
+#include <fcntl.h>
 
 static pthread_mutex_t nni_plat_lock = PTHREAD_MUTEX_INITIALIZER;
 static int nni_plat_inited = 0;
@@ -28,6 +29,11 @@ static int nni_plat_next = 0;
 pthread_condattr_t nni_cvattr;
 pthread_mutexattr_t nni_mxattr;
 
+// We open a /dev/null file descriptor so that we can dup2() it to
+// cause MacOS X to wakeup.  This gives us a "safe" close semantic.
+
+int nni_plat_devnull = -1;
+
 uint32_t
 nni_plat_nextid(void)
 {
@@ -228,30 +234,39 @@ nni_plat_init(int (*helper)(void))
 	if (nni_plat_inited) {
 		return (0);     // fast path
 	}
+
+	if ((nni_plat_devnull = open("/dev/null", O_RDONLY)) < 0) {
+		return (nni_plat_errno(errno));
+	}
 	pthread_mutex_lock(&nni_plat_lock);
 	if (nni_plat_inited) {        // check again under the lock to be sure
 		pthread_mutex_unlock(&nni_plat_lock);
+		(void) close(nni_plat_devnull);
 		return (0);
 	}
 	if (pthread_condattr_init(&nni_cvattr) != 0) {
 		pthread_mutex_unlock(&nni_plat_lock);
+		(void) close(nni_plat_devnull);
 		return (NNG_ENOMEM);
 	}
 #if !defined(NNG_USE_GETTIMEOFDAY) && NNG_USE_CLOCKID != CLOCK_REALTIME
 	if (pthread_condattr_setclock(&nni_cvattr, NNG_USE_CLOCKID) != 0) {
 		pthread_mutex_unlock(&nni_plat_lock);
+		(void) close(nni_plat_devnull);
 		return (NNG_ENOMEM);
 	}
 #endif
 
 	if (pthread_mutexattr_init(&nni_mxattr) != 0) {
 		pthread_mutex_unlock(&nni_plat_lock);
+		(void) close(nni_plat_devnull);
 		return (NNG_ENOMEM);
 	}
 
 	rv = pthread_mutexattr_settype(&nni_mxattr, PTHREAD_MUTEX_ERRORCHECK);
 	if (rv != 0) {
 		pthread_mutex_unlock(&nni_plat_lock);
+		(void) close(nni_plat_devnull);
 		return (NNG_ENOMEM);
 	}
 
@@ -276,6 +291,7 @@ nni_plat_init(int (*helper)(void))
 
 	if (pthread_atfork(NULL, NULL, nni_atfork_child) != 0) {
 		pthread_mutex_unlock(&nni_plat_lock);
+		(void) close(nni_plat_devnull);
 		return (NNG_ENOMEM);
 	}
 	if ((rv = helper()) == 0) {
@@ -294,6 +310,8 @@ nni_plat_fini(void)
 	if (nni_plat_inited) {
 		pthread_mutexattr_destroy(&nni_mxattr);
 		pthread_condattr_destroy(&nni_cvattr);
+		(void) close(nni_plat_devnull);
+		nni_plat_devnull = -1;
 		nni_plat_inited = 0;
 	}
 	pthread_mutex_unlock(&nni_plat_lock);
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index b9981c3c..abd6e25c 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -139,6 +139,7 @@ static int
 nni_req_pipe_add(void *arg)
 {
 	nni_req_pipe *rp = arg;
+
 	if (nni_pipe_peer(rp->pipe) != NNG_PROTO_REP) {
 		return (NNG_EPROTO);
 	}
diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c
index 00d9a2f3..249c2329 100644
--- a/src/transport/tcp/tcp.c
+++ b/src/transport/tcp/tcp.c
@@ -55,7 +55,7 @@ nni_tcp_pipe_close(void *arg)
 {
 	nni_tcp_pipe *pipe = arg;
 
-	nni_plat_tcp_close(&pipe->fd);
+	nni_plat_tcp_shutdown(&pipe->fd);
 }
 
 
@@ -64,6 +64,7 @@ nni_tcp_pipe_destroy(void *arg)
 {
 	nni_tcp_pipe *pipe = arg;
 
+	nni_plat_tcp_fini(&pipe->fd);
 	NNI_FREE_STRUCT(pipe);
 }
 
@@ -176,6 +177,7 @@ nni_tcp_ep_init(void **epp, const char *url, uint16_t proto)
 	ep->closed = 0;
 	ep->proto = proto;
 	ep->ipv4only = 0;
+	nni_plat_tcp_init(&ep->fd);
 
 	(void) snprintf(ep->addr, sizeof (ep->addr), "%s", url);
 
@@ -189,6 +191,7 @@ nni_tcp_ep_fini(void *arg)
 {
 	nni_tcp_ep *ep = arg;
 
+	nni_plat_tcp_fini(&ep->fd);
 	NNI_FREE_STRUCT(ep);
 }
 
@@ -198,7 +201,7 @@ nni_tcp_ep_close(void *arg)
 {
 	nni_tcp_ep *ep = arg;
 
-	nni_plat_tcp_close(&ep->fd);
+	nni_plat_tcp_shutdown(&ep->fd);
 }
 
 
@@ -340,6 +343,7 @@ nni_tcp_ep_connect(void *arg, void **pipep)
 	if ((pipe = NNI_ALLOC_STRUCT(pipe)) == NULL) {
 		return (NNG_ENOMEM);
 	}
+	nni_plat_tcp_init(&pipe->fd);
 	pipe->proto = ep->proto;
 
 	// Port is in the same place for both v4 and v6.
@@ -353,7 +357,8 @@ nni_tcp_ep_connect(void *arg, void **pipep)
 	}
 
 	if ((rv = nni_tcp_negotiate(pipe)) != 0) {
-		nni_plat_tcp_close(&pipe->fd);
+		nni_plat_tcp_shutdown(&pipe->fd);
+		nni_plat_tcp_fini(&pipe->fd);
 		NNI_FREE_STRUCT(pipe);
 		return (rv);
 	}
@@ -405,13 +410,15 @@ nni_tcp_ep_accept(void *arg, void **pipep)
 		return (NNG_ENOMEM);
 	}
 	pipe->proto = ep->proto;
+	nni_plat_tcp_init(&pipe->fd);
 
 	if ((rv = nni_plat_tcp_accept(&pipe->fd, &ep->fd)) != 0) {
 		NNI_FREE_STRUCT(pipe);
 		return (rv);
 	}
 	if ((rv = nni_tcp_negotiate(pipe)) != 0) {
-		nni_plat_tcp_close(&pipe->fd);
+		nni_plat_tcp_shutdown(&pipe->fd);
+		nni_plat_tcp_fini(&pipe->fd);
 		NNI_FREE_STRUCT(pipe);
 		return (rv);
 	}