Make some RERR_* choices better, and another noop_io_until_death() tweak.

[rsync/rsync.git] / io.c

diff --git a/io.c b/io.c
index 413557a..4a6b95f 100644 (file)
--- a/io.c
+++ b/io.c
@@ -136,6 +136,7 @@ enum festatus { FES_SUCCESS, FES_REDO, FES_NO_SEND };
 
 static flist_ndx_list redo_list, hlink_list;
 
+static void drain_multiplex_messages(void);
 static void sleep_for_bwlimit(int bytes_written);
 
 static void check_timeout(void)
@@ -173,9 +174,9 @@ static void check_timeout(void)
  * There is another case for older protocol versions (< 24) where the module
  * listing was not terminated, so we must ignore an EOF error in that case and
  * exit.  In this situation, kluge_around_eof will be > 0. */
-static NORETURN void whine_about_eof(int fd)
+static NORETURN void whine_about_eof(BOOL allow_kluge)
 {
-       if (kluge_around_eof && fd == sock_f_in) {
+       if (kluge_around_eof && allow_kluge) {
                int i;
                if (kluge_around_eof > 0)
                        exit_cleanup(0);
@@ -661,11 +662,19 @@ static char *perform_io(size_t needed, int flags)
                if (max_fd < 0) {
                        switch (flags & PIO_NEED_FLAGS) {
                        case PIO_NEED_INPUT:
+                               iobuf.in.len = 0;
+                               if (kluge_around_eof == 2)
+                                       exit_cleanup(0);
+                               if (iobuf.in_fd == -2)
+                                       whine_about_eof(True);
                                rprintf(FERROR, "error in perform_io: no fd for input.\n");
                                exit_cleanup(RERR_PROTOCOL);
                        case PIO_NEED_OUTROOM:
                        case PIO_NEED_MSGROOM:
                                msgs2stderr = 1;
+                               drain_multiplex_messages();
+                               if (iobuf.out_fd == -2)
+                                       whine_about_eof(True);
                                rprintf(FERROR, "error in perform_io: no fd for output.\n");
                                exit_cleanup(RERR_PROTOCOL);
                        default:
@@ -709,8 +718,9 @@ static char *perform_io(size_t needed, int flags)
                        int n;
                        if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
                                if (n == 0) {
+                                       /* Signal that input has become invalid. */
                                        if (!read_batch || batch_fd < 0 || am_generator)
-                                               whine_about_eof(iobuf.in_fd); /* Doesn't return. */
+                                               iobuf.in_fd = -2;
                                        batch_fd = -1;
                                        continue;
                                }
@@ -724,7 +734,7 @@ static char *perform_io(size_t needed, int flags)
                                                rsyserr(FERROR_SOCKET, errno, "read error");
                                        } else
                                                rsyserr(FERROR, errno, "read error");
-                                       exit_cleanup(RERR_STREAMIO);
+                                       exit_cleanup(RERR_SOCKETIO);
                                }
                        }
                        if (msgs2stderr && DEBUG_GTE(IO, 2))
@@ -752,9 +762,11 @@ static char *perform_io(size_t needed, int flags)
                                else {
                                        /* Don't write errors on a dead socket. */
                                        msgs2stderr = 1;
-                                       out->len = iobuf.raw_flushing_ends_before = out->pos = 0;
+                                       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());
-                                       exit_cleanup(RERR_STREAMIO);
+                                       drain_multiplex_messages();
+                                       exit_cleanup(RERR_SOCKETIO);
                                }
                        }
                        if (msgs2stderr && DEBUG_GTE(IO, 2)) {
@@ -773,8 +785,7 @@ static char *perform_io(size_t needed, int flags)
                                if (iobuf.raw_flushing_ends_before)
                                        iobuf.raw_flushing_ends_before -= out->size;
                                out->pos = 0;
-                       }
-                       if (out->pos == iobuf.raw_flushing_ends_before)
+                       } else if (out->pos == iobuf.raw_flushing_ends_before)
                                iobuf.raw_flushing_ends_before = 0;
                        if ((out->len -= n) == empty_buf_len) {
                                out->pos = 0;
@@ -807,8 +818,11 @@ void noop_io_until_death(void)
 {
        char buf[1024];
 
-       kluge_around_eof = 1;
-       set_io_timeout(protocol_version >= 31 ? 10 : 1);
+       kluge_around_eof = 2;
+       /* Setting an I/O timeout ensures that if something inexplicably weird
+        * happens, we won't hang around forever. */
+       if (!io_timeout)
+               set_io_timeout(60);
 
        while (1)
                read_buf(iobuf.in_fd, buf, sizeof buf);
@@ -831,7 +845,7 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
        if (convert > 0 && ic_send == (iconv_t)-1)
                convert = 0;
        if (convert > 0) {
-               /* Ensuring double-size room leaves space for a potential conversion. */
+               /* Ensuring double-size room leaves space for maximal conversion expansion. */
                if (iobuf.msg.len + len*2 + 4 > iobuf.msg.size)
                        perform_io(len*2 + 4, PIO_NEED_MSGROOM);
        } else
@@ -844,7 +858,7 @@ int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
                pos -= iobuf.msg.size;
        hdr = iobuf.msg.buf + pos;
 
-       iobuf.msg.len += 4; /* Leave room for the coming header bytes. */
+       iobuf.msg.len += 4; /* Allocate room for the coming header bytes. */
 
 #ifdef ICONV_OPTION
        if (convert > 0) {
@@ -1165,7 +1179,8 @@ BOOL io_start_buffering_out(int f_out)
        if (iobuf.out.buf) {
                if (iobuf.out_fd == -1)
                        iobuf.out_fd = f_out;
-               assert(f_out == iobuf.out_fd);
+               else
+                       assert(f_out == iobuf.out_fd);
                return False;
        }
 
@@ -1187,7 +1202,8 @@ BOOL io_start_buffering_in(int f_in)
        if (iobuf.in.buf) {
                if (iobuf.in_fd == -1)
                        iobuf.in_fd = f_in;
-               assert(f_in == iobuf.in_fd);
+               else
+                       assert(f_in == iobuf.in_fd);
                return False;
        }
 
@@ -1469,6 +1485,23 @@ static void read_a_msg(void)
        }
 }
 
+static void drain_multiplex_messages(void)
+{
+       while (IN_MULTIPLEXED && iobuf.in.len) {
+               if (iobuf.raw_input_ends_before) {
+                       size_t raw_len = iobuf.raw_input_ends_before - iobuf.in.pos;
+                       iobuf.raw_input_ends_before = 0;
+                       if (raw_len >= iobuf.in.len) {
+                               iobuf.in.len = 0;
+                               break;
+                       }
+                       iobuf.in.pos += raw_len;
+                       iobuf.in.len -= raw_len;
+               }
+               read_a_msg();
+       }
+}
+
 void wait_for_receiver(void)
 {
        if (!iobuf.raw_input_ends_before)
@@ -1622,7 +1655,7 @@ void read_buf(int f, char *buf, size_t len)
 {
        if (f != iobuf.in_fd) {
                if (safe_read(f, buf, len) != len)
-                       whine_about_eof(f); /* Doesn't return. */
+                       whine_about_eof(False); /* Doesn't return. */
                goto batch_copy;
        }
 
@@ -2079,11 +2112,11 @@ void io_printf(int fd, const char *format, ...)
        va_end(ap);
 
        if (len < 0)
-               exit_cleanup(RERR_STREAMIO);
+               exit_cleanup(RERR_PROTOCOL);
 
        if (len > (int)sizeof buf) {
                rprintf(FERROR, "io_printf() was too long for the buffer.\n");
-               exit_cleanup(RERR_STREAMIO);
+               exit_cleanup(RERR_PROTOCOL);
        }
 
        write_sbuf(fd, buf);