+void io_end_buffering_in(BOOL free_buffers)
+{
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_buffering_in(IOBUF_%s_BUFS)\n",
+ who_am_i(), free_buffers ? "FREE" : "KEEP");
+ }
+
+ if (free_buffers)
+ free_xbuf(&iobuf.in);
+ else
+ iobuf.in.pos = iobuf.in.len = 0;
+
+ iobuf.in_fd = -1;
+}
+
+void io_end_buffering_out(BOOL free_buffers)
+{
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_buffering_out(IOBUF_%s_BUFS)\n",
+ who_am_i(), free_buffers ? "FREE" : "KEEP");
+ }
+
+ io_flush(FULL_FLUSH);
+
+ if (free_buffers) {
+ free_xbuf(&iobuf.out);
+ free_xbuf(&iobuf.msg);
+ }
+
+ iobuf.out_fd = -1;
+}
+
+void maybe_flush_socket(int important)
+{
+ if (flist_eof && iobuf.out.buf && iobuf.out.len > iobuf.out_empty_len
+ && (important || time(NULL) - last_io_out >= 5))
+ io_flush(NORMAL_FLUSH);
+}
+
+/* Older rsync versions used to send either a MSG_NOOP (protocol 30) or a
+ * raw-data-based keep-alive (protocol 29), both of which implied forwarding of
+ * the message through the sender. Since the new timeout method does not need
+ * any forwarding, we just send an empty MSG_DATA message, which works with all
+ * rsync versions. This avoids any message forwarding, and leaves the raw-data
+ * stream alone (since we can never be quite sure if that stream is in the
+ * right state for a keep-alive message). */
+void maybe_send_keepalive(time_t now, int flags)
+{
+ if (flags & MSK_ACTIVE_RECEIVER)
+ last_io_in = now; /* Fudge things when we're working hard on the files. */
+
+ if (now - last_io_out >= allowed_lull) {
+ /* The receiver is special: it only sends keep-alive messages if it is
+ * actively receiving data. Otherwise, it lets the generator timeout. */
+ if (am_receiver && now - last_io_in >= io_timeout)
+ return;
+
+ if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len)
+ send_msg(MSG_DATA, "", 0, 0);
+ if (!(flags & MSK_ALLOW_FLUSH)) {
+ /* Let the caller worry about writing out the data. */
+ } else if (iobuf.msg.len)
+ perform_io(iobuf.msg.size - iobuf.msg.len + 1, PIO_NEED_MSGROOM);
+ else if (iobuf.out.len > iobuf.out_empty_len)
+ io_flush(NORMAL_FLUSH);
+ }
+}
+
+void start_flist_forward(int ndx)
+{
+ write_int(iobuf.out_fd, ndx);
+ forward_flist_data = 1;
+}
+
+void stop_flist_forward(void)
+{
+ forward_flist_data = 0;
+}
+
+/* Read a message from a multiplexed source. */
+static void read_a_msg(void)
+{
+ char data[BIGPATHBUFLEN];
+ int tag, val;
+ size_t msg_bytes;
+
+ /* This ensures that perform_io() does not try to do any message reading
+ * until we've read all of the data for this message. We should also
+ * try to avoid calling things that will cause data to be written via
+ * perform_io() prior to this being reset to 1. */
+ iobuf.in_multiplexed = -1;
+
+ tag = raw_read_int();
+
+ msg_bytes = tag & 0xFFFFFF;
+ tag = (tag >> 24) - MPLEX_BASE;
+
+ if (DEBUG_GTE(IO, 1) && msgs2stderr)
+ rprintf(FINFO, "[%s] got msg=%d, len=%ld\n", who_am_i(), (int)tag, (long)msg_bytes);
+
+ switch (tag) {
+ case MSG_DATA:
+ assert(iobuf.raw_input_ends_before == 0);
+ /* Though this does not yet read the data, we do mark where in
+ * the buffer the msg data will end once it is read. It is
+ * possible that this points off the end of the buffer, in
+ * which case the gradual reading of the input stream will
+ * cause this value to wrap around and eventually become real. */
+ if (msg_bytes)
+ iobuf.raw_input_ends_before = iobuf.in.pos + msg_bytes;
+ iobuf.in_multiplexed = 1;
+ break;
+ case MSG_STATS:
+ if (msg_bytes != sizeof stats.total_read || !am_generator)
+ goto invalid_msg;
+ raw_read_buf((char*)&stats.total_read, sizeof stats.total_read);
+ iobuf.in_multiplexed = 1;
+ break;
+ case MSG_REDO:
+ if (msg_bytes != 4 || !am_generator)
+ goto invalid_msg;
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ got_flist_entry_status(FES_REDO, val);
+ break;
+ case MSG_IO_ERROR:
+ if (msg_bytes != 4)
+ goto invalid_msg;
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ io_error |= val;
+ if (am_receiver)
+ send_msg_int(MSG_IO_ERROR, val);
+ break;
+ case MSG_IO_TIMEOUT:
+ if (msg_bytes != 4 || am_server || am_generator)
+ goto invalid_msg;
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ if (!io_timeout || io_timeout > val) {
+ if (INFO_GTE(MISC, 2))
+ rprintf(FINFO, "Setting --timeout=%d to match server\n", val);
+ set_io_timeout(val);
+ }
+ break;
+ case MSG_NOOP:
+ /* Support protocol-30 keep-alive method. */
+ if (msg_bytes != 0)
+ goto invalid_msg;
+ iobuf.in_multiplexed = 1;
+ if (am_sender)
+ maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
+ break;
+ case MSG_DELETED:
+ if (msg_bytes >= sizeof data)
+ goto overflow;
+ if (am_generator) {
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ send_msg(MSG_DELETED, data, msg_bytes, 1);
+ break;
+ }
+#ifdef ICONV_OPTION
+ if (ic_recv != (iconv_t)-1) {
+ xbuf outbuf, inbuf;
+ char ibuf[512];
+ int add_null = 0;
+ int flags = ICB_INCLUDE_BAD | ICB_INIT;
+
+ INIT_CONST_XBUF(outbuf, data);
+ INIT_XBUF(inbuf, ibuf, 0, (size_t)-1);
+
+ while (msg_bytes) {
+ size_t len = msg_bytes > sizeof ibuf - inbuf.len ? sizeof ibuf - inbuf.len : msg_bytes;
+ raw_read_buf(ibuf + inbuf.len, len);
+ inbuf.pos = 0;
+ inbuf.len += len;
+ if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])
+ inbuf.len--, add_null = 1;
+ if (iconvbufs(ic_send, &inbuf, &outbuf, flags) < 0) {
+ if (errno == E2BIG)
+ goto overflow;
+ /* Buffer ended with an incomplete char, so move the
+ * bytes to the start of the buffer and continue. */
+ memmove(ibuf, ibuf + inbuf.pos, inbuf.len);
+ }
+ flags &= ~ICB_INIT;
+ }
+ if (add_null) {
+ if (outbuf.len == outbuf.size)
+ goto overflow;
+ outbuf.buf[outbuf.len++] = '\0';
+ }
+ msg_bytes = outbuf.len;
+ } else
+#endif
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ /* A directory name was sent with the trailing null */
+ if (msg_bytes > 0 && !data[msg_bytes-1])
+ log_delete(data, S_IFDIR);
+ else {
+ data[msg_bytes] = '\0';
+ log_delete(data, S_IFREG);
+ }
+ break;
+ case MSG_SUCCESS:
+ if (msg_bytes != 4) {
+ invalid_msg:
+ rprintf(FERROR, "invalid multi-message %d:%lu [%s%s]\n",
+ tag, (unsigned long)msg_bytes, who_am_i(),
+ inc_recurse ? "/inc" : "");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ if (am_generator)
+ got_flist_entry_status(FES_SUCCESS, val);
+ else
+ successful_send(val);
+ break;
+ case MSG_NO_SEND:
+ if (msg_bytes != 4)
+ goto invalid_msg;
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ if (am_generator)
+ got_flist_entry_status(FES_NO_SEND, val);
+ else
+ send_msg_int(MSG_NO_SEND, val);
+ break;
+ case MSG_ERROR_SOCKET:
+ case MSG_ERROR_UTF8:
+ case MSG_CLIENT:
+ case MSG_LOG:
+ if (!am_generator)
+ goto invalid_msg;
+ if (tag == MSG_ERROR_SOCKET)
+ msgs2stderr = 1;
+ /* FALL THROUGH */
+ case MSG_INFO:
+ case MSG_ERROR:
+ case MSG_ERROR_XFER:
+ case MSG_WARNING:
+ if (msg_bytes >= sizeof data) {
+ overflow:
+ rprintf(FERROR,
+ "multiplexing overflow %d:%lu [%s%s]\n",
+ tag, (unsigned long)msg_bytes, who_am_i(),
+ inc_recurse ? "/inc" : "");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ rwrite((enum logcode)tag, data, msg_bytes, !am_generator);
+ if (first_message) {
+ if (list_only && !am_sender && tag == 1 && msg_bytes < sizeof data) {
+ data[msg_bytes] = '\0';
+ check_for_d_option_error(data);
+ }
+ first_message = 0;
+ }
+ break;
+ case MSG_ERROR_EXIT:
+ if (msg_bytes == 4)
+ val = raw_read_int();
+ else if (msg_bytes == 0)
+ val = 0;
+ else
+ goto invalid_msg;
+ iobuf.in_multiplexed = 1;
+ if (DEBUG_GTE(EXIT, 3))
+ rprintf(FINFO, "[%s] got MSG_ERROR_EXIT with %ld bytes\n", who_am_i(), (long)msg_bytes);
+ if (msg_bytes == 0) {
+ if (!am_sender && !am_generator) {
+ if (DEBUG_GTE(EXIT, 3)) {
+ rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT (len 0)\n",
+ who_am_i());
+ }
+ send_msg(MSG_ERROR_EXIT, "", 0, 0);
+ io_flush(FULL_FLUSH);
+ }
+ } else if (protocol_version >= 31) {
+ if (am_generator) {
+ if (DEBUG_GTE(EXIT, 3)) {
+ rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT with exit_code %d\n",
+ who_am_i(), val);
+ }
+ send_msg_int(MSG_ERROR_EXIT, val);
+ } else {
+ if (DEBUG_GTE(EXIT, 3)) {
+ rprintf(FINFO, "[%s] sending MSG_ERROR_EXIT (len 0)\n",
+ who_am_i());
+ }
+ send_msg(MSG_ERROR_EXIT, "", 0, 0);
+ }
+ }
+ /* Send a negative linenum so that we don't end up
+ * with a duplicate exit message. */
+ _exit_cleanup(val, __FILE__, 0 - __LINE__);
+ default:
+ rprintf(FERROR, "unexpected tag %d [%s%s]\n",
+ tag, who_am_i(), inc_recurse ? "/inc" : "");
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ assert(iobuf.in_multiplexed > 0);
+}
+
+static void drain_multiplex_messages(void)
+{
+ while (IN_MULTIPLEXED_AND_READY && iobuf.in.len) {
+ if (iobuf.raw_input_ends_before) {
+ size_t raw_len = iobuf.raw_input_ends_before - iobuf.in.pos;
+ iobuf.raw_input_ends_before = 0;
+ if (raw_len >= iobuf.in.len) {
+ iobuf.in.len = 0;
+ break;
+ }
+ iobuf.in.len -= raw_len;
+ if ((iobuf.in.pos += raw_len) >= iobuf.in.size)
+ iobuf.in.pos -= iobuf.in.size;
+ }
+ read_a_msg();
+ }
+}
+
+void wait_for_receiver(void)
+{
+ if (!iobuf.raw_input_ends_before)
+ read_a_msg();
+
+ if (iobuf.raw_input_ends_before) {
+ int ndx = read_int(iobuf.in_fd);
+ if (ndx < 0) {
+ switch (ndx) {
+ case NDX_FLIST_EOF:
+ flist_eof = 1;
+ if (DEBUG_GTE(FLIST, 3))
+ rprintf(FINFO, "[%s] flist_eof=1\n", who_am_i());
+ break;
+ case NDX_DONE:
+ msgdone_cnt++;
+ break;
+ default:
+ exit_cleanup(RERR_STREAMIO);
+ }
+ } else {
+ struct file_list *flist;
+ flist_receiving_enabled = False;
+ if (DEBUG_GTE(FLIST, 2)) {
+ rprintf(FINFO, "[%s] receiving flist for dir %d\n",
+ who_am_i(), ndx);
+ }
+ flist = recv_file_list(iobuf.in_fd);
+ flist->parent_ndx = ndx;
+#ifdef SUPPORT_HARD_LINKS
+ if (preserve_hard_links)
+ match_hard_links(flist);
+#endif
+ flist_receiving_enabled = True;
+ }
+ }
+}
+
+unsigned short read_shortint(int f)
+{
+ char b[2];
+ read_buf(f, b, 2);
+ return (UVAL(b, 1) << 8) + UVAL(b, 0);
+}
+
+int32 read_int(int f)
+{
+ char b[4];
+ int32 num;
+
+ read_buf(f, b, 4);
+ num = IVAL(b, 0);
+#if SIZEOF_INT32 > 4
+ if (num & (int32)0x80000000)
+ num |= ~(int32)0xffffffff;
+#endif
+ return num;
+}
+
+int32 read_varint(int f)
+{
+ union {
+ char b[5];
+ int32 x;
+ } u;
+ uchar ch;
+ int extra;
+
+ u.x = 0;
+ ch = read_byte(f);
+ 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);
+ }
+ read_buf(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
+ read_buf(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);
+ }
+ read_buf(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
+ read_buf(f, b, 8);
+ return IVAL(b,0) | (((int64)IVAL(b,4))<<32);
+#endif
+}
+
+void read_buf(int f, char *buf, size_t len)
+{
+ if (f != iobuf.in_fd) {
+ if (safe_read(f, buf, len) != len)
+ whine_about_eof(False); /* Doesn't return. */
+ goto batch_copy;
+ }
+
+ if (!IN_MULTIPLEXED) {
+ raw_read_buf(buf, len);
+ total_data_read += len;
+ if (forward_flist_data)
+ write_buf(iobuf.out_fd, buf, len);
+ batch_copy:
+ if (f == write_batch_monitor_in)
+ safe_write(batch_fd, buf, len);
+ return;
+ }
+
+ while (1) {
+ size_t siz;
+
+ while (!iobuf.raw_input_ends_before)
+ read_a_msg();
+
+ siz = MIN(len, iobuf.raw_input_ends_before - iobuf.in.pos);
+ if (siz >= iobuf.in.size)
+ siz = iobuf.in.size;
+ raw_read_buf(buf, siz);
+ total_data_read += siz;
+
+ if (forward_flist_data)
+ write_buf(iobuf.out_fd, buf, siz);
+
+ if (f == write_batch_monitor_in)
+ safe_write(batch_fd, buf, siz);
+
+ if ((len -= siz) == 0)
+ break;
+ buf += siz;
+ }
+}
+
+void read_sbuf(int f, char *buf, size_t len)
+{
+ read_buf(f, buf, len);
+ buf[len] = '\0';
+}
+
+uchar read_byte(int f)
+{
+ uchar c;
+ read_buf(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)
+ read_buf(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)
+{
+ int32 max_blength = protocol_version < 30 ? OLD_MAX_BLOCK_SIZE : MAX_BLOCK_SIZE;
+ 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_blength) {
+ 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 */
+
+ 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);
+}
+
+void io_flush(int flush_it_all)
+{
+ if (iobuf.out.len > iobuf.out_empty_len) {
+ if (flush_it_all) /* FULL_FLUSH: flush everything in the output buffers */
+ perform_io(iobuf.out.size - iobuf.out_empty_len, PIO_NEED_OUTROOM);
+ else /* NORMAL_FLUSH: flush at least 1 byte */
+ perform_io(iobuf.out.size - iobuf.out.len + 1, PIO_NEED_OUTROOM);
+ }
+ if (iobuf.msg.len)
+ perform_io(iobuf.msg.size, PIO_NEED_MSGROOM);
+}
+
+void write_shortint(int f, unsigned short x)
+{
+ char b[2];
+ b[0] = (char)x;
+ b[1] = (char)(x >> 8);
+ write_buf(f, b, 2);
+}
+
+void write_int(int f, int32 x)
+{
+ char b[4];
+ SIVAL(b, 0, x);
+ write_buf(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];
+
+ write_buf(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);
+ }