X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/5d1e1dcf4b9d4d9e0308d3e7f7036b4e5191e323..08f15335b5b469b29cbdacd0fc5ba48d76acf280:/io.c diff --git a/io.c b/io.c index 4fd1a138..6f3854e3 100644 --- a/io.c +++ b/io.c @@ -18,7 +18,7 @@ */ /* - Utilities used in rsync + socket and pipe IO utilities used in rsync tridge, June 1996 */ @@ -38,6 +38,7 @@ extern int io_timeout; extern struct stats stats; static int buffer_f_in = -1; +static int io_error_fd = -1; void setup_readbuffer(int f_in) { @@ -64,13 +65,30 @@ static void check_timeout(void) } } +/* 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; -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 @@ -79,24 +97,39 @@ 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; 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) { @@ -121,7 +154,7 @@ static int read_timeout(int fd, char *buf, int len) } /* this prevents us trying to write errors on a dead socket */ - io_multiplexing_out = 0; + io_multiplexing_close(); rprintf(FERROR,"read error: %s\n", strerror(errno)); exit_cleanup(RERR_STREAMIO); @@ -196,40 +229,6 @@ 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 = 8192; - - if (f == -1) return; - - if (read_buffer_len == 0) { - read_buffer_p = read_buffer; - } - - if (n > MAX_READ_BUFFER/4) - n = MAX_READ_BUFFER/4; - - if (read_buffer_p != read_buffer) { - memmove(read_buffer,read_buffer_p,read_buffer_len); - read_buffer_p = read_buffer; - } - - if (n > (read_buffer_size - read_buffer_len)) { - read_buffer_size += n; - 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); - read_buffer_len += 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) @@ -237,23 +236,8 @@ static void readfd(int fd,char *buffer,int N) int ret; int total=0; - if ((read_buffer_len < N) && (N < 1024)) { - read_check(buffer_f_in); - } - while (total < N) { - if (read_buffer_len > 0 && buffer_f_in == fd) { - ret = MIN(read_buffer_len,N-total); - memcpy(buffer+total,read_buffer_p,ret); - read_buffer_p += ret; - read_buffer_len -= ret; - total += ret; - continue; - } - - no_flush_read++; io_flush(); - no_flush_read--; ret = read_unbuffered(fd,buffer + total,N-total); total += ret; @@ -327,7 +311,6 @@ 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=0; no_flush++; @@ -335,33 +318,34 @@ 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; - - if (!no_flush_read) { - reading = (buffer_f_in != -1); - } + fd_count = fd; - if (reading) { - FD_SET(buffer_f_in,&r_fds); - if (buffer_f_in > fd) - fd_count = buffer_f_in+1; + 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, + io_error_fd != -1?&r_fds:NULL, &w_fds,NULL, &tv); if (count <= 0) { + if (errno == EBADF) { + exit_cleanup(RERR_SOCKETIO); + } check_timeout(); continue; } - if (reading && FD_ISSET(buffer_f_in, &r_fds)) { - 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)) { @@ -433,7 +417,7 @@ static void writefd(int fd,char *buf,int len) { stats.total_written += len; - if (!io_buffer) { + if (!io_buffer || fd != multiplex_out_fd) { writefd_unbuffered(fd, buf, len); return; } @@ -551,15 +535,10 @@ void io_start_multiplex_in(int fd) { multiplex_in_fd = fd; io_flush(); - if (read_buffer_len) { - fprintf(stderr,"ERROR: data in read buffer at mplx start\n"); - 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; @@ -575,6 +554,20 @@ int io_multiplex_write(int f, char *buf, int len) 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;