X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/blobdiff_plain/eeea1bbd72cf881a1be1826e68ba97e483f774c5..ef7441669b194ba6804c2eda07beed90a29ad2b3:/io.c diff --git a/io.c b/io.c index 1ae25ce5..bf39ff91 100644 --- a/io.c +++ b/io.c @@ -39,6 +39,7 @@ extern size_t bwlimit_writemax; extern int io_timeout; extern int am_server; extern int am_sender; +extern int am_receiver; extern int am_generator; extern int msgs2stderr; extern int inc_recurse; @@ -49,6 +50,7 @@ extern int file_total; extern int file_old_total; extern int list_only; extern int read_batch; +extern int compat_flags; extern int protect_args; extern int checksum_seed; extern int protocol_version; @@ -64,7 +66,6 @@ extern iconv_t ic_send, ic_recv; int csum_length = SHORT_SUM_LENGTH; /* initial value */ int allowed_lull = 0; -int ignore_timeout = 0; int batch_fd = -1; int msgdone_cnt = 0; int forward_flist_data = 0; @@ -152,25 +153,43 @@ static void read_a_msg(void); static void drain_multiplex_messages(void); static void sleep_for_bwlimit(int bytes_written); -static void check_timeout(void) -{ - time_t t; - - if (!io_timeout || ignore_timeout) +static void check_timeout(BOOL allow_keepalive) +{ + time_t t, chk; + + /* On the receiving side, the generator is now the one that decides + * when a timeout has occurred. When it is sifting through a lot of + * files looking for work, it will be sending keep-alive messages to + * the sender, and even though the receiver won't be sending/receiving + * anything (not even keep-alive messages), the successful writes to + * the sender will keep things going. If the receiver is actively + * receiving data, it will ensure that the generator knows that it is + * not idle by sending the generator keep-alive messages (since the + * generator might be blocked trying to send checksums, it needs to + * know that the receiver is active). Thus, as long as one or the + * other is successfully doing work, the generator will not timeout. */ + if (!io_timeout) return; - if (!last_io_in) { - last_io_in = time(NULL); - return; + t = time(NULL); + + if (allow_keepalive) { + /* This may put data into iobuf.msg w/o flushing. */ + maybe_send_keepalive(t, 0); } - t = time(NULL); + if (!last_io_in) + last_io_in = t; - if (t - last_io_in >= io_timeout) { + if (am_receiver) + return; + + chk = MAX(last_io_out, last_io_in); + if (t - chk >= io_timeout) { if (am_server) msgs2stderr = 1; rprintf(FERROR, "[%s] io timeout after %d seconds -- exiting\n", - who_am_i(), (int)(t-last_io_in)); + who_am_i(), (int)(t-chk)); exit_cleanup(RERR_TIMEOUT); } } @@ -252,7 +271,8 @@ static size_t safe_read(int fd, char *buf, size_t len) who_am_i()); exit_cleanup(RERR_FILEIO); } - check_timeout(); + if (io_timeout) + maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH); continue; } @@ -336,7 +356,8 @@ static void safe_write(int fd, const char *buf, size_t len) what_fd_is(fd), who_am_i()); exit_cleanup(RERR_FILEIO); } - check_timeout(); + if (io_timeout) + maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH); continue; } @@ -733,7 +754,7 @@ static char *perform_io(size_t needed, int flags) send_extra_file_list(sock_f_out, -1); extra_flist_sending_enabled = !flist_eof; } else - check_timeout(); + check_timeout((flags & PIO_NEED_INPUT) != 0); FD_ZERO(&r_fds); /* Just in case... */ FD_ZERO(&w_fds); } @@ -770,8 +791,11 @@ static char *perform_io(size_t needed, int flags) if (msgs2stderr && DEBUG_GTE(IO, 2)) rprintf(FINFO, "[%s] recv=%ld\n", who_am_i(), (long)n); - if (io_timeout) + if (io_timeout) { last_io_in = time(NULL); + if (flags & PIO_NEED_INPUT) + maybe_send_keepalive(last_io_in, 0); + } stats.total_read += n; iobuf.in.len += n; @@ -1042,13 +1066,15 @@ void io_set_sock_fds(int f_in, int f_out) void set_io_timeout(int secs) { io_timeout = secs; + allowed_lull = (io_timeout + 1) / 2; - if (!io_timeout || io_timeout > SELECT_TIMEOUT) + if (!io_timeout || allowed_lull > SELECT_TIMEOUT) select_timeout = SELECT_TIMEOUT; else - select_timeout = io_timeout; + select_timeout = allowed_lull; - allowed_lull = read_batch ? 0 : (io_timeout + 1) / 2; + if (read_batch) + allowed_lull = 0; } static void check_for_d_option_error(const char *msg) @@ -1315,18 +1341,29 @@ void maybe_flush_socket(int important) io_flush(NORMAL_FLUSH); } -/* This never adds new non-msg-buffer data, since we don't know the state - * of the raw-data buffer. */ -void maybe_send_keepalive(void) +/* Older rsync versions used to send either a MSG_NOOP (protocol 30) or a + * raw-data-based keep-alive (protocol 29), both of which implied forwarding of + * the message through the sender. Since the new timeout method does not need + * any forwarding, we just send an empty MSG_DATA message, which works with all + * rsync versions. This avoids any message forwarding, and leaves the raw-data + * stream alone (since we can never be quite sure if that stream is in the + * right state for a keep-alive message). */ +void maybe_send_keepalive(time_t now, int flags) { - if (time(NULL) - last_io_out >= allowed_lull) { - if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len) { - if (protocol_version >= 30) - send_msg(MSG_NOOP, "", 0, 0); - else - send_msg(MSG_DATA, "", 0, 0); - } - if (iobuf.msg.len) + if (flags & MSK_ACTIVE_RECEIVER) + last_io_in = now; /* Fudge things when we're working hard on the files. */ + + if (now - last_io_out >= allowed_lull) { + /* The receiver is special: it only sends keep-alive messages if it is + * actively receiving data. Otherwise, it lets the generator timeout. */ + if (am_receiver && now - last_io_in >= io_timeout) + return; + + if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len) + send_msg(MSG_DATA, "", 0, 0); + if (!(flags & MSK_ALLOW_FLUSH)) { + /* Let the caller worry about writing out the data. */ + } else if (iobuf.msg.len) perform_io(iobuf.msg.size - iobuf.msg.len + 1, PIO_NEED_MSGROOM); else if (iobuf.out.len > iobuf.out_empty_len) io_flush(NORMAL_FLUSH); @@ -1391,12 +1428,12 @@ static void read_a_msg(void) got_flist_entry_status(FES_REDO, val); break; case MSG_IO_ERROR: - if (msg_bytes != 4 || am_sender) + if (msg_bytes != 4) goto invalid_msg; val = raw_read_int(); iobuf.in_multiplexed = 1; io_error |= val; - if (!am_generator) + if (am_receiver) send_msg_int(MSG_IO_ERROR, val); break; case MSG_IO_TIMEOUT: @@ -1411,11 +1448,12 @@ static void read_a_msg(void) } break; case MSG_NOOP: + /* Support protocol-30 keep-alive method. */ if (msg_bytes != 0) goto invalid_msg; iobuf.in_multiplexed = 1; if (am_sender) - maybe_send_keepalive(); + maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH); break; case MSG_DELETED: if (msg_bytes >= sizeof data) @@ -2281,7 +2319,7 @@ void start_write_batch(int fd) * is involved. */ write_int(batch_fd, protocol_version); if (protocol_version >= 30) - write_byte(batch_fd, inc_recurse); + write_byte(batch_fd, compat_flags); write_int(batch_fd, checksum_seed); if (am_sender)