fixes #154 underlyng TCP & IPC transports should support partial recv/send

fixes #155 POSIX TCP & IPC could avoid a lot of context switches

4 files changed, 78 insertions, 170 deletions

diff --git a/src/platform/posix/posix_epdesc.c b/src/platform/posix/posix_epdesc.c
index 0524fe30..80711ed0 100644
--- a/src/platform/posix/posix_epdesc.c
+++ b/src/platform/posix/posix_epdesc.c
@@ -279,6 +279,12 @@ nni_posix_epdesc_listen(nni_posix_epdesc *ed)
 	}
 	(void) fcntl(fd, F_SETFD, FD_CLOEXEC);
 
+#ifdef SO_NOSIGPIPE
+	// Darwin lacks MSG_NOSIGNAL, but has a socket option.
+	int one = 1;
+	(void) setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one));
+#endif
+
 	if (bind(fd, (struct sockaddr *) ss, len) < 0) {
 		nni_mtx_unlock(&ed->mtx);
 		rv = nni_plat_errno(errno);
diff --git a/src/platform/posix/posix_pipedesc.c b/src/platform/posix/posix_pipedesc.c
index d25a9d96..6ae0d752 100644
--- a/src/platform/posix/posix_pipedesc.c
+++ b/src/platform/posix/posix_pipedesc.c
@@ -66,93 +66,75 @@ nni_posix_pipedesc_doclose(nni_posix_pipedesc *pd)
 static void
 nni_posix_pipedesc_dowrite(nni_posix_pipedesc *pd)
 {
-	int           n;
-	int           rv;
-	int           i;
-	struct iovec  iovec[4];
-	struct iovec *iovp;
-	nni_aio *     aio;
+	int          n;
+	struct iovec iovec[4];
+	nni_aio *    aio;
+	int          niov;
 
 	while ((aio = nni_list_first(&pd->writeq)) != NULL) {
-		for (i = 0; i < aio->a_niov; i++) {
-			iovec[i].iov_len  = aio->a_iov[i].iov_len;
-			iovec[i].iov_base = aio->a_iov[i].iov_buf;
+		int i;
+		for (niov = 0, i = 0; i < aio->a_niov; i++) {
+			iovec[niov].iov_len  = aio->a_iov[i].iov_len;
+			iovec[niov].iov_base = aio->a_iov[i].iov_buf;
+			niov++;
+		}
+		if (niov == 0) {
+			nni_posix_pipedesc_finish(aio, NNG_EINVAL);
+			continue;
 		}
-		iovp = &iovec[0];
-		rv   = 0;
 
-		n = writev(pd->node.fd, iovp, aio->a_niov);
+		n = writev(pd->node.fd, iovec, niov);
 		if (n < 0) {
 			if ((errno == EAGAIN) || (errno == EINTR)) {
 				// Can't write more right now.  We're done
 				// on this fd for now.
 				return;
 			}
-			rv = nni_plat_errno(errno);
-
-			nni_posix_pipedesc_finish(aio, rv);
+			nni_posix_pipedesc_finish(aio, nni_plat_errno(errno));
 			nni_posix_pipedesc_doclose(pd);
 			return;
 		}
 
 		aio->a_count += n;
 
-		while (n > 0) {
-			// If we didn't write the first full iov,
-			// then we're done for now.  Record progress
-			// and return to caller.
-			if (n < aio->a_iov[0].iov_len) {
-				aio->a_iov[0].iov_buf += n;
-				aio->a_iov[0].iov_len -= n;
-				return;
-			}
-
-			// We consumed the full iovec, so just move the
-			// remaininng ones up, and decrement count handled.
-			n -= aio->a_iov[0].iov_len;
-			for (i = 1; i < aio->a_niov; i++) {
-				aio->a_iov[i - 1] = aio->a_iov[i];
-			}
-			NNI_ASSERT(aio->a_niov > 0);
-			aio->a_niov--;
-		}
-
 		// We completed the entire operation on this aioq.
 		nni_posix_pipedesc_finish(aio, 0);
 
 		// Go back to start of loop to see if there is another
-		// aioq ready for us to process.
+		// aio ready for us to process.
 	}
 }
 
 static void
 nni_posix_pipedesc_doread(nni_posix_pipedesc *pd)
 {
-	int           n;
-	int           rv;
-	int           i;
-	struct iovec  iovec[4];
-	struct iovec *iovp;
-	nni_aio *     aio;
+	int          n;
+	struct iovec iovec[4];
+	nni_aio *    aio;
+	int          niov;
 
 	while ((aio = nni_list_first(&pd->readq)) != NULL) {
-		for (i = 0; i < aio->a_niov; i++) {
-			iovec[i].iov_len  = aio->a_iov[i].iov_len;
-			iovec[i].iov_base = aio->a_iov[i].iov_buf;
+		int i;
+		for (i = 0, niov = 0; i < aio->a_niov; i++) {
+			if (aio->a_iov[i].iov_len != 0) {
+				iovec[niov].iov_len  = aio->a_iov[i].iov_len;
+				iovec[niov].iov_base = aio->a_iov[i].iov_buf;
+				niov++;
+			}
+		}
+		if (niov == 0) {
+			nni_posix_pipedesc_finish(aio, NNG_EINVAL);
+			continue;
 		}
-		iovp = &iovec[0];
-		rv   = 0;
 
-		n = readv(pd->node.fd, iovp, aio->a_niov);
+		n = readv(pd->node.fd, iovec, niov);
 		if (n < 0) {
 			if ((errno == EAGAIN) || (errno == EINTR)) {
 				// Can't write more right now.  We're done
 				// on this fd for now.
 				return;
 			}
-			rv = nni_plat_errno(errno);
-
-			nni_posix_pipedesc_finish(aio, rv);
+			nni_posix_pipedesc_finish(aio, nni_plat_errno(errno));
 			nni_posix_pipedesc_doclose(pd);
 			return;
 		}
@@ -160,36 +142,17 @@ nni_posix_pipedesc_doread(nni_posix_pipedesc *pd)
 		if (n == 0) {
 			// No bytes indicates a closed descriptor.
 			nni_posix_pipedesc_finish(aio, NNG_ECLOSED);
+			nni_posix_pipedesc_doclose(pd);
 			return;
 		}
 
 		aio->a_count += n;
 
-		while (n > 0) {
-			// If we didn't write the first full iov,
-			// then we're done for now.  Record progress
-			// and return to caller.
-			if (n < aio->a_iov[0].iov_len) {
-				aio->a_iov[0].iov_buf += n;
-				aio->a_iov[0].iov_len -= n;
-				return;
-			}
-
-			// We consumed the full iovec, so just move the
-			// remaininng ones up, and decrement count handled.
-			n -= aio->a_iov[0].iov_len;
-			for (i = 1; i < aio->a_niov; i++) {
-				aio->a_iov[i - 1] = aio->a_iov[i];
-			}
-			NNI_ASSERT(aio->a_niov > 0);
-			aio->a_niov--;
-		}
-
 		// We completed the entire operation on this aioq.
 		nni_posix_pipedesc_finish(aio, 0);
 
 		// Go back to start of loop to see if there is another
-		// aioq ready for us to process.
+		// aio ready for us to process.
 	}
 }
 
@@ -262,7 +225,17 @@ nni_posix_pipedesc_recv(nni_posix_pipedesc *pd, nni_aio *aio)
 	}
 
 	nni_aio_list_append(&pd->readq, aio);
-	nni_posix_pollq_arm(&pd->node, POLLIN);
+	// 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(&pd->readq) == aio) {
+		nni_posix_pipedesc_doread(pd);
+		// 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(&pd->readq) == aio) {
+			nni_posix_pollq_arm(&pd->node, POLLIN);
+		}
+	}
 	nni_mtx_unlock(&pd->mtx);
 }
 
@@ -283,7 +256,14 @@ nni_posix_pipedesc_send(nni_posix_pipedesc *pd, nni_aio *aio)
 	}
 
 	nni_aio_list_append(&pd->writeq, aio);
-	nni_posix_pollq_arm(&pd->node, POLLOUT);
+	if (nni_list_first(&pd->writeq) == aio) {
+		nni_posix_pipedesc_dowrite(pd);
+		// 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(&pd->writeq) == aio) {
+			nni_posix_pollq_arm(&pd->node, POLLOUT);
+		}
+	}
 	nni_mtx_unlock(&pd->mtx);
 }
 
diff --git a/src/platform/windows/win_ipc.c b/src/platform/windows/win_ipc.c
index c1e2cd31..68a6ea2c 100644
--- a/src/platform/windows/win_ipc.c
+++ b/src/platform/windows/win_ipc.c
@@ -59,11 +59,10 @@ nni_win_ipc_pipe_start(nni_win_event *evt, nni_aio *aio)
 	BOOL               ok;
 	int                rv;
 	nni_plat_ipc_pipe *pipe = evt->ptr;
+	int                idx;
 
 	NNI_ASSERT(aio != NULL);
 	NNI_ASSERT(aio->a_niov > 0);
-	NNI_ASSERT(aio->a_iov[0].iov_len > 0);
-	NNI_ASSERT(aio->a_iov[0].iov_buf != NULL);
 
 	if (pipe->p == INVALID_HANDLE_VALUE) {
 		evt->status = NNG_ECLOSED;
@@ -71,13 +70,20 @@ nni_win_ipc_pipe_start(nni_win_event *evt, nni_aio *aio)
 		return (1);
 	}
 
+	idx = 0;
+	while ((idx < aio->a_niov) && (aio->a_iov[idx].iov_len == 0)) {
+		idx++;
+	}
+	NNI_ASSERT(idx < aio->a_niov);
 	// Now start a transfer.  We assume that only one send can be
 	// outstanding on a pipe at a time.  This is important to avoid
 	// scrambling the data anyway.  Note that Windows named pipes do
 	// not appear to support scatter/gather, so we have to process
 	// each element in turn.
-	buf = aio->a_iov[0].iov_buf;
-	len = (DWORD) aio->a_iov[0].iov_len;
+	buf = aio->a_iov[idx].iov_buf;
+	len = (DWORD) aio->a_iov[idx].iov_len;
+	NNI_ASSERT(buf != NULL);
+	NNI_ASSERT(len != 0);
 
 	// We limit ourselves to writing 16MB at a time.  Named Pipes
 	// on Windows have limits of between 31 and 64MB.
@@ -115,50 +121,7 @@ nni_win_ipc_pipe_cancel(nni_win_event *evt)
 static void
 nni_win_ipc_pipe_finish(nni_win_event *evt, nni_aio *aio)
 {
-	int   rv;
-	DWORD cnt;
-
-	cnt = evt->count;
-	if ((rv = evt->status) == 0) {
-
-		int i;
-		aio->a_count += cnt;
-
-		while (cnt > 0) {
-			// If we didn't transfer the first full iov,
-			// then we're done for now.  Record progress
-			// and move on.
-			if (cnt < aio->a_iov[0].iov_len) {
-				aio->a_iov[0].iov_len -= cnt;
-				aio->a_iov[0].iov_buf =
-				    (char *) aio->a_iov[0].iov_buf + cnt;
-				break;
-			}
-
-			// We consumed the full iov, so just move the
-			// remaining ones up, and decrement count handled.
-			cnt -= aio->a_iov[0].iov_len;
-			for (i = 1; i < aio->a_niov; i++) {
-				aio->a_iov[i - 1] = aio->a_iov[i];
-			}
-			NNI_ASSERT(aio->a_niov > 0);
-			aio->a_niov--;
-		}
-
-		while (aio->a_niov > 0) {
-			// If we have more to do, submit it!
-			if (aio->a_iov[0].iov_len > 0) {
-				nni_win_event_resubmit(evt, aio);
-				return;
-			}
-			for (i = 1; i < aio->a_niov; i++) {
-				aio->a_iov[i - 1] = aio->a_iov[i];
-			}
-		}
-	}
-
-	// All done; hopefully successfully.
-	nni_aio_finish(aio, rv, aio->a_count);
+	nni_aio_finish(aio, evt->status, evt->count);
 }
 
 static int
diff --git a/src/platform/windows/win_tcp.c b/src/platform/windows/win_tcp.c
index 375a72c3..01818af4 100644
--- a/src/platform/windows/win_tcp.c
+++ b/src/platform/windows/win_tcp.c
@@ -112,12 +112,13 @@ nni_win_tcp_pipe_start(nni_win_event *evt, nni_aio *aio)
 	NNI_ASSERT(aio->a_iov[0].iov_len > 0);
 	NNI_ASSERT(aio->a_iov[0].iov_buf != NULL);
 
-	niov = aio->a_niov;
-
 	// Put the AIOs in Windows form.
-	for (i = 0; i < aio->a_niov; i++) {
-		iov[i].buf = aio->a_iov[i].iov_buf;
-		iov[i].len = (ULONG) aio->a_iov[i].iov_len;
+	for (niov = 0, i = 0; i < aio->a_niov; i++) {
+		if (aio->a_iov[i].iov_len != 0) {
+			iov[niov].buf = aio->a_iov[i].iov_buf;
+			iov[niov].len = (ULONG) aio->a_iov[i].iov_len;
+			niov++;
+		}
 	}
 
 	if ((s = pipe->s) == INVALID_SOCKET) {
@@ -162,49 +163,7 @@ nni_win_tcp_pipe_cancel(nni_win_event *evt)
 static void
 nni_win_tcp_pipe_finish(nni_win_event *evt, nni_aio *aio)
 {
-	int    rv;
-	size_t cnt;
-
-	cnt = evt->count;
-	if ((rv = evt->status) == 0) {
-		int i;
-		aio->a_count += cnt;
-
-		while (cnt > 0) {
-			// If we didn't transfer the first full iov,
-			// then we're done for now.  Record progress
-			// and move on.
-			if (cnt < aio->a_iov[0].iov_len) {
-				aio->a_iov[0].iov_len -= cnt;
-				aio->a_iov[0].iov_buf =
-				    (char *) aio->a_iov[0].iov_buf + cnt;
-				break;
-			}
-
-			// We consumed the full iov, so just move the
-			// remaining ones up, and decrement count handled.
-			cnt -= aio->a_iov[0].iov_len;
-			for (i = 1; i < aio->a_niov; i++) {
-				aio->a_iov[i - 1] = aio->a_iov[i];
-			}
-			NNI_ASSERT(aio->a_niov > 0);
-			aio->a_niov--;
-		}
-
-		while (aio->a_niov > 0) {
-			// If we have more to do, submit it!
-			if (aio->a_iov[0].iov_len > 0) {
-				nni_win_event_resubmit(evt, aio);
-				return;
-			}
-			for (i = 1; i < aio->a_niov; i++) {
-				aio->a_iov[i - 1] = aio->a_iov[i];
-			}
-		}
-	}
-
-	// All done; hopefully successfully.
-	nni_aio_finish(aio, rv, aio->a_count);
+	nni_aio_finish(aio, evt->status, evt->count);
 }
 
 static int