Don't prematurely return short completions on blocking sockets.

Always requeue an AIO job at the head of the socket buffer's queue if
sosend() or soreceive() returns EWOULDBLOCK on a blocking socket.
Previously, requests were only requeued if they returned EWOULDBLOCK
and completed no data.  Now after a partial completion on a blocking
socket the request is queued and the remaining request is retried when
the socket is ready.  This allows writes larger than the currently
available space on a blocking socket to fully complete.  Reads on a
blocking socket that satifsy the low watermark can still return a short
read (same as read()).

In order to track previously completed data, the internal 'status'
field of the AIO job is used to store the amount of previously
computed data.

Non-blocking sockets continue to return short completions for both
reads and writes.

Add a test for a "large" AIO write on a blocking socket that writes
twice the socket buffer size to a UNIX domain socket.

Sponsored by:	Chelsio Communications

1 files changed, 65 insertions, 0 deletions

diff --git a/tests/sys/aio/aio_test.c b/tests/sys/aio/aio_test.c
index 1b10088..c81b8d6 100644
--- a/tests/sys/aio/aio_test.c
+++ b/tests/sys/aio/aio_test.c
@@ -781,6 +781,70 @@ ATF_TC_BODY(aio_socket_two_reads, tc)
 	close(s[0]);
 }
 
+/*
+ * This test ensures that aio_write() on a blocking socket of a "large"
+ * buffer does not return a short completion.
+ */
+ATF_TC_WITHOUT_HEAD(aio_socket_blocking_short_write);
+ATF_TC_BODY(aio_socket_blocking_short_write, tc)
+{
+	struct aiocb iocb, *iocbp;
+	char *buffer[2];
+	ssize_t done;
+	int buffer_size, sb_size;
+	socklen_t len;
+	int s[2];
+
+	ATF_REQUIRE_KERNEL_MODULE("aio");
+
+	ATF_REQUIRE(socketpair(PF_UNIX, SOCK_STREAM, 0, s) != -1);
+
+	len = sizeof(sb_size);
+	ATF_REQUIRE(getsockopt(s[0], SOL_SOCKET, SO_RCVBUF, &sb_size, &len) !=
+	    -1);
+	ATF_REQUIRE(len == sizeof(sb_size));
+	buffer_size = sb_size;
+
+	ATF_REQUIRE(getsockopt(s[1], SOL_SOCKET, SO_SNDBUF, &sb_size, &len) !=
+	    -1);
+	ATF_REQUIRE(len == sizeof(sb_size));
+	if (sb_size > buffer_size)
+		buffer_size = sb_size;
+
+	/*
+	 * Use twice the size of the MAX(receive buffer, send buffer)
+	 * to ensure that the write is split up into multiple writes
+	 * internally.
+	 */
+	buffer_size *= 2;
+
+	buffer[0] = malloc(buffer_size);
+	ATF_REQUIRE(buffer[0] != NULL);
+	buffer[1] = malloc(buffer_size);
+	ATF_REQUIRE(buffer[1] != NULL);
+
+	srandomdev();
+	aio_fill_buffer(buffer[1], buffer_size, random());
+
+	memset(&iocb, 0, sizeof(iocb));
+	iocb.aio_fildes = s[1];
+	iocb.aio_buf = buffer[1];
+	iocb.aio_nbytes = buffer_size;
+	ATF_REQUIRE(aio_write(&iocb) == 0);
+
+	done = recv(s[0], buffer[0], buffer_size, MSG_WAITALL);
+	ATF_REQUIRE(done == buffer_size);
+
+	done = aio_waitcomplete(&iocbp, NULL);
+	ATF_REQUIRE(iocbp == &iocb);
+	ATF_REQUIRE(done == buffer_size);
+
+	ATF_REQUIRE(memcmp(buffer[0], buffer[1], buffer_size) == 0);
+
+	close(s[1]);
+	close(s[0]);
+}
+
 ATF_TP_ADD_TCS(tp)
 {
 
@@ -792,6 +856,7 @@ ATF_TP_ADD_TCS(tp)
 	ATF_TP_ADD_TC(tp, aio_md_test);
 	ATF_TP_ADD_TC(tp, aio_large_read_test);
 	ATF_TP_ADD_TC(tp, aio_socket_two_reads);
+	ATF_TP_ADD_TC(tp, aio_socket_blocking_short_write);
 
 	return (atf_no_error());
 }