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;
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;
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;
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)
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);
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);
ff_forward_fd = -1;
write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
free_xbuf(&ff_xb);
- if (protocol_version == 30)
- io_start_multiplex_out(iobuf.out_fd);
+ if (ff_reenable_multiplex >= 0)
+ io_start_multiplex_out(ff_reenable_multiplex);
}
return;
}
}
}
+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;
+ }
+}
+
/* 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
* 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) {
- if (!(iobuf.in.buf = realloc_array(iobuf.in.buf, char, needed)))
- out_of_memory("perform_io");
- if (DEBUG_GTE(IO, 4)) {
- rprintf(FINFO, "[%s] resized input buffer from %ld to %ld bytes.\n",
- who_am_i(), (long)iobuf.in.size, (long)needed);
- }
- iobuf.in.size = needed;
- }
- 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:
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;
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;
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;
}
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.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",
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;
}
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:
}
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;
}
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. */
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;
}
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))
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)) {
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;
}
}
+ /* 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)) {
+ 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];
+
+ 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)
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) {
}
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);
* for recv_file_list() to use. */
void start_filesfrom_forwarding(int fd)
{
- if (protocol_version == 30) {
- /* Older protocols send the files-from data w/o packaging it in
- * multiplexed I/O packets, but protocol 30 messed up and did
- * this after starting multiplexing. We'll temporarily switch
+ 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. */
- io_end_multiplex_out(MPLX_TO_BUFFERED);
+ ff_reenable_multiplex = io_end_multiplex_out(MPLX_TO_BUFFERED);
}
ff_forward_fd = fd;
*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);
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;
+ return True;
}
void io_end_buffering_in(BOOL free_buffers)
/* Read a message from a multiplexed source. */
static void read_a_msg(void)
{
- char *data, line[BIGPATHBUFLEN];
- int tag;
+ char data[BIGPATHBUFLEN];
+ int tag, val;
size_t msg_bytes;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- tag = IVAL(data, 0);
+ tag = raw_read_int();
msg_bytes = tag & 0xFFFFFF;
tag = (tag >> 24) - MPLEX_BASE;
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);
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));
+ got_flist_entry_status(FES_REDO, raw_read_int());
break;
case MSG_IO_ERROR:
if (msg_bytes != 4 || am_sender)
goto invalid_msg;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- io_error |= IVAL(data, 0);
+ val = raw_read_int();
+ 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;
+ val = raw_read_int();
+ 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)
maybe_send_keepalive();
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);
+ 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);
/* 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 = raw_read_int();
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 = raw_read_int();
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:
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);
+ 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 (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) {
+ 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
+ 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" : "");
}
}
+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.len -= raw_len;
+ if ((iobuf.in.pos += raw_len) >= iobuf.in.size)
+ iobuf.in.pos -= iobuf.in.size;
+ }
+ read_a_msg();
+ }
+}
+
void wait_for_receiver(void)
{
if (!iobuf.raw_input_ends_before)
}
} 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 (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);
+ raw_read_buf(buf, len);
+ total_data_read += len;
if (forward_flist_data)
write_buf(iobuf.out_fd, buf, len);
batch_copy:
}
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)
write_buf(iobuf.out_fd, buf, siz);
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. */
memcpy(iobuf.out.buf + pos, buf, len);
iobuf.out.len += len;
+ total_data_written += len;
batch_copy:
if (f == write_batch_monitor_out)
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);
if (msgs2stderr && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_multiplex_out(%d)\n", who_am_i(), fd);
+ 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);
io_start_buffering_in(fd);
}
-void io_end_multiplex_in(int mode)
+int io_end_multiplex_in(int mode)
{
+ 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);
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(int mode)
+int io_end_multiplex_out(int mode)
{
+ 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);
iobuf.out.len = 0;
iobuf.out_empty_len = 0;
+
+ return ret;
}
void start_write_batch(int fd)
Matt McCutchen's Web Site / rsync / rsync.git / blobdiff