+ 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((flags & PIO_NEED_INPUT) != 0);
+ 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 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)
+ iobuf.in_fd = -2;
+ 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_SOCKETIO);
+ }
+ }
+ 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);
+ if (flags & PIO_NEED_INPUT)
+ maybe_send_keepalive(last_io_in, 0);
+ }
+ stats.total_read += n;
+
+ iobuf.in.len += n;
+ }
+
+ if (out && 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;
+ 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)) {
+ 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;
+ 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_AND_READY && !(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
+ * 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);
+ }
+ len = iobuf.msg.len - len;
+ } else
+#endif
+ {
+ size_t siz;
+
+ if ((pos += 4) == iobuf.msg.size)
+ pos = 0;
+
+ /* Handle a split copy if we wrap around the end of the circular buffer. */
+ if (pos >= iobuf.msg.pos && (siz = iobuf.msg.size - pos) < len) {
+ memcpy(iobuf.msg.buf + pos, buf, siz);
+ memcpy(iobuf.msg.buf, buf + siz, len - siz);
+ } else
+ memcpy(iobuf.msg.buf + pos, buf, len);
+
+ iobuf.msg.len += len;
+ }
+
+ SIVAL(hdr, 0, ((MPLEX_BASE + (int)code)<<24) + len);
+
+ if (want_debug && convert > 0)
+ rprintf(FINFO, "[%s] converted msg len=%ld\n", who_am_i(), (long)len);
+
+ return 1;
+}
+
+void send_msg_int(enum msgcode code, int num)
+{
+ char numbuf[4];
+
+ if (DEBUG_GTE(IO, 1))
+ rprintf(FINFO, "[%s] send_msg_int(%d, %d)\n", who_am_i(), (int)code, num);
+
+ SIVAL(numbuf, 0, num);
+ send_msg(code, numbuf, 4, -1);
+}
+
+static void got_flist_entry_status(enum festatus status, int ndx)
+{
+ struct file_list *flist = flist_for_ndx(ndx, "got_flist_entry_status");
+
+ if (remove_source_files) {
+ active_filecnt--;
+ active_bytecnt -= F_LENGTH(flist->files[ndx - flist->ndx_start]);
+ }
+
+ if (inc_recurse)
+ flist->in_progress--;
+
+ switch (status) {
+ case FES_SUCCESS:
+ if (remove_source_files)
+ send_msg_int(MSG_SUCCESS, ndx);
+ if (preserve_hard_links) {
+ struct file_struct *file = flist->files[ndx - flist->ndx_start];
+ if (F_IS_HLINKED(file)) {
+ flist_ndx_push(&hlink_list, ndx);
+ flist->in_progress++;
+ }
+ }
+ break;
+ case FES_REDO:
+ if (read_batch) {
+ if (inc_recurse)
+ flist->in_progress++;
+ break;
+ }
+ if (inc_recurse)
+ flist->to_redo++;
+ flist_ndx_push(&redo_list, ndx);
+ break;
+ case FES_NO_SEND:
+ break;
+ }
+}
+
+/* Note the fds used for the main socket (which might really be a pipe
+ * for a local transfer, but we can ignore that). */
+void io_set_sock_fds(int f_in, int f_out)
+{
+ sock_f_in = f_in;
+ sock_f_out = f_out;
+}
+
+void set_io_timeout(int secs)
+{
+ io_timeout = secs;
+ allowed_lull = (io_timeout + 1) / 2;
+
+ if (!io_timeout || allowed_lull > SELECT_TIMEOUT)
+ select_timeout = SELECT_TIMEOUT;
+ else
+ select_timeout = allowed_lull;
+
+ if (read_batch)
+ allowed_lull = 0;
+}
+
+static void check_for_d_option_error(const char *msg)
+{
+ static char rsync263_opts[] = "BCDHIKLPRSTWabceghlnopqrtuvxz";
+ char *colon;
+ int saw_d = 0;
+
+ if (*msg != 'r'
+ || strncmp(msg, REMOTE_OPTION_ERROR, sizeof REMOTE_OPTION_ERROR - 1) != 0)
+ return;
+
+ msg += sizeof REMOTE_OPTION_ERROR - 1;
+ if (*msg == '-' || (colon = strchr(msg, ':')) == NULL
+ || strncmp(colon, REMOTE_OPTION_ERROR2, sizeof REMOTE_OPTION_ERROR2 - 1) != 0)
+ return;
+
+ for ( ; *msg != ':'; msg++) {
+ if (*msg == 'd')
+ saw_d = 1;
+ else if (*msg == 'e')
+ break;
+ else if (strchr(rsync263_opts, *msg) == NULL)
+ return;
+ }
+
+ if (saw_d) {
+ rprintf(FWARNING,
+ "*** Try using \"--old-d\" if remote rsync is <= 2.6.3 ***\n");
+ }
+}
+
+/* This is used by the generator to limit how many file transfers can
+ * be active at once when --remove-source-files is specified. Without
+ * this, sender-side deletions were mostly happening at the end. */
+void increment_active_files(int ndx, int itemizing, enum logcode code)
+{
+ while (1) {
+ /* TODO: tune these limits? */
+ int limit = active_bytecnt >= 128*1024 ? 10 : 50;
+ if (active_filecnt < limit)
+ break;
+ check_for_finished_files(itemizing, code, 0);
+ if (active_filecnt < limit)
+ break;
+ wait_for_receiver();
+ }
+
+ active_filecnt++;
+ active_bytecnt += F_LENGTH(cur_flist->files[ndx - cur_flist->ndx_start]);
+}
+
+int get_redo_num(void)
+{
+ return flist_ndx_pop(&redo_list);
+}
+
+int get_hlink_num(void)
+{
+ return flist_ndx_pop(&hlink_list);
+}
+
+/* When we're the receiver and we have a local --files-from list of names
+ * that needs to be sent over the socket to the sender, we have to do two
+ * things at the same time: send the sender a list of what files we're
+ * processing and read the incoming file+info list from the sender. We do
+ * this by making recv_file_list() call forward_filesfrom_data(), which
+ * will ensure that we forward data to the sender until we get some data
+ * for recv_file_list() to use. */
+void start_filesfrom_forwarding(int fd)
+{
+ 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. */
+ ff_reenable_multiplex = io_end_multiplex_out(MPLX_TO_BUFFERED);
+ }
+ ff_forward_fd = fd;
+
+ alloc_xbuf(&ff_xb, FILESFROM_BUFLEN);
+}
+
+/* Read a line into the "buf" buffer. */
+int read_line(int fd, char *buf, size_t bufsiz, int flags)
+{
+ char ch, *s, *eob;
+
+#ifdef ICONV_OPTION
+ if (flags & RL_CONVERT && iconv_buf.size < bufsiz)
+ realloc_xbuf(&iconv_buf, bufsiz + 1024);
+#endif
+
+ start:
+#ifdef ICONV_OPTION
+ s = flags & RL_CONVERT ? iconv_buf.buf : buf;
+#else
+ s = buf;
+#endif
+ eob = s + bufsiz - 1;
+ while (1) {
+ /* We avoid read_byte() for files because files can return an EOF. */
+ if (fd == iobuf.in_fd)
+ ch = read_byte(fd);
+ else if (safe_read(fd, &ch, 1) == 0)
+ break;
+ if (flags & RL_EOL_NULLS ? ch == '\0' : (ch == '\r' || ch == '\n')) {
+ /* Skip empty lines if dumping comments. */
+ if (flags & RL_DUMP_COMMENTS && s == buf)
+ continue;
+ break;
+ }
+ if (s < eob)
+ *s++ = ch;
+ }
+ *s = '\0';
+
+ if (flags & RL_DUMP_COMMENTS && (*buf == '#' || *buf == ';'))
+ goto start;
+
+#ifdef ICONV_OPTION
+ if (flags & RL_CONVERT) {
+ xbuf outbuf;
+ INIT_XBUF(outbuf, buf, 0, bufsiz);
+ iconv_buf.pos = 0;
+ iconv_buf.len = s - iconv_buf.buf;
+ iconvbufs(ic_recv, &iconv_buf, &outbuf,
+ ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_INIT);
+ outbuf.buf[outbuf.len] = '\0';
+ return outbuf.len;
+ }
+#endif
+
+ return s - buf;
+}
+
+void read_args(int f_in, char *mod_name, char *buf, size_t bufsiz, int rl_nulls,
+ char ***argv_p, int *argc_p, char **request_p)
+{
+ int maxargs = MAX_ARGS;
+ int dot_pos = 0;
+ int argc = 0;
+ char **argv, *p;
+ int rl_flags = (rl_nulls ? RL_EOL_NULLS : 0);
+
+#ifdef ICONV_OPTION
+ rl_flags |= (protect_args && ic_recv != (iconv_t)-1 ? RL_CONVERT : 0);
+#endif
+
+ if (!(argv = new_array(char *, maxargs)))
+ out_of_memory("read_args");
+ if (mod_name && !protect_args)
+ argv[argc++] = "rsyncd";
+
+ while (1) {
+ if (read_line(f_in, buf, bufsiz, rl_flags) == 0)
+ break;
+
+ if (argc == maxargs-1) {
+ maxargs += MAX_ARGS;
+ if (!(argv = realloc_array(argv, char *, maxargs)))
+ out_of_memory("read_args");
+ }
+
+ if (dot_pos) {
+ if (request_p) {
+ *request_p = strdup(buf);
+ request_p = NULL;
+ }
+ if (mod_name)
+ glob_expand_module(mod_name, buf, &argv, &argc, &maxargs);
+ else
+ glob_expand(buf, &argv, &argc, &maxargs);
+ } else {
+ if (!(p = strdup(buf)))
+ out_of_memory("read_args");
+ argv[argc++] = p;
+ if (*p == '.' && p[1] == '\0')
+ dot_pos = argc;
+ }
+ }
+ argv[argc] = NULL;
+
+ glob_expand(NULL, NULL, NULL, NULL);
+
+ *argc_p = argc;
+ *argv_p = argv;
+}
+
+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 (iobuf.out.buf) {
+ if (iobuf.out_fd == -1)
+ iobuf.out_fd = f_out;
+ else
+ assert(f_out == iobuf.out_fd);
+ return False;
+ }
+
+ alloc_xbuf(&iobuf.out, ROUND_UP_1024(IO_BUFFER_SIZE * 2));
+ iobuf.out_fd = f_out;
+
+ return True;
+}
+
+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;
+ else
+ assert(f_in == iobuf.in_fd);
+ return False;
+ }
+
+ alloc_xbuf(&iobuf.in, ROUND_UP_1024(IO_BUFFER_SIZE));
+ iobuf.in_fd = f_in;
+
+ return True;
+}
+
+void io_end_buffering_in(BOOL free_buffers)
+{
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_buffering_in(IOBUF_%s_BUFS)\n",
+ who_am_i(), free_buffers ? "FREE" : "KEEP");
+ }
+
+ if (free_buffers)
+ free_xbuf(&iobuf.in);
+ else
+ iobuf.in.pos = iobuf.in.len = 0;
+
+ iobuf.in_fd = -1;
+}
+
+void io_end_buffering_out(BOOL free_buffers)
+{
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_buffering_out(IOBUF_%s_BUFS)\n",
+ who_am_i(), free_buffers ? "FREE" : "KEEP");
+ }
+
+ io_flush(FULL_FLUSH);
+
+ if (free_buffers) {
+ free_xbuf(&iobuf.out);
+ free_xbuf(&iobuf.msg);
+ }
+
+ iobuf.out_fd = -1;
+}
+
+void maybe_flush_socket(int important)
+{
+ if (flist_eof && iobuf.out.buf && iobuf.out.len > iobuf.out_empty_len
+ && (important || time(NULL) - last_io_out >= 5))
+ io_flush(NORMAL_FLUSH);
+}
+
+/* Older rsync versions used to send either a MSG_NOOP (protocol 30) or a
+ * raw-data-based keep-alive (protocol 29), both of which implied forwarding of
+ * the message through the sender. Since the new timeout method does not need
+ * any forwarding, we just send an empty MSG_DATA message, which works with all
+ * rsync versions. This avoids any message forwarding, and leaves the raw-data
+ * stream alone (since we can never be quite sure if that stream is in the
+ * right state for a keep-alive message). */
+void maybe_send_keepalive(time_t now, int flags)
+{
+ if (flags & MSK_ACTIVE_RECEIVER)
+ last_io_in = now; /* Fudge things when we're working hard on the files. */
+
+ if (now - last_io_out >= allowed_lull) {
+ /* The receiver is special: it only sends keep-alive messages if it is
+ * actively receiving data. Otherwise, it lets the generator timeout. */
+ if (am_receiver && now - last_io_in >= io_timeout)
+ return;
+
+ if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len)
+ send_msg(MSG_DATA, "", 0, 0);
+ if (!(flags & MSK_ALLOW_FLUSH)) {
+ /* Let the caller worry about writing out the data. */
+ } else if (iobuf.msg.len)
+ perform_io(iobuf.msg.size - iobuf.msg.len + 1, PIO_NEED_MSGROOM);
+ else if (iobuf.out.len > iobuf.out_empty_len)
+ io_flush(NORMAL_FLUSH);
+ }
+}
+
+void start_flist_forward(int ndx)
+{
+ write_int(iobuf.out_fd, ndx);
+ forward_flist_data = 1;
+}
+
+void stop_flist_forward(void)
+{
+ forward_flist_data = 0;
+}
+
+/* Read a message from a multiplexed source. */
+static void read_a_msg(void)
+{
+ char data[BIGPATHBUFLEN];
+ int tag, val;
+ size_t msg_bytes;
+
+ /* This ensures that perform_io() does not try to do any message reading
+ * until we've read all of the data for this message. We should also
+ * try to avoid calling things that will cause data to be written via
+ * perform_io() prior to this being reset to 1. */
+ iobuf.in_multiplexed = -1;
+
+ tag = raw_read_int();
+
+ msg_bytes = tag & 0xFFFFFF;
+ tag = (tag >> 24) - MPLEX_BASE;
+
+ if (DEBUG_GTE(IO, 1) && msgs2stderr)
+ rprintf(FINFO, "[%s] got msg=%d, len=%ld\n", who_am_i(), (int)tag, (long)msg_bytes);
+
+ switch (tag) {
+ case MSG_DATA:
+ assert(iobuf.raw_input_ends_before == 0);
+ /* Though this does not yet read the data, we do mark where in
+ * the buffer the msg data will end once it is read. It is
+ * possible that this points off the end of the buffer, in
+ * which case the gradual reading of the input stream will
+ * cause this value to wrap around and eventually become real. */
+ if (msg_bytes)
+ iobuf.raw_input_ends_before = iobuf.in.pos + msg_bytes;
+ iobuf.in_multiplexed = 1;
+ break;
+ case MSG_STATS:
+ if (msg_bytes != sizeof stats.total_read || !am_generator)
+ goto invalid_msg;
+ raw_read_buf((char*)&stats.total_read, sizeof stats.total_read);
+ iobuf.in_multiplexed = 1;
+ break;
+ case MSG_REDO:
+ if (msg_bytes != 4 || !am_generator)
+ goto invalid_msg;
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ got_flist_entry_status(FES_REDO, val);
+ break;
+ case MSG_IO_ERROR:
+ if (msg_bytes != 4)
+ goto invalid_msg;
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ io_error |= val;
+ if (am_receiver)
+ 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();
+ iobuf.in_multiplexed = 1;
+ 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:
+ /* Support protocol-30 keep-alive method. */
+ if (msg_bytes != 0)
+ goto invalid_msg;
+ iobuf.in_multiplexed = 1;
+ if (am_sender)
+ maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
+ break;
+ case MSG_DELETED:
+ if (msg_bytes >= sizeof data)
+ goto overflow;
+ if (am_generator) {
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ send_msg(MSG_DELETED, data, msg_bytes, 1);
+ break;
+ }
+#ifdef ICONV_OPTION
+ if (ic_recv != (iconv_t)-1) {
+ xbuf outbuf, inbuf;
+ char ibuf[512];
+ int add_null = 0;
+ int flags = ICB_INCLUDE_BAD | ICB_INIT;
+
+ 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;
+ raw_read_buf(ibuf + inbuf.len, len);
+ inbuf.pos = 0;
+ inbuf.len += len;
+ if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])
+ inbuf.len--, add_null = 1;
+ if (iconvbufs(ic_send, &inbuf, &outbuf, flags) < 0) {
+ if (errno == E2BIG)
+ goto overflow;
+ /* Buffer ended with an incomplete char, so move the
+ * bytes to the start of the buffer and continue. */
+ memmove(ibuf, ibuf + inbuf.pos, inbuf.len);
+ }
+ flags &= ~ICB_INIT;
+ }
+ if (add_null) {
+ if (outbuf.len == outbuf.size)
+ goto overflow;
+ outbuf.buf[outbuf.len++] = '\0';
+ }
+ msg_bytes = outbuf.len;
+ } else
+#endif
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ /* A directory name was sent with the trailing null */
+ if (msg_bytes > 0 && !data[msg_bytes-1])
+ log_delete(data, S_IFDIR);
+ else {
+ data[msg_bytes] = '\0';
+ log_delete(data, 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);
+ }
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ if (am_generator)
+ got_flist_entry_status(FES_SUCCESS, val);
+ else
+ successful_send(val);
+ break;
+ case MSG_NO_SEND:
+ if (msg_bytes != 4)
+ goto invalid_msg;
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ if (am_generator)
+ got_flist_entry_status(FES_NO_SEND, val);
+ else
+ send_msg_int(MSG_NO_SEND, val);
+ 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 data) {
+ 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);
+ }
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ rwrite((enum logcode)tag, data, msg_bytes, !am_generator);
+ if (first_message) {
+ 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 (msg_bytes == 4)
+ val = raw_read_int();
+ else if (msg_bytes == 0)
+ val = 0;
+ else
+ goto invalid_msg;
+ iobuf.in_multiplexed = 1;
+ if (DEBUG_GTE(EXIT, 3))
+ rprintf(FINFO, "[%s] got MSG_ERROR_EXIT with %ld bytes\n", who_am_i(), (long)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);
+ }
+ } else 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);