* Copyright (C) 1996-2001 Andrew Tridgell
* Copyright (C) 1996 Paul Mackerras
* Copyright (C) 2001, 2002 Martin Pool <mbp@samba.org>
- * Copyright (C) 2003, 2004, 2005, 2006 Wayne Davison
+ * Copyright (C) 2003-2007 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
- * the Free Software Foundation; either version 2 of the License, or
- * (at your option) any later version.
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
extern int am_daemon;
extern int am_sender;
extern int am_generator;
-extern int incremental;
+extern int inc_recurse;
extern int io_error;
extern int eol_nulls;
extern int flist_eof;
const char phase_unknown[] = "unknown";
int ignore_timeout = 0;
int batch_fd = -1;
-int done_cnt = 0;
+int msgdone_cnt = 0;
/* Ignore an EOF error if non-zero. See whine_about_eof(). */
int kluge_around_eof = 0;
switch (tag) {
case MSG_DONE:
- if (len != 0 || !am_generator) {
+ if (len < 0 || len > 1 || !am_generator) {
invalid_msg:
rprintf(FERROR, "invalid message %d:%d [%s%s]\n",
tag, len, who_am_i(),
- incremental ? "/incremental" : "");
+ inc_recurse ? "/inc" : "");
exit_cleanup(RERR_STREAMIO);
}
- done_cnt++;
+ if (len) {
+ readfd(fd, buf, len);
+ stats.total_read = read_longint(fd);
+ }
+ msgdone_cnt++;
break;
case MSG_REDO:
if (len != 4 || !am_generator)
if (remove_source_files)
decrement_active_files(IVAL(buf,0));
flist_ndx_push(&redo_list, IVAL(buf,0));
- if (incremental)
+ if (inc_recurse)
decrement_flist_in_progress(IVAL(buf,0), 1);
break;
case MSG_FLIST:
- if (len != 4 || !am_generator || !incremental)
+ if (len != 4 || !am_generator || !inc_recurse)
goto invalid_msg;
readfd(fd, buf, 4);
/* Read extra file list from receiver. */
flist->parent_ndx = IVAL(buf,0);
break;
case MSG_FLIST_EOF:
- if (len != 0 || !am_generator || !incremental)
+ if (len != 0 || !am_generator || !inc_recurse)
goto invalid_msg;
flist_eof = 1;
break;
}
if (preserve_hard_links)
flist_ndx_push(&hlink_list, IVAL(buf,0));
- if (incremental)
+ if (inc_recurse)
decrement_flist_in_progress(IVAL(buf,0), 0);
break;
case MSG_NO_SEND:
if (len != 4 || !am_generator)
goto invalid_msg;
readfd(fd, buf, len);
- if (incremental)
+ if (inc_recurse)
decrement_flist_in_progress(IVAL(buf,0), 0);
break;
case MSG_SOCKERR:
{
/* TODO: tune these limits? */
while (active_filecnt >= (active_bytecnt >= 128*1024 ? 10 : 50)) {
-#ifdef SUPPORT_HARD_LINKS
- if (hlink_list.head)
- check_for_finished_hlinks(itemizing, code);
-#endif
- io_flush(NORMAL_FLUSH);
- read_msg_fd();
+ check_for_finished_files(itemizing, code, 0);
+ if (iobuf_out_cnt)
+ io_flush(NORMAL_FLUSH);
+ else
+ read_msg_fd();
}
active_filecnt++;
- active_bytecnt += F_LENGTH(cur_flist->files[ndx]);
+ active_bytecnt += F_LENGTH(cur_flist->files[ndx - cur_flist->ndx_start]);
}
static void decrement_active_files(int ndx)
void wait_for_receiver(void)
{
- io_flush(NORMAL_FLUSH);
- read_msg_fd();
+ if (iobuf_out_cnt)
+ io_flush(NORMAL_FLUSH);
+ else
+ read_msg_fd();
}
int get_redo_num(void)
count = select(maxfd + 1, &r_fds, &w_fds, NULL, &tv);
if (count <= 0) {
- if (errno == EBADF)
+ if (errno == EBADF) {
+ defer_forwarding_messages = 0;
exit_cleanup(RERR_SOCKETIO);
+ }
check_timeout();
continue;
}
/* Don't try to write errors back across the stream. */
if (fd == sock_f_out)
io_end_multiplex_out();
+ /* Don't try to write errors down a failing msg pipe. */
+ if (am_server && fd == msg_fd_out)
+ exit_cleanup(RERR_STREAMIO);
rsyserr(FERROR, errno,
"writefd_unbuffered failed to write %ld bytes [%s]",
(long)len, who_am_i());