The "ndx" variable now holds the unique, over-the-wire value, not

[rsync/rsync.git] / io.c

diff --git a/io.c b/io.c
index 7bf0afc..3b857bf 100644 (file)
--- a/io.c
+++ b/io.c
@@ -4,7 +4,7 @@
  * 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, 2004, 2005, 2006, 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
@@ -41,7 +41,7 @@ extern int am_server;
 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;
@@ -58,7 +58,7 @@ extern struct file_list *cur_flist, *first_flist;
 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;
@@ -288,14 +288,18 @@ static void read_msg_fd(void)
 
        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)
@@ -304,11 +308,11 @@ static void read_msg_fd(void)
                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. */
@@ -318,7 +322,7 @@ static void read_msg_fd(void)
                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;
@@ -338,14 +342,14 @@ static void read_msg_fd(void)
                }
                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:
@@ -386,16 +390,15 @@ void increment_active_files(int ndx, int itemizing, enum logcode code)
 {
        /* 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)
@@ -481,8 +484,10 @@ void send_msg_int(enum msgcode code, int num)
 
 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)
@@ -597,8 +602,10 @@ static int read_timeout(int fd, char *buf, size_t len)
                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;
                }
@@ -1282,6 +1289,9 @@ static void writefd_unbuffered(int fd, const char *buf, size_t len)
                        /* 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());