More error messages.
[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                 int count;
178
179                 FD_ZERO(&fds);
180                 FD_SET(fd, &fds);
181                 if (io_error_fd != -1) {
182                         FD_SET(io_error_fd, &fds);
183                         if (io_error_fd > fd) fd_count = io_error_fd+1;
184                 }
185
186                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
187                 tv.tv_usec = 0;
188
189                 errno = 0;
190
191                 count = select(fd_count, &fds, NULL, NULL, &tv);
192
193                 if (count == 0) {
194                         check_timeout();
195                 }
196
197                 if (count <= 0) {
198                         if (errno == EBADF) {
199                                 exit_cleanup(RERR_SOCKETIO);
200                         }
201                         continue;
202                 }
203
204                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &fds)) {
205                         read_error_fd();
206                 }
207
208                 if (!FD_ISSET(fd, &fds)) continue;
209
210                 n = read(fd, buf, len);
211
212                 if (n > 0) {
213                         buf += n;
214                         len -= n;
215                         ret += n;
216                         if (io_timeout)
217                                 last_io = time(NULL);
218                         continue;
219                 } else if (n == 0) {
220                         whine_about_eof ();
221                         return -1; /* doesn't return */
222                 } else if (n == -1) {
223                         if (errno == EINTR || errno == EWOULDBLOCK ||
224                             errno == EAGAIN) 
225                                 continue;
226                         else
227                                 die_from_readerr (errno);
228                 }
229         }
230
231         return ret;
232 }
233
234
235
236
237 /*! Continue trying to read len bytes - don't return until len has
238   been read.   */
239 static void read_loop (int fd, char *buf, int len)
240 {
241         while (len) {
242                 int n = read_timeout(fd, buf, len);
243
244                 buf += n;
245                 len -= n;
246         }
247 }
248
249
250 /**
251  * Read from the file descriptor handling multiplexing - return number
252  * of bytes read.
253  * 
254  * Never returns <= 0. 
255  */
256 static int read_unbuffered(int fd, char *buf, int len)
257 {
258         static int remaining;
259         int tag, ret=0;
260         char line[1024];
261
262         if (!io_multiplexing_in || fd != multiplex_in_fd)
263                 return read_timeout(fd, buf, len);
264
265         while (ret == 0) {
266                 if (remaining) {
267                         len = MIN(len, remaining);
268                         read_loop(fd, buf, len);
269                         remaining -= len;
270                         ret = len;
271                         continue;
272                 }
273
274                 read_loop (fd, line, 4);
275                 tag = IVAL(line, 0);
276
277                 remaining = tag & 0xFFFFFF;
278                 tag = tag >> 24;
279
280                 if (tag == MPLEX_BASE) continue;
281
282                 tag -= MPLEX_BASE;
283
284                 if (tag != FERROR && tag != FINFO) {
285                         rprintf(FERROR,"unexpected tag %d\n", tag);
286                         exit_cleanup(RERR_STREAMIO);
287                 }
288
289                 if (remaining > sizeof(line)-1) {
290                         rprintf(FERROR,"multiplexing overflow %d\n\n", 
291                                 remaining);
292                         exit_cleanup(RERR_STREAMIO);
293                 }
294
295                 read_loop(fd, line, remaining);
296                 line[remaining] = 0;
297
298                 rprintf((enum logcode)tag,"%s", line);
299                 remaining = 0;
300         }
301
302         return ret;
303 }
304
305
306 /* do a buffered read from fd. don't return until all N bytes
307    have been read. If all N can't be read then exit with an error */
308 static void readfd (int fd, char *buffer, int N)
309 {
310         int  ret;
311         int total=0;  
312         
313         while (total < N) {
314                 io_flush();
315
316                 ret = read_unbuffered (fd, buffer + total, N-total);
317                 total += ret;
318         }
319
320         stats.total_read += total;
321 }
322
323
324 int32 read_int(int f)
325 {
326         char b[4];
327         int32 ret;
328
329         readfd(f,b,4);
330         ret = IVAL(b,0);
331         if (ret == (int32)0xffffffff) return -1;
332         return ret;
333 }
334
335 int64 read_longint(int f)
336 {
337         extern int remote_version;
338         int64 ret;
339         char b[8];
340         ret = read_int(f);
341
342         if ((int32)ret != (int32)0xffffffff) {
343                 return ret;
344         }
345
346 #ifdef NO_INT64
347         rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
348         exit_cleanup(RERR_UNSUPPORTED);
349 #else
350         if (remote_version >= 16) {
351                 readfd(f,b,8);
352                 ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
353         }
354 #endif
355
356         return ret;
357 }
358
359 void read_buf(int f,char *buf,int len)
360 {
361         readfd(f,buf,len);
362 }
363
364 void read_sbuf(int f,char *buf,int len)
365 {
366         read_buf (f,buf,len);
367         buf[len] = 0;
368 }
369
370 unsigned char read_byte(int f)
371 {
372         unsigned char c;
373         read_buf (f, (char *)&c, 1);
374         return c;
375 }
376
377 /* write len bytes to fd */
378 static void writefd_unbuffered(int fd,char *buf,int len)
379 {
380         int total = 0;
381         fd_set w_fds, r_fds;
382         int fd_count, count;
383         struct timeval tv;
384
385         err_list_push();
386
387         no_flush++;
388
389         while (total < len) {
390                 FD_ZERO(&w_fds);
391                 FD_ZERO(&r_fds);
392                 FD_SET(fd,&w_fds);
393                 fd_count = fd;
394
395                 if (io_error_fd != -1) {
396                         FD_SET(io_error_fd,&r_fds);
397                         if (io_error_fd > fd_count) 
398                                 fd_count = io_error_fd;
399                 }
400
401                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
402                 tv.tv_usec = 0;
403
404                 errno = 0;
405
406                 count = select(fd_count+1,
407                                io_error_fd != -1?&r_fds:NULL,
408                                &w_fds,NULL,
409                                &tv);
410
411                 if (count == 0) {
412                         check_timeout();
413                 }
414
415                 if (count <= 0) {
416                         if (errno == EBADF) {
417                                 exit_cleanup(RERR_SOCKETIO);
418                         }
419                         continue;
420                 }
421
422                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &r_fds)) {
423                         read_error_fd();
424                 }
425
426                 if (FD_ISSET(fd, &w_fds)) {
427                         int ret, n = len-total;
428                         ret = write(fd,buf+total,n);
429
430                         if (ret == -1 && errno == EINTR) {
431                                 continue;
432                         }
433
434                         if (ret == -1 && 
435                             (errno == EWOULDBLOCK || errno == EAGAIN)) {
436                                 msleep(1);
437                                 continue;
438                         }
439
440                         if (ret <= 0) {
441                                 rprintf(FERROR,
442                                         "error writing %d unbuffered bytes"
443                                         " - exiting: %s\n", len,
444                                         strerror(errno));
445                                 exit_cleanup(RERR_STREAMIO);
446                         }
447
448                         /* Sleep after writing to limit I/O bandwidth */
449                         if (bwlimit)
450                         {
451                             tv.tv_sec = 0;
452                             tv.tv_usec = ret * 1000 / bwlimit;
453                             while (tv.tv_usec > 1000000)
454                             {
455                                 tv.tv_sec++;
456                                 tv.tv_usec -= 1000000;
457                             }
458                             select(0, NULL, NULL, NULL, &tv);
459                         }
460  
461                         total += ret;
462
463                         if (io_timeout)
464                                 last_io = time(NULL);
465                 }
466         }
467
468         no_flush--;
469 }
470
471
472 static char *io_buffer;
473 static int io_buffer_count;
474
475 void io_start_buffering(int fd)
476 {
477         if (io_buffer) return;
478         multiplex_out_fd = fd;
479         io_buffer = (char *)malloc(IO_BUFFER_SIZE);
480         if (!io_buffer) out_of_memory("writefd");
481         io_buffer_count = 0;
482 }
483
484 /* write an message to a multiplexed stream. If this fails then rsync
485    exits */
486 static void mplex_write(int fd, enum logcode code, char *buf, int len)
487 {
488         char buffer[4096];
489         int n = len;
490
491         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
492
493         if (n > (sizeof(buffer)-4)) {
494                 n = sizeof(buffer)-4;
495         }
496
497         memcpy(&buffer[4], buf, n);
498         writefd_unbuffered(fd, buffer, n+4);
499
500         len -= n;
501         buf += n;
502
503         if (len) {
504                 writefd_unbuffered(fd, buf, len);
505         }
506 }
507
508
509 void io_flush(void)
510 {
511         int fd = multiplex_out_fd;
512
513         err_list_push();
514
515         if (!io_buffer_count || no_flush) return;
516
517         if (io_multiplexing_out) {
518                 mplex_write(fd, FNONE, io_buffer, io_buffer_count);
519         } else {
520                 writefd_unbuffered(fd, io_buffer, io_buffer_count);
521         }
522         io_buffer_count = 0;
523 }
524
525
526 /* XXX: fd is ignored, which seems a little strange. */
527 void io_end_buffering(int fd)
528 {
529         io_flush();
530         if (!io_multiplexing_out) {
531                 free(io_buffer);
532                 io_buffer = NULL;
533         }
534 }
535
536 static void writefd(int fd,char *buf,int len)
537 {
538         stats.total_written += len;
539
540         err_list_push();
541
542         if (!io_buffer || fd != multiplex_out_fd) {
543                 writefd_unbuffered(fd, buf, len);
544                 return;
545         }
546
547         while (len) {
548                 int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count);
549                 if (n > 0) {
550                         memcpy(io_buffer+io_buffer_count, buf, n);
551                         buf += n;
552                         len -= n;
553                         io_buffer_count += n;
554                 }
555                 
556                 if (io_buffer_count == IO_BUFFER_SIZE) io_flush();
557         }
558 }
559
560
561 void write_int(int f,int32 x)
562 {
563         char b[4];
564         SIVAL(b,0,x);
565         writefd(f,b,4);
566 }
567
568
569 /*
570  * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
571  * 64-bit types on this platform.
572  */
573 void write_longint(int f, int64 x)
574 {
575         extern int remote_version;
576         char b[8];
577
578         if (remote_version < 16 || x <= 0x7FFFFFFF) {
579                 write_int(f, (int)x);
580                 return;
581         }
582
583         write_int(f, (int32)0xFFFFFFFF);
584         SIVAL(b,0,(x&0xFFFFFFFF));
585         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
586
587         writefd(f,b,8);
588 }
589
590 void write_buf(int f,char *buf,int len)
591 {
592         writefd(f,buf,len);
593 }
594
595 /* write a string to the connection */
596 static void write_sbuf(int f,char *buf)
597 {
598         write_buf(f, buf, strlen(buf));
599 }
600
601
602 void write_byte(int f,unsigned char c)
603 {
604         write_buf(f,(char *)&c,1);
605 }
606
607
608
609 int read_line(int f, char *buf, int maxlen)
610 {
611         while (maxlen) {
612                 buf[0] = 0;
613                 read_buf(f, buf, 1);
614                 if (buf[0] == 0) return 0;
615                 if (buf[0] == '\n') {
616                         buf[0] = 0;
617                         break;
618                 }
619                 if (buf[0] != '\r') {
620                         buf++;
621                         maxlen--;
622                 }
623         }
624         if (maxlen == 0) {
625                 *buf = 0;
626                 return 0;
627         }
628
629         return 1;
630 }
631
632
633 void io_printf(int fd, const char *format, ...)
634 {
635         va_list ap;  
636         char buf[1024];
637         int len;
638         
639         va_start(ap, format);
640         len = vsnprintf(buf, sizeof(buf), format, ap);
641         va_end(ap);
642
643         if (len < 0) exit_cleanup(RERR_STREAMIO);
644
645         write_sbuf(fd, buf);
646 }
647
648
649 /* setup for multiplexing an error stream with the data stream */
650 void io_start_multiplex_out(int fd)
651 {
652         multiplex_out_fd = fd;
653         io_flush();
654         io_start_buffering(fd);
655         io_multiplexing_out = 1;
656 }
657
658 /* setup for multiplexing an error stream with the data stream */
659 void io_start_multiplex_in(int fd)
660 {
661         multiplex_in_fd = fd;
662         io_flush();
663         io_multiplexing_in = 1;
664 }
665
666 /* write an message to the multiplexed error stream */
667 int io_multiplex_write(enum logcode code, char *buf, int len)
668 {
669         if (!io_multiplexing_out) return 0;
670
671         io_flush();
672         stats.total_written += (len+4);
673         mplex_write(multiplex_out_fd, code, buf, len);
674         return 1;
675 }
676
677 /* stop output multiplexing */
678 void io_multiplexing_close(void)
679 {
680         io_multiplexing_out = 0;
681 }
682