fix: avoid EOF hang on unread pipe output (#189)

At EOF, pv waits for bytes already written to an output pipe to be consumed before finishing. If the downstream reader exits without reading those bytes, the pipe can keep reporting unread data forever. Since pv has no more input, there is no later write that would fail with EPIPE and break the wait.

Detect this EOF drain-wait case with a non-blocking poll() on the output pipe. When the pipe reports that its readers are gone, mark the pipe closed and stop waiting for bytes that can no longer be consumed. While a reader is still present, keep the existing behaviour and continue waiting for the pipe buffer to drain.

Add regression coverage for both ways pv can reach this state: direct input that leaves unread output bytes behind, and a pipeline where the original producer fails but an intermediate filter such as gzip still emits a valid empty stream. The latter matters because pv can still have pending output at EOF even when the command at the start of the pipeline failed immediately.

Signed-off-by: Wolfgang Hoschek <wolfgang.hoschek@mac.com>

Makefile.am

@@ -82,6 +82,8 @@ tests/Integrity_-_Basic.test \
 tests/Integrity_-_Binary_data.test \
 tests/Integrity_-_From_bursty_source.test \
 tests/Integrity_-_Large_file_support.test \
+tests/Integrity_-_On_closed_output_pipe_at_EOF.test \
+tests/Integrity_-_On_failed_input_pipe_at_EOF.test \
 tests/Integrity_-_On_output_pipe_close.test \
 tests/Integrity_-_When_adjusted_remotely.test \
 tests/Memory_safety_-_Basic.test \

configure.ac

@@ -45,6 +45,8 @@ AC_CHECK_HEADERS([sys/ioctl.h])
 AC_CHECK_HEADERS([libintl.h locale.h])
 AC_CHECK_HEADERS([sys/sysmacros.h])
 AC_CHECK_HEADERS([ftw.h])
+AC_CHECK_HEADERS([poll.h sys/poll.h])
+AC_CHECK_FUNCS([poll])
 AC_CHECK_MEMBERS([struct stat.st_blksize])
 AC_CHECK_DECLS([SA_SIGINFO], [], [], [[#include <signal.h>]])
 

src/pv/loop.c

@@ -25,6 +25,12 @@
 #include <sys/ioctl.h>
 #include <sys/stat.h>
 
+#if HAVE_POLL_H
+#include <poll.h>
+#elif HAVE_SYS_POLL_H
+#include <sys/poll.h>
+#endif
+
 #if HAVE_MATH_H
 #include <math.h>
 #endif
@@ -84,6 +90,48 @@ static bool pv__resize_display_on_signal(pvstate_t state)
 }
 
 
+/*
+ * Return true if the write end of a pipe reports that its readers have gone
+ * away.  This is only used after PV has reached EOF and is waiting for the
+ * output pipe buffer to drain; live readers still get the existing behaviour.
+ */
+static bool pv__output_pipe_has_no_reader(int output_fd)
+{
+#if HAVE_POLL && (HAVE_POLL_H || HAVE_SYS_POLL_H)
+	struct pollfd pfd;
+	int result;
+
+	memset(&pfd, 0, sizeof(pfd));
+	pfd.fd = output_fd;
+	pfd.events = POLLOUT;
+
+	result = poll(&pfd, 1, 0);
+	if (result < 0) {
+		if (EINTR != errno) {
+			debug("%s(%d): %s", "poll", output_fd, strerror(errno));
+		}
+		return false;
+	}
+	if (0 == result)
+		return false;
+
+#if defined(POLLERR)
+	if (0 != (pfd.revents & POLLERR))
+		return true;
+#endif
+#if defined(POLLHUP)
+	if (0 != (pfd.revents & POLLHUP))
+		return true;
+#endif
+
+	return false;
+#else
+	(void) output_fd;
+	return false;
+#endif
+}
+
+
 /*
  * Calculate and display the transfer statistics at the end of the transfer,
  * if stats are enabled and any measurements were taken.
@@ -673,8 +721,15 @@ int pv_main_loop(pvstate_t state)
 		 * buffer to empty (#164).
 		 */
 		if (eof_in && eof_out && state->transfer.written_but_not_consumed > 0) {
-			debug("%s", "EOF but bytes remain in output pipe - sleeping");
-			pv_nanosleep(50000000);
+			if (pv__output_pipe_has_no_reader(output_fd)) {
+				debug("%s",
+				      "EOF but output pipe readers are gone - clearing written_but_not_consumed");
+				state->flags.pipe_closed = 1;
+				state->transfer.written_but_not_consumed = 0;
+			} else {
+				debug("%s", "EOF but bytes remain in output pipe - sleeping");
+				pv_nanosleep(50000000);
+			}
 		}
 
 		/*

tests/Integrity_-_On_closed_output_pipe_at_EOF.test

@@ -0,0 +1,32 @@
+#!/bin/sh
+#
+# Check that EOF does not wait forever when the output pipe reader exits
+# without consuming bytes already written into the pipe.
+
+# Allow all tests to be skipped, e.g. during a release build
+test "${SKIP_ALL_TESTS}" = "1" && exit 77
+
+true "${testSubject:?not set - call this from 'make check'}"
+
+TIMEOUT=""
+if command -v timeout >/dev/null 2>&1; then
+	TIMEOUT="timeout"
+elif command -v gtimeout >/dev/null 2>&1; then
+	TIMEOUT="gtimeout"
+else
+	echo "timeout command not available"
+	exit 77
+fi
+
+# The reader keeps the pipe open briefly but reads nothing.  When it exits,
+# PV has no further input, so it must notice the closed reader without
+# relying on a later write returning EPIPE.
+if ! "${TIMEOUT}" 4 sh <<EOF
+printf "x" | "${testSubject}" -q -B 1 | sleep 1
+EOF
+then
+	echo "PV did not finish after the output pipe reader exited"
+	exit 1
+fi
+
+exit 0

tests/Integrity_-_On_failed_input_pipe_at_EOF.test

@@ -0,0 +1,37 @@
+#!/bin/sh
+#
+# Check that EOF does not wait forever when the upstream producer exits with
+# failure and the output pipe reader exits without consuming buffered bytes.
+
+# Allow all tests to be skipped, e.g. during a release build
+test "${SKIP_ALL_TESTS}" = "1" && exit 77
+
+true "${testSubject:?not set - call this from 'make check'}"
+
+TIMEOUT=""
+if command -v timeout >/dev/null 2>&1; then
+	TIMEOUT="timeout"
+elif command -v gtimeout >/dev/null 2>&1; then
+	TIMEOUT="gtimeout"
+else
+	echo "timeout command not available"
+	exit 77
+fi
+
+if ! command -v gzip >/dev/null 2>&1; then
+	echo "gzip command not available"
+	exit 77
+fi
+
+# The producer fails immediately and writes no data.  gzip still emits a small
+# valid empty stream, so PV writes bytes to its output pipe before reaching EOF.
+# The reader keeps the pipe open briefly but consumes nothing.
+if ! "${TIMEOUT}" 4 sh <<EOF
+false | gzip -1 -c -n | "${testSubject}" -q -B 1 | sleep 1
+EOF
+then
+	echo "PV did not finish after the input producer failed and the output reader exited"
+	exit 1
+fi
+
+exit 0