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