X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/22d6234e04e087b9111243d1f940db7c6d59fd76..b35d0d8e9ae9c5407c9f781b545f8a66b9caa9d0:/io.c diff --git a/io.c b/io.c index b48ae0f9..ed6a70c3 100644 --- a/io.c +++ b/io.c @@ -1,6 +1,8 @@ -/* - Copyright (C) Andrew Tridgell 1996 +/* -*- c-file-style: "linux" -*- + + Copyright (C) 1996-2001 by Andrew Tridgell Copyright (C) Paul Mackerras 1996 + Copyright (C) 2001, 2002 by Martin Pool This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -17,169 +19,330 @@ Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. */ -/* - Utilities used in rsync +/** + * @file io.c + * + * Socket and pipe IO utilities used in rsync. + * + * rsync provides its own multiplexing system, which is used to send + * stderr and stdout over a single socket. We need this because + * stdout normally carries the binary data stream, and stderr all our + * error messages. + * + * For historical reasons this is off during the start of the + * connection, but it's switched on quite early using + * io_start_multiplex_out() and io_start_multiplex_in(). + **/ - tridge, June 1996 - */ #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 no_flush; +extern int bwlimit; extern int verbose; -extern int sparse_files; extern int io_timeout; +extern struct stats stats; + + +/** Ignore EOF errors while reading a module listing if the remote + version is 24 or less. */ +int kludge_around_eof = False; + + +static int io_error_fd = -1; + +static void read_loop(int fd, char *buf, size_t len); + +static void check_timeout(void) +{ + extern int am_server, am_daemon; + time_t t; + + err_list_push(); + + if (!io_timeout) return; + + if (!last_io) { + last_io = time(NULL); + return; + } + + t = time(NULL); + + if (last_io && io_timeout && (t-last_io) >= io_timeout) { + if (!am_server && !am_daemon) { + rprintf(FERROR,"io timeout after %d seconds - exiting\n", + (int)(t-last_io)); + } + exit_cleanup(RERR_TIMEOUT); + } +} -int64 write_total(void) +/* setup the fd used to propogate errors */ +void io_set_error_fd(int fd) { - return total_written; + io_error_fd = fd; } -int64 read_total(void) +/* read some data from the error fd and write it to the write log code */ +static void read_error_fd(void) { - return total_read; + char buf[200]; + size_t n; + int fd = io_error_fd; + int tag, len; + + /* io_error_fd is temporarily disabled -- is this meant to + * prevent indefinite recursion? */ + io_error_fd = -1; + + read_loop(fd, buf, 4); + tag = IVAL(buf, 0); + + len = tag & 0xFFFFFF; + tag = tag >> 24; + tag -= MPLEX_BASE; + + while (len) { + n = len; + if (n > (sizeof(buf)-1)) + n = sizeof(buf)-1; + read_loop(fd, buf, n); + rwrite((enum logcode)tag, buf, n); + len -= n; + } + + io_error_fd = fd; } -static int buffer_f_in = -1; -void setup_nonblocking(int f_in,int f_out) +static void whine_about_eof (void) { - set_blocking(f_out,0); - buffer_f_in = f_in; + /** + It's almost always an error to get an EOF when we're trying + to read from the network, because the protocol is + self-terminating. + + However, there is one unfortunate cases where it is not, + which is rsync <2.4.6 sending a list of modules on a + server, since the list is terminated by closing the socket. + So, for the section of the program where that is a problem + (start_socket_client), kludge_around_eof is True and we + just exit. + */ + + if (kludge_around_eof) + exit_cleanup (0); + else { + rprintf (FERROR, + "%s: connection unexpectedly closed " + "(%.0f bytes read so far)\n", + RSYNC_NAME, (double)stats.total_read); + + exit_cleanup (RERR_STREAMIO); + } } -static char *read_buffer; -static char *read_buffer_p; -static int read_buffer_len; -static int read_buffer_size; +static void die_from_readerr (int err) +{ + /* this prevents us trying to write errors on a dead socket */ + io_multiplexing_close(); + + rprintf(FERROR, "%s: read error: %s\n", + RSYNC_NAME, strerror (err)); + exit_cleanup(RERR_STREAMIO); +} -/* 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) +/*! + * 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. + * + * TODO: If the remote shell connection fails, then current versions + * actually report an "unexpected EOF" error here. Since it's a + * fairly common mistake to try to use rsh when ssh is required, we + * should trap that: if we fail to read any data at all, we should + * give a better explanation. We can tell whether the connection has + * started by looking e.g. at whether the remote version is known yet. + */ +static int read_timeout (int fd, char *buf, size_t len) { - int n; + int n, ret=0; + + io_flush(); + + while (ret == 0) { + /* until we manage to read *something* */ + fd_set fds; + struct timeval tv; + int fd_count = fd+1; + int count; + + 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; + } - if (f == -1) return; + tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT; + tv.tv_usec = 0; - if (read_buffer_len == 0) { - read_buffer_p = read_buffer; - } + errno = 0; - if ((n=num_waiting(f)) <= 0) - return; + count = select(fd_count, &fds, NULL, NULL, &tv); + + if (count == 0) { + check_timeout(); + } + + if (count <= 0) { + if (errno == EBADF) { + exit_cleanup(RERR_SOCKETIO); + } + continue; + } - /* things could deteriorate if we read in really small chunks */ - if (n < 10) n = 1024; + if (io_error_fd != -1 && FD_ISSET(io_error_fd, &fds)) { + read_error_fd(); + } - if (n > MAX_READ_BUFFER/4) - n = MAX_READ_BUFFER/4; + if (!FD_ISSET(fd, &fds)) continue; - if (read_buffer_p != read_buffer) { - memmove(read_buffer,read_buffer_p,read_buffer_len); - read_buffer_p = read_buffer; - } + n = read(fd, buf, len); - 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); - if (!read_buffer) out_of_memory("read check"); - read_buffer_p = read_buffer; + if (n > 0) { + buf += n; + len -= n; + ret += n; + if (io_timeout) + last_io = time(NULL); + continue; + } else if (n == 0) { + whine_about_eof (); + return -1; /* doesn't return */ + } else if (n == -1) { + if (errno == EINTR || errno == EWOULDBLOCK || + errno == EAGAIN) + continue; + else + die_from_readerr (errno); + } } - n = read(f,read_buffer+read_buffer_len,n); - if (n > 0) { - read_buffer_len += n; - } + return ret; } -static time_t last_io; -static void check_timeout(void) + +/*! Continue trying to read len bytes - don't return until len has + been read. */ +static void read_loop (int fd, char *buf, size_t len) { - time_t t; - - if (!io_timeout) return; + while (len) { + int n = read_timeout(fd, buf, len); - if (!last_io) { - last_io = time(NULL); - return; + buf += n; + len -= n; } +} - t = time(NULL); - if (last_io && io_timeout && (t-last_io)>io_timeout) { - rprintf(FERROR,"read timeout after %d second - exiting\n", - (int)(t-last_io)); - exit_cleanup(1); +/** + * Read from the file descriptor handling multiplexing - return number + * of bytes read. + * + * Never returns <= 0. + */ +static int read_unbuffered(int fd, char *buf, size_t len) +{ + static size_t remaining; + int tag, ret = 0; + char line[1024]; + + if (!io_multiplexing_in || fd != multiplex_in_fd) + return read_timeout(fd, buf, len); + + while (ret == 0) { + if (remaining) { + len = MIN(len, remaining); + read_loop(fd, buf, len); + remaining -= len; + ret = len; + continue; + } + + read_loop(fd, line, 4); + tag = IVAL(line, 0); + + remaining = tag & 0xFFFFFF; + tag = tag >> 24; + + if (tag == MPLEX_BASE) + continue; + + tag -= MPLEX_BASE; + + if (tag != FERROR && tag != FINFO) { + rprintf(FERROR, "unexpected tag %d\n", tag); + exit_cleanup(RERR_STREAMIO); + } + + if (remaining > sizeof(line) - 1) { + rprintf(FERROR, "multiplexing overflow %d\n\n", + remaining); + exit_cleanup(RERR_STREAMIO); + } + + read_loop(fd, line, remaining); + line[remaining] = 0; + + rprintf((enum logcode) tag, "%s", line); + remaining = 0; } + + return ret; } -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, size_t N) { int ret; - int total=0; - struct timeval tv; - - if (read_buffer_len < N) - read_check(buffer_f_in); + size_t total=0; 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; - } - - while ((ret = read(fd,buffer + total,N-total)) == -1) { - fd_set fds; + io_flush(); - 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) 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) @@ -189,20 +352,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) 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 @@ -210,209 +369,200 @@ int64 read_longint(int f) return ret; } -void read_buf(int f,char *buf,int len) +void read_buf(int f,char *buf,size_t len) { - int ret; - if ((ret=readfd(f,buf,len)) != len) { - if (verbose > 1) - rprintf(FERROR,"(%d) 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) +void read_sbuf(int f,char *buf,size_t len) { - read_buf(f,buf,len); + read_buf (f,buf,len); buf[len] = 0; } 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; } +/* Write len bytes to fd. This underlies the multiplexing system, + * which is always called by application code. */ +static void writefd_unbuffered(int fd,char *buf,size_t len) +{ + size_t total = 0; + fd_set w_fds, r_fds; + int fd_count, count; + struct timeval tv; -static char last_byte; -static int last_sparse; + err_list_push(); -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; -} + no_flush++; + while (total < len) { + FD_ZERO(&w_fds); + FD_ZERO(&r_fds); + FD_SET(fd,&w_fds); + fd_count = fd; -static int write_sparse(int f,char *buf,int len) -{ - int l1=0,l2=0; - int ret; + if (io_error_fd != -1) { + FD_SET(io_error_fd,&r_fds); + if (io_error_fd > fd_count) + fd_count = io_error_fd; + } - for (l1=0;l1 0) - last_sparse=1; + count = select(fd_count+1, + io_error_fd != -1?&r_fds:NULL, + &w_fds,NULL, + &tv); - if (l1 > 0) - do_lseek(f,l1,SEEK_CUR); + if (count == 0) { + check_timeout(); + } - if (l1 == len) - return len; + if (count <= 0) { + if (errno == EBADF) { + exit_cleanup(RERR_SOCKETIO); + } + continue; + } - if ((ret=write(f,buf+l1,len-(l1+l2))) != len-(l1+l2)) { - if (ret == -1 || ret == 0) return ret; - return (l1+ret); - } + if (io_error_fd != -1 && FD_ISSET(io_error_fd, &r_fds)) { + read_error_fd(); + } - if (l2 > 0) - do_lseek(f,l2,SEEK_CUR); - - return len; -} + if (FD_ISSET(fd, &w_fds)) { + int ret; + size_t n = len-total; + ret = write(fd,buf+total,n); + if (ret == -1 && errno == EINTR) { + continue; + } + if (ret == -1 && + (errno == EWOULDBLOCK || errno == EAGAIN)) { + msleep(1); + continue; + } -int write_file(int f,char *buf,int len) -{ - int ret = 0; + if (ret <= 0) { + /* Don't try to write errors back + * across the stream */ + io_multiplexing_close(); + rprintf(FERROR, RSYNC_NAME + ": error writing %d unbuffered bytes" + " - exiting: %s\n", len, + strerror(errno)); + exit_cleanup(RERR_STREAMIO); + } - if (!sparse_files) - return write(f,buf,len); + /* Sleep after writing to limit I/O bandwidth */ + if (bwlimit) + { + tv.tv_sec = 0; + tv.tv_usec = ret * 1000 / bwlimit; + while (tv.tv_usec > 1000000) + { + tv.tv_sec++; + tv.tv_usec -= 1000000; + } + select(0, NULL, NULL, NULL, &tv); + } + + total += ret; - 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; + if (io_timeout) + last_io = time(NULL); } - len -= r1; - buf += r1; - ret += r1; } - return ret; -} - -static int writefd_unbuffered(int fd,char *buf,int len) -{ - int total = 0; - fd_set w_fds, r_fds; - int fd_count, count, got_select=0; - struct timeval tv; - - if (buffer_f_in == -1) - return write(fd,buf,len); - - 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); -#if 0 - rprintf(FERROR,"write exception\n"); - exit_cleanup(1); -#endif - } - - got_select = 0; - - - if (ret == -1) { - if (read_buffer_len < MAX_READ_BUFFER) - read_check(buffer_f_in); - - fd_count = fd+1; - FD_ZERO(&w_fds); - FD_ZERO(&r_fds); - FD_SET(fd,&w_fds); - if (buffer_f_in != -1) { - 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_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) { - check_timeout(); - continue; - } - - if (FD_ISSET(fd, &w_fds)) { - got_select = 1; - } - } else { - total += ret; - } - } - - if (io_timeout) - last_io = time(NULL); - - return total; + no_flush--; } + static char *io_buffer; static int io_buffer_count; void io_start_buffering(int fd) { + if (io_buffer) return; + multiplex_out_fd = fd; io_buffer = (char *)malloc(IO_BUFFER_SIZE); if (!io_buffer) out_of_memory("writefd"); io_buffer_count = 0; } -void io_end_buffering(int fd) +/* write an message to a multiplexed stream. If this fails then rsync + exits */ +static void mplex_write(int fd, enum logcode code, char *buf, size_t len) { - if (io_buffer_count) { - if (writefd_unbuffered(fd, io_buffer, - io_buffer_count) != - io_buffer_count) { - rprintf(FERROR,"write failed\n"); - exit_cleanup(1); - } - io_buffer_count = 0; + char buffer[4096]; + size_t n = len; + + SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len); + + if (n > (sizeof(buffer)-4)) { + n = sizeof(buffer)-4; + } + + memcpy(&buffer[4], buf, n); + writefd_unbuffered(fd, buffer, n+4); + + len -= n; + buf += n; + + if (len) { + writefd_unbuffered(fd, buf, len); + } +} + + +void io_flush(void) +{ + int fd = multiplex_out_fd; + + err_list_push(); + + if (!io_buffer_count || no_flush) return; + + if (io_multiplexing_out) { + mplex_write(fd, FNONE, io_buffer, io_buffer_count); + } else { + writefd_unbuffered(fd, io_buffer, io_buffer_count); + } + io_buffer_count = 0; +} + + +void io_end_buffering(void) +{ + io_flush(); + if (!io_multiplexing_out) { + free(io_buffer); + io_buffer = NULL; } - free(io_buffer); - io_buffer = NULL; } -static int writefd(int fd,char *buf,int len1) +static void writefd(int fd,char *buf,size_t len) { - int len = len1; + stats.total_written += len; - if (!io_buffer) return writefd_unbuffered(fd, buf, len); + err_list_push(); + + if (!io_buffer || fd != multiplex_out_fd) { + writefd_unbuffered(fd, buf, len); + return; + } while (len) { - int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count); + int n = MIN((int) len, IO_BUFFER_SIZE-io_buffer_count); if (n > 0) { memcpy(io_buffer+io_buffer_count, buf, n); buf += n; @@ -420,69 +570,47 @@ static int writefd(int fd,char *buf,int len1) io_buffer_count += n; } - if (io_buffer_count == IO_BUFFER_SIZE) { - if (writefd_unbuffered(fd, io_buffer, - io_buffer_count) != - io_buffer_count) { - return -1; - } - io_buffer_count = 0; - } + 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; + char b[4]; + SIVAL(b,0,x); + writefd(f,b,4); } + +/* + * Note: int64 may actually be a 32-bit type if ./configure couldn't find any + * 64-bit types on this platform. + */ 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) +void write_buf(int f,char *buf,size_t 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)); } @@ -493,15 +621,21 @@ 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) +/** + * Read a line of up to @p maxlen characters into @p buf. Does not + * contain a trailing newline or carriage return. + * + * @return 1 for success; 0 for io error or truncation. + **/ +int read_line(int f, char *buf, size_t maxlen) { while (maxlen) { + buf[0] = 0; read_buf(f, buf, 1); + if (buf[0] == 0) + return 0; if (buf[0] == '\n') { buf[0] = 0; break; @@ -515,6 +649,7 @@ int read_line(int f, char *buf, int maxlen) *buf = 0; return 0; } + return 1; } @@ -526,10 +661,46 @@ void io_printf(int fd, const char *format, ...) int len; va_start(ap, format); - len = vslprintf(buf, sizeof(buf)-1, format, ap); + len = vsnprintf(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); } + + +/* setup for multiplexing an error stream with the data stream */ +void io_start_multiplex_out(int fd) +{ + multiplex_out_fd = fd; + io_flush(); + io_start_buffering(fd); + io_multiplexing_out = 1; +} + +/* setup for multiplexing an error stream with the data stream */ +void io_start_multiplex_in(int fd) +{ + multiplex_in_fd = fd; + io_flush(); + io_multiplexing_in = 1; +} + +/* write an message to the multiplexed error stream */ +int io_multiplex_write(enum logcode code, char *buf, size_t len) +{ + if (!io_multiplexing_out) return 0; + + io_flush(); + stats.total_written += (len+4); + mplex_write(multiplex_out_fd, code, buf, len); + return 1; +} + +/* stop output multiplexing */ +void io_multiplexing_close(void) +{ + io_multiplexing_out = 0; +} +