X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/679e765768f85485818ecb33d084de677e3ac1a7..b0f3f5784c8635fe08d5d8e397f476a0c9ccba66:/io.c diff --git a/io.c b/io.c index a91b3db6..4fd1a138 100644 --- a/io.c +++ b/io.c @@ -24,34 +24,23 @@ */ #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; static int multiplex_in_fd; static int multiplex_out_fd; static time_t last_io; - +static int eof_error=1; extern int verbose; -extern int sparse_files; 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; -void setup_nonblocking(int f_in,int f_out) +void setup_readbuffer(int f_in) { - set_blocking(f_out,0); buffer_f_in = f_in; } @@ -68,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); } } @@ -80,45 +69,82 @@ 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; - -/* continue trying to read len bytes - don't return until len - has been read */ -static void read_loop(int fd, char *buf, int 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) { - while (len) { - int n = read(fd, buf, len); + int n, ret=0; + + no_flush_read++; + io_flush(); + no_flush_read--; + + while (ret == 0) { + fd_set fds; + struct timeval tv; + + FD_ZERO(&fds); + FD_SET(fd, &fds); + tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT; + tv.tv_usec = 0; + + if (select(fd+1, &fds, NULL, NULL, &tv) != 1) { + check_timeout(); + continue; + } + + n = read(fd, buf, len); + if (n > 0) { buf += n; len -= n; + ret += n; + if (io_timeout) + last_io = time(NULL); + continue; } - if (n == 0) { - rprintf(FERROR,"EOF in read_loop\n"); - exit_cleanup(1); + + if (n == -1 && errno == EINTR) { + continue; } - if (n == -1) { - fd_set fds; - struct timeval tv; - - if (errno != EAGAIN && errno != EWOULDBLOCK) { - rprintf(FERROR,"io error: %s\n", - strerror(errno)); - exit_cleanup(1); - } - FD_ZERO(&fds); - FD_SET(fd, &fds); - tv.tv_sec = io_timeout; - tv.tv_usec = 0; - if (select(fd+1, &fds, NULL, NULL, - io_timeout?&tv:NULL) != 1) { - check_timeout(); + if (n == 0) { + if (eof_error) { + rprintf(FERROR,"unexpected EOF in read_timeout\n"); } + exit_cleanup(RERR_STREAMIO); } + + /* this prevents us trying to write errors on a dead socket */ + io_multiplexing_out = 0; + + rprintf(FERROR,"read error: %s\n", strerror(errno)); + exit_cleanup(RERR_STREAMIO); } + + 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) +{ + while (len) { + int n = read_timeout(fd, buf, len); + + buf += n; + len -= n; + } +} + +/* 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) { static int remaining; @@ -127,7 +153,7 @@ static int read_unbuffered(int fd, char *buf, int len) char line[1024]; if (!io_multiplexing_in || fd != multiplex_in_fd) - return read(fd, buf, len); + return read_timeout(fd, buf, len); while (ret == 0) { if (remaining) { @@ -150,13 +176,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); @@ -170,12 +196,13 @@ 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; + int n = 8192; if (f == -1) return; @@ -183,12 +210,6 @@ static void read_check(int f) read_buffer_p = read_buffer; } - if ((n=num_waiting(f)) <= 0) - return; - - /* things could deteriorate if we read in really small chunks */ - if (n < 10) n = 1024; - if (n > MAX_READ_BUFFER/4) n = MAX_READ_BUFFER/4; @@ -199,28 +220,26 @@ static void read_check(int f) if (n > (read_buffer_size - read_buffer_len)) { read_buffer_size += n; - if (!read_buffer) - read_buffer = (char *)malloc(read_buffer_size); - else - read_buffer = (char *)realloc(read_buffer,read_buffer_size); + 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); - if (n > 0) { - read_buffer_len += n; - } + read_buffer_len += n; } -static int readfd(int fd,char *buffer,int 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) { int ret; int total=0; - struct timeval tv; - if (read_buffer_len < N) + if ((read_buffer_len < N) && (N < 1024)) { read_check(buffer_f_in); + } while (total < N) { if (read_buffer_len > 0 && buffer_f_in == fd) { @@ -232,47 +251,27 @@ static int readfd(int fd,char *buffer,int N) continue; } + no_flush_read++; io_flush(); + no_flush_read--; - while ((ret = read_unbuffered(fd,buffer + total,N-total)) == -1) { - fd_set fds; - - if (errno != EAGAIN && errno != EWOULDBLOCK) - return -1; - FD_ZERO(&fds); - FD_SET(fd, &fds); - tv.tv_sec = io_timeout; - tv.tv_usec = 0; - - if (select(fd+1, &fds, NULL, NULL, - io_timeout?&tv:NULL) != 1) { - check_timeout(); - } - } - - if (ret <= 0) - return total; + ret = read_unbuffered(fd,buffer + total,N-total); total += ret; } - if (io_timeout) - last_io = time(NULL); - return total; + stats.total_read += total; } int32 read_int(int f) { - int ret; - char b[4]; - if ((ret=readfd(f,b,4)) != 4) { - if (verbose > 1) - rprintf(FERROR,"(%d) read_int: Error reading %d bytes : %s\n", - getpid(),4,ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_read += 4; - return IVAL(b,0); + char b[4]; + int32 ret; + + readfd(f,b,4); + ret = IVAL(b,0); + if (ret == (int32)0xffffffff) return -1; + return ret; } int64 read_longint(int f) @@ -282,20 +281,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) { - if ((ret=readfd(f,b,8)) != 8) { - if (verbose > 1) - rprintf(FERROR,"(%d) read_longint: Error reading %d bytes : %s\n", - getpid(),8,ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_read += 8; + readfd(f,b,8); ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32); } #endif @@ -305,14 +300,7 @@ int64 read_longint(int f) void read_buf(int f,char *buf,int len) { - int ret; - if ((ret=readfd(f,buf,len)) != len) { - if (verbose > 1) - rprintf(FERROR,"(%d) read_buf: Error reading %d bytes : %s\n", - getpid(),len,ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_read += len; + readfd(f,buf,len); } void read_sbuf(int f,char *buf,int len) @@ -323,149 +311,83 @@ void read_sbuf(int f,char *buf,int len) unsigned char read_byte(int f) { - unsigned char c; - read_buf(f,(char *)&c,1); - return c; + unsigned char c; + read_buf(f,(char *)&c,1); + return c; } -static char last_byte; -static int last_sparse; - -int sparse_end(int f) -{ - if (last_sparse) { - do_lseek(f,-1,SEEK_CUR); - return (write(f,&last_byte,1) == 1 ? 0 : -1); - } - last_sparse = 0; - return 0; -} - -static int write_sparse(int f,char *buf,int len) -{ - int l1=0,l2=0; - int ret; - - for (l1=0;l1 0) - last_sparse=1; - - if (l1 > 0) - do_lseek(f,l1,SEEK_CUR); - - if (l1 == len) - return len; - - if ((ret=write(f,buf+l1,len-(l1+l2))) != len-(l1+l2)) { - if (ret == -1 || ret == 0) return ret; - return (l1+ret); - } - - if (l2 > 0) - do_lseek(f,l2,SEEK_CUR); - - return len; -} - - - -int write_file(int f,char *buf,int len) -{ - int ret = 0; - - if (!sparse_files) - return write(f,buf,len); - - while (len>0) { - int len1 = MIN(len, SPARSE_WRITE_SIZE); - int r1 = write_sparse(f, buf, len1); - if (r1 <= 0) { - if (ret > 0) return ret; - return r1; - } - len -= r1; - buf += r1; - ret += r1; - } - return ret; -} - - -static int writefd_unbuffered(int fd,char *buf,int len) +/* 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) { int total = 0; fd_set w_fds, r_fds; - int fd_count, count, got_select=0; + int fd_count, count; struct timeval tv; + int reading=0; - while (total < len) { - int ret = write(fd,buf+total,len-total); - - if (ret == 0) return total; - - if (ret == -1 && !(errno == EWOULDBLOCK || errno == EAGAIN)) - return -1; - - if (ret == -1 && got_select) { - /* hmmm, we got a write select on the fd and - then failed to write. Why doesn't that - mean that the fd is dead? It doesn't on - some systems it seems (eg. IRIX) */ - u_sleep(1000); - } - - got_select = 0; - - - if (ret != -1) { - total += ret; - continue; - } - - if (read_buffer_len < MAX_READ_BUFFER && buffer_f_in != -1) - read_check(buffer_f_in); + no_flush++; - fd_count = fd+1; + while (total < len) { FD_ZERO(&w_fds); FD_ZERO(&r_fds); FD_SET(fd,&w_fds); - if (buffer_f_in != -1) { + 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 = BLOCKING_TIMEOUT; + tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT; tv.tv_usec = 0; - count = select(fd_count,buffer_f_in == -1? NULL: &r_fds, - &w_fds,NULL,&tv); - - if (count == -1 && errno != EINTR) { - if (verbose > 1) - rprintf(FERROR,"select error: %s\n", strerror(errno)); - exit_cleanup(1); - } - - if (count == 0) { + + count = select(fd_count, + reading?&r_fds:NULL, + &w_fds,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)) { - got_select = 1; + int ret, n = len-total; + + if (n > PIPE_BUF) n = PIPE_BUF; + + ret = write(fd,buf+total,n?n:1); + + if (ret == -1 && errno == EINTR) { + continue; + } + + if (ret <= 0) { + rprintf(FERROR,"erroring writing %d bytes - exiting\n", len); + exit_cleanup(RERR_STREAMIO); + } + + total += ret; + + if (io_timeout) + last_io = time(NULL); } } - if (io_timeout) - last_io = time(NULL); - - return total; + no_flush--; } @@ -487,21 +409,13 @@ 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); - if (writefd_unbuffered(fd, io_buffer-4, io_buffer_count+4) != - io_buffer_count+4) { - rprintf(FERROR,"write failed\n"); - exit_cleanup(1); - } + writefd_unbuffered(fd, io_buffer-4, io_buffer_count+4); } else { - if (writefd_unbuffered(fd, io_buffer, io_buffer_count) != - io_buffer_count) { - rprintf(FERROR,"write failed\n"); - exit_cleanup(1); - } + writefd_unbuffered(fd, io_buffer, io_buffer_count); } io_buffer_count = 0; } @@ -515,11 +429,14 @@ void io_end_buffering(int fd) } } -static int writefd(int fd,char *buf,int len1) +static void writefd(int fd,char *buf,int len) { - int len = len1; + stats.total_written += len; - if (!io_buffer) return writefd_unbuffered(fd, buf, len); + if (!io_buffer) { + writefd_unbuffered(fd, buf, len); + return; + } while (len) { int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count); @@ -532,60 +449,40 @@ static int writefd(int fd,char *buf,int len1) if (io_buffer_count == IO_BUFFER_SIZE) io_flush(); } - - return len1; } void write_int(int f,int32 x) { - int ret; char b[4]; SIVAL(b,0,x); - if ((ret=writefd(f,b,4)) != 4) { - rprintf(FERROR,"write_int failed : %s\n", - ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_written += 4; + writefd(f,b,4); } void write_longint(int f, int64 x) { extern int remote_version; char b[8]; - int ret; if (remote_version < 16 || x <= 0x7FFFFFFF) { write_int(f, (int)x); return; } - write_int(f, -1); + write_int(f, (int32)0xFFFFFFFF); SIVAL(b,0,(x&0xFFFFFFFF)); SIVAL(b,4,((x>>32)&0xFFFFFFFF)); - if ((ret=writefd(f,b,8)) != 8) { - rprintf(FERROR,"write_longint failed : %s\n", - ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_written += 8; + writefd(f,b,8); } void write_buf(int f,char *buf,int len) { - int ret; - if ((ret=writefd(f,buf,len)) != len) { - rprintf(FERROR,"write_buf failed : %s\n", - ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_written += len; + writefd(f,buf,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)); } @@ -596,15 +493,14 @@ void write_byte(int f,unsigned char c) write_buf(f,(char *)&c,1); } -void write_flush(int f) -{ -} - - int read_line(int f, char *buf, int maxlen) { + eof_error = 0; + while (maxlen) { + buf[0] = 0; read_buf(f, buf, 1); + if (buf[0] == 0) return 0; if (buf[0] == '\n') { buf[0] = 0; break; @@ -618,6 +514,9 @@ int read_line(int f, char *buf, int maxlen) *buf = 0; return 0; } + + eof_error = 1; + return 1; } @@ -629,10 +528,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); } @@ -654,7 +553,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; @@ -670,6 +569,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; }