X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/cbce490e138d7bbc7bf982496861f093732dfba9..8b35435f7cc4708400c652434615ce48dc493de4:/io.c diff --git a/io.c b/io.c index 26b6e4b3..6cdbe6bf 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 */ @@ -39,7 +39,8 @@ extern struct stats stats; static int buffer_f_in = -1; static int io_error_fd = -1; -static int in_read_check; + +static void read_loop(int fd, char *buf, int len); void setup_readbuffer(int f_in) { @@ -72,29 +73,36 @@ void io_set_error_fd(int fd) io_error_fd = fd; } -/* read some data from the error fd and write it to FERROR */ +/* read some data from the error fd and write it to the write log code */ static void read_error_fd(void) { char buf[200]; int n; int fd = io_error_fd; + int tag, len; + io_error_fd = -1; - n = read(fd, buf, sizeof(buf)-1); - if (n > 0) { - rwrite(FERROR, buf, n); + 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 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 @@ -103,9 +111,7 @@ 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; @@ -189,7 +195,6 @@ static void read_loop(int fd, char *buf, int len) static int read_unbuffered(int fd, char *buf, int len) { static int remaining; - char ibuf[4]; int tag, ret=0; char line[1024]; @@ -205,8 +210,8 @@ static int read_unbuffered(int fd, char *buf, int len) continue; } - read_loop(fd, ibuf, 4); - tag = IVAL(ibuf, 0); + read_loop(fd, line, 4); + tag = IVAL(line, 0); remaining = tag & 0xFFFFFF; tag = tag >> 24; @@ -229,51 +234,14 @@ static int read_unbuffered(int fd, char *buf, int len) read_loop(fd, line, remaining); line[remaining] = 0; - rprintf(tag,"%s", line); + rprintf((enum logcode)tag,"%s", line); remaining = 0; - - if (in_read_check) break; } return ret; } - -/* 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; - - in_read_check = 1; - - if (f == -1) return; - - if (read_buffer_len == 0) { - read_buffer_p = read_buffer; - } - - 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; - - in_read_check = 0; -} - - /* 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) @@ -281,23 +249,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; @@ -371,7 +324,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++; @@ -381,17 +333,6 @@ static void writefd_unbuffered(int fd,char *buf,int len) FD_SET(fd,&w_fds); fd_count = fd; - if (!no_flush_read) { - reading = (buffer_f_in != -1) && - (read_buffer_len < MAX_READ_BUFFER); - } - - if (reading) { - FD_SET(buffer_f_in,&r_fds); - 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) @@ -404,7 +345,7 @@ static void writefd_unbuffered(int fd,char *buf,int len) errno = 0; count = select(fd_count+1, - (reading || io_error_fd != -1)?&r_fds:NULL, + io_error_fd != -1?&r_fds:NULL, &w_fds,NULL, &tv); @@ -416,10 +357,6 @@ static void writefd_unbuffered(int fd,char *buf,int len) 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(); } @@ -427,8 +364,6 @@ static void writefd_unbuffered(int fd,char *buf,int len) if (FD_ISSET(fd, &w_fds)) { 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) { @@ -458,22 +393,43 @@ void io_start_buffering(int fd) { if (io_buffer) return; multiplex_out_fd = fd; - io_buffer = (char *)malloc(IO_BUFFER_SIZE+4); + io_buffer = (char *)malloc(IO_BUFFER_SIZE); if (!io_buffer) out_of_memory("writefd"); io_buffer_count = 0; +} + +/* write an message to a multiplexed stream. If this fails then rsync + exits */ +static void mplex_write(int fd, enum logcode code, char *buf, int len) +{ + char buffer[4096]; + int 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); - /* leave room for the multiplex header in case it's needed */ - io_buffer += 4; + len -= n; + buf += n; + + if (len) { + writefd_unbuffered(fd, buf, len); + } } + void io_flush(void) { int fd = multiplex_out_fd; if (!io_buffer_count || no_flush) return; if (io_multiplexing_out) { - SIVAL(io_buffer-4, 0, (MPLEX_BASE<<24) + io_buffer_count); - writefd_unbuffered(fd, io_buffer-4, io_buffer_count+4); + mplex_write(fd, FNONE, io_buffer, io_buffer_count); } else { writefd_unbuffered(fd, io_buffer, io_buffer_count); } @@ -484,7 +440,7 @@ void io_end_buffering(int fd) { io_flush(); if (!io_multiplexing_out) { - free(io_buffer-4); + free(io_buffer); io_buffer = NULL; } } @@ -611,34 +567,25 @@ void io_start_multiplex_in(int fd) { multiplex_in_fd = fd; io_flush(); - if (read_buffer_len) { - exit_cleanup(RERR_STREAMIO); - } - io_multiplexing_in = 1; } /* write an message to the multiplexed error stream */ -int io_multiplex_write(int f, char *buf, int len) +int io_multiplex_write(enum logcode code, char *buf, int len) { if (!io_multiplexing_out) return 0; io_flush(); - - 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); + mplex_write(multiplex_out_fd, code, buf, len); return 1; } /* write a message to the special error fd */ -int io_error_write(int f, char *buf, int len) +int io_error_write(int f, enum logcode code, char *buf, int len) { if (f == -1) return 0; - writefd_unbuffered(f, buf, len); + mplex_write(f, code, buf, len); return 1; } @@ -652,3 +599,4 @@ void io_close_input(int fd) { buffer_f_in = -1; } +