Whoops, forgot to add the pipe implementations to git!!

4 files changed, 344 insertions, 0 deletions

diff --git a/src/platform/posix/posix_pipe.c b/src/platform/posix/posix_pipe.c
new file mode 100644
index 00000000..7f6a50cc
--- /dev/null
+++ b/src/platform/posix/posix_pipe.c
@@ -0,0 +1,138 @@
+//
+// Copyright 2017 Garrett D'Amore <garrett@damore.org>
+//
+// 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.
+//
+
+// POSIX pipes.
+#include "core/nng_impl.h"
+
+#ifdef PLATFORM_POSIX_PIPE
+
+// This implementation of notification pipes works ~everywhere on POSIX,
+// as it only relies on pipe() and non-blocking I/O.
+
+#ifdef NNG_USE_EVENTFD
+
+// Linux eventfd.  This is lighter weight than pipes, and has better semantics
+// to boot.  This is far better than say epoll().
+#include <sys/eventfd.h>
+#include <fcntl.h>
+#include <unistd.h>
+
+#ifdef EFD_CLOEXEC
+#define NNI_EVENTFD_FLAGS	EFD_CLOEXEC
+#else
+#define NNI_EVENTFD_FLAGS	0
+#endif
+
+int
+nni_plat_pipe_open(int *wfd, int *rfd)
+{
+	int fd;
+
+	if ((fd = eventfd(0, NNI_EVENTFD_FLAGS)) < 0) {
+		return (nni_plat_errno(errno));
+	}
+	(void) fcntl(fd, F_SETFD, FD_CLOEXEC);
+	(void) fcntl(fd, F_SETFL, O_NONBLOCK);
+
+	*wfd = *rfd = fd;
+	return (0);
+}
+
+
+void
+nni_plat_pipe_raise(int wfd)
+{
+	uint64_t one = 1;
+
+	(void) write(wfd, &one, sizeof (one));
+}
+
+
+void
+nni_plat_pipe_clear(int rfd)
+{
+	uint64_t val;
+
+	(void) read(rfd, &val, sizeof (val));
+}
+
+
+void
+nni_plat_pipe_close(int wfd, int rfd)
+{
+	NNI_ASSERT(wfd == rfd);
+	(void) close(wfd);
+}
+
+
+#else // NNG_USE_EVENTFD
+
+#include <unistd.h>
+#include <fcntl.h>
+
+int
+nni_plat_pipe_open(int *wfd, int *rfd)
+{
+	int fds[2];
+
+	if (pipe(fds) < 0) {
+		return (nni_plat_errno(errno));
+	}
+	*wfd = fds[0];
+	*rfd = fds[1];
+
+	(void) fcntl(fds[0], F_SETFD, FD_CLOEXEC);
+	(void) fcntl(fds[1], F_SETFD, FD_CLOEXEC);
+	(void) fcntl(fds[0], F_SETFL, O_NONBLOCK);
+	(void) fcntl(fds[1], F_SETFL, O_NONBLOCK);
+
+	return (0);
+}
+
+
+void
+nni_plat_pipe_raise(int wfd)
+{
+	char c = 1;
+
+	write(wfd, &c, 1);
+}
+
+
+void
+nni_plat_pipe_clear(int rfd)
+{
+	char buf[32];
+
+	for (;;) {
+		// Completely drain the pipe, but don't wait.  This coalesces
+		// events somewhat.
+		if (read(rfd, buf, sizeof (buf)) <= 0) {
+			return;
+		}
+	}
+}
+
+
+void
+nni_plat_pipe_close(int wfd, int rfd)
+{
+	close(wfd);
+	close(rfd);
+}
+
+
+#endif // NNG_USE_EVENTFD
+
+#else
+
+// Suppress empty symbols warnings in ranlib.
+int nni_posix_pipe_not_used = 0;
+
+#endif // PLATFORM_POSIX_PIPE
diff --git a/src/platform/windows/win_pipe.c b/src/platform/windows/win_pipe.c
new file mode 100644
index 00000000..f254d9bb
--- /dev/null
+++ b/src/platform/windows/win_pipe.c
@@ -0,0 +1,149 @@
+//
+// Copyright 2017 Garrett D'Amore <garrett@damore.org>
+//
+// 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"
+
+// Windows named pipes won't work for us; we *MUST* use sockets.  This is
+// a real sadness, but what can you do.  We use an anonymous socket bound
+// to localhost and a connected peer.
+
+#ifdef PLATFORM_WINDOWS
+
+int
+nni_plat_pipe_open(int *wfdp, int *rfdp)
+{
+	SOCKET afd = INVALID_SOCKET;
+	SOCKET rfd = INVALID_SOCKET;
+	SOCKET wfd = INVALID_SOCKET;
+
+	struct sockaddr_in addr;
+	socklen_t alen;
+	int one;
+	ULONG yes;
+	int rv;
+
+	ZeroMemory(&addr, sizeof (addr));
+
+	// Restrict our bind to the loopback address.  We bind to an
+	// ephemeral port.
+	addr.sin_family = AF_INET;
+	addr.sin_port = 0;
+	addr.sin_addr.s_addr = INADDR_LOOPBACK;
+
+	afd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+	if (afd == INVALID_SOCKET) {
+		goto fail;
+	}
+
+	// Make sure we have exclusive address use...
+	one = 1;
+	if (setsockopt(afd, SOL_SOCKET, SO_EXCLUSIVEADDRUSE,
+	    (char *) (&one), sizeof (one)) != 0) {
+		goto fail;
+	}
+	alen = sizeof (addr);
+	if (bind(afd, (struct sockaddr *) &addr, alen) != 0) {
+		goto fail;
+	}
+	// What port did we bind to?
+	if (getsockname(afd, (struct sockaddr *) &addr, &alen) != 0) {
+		goto fail;
+	}
+
+	// Minimum backlog -- we only expect one connection ever.
+	if (listen(afd, 1) != 0) {
+		goto fail;
+	}
+
+	rfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+	if (afd == INVALID_SOCKET) {
+		goto fail;
+	}
+
+	if (connect(rfd, (struct sockaddr *) &addr, alen) != 0) {
+		goto fail;
+	}
+
+	// Now we have to do the accept dance.  We don't care about the
+	// peer adddress, since know it.
+	wfd = accept(afd, NULL, 0);
+	if (wfd == INVALID_SOCKET) {
+		goto fail;
+	}
+
+	// Now that we are connected, mark everything non-blocking.
+	yes = 1;
+	if (ioctlsocket(rfd, FIONBIO, &yes) != 0) {
+		goto fail;
+	}
+	yes = 1;
+	if (ioctlsocket(wfd, FIONBIO, &yes) != 0) {
+		goto fail;
+	}
+
+	// Close the listener now that we have the connection.
+	closesocket((SOCKET) afd);
+	*rfdp = (int) rfd;
+	*wfdp = (int) wfd;
+	return (0);
+
+fail:
+	rv = nni_winsock_error(WSAGetLastError());
+	if (afd != INVALID_SOCKET) {
+		closesocket(afd);
+	}
+	if (rfd != INVALID_SOCKET) {
+		closesocket(rfd);
+	}
+	if (wfd != INVALID_SOCKET) {
+		closesocket(wfd);
+	}
+
+	return (0);
+}
+
+
+void
+nni_plat_pipe_raise(int wfd)
+{
+	char c = 1;
+
+	send((SOCKET) wfd, &c, 1, 0);
+}
+
+
+void
+nni_plat_pipe_clear(int rfd)
+{
+	char buf[32];
+
+	for (;;) {
+		// Completely drain the pipe, but don't wait.  This coalesces
+		// events somewhat.
+		if (recv((SOCKET) rfd, buf, sizeof (buf), 0) <= 0) {
+			return;
+		}
+	}
+}
+
+
+void
+nni_plat_pipe_close(int wfd, int rfd)
+{
+	closesocket((SOCKET) wfd);
+	closesocket((SOCKET) rfd);
+}
+
+
+#else
+
+// Suppress empty symbols warnings in ranlib.
+int nni_win_pipe_not_used = 0;
+
+#endif // PLATFORM_WINDOWS
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index f6751566..2b7a816e 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -64,6 +64,7 @@ add_nng_test(list 5)
 add_nng_test(platform 5)
 add_nng_test(reqrep 5)
 add_nng_test(pipeline 5)
+add_nng_test(pollfd 5)
 add_nng_test(pubsub 5)
 add_nng_test(sock 5)
 add_nng_test(survey 5)
diff --git a/tests/pollfd.c b/tests/pollfd.c
new file mode 100644
index 00000000..899547c6
--- /dev/null
+++ b/tests/pollfd.c
@@ -0,0 +1,56 @@
+//
+// Copyright 2017 Garrett D'Amore <garrett@damore.org>
+//
+// 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 "convey.h"
+#include "nng.h"
+
+#ifdef _WIN32
+#include <windows.h>
+#include <winsock2.h>
+#else
+#include <unistd.h>
+#define	INVALID_SOCKET -1
+#endif
+
+// Inproc tests.
+
+TestMain("Poll FDs", {
+
+	Convey("Given a connected pair of sockets", {
+		nng_socket s1;
+		nng_socket s2;
+
+		So(nng_open(&s1, NNG_PROTO_PAIR) == 0);
+		So(nng_open(&s2, NNG_PROTO_PAIR) == 0);
+		Reset({
+			nng_close(s1);
+			nng_close(s2);
+		})
+		So(nng_listen(s1, "inproc://yeahbaby", NULL, 0) == 0);
+		So(nng_dial(s2, "inproc://yeahbaby", NULL, NNG_FLAG_SYNCH) == 0);
+
+		Convey("We can get a recv FD", {
+			int fd;
+			size_t sz;
+
+			sz = sizeof (fd);
+			So(nng_getopt(s1, NNG_OPT_RECVFD, &fd, &sz) == 0);
+			So(fd != INVALID_SOCKET);
+		})
+
+		Convey("We can get a send FD", {
+			int fd;
+			size_t sz;
+
+			sz = sizeof (fd);
+			So(nng_getopt(s1, NNG_OPT_SENDFD, &fd, &sz) == 0);
+			So(fd != INVALID_SOCKET);
+		})
+	})
+})