mirrors/pv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

When splice() is available but neither the input nor the output are pipes, splice the input through an intermediary pipe (#188).

Browse Source
This commit is contained in:
Andrew Wood
2026-05-03 01:12:33 +01:00
parent 17cbab3045
commit 21c8237a73
5 changed files with 173 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/include/pv-internal.h
+11
View File
@@ -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
src/pv/file.c
+71 -1
View File
@@ -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;
}
src/pv/loop.c
+7 -8
View File
@@ -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
src/pv/state.c
+18
View File
@@ -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;
src/pv/transfer.c
+66 -1
View File
@@ -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@ */