From 856da0974138e4cb992aa25fb7733b2cf9ade4f7 Mon Sep 17 00:00:00 2001 From: Andrew Wood Date: Sat, 5 Oct 2024 23:04:18 +0100 Subject: [PATCH] Add a circular buffer of line positions so that in line mode we can work out how many lines are in the unconsumed portion of the output pipe buffer. --- docs/NEWS.md | 1 + src/include/pv-internal.h | 10 +++++++- src/pv/loop.c | 51 ++++++++++++++++++++++++++++++++++++++- src/pv/state.c | 8 ++++++ src/pv/transfer.c | 42 +++++++++++++++++++++++++++++--- 5 files changed, 107 insertions(+), 5 deletions(-) diff --git a/docs/NEWS.md b/docs/NEWS.md index 85544aa..4a241a7 100644 --- a/docs/NEWS.md +++ b/docs/NEWS.md @@ -6,6 +6,7 @@ * *feature:* **--gauge** with **--progress** to show rate gauge when size is unknown ([#46](https://codeberg.org/a-j-wood/pv/issues/46)) * *i18n:* comprehensive German translations update from Hartmut Goebel ([#98](https://codeberg.org/a-j-wood/pv/pulls/98)) * *fix:* resume stopped pipelines when running in the background (part of [#56](https://codeberg.org/a-j-wood/pv/issues/56)) + * *fix*: inspect the output pipe buffer to give a more accurate progress indicator of how much the next command has consumed * *fix:* prefix completion time (**--fineta**) with *FIN* rather than *ETA* ([#43](https://codeberg.org/a-j-wood/pv/issues/43)) * *fix:* surround average rate (**--average-rate**) with brackets rather than square brackets * *fix:* correct a memory leak in **--watchfd PID** diff --git a/src/include/pv-internal.h b/src/include/pv-internal.h index 853ec88..f4ff24a 100644 --- a/src/include/pv-internal.h +++ b/src/include/pv-internal.h @@ -49,6 +49,7 @@ typedef enum { #define MAX_WRITE_AT_ONCE (size_t) 524288 /* max to write() in one go */ #define TRANSFER_READ_TIMEOUT 0.09L /* seconds to time reads out at */ #define TRANSFER_WRITE_TIMEOUT 0.9L /* seconds to time writes out at */ +#define MAX_LINE_POSITIONS 100000 /* number of lines to remember positions of */ #define MAXIMISE_BUFFER_FILL 1 @@ -282,8 +283,8 @@ struct pvstate_s { * will always be less than or equal to read_position. */ struct { - /*@only@*/ /*@null@*/ char *transfer_buffer; /* data transfer buffer */ long double elapsed_seconds; /* how long we have been transferring data for */ + /*@only@*/ /*@null@*/ char *transfer_buffer; /* data transfer buffer */ size_t buffer_size; /* size of buffer */ size_t read_position; /* amount of data in buffer */ size_t write_position; /* buffered data written */ @@ -296,6 +297,13 @@ struct pvstate_s { off_t total_written; /* total bytes or lines written */ off_t transferred; /* amount transferred (written - unconsumed) */ + /* Keep track of line positions to backtrack written_but_not_consumed. */ + /*@null@*/ off_t *line_positions; /* line separator write positions (circular buffer) */ + size_t line_positions_capacity; /* total size of line position array */ + size_t line_positions_length; /* number of positions stored in array */ + size_t line_positions_head; /* index to use for next position */ + off_t last_output_position; /* write position last sent to output */ + /* * While reading from a file descriptor we keep track of how * many times in a row we've seen errors diff --git a/src/pv/loop.c b/src/pv/loop.c index 0ce68c4..cb166d8 100644 --- a/src/pv/loop.c +++ b/src/pv/loop.c @@ -314,9 +314,58 @@ int pv_main_loop(pvstate_t state) #endif state->transfer.transferred = state->transfer.total_written; - /* TODO: work out what to do in line mode. */ if (output_is_pipe && !state->control.linemode) { + /* + * Writing bytes to a pipe - the amount transferred + * to the receiver is the total amount we've + * written, minus what's sitting in the pipe buffer + * waiting for the receiver to consume it. + */ 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) { + /* + * Writing lines to a pipe - similar to above, but + * we have to work out how many lines the + * yet-to-be-consumed data in the buffer equates to. + * + * To do this, we walk backwards through our record + * of the line positions in the output we've + * written. + */ + off_t last_consumed_position = + state->transfer.last_output_position - state->transfer.written_but_not_consumed; + size_t lines_not_consumed = 0; + size_t line_from_end = 0; + + /* + * positions[head-1] = position of last separator written + * positions[head-2] = position of second last separator written + * etc + * + * We start at [head-1] and go backwards, wrapping + * around as it's a circular buffer, stopping at the + * length (number of positions stored), or when we + * have gone before the last consumed position. + */ + for (line_from_end = 0; line_from_end < state->transfer.line_positions_length; line_from_end++) { + size_t array_index; + array_index = + state->transfer.line_positions_head + state->transfer.line_positions_capacity - + line_from_end - 1; + while (array_index >= state->transfer.line_positions_capacity) + array_index -= state->transfer.line_positions_capacity; + if (state->transfer.line_positions[array_index] <= last_consumed_position) + break; + lines_not_consumed++; + } + + debug("%s: %lld -> %lld", "written_but_not_consumed bytes to lines", + (unsigned long long) (state->transfer.written_but_not_consumed), + (unsigned long long) lines_not_consumed); + + state->transfer.transferred -= lines_not_consumed; } /* diff --git a/src/pv/state.c b/src/pv/state.c index 3d202d8..9056568 100644 --- a/src/pv/state.c +++ b/src/pv/state.c @@ -95,6 +95,10 @@ void pv_state_reset(pvstate_t state) #ifdef HAVE_SPLICE state->transfer.splice_failed_fd = -1; #endif /* HAVE_SPLICE */ + + state->transfer.line_positions_length = 0; + state->transfer.line_positions_head = 0; + state->transfer.last_output_position = 0; } @@ -203,6 +207,10 @@ void pv_state_free(pvstate_t state) /*@+keeptrans@ */ /* splint - explicitly freeing this structure, so free() here is OK. */ + if (NULL != state->transfer.line_positions) + free(state->transfer.line_positions); + state->transfer.line_positions = NULL; + if (NULL != state->calc.history) free(state->calc.history); state->calc.history = NULL; diff --git a/src/pv/transfer.c b/src/pv/transfer.c index 7e37248..6e47991 100644 --- a/src/pv/transfer.c +++ b/src/pv/transfer.c @@ -726,6 +726,21 @@ static int pv__transfer_write(pvstate_t state, bool *eof_in, bool *eof_out, long char *ptr; long lines = 0; + /* + * Line mode - look through what we've just written + * to count how many lines there were. + */ + + /* Allocate buffer to remember line positions. */ + if (NULL == state->transfer.line_positions) { + state->transfer.line_positions_capacity = MAX_LINE_POSITIONS; + state->transfer.line_positions = + calloc((size_t) (state->transfer.line_positions_capacity), sizeof(off_t)); + if (NULL == state->transfer.line_positions) { + pv_error(state, "%s: alloc failed: %s", _("line positions"), strerror(errno)); + } + } + if (state->control.null_terminated_lines) { separator = '\0'; } else { @@ -735,9 +750,30 @@ static int pv__transfer_write(pvstate_t state, bool *eof_in, bool *eof_out, long ptr = (char *) (state->transfer.transfer_buffer + state->transfer.write_position - 1); for (ptr++; ptr - (char *) state->transfer.transfer_buffer - state->transfer.write_position < - (size_t) nwritten; ptr++) { - if (*ptr == separator) - ++lines; + (size_t) nwritten; ptr++, state->transfer.last_output_position++) { + if (*ptr != separator) + continue; + + /* Separator found - increment line count. */ + ++lines; + + if (NULL == state->transfer.line_positions) + continue; + + /* Store the position of the separator. */ + state->transfer.line_positions[state->transfer.line_positions_head] = + state->transfer.last_output_position; + state->transfer.line_positions_head++; + + /* Circular buffer - wrap around. */ + if (state->transfer.line_positions_head >= state->transfer.line_positions_capacity) { + state->transfer.line_positions_head = 0; + } + + /* Increment count of line positions, if below capacity. */ + if (state->transfer.line_positions_length < state->transfer.line_positions_capacity) { + state->transfer.line_positions_length++; + } } *lineswritten += lines;