More batch-mode fixes to handle redos properly (and without hanging).

diff --git a/io.c b/io.c
index 70756b8..a4e2435 100644 (file)
--- a/io.c
+++ b/io.c
@@ -164,6 +164,11 @@ static void got_flist_entry_status(enum festatus status, const char *buf)
                }
                break;
        case FES_REDO:
+               if (read_batch) {
+                       if (inc_recurse)
+                               flist->in_progress++;
+                       break;
+               }
                if (inc_recurse)
                        flist->to_redo++;
                flist_ndx_push(&redo_list, ndx);

diff --git a/receiver.c b/receiver.c
index 72972b0..b342afa 100644 (file)
--- a/receiver.c
+++ b/receiver.c
@@ -60,6 +60,7 @@ extern struct filter_list_struct daemon_filter_list;
 
 static struct bitbag *delayed_bits = NULL;
 static int phase = 0, redoing = 0;
+static flist_ndx_list batch_redo_list;
 /* We're either updating the basis file or an identical copy: */
 static int updating_basis_or_equiv;
 
@@ -349,29 +350,60 @@ static void handle_delayed_updates(char *local_name)
        }
 }
 
-static int get_next_gen_ndx(int fd, int next_gen_ndx, int desired_ndx)
+static void no_batched_update(int ndx, BOOL is_redo)
 {
-       static int batch_eof = 0;
-
-       while (next_gen_ndx < desired_ndx) {
-               if (next_gen_ndx >= 0) {
-                       struct file_struct *file = cur_flist->files[next_gen_ndx];
-                       rprintf(FERROR_XFER,
-                               "(No batched update for%s \"%s\")\n",
-                               file->flags & FLAG_FILE_SENT ? " resend of" : "",
-                               f_name(file, NULL));
-               }
-               next_gen_ndx = batch_eof ? -1 : read_int(fd);
-               if (next_gen_ndx == -1) {
-                       if (inc_recurse)
-                               next_gen_ndx = first_flist->prev->used + first_flist->prev->ndx_start;
-                       else
-                               next_gen_ndx = cur_flist->used;
-                       batch_eof = 1;
+       struct file_list *flist = flist_for_ndx(ndx, "no_batched_update");
+       struct file_struct *file = flist->files[ndx - flist->ndx_start];
+
+       rprintf(FERROR_XFER, "(No batched update for%s \"%s\")\n",
+               is_redo ? " resend of" : "", f_name(file, NULL));
+
+       if (inc_recurse)
+               send_msg_int(MSG_NO_SEND, ndx);
+}
+
+static int we_want_redo(int desired_ndx)
+{
+       static int redo_ndx = -1;
+
+       while (redo_ndx < desired_ndx) {
+               if (redo_ndx >= 0)
+                       no_batched_update(redo_ndx, True);
+               if ((redo_ndx = flist_ndx_pop(&batch_redo_list)) < 0)
+                       return 0;
+       }
+
+       if (redo_ndx == desired_ndx) {
+               redo_ndx = -1;
+               return 1;
+       }
+
+       return 0;
+}
+
+static int gen_wants_ndx(int desired_ndx)
+{
+       static int next_ndx = -1;
+       static BOOL got_eof = 0;
+
+       if (got_eof)
+               return 0;
+
+       while (next_ndx < desired_ndx) {
+               if (next_ndx >= 0)
+                       no_batched_update(next_ndx, False);
+               if ((next_ndx = read_int(batch_gen_fd)) < 0) {
+                       got_eof = True;
+                       return 0;
                }
        }
 
-       return next_gen_ndx;
+       if (next_ndx == desired_ndx) {
+               next_ndx = -1;
+               return 1;
+       }
+
+       return 0;
 }
 
 /**
@@ -380,7 +412,6 @@ static int get_next_gen_ndx(int fd, int next_gen_ndx, int desired_ndx)
  * Receiver process runs on the same host as the generator process. */
 int recv_files(int f_in, char *local_name)
 {
-       int next_gen_ndx = -1;
        int fd1,fd2;
        STRUCT_STAT st;
        int iflags, xlen;
@@ -419,17 +450,13 @@ int recv_files(int f_in, char *local_name)
                                end_progress(0);
                        }
                        if (inc_recurse && first_flist) {
+                               if (read_batch)
+                                       gen_wants_ndx(first_flist->used + first_flist->ndx_start);
                                flist_free(first_flist);
                                if (first_flist)
                                        continue;
-                       }
-                       if (read_batch && cur_flist) {
-                               int high = inc_recurse
-                                   ? first_flist->prev->used + first_flist->prev->ndx_start
-                                   : cur_flist->used;
-                               get_next_gen_ndx(batch_gen_fd, next_gen_ndx, high);
-                               next_gen_ndx = -1;
-                       }
+                       } else if (read_batch && first_flist)
+                               gen_wants_ndx(first_flist->used);
                        if (++phase > max_phase)
                                break;
                        if (DEBUG_GTE(RECV, 1))
@@ -518,15 +545,15 @@ int recv_files(int f_in, char *local_name)
                }
 
                if (read_batch) {
-                       next_gen_ndx = get_next_gen_ndx(batch_gen_fd, next_gen_ndx, ndx);
-                       if (ndx < next_gen_ndx) {
+                       if (!(redoing ? we_want_redo(ndx) : gen_wants_ndx(ndx))) {
                                rprintf(FINFO,
-                                       "(Skipping batched update for \"%s\")\n",
+                                       "(Skipping batched update for%s \"%s\")\n",
+                                       redoing ? " resend of" : "",
                                        fname);
                                discard_receive_data(f_in, F_LENGTH(file));
+                               file->flags |= FLAG_FILE_SENT;
                                continue;
                        }
-                       next_gen_ndx = -1;
                }
 
                partialptr = partial_dir ? partial_dir_fname(fname) : fname;
@@ -726,6 +753,9 @@ int recv_files(int f_in, char *local_name)
 
                cleanup_disable();
 
+               if (read_batch)
+                       file->flags |= FLAG_FILE_SENT;
+
                switch (recv_ok) {
                case 2:
                        break;
@@ -758,6 +788,8 @@ int recv_files(int f_in, char *local_name)
                                        keptstr, redostr);
                        }
                        if (!redoing) {
+                               if (read_batch)
+                                       flist_ndx_push(&batch_redo_list, ndx);
                                send_msg_int(MSG_REDO, ndx);
                                file->flags |= FLAG_FILE_SENT;
                        } else if (inc_recurse)