* Copyright (C) 1996-2001 Andrew Tridgell
* Copyright (C) 1996 Paul Mackerras
* Copyright (C) 2001, 2002 Martin Pool <mbp@samba.org>
- * Copyright (C) 2003-2013 Wayne Davison
+ * Copyright (C) 2003-2020 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 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 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 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;
static void drain_multiplex_messages(void);
static void sleep_for_bwlimit(int bytes_written);
-static void check_timeout(BOOL allow_keepalive)
+static void check_timeout(BOOL allow_keepalive, int keepalive_flags)
{
time_t t, chk;
if (allow_keepalive) {
/* This may put data into iobuf.msg w/o flushing. */
- maybe_send_keepalive(t, 0);
+ maybe_send_keepalive(t, keepalive_flags);
}
if (!last_io_in)
* 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);
}
- if (io_timeout)
- maybe_send_keepalive(time(NULL), MSK_ALLOW_FLUSH);
+ 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);
+ int 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);
if (n == 0)
if (n < 0) {
if (errno == EINTR)
continue;
- goto read_failed;
+ rsyserr(FERROR, errno, "safe_read failed to read %ld bytes", (long)len);
+ exit_cleanup(RERR_STREAMIO);
}
if ((got += (size_t)n) == len)
break;
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 %ld bytes to %s",
+ (long)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);
}
if (io_timeout)
}
}
+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
break;
}
+ if (got_kill_signal > 0)
+ handle_kill_signal(True);
+
if (extra_flist_sending_enabled) {
if (file_total - file_old_total < MAX_FILECNT_LOOKAHEAD && IN_MULTIPLEXED_AND_READY)
tv.tv_sec = 0;
send_extra_file_list(sock_f_out, -1);
extra_flist_sending_enabled = !flist_eof;
} else
- check_timeout((flags & PIO_NEED_INPUT) != 0);
+ check_timeout((flags & PIO_NEED_INPUT) != 0, 0);
FD_ZERO(&r_fds); /* Just in case... */
FD_ZERO(&w_fds);
}
if (msgs2stderr && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] recv=%ld\n", who_am_i(), (long)n);
- if (io_timeout) {
+ if (io_timeout || stop_at_utime) {
last_io_in = time(NULL);
- if (flags & PIO_NEED_INPUT)
+ if (stop_at_utime && last_io_in >= stop_at_utime) {
+ rprintf(FERROR, "stopping at requested limit\n");
+ exit_cleanup(RERR_TIMEOUT);
+ }
+ if (io_timeout && flags & PIO_NEED_INPUT)
maybe_send_keepalive(last_io_in, 0);
}
stats.total_read += n;
msgs2stderr = 1;
iobuf.out_fd = -2;
iobuf.out.len = iobuf.msg.len = iobuf.raw_flushing_ends_before = 0;
- rsyserr(FERROR_SOCKET, errno, "[%s] write error", who_am_i());
+ rsyserr(FERROR_SOCKET, errno, "write error");
drain_multiplex_messages();
exit_cleanup(RERR_SOCKETIO);
}
}
}
+ 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)) {
}
double_break:
+ if (got_kill_signal > 0)
+ handle_kill_signal(True);
+
data = iobuf.in.buf + iobuf.in.pos;
if (flags & PIO_CONSUME_INPUT) {
{
char buf[1024];
- if (!iobuf.in.buf || !iobuf.out.buf || iobuf.in_fd < 0 || iobuf.out_fd < 0 || kluge_around_eof)
+ if (!iobuf.in.buf || !iobuf.out.buf || iobuf.in_fd < 0 || iobuf.out_fd < 0 || kluge_around_eof || msgs2stderr)
return;
kluge_around_eof = 2;
} else
#endif
needed = len + 4 + 3;
- if (iobuf.msg.len + needed > iobuf.msg.size)
- perform_io(needed, PIO_NEED_MSGROOM);
+ if (iobuf.msg.len + needed > iobuf.msg.size) {
+ if (!am_receiver)
+ perform_io(needed, PIO_NEED_MSGROOM);
+ else { /* We allow the receiver to increase their iobuf.msg size 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)
}
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:
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 (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) {
int len = strlen(buf);
if (request_len)
request_p[0][request_len++] = ' ';
- if (!(*request_p = realloc_array(*request_p, char, request_len + len + 1)))
- out_of_memory("read_args");
+ *request_p = realloc_array(*request_p, char, request_len + len + 1);
memcpy(*request_p + request_len, buf, len + 1);
request_len += len;
}
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;
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. */
io_flush(FULL_FLUSH);
}
} else if (protocol_version >= 31) {
- if (am_generator) {
+ 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);
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)
#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;
}
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;
if (len < 0)
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_PROTOCOL);
}
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;
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, compat_flags);
+ write_varint(batch_fd, compat_flags);
write_int(batch_fd, checksum_seed);
if (am_sender)
git.samba.org / rsync.git / blobdiff