Allow the server to terminate the module list by just closing the
[rsync.git] / io.c
1 /* -*- c-file-style: "linux" -*-
2    
3    Copyright (C) 1996-2001 by Andrew Tridgell 
4    Copyright (C) Paul Mackerras 1996
5    Copyright (C) 2001 by Martin Pool <mbp@samba.org>
6    
7    This program is free software; you can redistribute it and/or modify
8    it under the terms of the GNU General Public License as published by
9    the Free Software Foundation; either version 2 of the License, or
10    (at your option) any later version.
11    
12    This program is distributed in the hope that it will be useful,
13    but WITHOUT ANY WARRANTY; without even the implied warranty of
14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15    GNU General Public License for more details.
16    
17    You should have received a copy of the GNU General Public License
18    along with this program; if not, write to the Free Software
19    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
20 */
21
22 /*
23   socket and pipe IO utilities used in rsync 
24
25   tridge, June 1996
26   */
27 #include "rsync.h"
28
29 /* if no timeout is specified then use a 60 second select timeout */
30 #define SELECT_TIMEOUT 60
31
32 static int io_multiplexing_out;
33 static int io_multiplexing_in;
34 static int multiplex_in_fd;
35 static int multiplex_out_fd;
36 static time_t last_io;
37 static int no_flush;
38
39 extern int bwlimit;
40 extern int verbose;
41 extern int io_timeout;
42 extern struct stats stats;
43
44
45 /** Ignore EOF errors while reading a module listing if the remote
46     version is 24 or less. */
47 int kludge_around_eof = False;
48
49
50 static int io_error_fd = -1;
51
52 static void read_loop(int fd, char *buf, int len);
53
54 static void check_timeout(void)
55 {
56         extern int am_server, am_daemon;
57         time_t t;
58
59         err_list_push();
60         
61         if (!io_timeout) return;
62
63         if (!last_io) {
64                 last_io = time(NULL);
65                 return;
66         }
67
68         t = time(NULL);
69
70         if (last_io && io_timeout && (t-last_io) >= io_timeout) {
71                 if (!am_server && !am_daemon) {
72                         rprintf(FERROR,"io timeout after %d seconds - exiting\n", 
73                                 (int)(t-last_io));
74                 }
75                 exit_cleanup(RERR_TIMEOUT);
76         }
77 }
78
79 /* setup the fd used to propogate errors */
80 void io_set_error_fd(int fd)
81 {
82         io_error_fd = fd;
83 }
84
85 /* read some data from the error fd and write it to the write log code */
86 static void read_error_fd(void)
87 {
88         char buf[200];
89         int n;
90         int fd = io_error_fd;
91         int tag, len;
92
93         /* io_error_fd is temporarily disabled -- is this meant to
94          * prevent indefinite recursion? */
95         io_error_fd = -1;
96
97         read_loop(fd, buf, 4);
98         tag = IVAL(buf, 0);
99
100         len = tag & 0xFFFFFF;
101         tag = tag >> 24;
102         tag -= MPLEX_BASE;
103
104         while (len) {
105                 n = len;
106                 if (n > (sizeof(buf)-1)) n = sizeof(buf)-1;
107                 read_loop(fd, buf, n);
108                 rwrite((enum logcode)tag, buf, n);
109                 len -= n;
110         }
111
112         io_error_fd = fd;
113 }
114
115
116 static void whine_about_eof (void)
117 {
118         /**
119            It's almost always an error to get an EOF when we're trying
120            to read from the network, because the protocol is
121            self-terminating.
122            
123            However, there is one unfortunate cases where it is not,
124            which is rsync <2.4.6 sending a list of modules on a
125            server, since the list is terminated by closing the socket.
126            So, for the section of the program where that is a problem
127            (start_socket_client), kludge_around_eof is True and we
128            just exit.
129         */
130
131         if (kludge_around_eof)
132                 exit_cleanup (0);
133         else {
134                 rprintf (FERROR,
135                          "%s: connection unexpectedly closed "
136                          "(%.0f bytes read so far)\n",
137                          RSYNC_NAME, (double)stats.total_read);
138         
139                 exit_cleanup (RERR_STREAMIO);
140         }
141 }
142
143
144 static void die_from_readerr (int err)
145 {
146         /* this prevents us trying to write errors on a dead socket */
147         io_multiplexing_close();
148                                 
149         rprintf(FERROR, "%s: read error: %s\n",
150                 RSYNC_NAME, strerror (err));
151         exit_cleanup(RERR_STREAMIO);
152 }
153
154
155 /*!
156  * Read from a socket with IO timeout. return the number of bytes
157  * read. If no bytes can be read then exit, never return a number <= 0.
158  *
159  * TODO: If the remote shell connection fails, then current versions
160  * actually report an "unexpected EOF" error here.  Since it's a
161  * fairly common mistake to try to use rsh when ssh is required, we
162  * should trap that: if we fail to read any data at all, we should
163  * give a better explanation.  We can tell whether the connection has
164  * started by looking e.g. at whether the remote version is known yet.
165  */
166 static int read_timeout (int fd, char *buf, int len)
167 {
168         int n, ret=0;
169
170         io_flush();
171
172         while (ret == 0) {
173                 /* until we manage to read *something* */
174                 fd_set fds;
175                 struct timeval tv;
176                 int fd_count = fd+1;
177
178                 FD_ZERO(&fds);
179                 FD_SET(fd, &fds);
180                 if (io_error_fd != -1) {
181                         FD_SET(io_error_fd, &fds);
182                         if (io_error_fd > fd) fd_count = io_error_fd+1;
183                 }
184
185                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
186                 tv.tv_usec = 0;
187
188                 errno = 0;
189
190                 if (select(fd_count, &fds, NULL, NULL, &tv) < 1) {
191                         if (errno == EBADF) {
192                                 exit_cleanup(RERR_SOCKETIO);
193                         }
194                         check_timeout();
195                         continue;
196                 }
197
198                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &fds)) {
199                         read_error_fd();
200                 }
201
202                 if (!FD_ISSET(fd, &fds)) continue;
203
204                 n = read(fd, buf, len);
205
206                 if (n > 0) {
207                         buf += n;
208                         len -= n;
209                         ret += n;
210                         if (io_timeout)
211                                 last_io = time(NULL);
212                         continue;
213                 } else if (n == 0) {
214                         whine_about_eof ();
215                         return -1; /* doesn't return */
216                 } else if (n == -1) {
217                         if (errno == EINTR || errno == EWOULDBLOCK ||
218                             errno == EAGAIN) 
219                                 continue;
220                         else
221                                 die_from_readerr (errno);
222                 }
223         }
224
225         return ret;
226 }
227
228
229
230
231 /*! Continue trying to read len bytes - don't return until len has
232   been read.   */
233 static void read_loop (int fd, char *buf, int len)
234 {
235         while (len) {
236                 int n = read_timeout(fd, buf, len);
237
238                 buf += n;
239                 len -= n;
240         }
241 }
242
243
244 /**
245  * Read from the file descriptor handling multiplexing - return number
246  * of bytes read.
247  * 
248  * Never returns <= 0. 
249  */
250 static int read_unbuffered(int fd, char *buf, int len)
251 {
252         static int remaining;
253         int tag, ret=0;
254         char line[1024];
255
256         if (!io_multiplexing_in || fd != multiplex_in_fd)
257                 return read_timeout(fd, buf, len);
258
259         while (ret == 0) {
260                 if (remaining) {
261                         len = MIN(len, remaining);
262                         read_loop(fd, buf, len);
263                         remaining -= len;
264                         ret = len;
265                         continue;
266                 }
267
268                 read_loop (fd, line, 4);
269                 tag = IVAL(line, 0);
270
271                 remaining = tag & 0xFFFFFF;
272                 tag = tag >> 24;
273
274                 if (tag == MPLEX_BASE) continue;
275
276                 tag -= MPLEX_BASE;
277
278                 if (tag != FERROR && tag != FINFO) {
279                         rprintf(FERROR,"unexpected tag %d\n", tag);
280                         exit_cleanup(RERR_STREAMIO);
281                 }
282
283                 if (remaining > sizeof(line)-1) {
284                         rprintf(FERROR,"multiplexing overflow %d\n\n", 
285                                 remaining);
286                         exit_cleanup(RERR_STREAMIO);
287                 }
288
289                 read_loop(fd, line, remaining);
290                 line[remaining] = 0;
291
292                 rprintf((enum logcode)tag,"%s", line);
293                 remaining = 0;
294         }
295
296         return ret;
297 }
298
299
300 /* do a buffered read from fd. don't return until all N bytes
301    have been read. If all N can't be read then exit with an error */
302 static void readfd (int fd, char *buffer, int N)
303 {
304         int  ret;
305         int total=0;  
306         
307         while (total < N) {
308                 io_flush();
309
310                 ret = read_unbuffered (fd, buffer + total, N-total);
311                 total += ret;
312         }
313
314         stats.total_read += total;
315 }
316
317
318 int32 read_int(int f)
319 {
320         char b[4];
321         int32 ret;
322
323         readfd(f,b,4);
324         ret = IVAL(b,0);
325         if (ret == (int32)0xffffffff) return -1;
326         return ret;
327 }
328
329 int64 read_longint(int f)
330 {
331         extern int remote_version;
332         int64 ret;
333         char b[8];
334         ret = read_int(f);
335
336         if ((int32)ret != (int32)0xffffffff) {
337                 return ret;
338         }
339
340 #ifdef NO_INT64
341         rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
342         exit_cleanup(RERR_UNSUPPORTED);
343 #else
344         if (remote_version >= 16) {
345                 readfd(f,b,8);
346                 ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
347         }
348 #endif
349
350         return ret;
351 }
352
353 void read_buf(int f,char *buf,int len)
354 {
355         readfd(f,buf,len);
356 }
357
358 void read_sbuf(int f,char *buf,int len)
359 {
360         read_buf (f,buf,len);
361         buf[len] = 0;
362 }
363
364 unsigned char read_byte(int f)
365 {
366         unsigned char c;
367         read_buf (f, (char *)&c, 1);
368         return c;
369 }
370
371 /* write len bytes to fd */
372 static void writefd_unbuffered(int fd,char *buf,int len)
373 {
374         int total = 0;
375         fd_set w_fds, r_fds;
376         int fd_count, count;
377         struct timeval tv;
378
379         err_list_push();
380
381         no_flush++;
382
383         while (total < len) {
384                 FD_ZERO(&w_fds);
385                 FD_ZERO(&r_fds);
386                 FD_SET(fd,&w_fds);
387                 fd_count = fd;
388
389                 if (io_error_fd != -1) {
390                         FD_SET(io_error_fd,&r_fds);
391                         if (io_error_fd > fd_count) 
392                                 fd_count = io_error_fd;
393                 }
394
395                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
396                 tv.tv_usec = 0;
397
398                 errno = 0;
399
400                 count = select(fd_count+1,
401                                io_error_fd != -1?&r_fds:NULL,
402                                &w_fds,NULL,
403                                &tv);
404
405                 if (count <= 0) {
406                         if (errno == EBADF) {
407                                 exit_cleanup(RERR_SOCKETIO);
408                         }
409                         check_timeout();
410                         continue;
411                 }
412
413                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &r_fds)) {
414                         read_error_fd();
415                 }
416
417                 if (FD_ISSET(fd, &w_fds)) {
418                         int ret, n = len-total;
419                         ret = write(fd,buf+total,n);
420
421                         if (ret == -1 && errno == EINTR) {
422                                 continue;
423                         }
424
425                         if (ret == -1 && 
426                             (errno == EWOULDBLOCK || errno == EAGAIN)) {
427                                 msleep(1);
428                                 continue;
429                         }
430
431                         if (ret <= 0) {
432                                 rprintf(FERROR,
433                                         "error writing %d unbuffered bytes"
434                                         " - exiting: %s\n", len,
435                                         strerror(errno));
436                                 exit_cleanup(RERR_STREAMIO);
437                         }
438
439                         /* Sleep after writing to limit I/O bandwidth */
440                         if (bwlimit)
441                         {
442                             tv.tv_sec = 0;
443                             tv.tv_usec = ret * 1000 / bwlimit;
444                             while (tv.tv_usec > 1000000)
445                             {
446                                 tv.tv_sec++;
447                                 tv.tv_usec -= 1000000;
448                             }
449                             select(0, NULL, NULL, NULL, &tv);
450                         }
451  
452                         total += ret;
453
454                         if (io_timeout)
455                                 last_io = time(NULL);
456                 }
457         }
458
459         no_flush--;
460 }
461
462
463 static char *io_buffer;
464 static int io_buffer_count;
465
466 void io_start_buffering(int fd)
467 {
468         if (io_buffer) return;
469         multiplex_out_fd = fd;
470         io_buffer = (char *)malloc(IO_BUFFER_SIZE);
471         if (!io_buffer) out_of_memory("writefd");
472         io_buffer_count = 0;
473 }
474
475 /* write an message to a multiplexed stream. If this fails then rsync
476    exits */
477 static void mplex_write(int fd, enum logcode code, char *buf, int len)
478 {
479         char buffer[4096];
480         int n = len;
481
482         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
483
484         if (n > (sizeof(buffer)-4)) {
485                 n = sizeof(buffer)-4;
486         }
487
488         memcpy(&buffer[4], buf, n);
489         writefd_unbuffered(fd, buffer, n+4);
490
491         len -= n;
492         buf += n;
493
494         if (len) {
495                 writefd_unbuffered(fd, buf, len);
496         }
497 }
498
499
500 void io_flush(void)
501 {
502         int fd = multiplex_out_fd;
503
504         err_list_push();
505
506         if (!io_buffer_count || no_flush) return;
507
508         if (io_multiplexing_out) {
509                 mplex_write(fd, FNONE, io_buffer, io_buffer_count);
510         } else {
511                 writefd_unbuffered(fd, io_buffer, io_buffer_count);
512         }
513         io_buffer_count = 0;
514 }
515
516
517 /* XXX: fd is ignored, which seems a little strange. */
518 void io_end_buffering(int fd)
519 {
520         io_flush();
521         if (!io_multiplexing_out) {
522                 free(io_buffer);
523                 io_buffer = NULL;
524         }
525 }
526
527 static void writefd(int fd,char *buf,int len)
528 {
529         stats.total_written += len;
530
531         err_list_push();
532
533         if (!io_buffer || fd != multiplex_out_fd) {
534                 writefd_unbuffered(fd, buf, len);
535                 return;
536         }
537
538         while (len) {
539                 int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count);
540                 if (n > 0) {
541                         memcpy(io_buffer+io_buffer_count, buf, n);
542                         buf += n;
543                         len -= n;
544                         io_buffer_count += n;
545                 }
546                 
547                 if (io_buffer_count == IO_BUFFER_SIZE) io_flush();
548         }
549 }
550
551
552 void write_int(int f,int32 x)
553 {
554         char b[4];
555         SIVAL(b,0,x);
556         writefd(f,b,4);
557 }
558
559
560 /*
561  * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
562  * 64-bit types on this platform.
563  */
564 void write_longint(int f, int64 x)
565 {
566         extern int remote_version;
567         char b[8];
568
569         if (remote_version < 16 || x <= 0x7FFFFFFF) {
570                 write_int(f, (int)x);
571                 return;
572         }
573
574         write_int(f, (int32)0xFFFFFFFF);
575         SIVAL(b,0,(x&0xFFFFFFFF));
576         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
577
578         writefd(f,b,8);
579 }
580
581 void write_buf(int f,char *buf,int len)
582 {
583         writefd(f,buf,len);
584 }
585
586 /* write a string to the connection */
587 static void write_sbuf(int f,char *buf)
588 {
589         write_buf(f, buf, strlen(buf));
590 }
591
592
593 void write_byte(int f,unsigned char c)
594 {
595         write_buf(f,(char *)&c,1);
596 }
597
598
599
600 int read_line(int f, char *buf, int maxlen)
601 {
602         while (maxlen) {
603                 buf[0] = 0;
604                 read_buf(f, buf, 1);
605                 if (buf[0] == 0) return 0;
606                 if (buf[0] == '\n') {
607                         buf[0] = 0;
608                         break;
609                 }
610                 if (buf[0] != '\r') {
611                         buf++;
612                         maxlen--;
613                 }
614         }
615         if (maxlen == 0) {
616                 *buf = 0;
617                 return 0;
618         }
619
620         return 1;
621 }
622
623
624 void io_printf(int fd, const char *format, ...)
625 {
626         va_list ap;  
627         char buf[1024];
628         int len;
629         
630         va_start(ap, format);
631         len = vsnprintf(buf, sizeof(buf), format, ap);
632         va_end(ap);
633
634         if (len < 0) exit_cleanup(RERR_STREAMIO);
635
636         write_sbuf(fd, buf);
637 }
638
639
640 /* setup for multiplexing an error stream with the data stream */
641 void io_start_multiplex_out(int fd)
642 {
643         multiplex_out_fd = fd;
644         io_flush();
645         io_start_buffering(fd);
646         io_multiplexing_out = 1;
647 }
648
649 /* setup for multiplexing an error stream with the data stream */
650 void io_start_multiplex_in(int fd)
651 {
652         multiplex_in_fd = fd;
653         io_flush();
654         io_multiplexing_in = 1;
655 }
656
657 /* write an message to the multiplexed error stream */
658 int io_multiplex_write(enum logcode code, char *buf, int len)
659 {
660         if (!io_multiplexing_out) return 0;
661
662         io_flush();
663         stats.total_written += (len+4);
664         mplex_write(multiplex_out_fd, code, buf, len);
665         return 1;
666 }
667
668 /* stop output multiplexing */
669 void io_multiplexing_close(void)
670 {
671         io_multiplexing_out = 0;
672 }
673