X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/ac32cdd7d4b718859a5a70b62c029ea286ba361b..ef7441669b194ba6804c2eda07beed90a29ad2b3:/io.c

diff --git a/io.c b/io.c
index 2c162a6e..bf39ff91 100644
--- a/io.c
+++ b/io.c
@@ -38,8 +38,8 @@ extern int bwlimit;
 extern size_t bwlimit_writemax;
 extern int io_timeout;
 extern int am_server;
-extern int am_daemon;
 extern int am_sender;
+extern int am_receiver;
 extern int am_generator;
 extern int msgs2stderr;
 extern int inc_recurse;
@@ -50,6 +50,7 @@ extern int file_total;
 extern int file_old_total;
 extern int list_only;
 extern int read_batch;
+extern int compat_flags;
 extern int protect_args;
 extern int checksum_seed;
 extern int protocol_version;
@@ -65,7 +66,6 @@ extern iconv_t ic_send, ic_recv;
 
 int csum_length = SHORT_SUM_LENGTH; /* initial value */
 int allowed_lull = 0;
-int ignore_timeout = 0;
 int batch_fd = -1;
 int msgdone_cnt = 0;
 int forward_flist_data = 0;
@@ -153,25 +153,43 @@ static void read_a_msg(void);
 static void drain_multiplex_messages(void);
 static void sleep_for_bwlimit(int bytes_written);
 
-static void check_timeout(void)
-{
-	time_t t;
-
-	if (!io_timeout || ignore_timeout)
+static void check_timeout(BOOL allow_keepalive)
+{
+	time_t t, chk;
+
+	/* On the receiving side, the generator is now the one that decides
+	 * when a timeout has occurred.  When it is sifting through a lot of
+	 * files looking for work, it will be sending keep-alive messages to
+	 * the sender, and even though the receiver won't be sending/receiving
+	 * anything (not even keep-alive messages), the successful writes to
+	 * the sender will keep things going.  If the receiver is actively
+	 * receiving data, it will ensure that the generator knows that it is
+	 * not idle by sending the generator keep-alive messages (since the
+	 * generator might be blocked trying to send checksums, it needs to
+	 * know that the receiver is active).  Thus, as long as one or the
+	 * other is successfully doing work, the generator will not timeout. */
+	if (!io_timeout)
 		return;
 
-	if (!last_io_in) {
-		last_io_in = time(NULL);
-		return;
+	t = time(NULL);
+
+	if (allow_keepalive) {
+		/* This may put data into iobuf.msg w/o flushing. */
+		maybe_send_keepalive(t, 0);
 	}
 
-	t = time(NULL);
+	if (!last_io_in)
+		last_io_in = t;
 
-	if (t - last_io_in >= io_timeout) {
-		if (!am_server && !am_daemon) {
-			rprintf(FERROR, "io timeout after %d seconds -- exiting\n",
-				(int)(t-last_io_in));
-		}
+	if (am_receiver)
+		return;
+
+	chk = MAX(last_io_out, last_io_in);
+	if (t - chk >= io_timeout) {
+		if (am_server)
+			msgs2stderr = 1;
+		rprintf(FERROR, "[%s] io timeout after %d seconds -- exiting\n",
+			who_am_i(), (int)(t-chk));
 		exit_cleanup(RERR_TIMEOUT);
 	}
 }
@@ -253,7 +271,8 @@ static size_t safe_read(int fd, char *buf, size_t len)
 					who_am_i());
 				exit_cleanup(RERR_FILEIO);
 			}
-			check_timeout();
+			if (io_timeout)
+				maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
 			continue;
 		}
 
@@ -337,7 +356,8 @@ static void safe_write(int fd, const char *buf, size_t len)
 					what_fd_is(fd), who_am_i());
 				exit_cleanup(RERR_FILEIO);
 			}
-			check_timeout();
+			if (io_timeout)
+				maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
 			continue;
 		}
 
@@ -490,23 +510,25 @@ void restore_iobuf_size(xbuf *out)
 	}
 }
 
-/* Perform buffered input and output until specified conditions are met.  When
- * given a "needed" read requirement, we'll return without doing any I/O if the
- * iobuf.in bytes are already available.  When reading, we'll read as many
- * bytes as we can into the buffer, and return as soon as we meet the minimum
- * read requirement.  When given a "needed" write requirement, we'll return
- * without doing any I/O if that many bytes will fit in the output buffer (we
- * check either iobuf.out or iobuf.msg, depending on the flags).  When writing,
- * we write out as much as we can, and return as soon as the given free-space
- * requirement is available.
+/* Perform buffered input and/or output until specified conditions are met.
+ * When given a "needed" read or write request, this returns without doing any
+ * I/O if the needed input bytes or write space is already available.  Once I/O
+ * is needed, this will try to do whatever reading and/or writing is currently
+ * possible, up to the maximum buffer allowances, no matter if this is a read
+ * or write request.  However, the I/O stops as soon as the required input
+ * bytes or output space is available.  If this is not a read request, the
+ * routine may also do some advantageous reading of messages from a multiplexed
+ * input source (which ensures that we don't jam up with everyone in their
+ * "need to write" code and nobody reading the accumulated data that would make
+ * writing possible).
  *
- * The iobuf.out and iobuf.msg buffers are circular, so some writes into them
- * will need to be split when the data needs to wrap around to the start.  In
- * order to help make this easier for some operations (such as the use of
- * SIVAL() into the buffer) a buffer may be temporarily shortened, but the
- * original size will be automatically restored.  The iobuf.in buffer is also
- * circular, so callers may need to split their reading of the data if it spans
- * the end.  See also the 3 raw_* iobuf vars that are used in the handling of
+ * The iobuf.in, .out and .msg buffers are all circular.  Callers need to be
+ * aware that some data copies will need to be split when the bytes wrap around
+ * from the end to the start.  In order to help make writing into the output
+ * buffers easier for some operations (such as the use of SIVAL() into the
+ * buffer) a buffer may be temporarily shortened by a small amount, but the
+ * original size will be automatically restored when the .pos wraps to the
+ * start.  See also the 3 raw_* iobuf vars that are used in the handling of
  * MSG_DATA bytes as they are read-from/written-into the buffers.
  *
  * When writing, we flush data in the following priority order:
@@ -732,7 +754,7 @@ static char *perform_io(size_t needed, int flags)
 				send_extra_file_list(sock_f_out, -1);
 				extra_flist_sending_enabled = !flist_eof;
 			} else
-				check_timeout();
+				check_timeout((flags & PIO_NEED_INPUT) != 0);
 			FD_ZERO(&r_fds); /* Just in case... */
 			FD_ZERO(&w_fds);
 		}
@@ -769,14 +791,17 @@ static char *perform_io(size_t needed, int flags)
 			if (msgs2stderr && DEBUG_GTE(IO, 2))
 				rprintf(FINFO, "[%s] recv=%ld\n", who_am_i(), (long)n);
 
-			if (io_timeout)
+			if (io_timeout) {
 				last_io_in = time(NULL);
+				if (flags & PIO_NEED_INPUT)
+					maybe_send_keepalive(last_io_in, 0);
+			}
 			stats.total_read += n;
 
 			iobuf.in.len += n;
 		}
 
-		if (iobuf.out_fd >= 0 && FD_ISSET(iobuf.out_fd, &w_fds)) {
+		if (out && FD_ISSET(iobuf.out_fd, &w_fds)) {
 			size_t len = iobuf.raw_flushing_ends_before ? iobuf.raw_flushing_ends_before - out->pos : out->len;
 			int n;
 
@@ -897,7 +922,7 @@ void noop_io_until_death(void)
 		read_buf(iobuf.in_fd, buf, sizeof buf);
 }
 
-/* Buffer a message for the multiplexed output stream.  Is never used for MSG_DATA. */
+/* Buffer a message for the multiplexed output stream.  Is not used for (normal) MSG_DATA. */
 int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
 {
 	char *hdr;
@@ -959,8 +984,8 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
 	{
 		size_t siz;
 
-		if ((pos += 4) >= iobuf.msg.size)
-			pos -= iobuf.msg.size;
+		if ((pos += 4) == iobuf.msg.size)
+			pos = 0;
 
 		/* Handle a split copy if we wrap around the end of the circular buffer. */
 		if (pos >= iobuf.msg.pos && (siz = iobuf.msg.size - pos) < len) {
@@ -1041,13 +1066,15 @@ void io_set_sock_fds(int f_in, int f_out)
 void set_io_timeout(int secs)
 {
 	io_timeout = secs;
+	allowed_lull = (io_timeout + 1) / 2;
 
-	if (!io_timeout || io_timeout > SELECT_TIMEOUT)
+	if (!io_timeout || allowed_lull > SELECT_TIMEOUT)
 		select_timeout = SELECT_TIMEOUT;
 	else
-		select_timeout = io_timeout;
+		select_timeout = allowed_lull;
 
-	allowed_lull = read_batch ? 0 : (io_timeout + 1) / 2;
+	if (read_batch)
+		allowed_lull = 0;
 }
 
 static void check_for_d_option_error(const char *msg)
@@ -1314,20 +1341,29 @@ void maybe_flush_socket(int important)
 		io_flush(NORMAL_FLUSH);
 }
 
-void maybe_send_keepalive(void)
+/* Older rsync versions used to send either a MSG_NOOP (protocol 30) or a
+ * raw-data-based keep-alive (protocol 29), both of which implied forwarding of
+ * the message through the sender.  Since the new timeout method does not need
+ * any forwarding, we just send an empty MSG_DATA message, which works with all
+ * rsync versions.  This avoids any message forwarding, and leaves the raw-data
+ * stream alone (since we can never be quite sure if that stream is in the
+ * right state for a keep-alive message). */
+void maybe_send_keepalive(time_t now, int flags)
 {
-	if (time(NULL) - last_io_out >= allowed_lull) {
-		if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len) {
-			if (protocol_version < 29)
-				return; /* there's nothing we can do */
-			if (protocol_version >= 30)
-				send_msg(MSG_NOOP, "", 0, 0);
-			else {
-				write_int(iobuf.out_fd, cur_flist->used);
-				write_shortint(iobuf.out_fd, ITEM_IS_NEW);
-			}
-		}
-		if (iobuf.msg.len)
+	if (flags & MSK_ACTIVE_RECEIVER)
+		last_io_in = now; /* Fudge things when we're working hard on the files. */
+
+	if (now - last_io_out >= allowed_lull) {
+		/* The receiver is special:  it only sends keep-alive messages if it is
+		 * actively receiving data.  Otherwise, it lets the generator timeout. */
+		if (am_receiver && now - last_io_in >= io_timeout)
+			return;
+
+		if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len)
+			send_msg(MSG_DATA, "", 0, 0);
+		if (!(flags & MSK_ALLOW_FLUSH)) {
+			/* Let the caller worry about writing out the data. */
+		} else if (iobuf.msg.len)
 			perform_io(iobuf.msg.size - iobuf.msg.len + 1, PIO_NEED_MSGROOM);
 		else if (iobuf.out.len > iobuf.out_empty_len)
 			io_flush(NORMAL_FLUSH);
@@ -1373,8 +1409,9 @@ static void read_a_msg(void)
 		 * the buffer the msg data will end once it is read.  It is
 		 * possible that this points off the end of the buffer, in
 		 * which case the gradual reading of the input stream will
-		 * cause this value to decrease and eventually become real. */
-		iobuf.raw_input_ends_before = iobuf.in.pos + msg_bytes;
+		 * cause this value to wrap around and eventually become real. */
+		if (msg_bytes)
+			iobuf.raw_input_ends_before = iobuf.in.pos + msg_bytes;
 		iobuf.in_multiplexed = 1;
 		break;
 	case MSG_STATS:
@@ -1391,12 +1428,12 @@ static void read_a_msg(void)
 		got_flist_entry_status(FES_REDO, val);
 		break;
 	case MSG_IO_ERROR:
-		if (msg_bytes != 4 || am_sender)
+		if (msg_bytes != 4)
 			goto invalid_msg;
 		val = raw_read_int();
 		iobuf.in_multiplexed = 1;
 		io_error |= val;
-		if (!am_generator)
+		if (am_receiver)
 			send_msg_int(MSG_IO_ERROR, val);
 		break;
 	case MSG_IO_TIMEOUT:
@@ -1411,9 +1448,12 @@ static void read_a_msg(void)
 		}
 		break;
 	case MSG_NOOP:
-		if (am_sender)
-			maybe_send_keepalive();
+		/* Support protocol-30 keep-alive method. */
+		if (msg_bytes != 0)
+			goto invalid_msg;
 		iobuf.in_multiplexed = 1;
+		if (am_sender)
+			maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
 		break;
 	case MSG_DELETED:
 		if (msg_bytes >= sizeof data)
@@ -1534,7 +1574,7 @@ static void read_a_msg(void)
 			goto invalid_msg;
 		iobuf.in_multiplexed = 1;
 		if (DEBUG_GTE(EXIT, 3))
-			rprintf(FINFO, "[%s] got MSG_ERROR_EXIT with %d bytes\n", who_am_i(), msg_bytes);
+			rprintf(FINFO, "[%s] got MSG_ERROR_EXIT with %ld bytes\n", who_am_i(), (long)msg_bytes);
 		if (msg_bytes == 0) {
 			if (!am_sender && !am_generator) {
 				if (DEBUG_GTE(EXIT, 3)) {
@@ -1544,23 +1584,19 @@ static void read_a_msg(void)
 				send_msg(MSG_ERROR_EXIT, "", 0, 0);
 				io_flush(FULL_FLUSH);
 			}
-			val = 0;
-		} else {
-			val = raw_read_int();
-			if (protocol_version >= 31) {
-				if (am_generator) {
-					if (DEBUG_GTE(EXIT, 3)) {
-						rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT with exit_code %d\n",
-							who_am_i(), val);
-					}
-					send_msg_int(MSG_ERROR_EXIT, val);
-				} else {
-					if (DEBUG_GTE(EXIT, 3)) {
-						rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT (len 0)\n",
-							who_am_i());
-					}
-					send_msg(MSG_ERROR_EXIT, "", 0, 0);
+		} else if (protocol_version >= 31) {
+			if (am_generator) {
+				if (DEBUG_GTE(EXIT, 3)) {
+					rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT with exit_code %d\n",
+						who_am_i(), val);
 				}
+				send_msg_int(MSG_ERROR_EXIT, val);
+			} else {
+				if (DEBUG_GTE(EXIT, 3)) {
+					rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT (len 0)\n",
+						who_am_i());
+				}
+				send_msg(MSG_ERROR_EXIT, "", 0, 0);
 			}
 		}
 		/* Send a negative linenum so that we don't end up
@@ -2283,7 +2319,7 @@ void start_write_batch(int fd)
 	 * is involved. */
 	write_int(batch_fd, protocol_version);
 	if (protocol_version >= 30)
-		write_byte(batch_fd, inc_recurse);
+		write_byte(batch_fd, compat_flags);
 	write_int(batch_fd, checksum_seed);
 
 	if (am_sender)