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