2 Copyright (C) Andrew Tridgell 1996
3 Copyright (C) Paul Mackerras 1996
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.
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.
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.
21 Utilities used in rsync
27 /* if no timeout is specified then use a 60 second select timeout */
28 #define SELECT_TIMEOUT 60
30 static int io_multiplexing_out
;
31 static int io_multiplexing_in
;
32 static int multiplex_in_fd
;
33 static int multiplex_out_fd
;
34 static time_t last_io
;
35 static int eof_error
=1;
37 extern int io_timeout
;
38 extern struct stats stats
;
40 static int buffer_f_in
= -1;
42 void setup_readbuffer(int f_in
)
47 static void check_timeout(void)
51 if (!io_timeout
) return;
60 if (last_io
&& io_timeout
&& (t
-last_io
) >= io_timeout
) {
61 rprintf(FERROR
,"io timeout after %d second - exiting\n",
63 exit_cleanup(RERR_TIMEOUT
);
68 static char *read_buffer
;
69 static char *read_buffer_p
;
70 static int read_buffer_len
;
71 static int read_buffer_size
;
73 static int no_flush_read
;
75 /* read from a socket with IO timeout. return the number of
76 bytes read. If no bytes can be read then exit, never return
78 static int read_timeout(int fd
, char *buf
, int len
)
92 tv
.tv_sec
= io_timeout
?io_timeout
:SELECT_TIMEOUT
;
95 if (select(fd
+1, &fds
, NULL
, NULL
, &tv
) != 1) {
100 n
= read(fd
, buf
, len
);
107 last_io
= time(NULL
);
111 if (n
== -1 && errno
== EINTR
) {
116 (errno
== EAGAIN
|| errno
== EWOULDBLOCK
)) {
117 /* this shouldn't happen, if it does then
118 sleep for a short time to prevent us
119 chewing too much CPU */
126 rprintf(FERROR
,"unexpected EOF in read_timeout\n");
128 exit_cleanup(RERR_STREAMIO
);
131 rprintf(FERROR
,"read error: %s\n", strerror(errno
));
132 exit_cleanup(RERR_STREAMIO
);
138 /* continue trying to read len bytes - don't return until len
140 static void read_loop(int fd
, char *buf
, int len
)
143 int n
= read_timeout(fd
, buf
, len
);
150 /* read from the file descriptor handling multiplexing -
151 return number of bytes read
153 static int read_unbuffered(int fd
, char *buf
, int len
)
155 static int remaining
;
160 if (!io_multiplexing_in
|| fd
!= multiplex_in_fd
)
161 return read_timeout(fd
, buf
, len
);
165 len
= MIN(len
, remaining
);
166 read_loop(fd
, buf
, len
);
172 read_loop(fd
, ibuf
, 4);
175 remaining
= tag
& 0xFFFFFF;
178 if (tag
== MPLEX_BASE
) continue;
182 if (tag
!= FERROR
&& tag
!= FINFO
) {
183 rprintf(FERROR
,"unexpected tag %d\n", tag
);
184 exit_cleanup(RERR_STREAMIO
);
187 if (remaining
> sizeof(line
)-1) {
188 rprintf(FERROR
,"multiplexing overflow %d\n\n",
190 exit_cleanup(RERR_STREAMIO
);
193 read_loop(fd
, line
, remaining
);
196 rprintf(tag
,"%s", line
);
205 /* This function was added to overcome a deadlock problem when using
206 * ssh. It looks like we can't allow our receive queue to get full or
207 * ssh will clag up. Uggh. */
208 static void read_check(int f
)
214 if (read_buffer_len
== 0) {
215 read_buffer_p
= read_buffer
;
218 if (n
> MAX_READ_BUFFER
/4)
219 n
= MAX_READ_BUFFER
/4;
221 if (read_buffer_p
!= read_buffer
) {
222 memmove(read_buffer
,read_buffer_p
,read_buffer_len
);
223 read_buffer_p
= read_buffer
;
226 if (n
> (read_buffer_size
- read_buffer_len
)) {
227 read_buffer_size
+= n
;
228 read_buffer
= (char *)Realloc(read_buffer
,read_buffer_size
);
229 if (!read_buffer
) out_of_memory("read check");
230 read_buffer_p
= read_buffer
;
233 n
= read_unbuffered(f
,read_buffer
+read_buffer_len
,n
);
234 read_buffer_len
+= n
;
238 /* do a buffered read from fd. don't return until all N bytes
239 have been read. If all N can't be read then exit with an error */
240 static void readfd(int fd
,char *buffer
,int N
)
245 if ((read_buffer_len
< N
) && (N
< 1024)) {
246 read_check(buffer_f_in
);
250 if (read_buffer_len
> 0 && buffer_f_in
== fd
) {
251 ret
= MIN(read_buffer_len
,N
-total
);
252 memcpy(buffer
+total
,read_buffer_p
,ret
);
253 read_buffer_p
+= ret
;
254 read_buffer_len
-= ret
;
263 ret
= read_unbuffered(fd
,buffer
+ total
,N
-total
);
267 stats
.total_read
+= total
;
271 int32
read_int(int f
)
278 if (ret
== (int32
)0xffffffff) return -1;
282 int64
read_longint(int f
)
284 extern int remote_version
;
289 if ((int32
)ret
!= (int32
)0xffffffff) {
294 rprintf(FERROR
,"Integer overflow - attempted 64 bit offset\n");
295 exit_cleanup(RERR_UNSUPPORTED
);
297 if (remote_version
>= 16) {
299 ret
= IVAL(b
,0) | (((int64
)IVAL(b
,4))<<32);
306 void read_buf(int f
,char *buf
,int len
)
311 void read_sbuf(int f
,char *buf
,int len
)
317 unsigned char read_byte(int f
)
320 read_buf(f
,(char *)&c
,1);
326 /* write len bytes to fd, possibly reading from buffer_f_in if set
327 in order to unclog the pipe. don't return until all len
328 bytes have been written */
329 static void writefd_unbuffered(int fd
,char *buf
,int len
)
340 while (total
< len
) {
346 if (!no_flush_read
) {
347 reading
= (buffer_f_in
!= -1);
351 FD_SET(buffer_f_in
,&r_fds
);
352 if (buffer_f_in
> fd
)
353 fd_count
= buffer_f_in
+1;
356 tv
.tv_sec
= io_timeout
?io_timeout
:SELECT_TIMEOUT
;
359 count
= select(fd_count
,
369 if (reading
&& FD_ISSET(buffer_f_in
, &r_fds
)) {
370 read_check(buffer_f_in
);
373 if (FD_ISSET(fd
, &w_fds
)) {
374 int n
= (len
-total
)>>blocked
;
375 int ret
= write(fd
,buf
+total
,n
?n
:1);
377 if (ret
== -1 && errno
== EINTR
) {
382 (errno
== EAGAIN
|| errno
== EWOULDBLOCK
)) {
388 rprintf(FERROR
,"erroring writing %d bytes - exiting\n", len
);
389 exit_cleanup(RERR_STREAMIO
);
396 last_io
= time(NULL
);
404 static char *io_buffer
;
405 static int io_buffer_count
;
407 void io_start_buffering(int fd
)
409 if (io_buffer
) return;
410 multiplex_out_fd
= fd
;
411 io_buffer
= (char *)malloc(IO_BUFFER_SIZE
+4);
412 if (!io_buffer
) out_of_memory("writefd");
415 /* leave room for the multiplex header in case it's needed */
421 int fd
= multiplex_out_fd
;
422 if (!io_buffer_count
|| no_flush
) return;
424 if (io_multiplexing_out
) {
425 SIVAL(io_buffer
-4, 0, (MPLEX_BASE
<<24) + io_buffer_count
);
426 writefd_unbuffered(fd
, io_buffer
-4, io_buffer_count
+4);
428 writefd_unbuffered(fd
, io_buffer
, io_buffer_count
);
433 void io_end_buffering(int fd
)
436 if (!io_multiplexing_out
) {
442 static void writefd(int fd
,char *buf
,int len
)
444 stats
.total_written
+= len
;
447 writefd_unbuffered(fd
, buf
, len
);
452 int n
= MIN(len
, IO_BUFFER_SIZE
-io_buffer_count
);
454 memcpy(io_buffer
+io_buffer_count
, buf
, n
);
457 io_buffer_count
+= n
;
460 if (io_buffer_count
== IO_BUFFER_SIZE
) io_flush();
465 void write_int(int f
,int32 x
)
472 void write_longint(int f
, int64 x
)
474 extern int remote_version
;
477 if (remote_version
< 16 || x
<= 0x7FFFFFFF) {
478 write_int(f
, (int)x
);
482 write_int(f
, (int32
)0xFFFFFFFF);
483 SIVAL(b
,0,(x
&0xFFFFFFFF));
484 SIVAL(b
,4,((x
>>32)&0xFFFFFFFF));
489 void write_buf(int f
,char *buf
,int len
)
494 /* write a string to the connection */
495 static void write_sbuf(int f
,char *buf
)
497 write_buf(f
, buf
, strlen(buf
));
501 void write_byte(int f
,unsigned char c
)
503 write_buf(f
,(char *)&c
,1);
506 int read_line(int f
, char *buf
, int maxlen
)
513 if (buf
[0] == 0) return 0;
514 if (buf
[0] == '\n') {
518 if (buf
[0] != '\r') {
534 void io_printf(int fd
, const char *format
, ...)
540 va_start(ap
, format
);
541 len
= vslprintf(buf
, sizeof(buf
), format
, ap
);
544 if (len
< 0) exit_cleanup(RERR_STREAMIO
);
550 /* setup for multiplexing an error stream with the data stream */
551 void io_start_multiplex_out(int fd
)
553 multiplex_out_fd
= fd
;
555 io_start_buffering(fd
);
556 io_multiplexing_out
= 1;
559 /* setup for multiplexing an error stream with the data stream */
560 void io_start_multiplex_in(int fd
)
562 multiplex_in_fd
= fd
;
564 if (read_buffer_len
) {
565 fprintf(stderr
,"ERROR: data in read buffer at mplx start\n");
566 exit_cleanup(RERR_STREAMIO
);
569 io_multiplexing_in
= 1;
572 /* write an message to the error stream */
573 int io_multiplex_write(int f
, char *buf
, int len
)
575 if (!io_multiplexing_out
) return 0;
579 SIVAL(io_buffer
-4, 0, ((MPLEX_BASE
+ f
)<<24) + len
);
580 memcpy(io_buffer
, buf
, len
);
582 stats
.total_written
+= (len
+4);
584 writefd_unbuffered(multiplex_out_fd
, io_buffer
-4, len
+4);
588 void io_close_input(int fd
)