X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/cbc63a09e8ea792a9eb39fc2f94536c7a01e525d..9270e88d76d3ea9e94d48977fbea57b997545577:/io.c

diff --git a/io.c b/io.c
index 0707ca21..b5fd3776 100644
--- a/io.c
+++ b/io.c
@@ -76,6 +76,9 @@ int kluge_around_eof = 0;
 int sock_f_in = -1;
 int sock_f_out = -1;
 
+int64 total_data_read = 0;
+int64 total_data_written = 0;
+
 static struct {
 	xbuf in, out, msg;
 	int in_fd;
@@ -94,7 +97,8 @@ static int write_batch_monitor_in = -1;
 static int write_batch_monitor_out = -1;
 
 static int ff_forward_fd = -1;
-static char ff_lastchar;
+static int ff_reenable_multiplex = -1;
+static char ff_lastchar = '\0';
 static xbuf ff_xb = EMPTY_XBUF;
 #ifdef ICONV_OPTION
 static xbuf iconv_buf = EMPTY_XBUF;
@@ -348,8 +352,8 @@ static void forward_filesfrom_data(void)
 			ff_forward_fd = -1;
 			write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
 			free_xbuf(&ff_xb);
-			if (protocol_version < 31)
-				io_start_multiplex_out(iobuf.out_fd);
+			if (ff_reenable_multiplex >= 0)
+				io_start_multiplex_out(ff_reenable_multiplex);
 		}
 		return;
 	}
@@ -748,7 +752,7 @@ static char *perform_io(size_t needed, int flags)
 					/* Don't write errors on a dead socket. */
 					msgs2stderr = 1;
 					out->len = iobuf.raw_flushing_ends_before = out->pos = 0;
-					rsyserr(FERROR_SOCKET, errno, "write error");
+					rsyserr(FERROR_SOCKET, errno, "[%s] write error", who_am_i());
 					exit_cleanup(RERR_STREAMIO);
 				}
 			}
@@ -798,6 +802,17 @@ static char *perform_io(size_t needed, int flags)
 	return data;
 }
 
+void noop_io_until_death(void)
+{
+	char buf[1024];
+
+	kluge_around_eof = 1;
+	set_io_timeout(10);
+
+	while (1)
+		read_buf(iobuf.in_fd, buf, sizeof buf);
+}
+
 /* Buffer a message for the multiplexed output stream.  Is never used for MSG_DATA. */
 int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
 {
@@ -1016,18 +1031,14 @@ int get_hlink_num(void)
  * for recv_file_list() to use. */
 void start_filesfrom_forwarding(int fd)
 {
-	ff_forward_fd = fd;
-	if (protocol_version < 31) {
-		int save_fd = iobuf.out_fd;
-		/* Older protocols send the files-from data w/o packaging it in
-		 * multiplexed I/O packets.  To match this, we temporarily turn
-		 * off the multiplexing of our output w/o disabling buffering. */
-		assert(OUT_MULTIPLEXED);
-		/* Be extra, extra sure no messages go out before files-from data. */
-		iobuf.msg.pos = iobuf.msg.len = 0;
-		io_end_multiplex_out(False);
-		iobuf.out_fd = save_fd;
+	if (protocol_version < 31 && OUT_MULTIPLEXED) {
+		/* Older protocols send the files-from data w/o packaging
+		 * it in multiplexed I/O packets, so temporarily switch
+		 * to buffered I/O to match this behavior. */
+		iobuf.msg.pos = iobuf.msg.len = 0; /* Be extra sure no messages go out. */
+		ff_reenable_multiplex = io_end_multiplex_out(MPLX_TO_BUFFERED);
 	}
+	ff_forward_fd = fd;
 
 	alloc_xbuf(&ff_xb, FILESFROM_BUFLEN);
 }
@@ -1138,7 +1149,7 @@ void read_args(int f_in, char *mod_name, char *buf, size_t bufsiz, int rl_nulls,
 	*argv_p = argv;
 }
 
-int io_start_buffering_out(int f_out)
+BOOL io_start_buffering_out(int f_out)
 {
 	if (msgs2stderr && DEBUG_GTE(IO, 2))
 		rprintf(FINFO, "[%s] io_start_buffering_out(%d)\n", who_am_i(), f_out);
@@ -1154,7 +1165,7 @@ int io_start_buffering_out(int f_out)
 		if (iobuf.out_fd == -1)
 			iobuf.out_fd = f_out;
 		assert(f_out == iobuf.out_fd);
-		return 0;
+		return False;
 	}
 
 	iobuf.out.size = IO_BUFFER_SIZE * 2 - 4;
@@ -1164,10 +1175,10 @@ int io_start_buffering_out(int f_out)
 	iobuf.out.pos = iobuf.out.len = 0;
 	iobuf.out_fd = f_out;
 
-	return 1;
+	return True;
 }
 
-int io_start_buffering_in(int f_in)
+BOOL io_start_buffering_in(int f_in)
 {
 	if (msgs2stderr && DEBUG_GTE(IO, 2))
 		rprintf(FINFO, "[%s] io_start_buffering_in(%d)\n", who_am_i(), f_in);
@@ -1176,7 +1187,7 @@ int io_start_buffering_in(int f_in)
 		if (iobuf.in_fd == -1)
 			iobuf.in_fd = f_in;
 		assert(f_in == iobuf.in_fd);
-		return 0;
+		return False;
 	}
 
 	iobuf.in.size = IO_BUFFER_SIZE;
@@ -1187,14 +1198,14 @@ int io_start_buffering_in(int f_in)
 
 	iobuf.in_fd = f_in;
 
-	return 1;
+	return True;
 }
 
 void io_end_buffering_in(BOOL free_buffers)
 {
-	if (DEBUG_GTE(IO, 2)) {
-		rprintf(FINFO, "[%s] io_end_buffering_in(%s)\n",
-			who_am_i(), free_buffers ? "True" : "False");
+	if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+		rprintf(FINFO, "[%s] io_end_buffering_in(IOBUF_%s_BUFS)\n",
+			who_am_i(), free_buffers ? "FREE" : "KEEP");
 	}
 
 	if (free_buffers)
@@ -1207,9 +1218,9 @@ void io_end_buffering_in(BOOL free_buffers)
 
 void io_end_buffering_out(BOOL free_buffers)
 {
-	if (DEBUG_GTE(IO, 2)) {
-		rprintf(FINFO, "[%s] io_end_buffering_out(%s)\n",
-			who_am_i(), free_buffers ? "True" : "False");
+	if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+		rprintf(FINFO, "[%s] io_end_buffering_out(IOBUF_%s_BUFS)\n",
+			who_am_i(), free_buffers ? "FREE" : "KEEP");
 	}
 
 	io_flush(FULL_FLUSH);
@@ -1217,9 +1228,6 @@ void io_end_buffering_out(BOOL free_buffers)
 	if (free_buffers) {
 		free_xbuf(&iobuf.out);
 		free_xbuf(&iobuf.msg);
-	} else {
-		iobuf.out.pos = iobuf.out.len = 0;
-		iobuf.msg.pos = iobuf.msg.len = 0;
 	}
 
 	iobuf.out_fd = -1;
@@ -1267,7 +1275,7 @@ void stop_flist_forward(void)
 static void read_a_msg(void)
 {
 	char *data, line[BIGPATHBUFLEN];
-	int tag;
+	int tag, val;
 	size_t msg_bytes;
 
 	data = perform_io(4, PIO_INPUT_AND_CONSUME);
@@ -1309,6 +1317,17 @@ static void read_a_msg(void)
 		if (!am_generator)
 			send_msg(MSG_IO_ERROR, data, 4, 0);
 		break;
+	case MSG_IO_TIMEOUT:
+		if (msg_bytes != 4 || am_server || am_generator)
+			goto invalid_msg;
+		data = perform_io(4, PIO_INPUT_AND_CONSUME);
+		val = IVAL(data, 0);
+		if (!io_timeout || io_timeout > val) {
+			if (INFO_GTE(MISC, 2))
+				rprintf(FINFO, "Setting --timeout=%d to match server\n", val);
+			set_io_timeout(val);
+		}
+		break;
 	case MSG_NOOP:
 		if (am_sender)
 			maybe_send_keepalive();
@@ -1418,6 +1437,16 @@ static void read_a_msg(void)
 			first_message = 0;
 		}
 		break;
+	case MSG_ERROR_EXIT:
+		if (msg_bytes != 4)
+			goto invalid_msg;
+		data = perform_io(4, PIO_INPUT_AND_CONSUME);
+		val = IVAL(data, 0);
+		if (am_generator && protocol_version >= 31)
+			send_msg_int(MSG_ERROR_EXIT, val);
+		if (am_generator)
+			val = RERR_RCVR_ERROR; /* avoids duplicate errors */
+		exit_cleanup(val);
 	default:
 		rprintf(FERROR, "unexpected tag %d [%s%s]\n",
 			tag, who_am_i(), inc_recurse ? "/inc" : "");
@@ -1584,6 +1613,7 @@ void read_buf(int f, char *buf, size_t len)
 
 	if (!IN_MULTIPLEXED) {
 		memcpy(buf, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+		total_data_read += len;
 		if (forward_flist_data)
 			write_buf(iobuf.out_fd, buf, len);
 	  batch_copy:
@@ -1607,6 +1637,7 @@ void read_buf(int f, char *buf, size_t len)
 		/* The bytes at the "data" pointer will survive long
 		 * enough to make a copy, but not past future I/O. */
 		memcpy(buf, data, siz);
+		total_data_read += siz;
 
 		if (forward_flist_data)
 			write_buf(iobuf.out_fd, buf, siz);
@@ -1879,6 +1910,7 @@ void write_buf(int f, const char *buf, size_t len)
 		memcpy(iobuf.out.buf + pos, buf, len);
 
 	iobuf.out.len += len;
+	total_data_written += len;
 
   batch_copy:
 	if (f == write_batch_monitor_out)
@@ -2045,11 +2077,11 @@ void io_printf(int fd, const char *format, ...)
 /* Setup for multiplexing a MSG_* stream with the data stream. */
 void io_start_multiplex_out(int fd)
 {
+	io_flush(FULL_FLUSH);
+
 	if (msgs2stderr && DEBUG_GTE(IO, 2))
 		rprintf(FINFO, "[%s] io_start_multiplex_out(%d)\n", who_am_i(), fd);
 
-	io_flush(FULL_FLUSH);
-
 	iobuf.out_empty_len = 4; /* See also OUT_MULTIPLEXED */
 	io_start_buffering_out(fd);
 
@@ -2067,28 +2099,40 @@ void io_start_multiplex_in(int fd)
 	io_start_buffering_in(fd);
 }
 
-void io_end_multiplex_in(BOOL free_buffers)
+int io_end_multiplex_in(int mode)
 {
-	if (DEBUG_GTE(IO, 2)) {
-		rprintf(FINFO, "[%s] io_end_multiplex_in(%s)\n",
-			who_am_i(), free_buffers ? "True" : "False");
-	}
+	int ret = iobuf.in_multiplexed ? iobuf.in_fd : -1;
+
+	if (msgs2stderr && DEBUG_GTE(IO, 2))
+		rprintf(FINFO, "[%s] io_end_multiplex_in(mode=%d)\n", who_am_i(), mode);
 
 	iobuf.in_multiplexed = False;
-	iobuf.raw_input_ends_before = 0;
-	io_end_buffering_in(free_buffers);
+	if (mode == MPLX_SWITCHING)
+		iobuf.raw_input_ends_before = 0;
+	else
+		assert(iobuf.raw_input_ends_before == 0);
+	if (mode != MPLX_TO_BUFFERED)
+		io_end_buffering_in(mode);
+
+	return ret;
 }
 
-/* Stop output multiplexing. */
-void io_end_multiplex_out(BOOL free_buffers)
+int io_end_multiplex_out(int mode)
 {
-	if (DEBUG_GTE(IO, 2)) {
-		rprintf(FINFO, "[%s] io_end_multiplex_out(%s)\n",
-			who_am_i(), free_buffers ? "True" : "False");
-	}
+	int ret = iobuf.out_empty_len ? iobuf.out_fd : -1;
 
-	io_end_buffering_out(free_buffers);
+	if (msgs2stderr && DEBUG_GTE(IO, 2))
+		rprintf(FINFO, "[%s] io_end_multiplex_out(mode=%d)\n", who_am_i(), mode);
+
+	if (mode != MPLX_TO_BUFFERED)
+		io_end_buffering_out(mode);
+	else
+		io_flush(FULL_FLUSH);
+
+	iobuf.out.len = 0;
 	iobuf.out_empty_len = 0;
+
+	return ret;
 }
 
 void start_write_batch(int fd)