Merge the changes for #188 and #191 which add an intermediate pipe on the input side to allow splice() to be used for all transfers.

docs/pv.1

@@ -291,6 +291,14 @@ automatically switches on \*(lq\fB\-\-no\-splice\fR\*(rq.
 Switching on this option results in a small loss of transfer efficiency.
 It has no effect on systems where \fBsplice\fR(2) is unavailable.
 .TP
+.BI \-J\  BYTES \fR,\ \fB\-\-pipe\-buffer\-size\  BYTES
+Attempt to set the size of the output pipe buffer to \fIBYTES\fR bytes.
+This will only have an effect if the output is a pipe, and will silently do
+nothing if the requested size cannot be achieved.
+If this option is not specified, and both the output and the first input are
+pipes, then the output pipe buffer size will automatically be increased to
+match the input pipe buffer, if it was smaller.
+.TP
 .B \-E, \-\-skip-errors
 Ignore read errors by attempting to skip past the offending sections.
 The corresponding parts of the output will be null bytes.
@@ -348,7 +356,6 @@ seekable, this option will have no effect other than to turn on
 Instead of transferring input data to standard output, discard it.
 This is equivalent to redirecting standard output to \fI/dev/null\fR,
 except that \fBwrite\fR(2) is never called.
-Implies \*(lq\fB\-\-no\-splice\fR\*(rq.
 .TP
 .BI \-U\  FILE \fR,\ \fB\-\-store\-and\-forward\  FILE
 Instead of passing data through immediately, do it in two stages - first
@@ -857,10 +864,10 @@ effective with small input files, and \*(lq\fB%nL\fR\*(rq may be a few lines
 out due to buffering within the pipeline itself.
 .PP
 Numbers passed to \*(lq\fB\-\-size\fR\*(rq, \*(lq\fB\-\-rate\-limit\fR\*(rq,
-\*(lq\fB\-\-buffer\-size\fR\*(rq, and \*(lq\fB\-\-error\-skip\-block\fR\*(rq
-may all be expressed as decimals if followed by a suffix, so for example
-\*(lq\fI\-\-size\~1.5G\fR\*(rq is equivalent to
-\*(lq\fI\-\-size\~1536M\fR\*(rq.
+\*(lq\fB\-\-buffer\-size\fR\*(rq, \*(lq\fB\-\-pipe\-buffer\-size\fR\*(rq,
+and \*(lq\fB\-\-error\-skip\-block\fR\*(rq may all be expressed as decimals
+if followed by a suffix, so for example \*(lq\fI\-\-size\~1.5G\fR\*(rq is
+equivalent to \*(lq\fI\-\-size\~1536M\fR\*(rq.
 .PP
 Numbers passed to \*(lq\fB\-\-interval\fR\*(rq and
 \*(lq\fB\-\-delay\-start\fR\*(rq may be integers or decimals, but may not

docs/pv.1.md

@@ -300,6 +300,15 @@ are explicitly switched on will be shown.
     option results in a small loss of transfer efficiency. It has no
     effect on systems where **splice**(2) is unavailable.
 
+**-J BYTES, \--pipe-buffer-size BYTES**
+
+:   Attempt to set the size of the output pipe buffer to *BYTES* bytes.
+    This will only have an effect if the output is a pipe, and will
+    silently do nothing if the requested size cannot be achieved. If
+    this option is not specified, and both the output and the first
+    input are pipes, then the output pipe buffer size will automatically
+    be increased to match the input pipe buffer, if it was smaller.
+
 **-E, \--skip-errors**
 
 :   Ignore read errors by attempting to skip past the offending
@@ -360,8 +369,7 @@ are explicitly switched on will be shown.
 
 :   Instead of transferring input data to standard output, discard it.
     This is equivalent to redirecting standard output to */dev/null*,
-    except that **write**(2) is never called. Implies
-    "**\--no-splice**".
+    except that **write**(2) is never called.
 
 **-U FILE, \--store-and-forward FILE**
 
@@ -814,9 +822,10 @@ small input files, and "**%nL**" may be a few lines out due to buffering
 within the pipeline itself.
 
 Numbers passed to "**\--size**", "**\--rate-limit**",
-"**\--buffer-size**", and "**\--error-skip-block**" may all be expressed
-as decimals if followed by a suffix, so for example "*\--size 1.5G*" is
-equivalent to "*\--size 1536M*".
+"**\--buffer-size**", "**\--pipe-buffer-size**", and
+"**\--error-skip-block**" may all be expressed as decimals if followed
+by a suffix, so for example "*\--size 1.5G*" is equivalent to
+"*\--size 1536M*".
 
 Numbers passed to "**\--interval**" and "**\--delay-start**" may be
 integers or decimals, but may not have a suffix.

src/include/options.h

@@ -72,6 +72,7 @@ struct opts_s {
 	size_t lastwritten;            /* show N bytes last written */
 	off_t rate_limit;              /* rate limit, in bytes per second */
 	size_t buffer_size;            /* buffer size, in bytes (0=default) */
+	size_t pipe_buffer_size;       /* pipe buffer size, in bytes (0=default) */
 	off_t size;                    /* total size of data */
 	off_t error_skip_block;        /* skip block size, 0 for adaptive */
 	pid_t remote;                  /* PID of pv to update settings of */

src/include/pv-internal.h

@@ -160,6 +160,8 @@ struct pvstate_s {
 		bool terminal_supports_utf8;	 /* whether the terminal supports UTF-8 */
 		bool terminal_supports_colour;	 /* whether the terminal supports colour */
 		bool checked_colour_support;	 /* whether we have checked colour support yet */
+		bool current_input_is_pipe;	 /* whether the current input file is a pipe */
+		bool output_is_pipe;		 /* whether the output is a pipe */
 	} status;
 
 	/***************
@@ -201,7 +203,8 @@ struct pvstate_s {
 		/*@null@*/ char *default_bar_style; /* which bar style to use by default */
 		off_t error_skip_block;          /* skip block size, 0 for adaptive */
 		off_t rate_limit;                /* rate limit, in bytes per second */
-		size_t target_buffer_size;       /* buffer size (0=default) */
+		size_t target_buffer_size;       /* transfer buffer size (0=default) */
+		size_t pipe_buffer_size;         /* pipe buffer size (0=default) */
 		off_t size;                      /* total size of data */
 		unsigned int skip_errors;        /* skip read errors counter */
 		int output_fd;                   /* fd to write output to */
@@ -455,6 +458,17 @@ struct pvstate_s {
 		int last_read_skip_fd;
 		/* read_error_warning_shown is defined below. */
 #ifdef HAVE_SPLICE
+		/*
+		 * File descriptors for an intermediate pipe, used when
+		 * neither input nor output are pipes; the size of the pipe
+		 * buffer; and how much input data is in the intermediate
+		 * pipe waiting to be passed to the output.
+		 */
+		int intermediate_pipe[2];
+		int intermediate_pipe_buffer_size;
+		int intermediate_pipe_buffer_used;
+		/* File descriptor to /dev/null for splicing with -X. */
+		int discard_fd;
 		/*
 		 * These variables are used to keep track of whether
 		 * splice() was used; splice_failed_fd is the file

src/include/pv.h

@@ -260,6 +260,7 @@ extern void pv_state_sparse_output_set(pvstate_t, bool);
 extern void pv_state_rate_limit_set(pvstate_t, off_t);
 extern void pv_state_target_buffer_size_set(pvstate_t, size_t);
 extern void pv_state_no_splice_set(pvstate_t, bool);
+extern void pv_state_pipe_buffer_size_set(pvstate_t, size_t);
 extern void pv_state_discard_input_set(pvstate_t, bool);
 extern void pv_state_size_set(pvstate_t, off_t);
 extern void pv_state_interval_set(pvstate_t, double);

src/main/help.c

@@ -400,6 +400,9 @@ void display_help(void)
 		{ "-C", "--no-splice", NULL,
 		 N_("never use splice(), always use read/write"),
 		 { 0, 0, 0, 0} },
+		{ "-J", "--pipe-buffer-size", N_("BYTES"),
+		 N_("set the pipe buffer size to BYTES"),
+		 { 0, 0, 0, 0} },
 		{ "-E", "--skip-errors", NULL,
 		 N_("skip read errors in input"),
 		 { 0, 0, 0, 0} },

src/main/main.c

@@ -906,6 +906,7 @@ int main(int argc, char **argv)
 	pv_state_rate_limit_set(state, opts->rate_limit);
 	pv_state_target_buffer_size_set(state, opts->buffer_size);
 	pv_state_no_splice_set(state, opts->no_splice);
+	pv_state_pipe_buffer_size_set(state, opts->pipe_buffer_size);
 	pv_state_size_set(state, opts->size);
 	pv_state_name_set(state, opts->name);
 	pv_state_default_bar_style_set(state, opts->default_bar_style);

src/main/options.c

@@ -721,6 +721,7 @@ opts_t opts_parse(unsigned int argc, char **argv)
 		{ "no-splice", 0, NULL, (int) 'C' },
 		{ "skip-errors", 0, NULL, (int) 'E' },
 		{ "error-skip-block", 1, NULL, (int) 'Z' },
+		{ "pipe-buffer-size", 1, NULL, (int) 'J' },
 		{ "stop-at-size", 0, NULL, (int) 'S' },
 		{ "sync", 0, NULL, (int) 'Y' },
 		{ "direct-io", 0, NULL, (int) 'K' },
@@ -743,7 +744,7 @@ opts_t opts_parse(unsigned int argc, char **argv)
 	/*@+nullassign@ */
 	int option_index = 0;
 #endif				/* HAVE_GETOPT_LONG */
-	char *short_options = "hVpteIrab8kTA:fvnqcWD:s:gl0i:w:H:N:u:F:x:L:B:CEZ:SYKOXU:R:Q:P:d:m:o:M:"
+	char *short_options = "hVpteIrab8kTA:fvnqcWD:s:gl0i:w:H:N:u:F:x:L:B:CEZ:J:SYKOXU:R:Q:P:d:m:o:M:"
 #ifdef ENABLE_DEBUGGING
 	    "!:"
 #endif
@@ -825,6 +826,7 @@ opts_t opts_parse(unsigned int argc, char **argv)
 		case 'L':
 		case 'B':
 		case 'Z':
+		case 'J':
 			if (!pv_getnum_check(optarg, PV_NUMTYPE_ANY_WITH_SUFFIX)) {
 				/*@-mustfreefresh@ *//* see above */
 				pv_error("-%c: %s: %s", c, optarg, _("numeric value not understood"));
@@ -1064,6 +1066,9 @@ opts_t opts_parse(unsigned int argc, char **argv)
 		case 'Z':
 			opts->error_skip_block = pv_getnum_size(optarg, opts->decimal_units);
 			break;
+		case 'J':
+			opts->pipe_buffer_size = (size_t) pv_getnum_size(optarg, opts->decimal_units);
+			break;
 		case 'S':
 			opts->stop_at_size = true;
 			break;
@@ -1079,7 +1084,6 @@ opts_t opts_parse(unsigned int argc, char **argv)
 			break;
 		case 'X':
 			opts->discard_input = true;
-			opts->no_splice = true;
 			break;
 		case 'U':
 			opts->store_and_forward_file = pv_strdup(optarg);

src/pv/file.c

@@ -290,7 +290,13 @@ off_t pv_calc_total_size(pvstate_t state)
  * error). It is an error if the next input file is the same as the file
  * the output is pointing to.
  *
- * Updates state->status.current_input_file in the process.
+ * Updates state->status.current_input_file and
+ * state->status.current_input_is_pipe.
+ *
+ * If the input is not a pipe, and state->status.output_is_pipe is false,
+ * and state->control.no_splice is false, also creates a pipe and populates
+ * state->transfer.intermediate_pipe[] with its file descriptors, if a pipe
+ * had not already been created (i.e.  that array contained [-1,-1]).
  */
 int pv_next_file(pvstate_t state, unsigned int filenum, int oldfd)
 {
@@ -375,6 +381,17 @@ int pv_next_file(pvstate_t state, unsigned int filenum, int oldfd)
 		return -1;
 	}
 
+	/*
+	 * Detect whether the input file is a pipe.  This is used later, in
+	 * pv__transfer_read(), to decide whether an intermediate pipe needs
+	 * to be used with splice().
+	 */
+	state->status.current_input_is_pipe = false;
+	if ((isb.st_mode & S_IFMT) == S_IFIFO) {
+		state->status.current_input_is_pipe = true;
+		debug("%s (fd %d)", "input is a pipe", fd);
+	}
+
 	state->status.current_input_file = filenum;
 #ifdef O_DIRECT
 	/*
@@ -398,6 +415,75 @@ int pv_next_file(pvstate_t state, unsigned int filenum, int oldfd)
 
 	debug("%s: %d: %s: fd=%d", "next file opened", filenum, pv_current_file_name(state), fd);
 
+#ifdef HAVE_SPLICE
+	if (!(state->status.output_is_pipe || state->status.current_input_is_pipe || state->control.no_splice)
+	    && (-1 == state->transfer.intermediate_pipe[0])) {
+		/*
+		 * Create an intermediate pipe to allow the input to be used
+		 * with splice() even though neither it nor the output are
+		 * themselves pipes.
+		 */
+		state->transfer.intermediate_pipe_buffer_used = 0;
+		if (pipe(state->transfer.intermediate_pipe) < 0) {
+			debug("%s: %s", "pipe()", strerror(errno));
+			state->transfer.intermediate_pipe[0] = -1;
+			state->transfer.intermediate_pipe[1] = -1;
+		} else {
+			state->transfer.intermediate_pipe_buffer_size = 64 * 1024;
+#if defined F_SETPIPE_SZ && defined F_GETPIPE_SZ
+			{
+				size_t target_pipe_buffer_size = state->control.pipe_buffer_size;
+				int new_size;
+
+				/* If no pipe buffer size was set, try for 1MiB. */
+				if (0 == target_pipe_buffer_size)
+					target_pipe_buffer_size = 1024 * 1024;
+
+				/*
+				 * Try to set the pipe buffer size, halving
+				 * repeatedly on failure.
+				 */
+				new_size = -1;
+				while (new_size < 0 && target_pipe_buffer_size >= 4096) {
+					new_size =
+					    fcntl(state->transfer.intermediate_pipe[1], F_SETPIPE_SZ,
+						  (int) target_pipe_buffer_size);
+					if (new_size < 0)
+						target_pipe_buffer_size = target_pipe_buffer_size / 2;
+				}
+				/* If all attempts failed, read the current size. */
+				if (new_size < 0) {
+					new_size = fcntl(state->transfer.intermediate_pipe[1], F_GETPIPE_SZ);
+					/* If unable to read, assume 64KiB. */
+					if (new_size < 0) {
+						new_size = 64 * 1024;
+					}
+				}
+
+				state->transfer.intermediate_pipe_buffer_size = new_size;
+			}
+#endif				/* defined F_SETPIPE_SZ && defined F_GETPIPE_SZ */
+		}
+		debug("%s: [%d,%d]", "intermediate pipe fds", state->transfer.intermediate_pipe[0],
+		      state->transfer.intermediate_pipe[1]);
+		debug("%s: %d", "intermediate pipe buffer size", state->transfer.intermediate_pipe_buffer_size);
+	}
+
+	if (state->control.discard_input && !state->control.no_splice && state->transfer.discard_fd < 0) {
+		/*
+		 * Open a file descriptor to /dev/null, so that input can be
+		 * spliced to it to implement -X.
+		 */
+		state->transfer.discard_fd = open("/dev/null", O_WRONLY);	/* flawfinder: ignore */
+		/* flawfinder: /dev/null is trusted. */
+		if (state->transfer.discard_fd < 0) {
+			pv_perror("%s", "/dev/null");
+			(void) close(fd);
+			fd = -1;
+		}
+	}
+#endif				/* HAVE_SPLICE */
+
 	return fd;
 }
 

src/pv/loop.c

@@ -355,7 +355,6 @@ int pv_main_loop(pvstate_t state)
 	struct timespec next_remotecheck, next_monitor_exchange;
 	int input_fd, output_fd;
 	unsigned int file_idx;
-	bool output_is_pipe;
 
 	/*
 	 * Notes on line mode:
@@ -379,14 +378,14 @@ int pv_main_loop(pvstate_t state)
 		output_fd = STDOUT_FILENO;
 
 	/* Determine whether the output is a pipe. */
-	output_is_pipe = false;
+	state->status.output_is_pipe = false;
 	{
 		struct stat sb;
 		memset(&sb, 0, sizeof(sb));
 		if (0 == fstat(output_fd, &sb)) {
 			/*@-type@ */
 			if ((sb.st_mode & S_IFMT) == S_IFIFO) {
-				output_is_pipe = true;
+				state->status.output_is_pipe = true;
 				debug("%s (fd %d)", "output is a pipe", output_fd);
 			}
 			/*@+type@ *//* splint says st_mode is __mode_t, not mode_t */
@@ -503,6 +502,46 @@ int pv_main_loop(pvstate_t state)
 	if (0 == state->control.target_buffer_size)
 		state->control.target_buffer_size = BUFFER_SIZE;
 
+#ifdef F_GETPIPE_SZ
+#ifdef F_SETPIPE_SZ
+	/*
+	 * If a pipe buffer size was explicitly set, then set the output
+	 * pipe buffer size to that.
+	 *
+	 * If no pipe buffer size was set, and the first input is a pipe,
+	 * then increase the output pipe buffer size to the same as the
+	 * input pipe buffer size, if the output buffer was smaller.
+	 */
+	if (state->status.output_is_pipe) {
+		size_t target_pipe_buffer_size = state->control.pipe_buffer_size;
+
+		if (0 == target_pipe_buffer_size) {
+			int input_pipe_size;
+			input_pipe_size = fcntl(input_fd, F_GETPIPE_SZ);
+
+			debug("%s %d: %s=%d", "input fd", input_fd, "pipe size", input_pipe_size);
+			if (input_pipe_size > 0) {
+				int output_pipe_size;
+				output_pipe_size = fcntl(output_fd, F_GETPIPE_SZ);
+				debug("%s %d: %s=%d", "output fd", output_fd, "pipe size", output_pipe_size);
+				if (output_pipe_size > 0 && output_pipe_size < input_pipe_size) {
+					target_pipe_buffer_size = (size_t) input_pipe_size;
+				}
+			}
+		}
+		if (target_pipe_buffer_size > 0) {
+			int output_pipe_size;
+
+			output_pipe_size = fcntl(output_fd, F_SETPIPE_SZ, (int) target_pipe_buffer_size);
+			debug("%s %d: %s=%d", "output fd", output_fd, "updated pipe size", output_pipe_size);
+			if (output_pipe_size < 0) {
+				debug("%s: %s", "failed to set pipe buffer size", strerror(errno));
+			}
+		}
+	}
+#endif				/* F_SETPIPE_SZ */
+#endif				/* F_GETPIPE_SZ */
+
 	/*
 	 * Repeat until eof_in is true, eof_out is true, and final_update is
 	 * true.
@@ -596,7 +635,7 @@ int pv_main_loop(pvstate_t state)
 		 * If writing to a pipe, look at how much is sitting in the
 		 * pipe buffer waiting for the receiver to read.
 		 */
-		if (output_is_pipe) {
+		if (state->status.output_is_pipe) {
 			int nbytes;
 			nbytes = 0;
 			if (0 != state->flags.pipe_closed) {
@@ -621,7 +660,7 @@ int pv_main_loop(pvstate_t state)
 #endif
 
 		state->transfer.transferred = state->transfer.total_written;
-		if (output_is_pipe && !state->control.linemode) {
+		if (state->status.output_is_pipe && !state->control.linemode) {
 			/*
 			 * Writing bytes to a pipe - the amount transferred
 			 * to the receiver is the total amount written,
@@ -630,8 +669,8 @@ int pv_main_loop(pvstate_t state)
 			 */
 			state->transfer.transferred -= state->transfer.written_but_not_consumed;
 
-		} else if (output_is_pipe && state->control.linemode && state->transfer.written_but_not_consumed > 0
-			   && NULL != state->transfer.line_positions) {
+		} else if (state->status.output_is_pipe && state->control.linemode
+			   && state->transfer.written_but_not_consumed > 0 && NULL != state->transfer.line_positions) {
 			/*
 			 * Writing lines to a pipe - similar to above, but
 			 * with the added complication of having to

src/pv/state.c

@@ -133,6 +133,9 @@ void pv_reset_transfer(pvtransferstate_t transfer)
 	transfer->last_read_skip_fd = 0;
 #ifdef HAVE_SPLICE
 	transfer->splice_failed_fd = -1;
+	transfer->intermediate_pipe[0] = -1;
+	transfer->intermediate_pipe[1] = -1;
+	transfer->discard_fd = -1;
 #endif				/* HAVE_SPLICE */
 
 	transfer->line_positions_length = 0;
@@ -271,6 +274,25 @@ void pv_freecontents_transfer(pvtransferstate_t transfer)
 	/*@+keeptrans@ */
 	/* splint - explicitly freeing this structure, so free() here is OK. */
 
+#ifdef HAVE_SPLICE
+	/* Close the intermediate pipe, if there was one. */
+	if (transfer->intermediate_pipe_buffer_size > 0) {
+		int idx;
+		for (idx = 0; idx < 2; idx++) {
+			if (transfer->intermediate_pipe[idx] >= 0) {
+				(void) close(transfer->intermediate_pipe[idx]);
+				transfer->intermediate_pipe[idx] = -1;
+			}
+		}
+	}
+
+	/* Close the discard file descriptor, if there was one. */
+	if (transfer->discard_fd >= 0) {
+		(void) close(transfer->discard_fd);
+		transfer->discard_fd = -1;
+	}
+#endif				/* HAVE_SPLICE */
+
 	if (NULL != transfer->line_positions)
 		free(transfer->line_positions);
 	transfer->line_positions = NULL;
@@ -642,6 +664,11 @@ void pv_state_no_splice_set(pvstate_t state, bool val)
 	state->control.no_splice = val;
 }
 
+void pv_state_pipe_buffer_size_set(pvstate_t state, size_t val)
+{
+	state->control.pipe_buffer_size = val;
+}
+
 void pv_state_size_set(pvstate_t state, off_t val)
 {
 	state->control.size = val;

src/pv/transfer.c

@@ -341,6 +341,13 @@ static bool pv__transfer_read(pvstate_t state, int fd, bool *eof_in, bool *eof_o
 	size_t bytes_can_read;
 	off_t amount_to_skip, amount_skipped, orig_offset, skip_offset;
 	ssize_t nread;
+#ifdef HAVE_SPLICE
+	int output_fd;
+
+	output_fd = state->control.output_fd;
+	if (state->control.discard_input && !state->control.no_splice)
+		output_fd = state->transfer.discard_fd;
+#endif				/* HAVE_SPLICE */
 
 	do_not_skip_errors = false;
 	if (0 == state->control.skip_errors)
@@ -382,7 +389,72 @@ static bool pv__transfer_read(pvstate_t state, int fd, bool *eof_in, bool *eof_o
 		/*@-nullpass@ */
 		/*@-type@ */
 		/* splint doesn't know about splice. */
-		nread = splice(fd, NULL, state->control.output_fd, NULL, bytes_to_splice, SPLICE_F_MORE);
+
+		if (!(state->status.output_is_pipe || state->status.current_input_is_pipe)
+		    && (-1 != state->transfer.intermediate_pipe[0]) && (-1 != state->transfer.intermediate_pipe[1])) {
+			/*
+			 * Neither the input nor the output is a pipe, but
+			 * there is an intermediate pipe to use, so splice
+			 * from the input to that pipe, and from the pipe to
+			 * the output.
+			 *
+			 * This has to be done in stages so as to keep
+			 * within the size of the pipe buffer, otherwise
+			 * splicing into it will block.
+			 */
+			int room_in_pipe_buffer;
+			size_t bytes_to_splice_in;
+
+			room_in_pipe_buffer =
+			    state->transfer.intermediate_pipe_buffer_size -
+			    state->transfer.intermediate_pipe_buffer_used;
+			if (room_in_pipe_buffer < 0)
+				room_in_pipe_buffer = 0;
+
+			bytes_to_splice_in = bytes_to_splice;
+			if (bytes_to_splice_in > (size_t) room_in_pipe_buffer)
+				bytes_to_splice_in = (size_t) room_in_pipe_buffer;
+
+			/*
+			 * Read into the intermediate pipe if it has room in
+			 * its buffer.
+			 */
+			if (bytes_to_splice_in > 0) {
+				ssize_t spliced_in;
+
+				spliced_in =
+				    splice(fd, NULL, state->transfer.intermediate_pipe[1], NULL, bytes_to_splice_in,
+					   SPLICE_F_MORE);
+				if (spliced_in > 0) {
+					state->transfer.intermediate_pipe_buffer_used += (int) spliced_in;
+				}
+				nread = spliced_in;
+			}
+
+			/*
+			 * Write from the intermediate pipe if it has
+			 * anything in its buffer and the read, above, did
+			 * not produce an error.
+			 */
+			if (nread >= 0 && state->transfer.intermediate_pipe_buffer_used > 0) {
+				size_t bytes_to_splice_out = (size_t) (state->transfer.intermediate_pipe_buffer_used);
+				ssize_t spliced_out;
+
+				if (bytes_to_splice_out > bytes_to_splice)
+					bytes_to_splice_out = bytes_to_splice;
+
+				spliced_out =
+				    splice(state->transfer.intermediate_pipe[0], NULL, output_fd, NULL,
+					   bytes_to_splice_out, SPLICE_F_MORE);
+				if (spliced_out > 0) {
+					state->transfer.intermediate_pipe_buffer_used -= (int) spliced_out;
+				}
+				nread = spliced_out;
+			}
+		} else {
+			/* Normal splice() from input to output. */
+			nread = splice(fd, NULL, output_fd, NULL, bytes_to_splice, SPLICE_F_MORE);
+		}
 		/*@+type@ */
 		/*@+nullpass@ */
 
@@ -409,7 +481,7 @@ static bool pv__transfer_read(pvstate_t state, int fd, bool *eof_in, bool *eof_o
 				 * error, it can't be skipped, so set
 				 * "do_not_skip_errors".
 				 */
-				if ((fdatasync(state->control.output_fd) < 0)
+				if ((fdatasync(output_fd) < 0)
 				    && (EIO == errno)) {
 					nread = -1;
 					do_not_skip_errors = true;