+ u.x = 0;
+ readfd(f, (char*)&ch, 1);
+ extra = int_byte_extra[ch / 4];
+ if (extra) {
+ uchar bit = ((uchar)1<<(8-extra));
+ if (extra >= (int)sizeof u.b) {
+ rprintf(FERROR, "Overflow in read_varint()\n");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ readfd(f, u.b, extra);
+ u.b[extra] = ch & (bit-1);
+ } else
+ u.b[0] = ch;
+#if CAREFUL_ALIGNMENT
+ u.x = IVAL(u.b,0);
+#endif
+#if SIZEOF_INT32 > 4
+ if (u.x & (int32)0x80000000)
+ u.x |= ~(int32)0xffffffff;
+#endif
+ return u.x;
+}
+
+int64 read_varlong(int f, uchar min_bytes)
+{
+ union {
+ char b[9];
+ int64 x;
+ } u;
+ char b2[8];
+ int extra;
+
+#if SIZEOF_INT64 < 8
+ memset(u.b, 0, 8);
+#else
+ u.x = 0;
+#endif
+ readfd(f, b2, min_bytes);
+ memcpy(u.b, b2+1, min_bytes-1);
+ extra = int_byte_extra[CVAL(b2, 0) / 4];
+ if (extra) {
+ uchar bit = ((uchar)1<<(8-extra));
+ if (min_bytes + extra > (int)sizeof u.b) {
+ rprintf(FERROR, "Overflow in read_varlong()\n");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ readfd(f, u.b + min_bytes - 1, extra);
+ u.b[min_bytes + extra - 1] = CVAL(b2, 0) & (bit-1);
+#if SIZEOF_INT64 < 8
+ if (min_bytes + extra > 5 || u.b[4] || CVAL(u.b,3) & 0x80) {
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+ }
+#endif
+ } else
+ u.b[min_bytes + extra - 1] = CVAL(b2, 0);
+#if SIZEOF_INT64 < 8
+ u.x = IVAL(u.b,0);
+#elif CAREFUL_ALIGNMENT
+ u.x = IVAL(u.b,0) | (((int64)IVAL(u.b,4))<<32);
+#endif
+ return u.x;
+}
+
+int64 read_longint(int f)
+{
+#if SIZEOF_INT64 >= 8
+ char b[9];
+#endif
+ int32 num = read_int(f);
+
+ if (num != (int32)0xffffffff)
+ return num;
+
+#if SIZEOF_INT64 < 8
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+#else
+ readfd(f, b, 8);
+ return IVAL(b,0) | (((int64)IVAL(b,4))<<32);
+#endif
+}
+
+void read_buf(int f, char *buf, size_t len)
+{
+ readfd(f,buf,len);
+}
+
+void read_sbuf(int f, char *buf, size_t len)
+{
+ readfd(f, buf, len);
+ buf[len] = '\0';
+}
+
+uchar read_byte(int f)
+{
+ uchar c;
+ readfd(f, (char *)&c, 1);
+ return c;
+}
+
+int read_vstring(int f, char *buf, int bufsize)
+{
+ int len = read_byte(f);
+
+ if (len & 0x80)
+ len = (len & ~0x80) * 0x100 + read_byte(f);
+
+ if (len >= bufsize) {
+ rprintf(FERROR, "over-long vstring received (%d > %d)\n",
+ len, bufsize - 1);
+ return -1;
+ }
+
+ if (len)
+ readfd(f, buf, len);
+ buf[len] = '\0';
+ return len;
+}
+
+/* Populate a sum_struct with values from the socket. This is
+ * called by both the sender and the receiver. */
+void read_sum_head(int f, struct sum_struct *sum)
+{
+ sum->count = read_int(f);
+ if (sum->count < 0) {
+ rprintf(FERROR, "Invalid checksum count %ld [%s]\n",
+ (long)sum->count, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ sum->blength = read_int(f);
+ if (sum->blength < 0 || sum->blength > MAX_BLOCK_SIZE) {
+ rprintf(FERROR, "Invalid block length %ld [%s]\n",
+ (long)sum->blength, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ sum->s2length = protocol_version < 27 ? csum_length : (int)read_int(f);
+ if (sum->s2length < 0 || sum->s2length > MAX_DIGEST_LEN) {
+ rprintf(FERROR, "Invalid checksum length %d [%s]\n",
+ sum->s2length, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+ sum->remainder = read_int(f);
+ if (sum->remainder < 0 || sum->remainder > sum->blength) {
+ rprintf(FERROR, "Invalid remainder length %ld [%s]\n",
+ (long)sum->remainder, who_am_i());
+ exit_cleanup(RERR_PROTOCOL);
+ }
+}
+
+/* Send the values from a sum_struct over the socket. Set sum to
+ * NULL if there are no checksums to send. This is called by both
+ * the generator and the sender. */
+void write_sum_head(int f, struct sum_struct *sum)
+{
+ static struct sum_struct null_sum;
+
+ if (sum == NULL)
+ sum = &null_sum;
+
+ write_int(f, sum->count);
+ write_int(f, sum->blength);
+ if (protocol_version >= 27)
+ write_int(f, sum->s2length);
+ write_int(f, sum->remainder);
+}
+
+/**
+ * Sleep after writing to limit I/O bandwidth usage.
+ *
+ * @todo Rather than sleeping after each write, it might be better to
+ * use some kind of averaging. The current algorithm seems to always
+ * use a bit less bandwidth than specified, because it doesn't make up
+ * for slow periods. But arguably this is a feature. In addition, we
+ * ought to take the time used to write the data into account.
+ *
+ * During some phases of big transfers (file FOO is uptodate) this is
+ * called with a small bytes_written every time. As the kernel has to
+ * round small waits up to guarantee that we actually wait at least the
+ * requested number of microseconds, this can become grossly inaccurate.
+ * We therefore keep track of the bytes we've written over time and only
+ * sleep when the accumulated delay is at least 1 tenth of a second.
+ **/
+static void sleep_for_bwlimit(int bytes_written)
+{
+ static struct timeval prior_tv;
+ static long total_written = 0;
+ struct timeval tv, start_tv;
+ long elapsed_usec, sleep_usec;
+
+#define ONE_SEC 1000000L /* # of microseconds in a second */
+
+ if (!bwlimit_writemax)
+ return;
+
+ total_written += bytes_written;
+
+ gettimeofday(&start_tv, NULL);
+ if (prior_tv.tv_sec) {
+ elapsed_usec = (start_tv.tv_sec - prior_tv.tv_sec) * ONE_SEC
+ + (start_tv.tv_usec - prior_tv.tv_usec);
+ total_written -= elapsed_usec * bwlimit / (ONE_SEC/1024);
+ if (total_written < 0)
+ total_written = 0;
+ }
+
+ sleep_usec = total_written * (ONE_SEC/1024) / bwlimit;
+ if (sleep_usec < ONE_SEC / 10) {
+ prior_tv = start_tv;
+ return;
+ }
+
+ tv.tv_sec = sleep_usec / ONE_SEC;
+ tv.tv_usec = sleep_usec % ONE_SEC;
+ select(0, NULL, NULL, NULL, &tv);
+
+ gettimeofday(&prior_tv, NULL);
+ elapsed_usec = (prior_tv.tv_sec - start_tv.tv_sec) * ONE_SEC
+ + (prior_tv.tv_usec - start_tv.tv_usec);
+ total_written = (sleep_usec - elapsed_usec) * bwlimit / (ONE_SEC/1024);
+}
+
+/* Write len bytes to the file descriptor fd, looping as necessary to get
+ * the job done and also (in certain circumstances) 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, const char *buf, size_t len)
+{
+ size_t n, total = 0;
+ fd_set w_fds, r_fds, e_fds;
+ int maxfd, count, cnt, using_r_fds;
+ int defer_inc = 0;
+ struct timeval tv;
+
+ if (no_flush++)
+ defer_forwarding_messages++, defer_inc++;
+
+ while (total < len) {
+ FD_ZERO(&w_fds);
+ FD_SET(fd, &w_fds);
+ FD_ZERO(&e_fds);
+ FD_SET(fd, &e_fds);
+ maxfd = fd;
+
+ if (msg_fd_in >= 0) {
+ FD_ZERO(&r_fds);
+ FD_SET(msg_fd_in, &r_fds);
+ if (msg_fd_in > maxfd)
+ maxfd = msg_fd_in;
+ using_r_fds = 1;
+ } else
+ using_r_fds = 0;
+
+ tv.tv_sec = select_timeout;
+ tv.tv_usec = 0;
+
+ errno = 0;
+ count = select(maxfd + 1, using_r_fds ? &r_fds : NULL,
+ &w_fds, &e_fds, &tv);
+
+ if (count <= 0) {
+ if (count < 0 && errno == EBADF)
+ exit_cleanup(RERR_SOCKETIO);
+ check_timeout();
+ continue;
+ }
+
+ /*if (FD_ISSET(fd, &e_fds))
+ rprintf(FINFO, "select exception on fd %d\n", fd); */
+
+ if (using_r_fds && FD_ISSET(msg_fd_in, &r_fds))
+ read_msg_fd();
+
+ if (!FD_ISSET(fd, &w_fds))
+ continue;
+
+ n = len - total;
+ if (bwlimit_writemax && n > bwlimit_writemax)
+ n = bwlimit_writemax;
+ cnt = write(fd, buf + total, n);
+
+ if (cnt <= 0) {
+ if (cnt < 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)
+ io_end_multiplex_out();
+ /* Don't try to write errors down a failing msg pipe. */
+ if (am_server && fd == msg_fd_out)
+ exit_cleanup(RERR_STREAMIO);
+ rsyserr(FERROR, errno,
+ "writefd_unbuffered failed to write %ld bytes [%s]",
+ (long)len, who_am_i());
+ /* If the other side is sending us error messages, try
+ * to grab any messages they sent before they died. */
+ while (!am_server && fd == sock_f_out && io_multiplexing_in) {
+ char buf[1024];
+ set_io_timeout(30);
+ ignore_timeout = 0;
+ readfd_unbuffered(sock_f_in, buf, sizeof buf);
+ }
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ total += cnt;
+ defer_forwarding_messages++, defer_inc++;
+
+ if (fd == sock_f_out) {
+ if (io_timeout || am_generator)
+ last_io_out = time(NULL);
+ sleep_for_bwlimit(cnt);
+ }
+ }
+
+ no_flush--;
+ defer_inc -= defer_forwarding_keep;
+ if (!(defer_forwarding_messages -= defer_inc) && !no_flush)
+ msg_flush();
+}
+
+void io_flush(int flush_it_all)
+{
+ if (!iobuf_out_cnt || no_flush)
+ return;
+
+ if (io_multiplexing_out)
+ mplex_write(sock_f_out, MSG_DATA, iobuf_out, iobuf_out_cnt, 0);
+ else
+ writefd_unbuffered(iobuf_f_out, iobuf_out, iobuf_out_cnt);
+ iobuf_out_cnt = 0;
+
+ if (flush_it_all && !defer_forwarding_messages)
+ msg_flush();
+}
+
+static void writefd(int fd, const char *buf, size_t len)
+{
+ 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 != iobuf_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, unsigned short x)
+{
+ char b[2];
+ b[0] = (char)x;
+ b[1] = (char)(x >> 8);
+ writefd(f, b, 2);
+}
+
+void write_int(int f, int32 x)
+{
+ char b[4];
+ SIVAL(b, 0, x);
+ writefd(f, b, 4);
+}
+
+void write_varint(int f, int32 x)
+{
+ char b[5];
+ uchar bit;
+ int cnt = 4;
+
+ SIVAL(b, 1, x);
+
+ while (cnt > 1 && b[cnt] == 0)
+ cnt--;
+ bit = ((uchar)1<<(7-cnt+1));
+ if (CVAL(b, cnt) >= bit) {
+ cnt++;
+ *b = ~(bit-1);
+ } else if (cnt > 1)
+ *b = b[cnt] | ~(bit*2-1);
+ else
+ *b = b[cnt];
+
+ writefd(f, b, cnt);
+}
+
+void write_varlong(int f, int64 x, uchar min_bytes)
+{
+ char b[9];
+ uchar bit;
+ int cnt = 8;
+
+ SIVAL(b, 1, x);
+#if SIZEOF_INT64 >= 8
+ SIVAL(b, 5, x >> 32);
+#else
+ if (x <= 0x7FFFFFFF && x >= 0)
+ memset(b + 5, 0, 4);
+ else {
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+ }
+#endif
+
+ while (cnt > min_bytes && b[cnt] == 0)
+ cnt--;
+ bit = ((uchar)1<<(7-cnt+min_bytes));
+ if (CVAL(b, cnt) >= bit) {
+ cnt++;
+ *b = ~(bit-1);
+ } else if (cnt > min_bytes)
+ *b = b[cnt] | ~(bit*2-1);
+ else
+ *b = b[cnt];
+
+ writefd(f, b, cnt);
+}
+
+/*
+ * 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[12], * const s = b+4;
+
+ SIVAL(s, 0, x);
+ if (x <= 0x7FFFFFFF && x >= 0) {
+ writefd(f, s, 4);
+ return;
+ }
+
+#if SIZEOF_INT64 < 8
+ rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
+ exit_cleanup(RERR_UNSUPPORTED);
+#else
+ memset(b, 0xFF, 4);
+ SIVAL(s, 4, x >> 32);
+ writefd(f, b, 12);
+#endif
+}
+
+void write_buf(int f, const char *buf, size_t len)
+{
+ writefd(f,buf,len);
+}
+
+/** Write a string to the connection */
+void write_sbuf(int f, const char *buf)
+{
+ writefd(f, buf, strlen(buf));
+}
+
+void write_byte(int f, uchar c)
+{
+ writefd(f, (char *)&c, 1);
+}
+
+void write_vstring(int f, const 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);
+}
+
+/* Send a file-list index using a byte-reduction method. */
+void write_ndx(int f, int32 ndx)
+{
+ static int32 prev_positive = -1, prev_negative = 1;
+ int32 diff, cnt = 0;
+ char b[6];
+
+ if (protocol_version < 30 || read_batch) {
+ write_int(f, ndx);
+ return;
+ }
+
+ /* Send NDX_DONE as a single-byte 0 with no side effects. Send
+ * negative nums as a positive after sending a leading 0xFF. */
+ if (ndx >= 0) {
+ diff = ndx - prev_positive;
+ prev_positive = ndx;
+ } else if (ndx == NDX_DONE) {
+ *b = 0;
+ writefd(f, b, 1);
+ return;
+ } else {
+ b[cnt++] = (char)0xFF;
+ ndx = -ndx;
+ diff = ndx - prev_negative;
+ prev_negative = ndx;
+ }
+
+ /* A diff of 1 - 253 is sent as a one-byte diff; a diff of 254 - 32767
+ * or 0 is sent as a 0xFE + a two-byte diff; otherwise we send 0xFE
+ * & all 4 bytes of the (non-negative) num with the high-bit set. */
+ if (diff < 0xFE && diff > 0)
+ b[cnt++] = (char)diff;
+ else if (diff < 0 || diff > 0x7FFF) {
+ b[cnt++] = (char)0xFE;
+ b[cnt++] = (char)((ndx >> 24) | 0x80);
+ b[cnt++] = (char)ndx;
+ b[cnt++] = (char)(ndx >> 8);
+ b[cnt++] = (char)(ndx >> 16);
+ } else {
+ b[cnt++] = (char)0xFE;
+ b[cnt++] = (char)(diff >> 8);
+ b[cnt++] = (char)diff;
+ }
+ writefd(f, b, cnt);
+}
+
+/* Receive a file-list index using a byte-reduction method. */
+int32 read_ndx(int f)