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

@@ -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**

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

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;
 		}
 
 		/*

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;

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;