+ 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) {
+ if (!(iobuf.in.buf = realloc_array(iobuf.in.buf, char, needed)))
+ out_of_memory("perform_io");
+ if (DEBUG_GTE(IO, 4)) {
+ rprintf(FINFO, "[%s] resized input buffer from %d to %d bytes.\n",
+ who_am_i(), iobuf.in.size, needed);
+ }
+ iobuf.in.size = needed;
+ }
+ 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 %d bytes from %d to 0 in the input buffer (size=%d, needed=%d).\n",
+ who_am_i(), iobuf.in.len, iobuf.in.pos, iobuf.in.size, 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 > iobuf.out.size - needed
+ ? (long)iobuf.out.len - (iobuf.out.size - needed) : 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.out.len > iobuf.msg.size - needed
+ ? (long)iobuf.out.len - (iobuf.msg.size - needed) : 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:
+ if (iobuf.out.len <= iobuf.out.size - needed)
+ goto double_break;
+ break;
+ case PIO_NEED_MSGROOM:
+ if (iobuf.msg.len <= iobuf.msg.size - needed)
+ 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) {
+ size_t val;
+
+ /* 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 ((val = iobuf.out.size - iobuf.raw_data_header_pos) < 4) {
+ /* We used some of the overflow bytes, so move them. */
+ if (DEBUG_GTE(IO, 4)) {
+ rprintf(FINFO, "[%s] wrap-bytes moved: %d (perform_io)\n",
+ who_am_i(), (int)val);
+ }
+ memcpy(iobuf.out.buf, iobuf.out.buf + iobuf.out.size, 4 - val);
+ }
+
+ 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;
+ 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:
+ 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;
+ 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
+ && file_total - file_old_total >= MIN_FILECNT_LOOKAHEAD)
+ tv.tv_sec = 0;
+ else {
+ extra_flist_sending_enabled = False;
+ tv.tv_sec = select_timeout;
+ }
+ } else
+ tv.tv_sec = select_timeout;
+ tv.tv_usec = 0;
+
+ cnt = select(max_fd + 1, &r_fds, &w_fds, &e_fds, &tv);
+
+ if (cnt <= 0) {
+ if (cnt < 0 && errno == EBADF) {
+ msgs2stderr = 1;
+ exit_cleanup(RERR_SOCKETIO);
+ }
+ if (extra_flist_sending_enabled) {
+ extra_flist_sending_enabled = False;
+ send_extra_file_list(sock_f_out, -1);
+ extra_flist_sending_enabled = !flist_eof;
+ } else
+ check_timeout();
+ FD_ZERO(&r_fds); /* Just in case... */
+ FD_ZERO(&w_fds);
+ }
+
+ if (iobuf.in_fd >= 0 && FD_ISSET(iobuf.in_fd, &r_fds)) {
+ size_t pos = iobuf.in.pos + iobuf.in.len;
+ size_t len = iobuf.in.size - pos;
+ int n;
+ if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
+ if (n == 0) {
+ if (!read_batch || batch_fd < 0 || am_generator)
+ whine_about_eof(iobuf.in_fd); /* Doesn't return. */
+ batch_fd = -1;
+ continue;
+ }
+ if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)
+ n = 0;
+ else {
+ /* Don't write errors on a dead socket. */
+ if (iobuf.in_fd == sock_f_in) {
+ if (am_sender)
+ msgs2stderr = 1;
+ rsyserr(FERROR_SOCKET, errno, "read error");
+ } else
+ rsyserr(FERROR, errno, "read error");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ }
+ if (msgs2stderr && DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] recv=%ld\n", who_am_i(), (long)n);
+
+ if (io_timeout)
+ last_io_in = time(NULL);
+ stats.total_read += n;
+
+ iobuf.in.len += n;
+ }
+
+ if (iobuf.out_fd >= 0 && FD_ISSET(iobuf.out_fd, &w_fds)) {
+ size_t len = iobuf.raw_flushing_ends_before ? iobuf.raw_flushing_ends_before - out->pos : out->len;
+ int n;
+
+ if (bwlimit_writemax && len > bwlimit_writemax)
+ len = bwlimit_writemax;
+
+ if (out->pos + len > out->size)
+ len = out->size - out->pos;
+ if ((n = write(iobuf.out_fd, out->buf + out->pos, len)) <= 0) {
+ if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)
+ n = 0;
+ else {
+ /* Don't write errors on a dead socket. */
+ msgs2stderr = 1;
+ out->len = iobuf.raw_flushing_ends_before = out->pos = 0;
+ rsyserr(FERROR_SOCKET, errno, "write error");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ }
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] %s sent=%ld\n",
+ who_am_i(), out == &iobuf.out ? "out" : "msg", (long)n);
+ }
+
+ if (io_timeout)
+ last_io_out = time(NULL);
+ stats.total_written += n;
+
+ if (bwlimit_writemax)
+ sleep_for_bwlimit(n);
+
+ if ((out->pos += n) == out->size) {
+ if (iobuf.raw_flushing_ends_before)
+ iobuf.raw_flushing_ends_before -= out->size;
+ out->pos = 0;
+ }
+ if (out->pos == iobuf.raw_flushing_ends_before)
+ iobuf.raw_flushing_ends_before = 0;
+ if ((out->len -= n) == empty_buf_len) {
+ out->pos = 0;
+ if (empty_buf_len)
+ iobuf.raw_data_header_pos = 0;
+ }
+ }
+
+ if (ff_forward_fd >= 0 && FD_ISSET(ff_forward_fd, &r_fds)) {
+ /* This can potentially flush all output and enable
+ * multiplexed output, so keep this last in the loop
+ * and be sure to not cache anything that would break
+ * such a change. */
+ forward_filesfrom_data();
+ }
+ }
+ double_break:
+
+ data = iobuf.in.buf + iobuf.in.pos;
+
+ if (flags & PIO_CONSUME_INPUT) {
+ iobuf.in.len -= needed;
+ iobuf.in.pos += needed;
+ }
+
+ return data;