+ 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) a buffer may be temporarily shortened, but the
+ * original size will be automatically restored. The iobuf.in buffer is also
+ * circular, so callers may need to split their reading of the data if it spans
+ * the end. See also the 3 raw_* iobuf vars that are used in the handling of
+ * MSG_DATA bytes as they are read-from/written-into the buffers.
+ *
+ * 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?
+ *
+ * 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:
+ /* We never resize the circular input buffer. */
+ if (iobuf.in.size < needed) {
+ rprintf(FERROR, "need to read %ld bytes, iobuf.in.buf is only %ld bytes.\n",
+ (long)needed, (long)iobuf.in.size);
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ if (DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%ld, %sinput)\n",
+ who_am_i(), (long)needed, flags & PIO_CONSUME_INPUT ? "consume&" : "");
+ }
+ 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.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;
+ }
+
+ empty_buf_len = iobuf.out_empty_len;
+ out = &iobuf.out;
+ } else if (iobuf.msg.len) {
+ empty_buf_len = 0;
+ out = &iobuf.msg;
+ } else
+ out = NULL;
+ if (out) {
+ FD_SET(iobuf.out_fd, &w_fds);
+ if (iobuf.out_fd > max_fd)
+ max_fd = iobuf.out_fd;
+ }
+ } else
+ out = NULL;
+
+ if (max_fd < 0) {
+ switch (flags & PIO_NEED_FLAGS) {
+ case PIO_NEED_INPUT:
+ iobuf.in.len = 0;
+ if (kluge_around_eof == 2)
+ exit_cleanup(0);
+ if (iobuf.in_fd == -2)
+ whine_about_eof(True);
+ rprintf(FERROR, "error in perform_io: no fd for input.\n");
+ exit_cleanup(RERR_PROTOCOL);
+ case PIO_NEED_OUTROOM:
+ case PIO_NEED_MSGROOM:
+ msgs2stderr = 1;
+ drain_multiplex_messages();
+ if (iobuf.out_fd == -2)
+ whine_about_eof(True);
+ rprintf(FERROR, "error in perform_io: no fd for output.\n");
+ exit_cleanup(RERR_PROTOCOL);
+ default:
+ /* No stated needs, so I guess this is OK. */
+ break;
+ }
+ break;
+ }
+
+ if (extra_flist_sending_enabled) {
+ if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD)
+ tv.tv_sec = 0;