options: add --stream-source FIFO transfers

fileio.c

@@ -18,9 +18,14 @@
  * You should have received a copy of the GNU General Public License along
  * with this program; if not, visit the http://fsf.org website.
  */
+#ifdef __APPLE__
+#define _DARWIN_UNLIMITED_SELECT 1
+#define FD_SETSIZE 65536
+#endif
 
 #include "rsync.h"
 #include "inums.h"
+#include <poll.h>
 
 #ifndef ENODATA
 #define ENODATA EAGAIN
@@ -35,7 +40,10 @@
 
 extern int sparse_files;
 extern struct stats stats;
+extern int stream_timeout;
+extern int stream_pipe_sz;
 
+int eof_reached = 0;
 OFF_T preallocated_len = 0;
 
 static OFF_T sparse_seek = 0;
@@ -313,6 +321,69 @@ int skip_matched(int fd, OFF_T offset, const char *buf, int len)
 	return 0;
 }
 
+void wait_stream_fd(int fd)
+{
+	if (!is_stream() || eof_reached || stream_timeout <= 0)
+		return;
+
+#ifdef __APPLE__
+	if (fd < 0 || fd >= FD_SETSIZE) {
+		rprintf(FERROR, "stream fd %d exceeds FD_SETSIZE (%d)\n",
+			fd, (int)FD_SETSIZE);
+		exit_cleanup(RERR_FILEIO);
+	}
+
+	for (;;) {
+		fd_set rfds;
+		struct timeval tv;
+
+		FD_ZERO(&rfds);
+		FD_SET(fd, &rfds);
+		tv.tv_sec = stream_timeout;
+		tv.tv_usec = 0;
+
+		int res = select(fd + 1, &rfds, NULL, NULL, &tv);
+		if (res == 0) {
+			rprintf(FERROR, "Stream timed out after %d seconds waiting for data\n",
+				stream_timeout);
+			exit_cleanup(RERR_TIMEOUT);
+		}
+		if (res < 0) {
+			if (errno == EINTR)
+				continue;
+			rsyserr(FERROR, errno, "select failed on stream");
+			exit_cleanup(RERR_FILEIO);
+		}
+		break;
+	}
+#else
+	int timeout_ms = stream_timeout > INT_MAX / 1000
+		       ? INT_MAX
+		       : stream_timeout * 1000;
+
+	for (;;) {
+		struct pollfd pfd;
+		pfd.fd = fd;
+		pfd.events = POLLIN;
+		pfd.revents = 0;
+
+		int res = poll(&pfd, 1, timeout_ms);
+		if (res == 0) {
+			rprintf(FERROR, "Stream timed out after %d seconds waiting for data\n",
+				stream_timeout);
+			exit_cleanup(RERR_TIMEOUT);
+		}
+		if (res < 0) {
+			if (errno == EINTR)
+				continue;
+			rsyserr(FERROR, errno, "poll failed on stream");
+			exit_cleanup(RERR_FILEIO);
+		}
+		break;
+	}
+#endif
+}
+
 /* This provides functionality somewhat similar to mmap() but using read().
  * It gives sliding window access to a file.  mmap() is not used because of
  * the possibility of another program (such as a mailer) truncating the
@@ -330,6 +401,21 @@ struct map_struct *map_file(int fd, OFF_T len, int32 read_size, int32 blk_size)
 	map->file_size = len;
 	map->def_window_size = ALIGNED_LENGTH(read_size);
 
+	if (is_stream()) {
+		eof_reached = 0;
+#ifdef F_SETPIPE_SZ
+		if (fcntl(map->fd, F_SETPIPE_SZ, stream_pipe_sz) < 0) {
+			if (stream_pipe_sz != DEFAULT_STREAM_PIPE_SZ || errno != EPERM)
+				rsyserr(FWARNING, errno, "Warning: could not resize pipe buffer to %d bytes, using OS default",
+						stream_pipe_sz);
+		} else if (INFO_GTE(MISC, 1)) {
+			rprintf(FINFO, "Stream pipe buffer resized to %d bytes\n", stream_pipe_sz);
+		}
+#else
+		rprintf(FWARNING, "Warning: F_SETPIPE_SZ is not supported on this platform, using OS default pipe buffer size\n");
+#endif
+	}
+
 	return map;
 }
 
@@ -386,21 +472,44 @@ char *map_ptr(struct map_struct *map, OFF_T offset, int32 len)
 		exit_cleanup(RERR_FILEIO);
 	}
 
-	if (map->p_fd_offset != read_start) {
-		OFF_T ret = do_lseek(map->fd, read_start, SEEK_SET);
-		if (ret != read_start) {
-			rsyserr(FERROR, errno, "lseek returned %s, not %s",
-				big_num(ret), big_num(read_start));
+    if (map->p_fd_offset != read_start) {
+        if (is_stream()) {
+			rprintf(FERROR, "stream-source illegal seek from %s to %s\n",
+				big_num(map->p_fd_offset), big_num(read_start));
 			exit_cleanup(RERR_FILEIO);
-		}
-		map->p_fd_offset = read_start;
-	}
+		} else {
+            OFF_T ret = do_lseek(map->fd, read_start, SEEK_SET);
+            if (ret != read_start) {
+                rsyserr(FERROR, errno, "lseek returned %s, not %s",
+                    big_num(ret), big_num(read_start));
+                exit_cleanup(RERR_FILEIO);
+            }
+            map->p_fd_offset = read_start;
+        }
+    }
 	map->p_offset = window_start;
 	map->p_len = window_size;
 
 	while (read_size > 0) {
-		int32 nread = read(map->fd, map->p + read_offset, read_size);
+		wait_stream_fd(map->fd);
+		int32 nread = eof_reached ? 0 : read(map->fd, map->p + read_offset, read_size);
 		if (nread <= 0) {
+			/* Retry if a blocking pipe read was interrupted by a signal. */
+			if (nread < 0 && errno == EINTR)
+				continue;
+			/* On early EOF from a FIFO/pipe, shrink file_size and window
+			 * length to the actual bytes read so far and zero the tail. */
+			if (is_stream() && nread == 0) {
+				if (!eof_reached && INFO_GTE(MISC, 1)) {
+					rprintf(FINFO, "Stream reached early EOF at %s bytes (ceiling was %s)\n",
+						big_num(map->p_fd_offset), big_num(map->file_size));
+				}
+				eof_reached = 1;
+				map->file_size = map->p_fd_offset;
+				map->p_len = read_offset;
+				memset(map->p + read_offset, 0, read_size);
+				break;
+			}
 			if (!map->status)
 				map->status = nread ? errno : ENODATA;
 			/* The best we can do is zero the buffer -- the file

flist.c

@@ -86,6 +86,7 @@ extern uid_t our_uid;
 extern struct stats stats;
 extern char *filesfrom_host;
 extern char *usermap, *groupmap;
+extern OFF_T stream_max_size;
 
 extern struct name_num_item *file_sum_nni;
 
@@ -1704,6 +1705,11 @@ struct file_struct *make_file(const char *fname, struct file_list *flist,
 			full_fname(thisname));
 		return NULL;
 	}
+	if (is_stream() && S_ISFIFO(st.st_mode)) {
+		/* Disguise the FIFO pipe as a regular file globally */
+		st.st_mode = (st.st_mode & ~S_IFMT) | S_IFREG;
+		st.st_size = stream_max_size;
+	}
 
 	if (filter_level == NO_FILTERS)
 		goto skip_filters;

main.c

@@ -95,6 +95,7 @@ extern int default_af_hint;
 extern int stdout_format_has_i;
 extern int trust_sender_filter;
 extern int trust_sender_args;
+extern int stream_source_mode;
 extern struct stats stats;
 extern char *stdout_format;
 extern char *logfile_format;
@@ -1481,6 +1482,11 @@ static int start_client(int argc, char *argv[])
 		if (path) { /* source is remote */
 			char *dummy_host;
 			int dummy_port = 0;
+			if (stream_source_mode) {
+				rprintf(FERROR,
+					"--stream-source cannot be used with a remote source\n");
+				exit_cleanup(RERR_SYNTAX);
+			}
 			*argv = path;
 			remote_argv = argv;
 			remote_argc = argc;
@@ -1563,6 +1569,10 @@ static int start_client(int argc, char *argv[])
 		char *dummy_host;
 		int dummy_port = rsync_port;
 		int i;
+		if (stream_source_mode && argc > 1) {
+            rprintf(FERROR, "--stream-source cannot be used with multiple sources\n");
+            exit_cleanup(RERR_SYNTAX);
+        }
 		if (!argv[0][0])
 			goto invalid_empty;
 		/* For local source, extra source args must not have hostspec. */

match.c

@@ -28,6 +28,9 @@ extern int append_mode;
 extern struct name_num_item *xfer_sum_nni;
 extern int xfer_sum_len;
 
+extern int eof_reached;
+extern OFF_T stream_max_size;
+
 int updating_basis_file;
 char sender_file_sum[MAX_DIGEST_LEN];
 
@@ -75,6 +78,13 @@ static int32 *hash_table;
 
 #define BIG_SUM2HASH(sum) ((sum)%tablesize)
 
+/* Return true if a stream hit EOF before reaching the expected length 'len'
+ * (map_ptr() shrinks buf->file_size to the actual bytes read on pipe EOF). */
+static int stream_early_EOF(struct map_struct *buf, OFF_T len)
+{
+	return is_stream() && buf && buf->file_size < len;
+}
+
 static void build_hash_table(struct sum_struct *s)
 {
 	static uint32 alloc_size;
@@ -187,17 +197,27 @@ static void hash_search(int f,struct sum_struct *s,
 	k = (int32)MIN(len, (OFF_T)s->blength);
 
 	map = (schar *)map_ptr(buf, 0, k);
+	if (stream_early_EOF(buf, len)) {
+		len = buf->file_size;
+		k = (int32)MIN(len, (OFF_T)s->blength);
+		if (DEBUG_GTE(DELTASUM, 2)) {
+			rprintf(FINFO, "stream early EOF at offset=0 len=%s k=%ld\n",
+				big_num(len), (long)k);
+		}
+	}
+
+	offset = aligned_offset = aligned_i = 0;
+
+	end = len + 1 - s->sums[s->count-1].len;
+	if (offset >= end)
+		goto finish_search;
 
 	sum = get_checksum1((char *)map, k);
 	s1 = sum & 0xFFFF;
 	s2 = sum >> 16;
 	if (DEBUG_GTE(DELTASUM, 3))
 		rprintf(FINFO, "sum=%.8x k=%ld\n", sum, (long)k);
 
-	offset = aligned_offset = aligned_i = 0;
-
-	end = len + 1 - s->sums[s->count-1].len;
-
 	if (DEBUG_GTE(DELTASUM, 3)) {
 		rprintf(FINFO, "hash search s->blength=%ld len=%s count=%s\n",
 			(long)s->blength, big_num(len), big_num(s->count));
@@ -317,6 +337,8 @@ static void hash_search(int f,struct sum_struct *s,
 							backup = 0;
 						map = (schar *)map_ptr(buf, aligned_offset - backup, l + backup)
 						    + backup;
+						if (stream_early_EOF(buf, aligned_offset + l))
+							goto check_want_i;
 						sum = get_checksum1((char *)map, l);
 						if (sum != s->sums[i].sum1)
 							goto check_want_i;
@@ -352,6 +374,8 @@ static void hash_search(int f,struct sum_struct *s,
 			offset += s->sums[i].len - 1;
 			k = (int32)MIN((OFF_T)s->blength, len-offset);
 			map = (schar *)map_ptr(buf, offset, k);
+			if (stream_early_EOF(buf, len))
+				k = (int32)MIN((OFF_T)s->blength, buf->file_size-offset);
 			sum = get_checksum1((char *)map, k);
 			s1 = sum & 0xFFFF;
 			s2 = sum >> 16;
@@ -368,8 +392,16 @@ static void hash_search(int f,struct sum_struct *s,
 				OFF_T skip = MIN((OFF_T)(k - 1), end - 1 - offset);
 				if (skip > 0) {
 					offset += skip;
+					/* Anchor the map_ptr window at last_match (offset - backup) so
+					 * un-flushed literal bytes are not evicted from the sliding window,
+					 * which would force an illegal backward lseek() on a stream. */
+					backup = (int32)(offset - last_match);
+					if (backup < 0)
+						backup = 0;
 					k = (int32)MIN((OFF_T)s->blength, len-offset);
-					map = (schar *)map_ptr(buf, offset, k);
+					map = (schar *)map_ptr(buf, offset - backup, k + backup) + backup;
+					if (stream_early_EOF(buf, len))
+						k = (int32)MIN((OFF_T)s->blength, buf->file_size-offset);
 					sum = get_checksum1((char *)map, k);
 					s1 = sum & 0xFFFF;
 					s2 = sum >> 16;
@@ -389,6 +421,16 @@ static void hash_search(int f,struct sum_struct *s,
 		/* Trim off the first byte from the checksum */
 		more = offset + k < len;
 		map = (schar *)map_ptr(buf, offset - backup, k + more + backup) + backup;
+		if (stream_early_EOF(buf, len)) {
+			len = buf->file_size;
+			end = len + 1 - s->sums[s->count-1].len;
+			k = (int32)MIN((OFF_T)s->blength, len - offset);
+			more = offset + k < len;
+			if (DEBUG_GTE(DELTASUM, 2)) {
+				rprintf(FINFO, "stream early EOF at offset=%s len=%s end=%s k=%ld\n",
+					big_num(offset), big_num(len), big_num(end), (long)k);
+			}
+		}
 		s1 -= map[0] + CHAR_OFFSET;
 		s2 -= k * (map[0]+CHAR_OFFSET);
 
@@ -409,8 +451,13 @@ static void hash_search(int f,struct sum_struct *s,
 			matched(f, s, buf, offset - s->blength, -2);
 	} while (++offset < end);
 
+  finish_search:
 	matched(f, s, buf, len, -1);
-	map_ptr(buf, len-1, 1);
+	/* Ensure the map window has read through the final byte of the file.
+	 * Guard with len > 0 because an empty stream (0-byte early EOF) shrinks
+	 * len to 0, where len - 1 would pass a negative offset (-1) to map_ptr(). */
+	if (len > 0)
+		map_ptr(buf, len-1, 1);
 }
 
 
@@ -480,11 +527,42 @@ void match_sums(int f, struct sum_struct *s, struct map_struct *buf, OFF_T len)
 	} else {
 		OFF_T j;
 		/* by doing this in pieces we avoid too many seeks */
-		for (j = last_match + CHUNK_SIZE; j < len; j += CHUNK_SIZE)
+		for (j = last_match + CHUNK_SIZE; ; j += CHUNK_SIZE) {
+			/* Pre-read the next chunk on a stream before calling matched() so
+			 * that if the pipe hits early EOF mid-chunk, buf->file_size and len
+			 * shrink before matched() tries to flush up to 'j'. */
+			if (is_stream() && buf) {
+				map_ptr(buf, last_match, (int32)MIN((OFF_T)CHUNK_SIZE, len - last_match));
+				if (buf->file_size < len)
+					len = buf->file_size;
+			}
+			if (j >= len)
+				break;
 			matched(f, s, buf, j, -2);
+		}
 		matched(f, s, buf, len, -1);
 	}
 
+	if (is_stream() && buf && !eof_reached) {
+		char extra;
+		ssize_t n;
+		wait_stream_fd(buf->fd);
+		do {
+			n = read(buf->fd, &extra, 1);
+		} while (n < 0 && errno == EINTR);
+		if (n < 0) {
+			rsyserr(FERROR, errno, "read failed on stream");
+			exit_cleanup(RERR_FILEIO);
+		}
+		if (n > 0) {
+			rprintf(FERROR,
+				"stream source exceeded --stream-max-size=%s\n",
+				big_num(stream_max_size));
+			exit_cleanup(RERR_FILEIO);
+		}
+		eof_reached = 1;
+	}
+
 	sum_end(sender_file_sum);
 
 	/* If we had a read error, send a bad checksum.  We use all bits