X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/bf9f01689f9cba70ed595ddcfabdf92d4b03cb80..d1be231290b3867648ee417fad341fdf1caaa94b:/io.c diff --git a/io.c b/io.c index d7140d26..25d75168 100644 --- 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; @@ -54,10 +57,10 @@ static void check_timeout(void) t = time(NULL); - if (last_io && io_timeout && (t-last_io)>io_timeout) { - rprintf(FERROR,"read timeout after %d second - exiting\n", + 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)); } @@ -522,10 +538,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); } @@ -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; }