int batch_fd = -1;
int msgdone_cnt = 0;
int forward_flist_data = 0;
+BOOL flist_receiving_enabled = 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 flist_ndx_list redo_list, hlink_list;
+static void read_a_msg(void);
static void drain_multiplex_messages(void);
static void sleep_for_bwlimit(int bytes_written);
* 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) the buffers MUST have 4 bytes of overflow space at
- * the end that is not not counted in the "size". The iobuf.in buffer is not
- * (currently) circular. To facilitate the handling of MSG_DATA bytes as they
- * are read-from/written-into the buffers, see the three raw_* iobuf vars.
+ * 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
+ * MSG_DATA bytes as they are read-from/written-into the buffers.
*
* When writing, we flush data in the following priority order:
*
*
* - Make this routine able to read the generator-to-receiver batch flow?
*
- * - Make the input buffer circular?
- *
* Unlike the old routines that this replaces, it is OK to read ahead as far as
* we can because the read_a_msg() routine now reads its bytes out of the input
* buffer. In the old days, only raw data was in the input buffer, and any
switch (flags & PIO_NEED_FLAGS) {
case PIO_NEED_INPUT:
+ /* We never resize the circular input buffer. */
+ if (iobuf.in.size < needed) {
+ rprintf(FERROR, "need to read %ld bytes, iobuf.in.buf is only %ld bytes.\n",
+ (long)needed, (long)iobuf.in.size);
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
if (DEBUG_GTE(IO, 3)) {
rprintf(FINFO, "[%s] perform_io(%ld, %sinput)\n",
who_am_i(), (long)needed, flags & PIO_CONSUME_INPUT ? "consume&" : "");
}
-
- /* Make sure the input buffer is big enough to hold "needed" bytes.
- * 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) {
- size_t new_size = ROUND_UP_1024(needed);
- if (DEBUG_GTE(IO, 4)) {
- rprintf(FINFO, "[%s] resizing input buffer from %ld to %ld bytes.\n",
- who_am_i(), (long)iobuf.in.size, (long)new_size);
- }
- realloc_xbuf(&iobuf.in, new_size);
- }
- if (iobuf.in.size - iobuf.in.pos < needed
- || (iobuf.in.len < needed && iobuf.in.len < 1024
- && iobuf.in.size - (iobuf.in.pos + iobuf.in.len) < 1024)) {
- memmove(iobuf.in.buf, iobuf.in.buf + iobuf.in.pos, iobuf.in.len);
- if (DEBUG_GTE(IO, 4)) {
- rprintf(FINFO,
- "[%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;
- iobuf.in.pos = 0;
- }
break;
case PIO_NEED_OUTROOM:
FD_ZERO(&r_fds);
FD_ZERO(&e_fds);
- if (iobuf.in_fd >= 0 && iobuf.in.size - (iobuf.in.pos + iobuf.in.len)) {
+ if (iobuf.in_fd >= 0 && iobuf.in.size - iobuf.in.len) {
if (!read_batch || batch_fd >= 0) {
FD_SET(iobuf.in_fd, &r_fds);
FD_SET(iobuf.in_fd, &e_fds);
}
if (iobuf.in_fd >= 0 && FD_ISSET(iobuf.in_fd, &r_fds)) {
- size_t pos = iobuf.in.pos + iobuf.in.len;
- size_t len = iobuf.in.size - pos;
+ size_t len, pos = iobuf.in.pos + iobuf.in.len;
int n;
+ if (pos >= iobuf.in.size) {
+ pos -= iobuf.in.size;
+ len = iobuf.in.size - iobuf.in.len;
+ } else
+ len = iobuf.in.size - pos;
if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
if (n == 0) {
/* Signal that input has become invalid. */
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_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)
+ wait_for_receiver(); /* generator only */
+ }
+
if (ff_forward_fd >= 0 && FD_ISSET(ff_forward_fd, &r_fds)) {
/* This can potentially flush all output and enable
* multiplexed output, so keep this last in the loop
if (flags & PIO_CONSUME_INPUT) {
iobuf.in.len -= needed;
iobuf.in.pos += needed;
+ if (iobuf.in.pos == iobuf.raw_input_ends_before)
+ iobuf.raw_input_ends_before = 0;
+ if (iobuf.in.pos >= iobuf.in.size) {
+ iobuf.in.pos -= iobuf.in.size;
+ if (iobuf.raw_input_ends_before)
+ iobuf.raw_input_ends_before -= iobuf.in.size;
+ }
}
return data;
}
+static void raw_read_buf(char *buf, size_t len)
+{
+ size_t pos = iobuf.in.pos;
+ char *data = perform_io(len, PIO_INPUT_AND_CONSUME);
+ if (iobuf.in.pos <= pos && len) {
+ size_t siz = len - iobuf.in.pos;
+ memcpy(buf, data, siz);
+ memcpy(buf + siz, iobuf.in.buf, iobuf.in.pos);
+ } else
+ memcpy(buf, data, len);
+}
+
+static int32 raw_read_int(void)
+{
+ char *data, buf[4];
+ if (iobuf.in.size - iobuf.in.pos >= 4)
+ data = perform_io(4, PIO_INPUT_AND_CONSUME);
+ else
+ raw_read_buf(data = buf, 4);
+ return IVAL(data, 0);
+}
+
void noop_io_until_death(void)
{
char buf[1024];
/* Read a message from a multiplexed source. */
static void read_a_msg(void)
{
- char *data, line[BIGPATHBUFLEN];
+ char data[BIGPATHBUFLEN];
int tag, val;
size_t msg_bytes;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- tag = IVAL(data, 0);
+ /* 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) {
* 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;
+ iobuf.in_multiplexed = 1;
break;
case MSG_STATS:
if (msg_bytes != sizeof stats.total_read || !am_generator)
goto invalid_msg;
- data = perform_io(sizeof stats.total_read, PIO_INPUT_AND_CONSUME);
- memcpy((char*)&stats.total_read, data, sizeof stats.total_read);
+ 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;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- got_flist_entry_status(FES_REDO, IVAL(data, 0));
+ 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)
goto invalid_msg;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- val = IVAL(data, 0);
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
io_error |= val;
if (!am_generator)
send_msg_int(MSG_IO_ERROR, val);
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);
+ 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);
case MSG_NOOP:
if (am_sender)
maybe_send_keepalive();
+ iobuf.in_multiplexed = 1;
break;
case MSG_DELETED:
- if (msg_bytes >= sizeof line)
+ if (msg_bytes >= sizeof data)
goto overflow;
if (am_generator) {
- memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), 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) {
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);
+ raw_read_buf(ibuf + inbuf.len, len);
inbuf.pos = 0;
inbuf.len += len;
if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])
msg_bytes = outbuf.len;
} else
#endif
- memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), 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:
inc_recurse ? "/inc" : "");
exit_cleanup(RERR_STREAMIO);
}
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- val = IVAL(data, 0);
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
if (am_generator)
got_flist_entry_status(FES_SUCCESS, val);
else
case MSG_NO_SEND:
if (msg_bytes != 4)
goto invalid_msg;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- val = IVAL(data, 0);
+ 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);
}
- memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), 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 %ld bytes\n", who_am_i(), (long)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);
+ } else {
+ val = raw_read_int();
if (protocol_version >= 31) {
- if (am_generator)
+ 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
+ } 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;
iobuf.in.len = 0;
break;
}
- iobuf.in.pos += raw_len;
iobuf.in.len -= raw_len;
+ if ((iobuf.in.pos += raw_len) >= iobuf.in.size)
+ iobuf.in.pos -= iobuf.in.size;
}
read_a_msg();
}
}
} else {
struct file_list *flist;
+ flist_receiving_enabled = False;
if (DEBUG_GTE(FLIST, 2)) {
rprintf(FINFO, "[%s] receiving flist for dir %d\n",
who_am_i(), ndx);
if (preserve_hard_links)
match_hard_links(flist);
#endif
+ flist_receiving_enabled = True;
}
}
}
}
if (!IN_MULTIPLEXED) {
- memcpy(buf, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+ raw_read_buf(buf, len);
total_data_read += len;
if (forward_flist_data)
write_buf(iobuf.out_fd, buf, len);
}
while (1) {
- char *data;
size_t siz;
while (!iobuf.raw_input_ends_before)
read_a_msg();
siz = MIN(len, iobuf.raw_input_ends_before - iobuf.in.pos);
- data = perform_io(siz, PIO_INPUT_AND_CONSUME);
- if (iobuf.in.pos == iobuf.raw_input_ends_before)
- iobuf.raw_input_ends_before = 0;
-
- /* The bytes at the "data" pointer will survive long
- * enough to make a copy, but not past future I/O. */
- memcpy(buf, data, siz);
+ if (siz >= iobuf.in.size)
+ siz = iobuf.in.size;
+ raw_read_buf(buf, siz);
total_data_read += siz;
if (forward_flist_data)
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