X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/6e4fb64e6141727b8144bb77265bf0a4a8450a39..66203a982b3d249bafda9b9272c4c103c19e4a9b:/io.c diff --git a/io.c b/io.c index 5122a2a2..ae398f2d 100644 --- a/io.c +++ b/io.c @@ -38,6 +38,8 @@ extern int io_timeout; extern struct stats stats; static int buffer_f_in = -1; +static int io_error_fd = -1; +static int in_read_check; void setup_readbuffer(int f_in) { @@ -60,10 +62,32 @@ 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; @@ -86,21 +110,37 @@ static int read_timeout(int fd, char *buf, int len) 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 +153,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 +217,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); @@ -196,6 +231,8 @@ static int read_unbuffered(int fd, char *buf, int len) rprintf(tag,"%s", line); remaining = 0; + + if (in_read_check) break; } return ret; @@ -209,6 +246,8 @@ static int read_unbuffered(int fd, char *buf, int len) static void read_check(int f) { int n = 8192; + + in_read_check = 1; if (f == -1) return; @@ -233,6 +272,8 @@ static void read_check(int f) n = read_unbuffered(f,read_buffer+read_buffer_len,n); read_buffer_len += n; + + in_read_check = 0; } @@ -264,14 +305,20 @@ static void readfd(int fd,char *buffer,int N) 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 +328,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); @@ -326,7 +375,6 @@ static void writefd_unbuffered(int fd,char *buf,int len) int fd_count, count; struct timeval tv; int reading=0; - int blocked=0; no_flush++; @@ -334,7 +382,7 @@ 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); @@ -342,19 +390,30 @@ static void writefd_unbuffered(int fd,char *buf,int len) if (reading) { FD_SET(buffer_f_in,&r_fds); - if (buffer_f_in > fd) - fd_count = buffer_f_in+1; + if (buffer_f_in > fd_count) + fd_count = buffer_f_in; + } + + 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, + (reading || io_error_fd != -1)?&r_fds:NULL, &w_fds,NULL, &tv); if (count <= 0) { + if (errno == EBADF) { + exit_cleanup(RERR_SOCKETIO); + } check_timeout(); continue; } @@ -363,28 +422,27 @@ static void writefd_unbuffered(int fd,char *buf,int len) 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 +493,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 +531,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)); @@ -530,10 +590,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); } @@ -555,13 +615,13 @@ 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; } -/* 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 +631,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;