X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/20caffd2b361bcad51692998411e4cc566c04b40..0c2e8f93643fd94a8d388e4373c30331d0af75b4:/io.c diff --git a/io.c b/io.c index 7c50d5af..9aab9dce 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 @@ -110,6 +115,17 @@ static char int_byte_extra[64] = { 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 5, 6, /* (C0 - FF)/4 */ }; +/* Our I/O buffers are sized with no bits on in the lowest byte of the "size" + * (indeed, our rounding of sizes in 1024-byte units assures more than this). + * This allows the code that is storing bytes near the physical end of a + * circular buffer to temporarily reduce the buffer's size (in order to make + * some storing idioms easier), while also making it simple to restore the + * buffer's actual size when the buffer's "pos" wraps around to the start (we + * just round the buffer's size up again). */ + +#define IOBUF_WAS_REDUCED(siz) ((siz) & 0xFF) +#define IOBUF_RESTORE_SIZE(siz) (((siz) | 0xFF) + 1) + #define IN_MULTIPLEXED (iobuf.in_multiplexed) #define OUT_MULTIPLEXED (iobuf.out_empty_len != 0) @@ -131,6 +147,7 @@ enum festatus { FES_SUCCESS, FES_REDO, FES_NO_SEND }; static flist_ndx_list redo_list, hlink_list; +static void drain_multiplex_messages(void); static void sleep_for_bwlimit(int bytes_written); static void check_timeout(void) @@ -168,9 +185,9 @@ static void check_timeout(void) * There is another case for older protocol versions (< 24) where the module * listing was not terminated, so we must ignore an EOF error in that case and * exit. In this situation, kluge_around_eof will be > 0. */ -static NORETURN void whine_about_eof(int fd) +static NORETURN void whine_about_eof(BOOL allow_kluge) { - if (kluge_around_eof && fd == sock_f_in) { + if (kluge_around_eof && allow_kluge) { int i; if (kluge_around_eof > 0) exit_cleanup(0); @@ -338,20 +355,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 +373,117 @@ 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 = f; /* Keep any non-zero offset to avoid iconv reset. */ + 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]; + 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); + } } -#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); - } 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); +} + +void reduce_iobuf_size(xbuf *out, size_t new_size) +{ + if (new_size < out->size) { + if (DEBUG_GTE(IO, 4)) { + const char *name = out == &iobuf.out ? "iobuf.out" + : out == &iobuf.msg ? "iobuf.msg" + : NULL; + if (name) { + rprintf(FINFO, "[%s] reduced size of %s (-%d)\n", + who_am_i(), name, (int)(out->size - new_size)); + } + } + out->size = new_size; + } +} + +void restore_iobuf_size(xbuf *out) +{ + if (IOBUF_WAS_REDUCED(out->size)) { + size_t new_size = IOBUF_RESTORE_SIZE(out->size); + if (DEBUG_GTE(IO, 4)) { + const char *name = out == &iobuf.out ? "iobuf.out" + : out == &iobuf.msg ? "iobuf.msg" + : NULL; + if (name) { + rprintf(FINFO, "[%s] restored size of %s (+%d)\n", + who_am_i(), name, (int)(new_size - out->size)); + } + } + out->size = new_size; } } @@ -468,13 +552,12 @@ static char *perform_io(size_t needed, int flags) * Also make sure it will fit in the free space at the end, or * else we need to shift some bytes. */ if (needed && iobuf.in.size < needed) { - if (!(iobuf.in.buf = realloc_array(iobuf.in.buf, char, needed))) - out_of_memory("perform_io"); + size_t new_size = ROUND_UP_1024(needed); if (DEBUG_GTE(IO, 4)) { - rprintf(FINFO, "[%s] resized input buffer from %d to %d bytes.\n", - who_am_i(), iobuf.in.size, needed); + rprintf(FINFO, "[%s] resizing input buffer from %ld to %ld bytes.\n", + who_am_i(), (long)iobuf.in.size, (long)new_size); } - iobuf.in.size = needed; + realloc_xbuf(&iobuf.in, new_size); } if (iobuf.in.size - iobuf.in.pos < needed || (iobuf.in.len < needed && iobuf.in.len < 1024 @@ -482,8 +565,8 @@ static char *perform_io(size_t needed, int flags) memmove(iobuf.in.buf, iobuf.in.buf + iobuf.in.pos, iobuf.in.len); if (DEBUG_GTE(IO, 4)) { rprintf(FINFO, - "[%s] moved %d bytes from %d to 0 in the input buffer (size=%d, needed=%d).\n", - who_am_i(), iobuf.in.len, iobuf.in.pos, iobuf.in.size, needed); + "[%s] moved %ld bytes from %ld to 0 in the input buffer (size=%ld, needed=%ld).\n", + who_am_i(), (long)iobuf.in.len, (long)iobuf.in.pos, (long)iobuf.in.size, (long)needed); } if (iobuf.raw_input_ends_before) iobuf.raw_input_ends_before -= iobuf.in.pos; @@ -502,8 +585,8 @@ static char *perform_io(size_t needed, int flags) if (DEBUG_GTE(IO, 3)) { rprintf(FINFO, "[%s] perform_io(%ld, outroom) needs to flush %ld\n", who_am_i(), (long)needed, - iobuf.out.len > iobuf.out.size - needed - ? (long)iobuf.out.len - (iobuf.out.size - needed) : 0L); + iobuf.out.len + needed > iobuf.out.size + ? (long)(iobuf.out.len + needed - iobuf.out.size) : 0L); } break; @@ -518,8 +601,8 @@ static char *perform_io(size_t needed, int flags) if (DEBUG_GTE(IO, 3)) { rprintf(FINFO, "[%s] perform_io(%ld, msgroom) needs to flush %ld\n", who_am_i(), (long)needed, - iobuf.out.len > iobuf.msg.size - needed - ? (long)iobuf.out.len - (iobuf.msg.size - needed) : 0L); + iobuf.msg.len + needed > iobuf.msg.size + ? (long)(iobuf.msg.len + needed - iobuf.msg.size) : 0L); } break; @@ -539,11 +622,13 @@ static char *perform_io(size_t needed, int flags) goto double_break; break; case PIO_NEED_OUTROOM: - if (iobuf.out.len <= iobuf.out.size - needed) + /* Note that iobuf.out_empty_len doesn't factor into this check + * because iobuf.out.len already holds any needed header len. */ + if (iobuf.out.len + needed <= iobuf.out.size) goto double_break; break; case PIO_NEED_MSGROOM: - if (iobuf.msg.len <= iobuf.msg.size - needed) + if (iobuf.msg.len + needed <= iobuf.msg.size) goto double_break; break; } @@ -573,22 +658,12 @@ static char *perform_io(size_t needed, int flags) if (iobuf.raw_flushing_ends_before || (!iobuf.msg.len && iobuf.out.len > iobuf.out_empty_len && !(flags & PIO_NEED_MSGROOM))) { if (OUT_MULTIPLEXED && !iobuf.raw_flushing_ends_before) { - size_t val; - /* The iobuf.raw_flushing_ends_before value can point off the end * of the iobuf.out buffer for a while, for easier subtracting. */ iobuf.raw_flushing_ends_before = iobuf.out.pos + iobuf.out.len; SIVAL(iobuf.out.buf + iobuf.raw_data_header_pos, 0, ((MPLEX_BASE + (int)MSG_DATA)<<24) + iobuf.out.len - 4); - if ((val = iobuf.out.size - iobuf.raw_data_header_pos) < 4) { - /* We used some of the overflow bytes, so move them. */ - if (DEBUG_GTE(IO, 4)) { - rprintf(FINFO, "[%s] wrap-bytes moved: %d (perform_io)\n", - who_am_i(), (int)val); - } - memcpy(iobuf.out.buf, iobuf.out.buf + iobuf.out.size, 4 - val); - } if (DEBUG_GTE(IO, 1)) { rprintf(FINFO, "[%s] send_msg(%d, %ld)\n", @@ -599,6 +674,14 @@ static char *perform_io(size_t needed, int flags) iobuf.raw_data_header_pos = iobuf.raw_flushing_ends_before; if (iobuf.raw_data_header_pos >= iobuf.out.size) iobuf.raw_data_header_pos -= iobuf.out.size; + else if (iobuf.raw_data_header_pos + 4 > iobuf.out.size) { + /* The 4-byte header won't fit at the end of the buffer, + * so we'll temporarily reduce the output buffer's size + * and put the header at the start of the buffer. */ + reduce_iobuf_size(&iobuf.out, iobuf.raw_data_header_pos); + iobuf.raw_data_header_pos = 0; + } + /* Yes, it is possible for this to make len > size for a while. */ iobuf.out.len += 4; } @@ -620,11 +703,19 @@ static char *perform_io(size_t needed, int flags) if (max_fd < 0) { switch (flags & PIO_NEED_FLAGS) { case PIO_NEED_INPUT: + iobuf.in.len = 0; + if (kluge_around_eof == 2) + exit_cleanup(0); + if (iobuf.in_fd == -2) + whine_about_eof(True); rprintf(FERROR, "error in perform_io: no fd for input.\n"); exit_cleanup(RERR_PROTOCOL); case PIO_NEED_OUTROOM: case PIO_NEED_MSGROOM: msgs2stderr = 1; + drain_multiplex_messages(); + if (iobuf.out_fd == -2) + whine_about_eof(True); rprintf(FERROR, "error in perform_io: no fd for output.\n"); exit_cleanup(RERR_PROTOCOL); default: @@ -635,8 +726,7 @@ static char *perform_io(size_t needed, int flags) } if (extra_flist_sending_enabled) { - if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD - && file_total - file_old_total >= MIN_FILECNT_LOOKAHEAD) + if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD) tv.tv_sec = 0; else { extra_flist_sending_enabled = False; @@ -669,8 +759,9 @@ static char *perform_io(size_t needed, int flags) int n; if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) { if (n == 0) { + /* Signal that input has become invalid. */ if (!read_batch || batch_fd < 0 || am_generator) - whine_about_eof(iobuf.in_fd); /* Doesn't return. */ + iobuf.in_fd = -2; batch_fd = -1; continue; } @@ -684,7 +775,7 @@ static char *perform_io(size_t needed, int flags) rsyserr(FERROR_SOCKET, errno, "read error"); } else rsyserr(FERROR, errno, "read error"); - exit_cleanup(RERR_STREAMIO); + exit_cleanup(RERR_SOCKETIO); } } if (msgs2stderr && DEBUG_GTE(IO, 2)) @@ -712,9 +803,11 @@ static char *perform_io(size_t needed, int flags) else { /* 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"); - exit_cleanup(RERR_STREAMIO); + iobuf.out_fd = -2; + iobuf.out.len = iobuf.msg.len = iobuf.raw_flushing_ends_before = 0; + rsyserr(FERROR_SOCKET, errno, "[%s] write error", who_am_i()); + drain_multiplex_messages(); + exit_cleanup(RERR_SOCKETIO); } } if (msgs2stderr && DEBUG_GTE(IO, 2)) { @@ -733,11 +826,12 @@ static char *perform_io(size_t needed, int flags) if (iobuf.raw_flushing_ends_before) iobuf.raw_flushing_ends_before -= out->size; out->pos = 0; - } - if (out->pos == iobuf.raw_flushing_ends_before) + restore_iobuf_size(out); + } else if (out->pos == iobuf.raw_flushing_ends_before) iobuf.raw_flushing_ends_before = 0; if ((out->len -= n) == empty_buf_len) { out->pos = 0; + restore_iobuf_size(out); if (empty_buf_len) iobuf.raw_data_header_pos = 0; } @@ -763,11 +857,25 @@ static char *perform_io(size_t needed, int flags) return data; } +void noop_io_until_death(void) +{ + char buf[1024]; + + kluge_around_eof = 2; + /* Setting an I/O timeout ensures that if something inexplicably weird + * happens, we won't hang around forever. */ + if (!io_timeout) + set_io_timeout(60); + + 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) { char *hdr; - size_t pos; + size_t needed, pos; BOOL want_debug = DEBUG_GTE(IO, 1) && convert >= 0 && (msgs2stderr || code != MSG_INFO); if (!OUT_MULTIPLEXED) @@ -776,24 +884,35 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert) if (want_debug) rprintf(FINFO, "[%s] send_msg(%d, %ld)\n", who_am_i(), (int)code, (long)len); + /* When checking for enough free space for this message, we need to + * make sure that there is space for the 4-byte header, plus we'll + * assume that we may waste up to 3 bytes (if the header doesn't fit + * at the physical end of the buffer). */ #ifdef ICONV_OPTION if (convert > 0 && ic_send == (iconv_t)-1) convert = 0; if (convert > 0) { - /* Ensuring double-size room leaves space for a potential conversion. */ - if (iobuf.msg.len + len*2 + 4 > iobuf.msg.size) - perform_io(len*2 + 4, PIO_NEED_MSGROOM); + /* Ensuring double-size room leaves space for maximal conversion expansion. */ + needed = len*2 + 4 + 3; } else #endif - if (iobuf.msg.len + len + 4 > iobuf.msg.size) - perform_io(len + 4, PIO_NEED_MSGROOM); + needed = len + 4 + 3; + if (iobuf.msg.len + needed > iobuf.msg.size) + perform_io(needed, PIO_NEED_MSGROOM); pos = iobuf.msg.pos + iobuf.msg.len; /* Must be set after any flushing. */ if (pos >= iobuf.msg.size) pos -= iobuf.msg.size; + else if (pos + 4 > iobuf.msg.size) { + /* The 4-byte header won't fit at the end of the buffer, + * so we'll temporarily reduce the message buffer's size + * and put the header at the start of the buffer. */ + reduce_iobuf_size(&iobuf.msg, pos); + pos = 0; + } hdr = iobuf.msg.buf + pos; - iobuf.msg.len += 4; /* Leave room for the coming header bytes. */ + iobuf.msg.len += 4; /* Allocate room for the coming header bytes. */ #ifdef ICONV_OPTION if (convert > 0) { @@ -803,7 +922,7 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert) len = iobuf.msg.len; iconvbufs(ic_send, &inbuf, &iobuf.msg, - ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT); + ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT | ICB_INIT); if (inbuf.len > 0) { rprintf(FERROR, "overflowed iobuf.msg buffer in send_msg"); exit_cleanup(RERR_UNSUPPORTED); @@ -828,13 +947,6 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert) } SIVAL(hdr, 0, ((MPLEX_BASE + (int)code)<<24) + len); - /* If the header used any overflow bytes, move them to the start. */ - if ((pos = hdr+4 - iobuf.msg.buf) > iobuf.msg.size) { - size_t siz = pos - iobuf.msg.size; - if (DEBUG_GTE(IO, 4)) - rprintf(FINFO, "[%s] wrap-bytes moved: %d (send_msg)\n", who_am_i(), (int)siz); - memcpy(iobuf.msg.buf, hdr+4 - siz, siz); - } if (want_debug && convert > 0) rprintf(FINFO, "[%s] converted msg len=%ld\n", who_am_i(), (long)len); @@ -981,18 +1093,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. */ @@ -1039,7 +1149,7 @@ int read_line(int fd, char *buf, size_t bufsiz, int flags) iconv_buf.pos = 0; iconv_buf.len = s - iconv_buf.buf; iconvbufs(ic_recv, &iconv_buf, &outbuf, - ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE); + ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_INIT); outbuf.buf[outbuf.len] = '\0'; return outbuf.len; } @@ -1101,36 +1211,26 @@ 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); - if (OUT_MULTIPLEXED && !iobuf.msg.buf) { - iobuf.msg.size = IO_BUFFER_SIZE - 4; - if (!(iobuf.msg.buf = new_array(char, iobuf.msg.size + 4))) - out_of_memory("io_start_buffering_out"); - iobuf.msg.pos = iobuf.msg.len = 0; - } - if (iobuf.out.buf) { if (iobuf.out_fd == -1) iobuf.out_fd = f_out; - assert(f_out == iobuf.out_fd); - return 0; + else + assert(f_out == iobuf.out_fd); + return False; } - iobuf.out.size = IO_BUFFER_SIZE * 2 - 4; - /* The 4 overflow bytes makes some circular-buffer wrapping operations easier. */ - if (!(iobuf.out.buf = new_array(char, iobuf.out.size + 4))) - out_of_memory("io_start_buffering_out"); - iobuf.out.pos = iobuf.out.len = 0; + alloc_xbuf(&iobuf.out, ROUND_UP_1024(IO_BUFFER_SIZE * 2)); 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); @@ -1138,33 +1238,22 @@ int io_start_buffering_in(int f_in) if (iobuf.in.buf) { if (iobuf.in_fd == -1) iobuf.in_fd = f_in; - assert(f_in == iobuf.in_fd); - return 0; + else + assert(f_in == iobuf.in_fd); + return False; } - iobuf.in.size = IO_BUFFER_SIZE; - if (!(iobuf.in.buf = new_array(char, iobuf.in.size))) - out_of_memory("io_start_buffering_in"); - - iobuf.in.pos = iobuf.in.len = 0; - + alloc_xbuf(&iobuf.in, ROUND_UP_1024(IO_BUFFER_SIZE)); 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 +1266,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 +1276,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; @@ -1197,7 +1283,8 @@ void io_end_buffering_out(BOOL free_buffers) void maybe_flush_socket(int important) { - if (iobuf.out.buf && iobuf.out.len && (important || time(NULL) - last_io_out >= 5)) + if (flist_eof && iobuf.out.buf && iobuf.out.len > iobuf.out_empty_len + && (important || time(NULL) - last_io_out >= 5)) io_flush(NORMAL_FLUSH); } @@ -1236,7 +1323,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); @@ -1274,9 +1361,21 @@ static void read_a_msg(void) if (msg_bytes != 4 || am_sender) goto invalid_msg; data = perform_io(4, PIO_INPUT_AND_CONSUME); - io_error |= IVAL(data, 0); + val = IVAL(data, 0); + io_error |= val; if (!am_generator) - send_msg(MSG_IO_ERROR, data, 4, 0); + send_msg_int(MSG_IO_ERROR, val); + 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) @@ -1295,19 +1394,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) { - 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]) + 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 += 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) < 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) @@ -1335,19 +1441,21 @@ static void read_a_msg(void) exit_cleanup(RERR_STREAMIO); } data = perform_io(4, PIO_INPUT_AND_CONSUME); + val = IVAL(data, 0); if (am_generator) - got_flist_entry_status(FES_SUCCESS, IVAL(data, 0)); + got_flist_entry_status(FES_SUCCESS, val); else - successful_send(IVAL(data, 0)); + successful_send(val); break; case MSG_NO_SEND: if (msg_bytes != 4) goto invalid_msg; data = perform_io(4, PIO_INPUT_AND_CONSUME); + val = IVAL(data, 0); if (am_generator) - got_flist_entry_status(FES_NO_SEND, IVAL(data, 0)); + got_flist_entry_status(FES_NO_SEND, val); else - send_msg(MSG_NO_SEND, data, 4, 0); + send_msg_int(MSG_NO_SEND, val); break; case MSG_ERROR_SOCKET: case MSG_ERROR_UTF8: @@ -1380,6 +1488,42 @@ static void read_a_msg(void) first_message = 0; } break; + case MSG_ERROR_EXIT: + if (DEBUG_GTE(EXIT, 3)) + rprintf(FINFO, "[%s] got MSG_ERROR_EXIT with %d bytes\n", who_am_i(), msg_bytes); + if (msg_bytes == 0) { + if (!am_sender && !am_generator) { + if (DEBUG_GTE(EXIT, 3)) { + rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT (len 0)\n", + who_am_i()); + } + send_msg(MSG_ERROR_EXIT, "", 0, 0); + io_flush(FULL_FLUSH); + } + val = 0; + } else if (msg_bytes == 4) { + data = perform_io(4, PIO_INPUT_AND_CONSUME); + val = IVAL(data, 0); + if (protocol_version >= 31) { + if (am_generator) { + if (DEBUG_GTE(EXIT, 3)) { + rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT with exit_code %d\n", + who_am_i(), val); + } + send_msg_int(MSG_ERROR_EXIT, val); + } else { + if (DEBUG_GTE(EXIT, 3)) { + rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT (len 0)\n", + who_am_i()); + } + send_msg(MSG_ERROR_EXIT, "", 0, 0); + } + } + } else + goto invalid_msg; + /* Send a negative linenum so that we don't end up + * with a duplicate exit message. */ + _exit_cleanup(val, __FILE__, 0 - __LINE__); default: rprintf(FERROR, "unexpected tag %d [%s%s]\n", tag, who_am_i(), inc_recurse ? "/inc" : ""); @@ -1387,6 +1531,23 @@ static void read_a_msg(void) } } +static void drain_multiplex_messages(void) +{ + while (IN_MULTIPLEXED && iobuf.in.len) { + if (iobuf.raw_input_ends_before) { + size_t raw_len = iobuf.raw_input_ends_before - iobuf.in.pos; + iobuf.raw_input_ends_before = 0; + if (raw_len >= iobuf.in.len) { + iobuf.in.len = 0; + break; + } + iobuf.in.pos += raw_len; + iobuf.in.len -= raw_len; + } + read_a_msg(); + } +} + void wait_for_receiver(void) { if (!iobuf.raw_input_ends_before) @@ -1540,12 +1701,13 @@ void read_buf(int f, char *buf, size_t len) { if (f != iobuf.in_fd) { if (safe_read(f, buf, len) != len) - whine_about_eof(f); /* Doesn't return. */ + whine_about_eof(False); /* Doesn't return. */ goto batch_copy; } 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: @@ -1569,6 +1731,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); @@ -1826,7 +1989,7 @@ void write_buf(int f, const char *buf, size_t len) goto batch_copy; } - if (iobuf.out.size - iobuf.out.len < len) + if (iobuf.out.len + len > iobuf.out.size) perform_io(len, PIO_NEED_OUTROOM); pos = iobuf.out.pos + iobuf.out.len; /* Must be set after any flushing. */ @@ -1841,6 +2004,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) @@ -1994,11 +2158,11 @@ void io_printf(int fd, const char *format, ...) va_end(ap); if (len < 0) - exit_cleanup(RERR_STREAMIO); + exit_cleanup(RERR_PROTOCOL); if (len > (int)sizeof buf) { rprintf(FERROR, "io_printf() was too long for the buffer.\n"); - exit_cleanup(RERR_STREAMIO); + exit_cleanup(RERR_PROTOCOL); } write_sbuf(fd, buf); @@ -2007,10 +2171,13 @@ 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); + if (!iobuf.msg.buf) + alloc_xbuf(&iobuf.msg, ROUND_UP_1024(IO_BUFFER_SIZE)); iobuf.out_empty_len = 4; /* See also OUT_MULTIPLEXED */ io_start_buffering_out(fd); @@ -2029,28 +2196,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); - io_end_buffering_out(free_buffers); + if (mode != MPLX_TO_BUFFERED) + io_end_buffering_out(mode); + else + io_flush(FULL_FLUSH); + + iobuf.out.len = 0; iobuf.out_empty_len = 0; + + return ret; } void start_write_batch(int fd)