static flist_ndx_list redo_list, hlink_list;
+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);
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) {
+ if (iobuf.raw_data_header_pos + 4 > iobuf.out.size) {
+ int siz = (int)(iobuf.raw_data_header_pos + 4 - iobuf.out.size);
/* 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);
+ who_am_i(), siz);
}
- memcpy(iobuf.out.buf, iobuf.out.buf + iobuf.out.size, 4 - val);
+ memcpy(iobuf.out.buf, iobuf.out.buf + iobuf.out.size, siz);
}
if (DEBUG_GTE(IO, 1)) {
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;
int n;
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;
+ 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());
- exit_cleanup(RERR_STREAMIO);
+ 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)
+ } 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;
{
char buf[1024];
- kluge_around_eof = 1;
- set_io_timeout(protocol_version >= 31 ? 10 : 1);
+ 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);
if (convert > 0 && ic_send == (iconv_t)-1)
convert = 0;
if (convert > 0) {
- /* Ensuring double-size room leaves space for a potential conversion. */
+ /* Ensuring double-size room leaves space for maximal conversion expansion. */
if (iobuf.msg.len + len*2 + 4 > iobuf.msg.size)
perform_io(len*2 + 4, PIO_NEED_MSGROOM);
} else
pos -= iobuf.msg.size;
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;
+ int siz = (int)(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);
+ rprintf(FINFO, "[%s] wrap-bytes moved: %d (send_msg)\n", who_am_i(), siz);
+ memcpy(iobuf.msg.buf, iobuf.msg.buf + iobuf.msg.size, siz);
}
if (want_debug && convert > 0)
if (iobuf.out.buf) {
if (iobuf.out_fd == -1)
iobuf.out_fd = f_out;
- assert(f_out == iobuf.out_fd);
+ else
+ assert(f_out == iobuf.out_fd);
return False;
}
if (iobuf.in.buf) {
if (iobuf.in_fd == -1)
iobuf.in_fd = f_in;
- assert(f_in == iobuf.in_fd);
+ else
+ assert(f_in == iobuf.in_fd);
return False;
}
}
}
+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.pos += raw_len;
+ iobuf.in.len -= raw_len;
+ }
+ read_a_msg();
+ }
+}
+
void wait_for_receiver(void)
{
if (!iobuf.raw_input_ends_before)
{
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;
}
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);