Merge branch 'fractional_rate_limit' (#193).

docs/pv.1

@@ -267,6 +267,9 @@ If the file already exists, it will be truncated.
 .BI \-L\  RATE \fR,\ \fB\-\-rate-limit\  RATE
 Limit the transfer to a maximum of \fIRATE\fR bytes per second.
 The same suffixes as \*(lq\fB\-\-size\fR\*(rq can be used.
+Decimal values are allowed.
+Specifying 0 will turn off rate limiting, which is useful with
+\*(lq\fB\-\-remote\fR\*(rq.
 .TP
 .BI \-B\  BYTES \fR,\ \fB\-\-buffer-size\  BYTES
 Use a transfer buffer size of \fIBYTES\fR bytes.

docs/pv.1.md

@@ -276,7 +276,9 @@ are explicitly switched on will be shown.
 **-L RATE, \--rate-limit RATE**
 
 :   Limit the transfer to a maximum of *RATE* bytes per second. The same
-    suffixes as "**\--size**" can be used.
+    suffixes as "**\--size**" can be used. Decimal values are allowed.
+    Specifying 0 will turn off rate limiting, which is useful with
+    "**\--remote**".
 
 **-B BYTES, \--buffer-size BYTES**
 

src/include/options.h

@@ -70,7 +70,7 @@ struct opts_s {
 	/*@keep@*/ /*@null@*/ int *watchfd_fd;  /* array of fds to watch in each one (0=all) */
 	/*@keep@*/ /*@null@*/ argv_string *argv;   /* array of non-option arguments */
 	size_t lastwritten;            /* show N bytes last written */
-	off_t rate_limit;              /* rate limit, in bytes per second */
+	long double 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 */
@@ -114,6 +114,8 @@ struct opts_s {
 	bool show_stats;	       /* set to write statistics at the end */
 	bool width_set_manually;       /* width was set manually, not detected */
 	bool height_set_manually;      /* height was set manually, not detected */
+	bool rate_limit_specified;     /* whether a rate limit value was given */
+	bool rate_limit_active;	       /* whether rate limiting is in effect (>0) */
 };
 
 /*@-exportlocal@*/

src/include/pv-internal.h

@@ -24,7 +24,7 @@ extern "C" {
 #endif
 
 #define RATE_GRANULARITY	100000000	 /* nsec between -L rate chunks */
-#define RATE_BURST_WINDOW	5	 	 /* rate burst window (multiples of rate) */
+#define RATE_BURST_WINDOW	5.0	 	 /* rate burst window (multiples of rate) */
 #define REMOTE_INTERVAL		100000000	 /* nsec between checks for -R and -Q */
 #define MONITOR_EXCHANGE_INTERVAL 100000000	 /* nsec between "-M both" data exchanges */
 #define BUFFER_SIZE		(size_t) 409600	 /* default transfer buffer size */
@@ -202,7 +202,7 @@ struct pvstate_s {
 		/*@null@*/ char *output_name;    /* name of the output, for diagnostics */
 		/*@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 */
+		long double rate_limit;          /* rate limit, in bytes per second */
 		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 */
@@ -237,6 +237,8 @@ struct pvstate_s {
 		bool show_stats;		 /* show statistics on exit */
 		bool width_set_manually;	 /* width was set manually, not detected */
 		bool height_set_manually;	 /* height was set manually, not detected */
+		bool rate_limit_specified;	 /* whether a rate limit value was given */
+		bool rate_limit_active;		 /* whether a rate limit is in effect (>0) */
 	} control;
 
 	/*******************

src/include/pv.h

@@ -72,8 +72,10 @@ extern double pv_getnum_interval(const char *);
  * Return the numeric value of a string representing a size, interpreting
  * suffixes in decimal units (multiples of 1000) instead of multiples of
  * 1024 if the second argument is true.
+ *
+ * Optionally also places the result in a long double.
  */
-extern off_t pv_getnum_size(const char *, bool);
+extern off_t pv_getnum_size(const char *, bool, /*@null@ */ long double *);
 
 /*
  * Return the numeric value of a string representing a count such as screen
@@ -264,7 +266,7 @@ extern void pv_state_stop_at_size_set(pvstate_t, bool);
 extern void pv_state_sync_after_write_set(pvstate_t, bool);
 extern void pv_state_direct_io_set(pvstate_t, bool);
 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_rate_limit_set(pvstate_t, long double, bool, bool);
 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);

src/main/main.c

@@ -903,7 +903,7 @@ int main(int argc, char **argv)
 	pv_state_sync_after_write_set(state, opts->sync_after_write);
 	pv_state_direct_io_set(state, opts->direct_io);
 	pv_state_discard_input_set(state, opts->discard_input);
-	pv_state_rate_limit_set(state, opts->rate_limit);
+	pv_state_rate_limit_set(state, opts->rate_limit, opts->rate_limit_specified, opts->rate_limit_active);
 	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);

src/main/options.c

@@ -35,6 +35,9 @@
 #ifdef HAVE_FTW_H
 #include <ftw.h>
 #endif
+#if HAVE_MATH_H
+#include <math.h>
+#endif
 
 
 void display_help(void);
@@ -798,6 +801,7 @@ opts_t opts_parse(unsigned int argc, char **argv)
 
 	opts->width_set_manually = false;
 	opts->height_set_manually = false;
+	opts->rate_limit_specified = false;
 
 	do {
 #ifdef HAVE_GETOPT_LONG
@@ -1005,7 +1009,7 @@ opts_t opts_parse(unsigned int argc, char **argv)
 		case 's':
 			if ('@' != *optarg) {
 				/* A number was passed, not "@<filename>". */
-				opts->size = pv_getnum_size(optarg, opts->decimal_units);
+				opts->size = pv_getnum_size(optarg, opts->decimal_units, NULL);
 			} else {
 				/* Permit "@<filename>". */
 				const char *size_file = 1 + optarg;
@@ -1054,10 +1058,24 @@ opts_t opts_parse(unsigned int argc, char **argv)
 			}
 			break;
 		case 'L':
-			opts->rate_limit = pv_getnum_size(optarg, opts->decimal_units);
+			(void) pv_getnum_size(optarg, opts->decimal_units, &(opts->rate_limit));
+			opts->rate_limit_specified = true;
+#if HAVE_MATH_H
+			/*@-unrecog@ *//* splint doesn't know nextafterl(). */
+			if (opts->rate_limit < nextafterl(0, INFINITY)) {
+				opts->rate_limit_active = false;
+			} else {
+				opts->rate_limit_active = true;
+			}
+			/*@+unrecog@ */
+#else
+			opts->rate_limit_active = false;
+			if (opts->rate_limit > 0.00001)
+				opts->rate_limit_active = true;
+#endif
 			break;
 		case 'B':
-			opts->buffer_size = (size_t) pv_getnum_size(optarg, opts->decimal_units);
+			opts->buffer_size = (size_t) pv_getnum_size(optarg, opts->decimal_units, NULL);
 			opts->no_splice = true;
 			break;
 		case 'C':
@@ -1067,10 +1085,10 @@ opts_t opts_parse(unsigned int argc, char **argv)
 			opts->skip_errors++;
 			break;
 		case 'Z':
-			opts->error_skip_block = pv_getnum_size(optarg, opts->decimal_units);
+			opts->error_skip_block = pv_getnum_size(optarg, opts->decimal_units, NULL);
 			break;
 		case 'J':
-			opts->pipe_buffer_size = (size_t) pv_getnum_size(optarg, opts->decimal_units);
+			opts->pipe_buffer_size = (size_t) pv_getnum_size(optarg, opts->decimal_units, NULL);
 			break;
 		case 'S':
 			opts->stop_at_size = true;
@@ -1214,7 +1232,7 @@ opts_t opts_parse(unsigned int argc, char **argv)
 	if (PV_ACTION_WATCHFD == opts->action) {
 		if (opts->linemode || opts->null_terminated_lines || opts->stop_at_size
 		    || (opts->skip_errors > 0) || (opts->buffer_size > 0)
-		    || (opts->rate_limit > 0)) {
+		    || (opts->rate_limit_active)) {
 			/*@-mustfreefresh@ *//* see above */
 			pv_error("%s",
 				 _("cannot use line mode or transfer modifier options when watching file descriptors"));

src/pv/loop.c

@@ -348,8 +348,8 @@ int pv_main_loop(pvstate_t state)
 {
 	long lineswritten;
 	off_t cansend;
+	long double rate_limited_target;
 	ssize_t written;
-	long double target;
 	bool eof_in, eof_out, final_update;
 	struct timespec start_time, next_update, next_ratecheck, cur_time;
 	struct timespec next_remotecheck, next_monitor_exchange;
@@ -443,7 +443,7 @@ int pv_main_loop(pvstate_t state)
 		pv_elapsedtime_add_nsec(&next_update, (long long) (1000000000.0 * state->control.interval));
 	}
 
-	target = 0;
+	rate_limited_target = 0.0;
 	final_update = false;
 	file_idx = 0;
 
@@ -572,20 +572,26 @@ int pv_main_loop(pvstate_t state)
 		if (1 == state->flags.trigger_exit)
 			break;
 
-		if (state->control.rate_limit > 0) {
+		if (state->control.rate_limit_active) {
 			pv_elapsedtime_read(&cur_time);
-			if (pv_elapsedtime_compare(&cur_time, &next_ratecheck) > 0) {
-				target +=
+			while (pv_elapsedtime_compare(&cur_time, &next_ratecheck) > 0) {
+				rate_limited_target +=
 				    ((long double) (state->control.rate_limit)) / (long double) (1000000000.0 /
 												 (long double)
 												 (RATE_GRANULARITY));
 				long double burst_max = ((long double) (state->control.rate_limit * RATE_BURST_WINDOW));
-				if (target > burst_max) {
-					target = burst_max;
+				if (burst_max > 1.0 && rate_limited_target > burst_max) {
+					/*
+					 * If the burst max is < 1 then
+					 * capping to that will mean nothing
+					 * ever gets sent, so turn off the
+					 * burst limit at 1 and below.
+					 */
+					rate_limited_target = burst_max;
 				}
 				pv_elapsedtime_add_nsec(&next_ratecheck, RATE_GRANULARITY);
 			}
-			cansend = (off_t) target;
+			cansend = (off_t) (rate_limited_target);
 		}
 
 		/*
@@ -595,7 +601,7 @@ int pv_main_loop(pvstate_t state)
 		if ((0 < state->control.size) && (state->control.stop_at_size)) {
 			if ((state->control.size < (state->transfer.total_written + cansend))
 			    || ((0 == cansend)
-				&& (0 == state->control.rate_limit))) {
+				&& (!state->control.rate_limit_active))) {
 				cansend = state->control.size - state->transfer.total_written;
 				if (0 >= cansend) {
 					debug("%s", "write limit reached (size explicitly set) - setting EOF flags");
@@ -622,12 +628,12 @@ int pv_main_loop(pvstate_t state)
 
 		if (state->control.linemode) {
 			state->transfer.total_written += lineswritten;
-			if (state->control.rate_limit > 0)
-				target -= lineswritten;
+			if (state->control.rate_limit_active)
+				rate_limited_target -= lineswritten;
 		} else {
 			state->transfer.total_written += written;
-			if (state->control.rate_limit > 0)
-				target -= written;
+			if (state->control.rate_limit_active)
+				rate_limited_target -= written;
 		}
 
 #ifdef FIONREAD

src/pv/number.c

@@ -29,8 +29,11 @@ bool pv_isdigit(char c)
  * a fractional part, optionally followed by a units suffix such as "K" for
  * kibibytes.  If "decimal_units" is true, suffixes are interpreted as
  * multiples of 1000, rather than multiples of 1024.
+ *
+ * If resultptr is not NULL, it is populated with the result as a long
+ * double.
  */
-off_t pv_getnum_size(const char *str, bool decimal_units)
+off_t pv_getnum_size(const char *str, bool decimal_units, /*@null@ */ long double *resultptr)
 {
 	off_t integral_part = 0;
 	off_t fractional_part = 0;
@@ -39,8 +42,11 @@ off_t pv_getnum_size(const char *str, bool decimal_units)
 	off_t decimal_multiplier = 0;
 	size_t readpos = 0;
 
-	if (NULL == str)
+	if (NULL == str) {
+		if (NULL != resultptr)
+			*resultptr = 0.0;
 		return (off_t) 0;
+	}
 
 	/* Skip any non-numeric leading characters. */
 	while (str[readpos] != '\0' && (!pv_isdigit(str[readpos])))
@@ -156,10 +162,19 @@ off_t pv_getnum_size(const char *str, bool decimal_units)
 	 * Add the fractional part, divided by its divisor, to the integral
 	 * part, now that the multiplier for the units has been applied.
 	 */
+	if (NULL != resultptr) {
+		*resultptr = ((long double) fractional_part) / (long double) fractional_divisor;
+		*resultptr += (long double) integral_part;
+	}
 	fractional_part = fractional_part / fractional_divisor;
 	integral_part += fractional_part;
 
-	debug("%s [%s] = %lld", str, decimal_units ? "decimal" : "binary", (long long) integral_part);
+	if (NULL != resultptr) {
+		debug("%s [%s] = %lld = %Lf", str, decimal_units ? "decimal" : "binary", (long long) integral_part,
+		      *resultptr);
+	} else {
+		debug("%s [%s] = %lld", str, decimal_units ? "decimal" : "binary", (long long) integral_part);
+	}
 
 	return integral_part;
 }
@@ -211,7 +226,7 @@ double pv_getnum_interval(const char *str)
  */
 unsigned int pv_getnum_count(const char *str, bool decimal_units)
 {
-	return (unsigned int) pv_getnum_size(str, decimal_units);
+	return (unsigned int) pv_getnum_size(str, decimal_units, NULL);
 }
 
 

src/pv/remote.c

@@ -34,7 +34,7 @@ struct remote_msg {
 	bool bytes;			 /* bytes transferred flag */
 	bool bufpercent;		 /* transfer buffer percentage flag */
 	size_t lastwritten;		 /* last-written bytes count */
-	off_t rate_limit;		 /* rate limit, in bytes per second */
+	long double rate_limit;		 /* rate limit, in bytes per second */
 	size_t buffer_size;		 /* buffer size, in bytes (0=default) */
 	off_t size;			 /* total size of data */
 	double interval;		 /* interval between updates */
@@ -42,6 +42,8 @@ struct remote_msg {
 	unsigned int height;		 /* screen height */
 	bool width_set_manually;	 /* width was set manually, not detected */
 	bool height_set_manually;	 /* height was set manually, not detected */
+	bool rate_limit_specified;	 /* whether a rate limit value was set */
+	bool rate_limit_active;		 /* whether rate limiting is in effect (>0) */
 	char name[256];			 /* flawfinder: ignore */
 	char format[256];		 /* flawfinder: ignore */
 	char extra_display[256];	 /* flawfinder: ignore */
@@ -106,6 +108,8 @@ int pv_remote_set(pvstate_t state, pid_t remote)
 	msgbuf.bufpercent = state->control.format_option.bufpercent;
 	msgbuf.lastwritten = state->control.format_option.lastwritten;
 	msgbuf.rate_limit = state->control.rate_limit;
+	msgbuf.rate_limit_specified = state->control.rate_limit_specified;
+	msgbuf.rate_limit_active = state->control.rate_limit_active;
 	msgbuf.buffer_size = state->control.target_buffer_size;
 	msgbuf.size = state->control.size;
 	msgbuf.interval = state->control.interval;
@@ -322,8 +326,9 @@ static bool pv__rxsignal_usr2(pvstate_t state)
 
 	pv_state_set_format_options(state, format_options);
 
-	if (msgbuf.rate_limit > 0)
-		pv_state_rate_limit_set(state, msgbuf.rate_limit);
+	if (msgbuf.rate_limit_specified)
+		pv_state_rate_limit_set(state, msgbuf.rate_limit, msgbuf.rate_limit_specified,
+					msgbuf.rate_limit_active);
 	if (msgbuf.buffer_size > 0) {
 		pv_state_target_buffer_size_set(state, msgbuf.buffer_size);
 	}

src/pv/state.c

@@ -649,9 +649,19 @@ void pv_state_discard_input_set(pvstate_t state, bool val)
 	state->control.discard_input = val;
 }
 
-void pv_state_rate_limit_set(pvstate_t state, off_t val)
+/*
+ * If "was_specified" is true, the limit was specified on the command line.
+ * This is used with --remote to determine whether to affect the remote
+ * process's rate limit settings.
+ *
+ * if "is_active" is true, the limit ("val") is not zero so rate limiting is
+ * active.
+ */
+void pv_state_rate_limit_set(pvstate_t state, long double val, bool was_specified, bool is_active)
 {
 	state->control.rate_limit = val;
+	state->control.rate_limit_specified = was_specified;
+	state->control.rate_limit_active = is_active;
 }
 
 void pv_state_target_buffer_size_set(pvstate_t state, size_t val)

src/pv/transfer.c

@@ -300,11 +300,11 @@ static ssize_t pv__transfer__write_repeated(int fd, char *buf, size_t count, boo
  *
  * At most, the number of bytes read will be the number of bytes remaining
  * in the input buffer, capped to the number of bytes left until
- * state->control.size is reached if state->control.stop_at_size is true.
- * If state->control.rate_limit is >0, and/or "max_to_write" is >0, and
- * splice() is used, then the maximum number of bytes read will be further
- * capped to the value of "max_to_write", since splice() writes as well as
- * reads.
+ * state->control.size is reached if state->control.stop_at_size is true. 
+ * If state->control.rate_limit_active is true, and/or "max_to_write" is >0,
+ * and splice() is used, then the maximum number of bytes read will be
+ * further capped to the value of "max_to_write", since splice() writes as
+ * well as reads.
  *
  * If splice() was successfully used, sets state->transfer.splice_used to
  * true; if it was unsuccessfully used, then
@@ -380,7 +380,7 @@ static bool pv__transfer_read(pvstate_t state, int fd, bool *eof_in, bool *eof_o
 	    && (0 == state->transfer.to_write)) {
 		size_t bytes_to_splice;
 
-		if (state->control.rate_limit > 0 || max_to_write != 0) {
+		if (state->control.rate_limit_active || max_to_write != 0) {
 			bytes_to_splice = (size_t) max_to_write;
 		} else {
 			bytes_to_splice = bytes_can_read;
@@ -1163,10 +1163,10 @@ static char *pv__allocate_aligned_buffer(int outfd, int infd, size_t target_size
 
 /*
  * Transfer some data from "fd" to standard output, timing out after 9/100
- * of a second.  If state->control.rate_limit is >0, and/or "allowed" is >0,
- * only up to "allowed" bytes can be written.  The variables that "eof_in"
- * and "eof_out" point to are used to flag that we've finished reading and
- * writing respectively.
+ * of a second.  If state->control.rate_limit_active is true, and/or
+ * "allowed" is >0, only up to "allowed" bytes can be written.  The
+ * variables that "eof_in" and "eof_out" point to are used to flag that
+ * we've finished reading and writing respectively.
  *
  * Returns the number of bytes written, or negative on error (in which case
  * state->status.exit_status is updated).  In line mode, the number of lines
@@ -1293,7 +1293,7 @@ ssize_t pv_transfer(pvstate_t state, int fd, bool *eof_in, bool *eof_out, off_t
 	 * limiting is active or if "allowed" is > 0.
 	 */
 	state->transfer.to_write = (ssize_t) (state->transfer.read_position - state->transfer.write_position);
-	if ((state->control.rate_limit > 0) || (allowed > 0)) {
+	if ((state->control.rate_limit_active) || (allowed > 0)) {
 		if ((off_t) (state->transfer.to_write) > allowed) {
 			state->transfer.to_write = (ssize_t) allowed;
 		}