Index: cleanup.c
--- cleanup.c	15 Dec 2003 08:08:44 -0000	1.19
+++ cleanup.c	15 Jan 2004 07:32:19 -0000
@@ -118,7 +118,7 @@ void _exit_cleanup(int code, const char 
 		if (cleanup_fd2 != -1) close(cleanup_fd2);
 		finish_transfer(cleanup_new_fname, fname, cleanup_file);
 	}
-	io_flush();
+	io_flush(FULL_FLUSH);
 	if (cleanup_fname)
 		do_unlink(cleanup_fname);
 	if (code) {
Index: generator.c
--- generator.c	12 Jan 2004 03:50:07 -0000	1.69
+++ generator.c	15 Jan 2004 07:32:19 -0000
@@ -522,7 +522,7 @@ void recv_generator(char *fname, struct 
 }
 
 
-void generate_files(int f,struct file_list *flist,char *local_name,int f_recv)
+void generate_files(int f, struct file_list *flist, char *local_name)
 {
 	int i;
 	int phase=0;
@@ -576,7 +576,7 @@ void generate_files(int f,struct file_li
 
 	/* files can cycle through the system more than once
 	 * to catch initial checksum errors */
-	while ((i = read_int(f_recv)) != -1) {
+	while ((i = get_redo_num()) != -1) {
 		struct file_struct *file = flist->files[i];
 		recv_generator(local_name? local_name
 			     : f_name_to(file,fbuf,sizeof fbuf), file, i, f);
Index: io.c
--- io.c	2 Jan 2004 08:33:57 -0000	1.117
+++ io.c	15 Jan 2004 07:32:19 -0000
@@ -49,6 +49,9 @@ static int no_flush;
 extern int bwlimit;
 extern int verbose;
 extern int io_timeout;
+extern int am_server;
+extern int am_daemon;
+extern int am_sender;
 extern struct stats stats;
 
 
@@ -74,8 +77,9 @@ const char *io_read_phase = phase_unknow
     version is 24 or less. */
 int kludge_around_eof = False;
 
+int msg_fd_in = -1;
+int msg_fd_out = -1;
 
-static int io_error_fd = -1;
 static int io_filesfrom_f_in = -1;
 static int io_filesfrom_f_out = -1;
 static char io_filesfrom_buf[2048];
@@ -85,14 +89,44 @@ static int io_filesfrom_buflen;
 
 static void read_loop(int fd, char *buf, size_t len);
 
+struct redo_list {
+	struct redo_list *next;
+	int num;
+};
+
+static struct redo_list *redo_list_head;
+static struct redo_list *redo_list_tail;
+
+struct msg_list {
+	struct msg_list *next;
+	char *buf;
+	int len;
+};
+
+static struct msg_list *msg_list_head;
+static struct msg_list *msg_list_tail;
+
+static void redo_list_add(int num)
+{
+	struct redo_list *rl;
+
+	if (!(rl = new(struct redo_list)))
+		exit_cleanup(RERR_MALLOC);
+	rl->next = NULL;
+	rl->num = num;
+	if (redo_list_tail)
+		redo_list_tail->next = rl;
+	else
+		redo_list_head = rl;
+	redo_list_tail = rl;
+}
+
 static void check_timeout(void)
 {
-	extern int am_server, am_daemon;
 	time_t t;
 
-	err_list_push();
-
-	if (!io_timeout) return;
+	if (!io_timeout)
+		return;
 
 	if (!last_io) {
 		last_io = time(NULL);
@@ -110,41 +144,155 @@ static void check_timeout(void)
 	}
 }
 
-/** Setup the fd used to propagate errors */
-void io_set_error_fd(int fd)
+/** Setup the fd used to receive MSG_* messages.  Only needed when
+ * we're the generator because the sender and receiver both use the
+ * multiplexed IO setup. */
+void set_msg_fd_in(int fd)
+{
+	msg_fd_in = fd;
+}
+
+/** Setup the fd used to send our MSG_* messages.  Only needed when
+ * we're the receiver because the generator and the sender both use
+ * the multiplexed IO setup. */
+void set_msg_fd_out(int fd)
+{
+	msg_fd_out = fd;
+	set_nonblocking(msg_fd_out);
+}
+
+/* Add a message to the pending MSG_* list. */
+static void msg_list_add(int code, char *buf, int len)
+{
+	struct msg_list *ml;
+
+	if (!(ml = new(struct msg_list)))
+		exit_cleanup(RERR_MALLOC);
+	ml->next = NULL;
+	if (!(ml->buf = new_array(char, len+4)))
+		exit_cleanup(RERR_MALLOC);
+	SIVAL(ml->buf, 0, ((code+MPLEX_BASE)<<24) | len);
+	memcpy(ml->buf+4, buf, len);
+	ml->len = len+4;
+	if (msg_list_tail)
+		msg_list_tail->next = ml;
+	else
+		msg_list_head = ml;
+	msg_list_tail = ml;
+}
+
+void send_msg(enum msgcode code, char *buf, int len)
 {
-	io_error_fd = fd;
+	msg_list_add(code, buf, len);
+	msg_list_push(NORMAL_FLUSH);
 }
 
-/** Read some data from the error fd and write it to the write log code */
-static void read_error_fd(void)
+/** Read a message from the MSG_* fd and dispatch it.  This is only
+ * called by the generator. */
+static void read_msg_fd(void)
 {
 	char buf[200];
 	size_t n;
-	int fd = io_error_fd;
+	int fd = msg_fd_in;
 	int tag, len;
 
-	/* io_error_fd is temporarily disabled -- is this meant to
-	 * prevent indefinite recursion? */
-	io_error_fd = -1;
+	/* Temporarily disable msg_fd_in.  This is needed because we
+	 * may call a write routine that could try to call us back. */
+	msg_fd_in = -1;
 
 	read_loop(fd, buf, 4);
 	tag = IVAL(buf, 0);
 
 	len = tag & 0xFFFFFF;
-	tag = tag >> 24;
-	tag -= MPLEX_BASE;
+	tag = (tag >> 24) - MPLEX_BASE;
 
-	while (len) {
-		n = len;
-		if (n > (sizeof buf - 1))
-			n = sizeof buf - 1;
-		read_loop(fd, buf, n);
-		rwrite((enum logcode)tag, buf, n);
-		len -= n;
+	switch (tag) {
+	case MSG_DONE:
+		if (len != 0)
+			exit_cleanup(RERR_STREAMIO);
+		redo_list_add(-1);
+		break;
+	case MSG_REDO:
+		if (len != 4)
+			exit_cleanup(RERR_STREAMIO);
+		read_loop(fd, buf, 4);
+		redo_list_add(IVAL(buf,0));
+		break;
+	case MSG_INFO:
+	case MSG_ERROR:
+	case MSG_LOG:
+		while (len) {
+			n = len;
+			if (n >= sizeof buf)
+				n = sizeof buf - 1;
+			read_loop(fd, buf, n);
+			rwrite((enum logcode)tag, buf, n);
+			len -= n;
+		}
+		break;
+	default:
+		exit_cleanup(RERR_STREAMIO);
+	}
+
+	msg_fd_in = fd;
+}
+
+/* Try to push messages off the list onto the wire.  If we leave with more
+ * to do, return 0.  On error, return -1.  If everything flushed, return 1.
+ * This is only called by the receiver. */
+int msg_list_push(int flush_it_all)
+{
+	static int written = 0;
+	struct timeval tv;
+	fd_set fds;
+
+	if (msg_fd_out < 0)
+		return -1;
+
+	while (msg_list_head) {
+		struct msg_list *ml = msg_list_head;
+		int n = write(msg_fd_out, ml->buf + written, ml->len - written);
+		if (n < 0) {
+			if (errno == EINTR)
+				continue;
+			if (errno != EWOULDBLOCK && errno != EAGAIN)
+				return -1;
+			if (!flush_it_all)
+				return 0;
+			FD_ZERO(&fds);
+			FD_SET(msg_fd_out, &fds);
+			tv.tv_sec = io_timeout ? io_timeout : SELECT_TIMEOUT;
+			tv.tv_usec = 0;
+			if (!select(msg_fd_out+1, NULL, &fds, NULL, &tv))
+				check_timeout();
+		} else if ((written += n) == ml->len) {
+			free(ml->buf);
+			msg_list_head = ml->next;
+			if (!msg_list_head)
+				msg_list_tail = NULL;
+			free(ml);
+			written = 0;
+		}
 	}
+	return 1;
+}
+
+int get_redo_num(void)
+{
+	struct redo_list *next;
+	int num;
+
+	while (!redo_list_head)
+		read_msg_fd();
 
-	io_error_fd = fd;
+	num = redo_list_head->num;
+	next = redo_list_head->next;
+	free(redo_list_head);
+	redo_list_head = next;
+	if (!next)
+		redo_list_tail = NULL;
+
+	return num;
 }
 
 /**
@@ -217,7 +365,7 @@ static int read_timeout(int fd, char *bu
 {
 	int n, ret=0;
 
-	io_flush();
+	io_flush(NORMAL_FLUSH);
 
 	while (ret == 0) {
 		/* until we manage to read *something* */
@@ -228,9 +376,10 @@ static int read_timeout(int fd, char *bu
 
 		FD_ZERO(&r_fds);
 		FD_SET(fd, &r_fds);
-		if (io_error_fd >= 0) {
-			FD_SET(io_error_fd, &r_fds);
-			if (io_error_fd >= fd_count) fd_count = io_error_fd+1;
+		if (msg_fd_in >= 0) {
+			FD_SET(msg_fd_in, &r_fds);
+			if (msg_fd_in >= fd_count)
+				fd_count = msg_fd_in+1;
 		}
 		if (io_filesfrom_f_out >= 0) {
 			int new_fd;
@@ -247,7 +396,8 @@ static int read_timeout(int fd, char *bu
 				FD_SET(io_filesfrom_f_out, &w_fds);
 				new_fd = io_filesfrom_f_out;
 			}
-			if (new_fd >= fd_count) fd_count = new_fd+1;
+			if (new_fd >= fd_count)
+				fd_count = new_fd+1;
 		}
 
 		tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
@@ -260,6 +410,7 @@ static int read_timeout(int fd, char *bu
 			       NULL, &tv);
 
 		if (count == 0) {
+			msg_list_push(NORMAL_FLUSH);
 			check_timeout();
 		}
 
@@ -270,9 +421,8 @@ static int read_timeout(int fd, char *bu
 			continue;
 		}
 
-		if (io_error_fd >= 0 && FD_ISSET(io_error_fd, &r_fds)) {
-			read_error_fd();
-		}
+		if (msg_fd_in >= 0 && FD_ISSET(msg_fd_in, &r_fds))
+			read_msg_fd();
 
 		if (io_filesfrom_f_out >= 0) {
 			if (io_filesfrom_buflen) {
@@ -471,9 +621,10 @@ static int read_unbuffered(int fd, char 
 		tag = IVAL(line, 0);
 
 		remaining = tag & 0xFFFFFF;
-		tag = tag >> 24;
+		tag = (tag >> 24) - MPLEX_BASE;
 
-		if (tag == MPLEX_BASE) {
+		switch (tag) {
+		case MSG_DATA:
 			if (!buffer || remaining > bufferSz) {
 				buffer = realloc_array(buffer, char, remaining);
 				if (!buffer) out_of_memory("read_unbuffered");
@@ -481,31 +632,26 @@ static int read_unbuffered(int fd, char 
 			}
 			read_loop(fd, buffer, remaining);
 			bufferIdx = 0;
-			continue;
-		}
-
-		tag -= MPLEX_BASE;
-
-		if (tag != FERROR && tag != FINFO) {
+			break;
+		case MSG_INFO:
+		case MSG_ERROR:
+			if (remaining >= sizeof line) {
+				rprintf(FERROR, "multiplexing overflow %d:%ld\n\n",
+					tag, (long)remaining);
+				exit_cleanup(RERR_STREAMIO);
+			}
+			read_loop(fd, line, remaining);
+			rwrite((enum logcode)tag, line, remaining);
+			remaining = 0;
+			break;
+		default:
 			rprintf(FERROR, "unexpected tag %d\n", tag);
 			exit_cleanup(RERR_STREAMIO);
 		}
-
-		if (remaining > sizeof line - 1) {
-			rprintf(FERROR, "multiplexing overflow %ld\n\n",
-				(long)remaining);
-			exit_cleanup(RERR_STREAMIO);
-		}
-
-		read_loop(fd, line, remaining);
-		line[remaining] = 0;
-
-		rprintf((enum logcode) tag, "%s", line);
-		remaining = 0;
 	}
 
 	if (remaining == 0)
-		io_flush();
+		io_flush(NORMAL_FLUSH);
 
 	return ret;
 }
@@ -622,7 +768,7 @@ static void writefd_unbuffered(int fd,ch
 	int fd_count, count;
 	struct timeval tv;
 
-	err_list_push();
+	msg_list_push(NORMAL_FLUSH);
 
 	no_flush++;
 
@@ -631,24 +777,22 @@ static void writefd_unbuffered(int fd,ch
 		FD_SET(fd,&w_fds);
 		fd_count = fd;
 
-		if (io_error_fd >= 0) {
+		if (msg_fd_in >= 0) {
 			FD_ZERO(&r_fds);
-			FD_SET(io_error_fd,&r_fds);
-			if (io_error_fd > fd_count) 
-				fd_count = io_error_fd;
+			FD_SET(msg_fd_in,&r_fds);
+			if (msg_fd_in > fd_count) 
+				fd_count = msg_fd_in;
 		}
 
 		tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 		tv.tv_usec = 0;
 
 		errno = 0;
-
-		count = select(fd_count+1,
-			       io_error_fd >= 0?&r_fds:NULL,
-			       &w_fds,NULL,
-			       &tv);
+		count = select(fd_count+1, msg_fd_in >= 0 ? &r_fds : NULL,
+			       &w_fds, NULL, &tv);
 
 		if (count == 0) {
+			msg_list_push(NORMAL_FLUSH);
 			check_timeout();
 		}
 
@@ -659,9 +803,8 @@ static void writefd_unbuffered(int fd,ch
 			continue;
 		}
 
-		if (io_error_fd >= 0 && FD_ISSET(io_error_fd, &r_fds)) {
-			read_error_fd();
-		}
+		if (msg_fd_in >= 0 && FD_ISSET(msg_fd_in, &r_fds))
+			read_msg_fd();
 
 		if (FD_ISSET(fd, &w_fds)) {
 			int ret;
@@ -722,7 +865,7 @@ void io_start_buffering_in(int fd)
  * Write an message to a multiplexed stream. If this fails then rsync
  * exits.
  **/
-static void mplex_write(int fd, enum logcode code, char *buf, size_t len)
+static void mplex_write(int fd, enum msgcode code, char *buf, size_t len)
 {
 	char buffer[4096];
 	size_t n = len;
@@ -745,26 +888,26 @@ static void mplex_write(int fd, enum log
 }
 
 
-void io_flush(void)
+void io_flush(int flush_it_all)
 {
 	int fd = multiplex_out_fd;
+	
+	msg_list_push(flush_it_all);
 
-	err_list_push();
-
-	if (!io_buffer_count || no_flush) return;
+	if (!io_buffer_count || no_flush)
+		return;
 
-	if (io_multiplexing_out) {
-		mplex_write(fd, FNONE, io_buffer, io_buffer_count);
-	} else {
+	if (io_multiplexing_out)
+		mplex_write(fd, MSG_DATA, io_buffer, io_buffer_count);
+	else
 		writefd_unbuffered(fd, io_buffer, io_buffer_count);
-	}
 	io_buffer_count = 0;
 }
 
 
 void io_end_buffering(void)
 {
-	io_flush();
+	io_flush(NORMAL_FLUSH);
 	if (!io_multiplexing_out) {
 		free(io_buffer);
 		io_buffer = NULL;
@@ -775,7 +918,7 @@ static void writefd(int fd,char *buf,siz
 {
 	stats.total_written += len;
 
-	err_list_push();
+	msg_list_push(NORMAL_FLUSH);
 
 	if (!io_buffer || fd != multiplex_out_fd) {
 		writefd_unbuffered(fd, buf, len);
@@ -791,7 +934,8 @@ static void writefd(int fd,char *buf,siz
 			io_buffer_count += n;
 		}
 
-		if (io_buffer_count == IO_BUFFER_SIZE) io_flush();
+		if (io_buffer_count == IO_BUFFER_SIZE)
+			io_flush(NORMAL_FLUSH);
 	}
 }
 
@@ -903,35 +1047,35 @@ void io_printf(int fd, const char *forma
 }
 
 
-/** Setup for multiplexing an error stream with the data stream */
+/** Setup for multiplexing a MSG_* stream with the data stream. */
 void io_start_multiplex_out(int fd)
 {
 	multiplex_out_fd = fd;
-	io_flush();
+	io_flush(NORMAL_FLUSH);
 	io_start_buffering_out(fd);
 	io_multiplexing_out = 1;
 }
 
-/** Setup for multiplexing an error stream with the data stream */
+/** Setup for multiplexing a MSG_* stream with the data stream. */
 void io_start_multiplex_in(int fd)
 {
 	multiplex_in_fd = fd;
-	io_flush();
+	io_flush(NORMAL_FLUSH);
 	io_multiplexing_in = 1;
 }
 
-/** Write an message to the multiplexed error stream */
-int io_multiplex_write(enum logcode code, char *buf, size_t len)
+/** Write an message to the multiplexed data stream. */
+int io_multiplex_write(enum msgcode code, char *buf, size_t len)
 {
 	if (!io_multiplexing_out) return 0;
 
-	io_flush();
+	io_flush(NORMAL_FLUSH);
 	stats.total_written += (len+4);
 	mplex_write(multiplex_out_fd, code, buf, len);
 	return 1;
 }
 
-/** Stop output multiplexing */
+/** Stop output multiplexing. */
 void io_multiplexing_close(void)
 {
 	io_multiplexing_out = 0;
Index: log.c
--- log.c	13 Jan 2004 04:50:45 -0000	1.69
+++ log.c	15 Jan 2004 07:32:20 -0000
@@ -30,6 +30,7 @@
 extern int am_daemon;
 extern int am_server;
 extern int am_sender;
+extern int msg_fd_out;
 extern int quiet;
 extern int module_id;
 extern char *auth_user;
@@ -38,7 +39,6 @@ extern char *log_format;
 static int log_initialised;
 static char *logfname;
 static FILE *logfile;
-static int log_error_fd = -1;
 struct stats stats;
 
 int log_got_error=0;
@@ -85,61 +85,6 @@ static char const *rerr_name(int code)
 	return NULL;
 }
 
-struct err_list {
-	struct err_list *next;
-	char *buf;
-	int len;
-	int written; /* how many bytes we have written so far */
-};
-
-static struct err_list *err_list_head;
-static struct err_list *err_list_tail;
-
-/* add an error message to the pending error list */
-static void err_list_add(int code, char *buf, int len)
-{
-	struct err_list *el;
-	el = new(struct err_list);
-	if (!el) exit_cleanup(RERR_MALLOC);
-	el->next = NULL;
-	el->buf = new_array(char, len+4);
-	if (!el->buf) exit_cleanup(RERR_MALLOC);
-	memcpy(el->buf+4, buf, len);
-	SIVAL(el->buf, 0, ((code+MPLEX_BASE)<<24) | len);
-	el->len = len+4;
-	el->written = 0;
-	if (err_list_tail) {
-		err_list_tail->next = el;
-	} else {
-		err_list_head = el;
-	}
-	err_list_tail = el;
-}
-
-
-/* try to push errors off the error list onto the wire */
-void err_list_push(void)
-{
-	if (log_error_fd == -1) return;
-
-	while (err_list_head) {
-		struct err_list *el = err_list_head;
-		int n = write(log_error_fd, el->buf+el->written, el->len - el->written);
-		/* don't check for an error if the best way of handling the error is
-		 * to ignore it */
-		if (n == -1) break;
-		if (n > 0) {
-			el->written += n;
-		}
-		if (el->written == el->len) {
-			free(el->buf);
-			err_list_head = el->next;
-			if (!err_list_head) err_list_tail = NULL;
-			free(el);
-		}
-	}
-}
-
 
 static void logit(int priority, char *buf)
 {
@@ -211,14 +156,6 @@ void log_close(void)
 	}
 }
 
-/* setup the error file descriptor - used when we are a server
- * that is receiving files */
-void set_error_fd(int fd)
-{
-	log_error_fd = fd;
-	set_nonblocking(log_error_fd);
-}
-
 /* this is the underlying (unformatted) rsync debugging function. Call
  * it with FINFO, FERROR or FLOG */
 void rwrite(enum logcode code, char *buf, int len)
@@ -226,9 +163,11 @@ void rwrite(enum logcode code, char *buf
 	FILE *f=NULL;
 	/* recursion can happen with certain fatal conditions */
 
-	if (quiet && code == FINFO) return;
+	if (quiet && code == FINFO)
+		return;
 
-	if (len < 0) exit_cleanup(RERR_MESSAGEIO);
+	if (len < 0)
+		exit_cleanup(RERR_MESSAGEIO);
 
 	buf[len] = 0;
 
@@ -237,17 +176,14 @@ void rwrite(enum logcode code, char *buf
 		return;
 	}
 
-	/* first try to pass it off to our sibling */
-	if (am_server && log_error_fd != -1) {
-		err_list_add(code, buf, len);
-		err_list_push();
-		return;
-	}
-
-	/* next, if we are a server and multiplexing is enabled,
-	 * pass it to the other side.  */
-	if (am_server && io_multiplex_write(code, buf, len)) {
-		return;
+	if (am_server) {
+		/* Pass it to non-server side, perhaps through our sibling. */
+		if (msg_fd_out >= 0) {
+			send_msg((enum msgcode)code, buf, len);
+			return;
+		}
+		if (io_multiplex_write((enum msgcode)code, buf, len))
+			return;
 	}
 
 	/* otherwise, if in daemon mode and either we are not a server
Index: main.c
--- main.c	2 Jan 2004 08:29:49 -0000	1.178
+++ main.c	15 Jan 2004 07:32:20 -0000
@@ -55,7 +55,7 @@ void wait_process(pid_t pid, int *status
 
 	while ((waited_pid = waitpid(pid, status, WNOHANG)) == 0) {
 		msleep(20);
-		io_flush();
+		io_flush(FULL_FLUSH);
 	}
 
 	if ((waited_pid == -1) && (errno == ECHILD)) {
@@ -383,13 +383,13 @@ static void do_server_sender(int f_in, i
 	io_start_buffering_in(f_in);
 	io_start_buffering_out(f_out);
 	send_files(flist,f_out,f_in);
-	io_flush();
+	io_flush(FULL_FLUSH);
 	report(f_out);
 	if (protocol_version >= 24) {
 		/* final goodbye message */
 		read_int(f_in);
 	}
-	io_flush();
+	io_flush(FULL_FLUSH);
 	exit_cleanup(0);
 }
 
@@ -398,7 +398,6 @@ static int do_recv(int f_in,int f_out,st
 {
 	int pid;
 	int status=0;
-	int recv_pipe[2];
 	int error_pipe[2];
 	extern int preserve_hard_links;
 	extern int delete_after;
@@ -415,20 +414,14 @@ static int do_recv(int f_in,int f_out,st
 		}
 	}
 
-	if (fd_pair(recv_pipe) < 0) {
-		rprintf(FERROR,"pipe failed in do_recv\n");
-		exit_cleanup(RERR_SOCKETIO);
-	}
-
 	if (fd_pair(error_pipe) < 0) {
 		rprintf(FERROR,"error pipe failed in do_recv\n");
 		exit_cleanup(RERR_SOCKETIO);
 	}
 
-	io_flush();
+	io_flush(NORMAL_FLUSH);
 
 	if ((pid=do_fork()) == 0) {
-		close(recv_pipe[0]);
 		close(error_pipe[0]);
 		if (f_in != f_out) close(f_out);
 
@@ -436,41 +429,39 @@ static int do_recv(int f_in,int f_out,st
 		io_multiplexing_close();
 
 		/* set place to send errors */
-		set_error_fd(error_pipe[1]);
+		set_msg_fd_out(error_pipe[1]);
 
-		recv_files(f_in,flist,local_name,recv_pipe[1]);
-		io_flush();
+		recv_files(f_in,flist,local_name);
+		io_flush(FULL_FLUSH);
 		report(f_in);
 
-		write_int(recv_pipe[1],1);
-		close(recv_pipe[1]);
-		io_flush();
+		send_msg(MSG_DONE, "", 0);
+		io_flush(FULL_FLUSH);
 		/* finally we go to sleep until our parent kills us
 		 * with a USR2 signal. We sleep for a short time as on
 		 * some OSes a signal won't interrupt a sleep! */
-		while (msleep(20))
-			;
+		while (1)
+			msleep(20);
 	}
 
-	close(recv_pipe[1]);
 	close(error_pipe[1]);
 	if (f_in != f_out) close(f_in);
 
 	io_start_buffering_out(f_out);
 
-	io_set_error_fd(error_pipe[0]);
+	set_msg_fd_in(error_pipe[0]);
 
-	generate_files(f_out,flist,local_name,recv_pipe[0]);
+	generate_files(f_out, flist, local_name);
 
-	read_int(recv_pipe[0]);
-	close(recv_pipe[0]);
+	get_redo_num(); /* Read final -1, and any prior messages. */
+	io_flush(FULL_FLUSH);
 	if (protocol_version >= 24) {
 		/* send a final goodbye message */
 		write_int(f_out, -1);
 	}
-	io_flush();
+	io_flush(FULL_FLUSH);
 
-	io_set_error_fd(-1);
+	set_msg_fd_in(-1);
 	kill(pid, SIGUSR2);
 	wait_process(pid, &status);
 	return status;
@@ -621,10 +612,10 @@ int client_run(int f_in, int f_out, pid_
 		if (verbose > 3)
 			rprintf(FINFO,"file list sent\n");
 
-		io_flush();
+		io_flush(NORMAL_FLUSH);
 		io_start_buffering_out(f_out);
 		send_files(flist,f_out,f_in);
-		io_flush();
+		io_flush(FULL_FLUSH);
 		if (protocol_version >= 24) {
 			/* final goodbye message */
 			read_int(f_in);
@@ -632,11 +623,11 @@ int client_run(int f_in, int f_out, pid_
 		if (pid != -1) {
 			if (verbose > 3)
 				rprintf(FINFO,"client_run waiting on %d\n", (int) pid);
-			io_flush();
+			io_flush(FULL_FLUSH);
 			wait_process(pid, &status);
 		}
 		report(-1);
-		io_flush();
+		io_flush(FULL_FLUSH);
 		exit_cleanup(status);
 	}
 
@@ -668,7 +659,7 @@ int client_run(int f_in, int f_out, pid_
 	if (pid != -1) {
 		if (verbose > 3)
 			rprintf(FINFO,"client_run2 waiting on %d\n", (int) pid);
-		io_flush();
+		io_flush(FULL_FLUSH);
 		wait_process(pid, &status);
 	}
 
Index: proto.h
--- proto.h	12 Jan 2004 03:50:16 -0000	1.171
+++ proto.h	15 Jan 2004 07:32:20 -0000
@@ -89,10 +89,14 @@ char *f_name_to(struct file_struct *f, c
 char *f_name(struct file_struct *f);
 void write_sum_head(int f, struct sum_struct *sum);
 void recv_generator(char *fname, struct file_struct *file, int i, int f_out);
-void generate_files(int f,struct file_list *flist,char *local_name,int f_recv);
+void generate_files(int f, struct file_list *flist, char *local_name);
 void init_hard_links(struct file_list *flist);
 void do_hard_links(void);
-void io_set_error_fd(int fd);
+void set_msg_fd_in(int fd);
+void set_msg_fd_out(int fd);
+void send_msg(enum msgcode code, char *buf, int len);
+int msg_list_push(int flush_it_all);
+int get_redo_num(void);
 void io_set_filesfrom_fds(int f_in, int f_out);
 int read_filesfrom_line(int fd, char *fname);
 int32 read_int(int f);
@@ -102,7 +106,7 @@ void read_sbuf(int f,char *buf,size_t le
 unsigned char read_byte(int f);
 void io_start_buffering_out(int fd);
 void io_start_buffering_in(int fd);
-void io_flush(void);
+void io_flush(int flush_it_all);
 void io_end_buffering(void);
 void write_int(int f,int32 x);
 void write_int_named(int f, int32 x, const char *phase);
@@ -113,7 +117,7 @@ int read_line(int f, char *buf, size_t m
 void io_printf(int fd, const char *format, ...);
 void io_start_multiplex_out(int fd);
 void io_start_multiplex_in(int fd);
-int io_multiplex_write(enum logcode code, char *buf, size_t len);
+int io_multiplex_write(enum msgcode code, char *buf, size_t len);
 void io_multiplexing_close(void);
 char *lp_motd_file(void);
 char *lp_log_file(void);
@@ -149,11 +153,9 @@ int lp_max_connections(int );
 BOOL lp_load(char *pszFname, int globals_only);
 int lp_numservices(void);
 int lp_number(char *name);
-void err_list_push(void);
 void log_init(void);
 void log_open(void);
 void log_close(void);
-void set_error_fd(int fd);
 void rwrite(enum logcode code, char *buf, int len);
 void rprintf(enum logcode code, const char *format, ...);
 void rsyserr(enum logcode code, int errcode, const char *format, ...);
@@ -183,7 +185,7 @@ pid_t local_child(int argc, char **argv,
 void end_progress(OFF_T size);
 void show_progress(OFF_T ofs, OFF_T size);
 void delete_files(struct file_list *flist);
-int recv_files(int f_in,struct file_list *flist,char *local_name,int f_gen);
+int recv_files(int f_in,struct file_list *flist,char *local_name);
 void free_sums(struct sum_struct *s);
 int delete_file(char *fname);
 int set_perms(char *fname,struct file_struct *file,STRUCT_STAT *st,
Index: receiver.c
--- receiver.c	13 Jan 2004 18:22:43 -0000	1.62
+++ receiver.c	15 Jan 2004 07:32:20 -0000
@@ -280,7 +280,7 @@ static int receive_data(int f_in,struct 
  * main routine for receiver process.
  *
  * Receiver process runs on the same host as the generator process. */
-int recv_files(int f_in,struct file_list *flist,char *local_name,int f_gen)
+int recv_files(int f_in,struct file_list *flist,char *local_name)
 {
 	int fd1,fd2;
 	STRUCT_STAT st;
@@ -314,7 +314,7 @@ int recv_files(int f_in,struct file_list
 				csum_length = SUM_LENGTH;
 				if (verbose > 2)
 					rprintf(FINFO,"recv_files phase=%d\n",phase);
-				write_int(f_gen,-1);
+				send_msg(MSG_DONE, "", 0);
 				continue;
 			}
 			break;
@@ -466,9 +466,11 @@ int recv_files(int f_in,struct file_list
 				rprintf(FERROR,"ERROR: file corruption in %s. File changed during transfer?\n",
 					full_fname(fname));
 			} else {
+				char buf[4];
 				if (verbose > 1)
 					rprintf(FINFO,"redoing %s(%d)\n",fname,i);
-				write_int(f_gen,i);
+				SIVAL(buf, 0, i);
+				send_msg(MSG_REDO, buf, 4);
 			}
 		}
 	}
Index: rsync.h
--- rsync.h	15 Jan 2004 02:25:38 -0000	1.172
+++ rsync.h	15 Jan 2004 07:32:20 -0000
@@ -106,10 +106,21 @@
 #define ADD_INCLUDE 1
 #define ADD_EXCLUDE 0
 
-/* Log values.  I *think* what these mean is: FLOG goes to the server
- * logfile; FERROR and FINFO try to end up on the client, with
- * different levels of filtering. */
-enum logcode {FNONE=0, FERROR=1, FINFO=2, FLOG=3 };
+#define FULL_FLUSH	1
+#define NORMAL_FLUSH	0
+
+/* Log-message categories.  FLOG is only used on the daemon side to
+ * output messages to the log file. */
+enum logcode { FERROR=1, FINFO=2, FLOG=3 };
+
+/* Messages types that are sent over the message channel.  The logcode
+ * values must all be present here with identical numbers. */
+enum msgcode {
+	MSG_DATA=0,	/* raw data on the multiplexed stream */
+	MSG_ERROR=FERROR, MSG_INFO=FINFO, MSG_LOG=FLOG, /* remote logging */
+	MSG_REDO=4,	/* reprocess indicated flist index */
+	MSG_DONE=5,	/* current phase is done */
+};
 
 #include "errcode.h"
 
