Clamp write to bufsize-1 bytes to afford sentinel value for no-progress

Author: z2oh

src/io.c

@@ -2518,13 +2518,40 @@ _dispatch_operation_perform(dispatch_operation_t op)
 				NTSTATUS status = _dispatch_NtQueryInformationFile(hFile,
 						&iosb, &fpli, sizeof(fpli), FilePipeLocalInformation);
 				if (NT_SUCCESS(status)) {
-					// WriteQuotaAvailable is unreliable in the presence
-					// of a blocking reader, when it can return zero, so only
-					// account for it otherwise
-					if (fpli.WriteQuotaAvailable > 0) {
-						len = MIN(len, fpli.WriteQuotaAvailable);
+					// WriteQuotaAvailable is the free space in the output buffer
+					// that has not already been reserved for reading. In other words,
+					// WriteQuotaAvailable =
+					//    OutboundQuota - WriteQuotaUsed - QueuedReadSize.
+					// It is not documented that QueuedReadSize is part of this
+					// calculation, but this behavior has been observed experimentally.
+					// Unfortunately, this means that it is not possible to distinguish
+					// between a full output buffer and a reader blocked waiting for a
+					// full buffer's worth of data. This is a problem because if the
+					// output buffer is full and no reader is waiting for data, then
+					// attempting to write to the buffer of a PIPE_WAIT, non-
+					// overlapped I/O pipe will block the dispatch queue thread.
+					//
+					// In order to work around this idiosyncrasy, we bound the size of
+					// the write to be OutboundQuota - 1. This affords us a sentinel value
+					// in WriteQuotaAvailable that can be used to detect if a reader is
+					// making progress or not.
+					// WriteQuotaAvailable = 0 => a reader is blocked waiting for data.
+					// WriteQuotaAvailable = 1 => the pipe has been written to, but no
+					//   reader is making progress.
+					// When we detect that WriteQuotaAvailable == 1, we write 0 bytes to
+					// avoid blocking the dispatch queue thread.
+					if (fpli.WriteQuotaAvailable == 0) {
+						// This condition can only occur when we have a reader blocked
+						// waiting for data on the pipe. In this case, write a full
+						// buffer's worth of data (less one byte to preserve this
+						// sentinel value of WriteQuotaAvailable == 0).
+						len = MIN(len, fpli.OutboundQuota - 1);
+					} else {
+						// Subtract 1 from WriteQuotaAvailable to ensure we do not fill
+						// the pipe and preserve the sentinel value of
+						// WriteQuotaAvailable == 1.
+						len = MIN(len, fpli.WriteQuotaAvailable - 1);
 					}
-					len = MIN(len, fpli.OutboundQuota);
 				}
 
 				OVERLAPPED ovlOverlapped = {};

tests/dispatch_io_pipe.c

@@ -404,7 +404,12 @@ test_dispatch_write(int kind, int delay)
 	dispatch_group_t g = dispatch_group_create();
 	dispatch_group_enter(g);
 
-	const size_t bufsize = test_get_pipe_buffer_size(kind);
+	// The libdispatch implementation writes at most bufsize-1 bytes
+	// before requiring a reader to start making progress. Because
+	// these tests operate serially, the reader will not make progress
+	// until the write finishes, and a write of >= bufsize will not
+	// finish until the reader starts draining the pipe.
+	const size_t bufsize = test_get_pipe_buffer_size(kind) - 1;
 
 	char *buf = calloc(bufsize, 1);
 	assert(buf);