+static void check_timeout(void)
+{
+ time_t t;
+
+ if (!io_timeout || ignore_timeout)
+ return;
+
+ if (!last_io_in) {
+ last_io_in = time(NULL);
+ return;
+ }
+
+ t = time(NULL);
+
+ 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));
+ }
+ exit_cleanup(RERR_TIMEOUT);
+ }
+}
+
+/* It's almost always an error to get an EOF when we're trying to read from the
+ * network, because the protocol is (for the most part) self-terminating.
+ *
+ * There is one case for the receiver when it is at the end of the transfer
+ * (hanging around reading any keep-alive packets that might come its way): if
+ * the sender dies before the generator's kill-signal comes through, we can end
+ * up here needing to loop until the kill-signal arrives. In this situation,
+ * kluge_around_eof will be < 0.
+ *
+ * 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)
+{
+ if (kluge_around_eof && fd == sock_f_in) {
+ int i;
+ if (kluge_around_eof > 0)
+ exit_cleanup(0);
+ /* If we're still here after 10 seconds, exit with an error. */
+ for (i = 10*1000/20; i--; )
+ msleep(20);
+ }
+
+ rprintf(FERROR, RSYNC_NAME ": connection unexpectedly closed "
+ "(%s bytes received so far) [%s]\n",
+ big_num(stats.total_read), who_am_i());
+
+ exit_cleanup(RERR_STREAMIO);
+}
+
+/* Do a safe read, handling any needed looping and error handling.
+ * Returns the count of the bytes read, which will only be different
+ * from "len" if we encountered an EOF. This routine is not used on
+ * the socket except very early in the transfer. */
+static size_t safe_read(int fd, char *buf, size_t len)
+{
+ size_t got;
+ int n;
+
+ assert(fd != iobuf.in_fd);
+
+ n = read(fd, buf, len);
+ if ((size_t)n == len || n == 0) {
+ if (DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] safe_read(%d)=%ld\n", who_am_i(), fd, (long)n);
+ return n;
+ }
+ if (n < 0) {
+ if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) {
+ read_failed:
+ rsyserr(FERROR, errno, "safe_read failed to read %ld bytes [%s]",
+ (long)len, who_am_i());
+ exit_cleanup(RERR_STREAMIO);
+ }
+ got = 0;
+ } else
+ got = n;
+
+ while (1) {
+ struct timeval tv;
+ fd_set r_fds, e_fds;
+ int cnt;
+
+ FD_ZERO(&r_fds);
+ FD_SET(fd, &r_fds);
+ FD_ZERO(&e_fds);
+ FD_SET(fd, &e_fds);
+ tv.tv_sec = select_timeout;
+ tv.tv_usec = 0;
+
+ cnt = select(fd+1, &r_fds, NULL, &e_fds, &tv);
+ if (cnt <= 0) {
+ if (cnt < 0 && errno == EBADF) {
+ rsyserr(FERROR, errno, "safe_read select failed [%s]",
+ who_am_i());
+ exit_cleanup(RERR_FILEIO);
+ }
+ check_timeout();
+ continue;
+ }
+
+ /*if (FD_ISSET(fd, &e_fds))
+ rprintf(FINFO, "select exception on fd %d\n", fd); */
+
+ if (FD_ISSET(fd, &r_fds)) {
+ n = read(fd, buf + got, len - got);
+ if (DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] safe_read(%d)=%ld\n", who_am_i(), fd, (long)n);
+ if (n == 0)
+ break;
+ if (n < 0) {
+ if (errno == EINTR)
+ continue;
+ goto read_failed;
+ }
+ if ((got += (size_t)n) == len)
+ break;
+ }
+ }
+
+ return got;
+}
+
+static const char *what_fd_is(int fd)
+{
+ static char buf[20];
+
+ if (fd == sock_f_out)
+ return "socket";
+ else if (fd == iobuf.out_fd)
+ return "message fd";
+ else if (fd == batch_fd)
+ return "batch file";
+ else {
+ snprintf(buf, sizeof buf, "fd %d", fd);
+ return buf;
+ }
+}
+
+/* Do a safe write, handling any needed looping and error handling.
+ * Returns only if everything was successfully written. This routine
+ * is not used on the socket except very early in the transfer. */
+static void safe_write(int fd, const char *buf, size_t len)
+{
+ int n;
+
+ assert(fd != iobuf.out_fd);
+
+ n = write(fd, buf, len);
+ if ((size_t)n == len)
+ return;
+ if (n < 0) {
+ if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) {
+ write_failed:
+ rsyserr(FERROR, errno,
+ "safe_write failed to write %ld bytes to %s [%s]",
+ (long)len, what_fd_is(fd), who_am_i());
+ exit_cleanup(RERR_STREAMIO);
+ }
+ } else {
+ buf += n;
+ len -= n;
+ }
+
+ while (len) {
+ struct timeval tv;
+ fd_set w_fds;
+ int cnt;
+
+ FD_ZERO(&w_fds);
+ FD_SET(fd, &w_fds);
+ tv.tv_sec = select_timeout;
+ tv.tv_usec = 0;
+
+ cnt = select(fd + 1, NULL, &w_fds, NULL, &tv);
+ if (cnt <= 0) {
+ if (cnt < 0 && errno == EBADF) {
+ rsyserr(FERROR, errno, "safe_write select failed on %s [%s]",
+ what_fd_is(fd), who_am_i());
+ exit_cleanup(RERR_FILEIO);
+ }
+ check_timeout();
+ continue;
+ }
+
+ if (FD_ISSET(fd, &w_fds)) {
+ n = write(fd, buf, len);
+ if (n < 0) {
+ if (errno == EINTR)
+ continue;
+ goto write_failed;
+ }
+ buf += n;
+ len -= n;
+ }
+ }
+}
+
+/* This is only called when files-from data is known to be available. We read
+ * a chunk of data and put it into the output buffer. */
+static void forward_filesfrom_data(void)
+{
+ int len;
+
+ len = read(ff_forward_fd, ff_xb.buf + ff_xb.len, ff_xb.size - ff_xb.len);
+ if (len <= 0) {
+ if (len == 0 || errno != EINTR) {
+ /* Send end-of-file marker */
+ ff_forward_fd = -1;
+ write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
+ free_xbuf(&ff_xb);
+ if (ff_reenable_multiplex >= 0)
+ io_start_multiplex_out(ff_reenable_multiplex);
+ }
+ return;
+ }
+
+ if (DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] files-from read=%ld\n", who_am_i(), (long)len);
+
+#ifdef ICONV_OPTION
+ len += ff_xb.len;
+#endif
+
+ if (!eol_nulls) {
+ char *s = ff_xb.buf + len;
+ /* Transform CR and/or LF into '\0' */
+ while (s-- > ff_xb.buf) {
+ if (*s == '\n' || *s == '\r')
+ *s = '\0';
+ }
+ }
+
+ if (ff_lastchar)
+ ff_xb.pos = 0;
+ else {
+ char *s = ff_xb.buf;
+ /* Last buf ended with a '\0', so don't let this buf start with one. */
+ while (len && *s == '\0')
+ s++, len--;
+ ff_xb.pos = s - ff_xb.buf;
+ }
+
+#ifdef ICONV_OPTION
+ if (filesfrom_convert && len) {
+ char *sob = ff_xb.buf + ff_xb.pos, *s = sob;
+ char *eob = sob + len;
+ int flags = ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT;
+ if (ff_lastchar == '\0')
+ flags |= ICB_INIT;
+ /* Convert/send each null-terminated string separately, skipping empties. */
+ while (s != eob) {
+ if (*s++ == '\0') {
+ ff_xb.len = s - sob - 1;
+ if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0)
+ exit_cleanup(RERR_PROTOCOL); /* impossible? */
+ write_buf(iobuf.out_fd, s-1, 1); /* Send the '\0'. */
+ while (s != eob && *s == '\0')
+ s++;
+ sob = s;
+ ff_xb.pos = sob - ff_xb.buf;
+ flags |= ICB_INIT;
+ }
+ }
+
+ if ((ff_xb.len = s - sob) == 0)
+ ff_lastchar = '\0';
+ else {
+ /* Handle a partial string specially, saving any incomplete chars. */
+ flags &= ~ICB_INCLUDE_INCOMPLETE;
+ if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0) {
+ if (errno == E2BIG)
+ exit_cleanup(RERR_PROTOCOL); /* impossible? */
+ if (ff_xb.pos)
+ memmove(ff_xb.buf, ff_xb.buf + ff_xb.pos, ff_xb.len);
+ }
+ ff_lastchar = 'x'; /* Anything non-zero. */
+ }
+ } else
+#endif
+
+ if (len) {
+ char *f = ff_xb.buf + ff_xb.pos;
+ char *t = ff_xb.buf;
+ char *eob = f + len;
+ /* Eliminate any multi-'\0' runs. */
+ while (f != eob) {
+ if (!(*t++ = *f++)) {
+ while (f != eob && *f == '\0')
+ f++;
+ }
+ }
+ ff_lastchar = f[-1];
+ if ((len = t - ff_xb.buf) != 0) {
+ /* This will not circle back to perform_io() because we only get
+ * called when there is plenty of room in the output buffer. */
+ write_buf(iobuf.out_fd, ff_xb.buf, len);
+ }
+ }
+}
+
+/* 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.
+ *
+ * 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.
+ *
+ * When writing, we flush data in the following priority order:
+ *
+ * 1. Finish writing any in-progress MSG_DATA sequence from iobuf.out.
+ *
+ * 2. Write out all the messages from the message buf (if iobuf.msg is active).
+ * Yes, this means that a PIO_NEED_OUTROOM call will completely flush any
+ * messages before getting to the iobuf.out flushing (except for rule 1).
+ *
+ * 3. Write out the raw data from iobuf.out, possibly filling in the multiplexed
+ * MSG_DATA header that was pre-allocated (when output is multiplexed).
+ *
+ * TODO: items for possible future work:
+ *
+ * - 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
+ * unused raw data in the buf would prevent the reading of socket data. */
+static char *perform_io(size_t needed, int flags)
+{
+ fd_set r_fds, e_fds, w_fds;
+ struct timeval tv;
+ int cnt, max_fd;
+ size_t empty_buf_len = 0;
+ xbuf *out;
+ char *data;
+
+ if (iobuf.in.len == 0 && iobuf.in.pos != 0) {
+ if (iobuf.raw_input_ends_before)
+ iobuf.raw_input_ends_before -= iobuf.in.pos;
+ iobuf.in.pos = 0;
+ }
+
+ switch (flags & PIO_NEED_FLAGS) {
+ case PIO_NEED_INPUT:
+ 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:
+ /* We never resize the circular output buffer. */
+ if (iobuf.out.size - iobuf.out_empty_len < needed) {
+ fprintf(stderr, "need to write %ld bytes, iobuf.out.buf is only %ld bytes.\n",
+ (long)needed, (long)(iobuf.out.size - iobuf.out_empty_len));
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ 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);
+ }
+ break;
+
+ case PIO_NEED_MSGROOM:
+ /* We never resize the circular message buffer. */
+ if (iobuf.msg.size < needed) {
+ fprintf(stderr, "need to write %ld bytes, iobuf.msg.buf is only %ld bytes.\n",
+ (long)needed, (long)iobuf.msg.size);
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ 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);
+ }
+ break;
+
+ case 0:
+ if (DEBUG_GTE(IO, 3))
+ rprintf(FINFO, "[%s] perform_io(%ld, %d)\n", who_am_i(), (long)needed, flags);
+ break;
+
+ default:
+ exit_cleanup(RERR_UNSUPPORTED);
+ }
+
+ while (1) {
+ switch (flags & PIO_NEED_FLAGS) {
+ case PIO_NEED_INPUT:
+ if (iobuf.in.len >= needed)
+ goto double_break;
+ break;
+ case PIO_NEED_OUTROOM:
+ if (iobuf.out.len <= iobuf.out.size - needed)
+ goto double_break;
+ break;
+ case PIO_NEED_MSGROOM:
+ if (iobuf.msg.len <= iobuf.msg.size - needed)
+ goto double_break;
+ break;
+ }
+
+ max_fd = -1;
+
+ FD_ZERO(&r_fds);
+ FD_ZERO(&e_fds);
+ if (iobuf.in_fd >= 0 && iobuf.in.size - (iobuf.in.pos + 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 > max_fd)
+ max_fd = iobuf.in_fd;
+ }
+
+ /* Only do more filesfrom processing if there is enough room in the out buffer. */
+ if (ff_forward_fd >= 0 && iobuf.out.size - iobuf.out.len > FILESFROM_BUFLEN*2) {
+ FD_SET(ff_forward_fd, &r_fds);
+ if (ff_forward_fd > max_fd)
+ max_fd = ff_forward_fd;
+ }
+
+ FD_ZERO(&w_fds);
+ if (iobuf.out_fd >= 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) {
+ /* 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",
+ who_am_i(), (int)MSG_DATA, (long)iobuf.out.len - 4);
+ }
+
+ /* reserve room for the next MSG_DATA header */
+ 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;
+ iobuf.out.len += 4;
+ }
+
+ empty_buf_len = iobuf.out_empty_len;
+ out = &iobuf.out;
+ } else if (iobuf.msg.len) {
+ empty_buf_len = 0;
+ out = &iobuf.msg;
+ } else
+ out = NULL;
+ if (out) {
+ FD_SET(iobuf.out_fd, &w_fds);
+ if (iobuf.out_fd > max_fd)
+ max_fd = iobuf.out_fd;
+ }
+ } else
+ out = NULL;
+
+ if (max_fd < 0) {
+ switch (flags & PIO_NEED_FLAGS) {
+ case PIO_NEED_INPUT:
+ 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;
+ rprintf(FERROR, "error in perform_io: no fd for output.\n");
+ exit_cleanup(RERR_PROTOCOL);
+ default:
+ /* No stated needs, so I guess this is OK. */
+ break;
+ }
+ break;
+ }
+
+ if (extra_flist_sending_enabled) {
+ if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD
+ && file_total - file_old_total >= MIN_FILECNT_LOOKAHEAD)
+ tv.tv_sec = 0;
+ else {
+ extra_flist_sending_enabled = False;
+ tv.tv_sec = select_timeout;
+ }
+ } else
+ tv.tv_sec = select_timeout;
+ tv.tv_usec = 0;
+
+ cnt = select(max_fd + 1, &r_fds, &w_fds, &e_fds, &tv);
+
+ if (cnt <= 0) {
+ if (cnt < 0 && errno == EBADF) {
+ msgs2stderr = 1;
+ exit_cleanup(RERR_SOCKETIO);
+ }
+ if (extra_flist_sending_enabled) {
+ extra_flist_sending_enabled = False;
+ send_extra_file_list(sock_f_out, -1);
+ extra_flist_sending_enabled = !flist_eof;
+ } else
+ check_timeout();
+ FD_ZERO(&r_fds); /* Just in case... */
+ FD_ZERO(&w_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;
+ int n;
+ if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
+ if (n == 0) {
+ if (!read_batch || batch_fd < 0 || am_generator)
+ whine_about_eof(iobuf.in_fd); /* Doesn't return. */
+ batch_fd = -1;
+ continue;
+ }
+ if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)
+ n = 0;
+ else {
+ /* Don't write errors on a dead socket. */
+ if (iobuf.in_fd == sock_f_in) {
+ if (am_sender)
+ msgs2stderr = 1;
+ rsyserr(FERROR_SOCKET, errno, "read error");
+ } else
+ rsyserr(FERROR, errno, "read error");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ }
+ if (msgs2stderr && DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] recv=%ld\n", who_am_i(), (long)n);
+
+ if (io_timeout)
+ last_io_in = time(NULL);
+ stats.total_read += n;
+
+ iobuf.in.len += n;
+ }
+
+ if (iobuf.out_fd >= 0 && 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;
+
+ if (bwlimit_writemax && len > bwlimit_writemax)
+ len = bwlimit_writemax;
+
+ if (out->pos + len > out->size)
+ len = out->size - out->pos;
+ if ((n = write(iobuf.out_fd, out->buf + out->pos, len)) <= 0) {
+ if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)
+ n = 0;
+ 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, "[%s] write error", who_am_i());
+ exit_cleanup(RERR_STREAMIO);
+ }
+ }
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] %s sent=%ld\n",
+ who_am_i(), out == &iobuf.out ? "out" : "msg", (long)n);
+ }
+
+ if (io_timeout)
+ last_io_out = time(NULL);
+ stats.total_written += n;
+
+ if (bwlimit_writemax)
+ sleep_for_bwlimit(n);
+
+ if ((out->pos += n) == out->size) {
+ if (iobuf.raw_flushing_ends_before)
+ iobuf.raw_flushing_ends_before -= out->size;
+ out->pos = 0;
+ }
+ if (out->pos == iobuf.raw_flushing_ends_before)
+ iobuf.raw_flushing_ends_before = 0;
+ if ((out->len -= n) == empty_buf_len) {
+ out->pos = 0;
+ if (empty_buf_len)
+ iobuf.raw_data_header_pos = 0;
+ }
+ }
+
+ 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
+ * and be sure to not cache anything that would break
+ * such a change. */
+ forward_filesfrom_data();
+ }
+ }
+ double_break:
+
+ data = iobuf.in.buf + iobuf.in.pos;
+
+ if (flags & PIO_CONSUME_INPUT) {
+ iobuf.in.len -= needed;
+ iobuf.in.pos += needed;
+ }
+
+ return data;
+}
+
+void noop_io_until_death(void)
+{
+ char buf[1024];
+
+ kluge_around_eof = 1;
+ set_io_timeout(protocol_version >= 31 ? 10 : 1);
+
+ 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;
+ BOOL want_debug = DEBUG_GTE(IO, 1) && convert >= 0 && (msgs2stderr || code != MSG_INFO);
+
+ if (!OUT_MULTIPLEXED)
+ return 0;
+
+ if (want_debug)
+ rprintf(FINFO, "[%s] send_msg(%d, %ld)\n", who_am_i(), (int)code, (long)len);
+
+#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);
+ } else
+#endif
+ if (iobuf.msg.len + len + 4 > iobuf.msg.size)
+ perform_io(len + 4, 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;
+ hdr = iobuf.msg.buf + pos;
+
+ iobuf.msg.len += 4; /* Leave room for the coming header bytes. */
+
+#ifdef ICONV_OPTION
+ if (convert > 0) {
+ xbuf inbuf;
+
+ INIT_XBUF(inbuf, (char*)buf, len, (size_t)-1);
+
+ len = iobuf.msg.len;
+ iconvbufs(ic_send, &inbuf, &iobuf.msg,
+ ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT | ICB_INIT);
+ if (inbuf.len > 0) {
+ rprintf(FERROR, "overflowed iobuf.msg buffer in send_msg");
+ exit_cleanup(RERR_UNSUPPORTED);
+ }
+ len = iobuf.msg.len - len;
+ } else
+#endif
+ {
+ size_t siz;
+
+ if ((pos += 4) >= iobuf.msg.size)
+ pos -= iobuf.msg.size;