extern int bwlimit;
extern size_t bwlimit_writemax;
-extern int verbose;
extern int io_timeout;
extern int allowed_lull;
extern int am_server;
static int io_filesfrom_buflen;
static size_t contiguous_write_len = 0;
static int select_timeout = SELECT_TIMEOUT;
+static int active_filecnt = 0;
+static OFF_T active_bytecnt = 0;
static void read_loop(int fd, char *buf, size_t len);
static struct flist_ndx_list redo_list, hlink_list;
-struct msg_list {
- struct msg_list *next;
+struct msg_list_item {
+ struct msg_list_item *next;
char *buf;
int len;
};
-static struct msg_list *msg_list_head;
-static struct msg_list *msg_list_tail;
+struct msg_list {
+ struct msg_list_item *head, *tail;
+};
+
+static struct msg_list msg_list;
static void flist_ndx_push(struct flist_ndx_list *lp, int ndx)
{
/* Add a message to the pending MSG_* list. */
static void msg_list_add(int code, char *buf, int len)
{
- struct msg_list *ml;
+ struct msg_list_item *ml;
- if (!(ml = new(struct msg_list)))
+ if (!(ml = new(struct msg_list_item)))
out_of_memory("msg_list_add");
ml->next = NULL;
if (!(ml->buf = new_array(char, len+4)))
SIVAL(ml->buf, 0, ((code+MPLEX_BASE)<<24) | len);
memcpy(ml->buf+4, buf, len);
ml->len = len+4;
- if (msg_list_tail)
- msg_list_tail->next = ml;
+ if (msg_list.tail)
+ msg_list.tail->next = ml;
else
- msg_list_head = ml;
- msg_list_tail = ml;
-}
-
-void send_msg(enum msgcode code, char *buf, int len)
-{
- if (msg_fd_out < 0) {
- io_multiplex_write(code, buf, len);
- return;
- }
- msg_list_add(code, buf, len);
- msg_list_push(NORMAL_FLUSH);
+ msg_list.head = ml;
+ msg_list.tail = ml;
}
/* Read a message from the MSG_* fd and handle it. This is called either
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, 4);
+ if (remove_sent_files)
+ decrement_active_files(IVAL(buf,0));
flist_ndx_push(&redo_list, IVAL(buf,0));
break;
case MSG_DELETED:
exit_cleanup(RERR_STREAMIO);
}
read_loop(fd, buf, len);
- if (remove_sent_files)
+ if (remove_sent_files) {
+ decrement_active_files(IVAL(buf,0));
io_multiplex_write(MSG_SUCCESS, buf, len);
+ }
if (preserve_hard_links)
flist_ndx_push(&hlink_list, IVAL(buf,0));
break;
+ case MSG_SOCKERR:
+ if (!am_generator) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ close_multiplexing_out();
+ /* FALL THROUGH */
case MSG_INFO:
case MSG_ERROR:
case MSG_LOG:
msg_fd_in = fd;
}
+/* This is used by the generator to limit how many file transfers can
+ * be active at once when --remove-sent-files is specified. Without
+ * this, sender-side deletions were mostly happening at the end. */
+void increment_active_files(int ndx, int itemizing, enum logcode code)
+{
+ /* TODO: tune these limits? */
+ while (active_filecnt >= (active_bytecnt >= 128*1024 ? 10 : 50)) {
+ if (hlink_list.head)
+ check_for_finished_hlinks(itemizing, code);
+ read_msg_fd();
+ }
+
+ active_filecnt++;
+ active_bytecnt += the_file_list->files[ndx]->length;
+}
+
+void decrement_active_files(int ndx)
+{
+ active_filecnt--;
+ active_bytecnt -= the_file_list->files[ndx]->length;
+}
+
/* Try to push messages off the list onto the wire. If we leave with more
* to do, return 0. On error, return -1. If everything flushed, return 1.
* This is only active in the receiver. */
-int msg_list_push(int flush_it_all)
+static int msg_list_flush(int flush_it_all)
{
static int written = 0;
struct timeval tv;
if (msg_fd_out < 0)
return -1;
- while (msg_list_head) {
- struct msg_list *ml = msg_list_head;
+ while (msg_list.head) {
+ struct msg_list_item *ml = msg_list.head;
int n = write(msg_fd_out, ml->buf + written, ml->len - written);
if (n < 0) {
if (errno == EINTR)
check_timeout();
} else if ((written += n) == ml->len) {
free(ml->buf);
- msg_list_head = ml->next;
- if (!msg_list_head)
- msg_list_tail = NULL;
+ msg_list.head = ml->next;
+ if (!msg_list.head)
+ msg_list.tail = NULL;
free(ml);
written = 0;
}
return 1;
}
+void send_msg(enum msgcode code, char *buf, int len)
+{
+ if (msg_fd_out < 0) {
+ io_multiplex_write(code, buf, len);
+ return;
+ }
+ msg_list_add(code, buf, len);
+ msg_list_flush(NORMAL_FLUSH);
+}
+
int get_redo_num(int itemizing, enum logcode code)
{
while (1) {
exit_cleanup(RERR_STREAMIO);
}
-
/**
* Read from a socket with I/O timeout. return the number of bytes
* read. If no bytes can be read then exit, never return a number <= 0.
*/
static int read_timeout(int fd, char *buf, size_t len)
{
- int n, ret = 0;
+ int n, cnt = 0;
io_flush(NORMAL_FLUSH);
- while (ret == 0) {
+ while (cnt == 0) {
/* until we manage to read *something* */
fd_set r_fds, w_fds;
struct timeval tv;
FD_ZERO(&r_fds);
FD_ZERO(&w_fds);
FD_SET(fd, &r_fds);
- if (msg_list_head) {
+ if (msg_list.head) {
FD_SET(msg_fd_out, &w_fds);
if (msg_fd_out > maxfd)
maxfd = msg_fd_out;
continue;
}
- if (msg_list_head && FD_ISSET(msg_fd_out, &w_fds))
- msg_list_push(NORMAL_FLUSH);
+ if (msg_list.head && FD_ISSET(msg_fd_out, &w_fds))
+ msg_list_flush(NORMAL_FLUSH);
if (io_filesfrom_f_out >= 0) {
if (io_filesfrom_buflen) {
continue;
/* Don't write errors on a dead socket. */
- if (fd == sock_f_in)
+ if (fd == sock_f_in) {
close_multiplexing_out();
- rsyserr(FERROR, errno, "read error");
+ rsyserr(FSOCKERR, errno, "read error");
+ } else
+ rsyserr(FERROR, errno, "read error");
exit_cleanup(RERR_STREAMIO);
}
buf += n;
len -= n;
- ret += n;
+ cnt += n;
if (fd == sock_f_in && io_timeout)
last_io_in = time(NULL);
}
- return ret;
+ return cnt;
}
/**
return s - fname;
}
-
static char *iobuf_out;
static int iobuf_out_cnt;
iobuf_out_cnt = 0;
}
-
static char *iobuf_in;
static size_t iobuf_in_siz;
out_of_memory("io_start_buffering_in");
}
-
void io_end_buffering(void)
{
io_flush(NORMAL_FLUSH);
}
}
-
void maybe_flush_socket(void)
{
if (iobuf_out && iobuf_out_cnt && time(NULL) - last_io_out >= 5)
io_flush(NORMAL_FLUSH);
}
-
void maybe_send_keepalive(void)
{
if (time(NULL) - last_io_out >= allowed_lull) {
}
}
-
/**
* Continue trying to read len bytes - don't return until len has been
* read.
}
}
-
/**
* Read from the file descriptor handling multiplexing - return number
* of bytes read.
static size_t remaining;
static size_t iobuf_in_ndx;
size_t msg_bytes;
- int tag, ret = 0;
+ int tag, cnt = 0;
char line[BIGPATHBUFLEN];
if (!iobuf_in || fd != sock_f_in)
iobuf_in_ndx = 0;
}
- while (ret == 0) {
+ while (cnt == 0) {
if (remaining) {
len = MIN(len, remaining);
memcpy(buf, iobuf_in + iobuf_in_ndx, len);
iobuf_in_ndx += len;
remaining -= len;
- ret = len;
+ cnt = len;
break;
}
if (msg_bytes >= sizeof line)
goto overflow;
read_loop(fd, line, msg_bytes);
- line[msg_bytes] = '\0';
/* A directory name was sent with the trailing null */
if (msg_bytes > 0 && !line[msg_bytes-1])
log_delete(line, S_IFDIR);
- else
+ else {
+ line[msg_bytes] = '\0';
log_delete(line, S_IFREG);
+ }
break;
case MSG_SUCCESS:
if (msg_bytes != 4) {
if (remaining == 0)
io_flush(NORMAL_FLUSH);
- return ret;
+ return cnt;
}
-
-
/**
* Do a buffered read from @p fd. Don't return until all @p n bytes
* have been read. If all @p n can't be read then exit with an
**/
static void readfd(int fd, char *buffer, size_t N)
{
- int ret;
+ int cnt;
size_t total = 0;
while (total < N) {
- ret = readfd_unbuffered(fd, buffer + total, N-total);
- total += ret;
+ cnt = readfd_unbuffered(fd, buffer + total, N-total);
+ total += cnt;
}
if (fd == write_batch_monitor_in) {
stats.total_read += total;
}
-
int read_shortint(int f)
{
uchar b[2];
return (b[1] << 8) + b[0];
}
-
int32 read_int(int f)
{
char b[4];
- int32 ret;
+ int32 num;
readfd(f,b,4);
- ret = IVAL(b,0);
- if (ret == (int32)0xffffffff)
+ num = IVAL(b,0);
+ if (num == (int32)0xffffffff)
return -1;
- return ret;
+ return num;
}
int64 read_longint(int f)
{
- int64 ret;
+ int64 num;
char b[8];
- ret = read_int(f);
+ num = read_int(f);
- if ((int32)ret != (int32)0xffffffff)
- return ret;
+ if ((int32)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);
- ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
+ num = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
#endif
- return ret;
+ return num;
}
void read_buf(int f,char *buf,size_t len)
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",
write_int(f, sum->remainder);
}
-
/**
* Sleep after writing to limit I/O bandwidth usage.
*
#define ONE_SEC 1000000L /* # of microseconds in a second */
- if (!bwlimit)
+ if (!bwlimit_writemax)
return;
total_written += bytes_written;
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.
{
size_t n, total = 0;
fd_set w_fds, r_fds;
- int maxfd, count, ret, using_r_fds;
+ int maxfd, count, cnt, using_r_fds;
struct timeval tv;
no_flush++;
continue;
n = len - total;
- if (bwlimit && n > bwlimit_writemax)
+ if (bwlimit_writemax && n > bwlimit_writemax)
n = bwlimit_writemax;
- ret = write(fd, buf + total, n);
+ cnt = write(fd, buf + total, n);
- if (ret <= 0) {
- if (ret < 0) {
+ if (cnt <= 0) {
+ if (cnt < 0) {
if (errno == EINTR)
continue;
if (errno == EWOULDBLOCK || errno == EAGAIN) {
exit_cleanup(RERR_STREAMIO);
}
- total += ret;
+ total += cnt;
if (fd == sock_f_out) {
if (io_timeout || am_generator)
last_io_out = time(NULL);
- sleep_for_bwlimit(ret);
+ sleep_for_bwlimit(cnt);
}
}
no_flush--;
}
-
/**
* Write an message to a multiplexed stream. If this fails then rsync
* exits.
**/
static void mplex_write(enum msgcode code, char *buf, size_t len)
{
- char buffer[BIGPATHBUFLEN];
+ char buffer[1024];
size_t n = len;
SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
contiguous_write_len = len + 4;
if (n > sizeof buffer - 4)
- n = sizeof buffer - 4;
+ n = 0;
+ else
+ memcpy(buffer + 4, buf, n);
- memcpy(&buffer[4], buf, n);
writefd_unbuffered(sock_f_out, buffer, n+4);
len -= n;
contiguous_write_len = 0;
}
-
void io_flush(int flush_it_all)
{
- msg_list_push(flush_it_all);
+ msg_list_flush(flush_it_all);
if (!iobuf_out_cnt || no_flush)
return;
iobuf_out_cnt = 0;
}
-
static void writefd(int fd,char *buf,size_t len)
{
if (fd == msg_fd_out) {
}
}
-
void write_shortint(int f, int x)
{
uchar b[2];
writefd(f, (char *)b, 2);
}
-
void write_int(int f,int32 x)
{
char b[4];
writefd(f,b,4);
}
-
void write_int_named(int f, int32 x, const char *phase)
{
io_write_phase = phase;
io_write_phase = phase_unknown;
}
-
/*
* Note: int64 may actually be a 32-bit type if ./configure couldn't find any
* 64-bit types on this platform.
writefd(f, str, len);
}
-
/**
* Read a line of up to @p maxlen characters into @p buf (not counting
* the trailing null). Strips the (required) trailing newline and all
return maxlen > 0;
}
-
void io_printf(int fd, const char *format, ...)
{
va_list ap;
write_sbuf(fd, buf);
}
-
/** Setup for multiplexing a MSG_* stream with the data stream. */
void io_start_multiplex_out(void)
{