+static void msg_flush(void)
+{
+ if (am_generator) {
+ while (msg_queue.head && io_multiplexing_out) {
+ struct msg_list_item *m = msg_queue.head;
+ int len = IVAL(m->buf, 0) & 0xFFFFFF;
+ int tag = *((uchar*)m->buf+3) - MPLEX_BASE;
+ if (!(msg_queue.head = m->next))
+ msg_queue.tail = NULL;
+ stats.total_written += len + 4;
+ defer_forwarding_messages++;
+ mplex_write(sock_f_out, tag, m->buf + 4, len, m->convert);
+ defer_forwarding_messages--;
+ free(m);
+ }
+ } else {
+ while (msg_queue.head) {
+ struct msg_list_item *m = msg_queue.head;
+ int len = IVAL(m->buf, 0) & 0xFFFFFF;
+ int tag = *((uchar*)m->buf+3) - MPLEX_BASE;
+ if (!(msg_queue.head = m->next))
+ msg_queue.tail = NULL;
+ defer_forwarding_messages++;
+ mplex_write(msg_fd_out, tag, m->buf + 4, len, m->convert);
+ defer_forwarding_messages--;
+ free(m);
+ }
+ }
+}
+
+/* Read a message from the MSG_* fd and handle it. This is called either
+ * during the early stages of being a local sender (up through the sending
+ * of the file list) or when we're the generator (to fetch the messages
+ * from the receiver). */
+static void read_msg_fd(void)
+{
+ char buf[2048];
+ size_t n;
+ struct file_list *flist;
+ int fd = msg_fd_in;
+ int tag, len;
+
+ /* Temporarily disable msg_fd_in. This is needed to avoid looping back
+ * to this routine from writefd_unbuffered(). */
+ no_flush++;
+ msg_fd_in = -1;
+ defer_forwarding_messages++;
+
+ readfd(fd, buf, 4);
+ tag = IVAL(buf, 0);
+
+ len = tag & 0xFFFFFF;
+ tag = (tag >> 24) - MPLEX_BASE;
+
+ switch (tag) {
+ case MSG_DONE:
+ if (len < 0 || len > 1 || !am_generator) {
+ invalid_msg:
+ rprintf(FERROR, "invalid message %d:%d [%s%s]\n",
+ tag, len, who_am_i(),
+ inc_recurse ? "/inc" : "");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ if (len) {
+ readfd(fd, buf, len);
+ stats.total_read = read_varlong(fd, 3);
+ }
+ msgdone_cnt++;
+ break;
+ case MSG_REDO:
+ if (len != 4 || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, 4);
+ if (remove_source_files)
+ decrement_active_files(IVAL(buf,0));
+ flist_ndx_push(&redo_list, IVAL(buf,0));
+ if (inc_recurse)
+ decrement_flist_in_progress(IVAL(buf,0), 1);
+ break;
+ case MSG_FLIST:
+ if (len != 4 || !am_generator || !inc_recurse)
+ goto invalid_msg;
+ readfd(fd, buf, 4);
+ /* Read extra file list from receiver. */
+ assert(iobuf_in != NULL);
+ assert(iobuf_f_in == fd);
+ if (verbose > 3) {
+ rprintf(FINFO, "[%s] receiving flist for dir %d\n",
+ who_am_i(), IVAL(buf,0));
+ }
+ flist = recv_file_list(fd);
+ flist->parent_ndx = IVAL(buf,0);
+ break;
+ case MSG_FLIST_EOF:
+ if (len != 0 || !am_generator || !inc_recurse)
+ goto invalid_msg;
+ flist_eof = 1;
+ break;
+ case MSG_DELETED:
+ if (len >= (int)sizeof buf || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, len);
+ send_msg(MSG_DELETED, buf, len, 1);
+ break;
+ case MSG_SUCCESS:
+ if (len != 4 || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, len);
+ if (remove_source_files) {
+ decrement_active_files(IVAL(buf,0));
+ send_msg(MSG_SUCCESS, buf, len, 0);
+ }
+ if (preserve_hard_links)
+ flist_ndx_push(&hlink_list, IVAL(buf,0));
+ if (inc_recurse)
+ decrement_flist_in_progress(IVAL(buf,0), 0);
+ break;
+ case MSG_NO_SEND:
+ if (len != 4 || !am_generator)
+ goto invalid_msg;
+ readfd(fd, buf, len);
+ if (inc_recurse)
+ decrement_flist_in_progress(IVAL(buf,0), 0);
+ break;
+ case MSG_SOCKERR:
+ case MSG_CLIENT:
+ if (!am_generator)
+ goto invalid_msg;
+ if (tag == MSG_SOCKERR)
+ io_end_multiplex_out();
+ /* FALL THROUGH */
+ case MSG_INFO:
+ case MSG_ERROR:
+ case MSG_LOG:
+ while (len) {
+ n = len;
+ if (n >= sizeof buf)
+ n = sizeof buf - 1;
+ readfd(fd, buf, n);
+ rwrite((enum logcode)tag, buf, n, !am_generator);