patch from Alberto Accomazzi <aaccomazzi@cfa.harvard.edu> to add

[rsync/rsync.git] / io.c

diff --git a/io.c b/io.c
index 469eef3..7191140 100644 (file)
--- a/io.c
+++ b/io.c
@@ -24,6 +24,9 @@
   */
 #include "rsync.h"
 
+/* if no timeout is specified then use a 60 second select timeout */
+#define SELECT_TIMEOUT 60
+
 static int io_multiplexing_out;
 static int io_multiplexing_in;
 static int multiplex_in_fd;
@@ -57,7 +60,7 @@ 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);
        }
 }
 
@@ -67,6 +70,7 @@ 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
@@ -75,7 +79,9 @@ 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;
@@ -83,11 +89,10 @@ static int read_timeout(int fd, char *buf, int len)
 
                FD_ZERO(&fds);
                FD_SET(fd, &fds);
-               tv.tv_sec = io_timeout;
+               tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
                tv.tv_usec = 0;
 
-               if (select(fd+1, &fds, NULL, NULL, 
-                          io_timeout?&tv:NULL) != 1) {
+               if (select(fd+1, &fds, NULL, NULL, &tv) != 1) {
                        check_timeout();
                        continue;
                }
@@ -95,7 +100,6 @@ static int read_timeout(int fd, char *buf, int len)
                n = read(fd, buf, len);
 
                if (n > 0) {
-                       stats.total_read += n;
                        buf += n;
                        len -= n;
                        ret += n;
@@ -119,13 +123,13 @@ static int read_timeout(int fd, char *buf, int len)
 
                if (n == 0) {
                        if (eof_error) {
-                               rprintf(FERROR,"EOF in read_timeout\n");
+                               rprintf(FERROR,"unexpected EOF in read_timeout\n");
                        }
-                       exit_cleanup(1);
+                       exit_cleanup(RERR_STREAMIO);
                }
 
                rprintf(FERROR,"read error: %s\n", strerror(errno));
-               exit_cleanup(1);
+               exit_cleanup(RERR_STREAMIO);
        }
 
        return ret;
@@ -177,13 +181,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);
@@ -238,7 +242,7 @@ static void readfd(int fd,char *buffer,int N)
        int  ret;
        int total=0;  
        
-       if (read_buffer_len < N && N < 1024) {
+       if ((read_buffer_len < N) && (N < 1024)) {
                read_check(buffer_f_in);
        }
        
@@ -252,19 +256,27 @@ static void readfd(int fd,char *buffer,int N)
                        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)
@@ -274,11 +286,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);
@@ -318,38 +332,44 @@ 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;
+       int reading=0;
        int blocked=0;
 
        no_flush++;
 
-       reading = (buffer_f_in != -1 && read_buffer_len < MAX_READ_BUFFER);
-
        while (total < 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);
+               }
+
                if (reading) {
                        FD_SET(buffer_f_in,&r_fds);
                        if (buffer_f_in > fd) 
                                fd_count = buffer_f_in+1;
                }
 
-               tv.tv_sec = io_timeout;
+               tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
                tv.tv_usec = 0;
 
                count = select(fd_count,
                               reading?&r_fds:NULL,
                               &w_fds,NULL,
-                              io_timeout?&tv:NULL);
+                              &tv);
 
                if (count <= 0) {
                        check_timeout();
                        continue;
                }
 
+               if (reading && FD_ISSET(buffer_f_in, &r_fds)) {
+                       read_check(buffer_f_in);
+               }
+
                if (FD_ISSET(fd, &w_fds)) {
                        int n = (len-total)>>blocked;
                        int ret = write(fd,buf+total,n?n:1);
@@ -366,20 +386,14 @@ static void writefd_unbuffered(int fd,char *buf,int len)
 
                        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);
-                       continue;
-               }
-
-               if (reading && FD_ISSET(buffer_f_in, &r_fds)) {
-                       read_check(buffer_f_in);
                }
        }
 
@@ -427,6 +441,8 @@ void io_end_buffering(int fd)
 
 static void writefd(int fd,char *buf,int len)
 {
+       stats.total_written += len;
+
        if (!io_buffer) {
                writefd_unbuffered(fd, buf, len);
                return;
@@ -463,7 +479,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));
 
@@ -476,7 +492,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));
 }
@@ -525,7 +541,7 @@ void io_printf(int fd, const char *format, ...)
        len = vslprintf(buf, sizeof(buf)-1, format, ap);
        va_end(ap);
 
-       if (len < 0) exit_cleanup(1);
+       if (len < 0) exit_cleanup(RERR_STREAMIO);
 
        write_sbuf(fd, buf);
 }
@@ -547,7 +563,7 @@ void io_start_multiplex_in(int fd)
        io_flush();
        if (read_buffer_len) {
                fprintf(stderr,"ERROR: data in read buffer at mplx start\n");
-               exit_cleanup(1);
+               exit_cleanup(RERR_STREAMIO);
        }
 
        io_multiplexing_in = 1;
@@ -563,6 +579,8 @@ 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;
 }