int sock_f_in = -1;
int sock_f_out = -1;
+int64 total_data_read = 0;
+int64 total_data_written = 0;
+
static struct {
xbuf in, out, msg;
int in_fd;
static int write_batch_monitor_out = -1;
static int ff_forward_fd = -1;
-static char ff_lastchar;
+static int ff_reenable_multiplex = -1;
+static char ff_lastchar = '\0';
+static xbuf ff_xb = EMPTY_XBUF;
#ifdef ICONV_OPTION
static xbuf iconv_buf = EMPTY_XBUF;
#endif
* a chunk of data and put it into the output buffer. */
static void forward_filesfrom_data(void)
{
- char buf[FILESFROM_BUFLEN];
int len;
- xbuf x;
-
- INIT_CONST_XBUF(x, buf);
- len = read(ff_forward_fd, x.buf, x.size);
+ len = read(ff_forward_fd, ff_xb.buf + ff_xb.len, ff_xb.size - ff_xb.len);
if (len <= 0) {
if (len == 0 || errno != EINTR) {
/* Send end-of-file marker */
- write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
ff_forward_fd = -1;
- if (protocol_version < 31)
- io_start_multiplex_out(iobuf.out_fd);
+ write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1);
+ free_xbuf(&ff_xb);
+ if (ff_reenable_multiplex >= 0)
+ io_start_multiplex_out(ff_reenable_multiplex);
}
return;
}
if (DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] files-from read=%ld\n", who_am_i(), (long)len);
+#ifdef ICONV_OPTION
+ len += ff_xb.len;
+#endif
+
if (!eol_nulls) {
- char *s = x.buf + len;
+ char *s = ff_xb.buf + len;
/* Transform CR and/or LF into '\0' */
- while (s-- > x.buf) {
+ while (s-- > ff_xb.buf) {
if (*s == '\n' || *s == '\r')
*s = '\0';
}
}
+
if (ff_lastchar)
- x.pos = 0;
+ ff_xb.pos = 0;
else {
- char *s = x.buf;
+ char *s = ff_xb.buf;
/* Last buf ended with a '\0', so don't let this buf start with one. */
while (len && *s == '\0')
s++, len--;
- x.pos = s - x.buf;
+ ff_xb.pos = s - ff_xb.buf;
}
+
+#ifdef ICONV_OPTION
+ if (filesfrom_convert && len) {
+ char *sob = ff_xb.buf + ff_xb.pos, *s = sob;
+ char *eob = sob + len;
+ int flags = ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_CIRCULAR_OUT;
+ if (ff_lastchar == '\0')
+ flags |= ICB_INIT;
+ /* Convert/send each null-terminated string separately, skipping empties. */
+ while (s != eob) {
+ if (*s++ == '\0') {
+ ff_xb.len = s - sob - 1;
+ if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0)
+ exit_cleanup(RERR_PROTOCOL); /* impossible? */
+ write_buf(iobuf.out_fd, s-1, 1); /* Send the '\0'. */
+ while (s != eob && *s == '\0')
+ s++;
+ sob = s;
+ ff_xb.pos = sob - ff_xb.buf;
+ flags |= ICB_INIT;
+ }
+ }
+
+ if ((ff_xb.len = s - sob) == 0)
+ ff_lastchar = '\0';
+ else {
+ /* Handle a partial string specially, saving any incomplete chars. */
+ flags &= ~ICB_INCLUDE_INCOMPLETE;
+ if (iconvbufs(ic_send, &ff_xb, &iobuf.out, flags) < 0) {
+ if (errno == E2BIG)
+ exit_cleanup(RERR_PROTOCOL); /* impossible? */
+ if (ff_xb.pos)
+ memmove(ff_xb.buf, ff_xb.buf + ff_xb.pos, ff_xb.len);
+ }
+ ff_lastchar = 'x'; /* Anything non-zero. */
+ }
+ } else
+#endif
+
if (len) {
- char *f = x.buf + x.pos;
- char *t = x.buf;
+ char *f = ff_xb.buf + ff_xb.pos;
+ char *t = ff_xb.buf;
char *eob = f + len;
/* Eliminate any multi-'\0' runs. */
while (f != eob) {
if (!(*t++ = *f++)) {
- while (f != eob && !*f)
- f++, len--;
+ while (f != eob && *f == '\0')
+ f++;
}
}
ff_lastchar = f[-1];
- }
-#ifdef ICONV_OPTION
- if (filesfrom_convert) {
- /* TODO would it help to translate each string between nulls separately? */
- x.len = len;
- iconvbufs(ic_send, &x, &iobuf.out, ICB_INCLUDE_BAD|ICB_INCLUDE_INCOMPLETE|ICB_CIRCULAR_OUT|ICB_INIT);
- } else
-#endif
- if (len) {
- /* This will not circle back to perform_io() because we only get
- * called when there is plenty of room in the output buffer. */
- write_buf(iobuf.out_fd, x.buf, len);
+ if ((len = t - ff_xb.buf) != 0) {
+ /* This will not circle back to perform_io() because we only get
+ * called when there is plenty of room in the output buffer. */
+ write_buf(iobuf.out_fd, ff_xb.buf, len);
+ }
}
}
/* Don't write errors on a dead socket. */
msgs2stderr = 1;
out->len = iobuf.raw_flushing_ends_before = out->pos = 0;
- rsyserr(FERROR_SOCKET, errno, "write error");
+ rsyserr(FERROR_SOCKET, errno, "[%s] write error", who_am_i());
exit_cleanup(RERR_STREAMIO);
}
}
return data;
}
+void noop_io_until_death(void)
+{
+ char buf[1024];
+
+ kluge_around_eof = 1;
+ set_io_timeout(10);
+
+ while (1)
+ read_buf(iobuf.in_fd, buf, sizeof buf);
+}
+
/* Buffer a message for the multiplexed output stream. Is never used for MSG_DATA. */
int send_msg(enum msgcode code, const char *buf, size_t len, int convert)
{
* for recv_file_list() to use. */
void start_filesfrom_forwarding(int fd)
{
- ff_forward_fd = fd;
- if (protocol_version < 31) {
- int save_fd = iobuf.out_fd;
- /* Older protocols send the files-from data w/o packaging it in
- * multiplexed I/O packets. To match this, we temporarily turn
- * off the multiplexing of our output w/o disabling buffering. */
- assert(OUT_MULTIPLEXED);
- /* Be extra, extra sure no messages go out before files-from data. */
- iobuf.msg.pos = iobuf.msg.len = 0;
- io_end_multiplex_out(False);
- iobuf.out_fd = save_fd;
+ if (protocol_version < 31 && OUT_MULTIPLEXED) {
+ /* Older protocols send the files-from data w/o packaging
+ * it in multiplexed I/O packets, so temporarily switch
+ * to buffered I/O to match this behavior. */
+ iobuf.msg.pos = iobuf.msg.len = 0; /* Be extra sure no messages go out. */
+ ff_reenable_multiplex = io_end_multiplex_out(MPLX_TO_BUFFERED);
}
+ ff_forward_fd = fd;
+
+ alloc_xbuf(&ff_xb, FILESFROM_BUFLEN);
}
/* Read a line into the "buf" buffer. */
*argv_p = argv;
}
-int io_start_buffering_out(int f_out)
+BOOL io_start_buffering_out(int f_out)
{
if (msgs2stderr && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_buffering_out(%d)\n", who_am_i(), f_out);
if (iobuf.out_fd == -1)
iobuf.out_fd = f_out;
assert(f_out == iobuf.out_fd);
- return 0;
+ return False;
}
iobuf.out.size = IO_BUFFER_SIZE * 2 - 4;
iobuf.out.pos = iobuf.out.len = 0;
iobuf.out_fd = f_out;
- return 1;
+ return True;
}
-int io_start_buffering_in(int f_in)
+BOOL io_start_buffering_in(int f_in)
{
if (msgs2stderr && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_buffering_in(%d)\n", who_am_i(), f_in);
if (iobuf.in_fd == -1)
iobuf.in_fd = f_in;
assert(f_in == iobuf.in_fd);
- return 0;
+ return False;
}
iobuf.in.size = IO_BUFFER_SIZE;
iobuf.in_fd = f_in;
- return 1;
+ return True;
}
void io_end_buffering_in(BOOL free_buffers)
{
- if (DEBUG_GTE(IO, 2)) {
- rprintf(FINFO, "[%s] io_end_buffering_in(%s)\n",
- who_am_i(), free_buffers ? "True" : "False");
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_buffering_in(IOBUF_%s_BUFS)\n",
+ who_am_i(), free_buffers ? "FREE" : "KEEP");
}
if (free_buffers)
void io_end_buffering_out(BOOL free_buffers)
{
- if (DEBUG_GTE(IO, 2)) {
- rprintf(FINFO, "[%s] io_end_buffering_out(%s)\n",
- who_am_i(), free_buffers ? "True" : "False");
+ if (msgs2stderr && DEBUG_GTE(IO, 2)) {
+ rprintf(FINFO, "[%s] io_end_buffering_out(IOBUF_%s_BUFS)\n",
+ who_am_i(), free_buffers ? "FREE" : "KEEP");
}
io_flush(FULL_FLUSH);
if (free_buffers) {
free_xbuf(&iobuf.out);
free_xbuf(&iobuf.msg);
- } else {
- iobuf.out.pos = iobuf.out.len = 0;
- iobuf.msg.pos = iobuf.msg.len = 0;
}
iobuf.out_fd = -1;
static void read_a_msg(void)
{
char *data, line[BIGPATHBUFLEN];
- int tag;
+ int tag, val;
size_t msg_bytes;
data = perform_io(4, PIO_INPUT_AND_CONSUME);
if (!am_generator)
send_msg(MSG_IO_ERROR, data, 4, 0);
break;
+ 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);
+ if (!io_timeout || io_timeout > val) {
+ if (INFO_GTE(MISC, 2))
+ rprintf(FINFO, "Setting --timeout=%d to match server\n", val);
+ set_io_timeout(val);
+ }
+ break;
case MSG_NOOP:
if (am_sender)
maybe_send_keepalive();
xbuf outbuf, inbuf;
char ibuf[512];
int add_null = 0;
+ int flags = ICB_INCLUDE_BAD | ICB_INIT;
INIT_CONST_XBUF(outbuf, line);
INIT_XBUF(inbuf, ibuf, 0, (size_t)-1);
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);
inbuf.pos = 0;
- inbuf.len = msg_bytes > sizeof ibuf ? sizeof ibuf : msg_bytes;
- memcpy(inbuf.buf, perform_io(inbuf.len, PIO_INPUT_AND_CONSUME), inbuf.len);
- if (!(msg_bytes -= inbuf.len)
- && !ibuf[inbuf.len-1])
+ inbuf.len += len;
+ if (!(msg_bytes -= len) && !ibuf[inbuf.len-1])
inbuf.len--, add_null = 1;
- if (iconvbufs(ic_send, &inbuf, &outbuf,
- ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE | ICB_INIT) < 0)
- goto overflow;
+ if (iconvbufs(ic_send, &inbuf, &outbuf, flags) < 0) {
+ if (errno == E2BIG)
+ goto overflow;
+ /* Buffer ended with an incomplete char, so move the
+ * bytes to the start of the buffer and continue. */
+ memmove(ibuf, ibuf + inbuf.pos, inbuf.len);
+ }
+ flags &= ~ICB_INIT;
}
if (add_null) {
if (outbuf.len == outbuf.size)
first_message = 0;
}
break;
+ case MSG_ERROR_EXIT:
+ if (msg_bytes != 4)
+ goto invalid_msg;
+ data = perform_io(4, PIO_INPUT_AND_CONSUME);
+ val = IVAL(data, 0);
+ if (am_generator && protocol_version >= 31)
+ send_msg_int(MSG_ERROR_EXIT, val);
+ if (am_generator)
+ val = RERR_RCVR_ERROR; /* avoids duplicate errors */
+ exit_cleanup(val);
default:
rprintf(FERROR, "unexpected tag %d [%s%s]\n",
tag, who_am_i(), inc_recurse ? "/inc" : "");
if (!IN_MULTIPLEXED) {
memcpy(buf, perform_io(len, PIO_INPUT_AND_CONSUME), len);
+ total_data_read += len;
if (forward_flist_data)
write_buf(iobuf.out_fd, buf, len);
batch_copy:
/* The bytes at the "data" pointer will survive long
* enough to make a copy, but not past future I/O. */
memcpy(buf, data, siz);
+ total_data_read += siz;
if (forward_flist_data)
write_buf(iobuf.out_fd, buf, siz);
memcpy(iobuf.out.buf + pos, buf, len);
iobuf.out.len += len;
+ total_data_written += len;
batch_copy:
if (f == write_batch_monitor_out)
/* Setup for multiplexing a MSG_* stream with the data stream. */
void io_start_multiplex_out(int fd)
{
+ io_flush(FULL_FLUSH);
+
if (msgs2stderr && DEBUG_GTE(IO, 2))
rprintf(FINFO, "[%s] io_start_multiplex_out(%d)\n", who_am_i(), fd);
- io_flush(FULL_FLUSH);
-
iobuf.out_empty_len = 4; /* See also OUT_MULTIPLEXED */
io_start_buffering_out(fd);
io_start_buffering_in(fd);
}
-void io_end_multiplex_in(BOOL free_buffers)
+int io_end_multiplex_in(int mode)
{
- if (DEBUG_GTE(IO, 2)) {
- rprintf(FINFO, "[%s] io_end_multiplex_in(%s)\n",
- who_am_i(), free_buffers ? "True" : "False");
- }
+ int ret = iobuf.in_multiplexed ? iobuf.in_fd : -1;
+
+ if (msgs2stderr && DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] io_end_multiplex_in(mode=%d)\n", who_am_i(), mode);
iobuf.in_multiplexed = False;
- iobuf.raw_input_ends_before = 0;
- io_end_buffering_in(free_buffers);
+ if (mode == MPLX_SWITCHING)
+ iobuf.raw_input_ends_before = 0;
+ else
+ assert(iobuf.raw_input_ends_before == 0);
+ if (mode != MPLX_TO_BUFFERED)
+ io_end_buffering_in(mode);
+
+ return ret;
}
-/* Stop output multiplexing. */
-void io_end_multiplex_out(BOOL free_buffers)
+int io_end_multiplex_out(int mode)
{
- if (DEBUG_GTE(IO, 2)) {
- rprintf(FINFO, "[%s] io_end_multiplex_out(%s)\n",
- who_am_i(), free_buffers ? "True" : "False");
- }
+ int ret = iobuf.out_empty_len ? iobuf.out_fd : -1;
+
+ if (msgs2stderr && DEBUG_GTE(IO, 2))
+ rprintf(FINFO, "[%s] io_end_multiplex_out(mode=%d)\n", who_am_i(), mode);
+
+ if (mode != MPLX_TO_BUFFERED)
+ io_end_buffering_out(mode);
+ else
+ io_flush(FULL_FLUSH);
- io_end_buffering_out(free_buffers);
+ iobuf.out.len = 0;
iobuf.out_empty_len = 0;
+
+ return ret;
}
void start_write_batch(int fd)
Matt McCutchen's Web Site / rsync / rsync.git / blobdiff