X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/d8a7290f8621b72139461ed6606cde31fa4d544f..9270e88d76d3ea9e94d48977fbea57b997545577:/io.c diff --git a/io.c b/io.c index 3ece9c3b..b5fd3776 100644 --- a/io.c +++ b/io.c @@ -76,6 +76,9 @@ int kluge_around_eof = 0; int sock_f_in = -1; int sock_f_out = -1; +int64 total_data_read = 0; +int64 total_data_written = 0; + static struct { xbuf in, out, msg; int in_fd; @@ -94,7 +97,9 @@ static int write_batch_monitor_in = -1; static int write_batch_monitor_out = -1; static int ff_forward_fd = -1; -static char ff_lastchar; +static int ff_reenable_multiplex = -1; +static char ff_lastchar = '\0'; +static xbuf ff_xb = EMPTY_XBUF; #ifdef ICONV_OPTION static xbuf iconv_buf = EMPTY_XBUF; #endif @@ -338,20 +343,17 @@ static void safe_write(int fd, const char *buf, size_t len) * a chunk of data and put it into the output buffer. */ static void forward_filesfrom_data(void) { - char buf[FILESFROM_BUFLEN]; int len; - xbuf x; - - INIT_CONST_XBUF(x, buf); - len = read(ff_forward_fd, x.buf, x.size); + len = read(ff_forward_fd, ff_xb.buf + ff_xb.len, ff_xb.size - ff_xb.len); if (len <= 0) { if (len == 0 || errno != EINTR) { /* Send end-of-file marker */ - write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1); ff_forward_fd = -1; - if (protocol_version < 31) - io_start_multiplex_out(iobuf.out_fd); + write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1); + free_xbuf(&ff_xb); + if (ff_reenable_multiplex >= 0) + io_start_multiplex_out(ff_reenable_multiplex); } return; } @@ -359,47 +361,84 @@ static void forward_filesfrom_data(void) if (DEBUG_GTE(IO, 2)) rprintf(FINFO, "[%s] files-from read=%ld\n", who_am_i(), (long)len); +#ifdef ICONV_OPTION + len += ff_xb.len; +#endif + if (!eol_nulls) { - char *s = x.buf + len; + char *s = ff_xb.buf + len; /* Transform CR and/or LF into '\0' */ - while (s-- > x.buf) { + while (s-- > ff_xb.buf) { if (*s == '\n' || *s == '\r') *s = '\0'; } } + if (ff_lastchar) - x.pos = 0; + ff_xb.pos = 0; else { - char *s = x.buf; + char *s = ff_xb.buf; /* Last buf ended with a '\0', so don't let this buf start with one. */ while (len && *s == '\0') s++, len--; - x.pos = s - x.buf; + ff_xb.pos = s - ff_xb.buf; } + +#ifdef ICONV_OPTION + if (filesfrom_convert && len) { + char *sob = ff_xb.buf + ff_xb.pos, *s = sob; + char *eob = sob + len; + int flags = ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT; + if (ff_lastchar == '\0') + flags |= ICB_INIT; + /* Convert/send each null-terminated string separately, skipping empties. */ + while (s != eob) { + if (*s++ == '\0') { + ff_xb.len = s - sob - 1; + if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0) + exit_cleanup(RERR_PROTOCOL); /* impossible? */ + write_buf(iobuf.out_fd, s-1, 1); /* Send the '\0'. */ + while (s != eob && *s == '\0') + s++; + sob = s; + ff_xb.pos = sob - ff_xb.buf; + flags |= ICB_INIT; + } + } + + if ((ff_xb.len = s - sob) == 0) + ff_lastchar = '\0'; + else { + /* Handle a partial string specially, saving any incomplete chars. */ + flags &= ~ICB_INCLUDE_INCOMPLETE; + if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0) { + if (errno == E2BIG) + exit_cleanup(RERR_PROTOCOL); /* impossible? */ + if (ff_xb.pos) + memmove(ff_xb.buf, ff_xb.buf + ff_xb.pos, ff_xb.len); + } + ff_lastchar = 'x'; /* Anything non-zero. */ + } + } else +#endif + if (len) { - char *f = x.buf + x.pos; - char *t = x.buf; + char *f = ff_xb.buf + ff_xb.pos; + char *t = ff_xb.buf; char *eob = f + len; /* Eliminate any multi-'\0' runs. */ while (f != eob) { if (!(*t++ = *f++)) { - while (f != eob && !*f) - f++, len--; + while (f != eob && *f == '\0') + f++; } } ff_lastchar = f[-1]; - } -#ifdef ICONV_OPTION - if (filesfrom_convert) { - /* TODO would it help to translate each string between nulls separately? */ - x.len = len; - iconvbufs(ic_send, &x, &iobuf.out, ICB_INCLUDE_BAD|ICB_INCLUDE_INCOMPLETE|ICB_CIRCULAR_OUT|ICB_INIT); - } else -#endif - if (len) { - /* This will not circle back to perform_io() because we only get - * called when there is plenty of room in the output buffer. */ - write_buf(iobuf.out_fd, x.buf, len); + if ((len = t - ff_xb.buf) != 0) { + /* This will not circle back to perform_io() because we only get + * called when there is plenty of room in the output buffer. */ + write_buf(iobuf.out_fd, ff_xb.buf, len); + } } } @@ -713,7 +752,7 @@ static char *perform_io(size_t needed, int flags) /* Don't write errors on a dead socket. */ msgs2stderr = 1; out->len = iobuf.raw_flushing_ends_before = out->pos = 0; - rsyserr(FERROR_SOCKET, errno, "write error"); + rsyserr(FERROR_SOCKET, errno, "[%s] write error", who_am_i()); exit_cleanup(RERR_STREAMIO); } } @@ -763,6 +802,17 @@ static char *perform_io(size_t needed, int flags) return data; } +void noop_io_until_death(void) +{ + char buf[1024]; + + kluge_around_eof = 1; + set_io_timeout(10); + + while (1) + read_buf(iobuf.in_fd, buf, sizeof buf); +} + /* Buffer a message for the multiplexed output stream. Is never used for MSG_DATA. */ int send_msg(enum msgcode code, const char *buf, size_t len, int convert) { @@ -981,18 +1031,16 @@ int get_hlink_num(void) * for recv_file_list() to use. */ void start_filesfrom_forwarding(int fd) { - ff_forward_fd = fd; - if (protocol_version < 31) { - int save_fd = iobuf.out_fd; - /* Older protocols send the files-from data w/o packaging it in - * multiplexed I/O packets. To match this, we temporarily turn - * off the multiplexing of our output w/o disabling buffering. */ - assert(OUT_MULTIPLEXED); - /* Be extra, extra sure no messages go out before files-from data. */ - iobuf.msg.pos = iobuf.msg.len = 0; - io_end_multiplex_out(False); - iobuf.out_fd = save_fd; + if (protocol_version < 31 && OUT_MULTIPLEXED) { + /* Older protocols send the files-from data w/o packaging + * it in multiplexed I/O packets, so temporarily switch + * to buffered I/O to match this behavior. */ + iobuf.msg.pos = iobuf.msg.len = 0; /* Be extra sure no messages go out. */ + ff_reenable_multiplex = io_end_multiplex_out(MPLX_TO_BUFFERED); } + ff_forward_fd = fd; + + alloc_xbuf(&ff_xb, FILESFROM_BUFLEN); } /* Read a line into the "buf" buffer. */ @@ -1101,7 +1149,7 @@ void read_args(int f_in, char *mod_name, char *buf, size_t bufsiz, int rl_nulls, *argv_p = argv; } -int io_start_buffering_out(int f_out) +BOOL io_start_buffering_out(int f_out) { if (msgs2stderr && DEBUG_GTE(IO, 2)) rprintf(FINFO, "[%s] io_start_buffering_out(%d)\n", who_am_i(), f_out); @@ -1117,7 +1165,7 @@ int io_start_buffering_out(int f_out) if (iobuf.out_fd == -1) iobuf.out_fd = f_out; assert(f_out == iobuf.out_fd); - return 0; + return False; } iobuf.out.size = IO_BUFFER_SIZE * 2 - 4; @@ -1127,10 +1175,10 @@ int io_start_buffering_out(int f_out) iobuf.out.pos = iobuf.out.len = 0; iobuf.out_fd = f_out; - return 1; + return True; } -int io_start_buffering_in(int f_in) +BOOL io_start_buffering_in(int f_in) { if (msgs2stderr && DEBUG_GTE(IO, 2)) rprintf(FINFO, "[%s] io_start_buffering_in(%d)\n", who_am_i(), f_in); @@ -1139,7 +1187,7 @@ int io_start_buffering_in(int f_in) if (iobuf.in_fd == -1) iobuf.in_fd = f_in; assert(f_in == iobuf.in_fd); - return 0; + return False; } iobuf.in.size = IO_BUFFER_SIZE; @@ -1150,21 +1198,14 @@ int io_start_buffering_in(int f_in) iobuf.in_fd = f_in; - return 1; -} - -static void free_xbuf(xbuf *x) -{ - if (x->buf) - free(x->buf); - memset(x, 0, sizeof (xbuf)); + return True; } void io_end_buffering_in(BOOL free_buffers) { - if (DEBUG_GTE(IO, 2)) { - rprintf(FINFO, "[%s] io_end_buffering_in(%s)\n", - who_am_i(), free_buffers ? "True" : "False"); + if (msgs2stderr && DEBUG_GTE(IO, 2)) { + rprintf(FINFO, "[%s] io_end_buffering_in(IOBUF_%s_BUFS)\n", + who_am_i(), free_buffers ? "FREE" : "KEEP"); } if (free_buffers) @@ -1177,9 +1218,9 @@ void io_end_buffering_in(BOOL free_buffers) void io_end_buffering_out(BOOL free_buffers) { - if (DEBUG_GTE(IO, 2)) { - rprintf(FINFO, "[%s] io_end_buffering_out(%s)\n", - who_am_i(), free_buffers ? "True" : "False"); + if (msgs2stderr && DEBUG_GTE(IO, 2)) { + rprintf(FINFO, "[%s] io_end_buffering_out(IOBUF_%s_BUFS)\n", + who_am_i(), free_buffers ? "FREE" : "KEEP"); } io_flush(FULL_FLUSH); @@ -1187,9 +1228,6 @@ void io_end_buffering_out(BOOL free_buffers) if (free_buffers) { free_xbuf(&iobuf.out); free_xbuf(&iobuf.msg); - } else { - iobuf.out.pos = iobuf.out.len = 0; - iobuf.msg.pos = iobuf.msg.len = 0; } iobuf.out_fd = -1; @@ -1237,7 +1275,7 @@ void stop_flist_forward(void) static void read_a_msg(void) { char *data, line[BIGPATHBUFLEN]; - int tag; + int tag, val; size_t msg_bytes; data = perform_io(4, PIO_INPUT_AND_CONSUME); @@ -1279,6 +1317,17 @@ static void read_a_msg(void) if (!am_generator) send_msg(MSG_IO_ERROR, data, 4, 0); break; + case MSG_IO_TIMEOUT: + if (msg_bytes != 4 || am_server || am_generator) + goto invalid_msg; + data = perform_io(4, PIO_INPUT_AND_CONSUME); + val = IVAL(data, 0); + if (!io_timeout || io_timeout > val) { + if (INFO_GTE(MISC, 2)) + rprintf(FINFO, "Setting --timeout=%d to match server\n", val); + set_io_timeout(val); + } + break; case MSG_NOOP: if (am_sender) maybe_send_keepalive(); @@ -1296,20 +1345,26 @@ static void read_a_msg(void) xbuf outbuf, inbuf; char ibuf[512]; int add_null = 0; + int flags = ICB_INCLUDE_BAD | ICB_INIT; INIT_CONST_XBUF(outbuf, line); INIT_XBUF(inbuf, ibuf, 0, (size_t)-1); while (msg_bytes) { + size_t len = msg_bytes > sizeof ibuf - inbuf.len ? sizeof ibuf - inbuf.len : msg_bytes; + memcpy(ibuf + inbuf.len, perform_io(len, PIO_INPUT_AND_CONSUME), len); inbuf.pos = 0; - inbuf.len = msg_bytes > sizeof ibuf ? sizeof ibuf : msg_bytes; - memcpy(inbuf.buf, perform_io(inbuf.len, PIO_INPUT_AND_CONSUME), inbuf.len); - if (!(msg_bytes -= inbuf.len) - && !ibuf[inbuf.len-1]) + inbuf.len += len; + if (!(msg_bytes -= len) && !ibuf[inbuf.len-1]) inbuf.len--, add_null = 1; - if (iconvbufs(ic_send, &inbuf, &outbuf, - ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_INIT) < 0) - goto overflow; + if (iconvbufs(ic_send, &inbuf, &outbuf, flags) < 0) { + if (errno == E2BIG) + goto overflow; + /* Buffer ended with an incomplete char, so move the + * bytes to the start of the buffer and continue. */ + memmove(ibuf, ibuf + inbuf.pos, inbuf.len); + } + flags &= ~ICB_INIT; } if (add_null) { if (outbuf.len == outbuf.size) @@ -1382,6 +1437,16 @@ static void read_a_msg(void) first_message = 0; } break; + case MSG_ERROR_EXIT: + if (msg_bytes != 4) + goto invalid_msg; + data = perform_io(4, PIO_INPUT_AND_CONSUME); + val = IVAL(data, 0); + if (am_generator && protocol_version >= 31) + send_msg_int(MSG_ERROR_EXIT, val); + if (am_generator) + val = RERR_RCVR_ERROR; /* avoids duplicate errors */ + exit_cleanup(val); default: rprintf(FERROR, "unexpected tag %d [%s%s]\n", tag, who_am_i(), inc_recurse ? "/inc" : ""); @@ -1548,6 +1613,7 @@ void read_buf(int f, char *buf, size_t len) if (!IN_MULTIPLEXED) { memcpy(buf, perform_io(len, PIO_INPUT_AND_CONSUME), len); + total_data_read += len; if (forward_flist_data) write_buf(iobuf.out_fd, buf, len); batch_copy: @@ -1571,6 +1637,7 @@ void read_buf(int f, char *buf, size_t len) /* The bytes at the "data" pointer will survive long * enough to make a copy, but not past future I/O. */ memcpy(buf, data, siz); + total_data_read += siz; if (forward_flist_data) write_buf(iobuf.out_fd, buf, siz); @@ -1843,6 +1910,7 @@ void write_buf(int f, const char *buf, size_t len) memcpy(iobuf.out.buf + pos, buf, len); iobuf.out.len += len; + total_data_written += len; batch_copy: if (f == write_batch_monitor_out) @@ -2009,11 +2077,11 @@ void io_printf(int fd, const char *format, ...) /* Setup for multiplexing a MSG_* stream with the data stream. */ void io_start_multiplex_out(int fd) { + io_flush(FULL_FLUSH); + if (msgs2stderr && DEBUG_GTE(IO, 2)) rprintf(FINFO, "[%s] io_start_multiplex_out(%d)\n", who_am_i(), fd); - io_flush(FULL_FLUSH); - iobuf.out_empty_len = 4; /* See also OUT_MULTIPLEXED */ io_start_buffering_out(fd); @@ -2031,28 +2099,40 @@ void io_start_multiplex_in(int fd) io_start_buffering_in(fd); } -void io_end_multiplex_in(BOOL free_buffers) +int io_end_multiplex_in(int mode) { - if (DEBUG_GTE(IO, 2)) { - rprintf(FINFO, "[%s] io_end_multiplex_in(%s)\n", - who_am_i(), free_buffers ? "True" : "False"); - } + int ret = iobuf.in_multiplexed ? iobuf.in_fd : -1; + + if (msgs2stderr && DEBUG_GTE(IO, 2)) + rprintf(FINFO, "[%s] io_end_multiplex_in(mode=%d)\n", who_am_i(), mode); iobuf.in_multiplexed = False; - iobuf.raw_input_ends_before = 0; - io_end_buffering_in(free_buffers); + if (mode == MPLX_SWITCHING) + iobuf.raw_input_ends_before = 0; + else + assert(iobuf.raw_input_ends_before == 0); + if (mode != MPLX_TO_BUFFERED) + io_end_buffering_in(mode); + + return ret; } -/* Stop output multiplexing. */ -void io_end_multiplex_out(BOOL free_buffers) +int io_end_multiplex_out(int mode) { - if (DEBUG_GTE(IO, 2)) { - rprintf(FINFO, "[%s] io_end_multiplex_out(%s)\n", - who_am_i(), free_buffers ? "True" : "False"); - } + int ret = iobuf.out_empty_len ? iobuf.out_fd : -1; + + if (msgs2stderr && DEBUG_GTE(IO, 2)) + rprintf(FINFO, "[%s] io_end_multiplex_out(mode=%d)\n", who_am_i(), mode); + + if (mode != MPLX_TO_BUFFERED) + io_end_buffering_out(mode); + else + io_flush(FULL_FLUSH); - io_end_buffering_out(free_buffers); + iobuf.out.len = 0; iobuf.out_empty_len = 0; + + return ret; } void start_write_batch(int fd)