+/* Write len bytes to the file descriptor fd, looping as necessary to get
+ * the job done and also (in certain circumstnces) reading any data on
+ * msg_fd_in to avoid deadlock.
+ *
+ * This function underlies the multiplexing system. The body of the
+ * application never calls this function directly. */
+static void writefd_unbuffered(int fd,char *buf,size_t len)
+{
+ size_t n, total = 0;
+ fd_set w_fds, r_fds;
+ int maxfd, count, ret;
+ struct timeval tv;
+
+ no_flush++;
+
+ while (total < len) {
+ FD_ZERO(&w_fds);
+ FD_SET(fd,&w_fds);
+ maxfd = fd;
+
+ if (msg_fd_in >= 0 && len-total >= contiguous_write_len) {
+ FD_ZERO(&r_fds);
+ FD_SET(msg_fd_in,&r_fds);
+ if (msg_fd_in > maxfd)
+ maxfd = msg_fd_in;
+ }
+ if (fd != sock_f_out && iobuf_out_cnt && no_flush == 1) {
+ FD_SET(sock_f_out, &w_fds);
+ if (sock_f_out > maxfd)
+ maxfd = sock_f_out;
+ }
+
+ tv.tv_sec = select_timeout;
+ tv.tv_usec = 0;
+
+ errno = 0;
+ count = select(maxfd + 1, msg_fd_in >= 0 ? &r_fds : NULL,
+ &w_fds, NULL, &tv);
+
+ if (count <= 0) {
+ if (count < 0 && errno == EBADF)
+ exit_cleanup(RERR_SOCKETIO);
+ check_timeout();
+ continue;
+ }
+
+ if (msg_fd_in >= 0 && FD_ISSET(msg_fd_in, &r_fds))
+ read_msg_fd();
+
+ if (!FD_ISSET(fd, &w_fds)) {
+ if (fd != sock_f_out && iobuf_out_cnt) {
+ no_flush--;
+ io_flush(NORMAL_FLUSH);
+ no_flush++;
+ }
+ continue;
+ }
+
+ n = len - total;
+ if (bwlimit && n > bwlimit_writemax)
+ n = bwlimit_writemax;
+ ret = write(fd, buf + total, n);
+
+ if (ret <= 0) {
+ if (ret < 0) {
+ if (errno == EINTR)
+ continue;
+ if (errno == EWOULDBLOCK || errno == EAGAIN) {
+ msleep(1);
+ continue;
+ }
+ }
+
+ /* Don't try to write errors back across the stream. */
+ if (fd == sock_f_out)
+ close_multiplexing_out();
+ rsyserr(FERROR, errno,
+ "writefd_unbuffered failed to write %ld bytes: phase \"%s\" [%s]",
+ (long)len, io_write_phase, who_am_i());
+ /* If the other side is sending us error messages, try
+ * to grab any messages they sent before they died. */
+ while (fd == sock_f_out && io_multiplexing_in) {
+ io_timeout = select_timeout = 30;
+ ignore_timeout = 0;
+ readfd_unbuffered(sock_f_in, io_filesfrom_buf,
+ sizeof io_filesfrom_buf);
+ }
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ total += ret;
+
+ if (fd == sock_f_out) {
+ if (io_timeout)
+ last_io = time(NULL);
+ sleep_for_bwlimit(ret);
+ }
+ }
+
+ no_flush--;
+}
+
+
+/**
+ * Write an message to a multiplexed stream. If this fails then rsync
+ * exits.
+ **/
+static void mplex_write(enum msgcode code, char *buf, size_t len)
+{
+ char buffer[4096];
+ size_t n = len;
+
+ SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
+
+ /* When the generator reads messages from the msg_fd_in pipe, it can
+ * cause output to occur down the socket. Setting contiguous_write_len
+ * prevents the reading of msg_fd_in once we actually start to write
+ * this sequence of data (though we might read it before the start). */
+ if (am_generator && msg_fd_in >= 0)
+ contiguous_write_len = len + 4;
+
+ if (n > sizeof buffer - 4)
+ n = sizeof buffer - 4;
+
+ memcpy(&buffer[4], buf, n);
+ writefd_unbuffered(sock_f_out, buffer, n+4);
+
+ len -= n;
+ buf += n;
+
+ if (len)
+ writefd_unbuffered(sock_f_out, buf, len);
+
+ if (am_generator && msg_fd_in >= 0)
+ contiguous_write_len = 0;
+}
+
+
+void io_flush(int flush_it_all)
+{
+ msg_list_push(flush_it_all);
+
+ if (!iobuf_out_cnt || no_flush)
+ return;
+
+ if (io_multiplexing_out)
+ mplex_write(MSG_DATA, iobuf_out, iobuf_out_cnt);
+ else
+ writefd_unbuffered(sock_f_out, iobuf_out, iobuf_out_cnt);
+ iobuf_out_cnt = 0;
+}
+
+
+static void writefd(int fd,char *buf,size_t len)
+{
+ if (fd == msg_fd_out) {
+ rprintf(FERROR, "Internal error: wrong write used in receiver.\n");
+ exit_cleanup(RERR_PROTOCOL);
+ }
+
+ if (fd == sock_f_out)
+ stats.total_written += len;
+
+ if (fd == write_batch_monitor_out) {
+ if ((size_t)write(batch_fd, buf, len) != len)
+ exit_cleanup(RERR_FILEIO);
+ }
+
+ if (!iobuf_out || fd != sock_f_out) {
+ writefd_unbuffered(fd, buf, len);
+ return;
+ }
+
+ while (len) {
+ int n = MIN((int)len, IO_BUFFER_SIZE - iobuf_out_cnt);
+ if (n > 0) {
+ memcpy(iobuf_out+iobuf_out_cnt, buf, n);
+ buf += n;
+ len -= n;
+ iobuf_out_cnt += n;
+ }
+
+ if (iobuf_out_cnt == IO_BUFFER_SIZE)
+ io_flush(NORMAL_FLUSH);
+ }
+}
+
+
+void write_shortint(int f, int x)
+{
+ uchar b[2];
+ b[0] = x;
+ b[1] = x >> 8;
+ writefd(f, (char *)b, 2);
+}
+
+
+void write_int(int f,int32 x)
+{
+ char b[4];
+ SIVAL(b,0,x);
+ writefd(f,b,4);
+}
+
+
+void write_int_named(int f, int32 x, const char *phase)
+{
+ io_write_phase = phase;
+ write_int(f, x);
+ io_write_phase = phase_unknown;
+}
+
+
+/*
+ * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
+ * 64-bit types on this platform.
+ */
+void write_longint(int f, int64 x)
+{
+ char b[8];
+
+ if (x <= 0x7FFFFFFF) {
+ write_int(f, (int)x);
+ return;
+ }
+
+#if SIZEOF_INT64 < 8
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+#else
+ write_int(f, (int32)0xFFFFFFFF);
+ SIVAL(b,0,(x&0xFFFFFFFF));
+ SIVAL(b,4,((x>>32)&0xFFFFFFFF));
+
+ writefd(f,b,8);
+#endif
+}
+
+void write_buf(int f,char *buf,size_t len)
+{
+ writefd(f,buf,len);
+}
+
+/** Write a string to the connection */
+void write_sbuf(int f, char *buf)
+{
+ writefd(f, buf, strlen(buf));
+}
+
+void write_byte(int f, uchar c)
+{
+ writefd(f, (char *)&c, 1);
+}
+
+void write_vstring(int f, char *str, int len)
+{
+ uchar lenbuf[3], *lb = lenbuf;
+
+ if (len > 0x7F) {
+ if (len > 0x7FFF) {
+ rprintf(FERROR,
+ "attempting to send over-long vstring (%d > %d)\n",
+ len, 0x7FFF);
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ *lb++ = len / 0x100 + 0x80;
+ }
+ *lb = len;
+
+ writefd(f, (char*)lenbuf, lb - lenbuf + 1);
+ if (len)
+ writefd(f, str, len);
+}
+
+
+/**
+ * Read a line of up to @p maxlen characters into @p buf (not counting
+ * the trailing null). Strips the (required) trailing newline and all
+ * carriage returns.
+ *
+ * @return 1 for success; 0 for I/O error or truncation.
+ **/
+int read_line(int f, char *buf, size_t maxlen)
+{
+ while (maxlen) {
+ buf[0] = 0;
+ read_buf(f, buf, 1);
+ if (buf[0] == 0)
+ return 0;
+ if (buf[0] == '\n')
+ break;
+ if (buf[0] != '\r') {
+ buf++;
+ maxlen--;
+ }
+ }
+ *buf = '\0';
+ return maxlen > 0;
+}
+
+
+void io_printf(int fd, const char *format, ...)
+{
+ va_list ap;
+ char buf[1024];
+ int len;
+
+ va_start(ap, format);
+ len = vsnprintf(buf, sizeof buf, format, ap);
+ va_end(ap);
+
+ if (len < 0)
+ exit_cleanup(RERR_STREAMIO);
+
+ write_sbuf(fd, buf);
+}
+
+
+/** Setup for multiplexing a MSG_* stream with the data stream. */
+void io_start_multiplex_out(void)
+{
+ io_flush(NORMAL_FLUSH);
+ io_start_buffering_out();
+ io_multiplexing_out = 1;
+}
+
+/** Setup for multiplexing a MSG_* stream with the data stream. */
+void io_start_multiplex_in(void)
+{
+ io_flush(NORMAL_FLUSH);
+ io_start_buffering_in();
+ io_multiplexing_in = 1;
+}
+
+/** Write an message to the multiplexed data stream. */
+int io_multiplex_write(enum msgcode code, char *buf, size_t len)
+{
+ if (!io_multiplexing_out)
+ return 0;
+
+ io_flush(NORMAL_FLUSH);
+ stats.total_written += (len+4);
+ mplex_write(code, buf, len);
+ return 1;
+}
+
+void close_multiplexing_in(void)
+{
+ io_multiplexing_in = 0;
+}
+
+/** Stop output multiplexing. */
+void close_multiplexing_out(void)
+{
+ io_multiplexing_out = 0;
+}
+
+void start_write_batch(int fd)
+{
+ write_stream_flags(batch_fd);
+
+ /* Some communication has already taken place, but we don't
+ * enable batch writing until here so that we can write a
+ * canonical record of the communication even though the
+ * actual communication so far depends on whether a daemon
+ * is involved. */
+ write_int(batch_fd, protocol_version);
+ write_int(batch_fd, checksum_seed);
+
+ if (am_sender)
+ write_batch_monitor_out = fd;
+ else
+ write_batch_monitor_in = fd;
+}
+
+void stop_write_batch(void)
+{
+ write_batch_monitor_out = -1;
+ write_batch_monitor_in = -1;
+}