X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/1b7c47cb550e7600a66528c9cbd4bc1e880fb91e..b35d0d8e9ae9c5407c9f781b545f8a66b9caa9d0:/io.c

diff --git a/io.c b/io.c
index dbd9f643..ed6a70c3 100644
--- a/io.c
+++ b/io.c
@@ -1,6 +1,8 @@
-/* 
-   Copyright (C) Andrew Tridgell 1996
+/* -*- c-file-style: "linux" -*-
+   
+   Copyright (C) 1996-2001 by Andrew Tridgell 
    Copyright (C) Paul Mackerras 1996
+   Copyright (C) 2001, 2002 by Martin Pool <mbp@samba.org>
    
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -17,11 +19,21 @@
    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */
 
-/*
-  Utilities used in rsync 
+/**
+ * @file io.c
+ *
+ * Socket and pipe IO utilities used in rsync.
+ *
+ * rsync provides its own multiplexing system, which is used to send
+ * stderr and stdout over a single socket.  We need this because
+ * stdout normally carries the binary data stream, and stderr all our
+ * error messages.
+ *
+ * For historical reasons this is off during the start of the
+ * connection, but it's switched on quite early using
+ * io_start_multiplex_out() and io_start_multiplex_in().
+ **/
 
-  tridge, June 1996
-  */
 #include "rsync.h"
 
 /* if no timeout is specified then use a 60 second select timeout */
@@ -32,21 +44,29 @@ static int io_multiplexing_in;
 static int multiplex_in_fd;
 static int multiplex_out_fd;
 static time_t last_io;
-static int eof_error=1;
+static int no_flush;
+
+extern int bwlimit;
 extern int verbose;
 extern int io_timeout;
 extern struct stats stats;
 
-static int buffer_f_in = -1;
 
-void setup_readbuffer(int f_in)
-{
-	buffer_f_in = f_in;
-}
+/** Ignore EOF errors while reading a module listing if the remote
+    version is 24 or less. */
+int kludge_around_eof = False;
+
+
+static int io_error_fd = -1;
+
+static void read_loop(int fd, char *buf, size_t len);
 
 static void check_timeout(void)
 {
+	extern int am_server, am_daemon;
 	time_t t;
+
+	err_list_push();
 	
 	if (!io_timeout) return;
 
@@ -58,45 +78,146 @@ static void check_timeout(void)
 	t = time(NULL);
 
 	if (last_io && io_timeout && (t-last_io) >= io_timeout) {
-		rprintf(FERROR,"io timeout after %d second - exiting\n", 
-			(int)(t-last_io));
-		exit_cleanup(1);
+		if (!am_server && !am_daemon) {
+			rprintf(FERROR,"io timeout after %d seconds - exiting\n", 
+				(int)(t-last_io));
+		}
+		exit_cleanup(RERR_TIMEOUT);
 	}
 }
 
+/* setup the fd used to propogate errors */
+void io_set_error_fd(int fd)
+{
+	io_error_fd = fd;
+}
 
-static char *read_buffer;
-static char *read_buffer_p;
-static int read_buffer_len;
-static int read_buffer_size;
-static int no_flush;
-static int no_flush_read;
+/* read some data from the error fd and write it to the write log code */
+static void read_error_fd(void)
+{
+	char buf[200];
+	size_t n;
+	int fd = io_error_fd;
+	int tag, len;
 
-/* read from a socket with IO timeout. return the number of
-   bytes read. If no bytes can be read then exit, never return
-   a number <= 0 */
-static int read_timeout(int fd, char *buf, int len)
+        /* io_error_fd is temporarily disabled -- is this meant to
+         * prevent indefinite recursion? */
+	io_error_fd = -1;
+
+	read_loop(fd, buf, 4);
+	tag = IVAL(buf, 0);
+
+	len = tag & 0xFFFFFF;
+	tag = tag >> 24;
+	tag -= 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;
+	}
+
+	io_error_fd = fd;
+}
+
+
+static void whine_about_eof (void)
+{
+	/**
+	   It's almost always an error to get an EOF when we're trying
+	   to read from the network, because the protocol is
+	   self-terminating.
+	   
+	   However, there is one unfortunate cases where it is not,
+	   which is rsync <2.4.6 sending a list of modules on a
+	   server, since the list is terminated by closing the socket.
+	   So, for the section of the program where that is a problem
+	   (start_socket_client), kludge_around_eof is True and we
+	   just exit.
+	*/
+
+	if (kludge_around_eof)
+		exit_cleanup (0);
+	else {
+		rprintf (FERROR,
+			 "%s: connection unexpectedly closed "
+			 "(%.0f bytes read so far)\n",
+			 RSYNC_NAME, (double)stats.total_read);
+	
+		exit_cleanup (RERR_STREAMIO);
+	}
+}
+
+
+static void die_from_readerr (int err)
+{
+	/* this prevents us trying to write errors on a dead socket */
+	io_multiplexing_close();
+				
+	rprintf(FERROR, "%s: read error: %s\n",
+		RSYNC_NAME, strerror (err));
+	exit_cleanup(RERR_STREAMIO);
+}
+
+
+/*!
+ * Read from a socket with IO timeout. return the number of bytes
+ * read. If no bytes can be read then exit, never return a number <= 0.
+ *
+ * TODO: If the remote shell connection fails, then current versions
+ * actually report an "unexpected EOF" error here.  Since it's a
+ * fairly common mistake to try to use rsh when ssh is required, we
+ * should trap that: if we fail to read any data at all, we should
+ * give a better explanation.  We can tell whether the connection has
+ * started by looking e.g. at whether the remote version is known yet.
+ */
+static int read_timeout (int fd, char *buf, size_t len)
 {
 	int n, ret=0;
 
-	no_flush_read++;
 	io_flush();
-	no_flush_read--;
 
 	while (ret == 0) {
+		/* until we manage to read *something* */
 		fd_set fds;
 		struct timeval tv;
+		int fd_count = fd+1;
+		int count;
 
 		FD_ZERO(&fds);
 		FD_SET(fd, &fds);
+		if (io_error_fd != -1) {
+			FD_SET(io_error_fd, &fds);
+			if (io_error_fd > fd) fd_count = io_error_fd+1;
+		}
+
 		tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 		tv.tv_usec = 0;
 
-		if (select(fd+1, &fds, NULL, NULL, &tv) != 1) {
+		errno = 0;
+
+		count = select(fd_count, &fds, NULL, NULL, &tv);
+
+		if (count == 0) {
 			check_timeout();
+		}
+
+		if (count <= 0) {
+			if (errno == EBADF) {
+				exit_cleanup(RERR_SOCKETIO);
+			}
 			continue;
 		}
 
+		if (io_error_fd != -1 && FD_ISSET(io_error_fd, &fds)) {
+			read_error_fd();
+		}
+
+		if (!FD_ISSET(fd, &fds)) continue;
+
 		n = read(fd, buf, len);
 
 		if (n > 0) {
@@ -106,38 +227,27 @@ static int read_timeout(int fd, char *buf, int len)
 			if (io_timeout)
 				last_io = time(NULL);
 			continue;
+		} else if (n == 0) {
+			whine_about_eof ();
+			return -1; /* doesn't return */
+		} else if (n == -1) {
+			if (errno == EINTR || errno == EWOULDBLOCK ||
+			    errno == EAGAIN) 
+				continue;
+			else
+				die_from_readerr (errno);
 		}
-
-		if (n == -1 && errno == EINTR) {
-			continue;
-		}
-
-		if (n == -1 && 
-		    (errno == EAGAIN || errno == EWOULDBLOCK)) {
-			/* this shouldn't happen, if it does then
-			   sleep for a short time to prevent us
-			   chewing too much CPU */
-			u_sleep(100);
-			continue;
-		}
-
-		if (n == 0) {
-			if (eof_error) {
-				rprintf(FERROR,"unexpected EOF in read_timeout\n");
-			}
-			exit_cleanup(1);
-		}
-
-		rprintf(FERROR,"read error: %s\n", strerror(errno));
-		exit_cleanup(1);
 	}
 
 	return ret;
 }
 
-/* continue trying to read len bytes - don't return until len
-   has been read */
-static void read_loop(int fd, char *buf, int len)
+
+
+
+/*! Continue trying to read len bytes - don't return until len has
+  been read.   */
+static void read_loop (int fd, char *buf, size_t len)
 {
 	while (len) {
 		int n = read_timeout(fd, buf, len);
@@ -147,17 +257,20 @@ static void read_loop(int fd, char *buf, int len)
 	}
 }
 
-/* read from the file descriptor handling multiplexing - 
-   return number of bytes read
-   never return <= 0 */
-static int read_unbuffered(int fd, char *buf, int len)
+
+/**
+ * Read from the file descriptor handling multiplexing - return number
+ * of bytes read.
+ * 
+ * Never returns <= 0. 
+ */
+static int read_unbuffered(int fd, char *buf, size_t len)
 {
-	static int remaining;
-	char ibuf[4];
-	int tag, ret=0;
+	static size_t remaining;
+	int tag, ret = 0;
 	char line[1024];
 
-	if (!io_multiplexing_in || fd != multiplex_in_fd) 
+	if (!io_multiplexing_in || fd != multiplex_in_fd)
 		return read_timeout(fd, buf, len);
 
 	while (ret == 0) {
@@ -169,31 +282,32 @@ static int read_unbuffered(int fd, char *buf, int len)
 			continue;
 		}
 
-		read_loop(fd, ibuf, 4);
-		tag = IVAL(ibuf, 0);
+		read_loop(fd, line, 4);
+		tag = IVAL(line, 0);
 
 		remaining = tag & 0xFFFFFF;
 		tag = tag >> 24;
 
-		if (tag == MPLEX_BASE) continue;
+		if (tag == MPLEX_BASE)
+			continue;
 
 		tag -= MPLEX_BASE;
 
 		if (tag != FERROR && tag != FINFO) {
-			rprintf(FERROR,"unexpected tag %d\n", tag);
-			exit_cleanup(1);
+			rprintf(FERROR, "unexpected tag %d\n", tag);
+			exit_cleanup(RERR_STREAMIO);
 		}
 
-		if (remaining > sizeof(line)-1) {
-			rprintf(FERROR,"multiplexing overflow %d\n\n", 
+		if (remaining > sizeof(line) - 1) {
+			rprintf(FERROR, "multiplexing overflow %d\n\n",
 				remaining);
-			exit_cleanup(1);
+			exit_cleanup(RERR_STREAMIO);
 		}
 
 		read_loop(fd, line, remaining);
 		line[remaining] = 0;
 
-		rprintf(tag,"%s", line);
+		rprintf((enum logcode) tag, "%s", line);
 		remaining = 0;
 	}
 
@@ -202,65 +316,17 @@ static int read_unbuffered(int fd, char *buf, int len)
 
 
 
-/* This function was added to overcome a deadlock problem when using
- * ssh.  It looks like we can't allow our receive queue to get full or
- * ssh will clag up. Uggh.  */
-static void read_check(int f)
-{
-	int n = 8192;
-
-	if (f == -1) return;
-
-	if (read_buffer_len == 0) {
-		read_buffer_p = read_buffer;
-	}
-
-	if (n > MAX_READ_BUFFER/4)
-		n = MAX_READ_BUFFER/4;
-
-	if (read_buffer_p != read_buffer) {
-		memmove(read_buffer,read_buffer_p,read_buffer_len);
-		read_buffer_p = read_buffer;
-	}
-
-	if (n > (read_buffer_size - read_buffer_len)) {
-		read_buffer_size += n;
-		read_buffer = (char *)Realloc(read_buffer,read_buffer_size);
-		if (!read_buffer) out_of_memory("read check");      
-		read_buffer_p = read_buffer;      
-	}
-
-	n = read_unbuffered(f,read_buffer+read_buffer_len,n);
-	read_buffer_len += n;
-}
-
-
 /* do a buffered read from fd. don't return until all N bytes
    have been read. If all N can't be read then exit with an error */
-static void readfd(int fd,char *buffer,int N)
+static void readfd (int fd, char *buffer, size_t N)
 {
 	int  ret;
-	int total=0;  
-	
-	if ((read_buffer_len < N) && (N < 1024)) {
-		read_check(buffer_f_in);
-	}
+	size_t total=0;  
 	
 	while (total < N) {
-		if (read_buffer_len > 0 && buffer_f_in == fd) {
-			ret = MIN(read_buffer_len,N-total);
-			memcpy(buffer+total,read_buffer_p,ret);
-			read_buffer_p += ret;
-			read_buffer_len -= ret;
-			total += ret;
-			continue;
-		} 
-
-		no_flush_read++;
 		io_flush();
-		no_flush_read--;
 
-		ret = read_unbuffered(fd,buffer + total,N-total);
+		ret = read_unbuffered (fd, buffer + total, N-total);
 		total += ret;
 	}
 
@@ -292,7 +358,7 @@ int64 read_longint(int f)
 
 #ifdef NO_INT64
 	rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
-	exit_cleanup(1);
+	exit_cleanup(RERR_UNSUPPORTED);
 #else
 	if (remote_version >= 16) {
 		readfd(f,b,8);
@@ -303,37 +369,34 @@ int64 read_longint(int f)
 	return ret;
 }
 
-void read_buf(int f,char *buf,int len)
+void read_buf(int f,char *buf,size_t len)
 {
 	readfd(f,buf,len);
 }
 
-void read_sbuf(int f,char *buf,int len)
+void read_sbuf(int f,char *buf,size_t len)
 {
-	read_buf(f,buf,len);
+	read_buf (f,buf,len);
 	buf[len] = 0;
 }
 
 unsigned char read_byte(int f)
 {
 	unsigned char c;
-	read_buf(f,(char *)&c,1);
+	read_buf (f, (char *)&c, 1);
 	return c;
 }
 
-
-
-/* write len bytes to fd, possibly reading from buffer_f_in if set
-   in order to unclog the pipe. don't return until all len
-   bytes have been written */
-static void writefd_unbuffered(int fd,char *buf,int len)
+/* Write len bytes to fd.  This underlies the multiplexing system,
+ * which is always called by application code.  */
+static void writefd_unbuffered(int fd,char *buf,size_t len)
 {
-	int total = 0;
+	size_t total = 0;
 	fd_set w_fds, r_fds;
 	int fd_count, count;
 	struct timeval tv;
-	int reading=0;
-	int blocked=0;
+
+	err_list_push();
 
 	no_flush++;
 
@@ -341,55 +404,78 @@ static void writefd_unbuffered(int fd,char *buf,int len)
 		FD_ZERO(&w_fds);
 		FD_ZERO(&r_fds);
 		FD_SET(fd,&w_fds);
-		fd_count = fd+1;
-
-		if (!no_flush_read) {
-			reading = (buffer_f_in != -1);
-		}
+		fd_count = fd;
 
-		if (reading) {
-			FD_SET(buffer_f_in,&r_fds);
-			if (buffer_f_in > fd) 
-				fd_count = buffer_f_in+1;
+		if (io_error_fd != -1) {
+			FD_SET(io_error_fd,&r_fds);
+			if (io_error_fd > fd_count) 
+				fd_count = io_error_fd;
 		}
 
 		tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 		tv.tv_usec = 0;
 
-		count = select(fd_count,
-			       reading?&r_fds:NULL,
+		errno = 0;
+
+		count = select(fd_count+1,
+			       io_error_fd != -1?&r_fds:NULL,
 			       &w_fds,NULL,
 			       &tv);
 
-		if (count <= 0) {
+		if (count == 0) {
 			check_timeout();
+		}
+
+		if (count <= 0) {
+			if (errno == EBADF) {
+				exit_cleanup(RERR_SOCKETIO);
+			}
 			continue;
 		}
 
-		if (reading && FD_ISSET(buffer_f_in, &r_fds)) {
-			read_check(buffer_f_in);
+		if (io_error_fd != -1 && FD_ISSET(io_error_fd, &r_fds)) {
+			read_error_fd();
 		}
 
 		if (FD_ISSET(fd, &w_fds)) {
-			int n = (len-total)>>blocked;
-			int ret = write(fd,buf+total,n?n:1);
+			int ret;
+			size_t n = len-total;
+			ret = write(fd,buf+total,n);
 
 			if (ret == -1 && errno == EINTR) {
 				continue;
 			}
 
 			if (ret == -1 && 
-			    (errno == EAGAIN || errno == EWOULDBLOCK)) {
-				blocked++;
+			    (errno == EWOULDBLOCK || errno == EAGAIN)) {
+				msleep(1);
 				continue;
 			}
 
 			if (ret <= 0) {
-				rprintf(FERROR,"erroring writing %d bytes - exiting\n", len);
-				exit_cleanup(1);
+				/* Don't try to write errors back
+				 * across the stream */
+				io_multiplexing_close();
+				rprintf(FERROR, RSYNC_NAME
+					": error writing %d unbuffered bytes"
+					" - exiting: %s\n", len,
+					strerror(errno));
+				exit_cleanup(RERR_STREAMIO);
 			}
 
-			blocked = 0;
+			/* Sleep after writing to limit I/O bandwidth */
+			if (bwlimit)
+			{
+			    tv.tv_sec = 0;
+			    tv.tv_usec = ret * 1000 / bwlimit;
+			    while (tv.tv_usec > 1000000)
+			    {
+				tv.tv_sec++;
+				tv.tv_usec -= 1000000;
+			    }
+			    select(0, NULL, NULL, NULL, &tv);
+ 			}
+ 
 			total += ret;
 
 			if (io_timeout)
@@ -408,48 +494,75 @@ void io_start_buffering(int fd)
 {
 	if (io_buffer) return;
 	multiplex_out_fd = fd;
-	io_buffer = (char *)malloc(IO_BUFFER_SIZE+4);
+	io_buffer = (char *)malloc(IO_BUFFER_SIZE);
 	if (!io_buffer) out_of_memory("writefd");
 	io_buffer_count = 0;
+}
+
+/* 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)
+{
+	char buffer[4096];
+	size_t n = len;
+
+	SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
+
+	if (n > (sizeof(buffer)-4)) {
+		n = sizeof(buffer)-4;
+	}
 
-	/* leave room for the multiplex header in case it's needed */
-	io_buffer += 4;
+	memcpy(&buffer[4], buf, n);
+	writefd_unbuffered(fd, buffer, n+4);
+
+	len -= n;
+	buf += n;
+
+	if (len) {
+		writefd_unbuffered(fd, buf, len);
+	}
 }
 
+
 void io_flush(void)
 {
 	int fd = multiplex_out_fd;
+
+	err_list_push();
+
 	if (!io_buffer_count || no_flush) return;
 
 	if (io_multiplexing_out) {
-		SIVAL(io_buffer-4, 0, (MPLEX_BASE<<24) + io_buffer_count);
-		writefd_unbuffered(fd, io_buffer-4, io_buffer_count+4);
+		mplex_write(fd, FNONE, io_buffer, io_buffer_count);
 	} else {
 		writefd_unbuffered(fd, io_buffer, io_buffer_count);
 	}
 	io_buffer_count = 0;
 }
 
-void io_end_buffering(int fd)
+
+void io_end_buffering(void)
 {
 	io_flush();
 	if (!io_multiplexing_out) {
-		free(io_buffer-4);
+		free(io_buffer);
 		io_buffer = NULL;
 	}
 }
 
-static void writefd(int fd,char *buf,int len)
+static void writefd(int fd,char *buf,size_t len)
 {
 	stats.total_written += len;
 
-	if (!io_buffer) {
+	err_list_push();
+
+	if (!io_buffer || fd != multiplex_out_fd) {
 		writefd_unbuffered(fd, buf, len);
 		return;
 	}
 
 	while (len) {
-		int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count);
+		int n = MIN((int) len, IO_BUFFER_SIZE-io_buffer_count);
 		if (n > 0) {
 			memcpy(io_buffer+io_buffer_count, buf, n);
 			buf += n;
@@ -469,6 +582,11 @@ void write_int(int f,int32 x)
 	writefd(f,b,4);
 }
 
+
+/*
+ * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
+ * 64-bit types on this platform.
+ */
 void write_longint(int f, int64 x)
 {
 	extern int remote_version;
@@ -486,7 +604,7 @@ void write_longint(int f, int64 x)
 	writefd(f,b,8);
 }
 
-void write_buf(int f,char *buf,int len)
+void write_buf(int f,char *buf,size_t len)
 {
 	writefd(f,buf,len);
 }
@@ -503,14 +621,21 @@ void write_byte(int f,unsigned char c)
 	write_buf(f,(char *)&c,1);
 }
 
-int read_line(int f, char *buf, int maxlen)
-{
-	eof_error = 0;
 
+
+/**
+ * Read a line of up to @p maxlen characters into @p buf.  Does not
+ * contain a trailing newline or carriage return.
+ *
+ * @return 1 for success; 0 for io error or truncation.
+ **/
+int read_line(int f, char *buf, size_t maxlen)
+{
 	while (maxlen) {
 		buf[0] = 0;
 		read_buf(f, buf, 1);
-		if (buf[0] == 0) return 0;
+		if (buf[0] == 0)
+			return 0;
 		if (buf[0] == '\n') {
 			buf[0] = 0;
 			break;
@@ -525,8 +650,6 @@ int read_line(int f, char *buf, int maxlen)
 		return 0;
 	}
 
-	eof_error = 1;
-
 	return 1;
 }
 
@@ -538,10 +661,10 @@ void io_printf(int fd, const char *format, ...)
 	int len;
 	
 	va_start(ap, format);
-	len = vslprintf(buf, sizeof(buf)-1, format, ap);
+	len = vsnprintf(buf, sizeof(buf), format, ap);
 	va_end(ap);
 
-	if (len < 0) exit_cleanup(1);
+	if (len < 0) exit_cleanup(RERR_STREAMIO);
 
 	write_sbuf(fd, buf);
 }
@@ -561,31 +684,23 @@ void io_start_multiplex_in(int fd)
 {
 	multiplex_in_fd = fd;
 	io_flush();
-	if (read_buffer_len) {
-		fprintf(stderr,"ERROR: data in read buffer at mplx start\n");
-		exit_cleanup(1);
-	}
-
 	io_multiplexing_in = 1;
 }
 
-/* write an message to the error stream */
-int io_multiplex_write(int f, char *buf, int len)
+/* write an message to the multiplexed error stream */
+int io_multiplex_write(enum logcode code, char *buf, size_t len)
 {
 	if (!io_multiplexing_out) return 0;
 
 	io_flush();
-
-	SIVAL(io_buffer-4, 0, ((MPLEX_BASE + f)<<24) + len);
-	memcpy(io_buffer, buf, len);
-
 	stats.total_written += (len+4);
-
-	writefd_unbuffered(multiplex_out_fd, io_buffer-4, len+4);
+	mplex_write(multiplex_out_fd, code, buf, len);
 	return 1;
 }
 
-void io_close_input(int fd)
+/* stop output multiplexing */
+void io_multiplexing_close(void)
 {
-	buffer_f_in = -1;
+	io_multiplexing_out = 0;
 }
+