X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/4e2a7e59e540a7a66da0c3e159f0464b05559f7f..e4c598c830234fe3def0bae6ba45ca54071a9ee5:/io.c

diff --git a/io.c b/io.c
index b717b431..4a6b95fb 100644
--- a/io.c
+++ b/io.c
@@ -136,6 +136,7 @@ enum festatus { FES_SUCCESS, FES_REDO, FES_NO_SEND };
 
 static flist_ndx_list redo_list, hlink_list;
 
+static void drain_multiplex_messages(void);
 static void sleep_for_bwlimit(int bytes_written);
 
 static void check_timeout(void)
@@ -173,9 +174,9 @@ static void check_timeout(void)
  * There is another case for older protocol versions (< 24) where the module
  * listing was not terminated, so we must ignore an EOF error in that case and
  * exit.  In this situation, kluge_around_eof will be > 0. */
-static NORETURN void whine_about_eof(int fd)
+static NORETURN void whine_about_eof(BOOL allow_kluge)
 {
-	if (kluge_around_eof && fd == sock_f_in) {
+	if (kluge_around_eof && allow_kluge) {
 		int i;
 		if (kluge_around_eof > 0)
 			exit_cleanup(0);
@@ -614,21 +615,20 @@ static char *perform_io(size_t needed, int flags)
 			if (iobuf.raw_flushing_ends_before
 			 || (!iobuf.msg.len && iobuf.out.len > iobuf.out_empty_len && !(flags & PIO_NEED_MSGROOM))) {
 				if (OUT_MULTIPLEXED && !iobuf.raw_flushing_ends_before) {
-					size_t val;
-
 					/* The iobuf.raw_flushing_ends_before value can point off the end
 					 * of the iobuf.out buffer for a while, for easier subtracting. */
 					iobuf.raw_flushing_ends_before = iobuf.out.pos + iobuf.out.len;
 
 					SIVAL(iobuf.out.buf + iobuf.raw_data_header_pos, 0,
 					      ((MPLEX_BASE + (int)MSG_DATA)<<24) + iobuf.out.len - 4);
-					if ((val = iobuf.out.size - iobuf.raw_data_header_pos) < 4) {
+					if (iobuf.raw_data_header_pos + 4 > iobuf.out.size) {
+						int siz = (int)(iobuf.raw_data_header_pos + 4 - iobuf.out.size);
 						/* We used some of the overflow bytes, so move them. */
 						if (DEBUG_GTE(IO, 4)) {
 							rprintf(FINFO, "[%s] wrap-bytes moved: %d (perform_io)\n",
-								who_am_i(), (int)val);
+								who_am_i(), siz);
 						}
-						memcpy(iobuf.out.buf, iobuf.out.buf + iobuf.out.size, 4 - val);
+						memcpy(iobuf.out.buf, iobuf.out.buf + iobuf.out.size, siz);
 					}
 
 					if (DEBUG_GTE(IO, 1)) {
@@ -662,11 +662,19 @@ static char *perform_io(size_t needed, int flags)
 		if (max_fd < 0) {
 			switch (flags & PIO_NEED_FLAGS) {
 			case PIO_NEED_INPUT:
+				iobuf.in.len = 0;
+				if (kluge_around_eof == 2)
+					exit_cleanup(0);
+				if (iobuf.in_fd == -2)
+					whine_about_eof(True);
 				rprintf(FERROR, "error in perform_io: no fd for input.\n");
 				exit_cleanup(RERR_PROTOCOL);
 			case PIO_NEED_OUTROOM:
 			case PIO_NEED_MSGROOM:
 				msgs2stderr = 1;
+				drain_multiplex_messages();
+				if (iobuf.out_fd == -2)
+					whine_about_eof(True);
 				rprintf(FERROR, "error in perform_io: no fd for output.\n");
 				exit_cleanup(RERR_PROTOCOL);
 			default:
@@ -677,8 +685,7 @@ static char *perform_io(size_t needed, int flags)
 		}
 
 		if (extra_flist_sending_enabled) {
-			if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD
-			 && file_total - file_old_total >= MIN_FILECNT_LOOKAHEAD)
+			if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD)
 				tv.tv_sec = 0;
 			else {
 				extra_flist_sending_enabled = False;
@@ -711,8 +718,9 @@ static char *perform_io(size_t needed, int flags)
 			int n;
 			if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
 				if (n == 0) {
+					/* Signal that input has become invalid. */
 					if (!read_batch || batch_fd < 0 || am_generator)
-						whine_about_eof(iobuf.in_fd); /* Doesn't return. */
+						iobuf.in_fd = -2;
 					batch_fd = -1;
 					continue;
 				}
@@ -726,7 +734,7 @@ static char *perform_io(size_t needed, int flags)
 						rsyserr(FERROR_SOCKET, errno, "read error");
 					} else
 						rsyserr(FERROR, errno, "read error");
-					exit_cleanup(RERR_STREAMIO);
+					exit_cleanup(RERR_SOCKETIO);
 				}
 			}
 			if (msgs2stderr && DEBUG_GTE(IO, 2))
@@ -754,9 +762,11 @@ static char *perform_io(size_t needed, int flags)
 				else {
 					/* Don't write errors on a dead socket. */
 					msgs2stderr = 1;
-					out->len = iobuf.raw_flushing_ends_before = out->pos = 0;
+					iobuf.out_fd = -2;
+					iobuf.out.len = iobuf.msg.len = iobuf.raw_flushing_ends_before = 0;
 					rsyserr(FERROR_SOCKET, errno, "[%s] write error", who_am_i());
-					exit_cleanup(RERR_STREAMIO);
+					drain_multiplex_messages();
+					exit_cleanup(RERR_SOCKETIO);
 				}
 			}
 			if (msgs2stderr && DEBUG_GTE(IO, 2)) {
@@ -775,8 +785,7 @@ static char *perform_io(size_t needed, int flags)
 				if (iobuf.raw_flushing_ends_before)
 					iobuf.raw_flushing_ends_before -= out->size;
 				out->pos = 0;
-			}
-			if (out->pos == iobuf.raw_flushing_ends_before)
+			} else if (out->pos == iobuf.raw_flushing_ends_before)
 				iobuf.raw_flushing_ends_before = 0;
 			if ((out->len -= n) == empty_buf_len) {
 				out->pos = 0;
@@ -809,8 +818,11 @@ void noop_io_until_death(void)
 {
 	char buf[1024];
 
-	kluge_around_eof = 1;
-	set_io_timeout(protocol_version >= 31 ? 10 : 1);
+	kluge_around_eof = 2;
+	/* Setting an I/O timeout ensures that if something inexplicably weird
+	 * happens, we won't hang around forever. */
+	if (!io_timeout)
+		set_io_timeout(60);
 
 	while (1)
 		read_buf(iobuf.in_fd, buf, sizeof buf);
@@ -833,7 +845,7 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
 	if (convert > 0 && ic_send == (iconv_t)-1)
 		convert = 0;
 	if (convert > 0) {
-		/* Ensuring double-size room leaves space for a potential conversion. */
+		/* Ensuring double-size room leaves space for maximal conversion expansion. */
 		if (iobuf.msg.len + len*2 + 4 > iobuf.msg.size)
 			perform_io(len*2 + 4, PIO_NEED_MSGROOM);
 	} else
@@ -846,7 +858,7 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
 		pos -= iobuf.msg.size;
 	hdr = iobuf.msg.buf + pos;
 
-	iobuf.msg.len += 4; /* Leave room for the coming header bytes. */
+	iobuf.msg.len += 4; /* Allocate room for the coming header bytes. */
 
 #ifdef ICONV_OPTION
 	if (convert > 0) {
@@ -883,10 +895,10 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
 	SIVAL(hdr, 0, ((MPLEX_BASE + (int)code)<<24) + len);
 	/* If the header used any overflow bytes, move them to the start. */
 	if ((pos = hdr+4 - iobuf.msg.buf) > iobuf.msg.size) {
-		size_t siz = pos - iobuf.msg.size;
+		int siz = (int)(pos - iobuf.msg.size);
 		if (DEBUG_GTE(IO, 4))
-			rprintf(FINFO, "[%s] wrap-bytes moved: %d (send_msg)\n", who_am_i(), (int)siz);
-		memcpy(iobuf.msg.buf, hdr+4 - siz, siz);
+			rprintf(FINFO, "[%s] wrap-bytes moved: %d (send_msg)\n", who_am_i(), siz);
+		memcpy(iobuf.msg.buf, iobuf.msg.buf + iobuf.msg.size, siz);
 	}
 
 	if (want_debug && convert > 0)
@@ -1167,7 +1179,8 @@ BOOL io_start_buffering_out(int f_out)
 	if (iobuf.out.buf) {
 		if (iobuf.out_fd == -1)
 			iobuf.out_fd = f_out;
-		assert(f_out == iobuf.out_fd);
+		else
+			assert(f_out == iobuf.out_fd);
 		return False;
 	}
 
@@ -1189,7 +1202,8 @@ BOOL io_start_buffering_in(int f_in)
 	if (iobuf.in.buf) {
 		if (iobuf.in_fd == -1)
 			iobuf.in_fd = f_in;
-		assert(f_in == iobuf.in_fd);
+		else
+			assert(f_in == iobuf.in_fd);
 		return False;
 	}
 
@@ -1471,6 +1485,23 @@ static void read_a_msg(void)
 	}
 }
 
+static void drain_multiplex_messages(void)
+{
+	while (IN_MULTIPLEXED && iobuf.in.len) {
+		if (iobuf.raw_input_ends_before) {
+			size_t raw_len = iobuf.raw_input_ends_before - iobuf.in.pos;
+			iobuf.raw_input_ends_before = 0;
+			if (raw_len >= iobuf.in.len) {
+				iobuf.in.len = 0;
+				break;
+			}
+			iobuf.in.pos += raw_len;
+			iobuf.in.len -= raw_len;
+		}
+		read_a_msg();
+	}
+}
+
 void wait_for_receiver(void)
 {
 	if (!iobuf.raw_input_ends_before)
@@ -1624,7 +1655,7 @@ void read_buf(int f, char *buf, size_t len)
 {
 	if (f != iobuf.in_fd) {
 		if (safe_read(f, buf, len) != len)
-			whine_about_eof(f); /* Doesn't return. */
+			whine_about_eof(False); /* Doesn't return. */
 		goto batch_copy;
 	}
 
@@ -2081,11 +2112,11 @@ void io_printf(int fd, const char *format, ...)
 	va_end(ap);
 
 	if (len < 0)
-		exit_cleanup(RERR_STREAMIO);
+		exit_cleanup(RERR_PROTOCOL);
 
 	if (len > (int)sizeof buf) {
 		rprintf(FERROR, "io_printf() was too long for the buffer.\n");
-		exit_cleanup(RERR_STREAMIO);
+		exit_cleanup(RERR_PROTOCOL);
 	}
 
 	write_sbuf(fd, buf);