X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/08f15335b5b469b29cbdacd0fc5ba48d76acf280..c3563c46ed40c3021b61f5ed228c9eca2d64e1f3:/io.c diff --git a/io.c b/io.c index 6f3854e3..39996885 100644 --- a/io.c +++ b/io.c @@ -1,5 +1,6 @@ -/* - Copyright (C) Andrew Tridgell 1996 +/* -*- c-file-style: "linux" -*- + + Copyright (C) 1996-2000 by Andrew Tridgell Copyright (C) Paul Mackerras 1996 This program is free software; you can redistribute it and/or modify @@ -27,6 +28,8 @@ /* if no timeout is specified then use a 60 second select timeout */ #define SELECT_TIMEOUT 60 +extern int bwlimit; + static int io_multiplexing_out; static int io_multiplexing_in; static int multiplex_in_fd; @@ -37,16 +40,13 @@ extern int verbose; 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) -{ - buffer_f_in = f_in; -} +static void read_loop(int fd, char *buf, int len); static void check_timeout(void) { + extern int am_server, am_daemon; time_t t; if (!io_timeout) return; @@ -59,8 +59,10 @@ static void check_timeout(void) t = time(NULL); if (last_io && io_timeout && (t-last_io) >= io_timeout) { - rprintf(FERROR,"io timeout after %d second - exiting\n", - (int)(t-last_io)); + if (!am_server && !am_daemon) { + rprintf(FERROR,"io timeout after %d second - exiting\n", + (int)(t-last_io)); + } exit_cleanup(RERR_TIMEOUT); } } @@ -71,17 +73,29 @@ 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; @@ -90,9 +104,15 @@ static void read_error_fd(void) static int no_flush; -/* 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 */ +/* + * 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. + */ static int read_timeout(int fd, char *buf, int len) { int n, ret=0; @@ -145,6 +165,11 @@ static int read_timeout(int fd, char *buf, int len) continue; } + if (n == -1 && + (errno == EWOULDBLOCK || errno == EAGAIN)) { + continue; + } + if (n == 0) { if (eof_error) { @@ -181,7 +206,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]; @@ -197,8 +221,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; @@ -221,7 +245,7 @@ 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; } @@ -300,11 +324,7 @@ unsigned char read_byte(int f) return c; } - - -/* 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 */ +/* write len bytes to fd */ static void writefd_unbuffered(int fd,char *buf,int len) { int total = 0; @@ -351,19 +371,36 @@ 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); + ret = write(fd,buf+total,n); if (ret == -1 && errno == EINTR) { continue; } + if (ret == -1 && + (errno == EWOULDBLOCK || errno == EAGAIN)) { + msleep(1); + continue; + } + if (ret <= 0) { rprintf(FERROR,"erroring writing %d bytes - exiting\n", len); exit_cleanup(RERR_STREAMIO); } + /* 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; if (io_timeout) @@ -382,37 +419,71 @@ 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); + + len -= n; + buf += n; - /* leave room for the multiplex header in case it's needed */ - io_buffer += 4; + 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); } io_buffer_count = 0; } + +/* XXX: fd is ignored, which seems a little strange. */ void io_end_buffering(int fd) { io_flush(); if (!io_multiplexing_out) { - free(io_buffer-4); + free(io_buffer); io_buffer = NULL; } } +/* some OSes have a bug where an exit causes the pending writes on + a socket to be flushed. Do an explicit shutdown to try to prevent this */ +void io_shutdown(void) +{ + if (multiplex_out_fd != -1) close(multiplex_out_fd); + if (io_error_fd != -1) close(io_error_fd); + multiplex_out_fd = -1; + io_error_fd = -1; +} + + static void writefd(int fd,char *buf,int len) { stats.total_written += len; @@ -443,6 +514,11 @@ void write_int(int f,int32 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; @@ -539,26 +615,21 @@ void io_start_multiplex_in(int fd) } /* 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; } @@ -568,7 +639,3 @@ void io_multiplexing_close(void) io_multiplexing_out = 0; } -void io_close_input(int fd) -{ - buffer_f_in = -1; -}