Add an option "-J" / "--pipe-buffer-size" to set the size of the output pipe buffer when the output is a pipe, and when that option is not specified, automatically increase the output pipe buffer size to match that of the input, when the first input is also a pipe (#188).

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

@@ -201,7 +201,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 */

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;

src/pv/loop.c

@@ -503,6 +503,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 (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.

src/pv/state.c

@@ -642,6 +642,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;