* Copyright (C) 1996-2001 Andrew Tridgell
* Copyright (C) 1996 Paul Mackerras
* Copyright (C) 2001, 2002 Martin Pool <mbp@samba.org>
- * Copyright (C) 2003-2009 Wayne Davison
+ * Copyright (C) 2003-2022 Wayne Davison
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
extern size_t bwlimit_writemax;
extern int io_timeout;
extern int am_server;
-extern int am_daemon;
extern int am_sender;
+extern int am_receiver;
extern int am_generator;
+extern int local_server;
extern int msgs2stderr;
extern int inc_recurse;
extern int io_error;
+extern int batch_fd;
extern int eol_nulls;
extern int flist_eof;
extern int file_total;
extern int file_old_total;
extern int list_only;
extern int read_batch;
+extern int compat_flags;
extern int protect_args;
extern int checksum_seed;
+extern int daemon_connection;
extern int protocol_version;
extern int remove_source_files;
extern int preserve_hard_links;
extern BOOL extra_flist_sending_enabled;
+extern BOOL flush_ok_after_signal;
extern struct stats stats;
+extern time_t stop_at_utime;
extern struct file_list *cur_flist;
#ifdef ICONV_OPTION
extern int filesfrom_convert;
int csum_length = SHORT_SUM_LENGTH; /* initial value */
int allowed_lull = 0;
-int ignore_timeout = 0;
-int batch_fd = -1;
int msgdone_cnt = 0;
int forward_flist_data = 0;
+BOOL flist_receiving_enabled = False;
/* Ignore an EOF error if non-zero. See whine_about_eof(). */
int kluge_around_eof = 0;
+int got_kill_signal = -1; /* is set to 0 only after multiplexed I/O starts */
int sock_f_in = -1;
int sock_f_out = -1;
int64 total_data_read = 0;
int64 total_data_written = 0;
+char num_dev_ino_buf[4 + 8 + 8];
+
static struct {
xbuf in, out, msg;
int in_fd;
int out_fd; /* Both "out" and "msg" go to this fd. */
- BOOL in_multiplexed;
+ int in_multiplexed;
unsigned out_empty_len;
size_t raw_data_header_pos; /* in the out xbuf */
size_t raw_flushing_ends_before; /* in the out xbuf */
2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 5, 6, /* (C0 - FF)/4 */
};
-#define IN_MULTIPLEXED (iobuf.in_multiplexed)
+/* Our I/O buffers are sized with no bits on in the lowest byte of the "size"
+ * (indeed, our rounding of sizes in 1024-byte units assures more than this).
+ * This allows the code that is storing bytes near the physical end of a
+ * circular buffer to temporarily reduce the buffer's size (in order to make
+ * some storing idioms easier), while also making it simple to restore the
+ * buffer's actual size when the buffer's "pos" wraps around to the start (we
+ * just round the buffer's size up again). */
+
+#define IOBUF_WAS_REDUCED(siz) ((siz) & 0xFF)
+#define IOBUF_RESTORE_SIZE(siz) (((siz) | 0xFF) + 1)
+
+#define IN_MULTIPLEXED (iobuf.in_multiplexed != 0)
+#define IN_MULTIPLEXED_AND_READY (iobuf.in_multiplexed > 0)
#define OUT_MULTIPLEXED (iobuf.out_empty_len != 0)
#define PIO_NEED_INPUT (1<<0) /* The *_NEED_* flags are mutually exclusive. */
static flist_ndx_list redo_list, hlink_list;
+static void read_a_msg(void);
+static void drain_multiplex_messages(void);
static void sleep_for_bwlimit(int bytes_written);
-static void check_timeout(void)
+static void check_timeout(BOOL allow_keepalive, int keepalive_flags)
{
- time_t t;
-
- if (!io_timeout || ignore_timeout)
+ time_t t, chk;
+
+ /* On the receiving side, the generator is now the one that decides
+ * when a timeout has occurred. When it is sifting through a lot of
+ * files looking for work, it will be sending keep-alive messages to
+ * the sender, and even though the receiver won't be sending/receiving
+ * anything (not even keep-alive messages), the successful writes to
+ * the sender will keep things going. If the receiver is actively
+ * receiving data, it will ensure that the generator knows that it is
+ * not idle by sending the generator keep-alive messages (since the
+ * generator might be blocked trying to send checksums, it needs to
+ * know that the receiver is active). Thus, as long as one or the
+ * other is successfully doing work, the generator will not timeout. */
+ if (!io_timeout)
return;
- if (!last_io_in) {
- last_io_in = time(NULL);
- return;
+ t = time(NULL);
+
+ if (allow_keepalive) {
+ /* This may put data into iobuf.msg w/o flushing. */
+ maybe_send_keepalive(t, keepalive_flags);
}
- t = time(NULL);
+ if (!last_io_in)
+ last_io_in = t;
- if (t - last_io_in >= io_timeout) {
- if (!am_server && !am_daemon) {
- rprintf(FERROR, "io timeout after %d seconds -- exiting\n",
- (int)(t-last_io_in));
- }
+ if (am_receiver)
+ return;
+
+ chk = MAX(last_io_out, last_io_in);
+ if (t - chk >= io_timeout) {
+ if (am_server)
+ msgs2stderr = 1;
+ rprintf(FERROR, "[%s] io timeout after %d seconds -- exiting\n",
+ who_am_i(), (int)(t-chk));
exit_cleanup(RERR_TIMEOUT);
}
}
* There is another case for older protocol versions (< 24) where the module
* listing was not terminated, so we must ignore an EOF error in that case and
* exit. In this situation, kluge_around_eof will be > 0. */
-static NORETURN void whine_about_eof(int fd)
+static NORETURN void whine_about_eof(BOOL allow_kluge)
{
- if (kluge_around_eof && fd == sock_f_in) {
+ if (kluge_around_eof && allow_kluge) {
int i;
if (kluge_around_eof > 0)
exit_cleanup(0);
* the socket except very early in the transfer. */
static size_t safe_read(int fd, char *buf, size_t len)
{
- size_t got;
- int n;
+ size_t got = 0;
assert(fd != iobuf.in_fd);
- n = read(fd, buf, len);
- if ((size_t)n == len || n == 0) {
- if (DEBUG_GTE(IO, 2))
- rprintf(FINFO, "[%s] safe_read(%d)=%ld\n", who_am_i(), fd, (long)n);
- return n;
- }
- if (n < 0) {
- if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) {
- read_failed:
- rsyserr(FERROR, errno, "safe_read failed to read %ld bytes [%s]",
- (long)len, who_am_i());
- exit_cleanup(RERR_STREAMIO);
- }
- got = 0;
- } else
- got = n;
-
while (1) {
struct timeval tv;
fd_set r_fds, e_fds;
cnt = select(fd+1, &r_fds, NULL, &e_fds, &tv);
if (cnt <= 0) {
if (cnt < 0 && errno == EBADF) {
- rsyserr(FERROR, errno, "safe_read select failed [%s]",
- who_am_i());
+ rsyserr(FERROR, errno, "safe_read select failed");
exit_cleanup(RERR_FILEIO);
}
- check_timeout();
+ check_timeout(1, MSK_ALLOW_FLUSH);
continue;
}
rprintf(FINFO, "select exception on fd %d\n", fd); */
if (FD_ISSET(fd, &r_fds)) {
- n = read(fd, buf + got, len - got);
- if (DEBUG_GTE(IO, 2))
- rprintf(FINFO, "[%s] safe_read(%d)=%ld\n", who_am_i(), fd, (long)n);
+ ssize_t n = read(fd, buf + got, len - got);
+ if (DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] safe_read(%d)=%" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), fd, (SIZE_T_FMT_CAST)n);
+ }
if (n == 0)
break;
if (n < 0) {
if (errno == EINTR)
continue;
- goto read_failed;
+ rsyserr(FERROR, errno, "safe_read failed to read %" SIZE_T_FMT_MOD "d bytes",
+ (SIZE_T_FMT_CAST)len);
+ exit_cleanup(RERR_STREAMIO);
}
if ((got += (size_t)n) == len)
break;
* is not used on the socket except very early in the transfer. */
static void safe_write(int fd, const char *buf, size_t len)
{
- int n;
+ ssize_t n;
assert(fd != iobuf.out_fd);
if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) {
write_failed:
rsyserr(FERROR, errno,
- "safe_write failed to write %ld bytes to %s [%s]",
- (long)len, what_fd_is(fd), who_am_i());
+ "safe_write failed to write %" SIZE_T_FMT_MOD "d bytes to %s",
+ (SIZE_T_FMT_CAST)len, what_fd_is(fd));
exit_cleanup(RERR_STREAMIO);
}
} else {
cnt = select(fd + 1, NULL, &w_fds, NULL, &tv);
if (cnt <= 0) {
if (cnt < 0 && errno == EBADF) {
- rsyserr(FERROR, errno, "safe_write select failed on %s [%s]",
- what_fd_is(fd), who_am_i());
+ rsyserr(FERROR, errno, "safe_write select failed on %s", what_fd_is(fd));
exit_cleanup(RERR_FILEIO);
}
- check_timeout();
+ if (io_timeout)
+ maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
continue;
}
* a chunk of data and put it into the output buffer. */
static void forward_filesfrom_data(void)
{
- int len;
+ ssize_t len;
len = read(ff_forward_fd, ff_xb.buf + ff_xb.len, ff_xb.size - ff_xb.len);
if (len <= 0) {
free_xbuf(&ff_xb);
if (ff_reenable_multiplex >= 0)
io_start_multiplex_out(ff_reenable_multiplex);
+ free_implied_include_partial_string();
}
return;
}
- if (DEBUG_GTE(IO, 2))
- rprintf(FINFO, "[%s] files-from read=%ld\n", who_am_i(), (long)len);
+ if (DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] files-from read=%" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), (SIZE_T_FMT_CAST)len);
+ }
#ifdef ICONV_OPTION
len += ff_xb.len;
while (s != eob) {
if (*s++ == '\0') {
ff_xb.len = s - sob - 1;
+ add_implied_include(sob, 0);
if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0)
exit_cleanup(RERR_PROTOCOL); /* impossible? */
write_buf(iobuf.out_fd, s-1, 1); /* Send the '\0'. */
ff_lastchar = '\0';
else {
/* Handle a partial string specially, saving any incomplete chars. */
+ implied_include_partial_string(sob, s);
flags &= ~ICB_INCLUDE_INCOMPLETE;
if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0) {
if (errno == E2BIG)
char *f = ff_xb.buf + ff_xb.pos;
char *t = ff_xb.buf;
char *eob = f + len;
+ char *cur = t;
/* Eliminate any multi-'\0' runs. */
while (f != eob) {
if (!(*t++ = *f++)) {
+ add_implied_include(cur, 0);
+ cur = t;
while (f != eob && *f == '\0')
f++;
}
}
+ implied_include_partial_string(cur, t);
ff_lastchar = f[-1];
if ((len = t - ff_xb.buf) != 0) {
/* This will not circle back to perform_io() because we only get
}
}
-/* 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.
+void reduce_iobuf_size(xbuf *out, size_t new_size)
+{
+ if (new_size < out->size) {
+ /* Avoid weird buffer interactions by only outputting this to stderr. */
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 4)) {
+ const char *name = out == &iobuf.out ? "iobuf.out"
+ : out == &iobuf.msg ? "iobuf.msg"
+ : NULL;
+ if (name) {
+ rprintf(FINFO, "[%s] reduced size of %s (-%d)\n",
+ who_am_i(), name, (int)(out->size - new_size));
+ }
+ }
+ out->size = new_size;
+ }
+}
+
+void restore_iobuf_size(xbuf *out)
+{
+ if (IOBUF_WAS_REDUCED(out->size)) {
+ size_t new_size = IOBUF_RESTORE_SIZE(out->size);
+ /* Avoid weird buffer interactions by only outputting this to stderr. */
+ if (msgs2stderr == 1 && 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;
+ }
+}
+
+static void handle_kill_signal(BOOL flush_ok)
+{
+ got_kill_signal = -1;
+ flush_ok_after_signal = flush_ok;
+ exit_cleanup(RERR_SIGNAL);
+}
+
+/* Perform buffered input and/or output until specified conditions are met.
+ * When given a "needed" read or write request, this returns without doing any
+ * I/O if the needed input bytes or write space is already available. Once I/O
+ * is needed, this will try to do whatever reading and/or writing is currently
+ * possible, up to the maximum buffer allowances, no matter if this is a read
+ * or write request. However, the I/O stops as soon as the required input
+ * bytes or output space is available. If this is not a read request, the
+ * routine may also do some advantageous reading of messages from a multiplexed
+ * input source (which ensures that we don't jam up with everyone in their
+ * "need to write" code and nobody reading the accumulated data that would make
+ * writing possible).
*
- * 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) the buffers MUST have 4 bytes of overflow space at
- * the end that is not not counted in the "size". The iobuf.in buffer is not
- * (currently) circular. To facilitate the handling of MSG_DATA bytes as they
- * are read-from/written-into the buffers, see the three raw_* iobuf vars.
+ * The iobuf.in, .out and .msg buffers are all circular. Callers need to be
+ * aware that some data copies will need to be split when the bytes wrap around
+ * from the end to the start. In order to help make writing into the output
+ * buffers easier for some operations (such as the use of SIVAL() into the
+ * buffer) a buffer may be temporarily shortened by a small amount, but the
+ * original size will be automatically restored when the .pos wraps to the
+ * start. 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:
*
*
* - Make this routine able to read the generator-to-receiver batch flow?
*
- * - Make the input buffer circular?
- *
* 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
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&" : "");
+ /* We never resize the circular input buffer. */
+ if (iobuf.in.size < needed) {
+ rprintf(FERROR, "need to read %" SIZE_T_FMT_MOD "d bytes,"
+ " iobuf.in.buf is only %" SIZE_T_FMT_MOD "d bytes.\n",
+ (SIZE_T_FMT_CAST)needed, (SIZE_T_FMT_CAST)iobuf.in.size);
+ exit_cleanup(RERR_PROTOCOL);
}
- /* 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 %ld to %ld bytes.\n",
- who_am_i(), (long)iobuf.in.size, (long)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 %ld bytes from %ld to 0 in the input buffer (size=%ld, needed=%ld).\n",
- who_am_i(), (long)iobuf.in.len, (long)iobuf.in.pos, (long)iobuf.in.size, (long)needed);
- }
- if (iobuf.raw_input_ends_before)
- iobuf.raw_input_ends_before -= iobuf.in.pos;
- iobuf.in.pos = 0;
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%" SIZE_T_FMT_MOD "d, %sinput)\n",
+ who_am_i(), (SIZE_T_FMT_CAST)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));
+ fprintf(stderr, "need to write %" SIZE_T_FMT_MOD "d bytes,"
+ " iobuf.out.buf is only %" SIZE_T_FMT_MOD "d bytes.\n",
+ (SIZE_T_FMT_CAST)needed, (SIZE_T_FMT_CAST)(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,
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%" SIZE_T_FMT_MOD "d,"
+ " outroom) needs to flush %" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), (SIZE_T_FMT_CAST)needed,
iobuf.out.len + needed > iobuf.out.size
- ? (long)(iobuf.out.len + needed - iobuf.out.size) : 0L);
+ ? (SIZE_T_FMT_CAST)(iobuf.out.len + needed - iobuf.out.size) : (SIZE_T_FMT_CAST)0);
}
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);
+ fprintf(stderr, "need to write %" SIZE_T_FMT_MOD "d bytes,"
+ " iobuf.msg.buf is only %" SIZE_T_FMT_MOD "d bytes.\n",
+ (SIZE_T_FMT_CAST)needed, (SIZE_T_FMT_CAST)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,
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%" SIZE_T_FMT_MOD "d,"
+ " msgroom) needs to flush %" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), (SIZE_T_FMT_CAST)needed,
iobuf.msg.len + needed > iobuf.msg.size
- ? (long)(iobuf.msg.len + needed - iobuf.msg.size) : 0L);
+ ? (SIZE_T_FMT_CAST)(iobuf.msg.len + needed - iobuf.msg.size) : (SIZE_T_FMT_CAST)0);
}
break;
case 0:
- if (DEBUG_GTE(IO, 3))
- rprintf(FINFO, "[%s] perform_io(%ld, %d)\n", who_am_i(), (long)needed, flags);
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 3)) {
+ rprintf(FINFO, "[%s] perform_io(%" SIZE_T_FMT_MOD "d, %d)\n",
+ who_am_i(), (SIZE_T_FMT_CAST)needed, flags);
+ }
break;
default:
FD_ZERO(&r_fds);
FD_ZERO(&e_fds);
- if (iobuf.in_fd >= 0 && iobuf.in.size - (iobuf.in.pos + iobuf.in.len)) {
+ 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.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);
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 1)) {
+ rprintf(FINFO, "[%s] send_msg(%d, %" SIZE_T_FMT_MOD "d)\n",
+ who_am_i(), (int)MSG_DATA, (SIZE_T_FMT_CAST)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;
}
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:
break;
}
+ if (got_kill_signal > 0)
+ handle_kill_signal(True);
+
if (extra_flist_sending_enabled) {
- if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD
- && file_total - file_old_total >= MIN_FILECNT_LOOKAHEAD)
+ if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD && IN_MULTIPLEXED_AND_READY)
tv.tv_sec = 0;
else {
extra_flist_sending_enabled = False;
send_extra_file_list(sock_f_out, -1);
extra_flist_sending_enabled = !flist_eof;
} else
- check_timeout();
+ check_timeout((flags & PIO_NEED_INPUT) != 0, 0);
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;
+ size_t len, pos = iobuf.in.pos + iobuf.in.len;
+ ssize_t n;
+ if (pos >= iobuf.in.size) {
+ pos -= iobuf.in.size;
+ len = iobuf.in.size - iobuf.in.len;
+ } else
+ len = iobuf.in.size - pos;
if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
if (n == 0) {
+ /* Signal that input has become invalid. */
if (!read_batch || batch_fd < 0 || am_generator)
- whine_about_eof(iobuf.in_fd); /* Doesn't return. */
+ iobuf.in_fd = -2;
batch_fd = -1;
continue;
}
rsyserr(FERROR_SOCKET, errno, "read error");
} else
rsyserr(FERROR, errno, "read error");
- exit_cleanup(RERR_STREAMIO);
+ exit_cleanup(RERR_SOCKETIO);
}
}
- if (msgs2stderr && DEBUG_GTE(IO, 2))
- rprintf(FINFO, "[%s] recv=%ld\n", who_am_i(), (long)n);
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] recv=%" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), (SIZE_T_FMT_CAST)n);
+ }
- if (io_timeout)
+ if (io_timeout) {
last_io_in = time(NULL);
+ if (io_timeout && flags & PIO_NEED_INPUT)
+ maybe_send_keepalive(last_io_in, 0);
+ }
stats.total_read += n;
iobuf.in.len += n;
}
- if (iobuf.out_fd >= 0 && FD_ISSET(iobuf.out_fd, &w_fds)) {
+ if (stop_at_utime && time(NULL) >= stop_at_utime) {
+ rprintf(FERROR, "stopping at requested limit\n");
+ exit_cleanup(RERR_TIMEOUT);
+ }
+
+ if (out && 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;
+ ssize_t n;
if (bwlimit_writemax && len > bwlimit_writemax)
len = bwlimit_writemax;
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, "[%s] write error", who_am_i());
- exit_cleanup(RERR_STREAMIO);
+ iobuf.out_fd = -2;
+ iobuf.out.len = iobuf.msg.len = iobuf.raw_flushing_ends_before = 0;
+ rsyserr(FERROR_SOCKET, errno, "write error");
+ drain_multiplex_messages();
+ exit_cleanup(RERR_SOCKETIO);
}
}
- if (msgs2stderr && DEBUG_GTE(IO, 2)) {
- rprintf(FINFO, "[%s] %s sent=%ld\n",
- who_am_i(), out == &iobuf.out ? "out" : "msg", (long)n);
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] %s sent=%" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), out == &iobuf.out ? "out" : "msg", (SIZE_T_FMT_CAST)n);
}
if (io_timeout)
if (iobuf.raw_flushing_ends_before)
iobuf.raw_flushing_ends_before -= out->size;
out->pos = 0;
- }
- if (out->pos == iobuf.raw_flushing_ends_before)
+ restore_iobuf_size(out);
+ } else if (out->pos == iobuf.raw_flushing_ends_before)
iobuf.raw_flushing_ends_before = 0;
if ((out->len -= n) == empty_buf_len) {
out->pos = 0;
+ restore_iobuf_size(out);
if (empty_buf_len)
iobuf.raw_data_header_pos = 0;
}
}
+ if (got_kill_signal > 0)
+ handle_kill_signal(True);
+
+ /* We need to help prevent deadlock by doing what reading
+ * we can whenever we are here trying to write. */
+ if (IN_MULTIPLEXED_AND_READY && !(flags & PIO_NEED_INPUT)) {
+ while (!iobuf.raw_input_ends_before && iobuf.in.len > 512)
+ read_a_msg();
+ if (flist_receiving_enabled && iobuf.in.len > 512)
+ wait_for_receiver(); /* generator only */
+ }
+
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
}
double_break:
+ if (got_kill_signal > 0)
+ handle_kill_signal(True);
+
data = iobuf.in.buf + iobuf.in.pos;
if (flags & PIO_CONSUME_INPUT) {
iobuf.in.len -= needed;
iobuf.in.pos += needed;
+ if (iobuf.in.pos == iobuf.raw_input_ends_before)
+ iobuf.raw_input_ends_before = 0;
+ if (iobuf.in.pos >= iobuf.in.size) {
+ iobuf.in.pos -= iobuf.in.size;
+ if (iobuf.raw_input_ends_before)
+ iobuf.raw_input_ends_before -= iobuf.in.size;
+ }
}
return data;
}
+static void raw_read_buf(char *buf, size_t len)
+{
+ size_t pos = iobuf.in.pos;
+ char *data = perform_io(len, PIO_INPUT_AND_CONSUME);
+ if (iobuf.in.pos <= pos && len) {
+ size_t siz = len - iobuf.in.pos;
+ memcpy(buf, data, siz);
+ memcpy(buf + siz, iobuf.in.buf, iobuf.in.pos);
+ } else
+ memcpy(buf, data, len);
+}
+
+static int32 raw_read_int(void)
+{
+ char *data, buf[4];
+ if (iobuf.in.size - iobuf.in.pos >= 4)
+ data = perform_io(4, PIO_INPUT_AND_CONSUME);
+ else
+ raw_read_buf(data = buf, 4);
+ return IVAL(data, 0);
+}
+
void noop_io_until_death(void)
{
char buf[1024];
- kluge_around_eof = 1;
- set_io_timeout(protocol_version >= 31 ? 10 : 1);
+ if (!iobuf.in.buf || !iobuf.out.buf || iobuf.in_fd < 0 || iobuf.out_fd < 0 || kluge_around_eof)
+ return;
+
+ /* If we're talking to a daemon over a socket, don't short-circuit this logic */
+ if (msgs2stderr && daemon_connection >= 0)
+ return;
+
+ kluge_around_eof = 2;
+ /* Setting an I/O timeout ensures that if something inexplicably weird
+ * happens, we won't hang around forever. */
+ if (!io_timeout)
+ set_io_timeout(60);
while (1)
read_buf(iobuf.in_fd, buf, sizeof buf);
}
-/* Buffer a message for the multiplexed output stream. Is never used for MSG_DATA. */
+/* Buffer a message for the multiplexed output stream. Is not used for (normal) MSG_DATA. */
int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
{
char *hdr;
- size_t pos;
- BOOL want_debug = DEBUG_GTE(IO, 1) && convert >= 0 && (msgs2stderr || code != MSG_INFO);
+ size_t needed, pos;
+ BOOL want_debug = DEBUG_GTE(IO, 1) && convert >= 0 && (msgs2stderr == 1 || code != MSG_INFO);
if (!OUT_MULTIPLEXED)
return 0;
- if (want_debug)
- rprintf(FINFO, "[%s] send_msg(%d, %ld)\n", who_am_i(), (int)code, (long)len);
+ if (want_debug) {
+ rprintf(FINFO, "[%s] send_msg(%d, %" SIZE_T_FMT_MOD "d)\n",
+ who_am_i(), (int)code, (SIZE_T_FMT_CAST)len);
+ }
+ /* When checking for enough free space for this message, we need to
+ * make sure that there is space for the 4-byte header, plus we'll
+ * assume that we may waste up to 3 bytes (if the header doesn't fit
+ * at the physical end of the buffer). */
#ifdef ICONV_OPTION
if (convert > 0 && ic_send == (iconv_t)-1)
convert = 0;
if (convert > 0) {
- /* Ensuring double-size room leaves space for a potential conversion. */
- if (iobuf.msg.len + len*2 + 4 > iobuf.msg.size)
- perform_io(len*2 + 4, PIO_NEED_MSGROOM);
+ /* Ensuring double-size room leaves space for maximal conversion expansion. */
+ needed = len*2 + 4 + 3;
} else
#endif
- if (iobuf.msg.len + len + 4 > iobuf.msg.size)
- perform_io(len + 4, PIO_NEED_MSGROOM);
+ needed = len + 4 + 3;
+ if (iobuf.msg.len + needed > iobuf.msg.size) {
+ if (am_sender)
+ perform_io(needed, PIO_NEED_MSGROOM);
+ else { /* We sometimes allow the iobuf.msg size to increase to avoid a deadlock. */
+ size_t old_size = iobuf.msg.size;
+ restore_iobuf_size(&iobuf.msg);
+ realloc_xbuf(&iobuf.msg, iobuf.msg.size * 2);
+ if (iobuf.msg.pos + iobuf.msg.len > old_size)
+ memcpy(iobuf.msg.buf + old_size, iobuf.msg.buf, iobuf.msg.pos + iobuf.msg.len - old_size);
+ }
+ }
pos = iobuf.msg.pos + iobuf.msg.len; /* Must be set after any flushing. */
if (pos >= iobuf.msg.size)
pos -= iobuf.msg.size;
+ else if (pos + 4 > iobuf.msg.size) {
+ /* The 4-byte header won't fit at the end of the buffer,
+ * so we'll temporarily reduce the message buffer's size
+ * and put the header at the start of the buffer. */
+ reduce_iobuf_size(&iobuf.msg, pos);
+ pos = 0;
+ }
hdr = iobuf.msg.buf + pos;
- iobuf.msg.len += 4; /* Leave room for the coming header bytes. */
+ iobuf.msg.len += 4; /* Allocate room for the coming header bytes. */
#ifdef ICONV_OPTION
if (convert > 0) {
{
size_t siz;
- if ((pos += 4) >= iobuf.msg.size)
- pos -= iobuf.msg.size;
+ if ((pos += 4) == iobuf.msg.size)
+ pos = 0;
/* Handle a split copy if we wrap around the end of the circular buffer. */
if (pos >= iobuf.msg.pos && (siz = iobuf.msg.size - pos) < len) {
}
SIVAL(hdr, 0, ((MPLEX_BASE + (int)code)<<24) + len);
- /* If the header used any overflow bytes, move them to the start. */
- if ((pos = hdr+4 - iobuf.msg.buf) > iobuf.msg.size) {
- size_t siz = pos - iobuf.msg.size;
- if (DEBUG_GTE(IO, 4))
- rprintf(FINFO, "[%s] wrap-bytes moved: %d (send_msg)\n", who_am_i(), (int)siz);
- memcpy(iobuf.msg.buf, hdr+4 - siz, siz);
- }
- if (want_debug && convert > 0)
- rprintf(FINFO, "[%s] converted msg len=%ld\n", who_am_i(), (long)len);
+ if (want_debug && convert > 0) {
+ rprintf(FINFO, "[%s] converted msg len=%" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), (SIZE_T_FMT_CAST)len);
+ }
return 1;
}
send_msg(code, numbuf, 4, -1);
}
+void send_msg_success(const char *fname, int num)
+{
+ if (local_server) {
+ STRUCT_STAT st;
+
+ if (DEBUG_GTE(IO, 1))
+ rprintf(FINFO, "[%s] send_msg_success(%d)\n", who_am_i(), num);
+
+ if (stat(fname, &st) < 0)
+ memset(&st, 0, sizeof (STRUCT_STAT));
+ SIVAL(num_dev_ino_buf, 0, num);
+ SIVAL64(num_dev_ino_buf, 4, st.st_dev);
+ SIVAL64(num_dev_ino_buf, 4+8, st.st_ino);
+ send_msg(MSG_SUCCESS, num_dev_ino_buf, sizeof num_dev_ino_buf, -1);
+ } else
+ send_msg_int(MSG_SUCCESS, num);
+}
+
static void got_flist_entry_status(enum festatus status, int ndx)
{
struct file_list *flist = flist_for_ndx(ndx, "got_flist_entry_status");
switch (status) {
case FES_SUCCESS:
- if (remove_source_files)
- send_msg_int(MSG_SUCCESS, ndx);
+ if (remove_source_files) {
+ if (local_server)
+ send_msg(MSG_SUCCESS, num_dev_ino_buf, sizeof num_dev_ino_buf, -1);
+ else
+ send_msg_int(MSG_SUCCESS, ndx);
+ }
+ /* FALL THROUGH */
+ case FES_NO_SEND:
+#ifdef SUPPORT_HARD_LINKS
if (preserve_hard_links) {
struct file_struct *file = flist->files[ndx - flist->ndx_start];
if (F_IS_HLINKED(file)) {
+ if (status == FES_NO_SEND)
+ flist_ndx_push(&hlink_list, -2); /* indicates a failure follows */
flist_ndx_push(&hlink_list, ndx);
- flist->in_progress++;
+ if (inc_recurse)
+ flist->in_progress++;
}
}
+#endif
break;
case FES_REDO:
if (read_batch) {
flist->to_redo++;
flist_ndx_push(&redo_list, ndx);
break;
- case FES_NO_SEND:
- break;
}
}
void set_io_timeout(int secs)
{
io_timeout = secs;
+ allowed_lull = (io_timeout + 1) / 2;
- if (!io_timeout || io_timeout > SELECT_TIMEOUT)
+ if (!io_timeout || allowed_lull > SELECT_TIMEOUT)
select_timeout = SELECT_TIMEOUT;
else
- select_timeout = io_timeout;
+ select_timeout = allowed_lull;
- allowed_lull = read_batch ? 0 : (io_timeout + 1) / 2;
+ if (read_batch)
+ allowed_lull = 0;
}
static void check_for_d_option_error(const char *msg)
}
if (saw_d) {
- rprintf(FWARNING,
- "*** Try using \"--old-d\" if remote rsync is <= 2.6.3 ***\n");
+ rprintf(FWARNING, "*** Try using \"--old-d\" if remote rsync is <= 2.6.3 ***\n");
}
}
#ifdef ICONV_OPTION
if (flags & RL_CONVERT && iconv_buf.size < bufsiz)
- realloc_xbuf(&iconv_buf, bufsiz + 1024);
+ realloc_xbuf(&iconv_buf, ROUND_UP_1024(bufsiz) + 1024);
#endif
start:
char ***argv_p, int *argc_p, char **request_p)
{
int maxargs = MAX_ARGS;
- int dot_pos = 0;
- int argc = 0;
+ int dot_pos = 0, argc = 0, request_len = 0;
char **argv, *p;
int rl_flags = (rl_nulls ? RL_EOL_NULLS : 0);
rl_flags |= (protect_args && ic_recv != (iconv_t)-1 ? RL_CONVERT : 0);
#endif
- if (!(argv = new_array(char *, maxargs)))
- out_of_memory("read_args");
+ argv = new_array(char *, maxargs);
if (mod_name && !protect_args)
argv[argc++] = "rsyncd";
+ if (request_p)
+ *request_p = NULL;
+
while (1) {
if (read_line(f_in, buf, bufsiz, rl_flags) == 0)
break;
if (argc == maxargs-1) {
maxargs += MAX_ARGS;
- if (!(argv = realloc_array(argv, char *, maxargs)))
- out_of_memory("read_args");
+ argv = realloc_array(argv, char *, maxargs);
}
if (dot_pos) {
- if (request_p) {
- *request_p = strdup(buf);
- request_p = NULL;
+ if (request_p && request_len < 1024) {
+ int len = strlen(buf);
+ if (request_len)
+ request_p[0][request_len++] = ' ';
+ *request_p = realloc_array(*request_p, char, request_len + len + 1);
+ memcpy(*request_p + request_len, buf, len + 1);
+ request_len += len;
}
if (mod_name)
glob_expand_module(mod_name, buf, &argv, &argc, &maxargs);
else
glob_expand(buf, &argv, &argc, &maxargs);
} else {
- if (!(p = strdup(buf)))
- out_of_memory("read_args");
+ p = strdup(buf);
argv[argc++] = p;
if (*p == '.' && p[1] == '\0')
dot_pos = argc;
BOOL io_start_buffering_out(int f_out)
{
- if (msgs2stderr && DEBUG_GTE(IO, 2))
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_buffering_out(%d)\n", who_am_i(), f_out);
- if (OUT_MULTIPLEXED && !iobuf.msg.buf) {
- iobuf.msg.size = IO_BUFFER_SIZE - 4;
- if (!(iobuf.msg.buf = new_array(char, iobuf.msg.size + 4)))
- out_of_memory("io_start_buffering_out");
- iobuf.msg.pos = iobuf.msg.len = 0;
- }
-
if (iobuf.out.buf) {
if (iobuf.out_fd == -1)
iobuf.out_fd = f_out;
- assert(f_out == iobuf.out_fd);
+ else
+ assert(f_out == iobuf.out_fd);
return False;
}
- iobuf.out.size = IO_BUFFER_SIZE * 2 - 4;
- /* The 4 overflow bytes makes some circular-buffer wrapping operations easier. */
- if (!(iobuf.out.buf = new_array(char, iobuf.out.size + 4)))
- out_of_memory("io_start_buffering_out");
- iobuf.out.pos = iobuf.out.len = 0;
+ alloc_xbuf(&iobuf.out, ROUND_UP_1024(IO_BUFFER_SIZE * 2));
iobuf.out_fd = f_out;
return True;
BOOL io_start_buffering_in(int f_in)
{
- if (msgs2stderr && DEBUG_GTE(IO, 2))
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_buffering_in(%d)\n", who_am_i(), f_in);
if (iobuf.in.buf) {
if (iobuf.in_fd == -1)
iobuf.in_fd = f_in;
- assert(f_in == iobuf.in_fd);
+ else
+ assert(f_in == iobuf.in_fd);
return False;
}
- iobuf.in.size = IO_BUFFER_SIZE;
- if (!(iobuf.in.buf = new_array(char, iobuf.in.size)))
- out_of_memory("io_start_buffering_in");
-
- iobuf.in.pos = iobuf.in.len = 0;
-
+ alloc_xbuf(&iobuf.in, ROUND_UP_1024(IO_BUFFER_SIZE));
iobuf.in_fd = f_in;
return True;
void io_end_buffering_in(BOOL free_buffers)
{
- if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2)) {
rprintf(FINFO, "[%s] io_end_buffering_in(IOBUF_%s_BUFS)\n",
who_am_i(), free_buffers ? "FREE" : "KEEP");
}
void io_end_buffering_out(BOOL free_buffers)
{
- if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2)) {
rprintf(FINFO, "[%s] io_end_buffering_out(IOBUF_%s_BUFS)\n",
who_am_i(), free_buffers ? "FREE" : "KEEP");
}
io_flush(NORMAL_FLUSH);
}
-void maybe_send_keepalive(void)
+/* 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 (time(NULL) - last_io_out >= allowed_lull) {
- if (!iobuf.msg.len && iobuf.out.len == iobuf.out_empty_len) {
- if (protocol_version < 29)
- return; /* there's nothing we can do */
- if (protocol_version >= 30)
- send_msg(MSG_NOOP, "", 0, 0);
- else {
- write_int(iobuf.out_fd, cur_flist->used);
- write_shortint(iobuf.out_fd, ITEM_IS_NEW);
- }
- }
- if (iobuf.msg.len)
+ if (flags & MSK_ACTIVE_RECEIVER)
+ last_io_in = now; /* Fudge things when we're working hard on the files. */
+
+ /* Early in the transfer (before the receiver forks) the receiving side doesn't
+ * care if it hasn't sent data in a while as long as it is receiving data (in
+ * fact, a pre-3.1.0 rsync would die if we tried to send it a keep alive during
+ * this time). So, if we're an early-receiving proc, just return and let the
+ * incoming data determine if we timeout. */
+ if (!am_sender && !am_receiver && !am_generator)
+ return;
+
+ 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);
/* Read a message from a multiplexed source. */
static void read_a_msg(void)
{
- char *data, line[BIGPATHBUFLEN];
+ char data[BIGPATHBUFLEN];
int tag, val;
size_t msg_bytes;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- tag = IVAL(data, 0);
+ /* 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 || tag != MSG_INFO))
- rprintf(FINFO, "[%s] got msg=%d, len=%ld\n", who_am_i(), (int)tag, (long)msg_bytes);
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 1)) {
+ rprintf(FINFO, "[%s] got msg=%d, len=%" SIZE_T_FMT_MOD "d\n",
+ who_am_i(), (int)tag, (SIZE_T_FMT_CAST)msg_bytes);
+ }
switch (tag) {
case MSG_DATA:
* 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 decrease and eventually become real. */
- iobuf.raw_input_ends_before = iobuf.in.pos + msg_bytes;
+ * 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;
- data = perform_io(sizeof stats.total_read, PIO_INPUT_AND_CONSUME);
- memcpy((char*)&stats.total_read, data, sizeof stats.total_read);
+ 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;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- got_flist_entry_status(FES_REDO, IVAL(data, 0));
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
+ got_flist_entry_status(FES_REDO, val);
break;
case MSG_IO_ERROR:
- if (msg_bytes != 4 || am_sender)
+ if (msg_bytes != 4)
goto invalid_msg;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- io_error |= IVAL(data, 0);
- if (!am_generator)
- send_msg(MSG_IO_ERROR, data, 4, 0);
+ 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;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- val = IVAL(data, 0);
+ 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);
}
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();
+ maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
break;
case MSG_DELETED:
- if (msg_bytes >= sizeof line)
+ if (msg_bytes >= sizeof data)
goto overflow;
if (am_generator) {
- memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), msg_bytes);
- send_msg(MSG_DELETED, line, msg_bytes, 1);
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
+ send_msg(MSG_DELETED, data, msg_bytes, 1);
break;
}
#ifdef ICONV_OPTION
int add_null = 0;
int flags = ICB_INCLUDE_BAD | ICB_INIT;
- INIT_CONST_XBUF(outbuf, line);
+ 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;
- memcpy(ibuf + inbuf.len, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+ raw_read_buf(ibuf + inbuf.len, len);
inbuf.pos = 0;
inbuf.len += len;
if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])
msg_bytes = outbuf.len;
} else
#endif
- memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), msg_bytes);
+ raw_read_buf(data, msg_bytes);
+ iobuf.in_multiplexed = 1;
/* A directory name was sent with the trailing null */
- if (msg_bytes > 0 && !line[msg_bytes-1])
- log_delete(line, S_IFDIR);
+ if (msg_bytes > 0 && !data[msg_bytes-1])
+ log_delete(data, S_IFDIR);
else {
- line[msg_bytes] = '\0';
- log_delete(line, S_IFREG);
+ data[msg_bytes] = '\0';
+ log_delete(data, S_IFREG);
}
break;
case MSG_SUCCESS:
- if (msg_bytes != 4) {
+ if (msg_bytes != (local_server ? 4+8+8 : 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);
}
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
+ raw_read_buf(num_dev_ino_buf, msg_bytes);
+ val = IVAL(num_dev_ino_buf, 0);
+ iobuf.in_multiplexed = 1;
if (am_generator)
- got_flist_entry_status(FES_SUCCESS, IVAL(data, 0));
+ got_flist_entry_status(FES_SUCCESS, val);
else
- successful_send(IVAL(data, 0));
+ successful_send(val);
break;
case MSG_NO_SEND:
if (msg_bytes != 4)
goto invalid_msg;
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
+ val = raw_read_int();
+ iobuf.in_multiplexed = 1;
if (am_generator)
- got_flist_entry_status(FES_NO_SEND, IVAL(data, 0));
+ got_flist_entry_status(FES_NO_SEND, val);
else
- send_msg(MSG_NO_SEND, data, 4, 0);
+ send_msg_int(MSG_NO_SEND, val);
break;
case MSG_ERROR_SOCKET:
case MSG_ERROR_UTF8:
case MSG_ERROR:
case MSG_ERROR_XFER:
case MSG_WARNING:
- if (msg_bytes >= sizeof line) {
+ if (msg_bytes >= sizeof data) {
overflow:
rprintf(FERROR,
"multiplexing overflow %d:%lu [%s%s]\n",
inc_recurse ? "/inc" : "");
exit_cleanup(RERR_STREAMIO);
}
- memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), msg_bytes);
- rwrite((enum logcode)tag, line, msg_bytes, !am_generator);
+ raw_read_buf(data, msg_bytes);
+ /* We don't set in_multiplexed value back to 1 before writing this message
+ * because the write might loop back and read yet another message, over and
+ * over again, while waiting for room to put the message in the msg buffer. */
+ rwrite((enum logcode)tag, data, msg_bytes, !am_generator);
+ iobuf.in_multiplexed = 1;
if (first_message) {
- if (list_only && !am_sender && tag == 1 && msg_bytes < sizeof line) {
- line[msg_bytes] = '\0';
- check_for_d_option_error(line);
+ 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 %" SIZE_T_FMT_MOD "d bytes\n",
+ who_am_i(), (SIZE_T_FMT_CAST)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);
}
- val = 0;
- } else if (msg_bytes == 4) {
- data = perform_io(4, PIO_INPUT_AND_CONSUME);
- val = IVAL(data, 0);
- if (protocol_version >= 31) {
- if (am_generator)
- send_msg_int(MSG_ERROR_EXIT, val);
- else
- send_msg(MSG_ERROR_EXIT, "", 0, 0);
+ } else if (protocol_version >= 31) {
+ if (am_generator || am_receiver) {
+ 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);
}
- } else
- goto invalid_msg;
+ }
/* Send a negative linenum so that we don't end up
* with a duplicate exit message. */
_exit_cleanup(val, __FILE__, 0 - __LINE__);
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)
}
} 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 = recv_file_list(iobuf.in_fd, ndx);
flist->parent_ndx = ndx;
#ifdef SUPPORT_HARD_LINKS
if (preserve_hard_links)
match_hard_links(flist);
#endif
+ flist_receiving_enabled = True;
}
}
}
#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);
+ u.x = IVAL64(u.b,0);
#endif
return u.x;
}
#endif
}
+/* Debugging note: this will be named read_buf_() when using an external zlib. */
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(f); /* Doesn't return. */
+ whine_about_eof(False); /* Doesn't return. */
goto batch_copy;
}
if (!IN_MULTIPLEXED) {
- memcpy(buf, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+ raw_read_buf(buf, len);
total_data_read += len;
if (forward_flist_data)
write_buf(iobuf.out_fd, buf, len);
}
while (1) {
- char *data;
size_t siz;
while (!iobuf.raw_input_ends_before)
read_a_msg();
siz = MIN(len, iobuf.raw_input_ends_before - iobuf.in.pos);
- data = perform_io(siz, PIO_INPUT_AND_CONSUME);
- if (iobuf.in.pos == iobuf.raw_input_ends_before)
- iobuf.raw_input_ends_before = 0;
-
- /* The bytes at the "data" pointer will survive long
- * enough to make a copy, but not past future I/O. */
- memcpy(buf, data, siz);
+ if (siz >= iobuf.in.size)
+ siz = iobuf.in.size;
+ raw_read_buf(buf, siz);
total_data_read += siz;
if (forward_flist_data)
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);
+ total_written -= (int64)elapsed_usec * bwlimit / (ONE_SEC/1024);
if (total_written < 0)
total_written = 0;
}
total_written = (sleep_usec - elapsed_usec) * bwlimit / (ONE_SEC/1024);
}
-void io_flush(int flush_it_all)
+void io_flush(int flush_type)
{
if (iobuf.out.len > iobuf.out_empty_len) {
- if (flush_it_all) /* FULL_FLUSH: flush everything in the output buffers */
+ if (flush_type == 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 */
+ else if (flush_type == NORMAL_FLUSH) /* flush at least 1 byte */
perform_io(iobuf.out.size - iobuf.out.len + 1, PIO_NEED_OUTROOM);
+ /* MSG_FLUSH: flush iobuf.msg only */
}
if (iobuf.msg.len)
perform_io(iobuf.msg.size, PIO_NEED_MSGROOM);
{
char b[5];
uchar bit;
- int cnt = 4;
+ int cnt;
SIVAL(b, 1, x);
- while (cnt > 1 && b[cnt] == 0)
- cnt--;
+ for (cnt = 4; 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];
+ *b = b[1];
write_buf(f, b, cnt);
}
uchar bit;
int cnt = 8;
- SIVAL(b, 1, x);
#if SIZEOF_INT64 >= 8
- SIVAL(b, 5, x >> 32);
+ SIVAL64(b, 1, x);
#else
+ SIVAL(b, 1, x);
if (x <= 0x7FFFFFFF && x >= 0)
memset(b + 5, 0, 4);
else {
#endif
}
+void write_bigbuf(int f, const char *buf, size_t len)
+{
+ size_t half_max = (iobuf.out.size - iobuf.out_empty_len) / 2;
+
+ while (len > half_max + 1024) {
+ write_buf(f, buf, half_max);
+ buf += half_max;
+ len -= half_max;
+ }
+
+ write_buf(f, buf, len);
+}
+
void write_buf(int f, const char *buf, size_t len)
{
size_t pos, siz;
/* Read a line of up to bufsiz-1 characters into buf. Strips
* the (required) trailing newline and all carriage returns.
* Returns 1 for success; 0 for I/O error or truncation. */
-int read_line_old(int fd, char *buf, size_t bufsiz)
+int read_line_old(int fd, char *buf, size_t bufsiz, int eof_ok)
{
+ assert(fd != iobuf.in_fd);
bufsiz--; /* leave room for the null */
while (bufsiz > 0) {
- assert(fd != iobuf.in_fd);
- if (safe_read(fd, buf, 1) == 0)
+ if (safe_read(fd, buf, 1) == 0) {
+ if (eof_ok)
+ break;
return 0;
+ }
if (*buf == '\0')
return 0;
if (*buf == '\n')
va_end(ap);
if (len < 0)
- exit_cleanup(RERR_STREAMIO);
+ exit_cleanup(RERR_PROTOCOL);
- if (len > (int)sizeof buf) {
+ if (len >= (int)sizeof buf) {
rprintf(FERROR, "io_printf() was too long for the buffer.\n");
- exit_cleanup(RERR_STREAMIO);
+ exit_cleanup(RERR_PROTOCOL);
}
write_sbuf(fd, buf);
{
io_flush(FULL_FLUSH);
- if (msgs2stderr && DEBUG_GTE(IO, 2))
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_multiplex_out(%d)\n", who_am_i(), fd);
+ if (!iobuf.msg.buf)
+ alloc_xbuf(&iobuf.msg, ROUND_UP_1024(IO_BUFFER_SIZE));
+
iobuf.out_empty_len = 4; /* See also OUT_MULTIPLEXED */
io_start_buffering_out(fd);
+ got_kill_signal = 0;
iobuf.raw_data_header_pos = iobuf.out.pos + iobuf.out.len;
iobuf.out.len += 4;
/* Setup for multiplexing a MSG_* stream with the data stream. */
void io_start_multiplex_in(int fd)
{
- if (msgs2stderr && DEBUG_GTE(IO, 2))
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_multiplex_in(%d)\n", who_am_i(), fd);
- iobuf.in_multiplexed = True; /* See also IN_MULTIPLEXED */
+ iobuf.in_multiplexed = 1; /* See also IN_MULTIPLEXED */
io_start_buffering_in(fd);
}
{
int ret = iobuf.in_multiplexed ? iobuf.in_fd : -1;
- if (msgs2stderr && DEBUG_GTE(IO, 2))
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_end_multiplex_in(mode=%d)\n", who_am_i(), mode);
- iobuf.in_multiplexed = False;
+ iobuf.in_multiplexed = 0;
if (mode == MPLX_SWITCHING)
iobuf.raw_input_ends_before = 0;
else
{
int ret = iobuf.out_empty_len ? iobuf.out_fd : -1;
- if (msgs2stderr && DEBUG_GTE(IO, 2))
+ if (msgs2stderr == 1 && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_end_multiplex_out(mode=%d)\n", who_am_i(), mode);
if (mode != MPLX_TO_BUFFERED)
iobuf.out.len = 0;
iobuf.out_empty_len = 0;
+ if (got_kill_signal > 0) /* Just in case... */
+ handle_kill_signal(False);
+ got_kill_signal = -1;
return ret;
}
* is involved. */
write_int(batch_fd, protocol_version);
if (protocol_version >= 30)
- write_byte(batch_fd, inc_recurse);
+ write_varint(batch_fd, compat_flags);
write_int(batch_fd, checksum_seed);
if (am_sender)