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;
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;
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;
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 (am_receiver)
+ return;
- if (t - last_io_in >= io_timeout) {
+ 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);
}
}
who_am_i());
exit_cleanup(RERR_FILEIO);
}
- check_timeout();
+ if (io_timeout)
+ maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
continue;
}
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;
}
}
}
-/* Perform buffered input and output until specified conditions are met. When
- * given a "needed" read requirement, we'll return without doing any I/O if the
- * iobuf.in bytes are already available. When reading, we'll read as many
- * bytes as we can into the buffer, and return as soon as we meet the minimum
- * read requirement. When given a "needed" write requirement, we'll return
- * without doing any I/O if that many bytes will fit in the output buffer (we
- * check either iobuf.out or iobuf.msg, depending on the flags). When writing,
- * we write out as much as we can, and return as soon as the given free-space
- * requirement is available.
+/* Perform buffered input and/or output until specified conditions are met.
+ * When given a "needed" read or write request, this returns without doing any
+ * I/O if the needed input bytes or write space is already available. Once I/O
+ * is needed, this will try to do whatever reading and/or writing is currently
+ * possible, up to the maximum buffer allowances, no matter if this is a read
+ * or write request. However, the I/O stops as soon as the required input
+ * bytes or output space is available. If this is not a read request, the
+ * routine may also do some advantageous reading of messages from a multiplexed
+ * input source (which ensures that we don't jam up with everyone in their
+ * "need to write" code and nobody reading the accumulated data that would make
+ * writing possible).
*
- * The iobuf.out and iobuf.msg buffers are circular, so some writes into them
- * will need to be split when the data needs to wrap around to the start. In
- * order to help make this easier for some operations (such as the use of
- * SIVAL() into the buffer) a buffer may be temporarily shortened, but the
- * original size will be automatically restored. The iobuf.in buffer is also
- * circular, so callers may need to split their reading of the data if it spans
- * the end. See also the 3 raw_* iobuf vars that are used in the handling of
+ * The iobuf.in, .out and .msg buffers are all circular. Callers need to be
+ * aware that some data copies will need to be split when the bytes wrap around
+ * from the end to the start. In order to help make writing into the output
+ * buffers easier for some operations (such as the use of SIVAL() into the
+ * buffer) a buffer may be temporarily shortened by a small amount, but the
+ * original size will be automatically restored when the .pos wraps to the
+ * start. See also the 3 raw_* iobuf vars that are used in the handling of
* MSG_DATA bytes as they are read-from/written-into the buffers.
*
* When writing, we flush data in the following priority order:
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);
}
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;
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)
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);
* the buffer the msg data will end once it is read. It is
* possible that this points off the end of the buffer, in
* which case the gradual reading of the input stream will
- * cause this value to decrease and eventually become real. */
+ * cause this value to wrap around and eventually become real. */
if (msg_bytes)
iobuf.raw_input_ends_before = iobuf.in.pos + msg_bytes;
iobuf.in_multiplexed = 1;
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:
}
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)
/* Read a line of up to bufsiz-1 characters into buf. Strips
* the (required) trailing newline and all carriage returns.
* Returns 1 for success; 0 for I/O error or truncation. */
-int read_line_old(int fd, char *buf, size_t bufsiz)
+int read_line_old(int fd, char *buf, size_t bufsiz, int eof_ok)
{
+ assert(fd != iobuf.in_fd);
bufsiz--; /* leave room for the null */
while (bufsiz > 0) {
- assert(fd != iobuf.in_fd);
- if (safe_read(fd, buf, 1) == 0)
+ if (safe_read(fd, buf, 1) == 0) {
+ if (eof_ok)
+ break;
return 0;
+ }
if (*buf == '\0')
return 0;
if (*buf == '\n')
* 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)