implemented -U / --unlimited-rate-until -- which doesn't start limiting the rate until a given number of bytes have been initially transferred

README

@@ -228,4 +228,7 @@ Credit is also due to:
   Eric A. Borisch
     - provided details of compatibility fix for "%Lu" in watchpid code
 
+  Davy Durham
+	- implemented "-U" / "--unlimited-rate-until" option
+
 -----------------------------------------------------------------------------

README.md

@@ -227,8 +227,14 @@ Michiel Van Herwegen
 Erkki Seppälä <http://www.inside.org/~flux/>
 - provided patch implementing "-I"
 
+Davy Durham
+- implemented "-U" / "--unlimited-rate-until" option
+
 ####News
 
+1.7.0 - ?
+	new option "-U" / "--unlimited-rate-until" to disable rate limiting until a given number of bytes have been transferred first, allowing for an initial burst of data.
+
 1.6.6 - 30 June 2017  
 
 	(r161) use %llu instead of %Lu for better compatibility (Eric A. Borisch)

doc/NEWS

@@ -1,3 +1,8 @@
+1.7.0 - ?
+  - new option "-U" / "--unlimited-rate-until" to disable rate limiting until
+    a given number of bytes have been transferred first, allowing for an
+    initial burst of data.
+
 1.6.6 - 30 June 2017
   - (r161) use %llu instead of %Lu for better compatibility (Eric A. Borisch)
   - (r162) (#1532) fix target buffer size (-B) being ignored (AndCycle, Ilya

doc/quickref.1.in

@@ -306,6 +306,12 @@ Limit the transfer to a maximum of
 bytes per second.  A suffix of "K", "M", "G", or "T" can be added to denote
 kibibytes (*1024), mebibytes, and so on.
 .TP
+.B \-U BYTES, \-\-unlimited-rate-until BYTES
+Do not limit the rate of transfer until
+.B BYTES
+bytes have been initially transferred.  A suffix of "K", "M", "G", or "T" can
+be added to denote kibibytes (*1024), mebibytes, and so on.
+.TP
 .B \-B BYTES, \-\-buffer-size BYTES
 Use a transfer buffer size of
 .B BYTES

src/include/options.h

@@ -32,6 +32,7 @@ struct opts_s {           /* structure describing run-time options */
 	unsigned char null;            /* lines are null-terminated */
 	unsigned char no_op;           /* do nothing other than pipe data */
 	unsigned long long rate_limit; /* rate limit, in bytes per second */
+	unsigned long long unlimited_rate_until; /* don't limit the rate until this many bytes have been transferred */
 	unsigned long long buffer_size;/* buffer size, in bytes (0=default) */
 	unsigned int remote;           /* PID of pv to update settings of */
 	unsigned long long size;       /* total size of data */

src/include/pv-internal.h

@@ -69,6 +69,7 @@ struct pvstate_s {
 	unsigned char stop_at_size;      /* set if we stop at "size" bytes */
 	unsigned char no_splice;         /* never use splice() */
 	unsigned long long rate_limit;   /* rate limit, in bytes per second */
+	unsigned long long unlimited_rate_until; /* don't limit the rate until this many bytes have been transferred */
 	unsigned long long target_buffer_size;  /* buffer size (0=default) */
 	unsigned long long size;         /* total size of data */
 	double interval;                 /* interval between updates */

src/include/pv.h

@@ -83,6 +83,7 @@ extern void pv_state_no_op_set(pvstate_t, unsigned char);
 extern void pv_state_skip_errors_set(pvstate_t, unsigned char);
 extern void pv_state_stop_at_size_set(pvstate_t, unsigned char);
 extern void pv_state_rate_limit_set(pvstate_t, unsigned long long);
+extern void pv_state_unlimited_rate_until_set(pvstate_t, unsigned long long);
 extern void pv_state_target_buffer_size_set(pvstate_t, unsigned long long);
 extern void pv_state_no_splice_set(pvstate_t, unsigned char);
 extern void pv_state_size_set(pvstate_t, unsigned long long);

src/main/help.c

@@ -78,6 +78,8 @@ void display_help(void)
 		{"", 0, 0, 0},
 		{"-L", "--rate-limit", N_("RATE"),
 		 N_("limit transfer to RATE bytes per second")},
+		{"-U", "--unlimited-rate-until", N_("BYTES"),
+		 N_("do not start limiting transfer rate until BYTES have been transferred")},
 		{"-B", "--buffer-size", N_("BYTES"),
 		 N_("use a buffer size of BYTES")},
 		{"-C", "--no-splice", 0,

src/main/main.c

@@ -195,6 +195,7 @@ int main(int argc, char **argv)
 	pv_state_skip_errors_set(state, opts->skip_errors);
 	pv_state_stop_at_size_set(state, opts->stop_at_size);
 	pv_state_rate_limit_set(state, opts->rate_limit);
+	pv_state_unlimited_rate_until_set(state, opts->unlimited_rate_until);
 	pv_state_target_buffer_size_set(state, opts->buffer_size);
 	pv_state_no_splice_set(state, opts->no_splice);
 	pv_state_size_set(state, opts->size);

src/main/options.c

@@ -73,6 +73,7 @@ opts_t opts_parse(int argc, char **argv)
 		{"name", 1, 0, 'N'},
 		{"format", 1, 0, 'F'},
 		{"rate-limit", 1, 0, 'L'},
+		{"unlimited-rate-until", 1, 0, 'U'},
 		{"buffer-size", 1, 0, 'B'},
 		{"no-splice", 0, 0, 'C'},
 		{"skip-errors", 0, 0, 'E'},
@@ -85,7 +86,7 @@ opts_t opts_parse(int argc, char **argv)
 	int option_index = 0;
 #endif
 	char *short_options =
-	    "hVpteIrabTA:fnqcWD:s:l0i:w:H:N:F:L:B:CESR:P:d:";
+	    "hVpteIrabTA:fnqcWD:s:l0i:w:H:N:F:L:U:B:CESR:P:d:";
 	int c, numopts;
 	unsigned int check_pid;
 	int check_fd;
@@ -145,6 +146,7 @@ opts_t opts_parse(int argc, char **argv)
 		case 'w':
 		case 'H':
 		case 'L':
+		case 'U':
 		case 'B':
 		case 'R':
 			if (pv_getnum_check(optarg, PV_NUMTYPE_INTEGER) !=
@@ -284,6 +286,9 @@ opts_t opts_parse(int argc, char **argv)
 		case 'L':
 			opts->rate_limit = pv_getnum_ll(optarg);
 			break;
+		case 'U':
+			opts->unlimited_rate_until = pv_getnum_ll(optarg);
+			break;
 		case 'B':
 			opts->buffer_size = pv_getnum_ll(optarg);
 			break;

src/main/remote.c

@@ -35,6 +35,7 @@ struct remote_msg {
 	unsigned char bufpercent;	 /* transfer buffer percentage flag */
 	unsigned int lastwritten;	 /* last-written bytes count */
 	unsigned long long rate_limit;	 /* rate limit, in bytes per second */
+	unsigned long long unlimited_rate_until;	 /* don't limit the rate until this many bytes have been transferred */
 	unsigned long long buffer_size;	 /* buffer size, in bytes (0=default) */
 	unsigned long long size;	 /* total size of data */
 	double interval;		 /* interval between updates */
@@ -139,6 +140,7 @@ int pv_remote_set(opts_t opts)
 	msgbuf.bufpercent = opts->bufpercent;
 	msgbuf.lastwritten = opts->lastwritten;
 	msgbuf.rate_limit = opts->rate_limit;
+	msgbuf.unlimited_rate_until = opts->unlimited_rate_until;
 	msgbuf.buffer_size = opts->buffer_size;
 	msgbuf.size = opts->size;
 	msgbuf.interval = opts->interval;
@@ -266,6 +268,8 @@ void pv_remote_check(pvstate_t state)
 
 	if (msgbuf.rate_limit > 0)
 		pv_state_rate_limit_set(state, msgbuf.rate_limit);
+	if (msgbuf.unlimited_rate_until > 0)
+		pv_state_unlimited_rate_until_set(state, msgbuf.unlimited_rate_until);
 	if (msgbuf.buffer_size > 0) {
 		pv_state_target_buffer_size_set(state, msgbuf.buffer_size);
 	}

src/pv/loop.c

@@ -145,7 +145,15 @@ int pv_main_loop(pvstate_t state)
 		if (state->pv_sig_abort)
 			break;
 
-		if (state->rate_limit > 0) {
+		if (0 < state->unlimited_rate_until) {
+			/*
+			 * If we're supposed to limit but not until a certain number of
+			 * bytes have been transferred, then don't transfer a chunk larger
+			 * than what should be transferred before rate limited kicks in
+			 * so we can come back here and compute rate limiting parameters
+			 */
+			cansend = state->unlimited_rate_until;
+		} else if (state->rate_limit > 0) {
 			gettimeofday(&cur_time, NULL);
 			if ((cur_time.tv_sec > next_ratecheck.tv_sec)
 			    || (cur_time.tv_sec == next_ratecheck.tv_sec
@@ -168,7 +176,7 @@ int pv_main_loop(pvstate_t state)
 		if ((0 < state->size) && (state->stop_at_size)) {
 			if ((state->size < (total_written + cansend))
 			    || ((0 == cansend)
-				&& (0 == state->rate_limit))) {
+				&& (0 >= state->unlimited_rate_until && 0 == state->rate_limit))) {
 				cansend = state->size - total_written;
 				if (0 >= cansend) {
 					eof_in = 1;
@@ -195,15 +203,19 @@ int pv_main_loop(pvstate_t state)
 		if (state->linemode) {
 			since_last += lineswritten;
 			total_written += lineswritten;
-			if (state->rate_limit > 0)
+			if (0 >= state->unlimited_rate_until && state->rate_limit > 0)
 				target -= lineswritten;
 		} else {
 			since_last += written;
 			total_written += written;
-			if (state->rate_limit > 0)
+			if (0 >= state->unlimited_rate_until && state->rate_limit > 0)
 				target -= written;
 		}
 
+		if (state->unlimited_rate_until > 0) {
+			state->unlimited_rate_until -= written;
+		}
+
 		if (eof_in && eof_out && n < (state->input_file_count - 1)) {
 			n++;
 			fd = pv_next_file(state, n, fd);

src/pv/state.c

@@ -154,6 +154,11 @@ void pv_state_rate_limit_set(pvstate_t state, unsigned long long val)
 	state->rate_limit = val;
 };
 
+void pv_state_unlimited_rate_until_set(pvstate_t state, unsigned long long val)
+{
+	state->unlimited_rate_until = val;
+};
+
 void pv_state_target_buffer_size_set(pvstate_t state,
 				     unsigned long long val)
 {

src/pv/transfer.c

@@ -231,7 +231,7 @@ static int pv__transfer_read(pvstate_t state, int fd,
 	if ((!state->linemode) && (!state->no_splice)
 	    && (fd != state->splice_failed_fd)
 	    && (0 == state->to_write)) {
-		if (state->rate_limit || allowed != 0)
+		if ((0 >= state->unlimited_rate_until && state->rate_limit) || allowed != 0)
 			bytes_to_splice = allowed;
 		else
 			bytes_to_splice = bytes_can_read;
@@ -737,7 +737,7 @@ long pv_transfer(pvstate_t state, int fd, int *eof_in, int *eof_out,
 	 * write.
 	 */
 	state->to_write = state->read_position - state->write_position;
-	if ((state->rate_limit > 0) || (allowed > 0)) {
+	if ((0 >= state->unlimited_rate_until && state->rate_limit > 0) || (allowed > 0)) {
 		if (state->to_write > allowed) {
 			state->to_write = allowed;
 		}