+ 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;
+ 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 not used for (normal) MSG_DATA. */
+int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
+{
+ char *hdr;
+ size_t needed, 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);
+
+ /* 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 maximal conversion expansion. */
+ needed = len*2 + 4 + 3;
+ } else
+#endif
+ 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; /* Allocate 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);