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