From: Wayne Davison Date: Sat, 12 Dec 2009 23:16:46 +0000 (-0800) Subject: Turn iobuf.in into a circular input buffer. X-Git-Url: https://mattmccutchen.net/rsync/rsync.git/commitdiff_plain/763880ba81b469955b8653b0bb61a69d75cec0e8 Turn iobuf.in into a circular input buffer. --- diff --git a/io.c b/io.c index 4c700244..91815ad3 100644 --- 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 @@ -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 - * 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: * @@ -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 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 @@ -558,24 +544,17 @@ static char *perform_io(size_t needed, int flags) 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&" : ""); } - - /* 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: @@ -637,14 +616,11 @@ static char *perform_io(size_t needed, int flags) 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); - 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); @@ -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)) { - 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; + 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. */ @@ -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 (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; } +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]; @@ -1338,12 +1347,11 @@ void stop_flist_forward(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; - 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; @@ -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; - 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; - 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; - 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); @@ -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; - 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); @@ -1401,7 +1405,7 @@ static void read_a_msg(void) 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; } @@ -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; - 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]) @@ -1439,7 +1443,7 @@ static void read_a_msg(void) 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); @@ -1456,8 +1460,7 @@ static void read_a_msg(void) 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 @@ -1466,8 +1469,7 @@ static void read_a_msg(void) 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 @@ -1494,7 +1496,7 @@ static void read_a_msg(void) 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) { @@ -1518,8 +1520,7 @@ static void read_a_msg(void) } 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)) { @@ -1557,8 +1558,9 @@ static void drain_multiplex_messages(void) iobuf.in.len = 0; break; } - iobuf.in.pos += 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(); } @@ -1724,7 +1726,7 @@ void read_buf(int f, char *buf, size_t len) } 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); @@ -1735,20 +1737,15 @@ void read_buf(int f, char *buf, size_t len) } 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); - 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)