extern size_t bwlimit_writemax;
extern int io_timeout;
extern int am_server;
-extern int am_daemon;
extern int am_sender;
+extern int am_receiver;
extern int am_generator;
extern int msgs2stderr;
extern int inc_recurse;
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;
BOOL flist_receiving_enabled = False;
+BOOL we_send_keepalive_messages = False;
/* Ignore an EOF error if non-zero. See whine_about_eof(). */
int kluge_around_eof = 0;
xbuf in, out, msg;
int in_fd;
int out_fd; /* Both "out" and "msg" go to this fd. */
- BOOL in_multiplexed;
+ int in_multiplexed;
unsigned out_empty_len;
size_t raw_data_header_pos; /* in the out xbuf */
size_t raw_flushing_ends_before; /* in the out xbuf */
#define IOBUF_WAS_REDUCED(siz) ((siz) & 0xFF)
#define IOBUF_RESTORE_SIZE(siz) (((siz) | 0xFF) + 1)
-#define IN_MULTIPLEXED (iobuf.in_multiplexed)
+#define IN_MULTIPLEXED (iobuf.in_multiplexed != 0)
+#define IN_MULTIPLEXED_AND_READY (iobuf.in_multiplexed > 0)
#define OUT_MULTIPLEXED (iobuf.out_empty_len != 0)
#define PIO_NEED_INPUT (1<<0) /* The *_NEED_* flags are mutually exclusive. */
static void drain_multiplex_messages(void);
static void sleep_for_bwlimit(int bytes_written);
-static void check_timeout(void)
+static void check_timeout(BOOL allow_keepalive)
{
- time_t t;
+ time_t t, chk;
- if (!io_timeout || ignore_timeout)
+ /* On the receiving side, the generator is now handling timeouts, so
+ * the receiver ignores them. Note that the am_receiver flag is not
+ * set until the receiver forks from the generator, so timeouts will be
+ * based on receiving data on the receiving side until that event. */
+ if (!io_timeout || am_receiver)
return;
- if (!last_io_in) {
- last_io_in = time(NULL);
- return;
+ t = time(NULL);
+
+ if (allow_keepalive && we_send_keepalive_messages) {
+ /* This may put data into iobuf.msg w/o flushing. */
+ maybe_send_keepalive(t, False);
}
- t = time(NULL);
+ if (!last_io_in)
+ last_io_in = t;
- if (t - last_io_in >= io_timeout) {
- if (!am_server && !am_daemon) {
- rprintf(FERROR, "io timeout after %d seconds -- exiting\n",
- (int)(t-last_io_in));
- }
+ 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-chk));
exit_cleanup(RERR_TIMEOUT);
}
}
who_am_i());
exit_cleanup(RERR_FILEIO);
}
- check_timeout();
+ if (we_send_keepalive_messages)
+ maybe_send_keepalive(time(NULL), True);
continue;
}
what_fd_is(fd), who_am_i());
exit_cleanup(RERR_FILEIO);
}
- check_timeout();
+ if (we_send_keepalive_messages)
+ maybe_send_keepalive(time(NULL), True);
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);
}
iobuf.in.len += n;
}
- if (iobuf.out_fd >= 0 && FD_ISSET(iobuf.out_fd, &w_fds)) {
+ if (out && FD_ISSET(iobuf.out_fd, &w_fds)) {
size_t len = iobuf.raw_flushing_ends_before ? iobuf.raw_flushing_ends_before - out->pos : out->len;
int n;
/* We need to help prevent deadlock by doing what reading
* we can whenever we are here trying to write. */
- if (IN_MULTIPLEXED && !(flags & PIO_NEED_INPUT)) {
+ if (IN_MULTIPLEXED_AND_READY && !(flags & PIO_NEED_INPUT)) {
while (!iobuf.raw_input_ends_before && iobuf.in.len > 512)
read_a_msg();
if (flist_receiving_enabled && iobuf.in.len > 512)
read_buf(iobuf.in_fd, buf, sizeof buf);
}
-/* Buffer a message for the multiplexed output stream. Is never used for MSG_DATA. */
+/* Buffer a message for the multiplexed output stream. Is not used for (normal) MSG_DATA. */
int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
{
char *hdr;
{
size_t siz;
- if ((pos += 4) >= iobuf.msg.size)
- pos -= iobuf.msg.size;
+ if ((pos += 4) == iobuf.msg.size)
+ pos = 0;
/* Handle a split copy if we wrap around the end of the circular buffer. */
if (pos >= iobuf.msg.pos && (siz = iobuf.msg.size - pos) < len) {
io_flush(NORMAL_FLUSH);
}
-void maybe_send_keepalive(void)
-{
- if (time(NULL) - last_io_out >= allowed_lull) {
- if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len) {
- if (protocol_version < 29)
- return; /* there's nothing we can do */
- if (protocol_version >= 30)
- send_msg(MSG_NOOP, "", 0, 0);
- else {
- write_int(iobuf.out_fd, cur_flist->used);
- write_shortint(iobuf.out_fd, ITEM_IS_NEW);
- }
- }
- if (iobuf.msg.len)
+/* 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, BOOL allow_flush)
+{
+ if (now - last_io_out >= allowed_lull) {
+ if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len)
+ send_msg(MSG_DATA, "", 0, 0);
+ if (!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);
/* Read a message from a multiplexed source. */
static void read_a_msg(void)
{
- char line[BIGPATHBUFLEN];
+ char data[BIGPATHBUFLEN];
int tag, val;
size_t msg_bytes;
+ /* This ensures that perform_io() does not try to do any message reading
+ * until we've read all of the data for this message. We should also
+ * try to avoid calling things that will cause data to be written via
+ * perform_io() prior to this being reset to 1. */
+ iobuf.in_multiplexed = -1;
+
tag = raw_read_int();
msg_bytes = tag & 0xFFFFFF;
tag = (tag >> 24) - MPLEX_BASE;
- if (DEBUG_GTE(IO, 1) && (msgs2stderr || tag != MSG_INFO))
+ if (DEBUG_GTE(IO, 1) && msgs2stderr)
rprintf(FINFO, "[%s] got msg=%d, len=%ld\n", who_am_i(), (int)tag, (long)msg_bytes);
switch (tag) {
* 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. */
- iobuf.raw_input_ends_before = iobuf.in.pos + msg_bytes;
+ * 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;
break;
case MSG_STATS:
if (msg_bytes != sizeof stats.total_read || !am_generator)
goto invalid_msg;
raw_read_buf((char*)&stats.total_read, sizeof stats.total_read);
+ iobuf.in_multiplexed = 1;
break;
case MSG_REDO:
if (msg_bytes != 4 || !am_generator)
goto invalid_msg;
- got_flist_entry_status(FES_REDO, raw_read_int());
+ val = raw_read_int();
+ 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:
if (msg_bytes != 4 || am_server || am_generator)
goto invalid_msg;
val = raw_read_int();
+ iobuf.in_multiplexed = 1;
if (!io_timeout || io_timeout > val) {
if (INFO_GTE(MISC, 2))
rprintf(FINFO, "Setting --timeout=%d to match server\n", val);
}
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), True);
break;
case MSG_DELETED:
- if (msg_bytes >= sizeof line)
+ if (msg_bytes >= sizeof data)
goto overflow;
if (am_generator) {
- raw_read_buf(line, msg_bytes);
- send_msg(MSG_DELETED, line, msg_bytes, 1);
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ send_msg(MSG_DELETED, data, msg_bytes, 1);
break;
}
#ifdef ICONV_OPTION
int add_null = 0;
int flags = ICB_INCLUDE_BAD | ICB_INIT;
- INIT_CONST_XBUF(outbuf, line);
+ INIT_CONST_XBUF(outbuf, data);
INIT_XBUF(inbuf, ibuf, 0, (size_t)-1);
while (msg_bytes) {
msg_bytes = outbuf.len;
} else
#endif
- raw_read_buf(line, msg_bytes);
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
/* A directory name was sent with the trailing null */
- if (msg_bytes > 0 && !line[msg_bytes-1])
- log_delete(line, S_IFDIR);
+ if (msg_bytes > 0 && !data[msg_bytes-1])
+ log_delete(data, S_IFDIR);
else {
- line[msg_bytes] = '\0';
- log_delete(line, S_IFREG);
+ data[msg_bytes] = '\0';
+ log_delete(data, S_IFREG);
}
break;
case MSG_SUCCESS:
exit_cleanup(RERR_STREAMIO);
}
val = raw_read_int();
+ iobuf.in_multiplexed = 1;
if (am_generator)
got_flist_entry_status(FES_SUCCESS, val);
else
if (msg_bytes != 4)
goto invalid_msg;
val = raw_read_int();
+ iobuf.in_multiplexed = 1;
if (am_generator)
got_flist_entry_status(FES_NO_SEND, val);
else
case MSG_ERROR:
case MSG_ERROR_XFER:
case MSG_WARNING:
- if (msg_bytes >= sizeof line) {
+ if (msg_bytes >= sizeof data) {
overflow:
rprintf(FERROR,
"multiplexing overflow %d:%lu [%s%s]\n",
inc_recurse ? "/inc" : "");
exit_cleanup(RERR_STREAMIO);
}
- raw_read_buf(line, msg_bytes);
- rwrite((enum logcode)tag, line, msg_bytes, !am_generator);
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ rwrite((enum logcode)tag, data, msg_bytes, !am_generator);
if (first_message) {
- if (list_only && !am_sender && tag == 1 && msg_bytes < sizeof line) {
- line[msg_bytes] = '\0';
- check_for_d_option_error(line);
+ if (list_only && !am_sender && tag == 1 && msg_bytes < sizeof data) {
+ data[msg_bytes] = '\0';
+ check_for_d_option_error(data);
}
first_message = 0;
}
break;
case MSG_ERROR_EXIT:
+ if (msg_bytes == 4)
+ val = raw_read_int();
+ else if (msg_bytes == 0)
+ val = 0;
+ else
+ goto invalid_msg;
+ iobuf.in_multiplexed = 1;
if (DEBUG_GTE(EXIT, 3))
- rprintf(FINFO, "[%s] got MSG_ERROR_EXIT with %d bytes\n", who_am_i(), msg_bytes);
+ rprintf(FINFO, "[%s] got MSG_ERROR_EXIT with %ld bytes\n", who_am_i(), (long)msg_bytes);
if (msg_bytes == 0) {
if (!am_sender && !am_generator) {
if (DEBUG_GTE(EXIT, 3)) {
send_msg(MSG_ERROR_EXIT, "", 0, 0);
io_flush(FULL_FLUSH);
}
- val = 0;
- } else if (msg_bytes == 4) {
- val = raw_read_int();
- 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 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__);
tag, who_am_i(), inc_recurse ? "/inc" : "");
exit_cleanup(RERR_STREAMIO);
}
+
+ assert(iobuf.in_multiplexed > 0);
}
static void drain_multiplex_messages(void)
{
- while (IN_MULTIPLEXED && iobuf.in.len) {
+ while (IN_MULTIPLEXED_AND_READY && 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 (msgs2stderr && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_multiplex_in(%d)\n", who_am_i(), fd);
- iobuf.in_multiplexed = True; /* See also IN_MULTIPLEXED */
+ iobuf.in_multiplexed = 1; /* See also IN_MULTIPLEXED */
io_start_buffering_in(fd);
}
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.in_multiplexed = 0;
if (mode == MPLX_SWITCHING)
iobuf.raw_input_ends_before = 0;
else
Matt McCutchen's Web Site / rsync / rsync.git / blobdiff