+ INIT_CONST_XBUF(outbuf, line);
+ INIT_XBUF(inbuf, ibuf, 0, (size_t)-1);
+
+ while (msg_bytes) {
+ inbuf.len = msg_bytes > sizeof ibuf ? sizeof ibuf : msg_bytes;
+ memcpy(inbuf.buf, perform_io(inbuf.len, PIO_INPUT_AND_CONSUME), inbuf.len);
+ if (!(msg_bytes -= inbuf.len)
+ && !ibuf[inbuf.len-1])
+ inbuf.len--, add_null = 1;
+ if (iconvbufs(ic_send, &inbuf, &outbuf,
+ ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE) < 0)
+ goto overflow;
+ }
+ if (add_null) {
+ if (outbuf.len == outbuf.size)
+ goto overflow;
+ outbuf.buf[outbuf.len++] = '\0';
+ }
+ msg_bytes = outbuf.len;
+ } else
+#endif
+ memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), msg_bytes);
+ /* A directory name was sent with the trailing null */
+ if (msg_bytes > 0 && !line[msg_bytes-1])
+ log_delete(line, S_IFDIR);
+ else {
+ line[msg_bytes] = '\0';
+ log_delete(line, S_IFREG);
+ }
+ break;
+ case MSG_SUCCESS:
+ if (msg_bytes != 4) {
+ invalid_msg:
+ rprintf(FERROR, "invalid multi-message %d:%lu [%s%s]\n",
+ tag, (unsigned long)msg_bytes, who_am_i(),
+ inc_recurse ? "/inc" : "");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ data = perform_io(4, PIO_INPUT_AND_CONSUME);
+ if (am_generator)
+ got_flist_entry_status(FES_SUCCESS, IVAL(data, 0));
+ else
+ successful_send(IVAL(data, 0));
+ break;
+ case MSG_NO_SEND:
+ if (msg_bytes != 4)
+ goto invalid_msg;
+ data = perform_io(4, PIO_INPUT_AND_CONSUME);
+ if (am_generator)
+ got_flist_entry_status(FES_NO_SEND, IVAL(data, 0));
+ else
+ send_msg(MSG_NO_SEND, data, 4, 0);
+ break;
+ case MSG_ERROR_SOCKET:
+ case MSG_ERROR_UTF8:
+ case MSG_CLIENT:
+ case MSG_LOG:
+ if (!am_generator)
+ goto invalid_msg;
+ if (tag == MSG_ERROR_SOCKET)
+ msgs2stderr = 1;
+ /* FALL THROUGH */
+ case MSG_INFO:
+ case MSG_ERROR:
+ case MSG_ERROR_XFER:
+ case MSG_WARNING:
+ if (msg_bytes >= sizeof line) {
+ overflow:
+ rprintf(FERROR,
+ "multiplexing overflow %d:%lu [%s%s]\n",
+ tag, (unsigned long)msg_bytes, who_am_i(),
+ 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);
+ if (first_message) {
+ if (list_only && !am_sender && tag == 1 && msg_bytes < sizeof line) {
+ line[msg_bytes] = '\0';
+ check_for_d_option_error(line);
+ }
+ first_message = 0;
+ }
+ break;
+ default:
+ rprintf(FERROR, "unexpected tag %d [%s%s]\n",
+ tag, who_am_i(), inc_recurse ? "/inc" : "");
+ exit_cleanup(RERR_STREAMIO);
+ }
+}
+
+void wait_for_receiver(void)
+{
+ if (!iobuf.raw_input_ends_before)
+ read_a_msg();
+
+ if (iobuf.raw_input_ends_before) {
+ int ndx = read_int(iobuf.in_fd);
+ if (ndx < 0) {
+ switch (ndx) {
+ case NDX_FLIST_EOF:
+ flist_eof = 1;
+ if (DEBUG_GTE(FLIST, 3))
+ rprintf(FINFO, "[%s] flist_eof=1\n", who_am_i());
+ break;
+ case NDX_DONE:
+ msgdone_cnt++;
+ break;
+ default:
+ exit_cleanup(RERR_STREAMIO);
+ }
+ } else {
+ struct file_list *flist;
+ if (DEBUG_GTE(FLIST, 2)) {
+ rprintf(FINFO, "[%s] receiving flist for dir %d\n",
+ who_am_i(), ndx);
+ }
+ flist = recv_file_list(iobuf.in_fd);
+ flist->parent_ndx = ndx;
+#ifdef SUPPORT_HARD_LINKS
+ if (preserve_hard_links)
+ match_hard_links(flist);
+#endif
+ }
+ }
+}
+
+unsigned short read_shortint(int f)
+{
+ char b[2];
+ read_buf(f, b, 2);
+ return (UVAL(b, 1) << 8) + UVAL(b, 0);
+}
+
+int32 read_int(int f)
+{
+ char b[4];
+ int32 num;
+
+ read_buf(f, b, 4);
+ num = IVAL(b, 0);
+#if SIZEOF_INT32 > 4
+ if (num & (int32)0x80000000)
+ num |= ~(int32)0xffffffff;
+#endif
+ return num;
+}
+
+int32 read_varint(int f)
+{
+ union {
+ char b[5];
+ int32 x;
+ } u;
+ uchar ch;
+ int extra;
+
+ u.x = 0;
+ ch = read_byte(f);
+ extra = int_byte_extra[ch / 4];
+ if (extra) {
+ uchar bit = ((uchar)1<<(8-extra));
+ if (extra >= (int)sizeof u.b) {
+ rprintf(FERROR, "Overflow in read_varint()\n");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ read_buf(f, u.b, extra);
+ u.b[extra] = ch & (bit-1);
+ } else
+ u.b[0] = ch;
+#if CAREFUL_ALIGNMENT
+ u.x = IVAL(u.b,0);
+#endif
+#if SIZEOF_INT32 > 4
+ if (u.x & (int32)0x80000000)
+ u.x |= ~(int32)0xffffffff;
+#endif
+ return u.x;
+}
+
+int64 read_varlong(int f, uchar min_bytes)
+{
+ union {
+ char b[9];
+ int64 x;
+ } u;
+ char b2[8];
+ int extra;
+
+#if SIZEOF_INT64 < 8
+ memset(u.b, 0, 8);
+#else
+ u.x = 0;
+#endif
+ read_buf(f, b2, min_bytes);
+ memcpy(u.b, b2+1, min_bytes-1);
+ extra = int_byte_extra[CVAL(b2, 0) / 4];
+ if (extra) {
+ uchar bit = ((uchar)1<<(8-extra));
+ if (min_bytes + extra > (int)sizeof u.b) {
+ rprintf(FERROR, "Overflow in read_varlong()\n");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ read_buf(f, u.b + min_bytes - 1, extra);
+ u.b[min_bytes + extra - 1] = CVAL(b2, 0) & (bit-1);
+#if SIZEOF_INT64 < 8
+ if (min_bytes + extra > 5 || u.b[4] || CVAL(u.b,3) & 0x80) {
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+ }
+#endif
+ } else
+ u.b[min_bytes + extra - 1] = CVAL(b2, 0);
+#if SIZEOF_INT64 < 8
+ u.x = IVAL(u.b,0);
+#elif CAREFUL_ALIGNMENT
+ u.x = IVAL(u.b,0) | (((int64)IVAL(u.b,4))<<32);
+#endif
+ return u.x;
+}
+
+int64 read_longint(int f)
+{
+#if SIZEOF_INT64 >= 8
+ char b[9];
+#endif
+ int32 num = read_int(f);
+
+ if (num != (int32)0xffffffff)
+ return num;
+
+#if SIZEOF_INT64 < 8
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+#else
+ read_buf(f, b, 8);
+ return IVAL(b,0) | (((int64)IVAL(b,4))<<32);
+#endif
+}
+
+void read_buf(int f, char *buf, size_t len)
+{
+ if (f != iobuf.in_fd) {
+ if (safe_read(f, buf, len) != len)
+ whine_about_eof(f); /* Doesn't return. */
+ goto batch_copy;
+ }
+
+ if (!IN_MULTIPLEXED) {
+ memcpy(buf, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+ if (forward_flist_data)
+ write_buf(iobuf.out_fd, buf, len);
+ batch_copy:
+ if (f == write_batch_monitor_in)
+ safe_write(batch_fd, buf, len);
+ return;
+ }
+
+ 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 (forward_flist_data)
+ write_buf(iobuf.out_fd, buf, siz);
+
+ if (f == write_batch_monitor_in)
+ safe_write(batch_fd, buf, siz);
+
+ if ((len -= siz) == 0)
+ break;
+ buf += siz;
+ }
+}
+
+void read_sbuf(int f, char *buf, size_t len)
+{
+ read_buf(f, buf, len);
+ buf[len] = '\0';
+}
+
+uchar read_byte(int f)
+{
+ uchar c;
+ read_buf(f, (char*)&c, 1);
+ return c;
+}
+
+int read_vstring(int f, char *buf, int bufsize)
+{
+ int len = read_byte(f);
+
+ if (len & 0x80)
+ len = (len & ~0x80) * 0x100 + read_byte(f);
+
+ if (len >= bufsize) {
+ rprintf(FERROR, "over-long vstring received (%d > %d)\n",
+ len, bufsize - 1);
+ return -1;
+ }
+
+ if (len)
+ read_buf(f, buf, len);
+ buf[len] = '\0';
+ return len;
+}
+
+/* Populate a sum_struct with values from the socket. This is
+ * called by both the sender and the receiver. */
+void read_sum_head(int f, struct sum_struct *sum)
+{
+ int32 max_blength = protocol_version < 30 ? OLD_MAX_BLOCK_SIZE : MAX_BLOCK_SIZE;
+ sum->count = read_int(f);
+ if (sum->count < 0) {
+ rprintf(FERROR, "Invalid checksum count %ld [%s]\n",
+ (long)sum->count, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ sum->blength = read_int(f);
+ if (sum->blength < 0 || sum->blength > max_blength) {
+ rprintf(FERROR, "Invalid block length %ld [%s]\n",
+ (long)sum->blength, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ sum->s2length = protocol_version < 27 ? csum_length : (int)read_int(f);
+ if (sum->s2length < 0 || sum->s2length > MAX_DIGEST_LEN) {
+ rprintf(FERROR, "Invalid checksum length %d [%s]\n",
+ sum->s2length, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ sum->remainder = read_int(f);
+ if (sum->remainder < 0 || sum->remainder > sum->blength) {
+ rprintf(FERROR, "Invalid remainder length %ld [%s]\n",
+ (long)sum->remainder, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+}
+
+/* Send the values from a sum_struct over the socket. Set sum to
+ * NULL if there are no checksums to send. This is called by both
+ * the generator and the sender. */
+void write_sum_head(int f, struct sum_struct *sum)
+{
+ static struct sum_struct null_sum;
+
+ if (sum == NULL)
+ sum = &null_sum;
+
+ write_int(f, sum->count);
+ write_int(f, sum->blength);
+ if (protocol_version >= 27)
+ write_int(f, sum->s2length);
+ write_int(f, sum->remainder);
+}
+
+/* Sleep after writing to limit I/O bandwidth usage.
+ *
+ * @todo Rather than sleeping after each write, it might be better to
+ * use some kind of averaging. The current algorithm seems to always
+ * use a bit less bandwidth than specified, because it doesn't make up
+ * for slow periods. But arguably this is a feature. In addition, we
+ * ought to take the time used to write the data into account.
+ *
+ * During some phases of big transfers (file FOO is uptodate) this is
+ * called with a small bytes_written every time. As the kernel has to
+ * round small waits up to guarantee that we actually wait at least the
+ * requested number of microseconds, this can become grossly inaccurate.
+ * We therefore keep track of the bytes we've written over time and only
+ * sleep when the accumulated delay is at least 1 tenth of a second. */
+static void sleep_for_bwlimit(int bytes_written)
+{
+ static struct timeval prior_tv;
+ static long total_written = 0;
+ struct timeval tv, start_tv;
+ long elapsed_usec, sleep_usec;
+
+#define ONE_SEC 1000000L /* # of microseconds in a second */
+
+ total_written += bytes_written;
+
+ gettimeofday(&start_tv, NULL);
+ if (prior_tv.tv_sec) {
+ elapsed_usec = (start_tv.tv_sec - prior_tv.tv_sec) * ONE_SEC
+ + (start_tv.tv_usec - prior_tv.tv_usec);
+ total_written -= elapsed_usec * bwlimit / (ONE_SEC/1024);
+ if (total_written < 0)
+ total_written = 0;
+ }
+
+ sleep_usec = total_written * (ONE_SEC/1024) / bwlimit;
+ if (sleep_usec < ONE_SEC / 10) {
+ prior_tv = start_tv;
+ return;
+ }
+
+ tv.tv_sec = sleep_usec / ONE_SEC;
+ tv.tv_usec = sleep_usec % ONE_SEC;
+ select(0, NULL, NULL, NULL, &tv);
+
+ gettimeofday(&prior_tv, NULL);
+ elapsed_usec = (prior_tv.tv_sec - start_tv.tv_sec) * ONE_SEC
+ + (prior_tv.tv_usec - start_tv.tv_usec);
+ total_written = (sleep_usec - elapsed_usec) * bwlimit / (ONE_SEC/1024);
+}
+
+void io_flush(int flush_it_all)
+{
+ if (iobuf.out.len > iobuf.out_empty_len) {
+ if (flush_it_all) /* FULL_FLUSH: flush everything in the output buffers */
+ perform_io(iobuf.out.size - iobuf.out_empty_len, PIO_NEED_OUTROOM);
+ else /* NORMAL_FLUSH: flush at least 1 byte */
+ perform_io(iobuf.out.size - iobuf.out.len + 1, PIO_NEED_OUTROOM);
+ }
+ if (iobuf.msg.len)
+ perform_io(iobuf.msg.size, PIO_NEED_MSGROOM);
+}
+
+void write_shortint(int f, unsigned short x)
+{
+ char b[2];
+ b[0] = (char)x;
+ b[1] = (char)(x >> 8);
+ write_buf(f, b, 2);
+}
+
+void write_int(int f, int32 x)
+{
+ char b[4];
+ SIVAL(b, 0, x);
+ write_buf(f, b, 4);
+}
+
+void write_varint(int f, int32 x)
+{
+ char b[5];
+ uchar bit;
+ int cnt = 4;
+
+ SIVAL(b, 1, x);
+
+ while (cnt > 1 && b[cnt] == 0)
+ cnt--;
+ bit = ((uchar)1<<(7-cnt+1));
+ if (CVAL(b, cnt) >= bit) {
+ cnt++;
+ *b = ~(bit-1);
+ } else if (cnt > 1)
+ *b = b[cnt] | ~(bit*2-1);
+ else
+ *b = b[cnt];
+
+ write_buf(f, b, cnt);
+}
+
+void write_varlong(int f, int64 x, uchar min_bytes)
+{
+ char b[9];
+ uchar bit;
+ int cnt = 8;
+
+ SIVAL(b, 1, x);
+#if SIZEOF_INT64 >= 8
+ SIVAL(b, 5, x >> 32);
+#else
+ if (x <= 0x7FFFFFFF && x >= 0)
+ memset(b + 5, 0, 4);
+ else {
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+ }
+#endif
+
+ while (cnt > min_bytes && b[cnt] == 0)
+ cnt--;
+ bit = ((uchar)1<<(7-cnt+min_bytes));
+ if (CVAL(b, cnt) >= bit) {
+ cnt++;
+ *b = ~(bit-1);
+ } else if (cnt > min_bytes)
+ *b = b[cnt] | ~(bit*2-1);
+ else
+ *b = b[cnt];
+
+ write_buf(f, b, cnt);
+}
+
+/*
+ * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
+ * 64-bit types on this platform.
+ */
+void write_longint(int f, int64 x)
+{
+ char b[12], * const s = b+4;
+
+ SIVAL(s, 0, x);
+ if (x <= 0x7FFFFFFF && x >= 0) {
+ write_buf(f, s, 4);
+ return;
+ }
+
+#if SIZEOF_INT64 < 8
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+#else
+ memset(b, 0xFF, 4);
+ SIVAL(s, 4, x >> 32);
+ write_buf(f, b, 12);
+#endif
+}
+
+void write_buf(int f, const char *buf, size_t len)
+{
+ size_t pos, siz;
+
+ if (f != iobuf.out_fd) {
+ safe_write(f, buf, len);
+ goto batch_copy;
+ }
+
+ if (iobuf.out.size - iobuf.out.len < len)
+ perform_io(len, PIO_NEED_OUTROOM);
+
+ pos = iobuf.out.pos + iobuf.out.len; /* Must be set after any flushing. */
+ if (pos >= iobuf.out.size)
+ pos -= iobuf.out.size;
+
+ /* Handle a split copy if we wrap around the end of the circular buffer. */
+ if (pos >= iobuf.out.pos && (siz = iobuf.out.size - pos) < len) {
+ memcpy(iobuf.out.buf + pos, buf, siz);
+ memcpy(iobuf.out.buf, buf + siz, len - siz);
+ } else
+ memcpy(iobuf.out.buf + pos, buf, len);
+
+ iobuf.out.len += len;
+
+ batch_copy:
+ if (f == write_batch_monitor_out)
+ safe_write(batch_fd, buf, len);
+}
+
+/* Write a string to the connection */
+void write_sbuf(int f, const char *buf)
+{
+ write_buf(f, buf, strlen(buf));
+}
+
+void write_byte(int f, uchar c)
+{
+ write_buf(f, (char *)&c, 1);
+}
+
+void write_vstring(int f, const char *str, int len)
+{
+ uchar lenbuf[3], *lb = lenbuf;
+
+ if (len > 0x7F) {
+ if (len > 0x7FFF) {
+ rprintf(FERROR,
+ "attempting to send over-long vstring (%d > %d)\n",
+ len, 0x7FFF);
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ *lb++ = len / 0x100 + 0x80;
+ }
+ *lb = len;
+
+ write_buf(f, (char*)lenbuf, lb - lenbuf + 1);
+ if (len)
+ write_buf(f, str, len);
+}
+
+/* Send a file-list index using a byte-reduction method. */
+void write_ndx(int f, int32 ndx)
+{
+ static int32 prev_positive = -1, prev_negative = 1;
+ int32 diff, cnt = 0;
+ char b[6];
+
+ if (protocol_version < 30 || read_batch) {
+ write_int(f, ndx);
+ return;
+ }
+
+ /* Send NDX_DONE as a single-byte 0 with no side effects. Send
+ * negative nums as a positive after sending a leading 0xFF. */
+ if (ndx >= 0) {
+ diff = ndx - prev_positive;
+ prev_positive = ndx;
+ } else if (ndx == NDX_DONE) {
+ *b = 0;
+ write_buf(f, b, 1);
+ return;
+ } else {
+ b[cnt++] = (char)0xFF;
+ ndx = -ndx;
+ diff = ndx - prev_negative;
+ prev_negative = ndx;
+ }
+
+ /* A diff of 1 - 253 is sent as a one-byte diff; a diff of 254 - 32767
+ * or 0 is sent as a 0xFE + a two-byte diff; otherwise we send 0xFE
+ * & all 4 bytes of the (non-negative) num with the high-bit set. */
+ if (diff < 0xFE && diff > 0)
+ b[cnt++] = (char)diff;
+ else if (diff < 0 || diff > 0x7FFF) {
+ b[cnt++] = (char)0xFE;
+ b[cnt++] = (char)((ndx >> 24) | 0x80);
+ b[cnt++] = (char)ndx;
+ b[cnt++] = (char)(ndx >> 8);
+ b[cnt++] = (char)(ndx >> 16);
+ } else {
+ b[cnt++] = (char)0xFE;
+ b[cnt++] = (char)(diff >> 8);
+ b[cnt++] = (char)diff;
+ }
+ write_buf(f, b, cnt);
+}
+
+/* Receive a file-list index using a byte-reduction method. */
+int32 read_ndx(int f)
+{
+ static int32 prev_positive = -1, prev_negative = 1;
+ int32 *prev_ptr, num;
+ char b[4];
+
+ if (protocol_version < 30)
+ return read_int(f);
+
+ read_buf(f, b, 1);
+ if (CVAL(b, 0) == 0xFF) {
+ read_buf(f, b, 1);
+ prev_ptr = &prev_negative;
+ } else if (CVAL(b, 0) == 0)
+ return NDX_DONE;
+ else
+ prev_ptr = &prev_positive;
+ if (CVAL(b, 0) == 0xFE) {
+ read_buf(f, b, 2);
+ if (CVAL(b, 0) & 0x80) {
+ b[3] = CVAL(b, 0) & ~0x80;
+ b[0] = b[1];
+ read_buf(f, b+1, 2);
+ num = IVAL(b, 0);
+ } else
+ num = (UVAL(b,0)<<8) + UVAL(b,1) + *prev_ptr;
+ } else
+ num = UVAL(b, 0) + *prev_ptr;
+ *prev_ptr = num;
+ if (prev_ptr == &prev_negative)
+ num = -num;
+ return num;
+}
+
+/* Read a line of up to bufsiz-1 characters into buf. Strips
+ * the (required) trailing newline and all carriage returns.
+ * Returns 1 for success; 0 for I/O error or truncation. */
+int read_line_old(int fd, char *buf, size_t bufsiz)
+{
+ bufsiz--; /* leave room for the null */
+ while (bufsiz > 0) {
+ assert(fd != iobuf.in_fd);
+ if (safe_read(fd, buf, 1) == 0)
+ return 0;
+ if (*buf == '\0')
+ return 0;
+ if (*buf == '\n')
+ break;
+ if (*buf != '\r') {
+ buf++;
+ bufsiz--;
+ }
+ }
+ *buf = '\0';
+ return bufsiz > 0;
+}
+
+void io_printf(int fd, const char *format, ...)
+{
+ va_list ap;
+ char buf[BIGPATHBUFLEN];
+ int len;
+
+ va_start(ap, format);
+ len = vsnprintf(buf, sizeof buf, format, ap);
+ va_end(ap);
+
+ if (len < 0)
+ exit_cleanup(RERR_STREAMIO);
+
+ if (len > (int)sizeof buf) {
+ rprintf(FERROR, "io_printf() was too long for the buffer.\n");
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ write_sbuf(fd, buf);
+}
+
+/* Setup for multiplexing a MSG_* stream with the data stream. */
+void io_start_multiplex_out(int fd)
+{
+ if (msgs2stderr && DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] io_start_multiplex_out(%d)\n", who_am_i(), fd);
+
+ io_flush(FULL_FLUSH);
+
+ iobuf.out_empty_len = 4; /* See also OUT_MULTIPLEXED */
+ io_start_buffering_out(fd);
+
+ iobuf.raw_data_header_pos = iobuf.out.pos + iobuf.out.len;
+ iobuf.out.len += 4;
+}
+
+/* Setup for multiplexing a MSG_* stream with the data stream. */
+void io_start_multiplex_in(int fd)
+{
+ 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 */
+ io_start_buffering_in(fd);
+}
+
+void io_end_multiplex_in(BOOL free_buffers)
+{
+ if (DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_multiplex_in(%s)\n",
+ who_am_i(), free_buffers ? "True" : "False");
+ }
+
+ iobuf.in_multiplexed = False;
+ iobuf.raw_input_ends_before = 0;
+ io_end_buffering_in(free_buffers);
+}
+
+/* Stop output multiplexing. */
+void io_end_multiplex_out(BOOL free_buffers)
+{
+ if (DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_multiplex_out(%s)\n",
+ who_am_i(), free_buffers ? "True" : "False");
+ }
+
+ io_end_buffering_out(free_buffers);
+ iobuf.out_empty_len = 0;
+}
+
+void start_write_batch(int fd)
+{
+ /* Some communication has already taken place, but we don't
+ * enable batch writing until here so that we can write a
+ * canonical record of the communication even though the
+ * actual communication so far depends on whether a daemon
+ * is involved. */
+ write_int(batch_fd, protocol_version);
+ if (protocol_version >= 30)
+ write_byte(batch_fd, inc_recurse);
+ write_int(batch_fd, checksum_seed);
+
+ if (am_sender)
+ write_batch_monitor_out = fd;
+ else
+ write_batch_monitor_in = fd;
+}
+
+void stop_write_batch(void)
+{
+ write_batch_monitor_out = -1;
+ write_batch_monitor_in = -1;
+}