+ 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);