+/** Setup the fd used to send our MSG_* messages. Only needed when
+ * we're the receiver because the generator and the sender both use
+ * the multiplexed I/O setup. */
+void set_msg_fd_out(int fd)
+{
+ msg_fd_out = fd;
+ set_nonblocking(msg_fd_out);
+}
+
+/* Add a message to the pending MSG_* list. */
+static void msg_list_add(int code, char *buf, int len)
+{
+ struct msg_list *ml;
+
+ if (!(ml = new(struct msg_list)))
+ exit_cleanup(RERR_MALLOC);
+ ml->next = NULL;
+ if (!(ml->buf = new_array(char, len+4)))
+ exit_cleanup(RERR_MALLOC);
+ SIVAL(ml->buf, 0, ((code+MPLEX_BASE)<<24) | len);
+ memcpy(ml->buf+4, buf, len);
+ ml->len = len+4;
+ if (msg_list_tail)
+ msg_list_tail->next = ml;
+ else
+ msg_list_head = ml;
+ msg_list_tail = ml;
+}
+
+void send_msg(enum msgcode code, char *buf, int len)
+{
+ msg_list_add(code, buf, len);
+ msg_list_push(NORMAL_FLUSH);
+}
+
+/** Read a message from the MSG_* fd and dispatch it. This is only
+ * called by the generator. */
+static void read_msg_fd(void)
+{
+ char buf[200];
+ size_t n;
+ int fd = msg_fd_in;
+ int tag, len;
+
+ /* Temporarily disable msg_fd_in. This is needed because we
+ * may call a write routine that could try to call us back. */
+ msg_fd_in = -1;
+
+ read_loop(fd, buf, 4);
+ tag = IVAL(buf, 0);
+
+ len = tag & 0xFFFFFF;
+ tag = (tag >> 24) - MPLEX_BASE;
+
+ switch (tag) {
+ case MSG_DONE:
+ if (len != 0) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ redo_list_add(-1);
+ break;
+ case MSG_REDO:
+ if (len != 4) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ read_loop(fd, buf, 4);
+ redo_list_add(IVAL(buf,0));
+ break;
+ case MSG_INFO:
+ case MSG_ERROR:
+ case MSG_LOG:
+ while (len) {
+ n = len;
+ if (n >= sizeof buf)
+ n = sizeof buf - 1;
+ read_loop(fd, buf, n);
+ rwrite((enum logcode)tag, buf, n);
+ len -= n;
+ }
+ break;
+ default:
+ rprintf(FERROR, "unknown message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ msg_fd_in = fd;
+}
+
+/* Try to push messages off the list onto the wire. If we leave with more
+ * to do, return 0. On error, return -1. If everything flushed, return 1.
+ * This is only called by the receiver. */
+int msg_list_push(int flush_it_all)
+{
+ static int written = 0;
+ struct timeval tv;
+ fd_set fds;
+
+ if (msg_fd_out < 0)
+ return -1;
+
+ while (msg_list_head) {
+ struct msg_list *ml = msg_list_head;
+ int n = write(msg_fd_out, ml->buf + written, ml->len - written);
+ if (n < 0) {
+ if (errno == EINTR)
+ continue;
+ if (errno != EWOULDBLOCK && errno != EAGAIN)
+ return -1;
+ if (!flush_it_all)
+ return 0;
+ FD_ZERO(&fds);
+ FD_SET(msg_fd_out, &fds);
+ tv.tv_sec = io_timeout ? io_timeout : SELECT_TIMEOUT;
+ tv.tv_usec = 0;
+ if (!select(msg_fd_out+1, NULL, &fds, NULL, &tv))
+ check_timeout();
+ } else if ((written += n) == ml->len) {
+ free(ml->buf);
+ msg_list_head = ml->next;
+ if (!msg_list_head)
+ msg_list_tail = NULL;
+ free(ml);
+ written = 0;
+ }
+ }
+ return 1;
+}
+
+int get_redo_num(void)
+{
+ struct redo_list *next;
+ int num;
+
+ while (!redo_list_head)
+ read_msg_fd();
+
+ num = redo_list_head->num;
+ next = redo_list_head->next;
+ free(redo_list_head);
+ redo_list_head = next;
+ if (!next)
+ redo_list_tail = NULL;
+
+ return num;
+}