Specifying a limit for total data transfer

cleanup.c

@@ -34,6 +34,7 @@ extern int protocol_version;
 extern int output_needs_newline;
 extern char *partial_dir;
 extern char *logfile_name;
+extern int data_transfer_limit_reached;
 
 int called_from_signal_handler = 0;
 BOOL shutting_down = False;
@@ -210,6 +211,8 @@ NORETURN void _exit_cleanup(int code, const char *file, int line)
 		if (exit_code == 0) {
 			if (code)
 				exit_code = code;
+			if (data_transfer_limit_reached)
+				exit_code = RERR_DATA_LIMIT;
 			if (io_error & IOERR_DEL_LIMIT)
 				exit_code = RERR_DEL_LIMIT;
 			if (io_error & IOERR_VANISHED)

errcode.h

@@ -43,6 +43,7 @@
 #define RERR_PARTIAL    23      /* partial transfer */
 #define RERR_VANISHED   24      /* file(s) vanished on sender side */
 #define RERR_DEL_LIMIT  25      /* skipped some deletes due to --max-delete */
+#define RERR_DATA_LIMIT 26      /* data transfer limit reached */
 
 #define RERR_TIMEOUT    30      /* timeout in data send/receive */
 #define RERR_CONTIMEOUT 35      /* timeout waiting for daemon connection */

flist.c

@@ -78,6 +78,7 @@ extern uid_t our_uid;
 extern struct stats stats;
 extern char *filesfrom_host;
 extern char *usermap, *groupmap;
+extern int data_transfer_limit_reached;
 
 extern struct name_num_item *file_sum_nni;
 
@@ -1714,6 +1715,9 @@ static void send_if_directory(int f, struct file_list *flist,
 {
 	char is_dot_dir = fbuf[ol-1] == '.' && (ol == 1 || fbuf[ol-2] == '/');
 
+	if (data_transfer_limit_reached)
+		return;
+
 	if (S_ISDIR(file->mode)
 	    && !(file->flags & FLAG_MOUNT_DIR) && f_name(file, fbuf)) {
 		void *save_filters;
@@ -1866,6 +1870,9 @@ static void send_directory(int f, struct file_list *flist, char *fbuf, int len,
 
 	assert(flist != NULL);
 
+	if (data_transfer_limit_reached)
+		return;
+
 	if (!(d = opendir(fbuf))) {
 		if (errno == ENOENT) {
 			if (am_sender) /* Can abuse this for vanished error w/ENOENT: */
@@ -1892,6 +1899,10 @@ static void send_directory(int f, struct file_list *flist, char *fbuf, int len,
 	for (errno = 0, di = readdir(d); di; errno = 0, di = readdir(d)) {
 		unsigned name_len;
 		char *dname = d_name(di);
+
+		if (data_transfer_limit_reached)
+			break;
+
 		if (dname[0] == '.' && (dname[1] == '\0'
 		    || (dname[1] == '.' && dname[2] == '\0')))
 			continue;
@@ -2050,6 +2061,9 @@ static void send1extra(int f, struct file_struct *file, struct file_list *flist)
 	int len, dlen, flags = FLAG_DIVERT_DIRS | FLAG_CONTENT_DIR;
 	size_t j;
 
+	if (data_transfer_limit_reached)
+		return;
+
 	f_name(file, fbuf);
 	dlen = strlen(fbuf);
 
@@ -2082,6 +2096,9 @@ static void send1extra(int f, struct file_struct *file, struct file_list *flist)
 		relnamecache *rnp = ((relnamecache**)relname_list->items)[j];
 		char name_type = rnp->name_type;
 
+		if (data_transfer_limit_reached)
+			break;
+
 		fbuf[dlen] = '/';
 		len = strlcpy(fbuf + dlen + 1, rnp->fname, sizeof fbuf - dlen - 1);
 		free(rnp);
@@ -2121,6 +2138,17 @@ static void write_end_of_flist(int f, int send_io_error)
 		write_byte(f, 0);
 }
 
+static void send_data_limit_flist_eof(int f)
+{
+	if (!flist_eof) {
+		write_ndx(f, NDX_FLIST_EOF);
+		flist_eof = 1;
+		if (DEBUG_GTE(FLIST, 3))
+			rprintf(FINFO, "[%s] flist_eof=1 due to data transfer limit\n", who_am_i());
+		change_local_filter_dir(NULL, 0, 0);
+	}
+}
+
 void send_extra_file_list(int f, int at_least)
 {
 	struct file_list *flist;
@@ -2131,6 +2159,11 @@ void send_extra_file_list(int f, int at_least)
 	if (flist_eof)
 		return;
 
+	if (data_transfer_limit_reached) {
+		send_data_limit_flist_eof(f);
+		return;
+	}
+
 	if (at_least < 0)
 		at_least = file_total - file_old_total + 1;
 
@@ -2142,6 +2175,11 @@ void send_extra_file_list(int f, int at_least)
 		const char *pathname = F_PATHNAME(file);
 		int32 *dp;
 
+		if (data_transfer_limit_reached) {
+			send_data_limit_flist_eof(f);
+			goto finish;
+		}
+
 		flist = flist_new(0, "send_extra_file_list");
 		start_write = stats.total_written;
 
@@ -2293,6 +2331,9 @@ struct file_list *send_file_list(int f, int argc, char *argv[])
 	while (1) {
 		char fbuf[MAXPATHLEN], *fn, name_type;
 
+		if (data_transfer_limit_reached)
+			break;
+
 		if (use_ff_fd) {
 			if (read_line(filesfrom_fd, fbuf, sizeof fbuf, rl_flags) == 0)
 				break;

generator.c

@@ -100,6 +100,21 @@ extern char *tmpdir;
 extern char *basis_dir[MAX_BASIS_DIRS+1];
 extern struct file_list *cur_flist, *first_flist, *dir_flist;
 extern filter_rule_list filter_list, daemon_filter_list;
+static int64 total_transferred_data = 0;
+extern int data_transfer_limit_reached;
+static int data_transfer_limit_msg_sent = 0;
+
+static int file_will_exceed_data_transfer_limit(struct file_struct *file)
+{
+	if (data_transfer_limit < 0)
+		return 0;
+
+	if (delay_transfer_limit_check)
+		return total_transferred_data >= data_transfer_limit;
+
+	return total_transferred_data + F_LENGTH(file) > data_transfer_limit;
+}
+
 
 int maybe_ATTRS_REPORT = 0;
 int maybe_ATTRS_ACCURATE_TIME = 0;
@@ -1245,6 +1260,15 @@ static void recv_generator(char *fname, struct file_struct *file, int ndx,
 	if (DEBUG_GTE(GENR, 1))
 		rprintf(FINFO, "recv_generator(%s,%d)\n", fname, ndx);
 
+
+	if (data_transfer_limit_reached) {
+#ifdef SUPPORT_HARD_LINKS
+		if (preserve_hard_links && F_IS_HLINKED(file))
+			handle_skipped_hlink(file, itemizing, code, f_out);
+#endif
+		return;
+	}
+
 	if (list_only) {
 		if (is_dir < 0
 		 || (is_dir && !implied_dirs && file->flags & FLAG_IMPLIED_DIR))
@@ -1929,7 +1953,19 @@ static void recv_generator(char *fname, struct file_struct *file, int ndx,
 	if (DEBUG_GTE(DELTASUM, 2))
 		rprintf(FINFO, "generating and sending sums for %d\n", ndx);
 
-  notify_others:
+notify_others:
+	if (file_will_exceed_data_transfer_limit(file)) {
+		rprintf(FINFO,
+			"recv_generator: transferring %s would exceed data transfer limit. Not transferred; stopping all remaining transfers.\n",
+			fname);
+		data_transfer_limit_reached = 1;
+		if (!data_transfer_limit_msg_sent) {
+			send_msg_int(MSG_DATA_LIMIT_REACHED, 0);
+			io_flush(MSG_FLUSH);
+			data_transfer_limit_msg_sent = 1;
+		}
+		goto cleanup;
+	}
 	if (remove_source_files && !delay_updates && !phase && !dry_run)
 		increment_active_files(ndx, itemizing, code);
 	if (inc_recurse && (!dry_run || write_batch < 0))
@@ -1952,6 +1988,8 @@ static void recv_generator(char *fname, struct file_struct *file, int ndx,
 		free_stat_x(&real_sx);
 	}
 
+	total_transferred_data += F_LENGTH(file);
+
 	if (!do_xfers) {
 #ifdef SUPPORT_HARD_LINKS
 		if (preserve_hard_links && F_IS_HLINKED(file))
@@ -2174,8 +2212,7 @@ void check_for_finished_files(int itemizing, enum logcode code, int check_redo)
 			continue;
 		}
 #endif
-
-		if (check_redo && (ndx = get_redo_num()) != -1) {
+		if (!data_transfer_limit_reached && check_redo && (ndx = get_redo_num()) != -1) {
 			OFF_T save_max_size = max_size;
 			OFF_T save_min_size = min_size;
 			csum_length = SUM_LENGTH;
@@ -2370,7 +2407,7 @@ void generate_files(int f_out, const char *local_name)
 		}
 	} while ((cur_flist = cur_flist->next) != NULL);
 
-	if (delete_during)
+	if (delete_during && !data_transfer_limit_reached)
 		delete_in_dir(NULL, NULL, dev_zero);
 	phase++;
 	if (DEBUG_GTE(GENR, 1))
@@ -2425,9 +2462,9 @@ void generate_files(int f_out, const char *local_name)
 	info_levels[INFO_FLIST] = save_info_flist;
 	info_levels[INFO_PROGRESS] = save_info_progress;
 
-	if (delete_during == 2)
+	if (delete_during == 2 && !data_transfer_limit_reached)
 		do_delayed_deletions(fbuf);
-	if (delete_after && !solo_file && file_total > 0)
+	if (delete_after && !solo_file && file_total > 0 && !data_transfer_limit_reached)
 		do_delete_pass();
 
 	if (max_delete >= 0 && skipped_deletes) {