+ int len;
+
+ len = read(ff_forward_fd, ff_xb.buf + ff_xb.len, ff_xb.size - ff_xb.len);
+ if (len <= 0) {
+ if (len == 0 || errno != EINTR) {
+ /* Send end-of-file marker */
+ ff_forward_fd = -1;
+ write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
+ free_xbuf(&ff_xb);
+ if (ff_reenable_multiplex >= 0)
+ io_start_multiplex_out(ff_reenable_multiplex);
+ }
+ return;
+ }
+
+ if (DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] files-from read=%ld\n", who_am_i(), (long)len);
+
+#ifdef ICONV_OPTION
+ len += ff_xb.len;
+#endif
+
+ if (!eol_nulls) {
+ char *s = ff_xb.buf + len;
+ /* Transform CR and/or LF into '\0' */
+ while (s-- > ff_xb.buf) {
+ if (*s == '\n' || *s == '\r')
+ *s = '\0';
+ }
+ }
+
+ if (ff_lastchar)
+ ff_xb.pos = 0;
+ else {
+ char *s = ff_xb.buf;
+ /* Last buf ended with a '\0', so don't let this buf start with one. */
+ while (len && *s == '\0')
+ s++, len--;
+ ff_xb.pos = s - ff_xb.buf;
+ }
+
+#ifdef ICONV_OPTION
+ if (filesfrom_convert && len) {
+ char *sob = ff_xb.buf + ff_xb.pos, *s = sob;
+ char *eob = sob + len;
+ int flags = ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT;
+ if (ff_lastchar == '\0')
+ flags |= ICB_INIT;
+ /* Convert/send each null-terminated string separately, skipping empties. */
+ while (s != eob) {
+ if (*s++ == '\0') {
+ ff_xb.len = s - sob - 1;
+ if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0)
+ exit_cleanup(RERR_PROTOCOL); /* impossible? */
+ write_buf(iobuf.out_fd, s-1, 1); /* Send the '\0'. */
+ while (s != eob && *s == '\0')
+ s++;
+ sob = s;
+ ff_xb.pos = sob - ff_xb.buf;
+ flags |= ICB_INIT;
+ }
+ }
+
+ if ((ff_xb.len = s - sob) == 0)
+ ff_lastchar = '\0';
+ else {
+ /* Handle a partial string specially, saving any incomplete chars. */
+ flags &= ~ICB_INCLUDE_INCOMPLETE;
+ if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0) {
+ if (errno == E2BIG)
+ exit_cleanup(RERR_PROTOCOL); /* impossible? */
+ if (ff_xb.pos)
+ memmove(ff_xb.buf, ff_xb.buf + ff_xb.pos, ff_xb.len);
+ }
+ ff_lastchar = 'x'; /* Anything non-zero. */
+ }
+ } else
+#endif
+
+ if (len) {
+ char *f = ff_xb.buf + ff_xb.pos;
+ char *t = ff_xb.buf;
+ char *eob = f + len;
+ /* Eliminate any multi-'\0' runs. */
+ while (f != eob) {
+ if (!(*t++ = *f++)) {
+ while (f != eob && *f == '\0')
+ f++;
+ }
+ }
+ ff_lastchar = f[-1];
+ if ((len = t - ff_xb.buf) != 0) {
+ /* This will not circle back to perform_io() because we only get
+ * called when there is plenty of room in the output buffer. */
+ write_buf(iobuf.out_fd, ff_xb.buf, len);
+ }
+ }
+}
+
+void reduce_iobuf_size(xbuf *out, size_t new_size)
+{
+ if (new_size < out->size) {
+ if (DEBUG_GTE(IO, 4)) {
+ const char *name = out == &iobuf.out ? "iobuf.out"
+ : out == &iobuf.msg ? "iobuf.msg"
+ : NULL;
+ if (name) {
+ rprintf(FINFO, "[%s] reduced size of %s (-%d)\n",
+ who_am_i(), name, (int)(out->size - new_size));
+ }
+ }
+ out->size = new_size;
+ }
+}
+
+void restore_iobuf_size(xbuf *out)
+{
+ if (IOBUF_WAS_REDUCED(out->size)) {
+ size_t new_size = IOBUF_RESTORE_SIZE(out->size);
+ if (DEBUG_GTE(IO, 4)) {
+ const char *name = out == &iobuf.out ? "iobuf.out"
+ : out == &iobuf.msg ? "iobuf.msg"
+ : NULL;
+ if (name) {
+ rprintf(FINFO, "[%s] restored size of %s (+%d)\n",
+ who_am_i(), name, (int)(new_size - out->size));
+ }
+ }
+ out->size = new_size;
+ }
+}
+
+/* Perform buffered input and output until specified conditions are met. When
+ * given a "needed" read requirement, we'll return without doing any I/O if the
+ * iobuf.in bytes are already available. When reading, we'll read as many
+ * bytes as we can into the buffer, and return as soon as we meet the minimum
+ * read requirement. When given a "needed" write requirement, we'll return
+ * without doing any I/O if that many bytes will fit in the output buffer (we
+ * check either iobuf.out or iobuf.msg, depending on the flags). When writing,
+ * we write out as much as we can, and return as soon as the given free-space
+ * requirement is available.
+ *
+ * The iobuf.out and iobuf.msg buffers are circular, so some writes into them
+ * will need to be split when the data needs to wrap around to the start. In
+ * order to help make this easier for some operations (such as the use of
+ * SIVAL() into the buffer) the buffers MUST have 4 bytes of overflow space at
+ * the end that is not not counted in the "size". The iobuf.in buffer is not
+ * (currently) circular. To facilitate the handling of MSG_DATA bytes as they
+ * are read-from/written-into the buffers, see the three raw_* iobuf vars.
+ *
+ * When writing, we flush data in the following priority order:
+ *
+ * 1. Finish writing any in-progress MSG_DATA sequence from iobuf.out.
+ *
+ * 2. Write out all the messages from the message buf (if iobuf.msg is active).
+ * Yes, this means that a PIO_NEED_OUTROOM call will completely flush any
+ * messages before getting to the iobuf.out flushing (except for rule 1).
+ *
+ * 3. Write out the raw data from iobuf.out, possibly filling in the multiplexed
+ * MSG_DATA header that was pre-allocated (when output is multiplexed).
+ *
+ * TODO: items for possible future work:
+ *
+ * - Make this routine able to read the generator-to-receiver batch flow?
+ *
+ * - Make the input buffer circular?
+ *
+ * Unlike the old routines that this replaces, it is OK to read ahead as far as
+ * we can because the read_a_msg() routine now reads its bytes out of the input
+ * buffer. In the old days, only raw data was in the input buffer, and any
+ * unused raw data in the buf would prevent the reading of socket data. */
+static char *perform_io(size_t needed, int flags)
+{
+ fd_set r_fds, e_fds, w_fds;
+ struct timeval tv;
+ int cnt, max_fd;
+ size_t empty_buf_len = 0;
+ xbuf *out;
+ char *data;
+
+ if (iobuf.in.len == 0 && iobuf.in.pos != 0) {
+ if (iobuf.raw_input_ends_before)
+ iobuf.raw_input_ends_before -= iobuf.in.pos;
+ iobuf.in.pos = 0;
+ }
+
+ switch (flags & PIO_NEED_FLAGS) {
+ case PIO_NEED_INPUT:
+ if (DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%ld, %sinput)\n",
+ who_am_i(), (long)needed, flags & PIO_CONSUME_INPUT ? "consume&" : "");
+ }
+
+ /* Make sure the input buffer is big enough to hold "needed" bytes.
+ * Also make sure it will fit in the free space at the end, or
+ * else we need to shift some bytes. */
+ if (needed && iobuf.in.size < needed) {
+ size_t new_size = ROUND_UP_1024(needed);
+ if (DEBUG_GTE(IO, 4)) {
+ rprintf(FINFO, "[%s] resizing input buffer from %ld to %ld bytes.\n",
+ who_am_i(), (long)iobuf.in.size, (long)new_size);
+ }
+ realloc_xbuf(&iobuf.in, new_size);
+ }
+ if (iobuf.in.size - iobuf.in.pos < needed
+ || (iobuf.in.len < needed && iobuf.in.len < 1024
+ && iobuf.in.size - (iobuf.in.pos + iobuf.in.len) < 1024)) {
+ memmove(iobuf.in.buf, iobuf.in.buf + iobuf.in.pos, iobuf.in.len);
+ if (DEBUG_GTE(IO, 4)) {
+ rprintf(FINFO,
+ "[%s] moved %ld bytes from %ld to 0 in the input buffer (size=%ld, needed=%ld).\n",
+ who_am_i(), (long)iobuf.in.len, (long)iobuf.in.pos, (long)iobuf.in.size, (long)needed);
+ }
+ if (iobuf.raw_input_ends_before)
+ iobuf.raw_input_ends_before -= iobuf.in.pos;
+ iobuf.in.pos = 0;
+ }
+ break;
+
+ case PIO_NEED_OUTROOM:
+ /* We never resize the circular output buffer. */
+ if (iobuf.out.size - iobuf.out_empty_len < needed) {
+ fprintf(stderr, "need to write %ld bytes, iobuf.out.buf is only %ld bytes.\n",
+ (long)needed, (long)(iobuf.out.size - iobuf.out_empty_len));
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ if (DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%ld, outroom) needs to flush %ld\n",
+ who_am_i(), (long)needed,
+ iobuf.out.len + needed > iobuf.out.size
+ ? (long)(iobuf.out.len + needed - iobuf.out.size) : 0L);
+ }
+ break;
+
+ case PIO_NEED_MSGROOM:
+ /* We never resize the circular message buffer. */
+ if (iobuf.msg.size < needed) {
+ fprintf(stderr, "need to write %ld bytes, iobuf.msg.buf is only %ld bytes.\n",
+ (long)needed, (long)iobuf.msg.size);
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ if (DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%ld, msgroom) needs to flush %ld\n",
+ who_am_i(), (long)needed,
+ iobuf.msg.len + needed > iobuf.msg.size
+ ? (long)(iobuf.msg.len + needed - iobuf.msg.size) : 0L);
+ }
+ break;
+
+ case 0:
+ if (DEBUG_GTE(IO, 3))
+ rprintf(FINFO, "[%s] perform_io(%ld, %d)\n", who_am_i(), (long)needed, flags);
+ break;
+
+ default:
+ exit_cleanup(RERR_UNSUPPORTED);
+ }
+
+ while (1) {
+ switch (flags & PIO_NEED_FLAGS) {
+ case PIO_NEED_INPUT:
+ if (iobuf.in.len >= needed)
+ goto double_break;
+ break;
+ case PIO_NEED_OUTROOM:
+ /* Note that iobuf.out_empty_len doesn't factor into this check
+ * because iobuf.out.len already holds any needed header len. */
+ if (iobuf.out.len + needed <= iobuf.out.size)
+ goto double_break;
+ break;
+ case PIO_NEED_MSGROOM:
+ if (iobuf.msg.len + needed <= iobuf.msg.size)
+ goto double_break;
+ break;
+ }
+
+ max_fd = -1;
+
+ FD_ZERO(&r_fds);
+ FD_ZERO(&e_fds);
+ if (iobuf.in_fd >= 0 && iobuf.in.size - (iobuf.in.pos + iobuf.in.len)) {
+ if (!read_batch || batch_fd >= 0) {
+ FD_SET(iobuf.in_fd, &r_fds);
+ FD_SET(iobuf.in_fd, &e_fds);
+ }
+ if (iobuf.in_fd > max_fd)
+ max_fd = iobuf.in_fd;
+ }
+
+ /* Only do more filesfrom processing if there is enough room in the out buffer. */
+ if (ff_forward_fd >= 0 && iobuf.out.size - iobuf.out.len > FILESFROM_BUFLEN*2) {
+ FD_SET(ff_forward_fd, &r_fds);
+ if (ff_forward_fd > max_fd)
+ max_fd = ff_forward_fd;
+ }
+
+ FD_ZERO(&w_fds);
+ if (iobuf.out_fd >= 0) {
+ if (iobuf.raw_flushing_ends_before
+ || (!iobuf.msg.len && iobuf.out.len > iobuf.out_empty_len && !(flags & PIO_NEED_MSGROOM))) {
+ if (OUT_MULTIPLEXED && !iobuf.raw_flushing_ends_before) {
+ /* The iobuf.raw_flushing_ends_before value can point off the end
+ * of the iobuf.out buffer for a while, for easier subtracting. */
+ iobuf.raw_flushing_ends_before = iobuf.out.pos + iobuf.out.len;
+
+ SIVAL(iobuf.out.buf + iobuf.raw_data_header_pos, 0,
+ ((MPLEX_BASE + (int)MSG_DATA)<<24) + iobuf.out.len - 4);
+
+ if (DEBUG_GTE(IO, 1)) {
+ rprintf(FINFO, "[%s] send_msg(%d, %ld)\n",
+ who_am_i(), (int)MSG_DATA, (long)iobuf.out.len - 4);
+ }
+
+ /* reserve room for the next MSG_DATA header */
+ iobuf.raw_data_header_pos = iobuf.raw_flushing_ends_before;
+ if (iobuf.raw_data_header_pos >= iobuf.out.size)
+ iobuf.raw_data_header_pos -= iobuf.out.size;
+ else if (iobuf.raw_data_header_pos + 4 > iobuf.out.size) {
+ /* The 4-byte header won't fit at the end of the buffer,
+ * so we'll temporarily reduce the output buffer's size
+ * and put the header at the start of the buffer. */
+ reduce_iobuf_size(&iobuf.out, iobuf.raw_data_header_pos);
+ iobuf.raw_data_header_pos = 0;
+ }
+ /* Yes, it is possible for this to make len > size for a while. */
+ iobuf.out.len += 4;
+ }