diff options
| -rw-r--r-- | flist.c | 13 | ||||
| -rw-r--r-- | io.c | 69 | ||||
| -rw-r--r-- | main.c | 10 | ||||
| -rw-r--r-- | rsync.h | 7 |
4 files changed, 61 insertions, 38 deletions
@@ -52,6 +52,7 @@ extern int preserve_hard_links; extern int preserve_devices; extern int preserve_specials; extern int missing_args; +extern int sock_f_in; extern int uid_ndx; extern int gid_ndx; extern int eol_nulls; @@ -2092,6 +2093,13 @@ struct file_list *send_file_list(int f, int argc, char *argv[]) full_fname(argv[0])); exit_cleanup(RERR_FILESELECT); } + if (protocol_version == 30) { + /* Older protocols send the files-from data w/o packaging it in + * multiplexed I/O packets, but protocol 30 messed up and did + * this after starting multiplexing. We'll temporarily switch + * to buffered I/O to match this behavior. */ + io_end_multiplex_in(MPLX_TO_BUFFERED); + } use_ff_fd = 1; } @@ -2297,6 +2305,9 @@ struct file_list *send_file_list(int f, int argc, char *argv[]) send_file_name(f, flist, fbuf, &st, flags, NO_FILTERS); } + if (use_ff_fd && protocol_version == 30) + io_start_multiplex_in(sock_f_in); + gettimeofday(&end_tv, NULL); stats.flist_buildtime = (int64)(end_tv.tv_sec - start_tv.tv_sec) * 1000 + (end_tv.tv_usec - start_tv.tv_usec) / 1000; @@ -2352,7 +2363,7 @@ struct file_list *send_file_list(int f, int argc, char *argv[]) send_msg_int(MSG_IO_ERROR, io_error); if (disable_buffering) - io_end_buffering_out(True); + io_end_buffering_out(IOBUF_FREE_BUFS); stats.flist_size = stats.total_written - start_write; stats.num_files = flist->used; @@ -348,7 +348,7 @@ static void forward_filesfrom_data(void) ff_forward_fd = -1; write_buf(iobuf.out_fd, "\0\0", ff_lastchar ? 2 : 1); free_xbuf(&ff_xb); - if (protocol_version < 31) + if (protocol_version == 30) io_start_multiplex_out(iobuf.out_fd); } return; @@ -1016,18 +1016,15 @@ int get_hlink_num(void) * 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; + if (protocol_version == 30) { /* 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; + * multiplexed I/O packets, but protocol 30 messed up and did + * this after starting multiplexing. We'll temporarily switch + * to buffered I/O to match this behavior. */ + iobuf.msg.pos = iobuf.msg.len = 0; /* Be extra sure no messages go out. */ + io_end_multiplex_out(MPLX_TO_BUFFERED); } + ff_forward_fd = fd; alloc_xbuf(&ff_xb, FILESFROM_BUFLEN); } @@ -1192,9 +1189,9 @@ int io_start_buffering_in(int f_in) 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) @@ -1207,9 +1204,9 @@ void io_end_buffering_in(BOOL 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); @@ -1217,9 +1214,6 @@ void io_end_buffering_out(BOOL free_buffers) 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; @@ -2045,11 +2039,11 @@ void io_printf(int fd, const char *format, ...) /* 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); @@ -2067,27 +2061,32 @@ void io_start_multiplex_in(int fd) io_start_buffering_in(fd); } -void io_end_multiplex_in(BOOL free_buffers) +void 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"); - } + 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); } /* Stop output multiplexing. */ -void io_end_multiplex_out(BOOL free_buffers) +void 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"); - } + 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; } @@ -848,7 +848,7 @@ static int do_recv(int f_in, int f_out, char *local_name) close(error_pipe[0]); /* We can't let two processes write to the socket at one time. */ - io_end_multiplex_out(False); + io_end_multiplex_out(MPLX_SWITCHING); if (f_in != f_out) close(f_out); sock_f_out = -1; @@ -895,7 +895,7 @@ static int do_recv(int f_in, int f_out, char *local_name) am_generator = 1; - io_end_multiplex_in(False); + io_end_multiplex_in(MPLX_SWITCHING); if (write_batch && !am_server) stop_write_batch(); @@ -1056,6 +1056,8 @@ void start_server(int f_in, int f_out, int argc, char *argv[]) keep_dirlinks = 0; /* Must be disabled on the sender. */ if (need_messages_from_generator) io_start_multiplex_in(f_in); + else + io_start_buffering_in(f_in); recv_filter_list(f_in); do_server_sender(f_in, f_out, argc, argv); } else @@ -1104,6 +1106,8 @@ int client_run(int f_in, int f_out, pid_t pid, int argc, char *argv[]) io_start_buffering_out(f_out); if (protocol_version >= 31 || (!filesfrom_host && protocol_version >= 23)) io_start_multiplex_in(f_in); + else + io_start_buffering_in(f_in); send_filter_list(f_out); if (filesfrom_host) filesfrom_fd = f_in; @@ -1139,6 +1143,8 @@ int client_run(int f_in, int f_out, pid_t pid, int argc, char *argv[]) io_start_multiplex_in(f_in); if (need_messages_from_generator) io_start_multiplex_out(f_out); + else + io_start_buffering_out(f_out); } send_filter_list(read_batch ? -1 : f_out); @@ -903,6 +903,13 @@ typedef struct { #define ICB_CIRCULAR_OUT (1<<3) #define ICB_INIT (1<<4) +#define IOBUF_KEEP_BUFS 0 +#define IOBUF_FREE_BUFS 1 + +#define MPLX_SWITCHING IOBUF_KEEP_BUFS +#define MPLX_ALL_DONE IOBUF_FREE_BUFS +#define MPLX_TO_BUFFERED 2 + #define RL_EOL_NULLS (1<<0) #define RL_DUMP_COMMENTS (1<<1) #define RL_CONVERT (1<<2) |