Turn iobuf.in into a circular input buffer.

diff --git a/io.c b/io.c
index 4c70024..91815ad 100644 (file)
--- a/io.c
+++ b/io.c
@@ -489,19 +489,6 @@ void restore_iobuf_size(xbuf *out)
        }
 }
 
        }
 }
 

-static void slide_iobuf_in(size_t needed)
-{
-       memmove(iobuf.in.buf, iobuf.in.buf + iobuf.in.pos, iobuf.in.len);
-       if (DEBUG_GTE(IO, 4)) {
-               rprintf(FINFO,
-                       "[%s] moved %ld bytes from %ld to 0 in the input buffer (size=%ld, needed=%ld).\n",
-                       who_am_i(), (long)iobuf.in.len, (long)iobuf.in.pos, (long)iobuf.in.size, (long)needed);
-       }
-       if (iobuf.raw_input_ends_before)
-               iobuf.raw_input_ends_before -= iobuf.in.pos;
-       iobuf.in.pos = 0;
-}
-

 /* Perform buffered input and output until specified conditions are met.  When
  * given a "needed" read requirement, we'll return without doing any I/O if the
  * iobuf.in bytes are already available.  When reading, we'll read as many
 /* Perform buffered input and output until specified conditions are met.  When
  * given a "needed" read requirement, we'll return without doing any I/O if the
  * iobuf.in bytes are already available.  When reading, we'll read as many


@@ -515,10 +502,11 @@ static void slide_iobuf_in(size_t needed)
  * The iobuf.out and iobuf.msg buffers are circular, so some writes into them
  * will need to be split when the data needs to wrap around to the start.  In
  * order to help make this easier for some operations (such as the use of
  * The iobuf.out and iobuf.msg buffers are circular, so some writes into them
  * will need to be split when the data needs to wrap around to the start.  In
  * order to help make this easier for some operations (such as the use of

- * SIVAL() into the buffer) the buffers MUST have 4 bytes of overflow space at
- * the end that is not not counted in the "size".  The iobuf.in buffer is not
- * (currently) circular.  To facilitate the handling of MSG_DATA bytes as they
- * are read-from/written-into the buffers, see the three raw_* iobuf vars.
+ * SIVAL() into the buffer) a buffer may be temporarily shortened, but the
+ * original size will be automatically restored.  The iobuf.in buffer is also
+ * circular, so callers may need to split their reading of the data if it spans
+ * the end.  See also the 3 raw_* iobuf vars that are used in the handling of
+ * MSG_DATA bytes as they are read-from/written-into the buffers.

  *
  * When writing, we flush data in the following priority order:
  *
  *
  * When writing, we flush data in the following priority order:
  *


@@ -535,8 +523,6 @@ static void slide_iobuf_in(size_t needed)
  *
  *    - Make this routine able to read the generator-to-receiver batch flow?
  *
  *
  *    - Make this routine able to read the generator-to-receiver batch flow?
  *

- *    - Make the input buffer circular?
- *

  * Unlike the old routines that this replaces, it is OK to read ahead as far as
  * we can because the read_a_msg() routine now reads its bytes out of the input
  * buffer.  In the old days, only raw data was in the input buffer, and any
  * Unlike the old routines that this replaces, it is OK to read ahead as far as
  * we can because the read_a_msg() routine now reads its bytes out of the input
  * buffer.  In the old days, only raw data was in the input buffer, and any


@@ -558,24 +544,17 @@ static char *perform_io(size_t needed, int flags)
 
        switch (flags & PIO_NEED_FLAGS) {
        case PIO_NEED_INPUT:
 
        switch (flags & PIO_NEED_FLAGS) {
        case PIO_NEED_INPUT:

+               /* We never resize the circular input buffer. */
+               if (iobuf.in.size < needed) {
+                       rprintf(FERROR, "need to read %ld bytes, iobuf.in.buf is only %ld bytes.\n",
+                               (long)needed, (long)iobuf.in.size);
+                       exit_cleanup(RERR_PROTOCOL);
+               }
+

                if (DEBUG_GTE(IO, 3)) {
                        rprintf(FINFO, "[%s] perform_io(%ld, %sinput)\n",
                                who_am_i(), (long)needed, flags & PIO_CONSUME_INPUT ? "consume&" : "");
                }
                if (DEBUG_GTE(IO, 3)) {
                        rprintf(FINFO, "[%s] perform_io(%ld, %sinput)\n",
                                who_am_i(), (long)needed, flags & PIO_CONSUME_INPUT ? "consume&" : "");
                }

-
-               /* Make sure the input buffer is big enough to hold "needed" bytes.
-                * Also make sure it will fit in the free space at the end, or
-                * else we need to shift some bytes. */
-               if (needed && iobuf.in.size < needed) {
-                       size_t new_size = ROUND_UP_1024(needed);
-                       if (DEBUG_GTE(IO, 4)) {
-                               rprintf(FINFO, "[%s] resizing input buffer from %ld to %ld bytes.\n",
-                                       who_am_i(), (long)iobuf.in.size, (long)new_size);
-                       }
-                       realloc_xbuf(&iobuf.in, new_size);
-               }
-               if (iobuf.in.size - iobuf.in.pos < needed)
-                       slide_iobuf_in(needed);

                break;
 
        case PIO_NEED_OUTROOM:
                break;
 
        case PIO_NEED_OUTROOM:


@@ -637,14 +616,11 @@ static char *perform_io(size_t needed, int flags)
                        break;
                }
 
                        break;
                }
 

-               if (iobuf.in.len < 1024 && iobuf.in.size - (iobuf.in.pos + iobuf.in.len) < 1024)
-                       slide_iobuf_in(flags & PIO_NEED_INPUT ? needed : 0);
-

                max_fd = -1;
 
                FD_ZERO(&r_fds);
                FD_ZERO(&e_fds);
                max_fd = -1;
 
                FD_ZERO(&r_fds);
                FD_ZERO(&e_fds);

-               if (iobuf.in_fd >= 0 && iobuf.in.size - (iobuf.in.pos + iobuf.in.len)) {
+               if (iobuf.in_fd >= 0 && iobuf.in.size - iobuf.in.len) {

                        if (!read_batch || batch_fd >= 0) {
                                FD_SET(iobuf.in_fd, &r_fds);
                                FD_SET(iobuf.in_fd, &e_fds);
                        if (!read_batch || batch_fd >= 0) {
                                FD_SET(iobuf.in_fd, &r_fds);
                                FD_SET(iobuf.in_fd, &e_fds);


@@ -761,9 +737,13 @@ static char *perform_io(size_t needed, int flags)
                }
 
                if (iobuf.in_fd >= 0 && FD_ISSET(iobuf.in_fd, &r_fds)) {
                }
 
                if (iobuf.in_fd >= 0 && FD_ISSET(iobuf.in_fd, &r_fds)) {

-                       size_t pos = iobuf.in.pos + iobuf.in.len;
-                       size_t len = iobuf.in.size - pos;
+                       size_t len, pos = iobuf.in.pos + iobuf.in.len;

                        int n;
                        int n;

+                       if (pos >= iobuf.in.size) {
+                               pos -= iobuf.in.size;
+                               len = iobuf.in.size - iobuf.in.len;
+                       } else
+                               len = iobuf.in.size - pos;

                        if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
                                if (n == 0) {
                                        /* Signal that input has become invalid. */
                        if ((n = read(iobuf.in_fd, iobuf.in.buf + pos, len)) <= 0) {
                                if (n == 0) {
                                        /* Signal that input has become invalid. */


@@ -868,11 +848,40 @@ static char *perform_io(size_t needed, int flags)
        if (flags & PIO_CONSUME_INPUT) {
                iobuf.in.len -= needed;
                iobuf.in.pos += needed;
        if (flags & PIO_CONSUME_INPUT) {
                iobuf.in.len -= needed;
                iobuf.in.pos += needed;

+               if (iobuf.in.pos == iobuf.raw_input_ends_before)
+                       iobuf.raw_input_ends_before = 0;
+               if (iobuf.in.pos >= iobuf.in.size) {
+                       iobuf.in.pos -= iobuf.in.size;
+                       if (iobuf.raw_input_ends_before)
+                               iobuf.raw_input_ends_before -= iobuf.in.size;
+               }

        }
 
        return data;
 }
 
        }
 
        return data;
 }
 

+static void raw_read_buf(char *buf, size_t len)
+{
+       size_t pos = iobuf.in.pos;
+       char *data = perform_io(len, PIO_INPUT_AND_CONSUME);
+       if (iobuf.in.pos <= pos && len) {
+               size_t siz = len - iobuf.in.pos;
+               memcpy(buf, data, siz);
+               memcpy(buf + siz, iobuf.in.buf, iobuf.in.pos);
+       } else
+               memcpy(buf, data, len);
+}
+
+static int32 raw_read_int(void)
+{
+       char *data, buf[4];
+       if (iobuf.in.size - iobuf.in.pos >= 4)
+               data = perform_io(4, PIO_INPUT_AND_CONSUME);
+       else
+               raw_read_buf(data = buf, 4);
+       return IVAL(data, 0);
+}
+

 void noop_io_until_death(void)
 {
        char buf[1024];
 void noop_io_until_death(void)
 {
        char buf[1024];


@@ -1338,12 +1347,11 @@ void stop_flist_forward(void)
 /* Read a message from a multiplexed source. */
 static void read_a_msg(void)
 {
 /* Read a message from a multiplexed source. */
 static void read_a_msg(void)
 {

-       char *data, line[BIGPATHBUFLEN];
+       char line[BIGPATHBUFLEN];

        int tag, val;
        size_t msg_bytes;
 
        int tag, val;
        size_t msg_bytes;
 

-       data = perform_io(4, PIO_INPUT_AND_CONSUME);
-       tag = IVAL(data, 0);
+       tag = raw_read_int();

 
        msg_bytes = tag & 0xFFFFFF;
        tag = (tag >> 24) - MPLEX_BASE;
 
        msg_bytes = tag & 0xFFFFFF;
        tag = (tag >> 24) - MPLEX_BASE;


@@ -1364,20 +1372,17 @@ static void read_a_msg(void)
        case MSG_STATS:
                if (msg_bytes != sizeof stats.total_read || !am_generator)
                        goto invalid_msg;
        case MSG_STATS:
                if (msg_bytes != sizeof stats.total_read || !am_generator)
                        goto invalid_msg;

-               data = perform_io(sizeof stats.total_read, PIO_INPUT_AND_CONSUME);
-               memcpy((char*)&stats.total_read, data, sizeof stats.total_read);
+               raw_read_buf((char*)&stats.total_read, sizeof stats.total_read);

                break;
        case MSG_REDO:
                if (msg_bytes != 4 || !am_generator)
                        goto invalid_msg;
                break;
        case MSG_REDO:
                if (msg_bytes != 4 || !am_generator)
                        goto invalid_msg;

-               data = perform_io(4, PIO_INPUT_AND_CONSUME);
-               got_flist_entry_status(FES_REDO, IVAL(data, 0));
+               got_flist_entry_status(FES_REDO, raw_read_int());

                break;
        case MSG_IO_ERROR:
                if (msg_bytes != 4 || am_sender)
                        goto invalid_msg;
                break;
        case MSG_IO_ERROR:
                if (msg_bytes != 4 || am_sender)
                        goto invalid_msg;

-               data = perform_io(4, PIO_INPUT_AND_CONSUME);
-               val = IVAL(data, 0);
+               val = raw_read_int();

                io_error |= val;
                if (!am_generator)
                        send_msg_int(MSG_IO_ERROR, val);
                io_error |= val;
                if (!am_generator)
                        send_msg_int(MSG_IO_ERROR, val);


@@ -1385,8 +1390,7 @@ static void read_a_msg(void)
        case MSG_IO_TIMEOUT:
                if (msg_bytes != 4 || am_server || am_generator)
                        goto invalid_msg;
        case MSG_IO_TIMEOUT:
                if (msg_bytes != 4 || am_server || am_generator)
                        goto invalid_msg;

-               data = perform_io(4, PIO_INPUT_AND_CONSUME);
-               val = IVAL(data, 0);
+               val = raw_read_int();

                if (!io_timeout || io_timeout > val) {
                        if (INFO_GTE(MISC, 2))
                                rprintf(FINFO, "Setting --timeout=%d to match server\n", val);
                if (!io_timeout || io_timeout > val) {
                        if (INFO_GTE(MISC, 2))
                                rprintf(FINFO, "Setting --timeout=%d to match server\n", val);


@@ -1401,7 +1405,7 @@ static void read_a_msg(void)
                if (msg_bytes >= sizeof line)
                        goto overflow;
                if (am_generator) {
                if (msg_bytes >= sizeof line)
                        goto overflow;
                if (am_generator) {

-                       memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), msg_bytes);
+                       raw_read_buf(line, msg_bytes);

                        send_msg(MSG_DELETED, line, msg_bytes, 1);
                        break;
                }
                        send_msg(MSG_DELETED, line, msg_bytes, 1);
                        break;
                }


@@ -1417,7 +1421,7 @@ static void read_a_msg(void)
 
                        while (msg_bytes) {
                                size_t len = msg_bytes > sizeof ibuf - inbuf.len ? sizeof ibuf - inbuf.len : msg_bytes;
 
                        while (msg_bytes) {
                                size_t len = msg_bytes > sizeof ibuf - inbuf.len ? sizeof ibuf - inbuf.len : msg_bytes;

-                               memcpy(ibuf + inbuf.len, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+                               raw_read_buf(ibuf + inbuf.len, len);

                                inbuf.pos = 0;
                                inbuf.len += len;
                                if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])
                                inbuf.pos = 0;
                                inbuf.len += len;
                                if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])


@@ -1439,7 +1443,7 @@ static void read_a_msg(void)
                        msg_bytes = outbuf.len;
                } else
 #endif
                        msg_bytes = outbuf.len;
                } else
 #endif

-                       memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), msg_bytes);
+                       raw_read_buf(line, msg_bytes);

                /* A directory name was sent with the trailing null */
                if (msg_bytes > 0 && !line[msg_bytes-1])
                        log_delete(line, S_IFDIR);
                /* A directory name was sent with the trailing null */
                if (msg_bytes > 0 && !line[msg_bytes-1])
                        log_delete(line, S_IFDIR);


@@ -1456,8 +1460,7 @@ static void read_a_msg(void)
                                inc_recurse ? "/inc" : "");
                        exit_cleanup(RERR_STREAMIO);
                }
                                inc_recurse ? "/inc" : "");
                        exit_cleanup(RERR_STREAMIO);
                }

-               data = perform_io(4, PIO_INPUT_AND_CONSUME);
-               val = IVAL(data, 0);
+               val = raw_read_int();

                if (am_generator)
                        got_flist_entry_status(FES_SUCCESS, val);
                else
                if (am_generator)
                        got_flist_entry_status(FES_SUCCESS, val);
                else


@@ -1466,8 +1469,7 @@ static void read_a_msg(void)
        case MSG_NO_SEND:
                if (msg_bytes != 4)
                        goto invalid_msg;
        case MSG_NO_SEND:
                if (msg_bytes != 4)
                        goto invalid_msg;

-               data = perform_io(4, PIO_INPUT_AND_CONSUME);
-               val = IVAL(data, 0);
+               val = raw_read_int();

                if (am_generator)
                        got_flist_entry_status(FES_NO_SEND, val);
                else
                if (am_generator)
                        got_flist_entry_status(FES_NO_SEND, val);
                else


@@ -1494,7 +1496,7 @@ static void read_a_msg(void)
                                inc_recurse ? "/inc" : "");
                        exit_cleanup(RERR_STREAMIO);
                }
                                inc_recurse ? "/inc" : "");
                        exit_cleanup(RERR_STREAMIO);
                }

-               memcpy(line, perform_io(msg_bytes, PIO_INPUT_AND_CONSUME), msg_bytes);
+               raw_read_buf(line, msg_bytes);

                rwrite((enum logcode)tag, line, msg_bytes, !am_generator);
                if (first_message) {
                        if (list_only && !am_sender && tag == 1 && msg_bytes < sizeof line) {
                rwrite((enum logcode)tag, line, msg_bytes, !am_generator);
                if (first_message) {
                        if (list_only && !am_sender && tag == 1 && msg_bytes < sizeof line) {


@@ -1518,8 +1520,7 @@ static void read_a_msg(void)
                        }
                        val = 0;
                } else if (msg_bytes == 4) {
                        }
                        val = 0;
                } else if (msg_bytes == 4) {

-                       data = perform_io(4, PIO_INPUT_AND_CONSUME);
-                       val = IVAL(data, 0);
+                       val = raw_read_int();

                        if (protocol_version >= 31) {
                                if (am_generator) {
                                        if (DEBUG_GTE(EXIT, 3)) {
                        if (protocol_version >= 31) {
                                if (am_generator) {
                                        if (DEBUG_GTE(EXIT, 3)) {


@@ -1557,8 +1558,9 @@ static void drain_multiplex_messages(void)
                                iobuf.in.len = 0;
                                break;
                        }
                                iobuf.in.len = 0;
                                break;
                        }

-                       iobuf.in.pos += raw_len;

                        iobuf.in.len -= raw_len;
                        iobuf.in.len -= raw_len;

+                       if ((iobuf.in.pos += raw_len) >= iobuf.in.size)
+                               iobuf.in.pos -= iobuf.in.size;

                }
                read_a_msg();
        }
                }
                read_a_msg();
        }


@@ -1724,7 +1726,7 @@ void read_buf(int f, char *buf, size_t len)
        }
 
        if (!IN_MULTIPLEXED) {
        }
 
        if (!IN_MULTIPLEXED) {

-               memcpy(buf, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+               raw_read_buf(buf, len);

                total_data_read += len;
                if (forward_flist_data)
                        write_buf(iobuf.out_fd, buf, len);
                total_data_read += len;
                if (forward_flist_data)
                        write_buf(iobuf.out_fd, buf, len);


@@ -1735,20 +1737,15 @@ void read_buf(int f, char *buf, size_t len)
        }
 
        while (1) {
        }
 
        while (1) {

-               char *data;

                size_t siz;
 
                while (!iobuf.raw_input_ends_before)
                        read_a_msg();
 
                siz = MIN(len, iobuf.raw_input_ends_before - iobuf.in.pos);
                size_t siz;
 
                while (!iobuf.raw_input_ends_before)
                        read_a_msg();
 
                siz = MIN(len, iobuf.raw_input_ends_before - iobuf.in.pos);

-               data = perform_io(siz, PIO_INPUT_AND_CONSUME);
-               if (iobuf.in.pos == iobuf.raw_input_ends_before)
-                       iobuf.raw_input_ends_before = 0;
-
-               /* The bytes at the "data" pointer will survive long
-                * enough to make a copy, but not past future I/O. */
-               memcpy(buf, data, siz);
+               if (siz >= iobuf.in.size)
+                       siz = iobuf.in.size;
+               raw_read_buf(buf, siz);

                total_data_read += siz;
 
                if (forward_flist_data)
                total_data_read += siz;
 
                if (forward_flist_data)