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

Author: z2oh

Diff for: src/io.c

@@ -2518,13 +2518,38 @@ _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.
+					// 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 idiosynchrasy, 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 == 1) {
+						len = 0;
+					} else if (fpli.WriteQuotaAvailable > 1) {
+						// 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);
+
+					// Always bound the write size by the OutboundQuota - 1 to preserve
+					// the sentinel value of WriteQuotaAvailable == 1.
+					len = MIN(len, fpli.OutboundQuota - 1);
 				}
 
 				OVERLAPPED ovlOverlapped = {};

Diff for: tests/dispatch_io_pipe.c

@@ -404,7 +404,14 @@ 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);
+	// We must subtract 1 from the bufsize because the write operation
+	// uses 1 byte of the buffer as a sentinel value to ensure that a
+	// reader is 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. In practice, a serial full-buffer
+	// read-after-write will not be encountered.
+	const size_t bufsize = test_get_pipe_buffer_size(kind) - 1;
 
 	char *buf = calloc(bufsize, 1);
 	assert(buf);