X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/fe8c0a9824bf990ae3ff692d25105dd2c9f0ebf4..654175798bdbdd6403e10c8fa74e8586b3612ea1:/io.c diff --git a/io.c b/io.c index 8eb62c38..71911408 100644 --- a/io.c +++ b/io.c @@ -24,8 +24,8 @@ */ #include "rsync.h" -static int64 total_written; -static int64 total_read; +/* 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; @@ -35,17 +35,7 @@ static time_t last_io; static int eof_error=1; extern int verbose; extern int io_timeout; - - -int64 write_total(void) -{ - return total_written; -} - -int64 read_total(void) -{ - return total_read; -} +extern struct stats stats; static int buffer_f_in = -1; @@ -67,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); } } @@ -79,6 +69,8 @@ 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 @@ -87,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; @@ -95,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; } @@ -119,15 +112,24 @@ 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,"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; @@ -145,7 +147,7 @@ static void read_loop(int fd, char *buf, int len) } } -/* read from the file descriptor handing multiplexing - +/* 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) @@ -179,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); @@ -240,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); } @@ -254,20 +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); - total_read += 4; - return IVAL(b,0); + ret = IVAL(b,0); + if (ret == (int32)0xffffffff) return -1; + return ret; } int64 read_longint(int f) @@ -277,15 +286,16 @@ 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); - total_read += 8; ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32); } #endif @@ -296,7 +306,6 @@ int64 read_longint(int f) void read_buf(int f,char *buf,int len) { readfd(f,buf,len); - total_read += len; } void read_sbuf(int f,char *buf,int len) @@ -323,9 +332,10 @@ 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; - reading = (buffer_f_in != -1 && read_buffer_len < MAX_READ_BUFFER); + no_flush++; while (total < len) { FD_ZERO(&w_fds); @@ -333,47 +343,61 @@ static void writefd_unbuffered(int fd,char *buf,int len) 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 ret = write(fd,buf+total,len-total); + int n = (len-total)>>blocked; + int ret = write(fd,buf+total,n?n:1); if (ret == -1 && errno == EINTR) { continue; } + if (ret == -1 && + (errno == EAGAIN || errno == EWOULDBLOCK)) { + blocked++; + continue; + } + if (ret <= 0) { rprintf(FERROR,"erroring writing %d bytes - exiting\n", len); - exit_cleanup(1); + exit_cleanup(RERR_STREAMIO); } + blocked = 0; total += ret; + if (io_timeout) last_io = time(NULL); - continue; - } - - if (reading && FD_ISSET(buffer_f_in, &r_fds)) { - read_check(buffer_f_in); } } + + no_flush--; } @@ -395,7 +419,7 @@ void io_start_buffering(int fd) void io_flush(void) { int fd = multiplex_out_fd; - if (!io_buffer_count) return; + if (!io_buffer_count || no_flush) return; if (io_multiplexing_out) { SIVAL(io_buffer-4, 0, (MPLEX_BASE<<24) + io_buffer_count); @@ -417,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; @@ -441,7 +467,6 @@ void write_int(int f,int32 x) char b[4]; SIVAL(b,0,x); writefd(f,b,4); - total_written += 4; } void write_longint(int f, int64 x) @@ -454,22 +479,20 @@ 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)); writefd(f,b,8); - total_written += 8; } void write_buf(int f,char *buf,int len) { writefd(f,buf,len); - total_written += 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)); } @@ -518,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); } @@ -540,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; @@ -556,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; }