From 21c8237a73db1232bb27f48051e9c914831b6fde Mon Sep 17 00:00:00 2001 From: Andrew Wood Date: Sun, 3 May 2026 01:12:33 +0100 Subject: [PATCH] When splice() is available but neither the input nor the output are pipes, splice the input through an intermediary pipe (#188). --- src/include/pv-internal.h | 11 ++++++ src/pv/file.c | 72 ++++++++++++++++++++++++++++++++++++++- src/pv/loop.c | 15 ++++---- src/pv/state.c | 18 ++++++++++ src/pv/transfer.c | 67 +++++++++++++++++++++++++++++++++++- 5 files changed, 173 insertions(+), 10 deletions(-) diff --git a/src/include/pv-internal.h b/src/include/pv-internal.h index 65502b0..7f16c05 100644 --- a/src/include/pv-internal.h +++ b/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; /*************** @@ -456,6 +458,15 @@ 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; /* * These variables are used to keep track of whether * splice() was used; splice_failed_fd is the file diff --git a/src/pv/file.c b/src/pv/file.c index d6b8107..15af94c 100644 --- a/src/pv/file.c +++ b/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,59 @@ 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 512k. */ + if (0 == target_pipe_buffer_size) + target_pipe_buffer_size = 512 * 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 (new_size < 0) { + new_size = fcntl(state->transfer.intermediate_pipe[1], F_GETPIPE_SZ); + if (new_size < 0) { + new_size = 64 * 1024; + } + } + + state->transfer.intermediate_pipe_buffer_size = new_size; + } +#endif + } + 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); + } +#endif /* HAVE_SPLICE */ + return fd; } diff --git a/src/pv/loop.c b/src/pv/loop.c index 52f5709..672ef4c 100644 --- a/src/pv/loop.c +++ b/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 */ @@ -513,7 +512,7 @@ int pv_main_loop(pvstate_t state) * 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) { + if (state->status.output_is_pipe) { size_t target_pipe_buffer_size = state->control.pipe_buffer_size; if (0 == target_pipe_buffer_size) { @@ -636,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) { @@ -661,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, @@ -670,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 diff --git a/src/pv/state.c b/src/pv/state.c index 8e79f78..c067d77 100644 --- a/src/pv/state.c +++ b/src/pv/state.c @@ -208,6 +208,11 @@ pvstate_t pv_state_alloc(void) #endif /* HAVE_IPC */ state->cursor.lock_fd = -1; +#ifdef HAVE_SPLICE + state->transfer.intermediate_pipe[0] = -1; + state->transfer.intermediate_pipe[1] = -1; +#endif /* HAVE_SPLICE */ + pv_state_reset(state); #ifdef HAVE_GETCWD @@ -271,6 +276,19 @@ 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; + } + } + } +#endif /* HAVE_SPLICE */ + if (NULL != transfer->line_positions) free(transfer->line_positions); transfer->line_positions = NULL; diff --git a/src/pv/transfer.c b/src/pv/transfer.c index 1b12d35..80073d9 100644 --- a/src/pv/transfer.c +++ b/src/pv/transfer.c @@ -382,7 +382,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, state->control.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, state->control.output_fd, NULL, bytes_to_splice, SPLICE_F_MORE); + } /*@+type@ */ /*@+nullpass@ */