X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/2f9af9011887f585e152c28db52166a78a0b5d5d..08f15335b5b469b29cbdacd0fc5ba48d76acf280:/io.c

diff --git a/io.c b/io.c
index 0f8ab919..6f3854e3 100644
--- a/io.c
+++ b/io.c
@@ -18,7 +18,7 @@
 */
 
 /*
-  Utilities used in rsync 
+  socket and pipe IO utilities used in rsync 
 
   tridge, June 1996
   */
@@ -38,6 +38,7 @@ extern int io_timeout;
 extern struct stats stats;
 
 static int buffer_f_in = -1;
+static int io_error_fd = -1;
 
 void setup_readbuffer(int f_in)
 {
@@ -60,17 +61,34 @@ static void check_timeout(void)
 	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);
+		exit_cleanup(RERR_TIMEOUT);
 	}
 }
 
+/* setup the fd used to propogate errors */
+void io_set_error_fd(int fd)
+{
+	io_error_fd = fd;
+}
+
+/* read some data from the error fd and write it to FERROR */
+static void read_error_fd(void)
+{
+	char buf[200];
+	int n;
+	int fd = io_error_fd;
+	io_error_fd = -1;
+
+	n = read(fd, buf, sizeof(buf)-1);
+	if (n > 0) {
+		rwrite(FERROR, buf, n);
+	}
+
+	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 from a socket with IO timeout. return the number of
    bytes read. If no bytes can be read then exit, never return
@@ -79,28 +97,42 @@ static int read_timeout(int fd, char *buf, int len)
 {
 	int n, ret=0;
 
-	no_flush_read++;
 	io_flush();
-	no_flush_read--;
 
 	while (ret == 0) {
 		fd_set fds;
 		struct timeval tv;
+		int fd_count = fd+1;
 
 		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;
+
+		if (select(fd_count, &fds, NULL, NULL, &tv) < 1) {
+			if (errno == EBADF) {
+				exit_cleanup(RERR_SOCKETIO);
+			}
 			check_timeout();
 			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) {
-			stats.total_read += n;
 			buf += n;
 			len -= n;
 			ret += n;
@@ -113,24 +145,19 @@ static int read_timeout(int fd, char *buf, int len)
 			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);
+			exit_cleanup(RERR_STREAMIO);
 		}
 
+		/* this prevents us trying to write errors on a dead socket */
+		io_multiplexing_close();
+
 		rprintf(FERROR,"read error: %s\n", strerror(errno));
-		exit_cleanup(1);
+		exit_cleanup(RERR_STREAMIO);
 	}
 
 	return ret;
@@ -182,13 +209,13 @@ static int read_unbuffered(int fd, char *buf, int len)
 
 		if (tag != FERROR && tag != FINFO) {
 			rprintf(FERROR,"unexpected tag %d\n", tag);
-			exit_cleanup(1);
+			exit_cleanup(RERR_STREAMIO);
 		}
 
 		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);
@@ -202,40 +229,6 @@ 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)
@@ -243,35 +236,26 @@ static void readfd(int fd,char *buffer,int N)
 	int  ret;
 	int total=0;  
 	
-	if ((read_buffer_len < N) && (N < 1024)) {
-		read_check(buffer_f_in);
-	}
-	
 	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);
 		total += ret;
 	}
+
+	stats.total_read += total;
 }
 
 
 int32 read_int(int f)
 {
 	char b[4];
+	int32 ret;
+
 	readfd(f,b,4);
-	return IVAL(b,0);
+	ret = IVAL(b,0);
+	if (ret == (int32)0xffffffff) return -1;
+	return ret;
 }
 
 int64 read_longint(int f)
@@ -281,11 +265,13 @@ int64 read_longint(int f)
 	char b[8];
 	ret = read_int(f);
 
-	if ((int32)ret != (int32)0xffffffff) return ret;
+	if ((int32)ret != (int32)0xffffffff) {
+		return ret;
+	}
 
 #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);
@@ -325,8 +311,6 @@ static void writefd_unbuffered(int fd,char *buf,int len)
 	fd_set w_fds, r_fds;
 	int fd_count, count;
 	struct timeval tv;
-	int reading=0;
-	int blocked=0;
 
 	no_flush++;
 
@@ -334,57 +318,53 @@ 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;
+		fd_count = fd;
 
-		if (!no_flush_read) {
-			reading = (buffer_f_in != -1);
-		}
-
-		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 (errno == EBADF) {
+				exit_cleanup(RERR_SOCKETIO);
+			}
 			check_timeout();
 			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, n = len-total;
+			
+			if (n > PIPE_BUF) n = PIPE_BUF;
 
-			if (ret == -1 && errno == EINTR) {
-				continue;
-			}
+			ret = write(fd,buf+total,n?n:1);
 
-			if (ret == -1 && 
-			    (errno == EAGAIN || errno == EWOULDBLOCK)) {
-				blocked++;
+			if (ret == -1 && errno == EINTR) {
 				continue;
 			}
 
 			if (ret <= 0) {
 				rprintf(FERROR,"erroring writing %d bytes - exiting\n", len);
-				exit_cleanup(1);
+				exit_cleanup(RERR_STREAMIO);
 			}
 
-			blocked = 0;
 			total += ret;
-			stats.total_written += ret;
 
 			if (io_timeout)
 				last_io = time(NULL);
@@ -435,7 +415,9 @@ void io_end_buffering(int fd)
 
 static void writefd(int fd,char *buf,int len)
 {
-	if (!io_buffer) {
+	stats.total_written += len;
+
+	if (!io_buffer || fd != multiplex_out_fd) {
 		writefd_unbuffered(fd, buf, len);
 		return;
 	}
@@ -471,7 +453,7 @@ void write_longint(int f, int64 x)
 		return;
 	}
 
-	write_int(f, -1);
+	write_int(f, (int32)0xFFFFFFFF);
 	SIVAL(b,0,(x&0xFFFFFFFF));
 	SIVAL(b,4,((x>>32)&0xFFFFFFFF));
 
@@ -484,7 +466,7 @@ void write_buf(int f,char *buf,int len)
 }
 
 /* write a string to the connection */
-void write_sbuf(int f,char *buf)
+static void write_sbuf(int f,char *buf)
 {
 	write_buf(f, buf, strlen(buf));
 }
@@ -530,10 +512,10 @@ void io_printf(int fd, const char *format, ...)
 	int len;
 	
 	va_start(ap, format);
-	len = vslprintf(buf, sizeof(buf)-1, format, ap);
+	len = vslprintf(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);
 }
@@ -553,15 +535,10 @@ 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 */
+/* write an message to the multiplexed error stream */
 int io_multiplex_write(int f, char *buf, int len)
 {
 	if (!io_multiplexing_out) return 0;
@@ -571,10 +548,26 @@ int io_multiplex_write(int f, char *buf, int len)
 	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);
 	return 1;
 }
 
+/* write a message to the special error fd */
+int io_error_write(int f, char *buf, int len)
+{
+	if (f == -1) return 0;
+	writefd_unbuffered(f, buf, len);
+	return 1;
+}
+
+/* stop output multiplexing */
+void io_multiplexing_close(void)
+{
+	io_multiplexing_out = 0;
+}
+
 void io_close_input(int fd)
 {
 	buffer_f_in = -1;