+ no_flush--;
+ msg_fd_in = fd;
+ if (!--defer_forwarding_messages)
+ msg_flush();
+}
+
+/* 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)
+{
+ /* TODO: tune these limits? */
+ while (active_filecnt >= (active_bytecnt >= 128*1024 ? 10 : 50)) {
+ check_for_finished_files(itemizing, code, 0);
+ if (iobuf_out_cnt)
+ io_flush(NORMAL_FLUSH);
+ else
+ read_msg_fd();
+ }
+
+ active_filecnt++;
+ active_bytecnt += F_LENGTH(cur_flist->files[ndx - cur_flist->ndx_start]);
+}
+
+static void decrement_active_files(int ndx)
+{
+ struct file_list *flist = flist_for_ndx(ndx);
+ assert(flist != NULL);
+ active_filecnt--;
+ active_bytecnt -= F_LENGTH(flist->files[ndx - flist->ndx_start]);
+}
+
+static void decrement_flist_in_progress(int ndx, int redo)
+{
+ struct file_list *flist = cur_flist ? cur_flist : first_flist;
+
+ while (ndx < flist->ndx_start) {
+ if (flist == first_flist) {
+ invalid_ndx:
+ rprintf(FERROR,
+ "Invalid file index: %d (%d - %d) [%s]\n",
+ ndx, first_flist->ndx_start,
+ first_flist->prev->ndx_start + first_flist->prev->count - 1,
+ who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ flist = flist->prev;
+ }
+ while (ndx >= flist->ndx_start + flist->count) {
+ if (!(flist = flist->next))
+ goto invalid_ndx;
+ }
+
+ flist->in_progress--;
+ if (redo)
+ flist->to_redo++;
+}
+
+/* Write an message to a multiplexed stream. If this fails, rsync exits. */
+static void mplex_write(int fd, enum msgcode code, const char *buf, size_t len, int convert)
+{
+ char buffer[BIGPATHBUFLEN]; /* Oversized for use by iconv code. */
+ size_t n = len;
+
+ SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
+
+#ifdef ICONV_OPTION
+ if (convert && ic_send == (iconv_t)-1)
+#endif
+ convert = 0;
+
+ if (convert || n > 1024 - 4) /* BIGPATHBUFLEN can handle 1024 bytes */
+ n = 0;
+ else
+ memcpy(buffer + 4, buf, n);
+
+ writefd_unbuffered(fd, buffer, n+4);
+
+ len -= n;
+ buf += n;
+
+#ifdef ICONV_OPTION
+ if (convert) {
+ iconv(ic_send, NULL, 0, NULL, 0);
+ defer_forwarding_messages++;
+ while (len) {
+ ICONV_CONST char *ibuf = (ICONV_CONST char *)buf;
+ char *obuf = buffer;
+ size_t ocnt = sizeof buffer;
+ while (len && iconv(ic_send, &ibuf,&len,
+ &obuf,&ocnt) == (size_t)-1) {
+ if (errno == E2BIG || !ocnt)
+ break;
+ *obuf++ = *ibuf++;
+ ocnt--, len--;
+ }
+ n = obuf - buffer;
+ writefd_unbuffered(fd, buffer, n);
+ }
+ if (!--defer_forwarding_messages)
+ msg_flush();
+ } else
+#endif
+ if (len) {
+ defer_forwarding_messages++;
+ writefd_unbuffered(fd, buf, len);
+ if (!--defer_forwarding_messages)
+ msg_flush();
+ }
+}
+
+int send_msg(enum msgcode code, const char *buf, int len, int convert)
+{
+ if (msg_fd_out < 0) {
+ if (!defer_forwarding_messages)
+ return io_multiplex_write(code, buf, len, convert);
+ if (!io_multiplexing_out)
+ return 0;
+ msg_list_add(&msg_queue, code, buf, len, convert);
+ return 1;