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

diff --git a/io.c b/io.c
index 3ece9c3b..4a6b95fb 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,9 @@ 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;
 #endif
@@ -131,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)
@@ -168,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);
@@ -338,20 +344,17 @@ static void safe_write(int fd, const char *buf, size_t len)
  * a chunk of data and put it into the output buffer. */
 static void forward_filesfrom_data(void)
 {
-	char buf[FILESFROM_BUFLEN];
 	int len;
-	xbuf x;
-
-	INIT_CONST_XBUF(x, buf);
 
-	len = read(ff_forward_fd, x.buf, x.size);
+	len = read(ff_forward_fd, ff_xb.buf + ff_xb.len, ff_xb.size - ff_xb.len);
 	if (len <= 0) {
 		if (len == 0 || errno != EINTR) {
 			/* Send end-of-file marker */
-			write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
 			ff_forward_fd = -1;
-			if (protocol_version < 31)
-				io_start_multiplex_out(iobuf.out_fd);
+			write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
+			free_xbuf(&ff_xb);
+			if (ff_reenable_multiplex >= 0)
+				io_start_multiplex_out(ff_reenable_multiplex);
 		}
 		return;
 	}
@@ -359,47 +362,84 @@ static void forward_filesfrom_data(void)
 	if (DEBUG_GTE(IO, 2))
 		rprintf(FINFO, "[%s] files-from read=%ld\n", who_am_i(), (long)len);
 
+#ifdef ICONV_OPTION
+	len += ff_xb.len;
+#endif
+
 	if (!eol_nulls) {
-		char *s = x.buf + len;
+		char *s = ff_xb.buf + len;
 		/* Transform CR and/or LF into '\0' */
-		while (s-- > x.buf) {
+		while (s-- > ff_xb.buf) {
 			if (*s == '\n' || *s == '\r')
 				*s = '\0';
 		}
 	}
+
 	if (ff_lastchar)
-		x.pos = 0;
+		ff_xb.pos = 0;
 	else {
-		char *s = x.buf;
+		char *s = ff_xb.buf;
 		/* Last buf ended with a '\0', so don't let this buf start with one. */
 		while (len && *s == '\0')
 			s++, len--;
-		x.pos = s - x.buf;
+		ff_xb.pos = s - ff_xb.buf;
 	}
+
+#ifdef ICONV_OPTION
+	if (filesfrom_convert && len) {
+		char *sob = ff_xb.buf + ff_xb.pos, *s = sob;
+		char *eob = sob + len;
+		int flags = ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT;
+		if (ff_lastchar == '\0')
+			flags |= ICB_INIT;
+		/* Convert/send each null-terminated string separately, skipping empties. */
+		while (s != eob) {
+			if (*s++ == '\0') {
+				ff_xb.len = s - sob - 1;
+				if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0)
+					exit_cleanup(RERR_PROTOCOL); /* impossible? */
+				write_buf(iobuf.out_fd, s-1, 1); /* Send the '\0'. */
+				while (s != eob && *s == '\0')
+					s++;
+				sob = s;
+				ff_xb.pos = sob - ff_xb.buf;
+				flags |= ICB_INIT;
+			}
+		}
+
+		if ((ff_xb.len = s - sob) == 0)
+			ff_lastchar = '\0';
+		else {
+			/* Handle a partial string specially, saving any incomplete chars. */
+			flags &= ~ICB_INCLUDE_INCOMPLETE;
+			if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0) {
+				if (errno == E2BIG)
+					exit_cleanup(RERR_PROTOCOL); /* impossible? */
+				if (ff_xb.pos)
+					memmove(ff_xb.buf, ff_xb.buf + ff_xb.pos, ff_xb.len);
+			}
+			ff_lastchar = 'x'; /* Anything non-zero. */
+		}
+	} else
+#endif
+
 	if (len) {
-		char *f = x.buf + x.pos;
-		char *t = x.buf;
+		char *f = ff_xb.buf + ff_xb.pos;
+		char *t = ff_xb.buf;
 		char *eob = f + len;
 		/* Eliminate any multi-'\0' runs. */
 		while (f != eob) {
 			if (!(*t++ = *f++)) {
-				while (f != eob && !*f)
-					f++, len--;
+				while (f != eob && *f == '\0')
+					f++;
 			}
 		}
 		ff_lastchar = f[-1];
-	}
-#ifdef ICONV_OPTION
-	if (filesfrom_convert) {
-		/* TODO would it help to translate each string between nulls separately? */
-		x.len = len;
-		iconvbufs(ic_send, &x, &iobuf.out, ICB_INCLUDE_BAD|ICB_INCLUDE_INCOMPLETE|ICB_CIRCULAR_OUT|ICB_INIT);
-	} else
-#endif
-	if (len) {
-		/* This will not circle back to perform_io() because we only get
-		 * called when there is plenty of room in the output buffer. */
-		write_buf(iobuf.out_fd, x.buf, len);
+		if ((len = t - ff_xb.buf) != 0) {
+			/* This will not circle back to perform_io() because we only get
+			 * called when there is plenty of room in the output buffer. */
+			write_buf(iobuf.out_fd, ff_xb.buf, len);
+		}
 	}
 }
 
@@ -502,8 +542,8 @@ static char *perform_io(size_t needed, int flags)
 		if (DEBUG_GTE(IO, 3)) {
 			rprintf(FINFO, "[%s] perform_io(%ld, outroom) needs to flush %ld\n",
 				who_am_i(), (long)needed,
-				iobuf.out.len > iobuf.out.size - needed
-				? (long)iobuf.out.len - (iobuf.out.size - needed) : 0L);
+				iobuf.out.len + needed > iobuf.out.size
+				? (long)(iobuf.out.len + needed - iobuf.out.size) : 0L);
 		}
 		break;
 
@@ -518,8 +558,8 @@ static char *perform_io(size_t needed, int flags)
 		if (DEBUG_GTE(IO, 3)) {
 			rprintf(FINFO, "[%s] perform_io(%ld, msgroom) needs to flush %ld\n",
 				who_am_i(), (long)needed,
-				iobuf.out.len > iobuf.msg.size - needed
-				? (long)iobuf.out.len - (iobuf.msg.size - needed) : 0L);
+				iobuf.msg.len + needed > iobuf.msg.size
+				? (long)(iobuf.msg.len + needed - iobuf.msg.size) : 0L);
 		}
 		break;
 
@@ -539,11 +579,13 @@ static char *perform_io(size_t needed, int flags)
 				goto double_break;
 			break;
 		case PIO_NEED_OUTROOM:
-			if (iobuf.out.len <= iobuf.out.size - needed)
+			/* Note that iobuf.out_empty_len doesn't factor into this check
+			 * because iobuf.out.len already holds any needed header len. */
+			if (iobuf.out.len + needed <= iobuf.out.size)
 				goto double_break;
 			break;
 		case PIO_NEED_MSGROOM:
-			if (iobuf.msg.len <= iobuf.msg.size - needed)
+			if (iobuf.msg.len + needed <= iobuf.msg.size)
 				goto double_break;
 			break;
 		}
@@ -573,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)) {
@@ -599,6 +640,7 @@ static char *perform_io(size_t needed, int flags)
 					iobuf.raw_data_header_pos = iobuf.raw_flushing_ends_before;
 					if (iobuf.raw_data_header_pos >= iobuf.out.size)
 						iobuf.raw_data_header_pos -= iobuf.out.size;
+					/* Yes, it is possible for this to make len > size for a while. */
 					iobuf.out.len += 4;
 				}
 
@@ -620,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:
@@ -635,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;
@@ -669,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;
 				}
@@ -684,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))
@@ -712,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;
-					rsyserr(FERROR_SOCKET, errno, "write error");
-					exit_cleanup(RERR_STREAMIO);
+					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());
+					drain_multiplex_messages();
+					exit_cleanup(RERR_SOCKETIO);
 				}
 			}
 			if (msgs2stderr && DEBUG_GTE(IO, 2)) {
@@ -733,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;
@@ -763,6 +814,20 @@ static char *perform_io(size_t needed, int flags)
 	return data;
 }
 
+void noop_io_until_death(void)
+{
+	char buf[1024];
+
+	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);
+}
+
 /* 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)
 {
@@ -780,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
@@ -793,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) {
@@ -830,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)
@@ -981,18 +1046,16 @@ 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);
 }
 
 /* Read a line into the "buf" buffer. */
@@ -1101,7 +1164,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);
@@ -1116,8 +1179,9 @@ int 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);
-		return 0;
+		else
+			assert(f_out == iobuf.out_fd);
+		return False;
 	}
 
 	iobuf.out.size = IO_BUFFER_SIZE * 2 - 4;
@@ -1127,10 +1191,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);
@@ -1138,8 +1202,9 @@ int 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);
-		return 0;
+		else
+			assert(f_in == iobuf.in_fd);
+		return False;
 	}
 
 	iobuf.in.size = IO_BUFFER_SIZE;
@@ -1150,21 +1215,14 @@ int io_start_buffering_in(int f_in)
 
 	iobuf.in_fd = f_in;
 
-	return 1;
-}
-
-static void free_xbuf(xbuf *x)
-{
-	if (x->buf)
-		free(x->buf);
-	memset(x, 0, sizeof (xbuf));
+	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)
@@ -1177,9 +1235,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);
@@ -1187,9 +1245,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;
@@ -1237,7 +1292,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);
@@ -1275,9 +1330,21 @@ static void read_a_msg(void)
 		if (msg_bytes != 4 || am_sender)
 			goto invalid_msg;
 		data = perform_io(4, PIO_INPUT_AND_CONSUME);
-		io_error |= IVAL(data, 0);
+		val = IVAL(data, 0);
+		io_error |= val;
 		if (!am_generator)
-			send_msg(MSG_IO_ERROR, data, 4, 0);
+			send_msg_int(MSG_IO_ERROR, val);
+		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)
@@ -1296,20 +1363,26 @@ static void read_a_msg(void)
 			xbuf outbuf, inbuf;
 			char ibuf[512];
 			int add_null = 0;
+			int flags = ICB_INCLUDE_BAD | ICB_INIT;
 
 			INIT_CONST_XBUF(outbuf, line);
 			INIT_XBUF(inbuf, ibuf, 0, (size_t)-1);
 
 			while (msg_bytes) {
+				size_t len = msg_bytes > sizeof ibuf - inbuf.len ? sizeof ibuf - inbuf.len : msg_bytes;
+				memcpy(ibuf + inbuf.len, perform_io(len, PIO_INPUT_AND_CONSUME), len);
 				inbuf.pos = 0;
-				inbuf.len = msg_bytes > sizeof ibuf ? sizeof ibuf : msg_bytes;
-				memcpy(inbuf.buf, perform_io(inbuf.len, PIO_INPUT_AND_CONSUME), inbuf.len);
-				if (!(msg_bytes -= inbuf.len)
-				 && !ibuf[inbuf.len-1])
+				inbuf.len += len;
+				if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])
 					inbuf.len--, add_null = 1;
-				if (iconvbufs(ic_send, &inbuf, &outbuf,
-				    ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_INIT) < 0)
-					goto overflow;
+				if (iconvbufs(ic_send, &inbuf, &outbuf, flags) < 0) {
+					if (errno == E2BIG)
+						goto overflow;
+					/* Buffer ended with an incomplete char, so move the
+					 * bytes to the start of the buffer and continue. */
+					memmove(ibuf, ibuf + inbuf.pos, inbuf.len);
+				}
+				flags &= ~ICB_INIT;
 			}
 			if (add_null) {
 				if (outbuf.len == outbuf.size)
@@ -1337,19 +1410,21 @@ static void read_a_msg(void)
 			exit_cleanup(RERR_STREAMIO);
 		}
 		data = perform_io(4, PIO_INPUT_AND_CONSUME);
+		val = IVAL(data, 0);
 		if (am_generator)
-			got_flist_entry_status(FES_SUCCESS, IVAL(data, 0));
+			got_flist_entry_status(FES_SUCCESS, val);
 		else
-			successful_send(IVAL(data, 0));
+			successful_send(val);
 		break;
 	case MSG_NO_SEND:
 		if (msg_bytes != 4)
 			goto invalid_msg;
 		data = perform_io(4, PIO_INPUT_AND_CONSUME);
+		val = IVAL(data, 0);
 		if (am_generator)
-			got_flist_entry_status(FES_NO_SEND, IVAL(data, 0));
+			got_flist_entry_status(FES_NO_SEND, val);
 		else
-			send_msg(MSG_NO_SEND, data, 4, 0);
+			send_msg_int(MSG_NO_SEND, val);
 		break;
 	case MSG_ERROR_SOCKET:
 	case MSG_ERROR_UTF8:
@@ -1382,6 +1457,27 @@ static void read_a_msg(void)
 			first_message = 0;
 		}
 		break;
+	case MSG_ERROR_EXIT:
+		if (msg_bytes == 0) {
+			if (!am_sender && !am_generator) {
+				send_msg(MSG_ERROR_EXIT, "", 0, 0);
+				io_flush(FULL_FLUSH);
+			}
+			val = 0;
+		} else if (msg_bytes == 4) {
+			data = perform_io(4, PIO_INPUT_AND_CONSUME);
+			val = IVAL(data, 0);
+			if (protocol_version >= 31) {
+				if (am_generator)
+					send_msg_int(MSG_ERROR_EXIT, val);
+				else
+					send_msg(MSG_ERROR_EXIT, "", 0, 0);
+			}
+		} else
+			goto invalid_msg;
+		/* Send a negative linenum so that we don't end up
+		 * with a duplicate exit message. */
+		_exit_cleanup(val, __FILE__, 0 - __LINE__);
 	default:
 		rprintf(FERROR, "unexpected tag %d [%s%s]\n",
 			tag, who_am_i(), inc_recurse ? "/inc" : "");
@@ -1389,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)
@@ -1542,12 +1655,13 @@ 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;
 	}
 
 	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:
@@ -1571,6 +1685,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);
@@ -1828,7 +1943,7 @@ void write_buf(int f, const char *buf, size_t len)
 		goto batch_copy;
 	}
 
-	if (iobuf.out.size - iobuf.out.len < len)
+	if (iobuf.out.len + len > iobuf.out.size)
 		perform_io(len, PIO_NEED_OUTROOM);
 
 	pos = iobuf.out.pos + iobuf.out.len; /* Must be set after any flushing. */
@@ -1843,6 +1958,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)
@@ -1996,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);
@@ -2009,11 +2125,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);
 
@@ -2031,28 +2147,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)