From 8d9dc9f99db13d42272b7c150f2e6849331ccdc1 Mon Sep 17 00:00:00 2001 From: Andrew Tridgell Date: Mon, 18 May 1998 08:20:26 +0000 Subject: [PATCH] this is a large commit which adds io multiplexing, thus giving error messages from the server when using the socket code. I've doen a fair bit of testing and I don't seem to have broken anything, but it is a very complex change. More testing needed! --- clientserver.c | 11 ++ io.c | 392 +++++++++++++++++++++++++++++++++---------------- log.c | 14 +- main.c | 58 +++++--- options.c | 1 + rsync.c | 1 + rsync.h | 3 +- socket.c | 8 +- util.c | 5 +- 9 files changed, 338 insertions(+), 155 deletions(-) diff --git a/clientserver.c b/clientserver.c index 44395357..b8baa331 100644 --- a/clientserver.c +++ b/clientserver.c @@ -33,6 +33,8 @@ int start_socket_client(char *host, char *path, int argc, char *argv[]) char line[MAXPATHLEN]; char *p, *user=NULL; extern int remote_version; + extern int am_client; + extern int am_sender; p = strchr(host, '@'); if (p) { @@ -44,6 +46,8 @@ int start_socket_client(char *host, char *path, int argc, char *argv[]) if (!user) user = getenv("USER"); if (!user) user = getenv("LOGNAME"); + am_client = 1; + fd = open_socket_out(host, rsync_port); if (fd == -1) { exit_cleanup(1); @@ -92,6 +96,9 @@ int start_socket_client(char *host, char *path, int argc, char *argv[]) } io_printf(fd,"\n"); + if (remote_version > 17 && !am_sender) + io_start_multiplex_in(fd); + return client_run(fd, fd, -1, argc, argv); } @@ -113,6 +120,7 @@ static int rsync_module(int fd, int i) int start_glob=0; char *request=NULL; extern int am_sender; + extern int remote_version; if (!allow_access(addr, host, lp_hosts_allow(i), lp_hosts_deny(i))) { rprintf(FERROR,"rsync denied on module %s from %s (%s)\n", @@ -255,6 +263,9 @@ static int rsync_module(int fd, int i) argp = argv + optind; optind = 0; + if (remote_version > 17 && am_sender) + io_start_multiplex_out(fd); + start_server(fd, fd, argc, argp); return 0; diff --git a/io.c b/io.c index b48ae0f9..a767e2ab 100644 --- a/io.c +++ b/io.c @@ -27,18 +27,22 @@ static int64 total_written; static int64 total_read; +static int io_multiplexing_out; +static int io_multiplexing_in; +static time_t last_io; + extern int verbose; extern int sparse_files; extern int io_timeout; int64 write_total(void) { - return total_written; + return total_written; } int64 read_total(void) { - return total_read; + return total_read; } static int buffer_f_in = -1; @@ -49,6 +53,26 @@ void setup_nonblocking(int f_in,int f_out) buffer_f_in = f_in; } +static void check_timeout(void) +{ + time_t t; + + if (!io_timeout) return; + + if (!last_io) { + last_io = time(NULL); + return; + } + + t = time(NULL); + + if (last_io && io_timeout && (t-last_io)>io_timeout) { + rprintf(FERROR,"read timeout after %d second - exiting\n", + (int)(t-last_io)); + exit_cleanup(1); + } +} + static char *read_buffer; static char *read_buffer_p; @@ -56,6 +80,92 @@ static int read_buffer_len; static int read_buffer_size; +/* continue trying to read len bytes - don't return until len + has been read */ +static void read_loop(int fd, char *buf, int len) +{ + while (len) { + int n = read(fd, buf, len); + if (n > 0) { + buf += n; + len -= n; + } + if (n == 0) { + rprintf(FERROR,"EOF in read_loop\n"); + exit_cleanup(1); + } + if (n == -1) { + fd_set fds; + struct timeval tv; + + if (errno != EAGAIN && errno != EWOULDBLOCK) { + rprintf(FERROR,"io error: %s\n", + strerror(errno)); + exit_cleanup(1); + } + + FD_ZERO(&fds); + FD_SET(fd, &fds); + tv.tv_sec = io_timeout; + tv.tv_usec = 0; + + if (select(fd+1, &fds, NULL, NULL, + io_timeout?&tv:NULL) != 1) { + check_timeout(); + } + } + } +} + +static int read_unbuffered(int fd, char *buf, int len) +{ + static int remaining; + char ibuf[4]; + int tag, ret=0; + char line[1024]; + + if (!io_multiplexing_in) return read(fd, buf, len); + + while (ret == 0) { + if (remaining) { + len = MIN(len, remaining); + read_loop(fd, buf, len); + remaining -= len; + ret = len; + continue; + } + + read_loop(fd, ibuf, 4); + tag = IVAL(ibuf, 0); + remaining = tag & 0xFFFFFF; + tag = tag >> 24; + + if (tag == MPLEX_BASE) continue; + + tag -= MPLEX_BASE; + + if (tag != FERROR && tag != FINFO) { + rprintf(FERROR,"unexpected tag %d\n", tag); + exit_cleanup(1); + } + + if (remaining > sizeof(line)-1) { + rprintf(FERROR,"multiplexing overflow %d\n\n", + remaining); + exit_cleanup(1); + } + + read_loop(fd, line, remaining); + line[remaining] = 0; + + rprintf(tag,"%s", line); + remaining = 0; + } + + return ret; +} + + /* This function was added to overcome a deadlock problem when using * ssh. It looks like we can't allow our receive queue to get full or * ssh will clag up. Uggh. */ @@ -93,35 +203,12 @@ static void read_check(int f) read_buffer_p = read_buffer; } - n = read(f,read_buffer+read_buffer_len,n); + n = read_unbuffered(f,read_buffer+read_buffer_len,n); if (n > 0) { read_buffer_len += n; } } -static time_t last_io; - - -static void check_timeout(void) -{ - time_t t; - - if (!io_timeout) return; - - if (!last_io) { - last_io = time(NULL); - return; - } - - t = time(NULL); - - if (last_io && io_timeout && (t-last_io)>io_timeout) { - rprintf(FERROR,"read timeout after %d second - exiting\n", - (int)(t-last_io)); - exit_cleanup(1); - } -} - static int readfd(int fd,char *buffer,int N) { int ret; @@ -141,7 +228,9 @@ static int readfd(int fd,char *buffer,int N) continue; } - while ((ret = read(fd,buffer + total,N-total)) == -1) { + io_flush(); + + while ((ret = read_unbuffered(fd,buffer + total,N-total)) == -1) { fd_set fds; if (errno != EAGAIN && errno != EWOULDBLOCK) @@ -174,7 +263,7 @@ int32 read_int(int f) char b[4]; if ((ret=readfd(f,b,4)) != 4) { if (verbose > 1) - rprintf(FERROR,"(%d) Error reading %d bytes : %s\n", + rprintf(FERROR,"(%d) read_int: Error reading %d bytes : %s\n", getpid(),4,ret==-1?strerror(errno):"EOF"); exit_cleanup(1); } @@ -198,7 +287,7 @@ int64 read_longint(int f) if (remote_version >= 16) { if ((ret=readfd(f,b,8)) != 8) { if (verbose > 1) - rprintf(FERROR,"(%d) Error reading %d bytes : %s\n", + rprintf(FERROR,"(%d) read_longint: Error reading %d bytes : %s\n", getpid(),8,ret==-1?strerror(errno):"EOF"); exit_cleanup(1); } @@ -215,7 +304,7 @@ void read_buf(int f,char *buf,int len) int ret; if ((ret=readfd(f,buf,len)) != len) { if (verbose > 1) - rprintf(FERROR,"(%d) Error reading %d bytes : %s\n", + rprintf(FERROR,"(%d) read_buf: Error reading %d bytes : %s\n", getpid(),len,ret==-1?strerror(errno):"EOF"); exit_cleanup(1); } @@ -306,103 +395,121 @@ int write_file(int f,char *buf,int len) static int writefd_unbuffered(int fd,char *buf,int len) { - int total = 0; - fd_set w_fds, r_fds; - int fd_count, count, got_select=0; - struct timeval tv; + int total = 0; + fd_set w_fds, r_fds; + int fd_count, count, got_select=0; + struct timeval tv; - if (buffer_f_in == -1) - return write(fd,buf,len); + while (total < len) { + int ret = write(fd,buf+total,len-total); - while (total < len) { - int ret = write(fd,buf+total,len-total); + if (ret == 0) return total; - if (ret == 0) return total; + if (ret == -1 && !(errno == EWOULDBLOCK || errno == EAGAIN)) + return -1; - if (ret == -1 && !(errno == EWOULDBLOCK || errno == EAGAIN)) - return -1; + if (ret == -1 && got_select) { + /* hmmm, we got a write select on the fd and + then failed to write. Why doesn't that + mean that the fd is dead? It doesn't on + some systems it seems (eg. IRIX) */ + u_sleep(1000); + } - if (ret == -1 && got_select) { - /* hmmm, we got a write select on the fd and then failed to write. - Why doesn't that mean that the fd is dead? It doesn't on some - systems it seems (eg. IRIX) */ - u_sleep(1000); -#if 0 - rprintf(FERROR,"write exception\n"); - exit_cleanup(1); -#endif - } - - got_select = 0; - - - if (ret == -1) { - if (read_buffer_len < MAX_READ_BUFFER) - read_check(buffer_f_in); - - fd_count = fd+1; - FD_ZERO(&w_fds); - FD_ZERO(&r_fds); - FD_SET(fd,&w_fds); - if (buffer_f_in != -1) { - FD_SET(buffer_f_in,&r_fds); - if (buffer_f_in > fd) - fd_count = buffer_f_in+1; - } - - tv.tv_sec = BLOCKING_TIMEOUT; - tv.tv_usec = 0; - count = select(fd_count,buffer_f_in == -1? NULL: &r_fds, - &w_fds,NULL,&tv); - - if (count == -1 && errno != EINTR) { - if (verbose > 1) - rprintf(FERROR,"select error: %s\n", strerror(errno)); - exit_cleanup(1); - } - - if (count == 0) { - check_timeout(); - continue; - } - - if (FD_ISSET(fd, &w_fds)) { - got_select = 1; - } - } else { - total += ret; - } - } + got_select = 0; - if (io_timeout) - last_io = time(NULL); - return total; + if (ret != -1) { + total += ret; + continue; + } + + if (read_buffer_len < MAX_READ_BUFFER && buffer_f_in != -1) + read_check(buffer_f_in); + + fd_count = fd+1; + FD_ZERO(&w_fds); + FD_ZERO(&r_fds); + FD_SET(fd,&w_fds); + if (buffer_f_in != -1) { + FD_SET(buffer_f_in,&r_fds); + if (buffer_f_in > fd) + fd_count = buffer_f_in+1; + } + + tv.tv_sec = BLOCKING_TIMEOUT; + tv.tv_usec = 0; + count = select(fd_count,buffer_f_in == -1? NULL: &r_fds, + &w_fds,NULL,&tv); + + if (count == -1 && errno != EINTR) { + if (verbose > 1) + rprintf(FERROR,"select error: %s\n", strerror(errno)); + exit_cleanup(1); + } + + if (count == 0) { + check_timeout(); + continue; + } + + if (FD_ISSET(fd, &w_fds)) { + got_select = 1; + } + } + + if (io_timeout) + last_io = time(NULL); + + return total; } + static char *io_buffer; static int io_buffer_count; +static int io_out_fd; void io_start_buffering(int fd) { - io_buffer = (char *)malloc(IO_BUFFER_SIZE); + if (io_buffer) return; + io_out_fd = fd; + io_buffer = (char *)malloc(IO_BUFFER_SIZE+4); if (!io_buffer) out_of_memory("writefd"); io_buffer_count = 0; + + /* leave room for the multiplex header in case it's needed */ + io_buffer += 4; } -void io_end_buffering(int fd) +void io_flush(void) { - if (io_buffer_count) { - if (writefd_unbuffered(fd, io_buffer, - io_buffer_count) != + int fd = io_out_fd; + if (!io_buffer_count) return; + + if (io_multiplexing_out) { + SIVAL(io_buffer-4, 0, (MPLEX_BASE<<24) + io_buffer_count); + if (writefd_unbuffered(fd, io_buffer-4, io_buffer_count+4) != + io_buffer_count+4) { + rprintf(FERROR,"write failed\n"); + exit_cleanup(1); + } + } else { + if (writefd_unbuffered(fd, io_buffer, io_buffer_count) != io_buffer_count) { rprintf(FERROR,"write failed\n"); exit_cleanup(1); } - io_buffer_count = 0; } - free(io_buffer); - io_buffer = NULL; + io_buffer_count = 0; +} + +void io_end_buffering(int fd) +{ + io_flush(); + if (!io_multiplexing_out) { + free(io_buffer-4); + io_buffer = NULL; + } } static int writefd(int fd,char *buf,int len1) @@ -420,14 +527,7 @@ static int writefd(int fd,char *buf,int len1) io_buffer_count += n; } - if (io_buffer_count == IO_BUFFER_SIZE) { - if (writefd_unbuffered(fd, io_buffer, - io_buffer_count) != - io_buffer_count) { - return -1; - } - io_buffer_count = 0; - } + if (io_buffer_count == IO_BUFFER_SIZE) io_flush(); } return len1; @@ -436,15 +536,15 @@ static int writefd(int fd,char *buf,int len1) void write_int(int f,int32 x) { - int ret; - char b[4]; - SIVAL(b,0,x); - if ((ret=writefd(f,b,4)) != 4) { - rprintf(FERROR,"write_int failed : %s\n", - ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_written += 4; + int ret; + char b[4]; + SIVAL(b,0,x); + if ((ret=writefd(f,b,4)) != 4) { + rprintf(FERROR,"write_int failed : %s\n", + ret==-1?strerror(errno):"EOF"); + exit_cleanup(1); + } + total_written += 4; } void write_longint(int f, int64 x) @@ -472,13 +572,13 @@ void write_longint(int f, int64 x) void write_buf(int f,char *buf,int len) { - int ret; - if ((ret=writefd(f,buf,len)) != len) { - rprintf(FERROR,"write_buf failed : %s\n", - ret==-1?strerror(errno):"EOF"); - exit_cleanup(1); - } - total_written += len; + int ret; + if ((ret=writefd(f,buf,len)) != len) { + rprintf(FERROR,"write_buf failed : %s\n", + ret==-1?strerror(errno):"EOF"); + exit_cleanup(1); + } + total_written += len; } /* write a string to the connection */ @@ -533,3 +633,41 @@ void io_printf(int fd, const char *format, ...) write_sbuf(fd, buf); } + + +/* setup for multiplexing an error stream with the data stream */ +void io_start_multiplex_out(int fd) +{ + io_start_buffering(fd); + io_multiplexing_out = 1; +} + +/* setup for multiplexing an error stream with the data stream */ +void io_start_multiplex_in(int fd) +{ + if (read_buffer_len) { + fprintf(stderr,"ERROR: data in read buffer at mplx start\n"); + exit_cleanup(1); + } + + io_multiplexing_in = 1; +} + +/* write an message to the error stream */ +int io_multiplex_write(int f, char *buf, int len) +{ + if (!io_multiplexing_out) return 0; + + io_flush(); + + SIVAL(io_buffer-4, 0, ((MPLEX_BASE + f)<<24) + len); + memcpy(io_buffer, buf, len); + + writefd_unbuffered(io_out_fd, io_buffer-4, len+4); + return 1; +} + +void io_close_input(int fd) +{ + buffer_f_in = -1; +} diff --git a/log.c b/log.c index 90b287e1..71eb234d 100644 --- a/log.c +++ b/log.c @@ -56,6 +56,12 @@ void rprintf(int fd, const char *format, ...) int len; FILE *f=NULL; extern int am_daemon; + /* recursion can happen with certain fatal conditions */ + static int depth; + + if (depth) return; + + depth++; va_start(ap, format); len = vslprintf(buf, sizeof(buf)-1, format, ap); @@ -72,7 +78,11 @@ void rprintf(int fd, const char *format, ...) if (fd == FERROR) priority = LOG_WARNING; log_open(); - syslog(priority, "%s", buf); + if (!io_multiplex_write(fd, buf, strlen(buf))) { + syslog(priority, "%s", buf); + } + + depth--; return; } @@ -91,6 +101,8 @@ void rprintf(int fd, const char *format, ...) if (!f) exit_cleanup(1); if (fwrite(buf, len, 1, f) != 1) exit_cleanup(1); + + depth--; } void rflush(int fd) diff --git a/main.c b/main.c index dbe6aab6..5a5d5153 100644 --- a/main.c +++ b/main.c @@ -35,9 +35,9 @@ static void report(int f) extern int am_daemon; if (am_daemon) { - rprintf(FINFO, "wrote %.0f bytes read %.0f bytes total size %.0f\n", - (double)write_total(),(double)read_total(), - (double)total_size); + syslog(LOG_INFO,"wrote %.0f bytes read %.0f bytes total size %.0f\n", + (double)write_total(),(double)read_total(), + (double)total_size); if (f == -1 || !am_sender) return; } @@ -218,8 +218,13 @@ static void do_server_sender(int f_in, int f_out, int argc,char *argv[]) } flist = send_file_list(f_out,argc,argv); + if (!flist || flist->count == 0) { + exit_cleanup(0); + } + send_files(flist,f_out,f_in); report(f_out); + io_flush(); exit_cleanup(0); } @@ -230,29 +235,40 @@ static int do_recv(int f_in,int f_out,struct file_list *flist,char *local_name) int status=0; int recv_pipe[2]; extern int preserve_hard_links; - extern int am_daemon; + extern int am_server; if (preserve_hard_links) init_hard_links(flist); if (pipe(recv_pipe) < 0) { rprintf(FERROR,"pipe failed in do_recv\n"); - exit(1); + exit_cleanup(1); } + io_flush(); if ((pid=do_fork()) == 0) { - close(recv_pipe[0]); - if (f_in != f_out) close(f_out); - recv_files(f_in,flist,local_name,recv_pipe[1]); - if (am_daemon) report(-1); - exit_cleanup(0); + close(recv_pipe[1]); + io_close_input(f_in); + if (f_in != f_out) close(f_in); + generate_files(f_out,flist,local_name,recv_pipe[0]); + + io_flush(); + _exit(0); } - close(recv_pipe[1]); - if (f_in != f_out) close(f_in); - generate_files(f_out,flist,local_name,recv_pipe[0]); + close(recv_pipe[0]); + if (f_in != f_out) close(f_out); + + recv_files(f_in,flist,local_name,recv_pipe[1]); + if (!am_server) + report(f_in); + + if (verbose > 3) + rprintf(FINFO,"do_recv waiting on %d\n",pid); + + io_flush(); waitpid(pid, &status, 0); return status; @@ -287,7 +303,7 @@ static void do_server_recv(int f_in, int f_out, int argc,char *argv[]) flist = recv_file_list(f_in); if (!flist || flist->count == 0) { - rprintf(FERROR,"nothing to do\n"); + rprintf(FERROR,"server_recv: nothing to do\n"); exit_cleanup(1); } @@ -344,7 +360,8 @@ int client_run(int f_in, int f_out, int pid, int argc, char *argv[]) send_files(flist,f_out,f_in); if (pid != -1) { if (verbose > 3) - rprintf(FINFO,"waiting on %d\n",pid); + rprintf(FINFO,"client_run waiting on %d\n",pid); + io_flush(); waitpid(pid, &status, 0); } report(-1); @@ -355,7 +372,7 @@ int client_run(int f_in, int f_out, int pid, int argc, char *argv[]) flist = recv_file_list(f_in); if (!flist || flist->count == 0) { - rprintf(FINFO,"nothing to do\n"); + rprintf(FINFO,"client: nothing to do\n"); exit_cleanup(0); } @@ -363,9 +380,10 @@ int client_run(int f_in, int f_out, int pid, int argc, char *argv[]) status2 = do_recv(f_in,f_out,flist,local_name); - report(f_in); - if (pid != -1) { + if (verbose > 3) + rprintf(FINFO,"client_run2 waiting on %d\n",pid); + io_flush(); waitpid(pid, &status, 0); } @@ -483,7 +501,7 @@ int main(int argc,char *argv[]) if (argc < 2) { usage(FERROR); - exit(1); + exit_cleanup(1); } /* we set a 0 umask so that correct file permissions can be @@ -507,7 +525,7 @@ int main(int argc,char *argv[]) if (argc < 1) { usage(FERROR); - exit(1); + exit_cleanup(1); } if (dry_run) diff --git a/options.c b/options.c index 3e2b7147..11dbcbea 100644 --- a/options.c +++ b/options.c @@ -54,6 +54,7 @@ int am_server = 0; int am_sender=0; int recurse = 0; int am_daemon=0; +int am_client=0; int block_size=BLOCK_SIZE; diff --git a/rsync.c b/rsync.c index a9c24387..82efef54 100644 --- a/rsync.c +++ b/rsync.c @@ -723,6 +723,7 @@ static char *cleanup_fname; void exit_cleanup(int code) { + io_flush(); if (cleanup_fname) do_unlink(cleanup_fname); signal(SIGUSR1, SIG_IGN); diff --git a/rsync.h b/rsync.h index 1a691586..931f6f03 100644 --- a/rsync.h +++ b/rsync.h @@ -41,7 +41,7 @@ #define SAME_TIME (1<<7) /* update this if you make incompatible changes */ -#define PROTOCOL_VERSION 17 +#define PROTOCOL_VERSION 18 #define MIN_PROTOCOL_VERSION 11 #define MAX_PROTOCOL_VERSION 30 @@ -58,6 +58,7 @@ #define BLOCKING_TIMEOUT 10 +#define MPLEX_BASE 7 #define FERROR 1 #define FINFO 2 diff --git a/socket.c b/socket.c index 3c876ed0..910c2dc0 100644 --- a/socket.c +++ b/socket.c @@ -125,12 +125,12 @@ void start_accept_loop(int port, int (*fn)(int )) /* open an incoming socket */ s = open_socket_in(SOCK_STREAM, port); if (s == -1) - exit(1); + exit_cleanup(1); /* ready to listen */ if (listen(s, 5) == -1) { close(s); - exit(1); + exit_cleanup(1); } @@ -308,7 +308,7 @@ char *client_addr(int fd) static char addr_buf[100]; if (getpeername(fd, &sa, &length)) { - exit(1); + exit_cleanup(1); } strlcpy(addr_buf,(char *)inet_ntoa(sockin->sin_addr), sizeof(addr_buf)-1); @@ -331,7 +331,7 @@ char *client_name(int fd) strcpy(name_buf,"UNKNOWN"); if (getpeername(fd, &sa, &length)) { - exit(1); + exit_cleanup(1); } /* Look up the remote host name. */ diff --git a/util.c b/util.c index 04420048..5b16fc4b 100644 --- a/util.c +++ b/util.c @@ -633,7 +633,7 @@ int vslprintf(char *str, int n, const char *format, va_list ap) buf = malloc(len); if (!buf) { /* can't call debug or we would recurse */ - exit(1); + exit_cleanup(1); } } @@ -641,7 +641,7 @@ int vslprintf(char *str, int n, const char *format, va_list ap) ret = strlen(buf); if (ret > n) { /* yikes! */ - exit(1); + exit_cleanup(1); } buf[ret] = 0; @@ -663,3 +663,4 @@ int slprintf(char *str, int n, char *format, ...) va_end(ap); return ret; } + -- 2.34.1