2 * Socket and pipe I/O utilities used in rsync.
4 * Copyright (C) 1996-2001 Andrew Tridgell
5 * Copyright (C) 1996 Paul Mackerras
6 * Copyright (C) 2001, 2002 Martin Pool <mbp@samba.org>
7 * Copyright (C) 2003-2020 Wayne Davison
9 * This program is free software; you can redistribute it and/or modify
10 * it under the terms of the GNU General Public License as published by
11 * the Free Software Foundation; either version 3 of the License, or
12 * (at your option) any later version.
14 * This program is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 * GNU General Public License for more details.
19 * You should have received a copy of the GNU General Public License along
20 * with this program; if not, visit the http://fsf.org website.
23 /* Rsync provides its own multiplexing system, which is used to send
24 * stderr and stdout over a single socket.
26 * For historical reasons this is off during the start of the
27 * connection, but it's switched on quite early using
28 * io_start_multiplex_out() and io_start_multiplex_in(). */
34 /** If no timeout is specified then use a 60 second select timeout */
35 #define SELECT_TIMEOUT 60
38 extern size_t bwlimit_writemax
;
39 extern int io_timeout
;
42 extern int am_receiver
;
43 extern int am_generator
;
44 extern int msgs2stderr
;
45 extern int inc_recurse
;
50 extern int file_total
;
51 extern int file_old_total
;
53 extern int read_batch
;
54 extern int compat_flags
;
55 extern int protect_args
;
56 extern int checksum_seed
;
57 extern int daemon_connection
;
58 extern int protocol_version
;
59 extern int remove_source_files
;
60 extern int preserve_hard_links
;
61 extern BOOL extra_flist_sending_enabled
;
62 extern BOOL flush_ok_after_signal
;
63 extern struct stats stats
;
64 extern time_t stop_at_utime
;
65 extern struct file_list
*cur_flist
;
67 extern int filesfrom_convert
;
68 extern iconv_t ic_send
, ic_recv
;
71 int csum_length
= SHORT_SUM_LENGTH
; /* initial value */
74 int forward_flist_data
= 0;
75 BOOL flist_receiving_enabled
= False
;
77 /* Ignore an EOF error if non-zero. See whine_about_eof(). */
78 int kluge_around_eof
= 0;
79 int got_kill_signal
= -1; /* is set to 0 only after multiplexed I/O starts */
84 int64 total_data_read
= 0;
85 int64 total_data_written
= 0;
90 int out_fd
; /* Both "out" and "msg" go to this fd. */
92 unsigned out_empty_len
;
93 size_t raw_data_header_pos
; /* in the out xbuf */
94 size_t raw_flushing_ends_before
; /* in the out xbuf */
95 size_t raw_input_ends_before
; /* in the in xbuf */
96 } iobuf
= { .in_fd
= -1, .out_fd
= -1 };
98 static time_t last_io_in
;
99 static time_t last_io_out
;
101 static int write_batch_monitor_in
= -1;
102 static int write_batch_monitor_out
= -1;
104 static int ff_forward_fd
= -1;
105 static int ff_reenable_multiplex
= -1;
106 static char ff_lastchar
= '\0';
107 static xbuf ff_xb
= EMPTY_XBUF
;
109 static xbuf iconv_buf
= EMPTY_XBUF
;
111 static int select_timeout
= SELECT_TIMEOUT
;
112 static int active_filecnt
= 0;
113 static OFF_T active_bytecnt
= 0;
114 static int first_message
= 1;
116 static char int_byte_extra
[64] = {
117 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, /* (00 - 3F)/4 */
118 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, /* (40 - 7F)/4 */
119 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, /* (80 - BF)/4 */
120 2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 5, 6, /* (C0 - FF)/4 */
123 /* Our I/O buffers are sized with no bits on in the lowest byte of the "size"
124 * (indeed, our rounding of sizes in 1024-byte units assures more than this).
125 * This allows the code that is storing bytes near the physical end of a
126 * circular buffer to temporarily reduce the buffer's size (in order to make
127 * some storing idioms easier), while also making it simple to restore the
128 * buffer's actual size when the buffer's "pos" wraps around to the start (we
129 * just round the buffer's size up again). */
131 #define IOBUF_WAS_REDUCED(siz) ((siz) & 0xFF)
132 #define IOBUF_RESTORE_SIZE(siz) (((siz) | 0xFF) + 1)
134 #define IN_MULTIPLEXED (iobuf.in_multiplexed != 0)
135 #define IN_MULTIPLEXED_AND_READY (iobuf.in_multiplexed > 0)
136 #define OUT_MULTIPLEXED (iobuf.out_empty_len != 0)
138 #define PIO_NEED_INPUT (1<<0) /* The *_NEED_* flags are mutually exclusive. */
139 #define PIO_NEED_OUTROOM (1<<1)
140 #define PIO_NEED_MSGROOM (1<<2)
142 #define PIO_CONSUME_INPUT (1<<4) /* Must becombined with PIO_NEED_INPUT. */
144 #define PIO_INPUT_AND_CONSUME (PIO_NEED_INPUT | PIO_CONSUME_INPUT)
145 #define PIO_NEED_FLAGS (PIO_NEED_INPUT | PIO_NEED_OUTROOM | PIO_NEED_MSGROOM)
147 #define REMOTE_OPTION_ERROR "rsync: on remote machine: -"
148 #define REMOTE_OPTION_ERROR2 ": unknown option"
150 #define FILESFROM_BUFLEN 2048
152 enum festatus
{ FES_SUCCESS
, FES_REDO
, FES_NO_SEND
};
154 static flist_ndx_list redo_list
, hlink_list
;
156 static void read_a_msg(void);
157 static void drain_multiplex_messages(void);
158 static void sleep_for_bwlimit(int bytes_written
);
160 static void check_timeout(BOOL allow_keepalive
, int keepalive_flags
)
164 /* On the receiving side, the generator is now the one that decides
165 * when a timeout has occurred. When it is sifting through a lot of
166 * files looking for work, it will be sending keep-alive messages to
167 * the sender, and even though the receiver won't be sending/receiving
168 * anything (not even keep-alive messages), the successful writes to
169 * the sender will keep things going. If the receiver is actively
170 * receiving data, it will ensure that the generator knows that it is
171 * not idle by sending the generator keep-alive messages (since the
172 * generator might be blocked trying to send checksums, it needs to
173 * know that the receiver is active). Thus, as long as one or the
174 * other is successfully doing work, the generator will not timeout. */
180 if (allow_keepalive
) {
181 /* This may put data into iobuf.msg w/o flushing. */
182 maybe_send_keepalive(t
, keepalive_flags
);
191 chk
= MAX(last_io_out
, last_io_in
);
192 if (t
- chk
>= io_timeout
) {
195 rprintf(FERROR
, "[%s] io timeout after %d seconds -- exiting\n",
196 who_am_i(), (int)(t
-chk
));
197 exit_cleanup(RERR_TIMEOUT
);
201 /* It's almost always an error to get an EOF when we're trying to read from the
202 * network, because the protocol is (for the most part) self-terminating.
204 * There is one case for the receiver when it is at the end of the transfer
205 * (hanging around reading any keep-alive packets that might come its way): if
206 * the sender dies before the generator's kill-signal comes through, we can end
207 * up here needing to loop until the kill-signal arrives. In this situation,
208 * kluge_around_eof will be < 0.
210 * There is another case for older protocol versions (< 24) where the module
211 * listing was not terminated, so we must ignore an EOF error in that case and
212 * exit. In this situation, kluge_around_eof will be > 0. */
213 static NORETURN
void whine_about_eof(BOOL allow_kluge
)
215 if (kluge_around_eof
&& allow_kluge
) {
217 if (kluge_around_eof
> 0)
219 /* If we're still here after 10 seconds, exit with an error. */
220 for (i
= 10*1000/20; i
--; )
224 rprintf(FERROR
, RSYNC_NAME
": connection unexpectedly closed "
225 "(%s bytes received so far) [%s]\n",
226 big_num(stats
.total_read
), who_am_i());
228 exit_cleanup(RERR_STREAMIO
);
231 /* Do a safe read, handling any needed looping and error handling.
232 * Returns the count of the bytes read, which will only be different
233 * from "len" if we encountered an EOF. This routine is not used on
234 * the socket except very early in the transfer. */
235 static size_t safe_read(int fd
, char *buf
, size_t len
)
239 assert(fd
!= iobuf
.in_fd
);
250 tv
.tv_sec
= select_timeout
;
253 cnt
= select(fd
+1, &r_fds
, NULL
, &e_fds
, &tv
);
255 if (cnt
< 0 && errno
== EBADF
) {
256 rsyserr(FERROR
, errno
, "safe_read select failed");
257 exit_cleanup(RERR_FILEIO
);
259 check_timeout(1, MSK_ALLOW_FLUSH
);
263 /*if (FD_ISSET(fd, &e_fds))
264 rprintf(FINFO, "select exception on fd %d\n", fd); */
266 if (FD_ISSET(fd
, &r_fds
)) {
267 int n
= read(fd
, buf
+ got
, len
- got
);
268 if (DEBUG_GTE(IO
, 2))
269 rprintf(FINFO
, "[%s] safe_read(%d)=%ld\n", who_am_i(), fd
, (long)n
);
275 rsyserr(FERROR
, errno
, "safe_read failed to read %ld bytes", (long)len
);
276 exit_cleanup(RERR_STREAMIO
);
278 if ((got
+= (size_t)n
) == len
)
286 static const char *what_fd_is(int fd
)
290 if (fd
== sock_f_out
)
292 else if (fd
== iobuf
.out_fd
)
294 else if (fd
== batch_fd
)
297 snprintf(buf
, sizeof buf
, "fd %d", fd
);
302 /* Do a safe write, handling any needed looping and error handling.
303 * Returns only if everything was successfully written. This routine
304 * is not used on the socket except very early in the transfer. */
305 static void safe_write(int fd
, const char *buf
, size_t len
)
309 assert(fd
!= iobuf
.out_fd
);
311 n
= write(fd
, buf
, len
);
312 if ((size_t)n
== len
)
315 if (errno
!= EINTR
&& errno
!= EWOULDBLOCK
&& errno
!= EAGAIN
) {
317 rsyserr(FERROR
, errno
,
318 "safe_write failed to write %ld bytes to %s",
319 (long)len
, what_fd_is(fd
));
320 exit_cleanup(RERR_STREAMIO
);
334 tv
.tv_sec
= select_timeout
;
337 cnt
= select(fd
+ 1, NULL
, &w_fds
, NULL
, &tv
);
339 if (cnt
< 0 && errno
== EBADF
) {
340 rsyserr(FERROR
, errno
, "safe_write select failed on %s", what_fd_is(fd
));
341 exit_cleanup(RERR_FILEIO
);
344 maybe_send_keepalive(time(NULL
), MSK_ALLOW_FLUSH
);
348 if (FD_ISSET(fd
, &w_fds
)) {
349 n
= write(fd
, buf
, len
);
361 /* This is only called when files-from data is known to be available. We read
362 * a chunk of data and put it into the output buffer. */
363 static void forward_filesfrom_data(void)
367 len
= read(ff_forward_fd
, ff_xb
.buf
+ ff_xb
.len
, ff_xb
.size
- ff_xb
.len
);
369 if (len
== 0 || errno
!= EINTR
) {
370 /* Send end-of-file marker */
372 write_buf(iobuf
.out_fd
, "\0\0", ff_lastchar
? 2 : 1);
374 if (ff_reenable_multiplex
>= 0)
375 io_start_multiplex_out(ff_reenable_multiplex
);
380 if (DEBUG_GTE(IO
, 2))
381 rprintf(FINFO
, "[%s] files-from read=%ld\n", who_am_i(), (long)len
);
388 char *s
= ff_xb
.buf
+ len
;
389 /* Transform CR and/or LF into '\0' */
390 while (s
-- > ff_xb
.buf
) {
391 if (*s
== '\n' || *s
== '\r')
400 /* Last buf ended with a '\0', so don't let this buf start with one. */
401 while (len
&& *s
== '\0')
403 ff_xb
.pos
= s
- ff_xb
.buf
;
407 if (filesfrom_convert
&& len
) {
408 char *sob
= ff_xb
.buf
+ ff_xb
.pos
, *s
= sob
;
409 char *eob
= sob
+ len
;
410 int flags
= ICB_INCLUDE_BAD
| ICB_INCLUDE_INCOMPLETE
| ICB_CIRCULAR_OUT
;
411 if (ff_lastchar
== '\0')
413 /* Convert/send each null-terminated string separately, skipping empties. */
416 ff_xb
.len
= s
- sob
- 1;
417 if (iconvbufs(ic_send
, &ff_xb
, &iobuf
.out
, flags
) < 0)
418 exit_cleanup(RERR_PROTOCOL
); /* impossible? */
419 write_buf(iobuf
.out_fd
, s
-1, 1); /* Send the '\0'. */
420 while (s
!= eob
&& *s
== '\0')
423 ff_xb
.pos
= sob
- ff_xb
.buf
;
428 if ((ff_xb
.len
= s
- sob
) == 0)
431 /* Handle a partial string specially, saving any incomplete chars. */
432 flags
&= ~ICB_INCLUDE_INCOMPLETE
;
433 if (iconvbufs(ic_send
, &ff_xb
, &iobuf
.out
, flags
) < 0) {
435 exit_cleanup(RERR_PROTOCOL
); /* impossible? */
437 memmove(ff_xb
.buf
, ff_xb
.buf
+ ff_xb
.pos
, ff_xb
.len
);
439 ff_lastchar
= 'x'; /* Anything non-zero. */
445 char *f
= ff_xb
.buf
+ ff_xb
.pos
;
448 /* Eliminate any multi-'\0' runs. */
450 if (!(*t
++ = *f
++)) {
451 while (f
!= eob
&& *f
== '\0')
456 if ((len
= t
- ff_xb
.buf
) != 0) {
457 /* This will not circle back to perform_io() because we only get
458 * called when there is plenty of room in the output buffer. */
459 write_buf(iobuf
.out_fd
, ff_xb
.buf
, len
);
464 void reduce_iobuf_size(xbuf
*out
, size_t new_size
)
466 if (new_size
< out
->size
) {
467 /* Avoid weird buffer interactions by only outputting this to stderr. */
468 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 4)) {
469 const char *name
= out
== &iobuf
.out
? "iobuf.out"
470 : out
== &iobuf
.msg
? "iobuf.msg"
473 rprintf(FINFO
, "[%s] reduced size of %s (-%d)\n",
474 who_am_i(), name
, (int)(out
->size
- new_size
));
477 out
->size
= new_size
;
481 void restore_iobuf_size(xbuf
*out
)
483 if (IOBUF_WAS_REDUCED(out
->size
)) {
484 size_t new_size
= IOBUF_RESTORE_SIZE(out
->size
);
485 /* Avoid weird buffer interactions by only outputting this to stderr. */
486 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 4)) {
487 const char *name
= out
== &iobuf
.out
? "iobuf.out"
488 : out
== &iobuf
.msg
? "iobuf.msg"
491 rprintf(FINFO
, "[%s] restored size of %s (+%d)\n",
492 who_am_i(), name
, (int)(new_size
- out
->size
));
495 out
->size
= new_size
;
499 static void handle_kill_signal(BOOL flush_ok
)
501 got_kill_signal
= -1;
502 flush_ok_after_signal
= flush_ok
;
503 exit_cleanup(RERR_SIGNAL
);
506 /* Perform buffered input and/or output until specified conditions are met.
507 * When given a "needed" read or write request, this returns without doing any
508 * I/O if the needed input bytes or write space is already available. Once I/O
509 * is needed, this will try to do whatever reading and/or writing is currently
510 * possible, up to the maximum buffer allowances, no matter if this is a read
511 * or write request. However, the I/O stops as soon as the required input
512 * bytes or output space is available. If this is not a read request, the
513 * routine may also do some advantageous reading of messages from a multiplexed
514 * input source (which ensures that we don't jam up with everyone in their
515 * "need to write" code and nobody reading the accumulated data that would make
518 * The iobuf.in, .out and .msg buffers are all circular. Callers need to be
519 * aware that some data copies will need to be split when the bytes wrap around
520 * from the end to the start. In order to help make writing into the output
521 * buffers easier for some operations (such as the use of SIVAL() into the
522 * buffer) a buffer may be temporarily shortened by a small amount, but the
523 * original size will be automatically restored when the .pos wraps to the
524 * start. See also the 3 raw_* iobuf vars that are used in the handling of
525 * MSG_DATA bytes as they are read-from/written-into the buffers.
527 * When writing, we flush data in the following priority order:
529 * 1. Finish writing any in-progress MSG_DATA sequence from iobuf.out.
531 * 2. Write out all the messages from the message buf (if iobuf.msg is active).
532 * Yes, this means that a PIO_NEED_OUTROOM call will completely flush any
533 * messages before getting to the iobuf.out flushing (except for rule 1).
535 * 3. Write out the raw data from iobuf.out, possibly filling in the multiplexed
536 * MSG_DATA header that was pre-allocated (when output is multiplexed).
538 * TODO: items for possible future work:
540 * - Make this routine able to read the generator-to-receiver batch flow?
542 * Unlike the old routines that this replaces, it is OK to read ahead as far as
543 * we can because the read_a_msg() routine now reads its bytes out of the input
544 * buffer. In the old days, only raw data was in the input buffer, and any
545 * unused raw data in the buf would prevent the reading of socket data. */
546 static char *perform_io(size_t needed
, int flags
)
548 fd_set r_fds
, e_fds
, w_fds
;
551 size_t empty_buf_len
= 0;
555 if (iobuf
.in
.len
== 0 && iobuf
.in
.pos
!= 0) {
556 if (iobuf
.raw_input_ends_before
)
557 iobuf
.raw_input_ends_before
-= iobuf
.in
.pos
;
561 switch (flags
& PIO_NEED_FLAGS
) {
563 /* We never resize the circular input buffer. */
564 if (iobuf
.in
.size
< needed
) {
565 rprintf(FERROR
, "need to read %ld bytes, iobuf.in.buf is only %ld bytes.\n",
566 (long)needed
, (long)iobuf
.in
.size
);
567 exit_cleanup(RERR_PROTOCOL
);
570 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 3)) {
571 rprintf(FINFO
, "[%s] perform_io(%ld, %sinput)\n",
572 who_am_i(), (long)needed
, flags
& PIO_CONSUME_INPUT
? "consume&" : "");
576 case PIO_NEED_OUTROOM
:
577 /* We never resize the circular output buffer. */
578 if (iobuf
.out
.size
- iobuf
.out_empty_len
< needed
) {
579 fprintf(stderr
, "need to write %ld bytes, iobuf.out.buf is only %ld bytes.\n",
580 (long)needed
, (long)(iobuf
.out
.size
- iobuf
.out_empty_len
));
581 exit_cleanup(RERR_PROTOCOL
);
584 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 3)) {
585 rprintf(FINFO
, "[%s] perform_io(%ld, outroom) needs to flush %ld\n",
586 who_am_i(), (long)needed
,
587 iobuf
.out
.len
+ needed
> iobuf
.out
.size
588 ? (long)(iobuf
.out
.len
+ needed
- iobuf
.out
.size
) : 0L);
592 case PIO_NEED_MSGROOM
:
593 /* We never resize the circular message buffer. */
594 if (iobuf
.msg
.size
< needed
) {
595 fprintf(stderr
, "need to write %ld bytes, iobuf.msg.buf is only %ld bytes.\n",
596 (long)needed
, (long)iobuf
.msg
.size
);
597 exit_cleanup(RERR_PROTOCOL
);
600 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 3)) {
601 rprintf(FINFO
, "[%s] perform_io(%ld, msgroom) needs to flush %ld\n",
602 who_am_i(), (long)needed
,
603 iobuf
.msg
.len
+ needed
> iobuf
.msg
.size
604 ? (long)(iobuf
.msg
.len
+ needed
- iobuf
.msg
.size
) : 0L);
609 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 3))
610 rprintf(FINFO
, "[%s] perform_io(%ld, %d)\n", who_am_i(), (long)needed
, flags
);
614 exit_cleanup(RERR_UNSUPPORTED
);
618 switch (flags
& PIO_NEED_FLAGS
) {
620 if (iobuf
.in
.len
>= needed
)
623 case PIO_NEED_OUTROOM
:
624 /* Note that iobuf.out_empty_len doesn't factor into this check
625 * because iobuf.out.len already holds any needed header len. */
626 if (iobuf
.out
.len
+ needed
<= iobuf
.out
.size
)
629 case PIO_NEED_MSGROOM
:
630 if (iobuf
.msg
.len
+ needed
<= iobuf
.msg
.size
)
639 if (iobuf
.in_fd
>= 0 && iobuf
.in
.size
- iobuf
.in
.len
) {
640 if (!read_batch
|| batch_fd
>= 0) {
641 FD_SET(iobuf
.in_fd
, &r_fds
);
642 FD_SET(iobuf
.in_fd
, &e_fds
);
644 if (iobuf
.in_fd
> max_fd
)
645 max_fd
= iobuf
.in_fd
;
648 /* Only do more filesfrom processing if there is enough room in the out buffer. */
649 if (ff_forward_fd
>= 0 && iobuf
.out
.size
- iobuf
.out
.len
> FILESFROM_BUFLEN
*2) {
650 FD_SET(ff_forward_fd
, &r_fds
);
651 if (ff_forward_fd
> max_fd
)
652 max_fd
= ff_forward_fd
;
656 if (iobuf
.out_fd
>= 0) {
657 if (iobuf
.raw_flushing_ends_before
658 || (!iobuf
.msg
.len
&& iobuf
.out
.len
> iobuf
.out_empty_len
&& !(flags
& PIO_NEED_MSGROOM
))) {
659 if (OUT_MULTIPLEXED
&& !iobuf
.raw_flushing_ends_before
) {
660 /* The iobuf.raw_flushing_ends_before value can point off the end
661 * of the iobuf.out buffer for a while, for easier subtracting. */
662 iobuf
.raw_flushing_ends_before
= iobuf
.out
.pos
+ iobuf
.out
.len
;
664 SIVAL(iobuf
.out
.buf
+ iobuf
.raw_data_header_pos
, 0,
665 ((MPLEX_BASE
+ (int)MSG_DATA
)<<24) + iobuf
.out
.len
- 4);
667 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 1)) {
668 rprintf(FINFO
, "[%s] send_msg(%d, %ld)\n",
669 who_am_i(), (int)MSG_DATA
, (long)iobuf
.out
.len
- 4);
672 /* reserve room for the next MSG_DATA header */
673 iobuf
.raw_data_header_pos
= iobuf
.raw_flushing_ends_before
;
674 if (iobuf
.raw_data_header_pos
>= iobuf
.out
.size
)
675 iobuf
.raw_data_header_pos
-= iobuf
.out
.size
;
676 else if (iobuf
.raw_data_header_pos
+ 4 > iobuf
.out
.size
) {
677 /* The 4-byte header won't fit at the end of the buffer,
678 * so we'll temporarily reduce the output buffer's size
679 * and put the header at the start of the buffer. */
680 reduce_iobuf_size(&iobuf
.out
, iobuf
.raw_data_header_pos
);
681 iobuf
.raw_data_header_pos
= 0;
683 /* Yes, it is possible for this to make len > size for a while. */
687 empty_buf_len
= iobuf
.out_empty_len
;
689 } else if (iobuf
.msg
.len
) {
695 FD_SET(iobuf
.out_fd
, &w_fds
);
696 if (iobuf
.out_fd
> max_fd
)
697 max_fd
= iobuf
.out_fd
;
703 switch (flags
& PIO_NEED_FLAGS
) {
706 if (kluge_around_eof
== 2)
708 if (iobuf
.in_fd
== -2)
709 whine_about_eof(True
);
710 rprintf(FERROR
, "error in perform_io: no fd for input.\n");
711 exit_cleanup(RERR_PROTOCOL
);
712 case PIO_NEED_OUTROOM
:
713 case PIO_NEED_MSGROOM
:
715 drain_multiplex_messages();
716 if (iobuf
.out_fd
== -2)
717 whine_about_eof(True
);
718 rprintf(FERROR
, "error in perform_io: no fd for output.\n");
719 exit_cleanup(RERR_PROTOCOL
);
721 /* No stated needs, so I guess this is OK. */
727 if (got_kill_signal
> 0)
728 handle_kill_signal(True
);
730 if (extra_flist_sending_enabled
) {
731 if (file_total
- file_old_total
< MAX_FILECNT_LOOKAHEAD
&& IN_MULTIPLEXED_AND_READY
)
734 extra_flist_sending_enabled
= False
;
735 tv
.tv_sec
= select_timeout
;
738 tv
.tv_sec
= select_timeout
;
741 cnt
= select(max_fd
+ 1, &r_fds
, &w_fds
, &e_fds
, &tv
);
744 if (cnt
< 0 && errno
== EBADF
) {
746 exit_cleanup(RERR_SOCKETIO
);
748 if (extra_flist_sending_enabled
) {
749 extra_flist_sending_enabled
= False
;
750 send_extra_file_list(sock_f_out
, -1);
751 extra_flist_sending_enabled
= !flist_eof
;
753 check_timeout((flags
& PIO_NEED_INPUT
) != 0, 0);
754 FD_ZERO(&r_fds
); /* Just in case... */
758 if (iobuf
.in_fd
>= 0 && FD_ISSET(iobuf
.in_fd
, &r_fds
)) {
759 size_t len
, pos
= iobuf
.in
.pos
+ iobuf
.in
.len
;
761 if (pos
>= iobuf
.in
.size
) {
762 pos
-= iobuf
.in
.size
;
763 len
= iobuf
.in
.size
- iobuf
.in
.len
;
765 len
= iobuf
.in
.size
- pos
;
766 if ((n
= read(iobuf
.in_fd
, iobuf
.in
.buf
+ pos
, len
)) <= 0) {
768 /* Signal that input has become invalid. */
769 if (!read_batch
|| batch_fd
< 0 || am_generator
)
774 if (errno
== EINTR
|| errno
== EWOULDBLOCK
|| errno
== EAGAIN
)
777 /* Don't write errors on a dead socket. */
778 if (iobuf
.in_fd
== sock_f_in
) {
781 rsyserr(FERROR_SOCKET
, errno
, "read error");
783 rsyserr(FERROR
, errno
, "read error");
784 exit_cleanup(RERR_SOCKETIO
);
787 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2))
788 rprintf(FINFO
, "[%s] recv=%ld\n", who_am_i(), (long)n
);
790 if (io_timeout
|| stop_at_utime
) {
791 last_io_in
= time(NULL
);
792 if (stop_at_utime
&& last_io_in
>= stop_at_utime
) {
793 rprintf(FERROR
, "stopping at requested limit\n");
794 exit_cleanup(RERR_TIMEOUT
);
796 if (io_timeout
&& flags
& PIO_NEED_INPUT
)
797 maybe_send_keepalive(last_io_in
, 0);
799 stats
.total_read
+= n
;
804 if (out
&& FD_ISSET(iobuf
.out_fd
, &w_fds
)) {
805 size_t len
= iobuf
.raw_flushing_ends_before
? iobuf
.raw_flushing_ends_before
- out
->pos
: out
->len
;
808 if (bwlimit_writemax
&& len
> bwlimit_writemax
)
809 len
= bwlimit_writemax
;
811 if (out
->pos
+ len
> out
->size
)
812 len
= out
->size
- out
->pos
;
813 if ((n
= write(iobuf
.out_fd
, out
->buf
+ out
->pos
, len
)) <= 0) {
814 if (errno
== EINTR
|| errno
== EWOULDBLOCK
|| errno
== EAGAIN
)
817 /* Don't write errors on a dead socket. */
820 iobuf
.out
.len
= iobuf
.msg
.len
= iobuf
.raw_flushing_ends_before
= 0;
821 rsyserr(FERROR_SOCKET
, errno
, "write error");
822 drain_multiplex_messages();
823 exit_cleanup(RERR_SOCKETIO
);
826 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2)) {
827 rprintf(FINFO
, "[%s] %s sent=%ld\n",
828 who_am_i(), out
== &iobuf
.out
? "out" : "msg", (long)n
);
832 last_io_out
= time(NULL
);
833 stats
.total_written
+= n
;
835 if (bwlimit_writemax
)
836 sleep_for_bwlimit(n
);
838 if ((out
->pos
+= n
) == out
->size
) {
839 if (iobuf
.raw_flushing_ends_before
)
840 iobuf
.raw_flushing_ends_before
-= out
->size
;
842 restore_iobuf_size(out
);
843 } else if (out
->pos
== iobuf
.raw_flushing_ends_before
)
844 iobuf
.raw_flushing_ends_before
= 0;
845 if ((out
->len
-= n
) == empty_buf_len
) {
847 restore_iobuf_size(out
);
849 iobuf
.raw_data_header_pos
= 0;
853 if (got_kill_signal
> 0)
854 handle_kill_signal(True
);
856 /* We need to help prevent deadlock by doing what reading
857 * we can whenever we are here trying to write. */
858 if (IN_MULTIPLEXED_AND_READY
&& !(flags
& PIO_NEED_INPUT
)) {
859 while (!iobuf
.raw_input_ends_before
&& iobuf
.in
.len
> 512)
861 if (flist_receiving_enabled
&& iobuf
.in
.len
> 512)
862 wait_for_receiver(); /* generator only */
865 if (ff_forward_fd
>= 0 && FD_ISSET(ff_forward_fd
, &r_fds
)) {
866 /* This can potentially flush all output and enable
867 * multiplexed output, so keep this last in the loop
868 * and be sure to not cache anything that would break
870 forward_filesfrom_data();
875 if (got_kill_signal
> 0)
876 handle_kill_signal(True
);
878 data
= iobuf
.in
.buf
+ iobuf
.in
.pos
;
880 if (flags
& PIO_CONSUME_INPUT
) {
881 iobuf
.in
.len
-= needed
;
882 iobuf
.in
.pos
+= needed
;
883 if (iobuf
.in
.pos
== iobuf
.raw_input_ends_before
)
884 iobuf
.raw_input_ends_before
= 0;
885 if (iobuf
.in
.pos
>= iobuf
.in
.size
) {
886 iobuf
.in
.pos
-= iobuf
.in
.size
;
887 if (iobuf
.raw_input_ends_before
)
888 iobuf
.raw_input_ends_before
-= iobuf
.in
.size
;
895 static void raw_read_buf(char *buf
, size_t len
)
897 size_t pos
= iobuf
.in
.pos
;
898 char *data
= perform_io(len
, PIO_INPUT_AND_CONSUME
);
899 if (iobuf
.in
.pos
<= pos
&& len
) {
900 size_t siz
= len
- iobuf
.in
.pos
;
901 memcpy(buf
, data
, siz
);
902 memcpy(buf
+ siz
, iobuf
.in
.buf
, iobuf
.in
.pos
);
904 memcpy(buf
, data
, len
);
907 static int32
raw_read_int(void)
910 if (iobuf
.in
.size
- iobuf
.in
.pos
>= 4)
911 data
= perform_io(4, PIO_INPUT_AND_CONSUME
);
913 raw_read_buf(data
= buf
, 4);
914 return IVAL(data
, 0);
917 void noop_io_until_death(void)
921 if (!iobuf
.in
.buf
|| !iobuf
.out
.buf
|| iobuf
.in_fd
< 0 || iobuf
.out_fd
< 0 || kluge_around_eof
)
924 /* If we're talking to a daemon over a socket, don't short-circuit this logic */
925 if (msgs2stderr
&& daemon_connection
>= 0)
928 kluge_around_eof
= 2;
929 /* Setting an I/O timeout ensures that if something inexplicably weird
930 * happens, we won't hang around forever. */
935 read_buf(iobuf
.in_fd
, buf
, sizeof buf
);
938 /* Buffer a message for the multiplexed output stream. Is not used for (normal) MSG_DATA. */
939 int send_msg(enum msgcode code
, const char *buf
, size_t len
, int convert
)
943 BOOL want_debug
= DEBUG_GTE(IO
, 1) && convert
>= 0 && (msgs2stderr
== 1 || code
!= MSG_INFO
);
945 if (!OUT_MULTIPLEXED
)
949 rprintf(FINFO
, "[%s] send_msg(%d, %ld)\n", who_am_i(), (int)code
, (long)len
);
951 /* When checking for enough free space for this message, we need to
952 * make sure that there is space for the 4-byte header, plus we'll
953 * assume that we may waste up to 3 bytes (if the header doesn't fit
954 * at the physical end of the buffer). */
956 if (convert
> 0 && ic_send
== (iconv_t
)-1)
959 /* Ensuring double-size room leaves space for maximal conversion expansion. */
960 needed
= len
*2 + 4 + 3;
963 needed
= len
+ 4 + 3;
964 if (iobuf
.msg
.len
+ needed
> iobuf
.msg
.size
) {
966 perform_io(needed
, PIO_NEED_MSGROOM
);
967 else { /* We sometimes allow the iobuf.msg size to increase to avoid a deadlock. */
968 size_t old_size
= iobuf
.msg
.size
;
969 restore_iobuf_size(&iobuf
.msg
);
970 realloc_xbuf(&iobuf
.msg
, iobuf
.msg
.size
* 2);
971 if (iobuf
.msg
.pos
+ iobuf
.msg
.len
> old_size
)
972 memcpy(iobuf
.msg
.buf
+ old_size
, iobuf
.msg
.buf
, iobuf
.msg
.pos
+ iobuf
.msg
.len
- old_size
);
976 pos
= iobuf
.msg
.pos
+ iobuf
.msg
.len
; /* Must be set after any flushing. */
977 if (pos
>= iobuf
.msg
.size
)
978 pos
-= iobuf
.msg
.size
;
979 else if (pos
+ 4 > iobuf
.msg
.size
) {
980 /* The 4-byte header won't fit at the end of the buffer,
981 * so we'll temporarily reduce the message buffer's size
982 * and put the header at the start of the buffer. */
983 reduce_iobuf_size(&iobuf
.msg
, pos
);
986 hdr
= iobuf
.msg
.buf
+ pos
;
988 iobuf
.msg
.len
+= 4; /* Allocate room for the coming header bytes. */
994 INIT_XBUF(inbuf
, (char*)buf
, len
, (size_t)-1);
997 iconvbufs(ic_send
, &inbuf
, &iobuf
.msg
,
998 ICB_INCLUDE_BAD
| ICB_INCLUDE_INCOMPLETE
| ICB_CIRCULAR_OUT
| ICB_INIT
);
1000 rprintf(FERROR
, "overflowed iobuf.msg buffer in send_msg");
1001 exit_cleanup(RERR_UNSUPPORTED
);
1003 len
= iobuf
.msg
.len
- len
;
1009 if ((pos
+= 4) == iobuf
.msg
.size
)
1012 /* Handle a split copy if we wrap around the end of the circular buffer. */
1013 if (pos
>= iobuf
.msg
.pos
&& (siz
= iobuf
.msg
.size
- pos
) < len
) {
1014 memcpy(iobuf
.msg
.buf
+ pos
, buf
, siz
);
1015 memcpy(iobuf
.msg
.buf
, buf
+ siz
, len
- siz
);
1017 memcpy(iobuf
.msg
.buf
+ pos
, buf
, len
);
1019 iobuf
.msg
.len
+= len
;
1022 SIVAL(hdr
, 0, ((MPLEX_BASE
+ (int)code
)<<24) + len
);
1024 if (want_debug
&& convert
> 0)
1025 rprintf(FINFO
, "[%s] converted msg len=%ld\n", who_am_i(), (long)len
);
1030 void send_msg_int(enum msgcode code
, int num
)
1034 if (DEBUG_GTE(IO
, 1))
1035 rprintf(FINFO
, "[%s] send_msg_int(%d, %d)\n", who_am_i(), (int)code
, num
);
1037 SIVAL(numbuf
, 0, num
);
1038 send_msg(code
, numbuf
, 4, -1);
1041 static void got_flist_entry_status(enum festatus status
, int ndx
)
1043 struct file_list
*flist
= flist_for_ndx(ndx
, "got_flist_entry_status");
1045 if (remove_source_files
) {
1047 active_bytecnt
-= F_LENGTH(flist
->files
[ndx
- flist
->ndx_start
]);
1051 flist
->in_progress
--;
1055 if (remove_source_files
)
1056 send_msg_int(MSG_SUCCESS
, ndx
);
1059 #ifdef SUPPORT_HARD_LINKS
1060 if (preserve_hard_links
) {
1061 struct file_struct
*file
= flist
->files
[ndx
- flist
->ndx_start
];
1062 if (F_IS_HLINKED(file
)) {
1063 if (status
== FES_NO_SEND
)
1064 flist_ndx_push(&hlink_list
, -2); /* indicates a failure follows */
1065 flist_ndx_push(&hlink_list
, ndx
);
1067 flist
->in_progress
++;
1075 flist
->in_progress
++;
1080 flist_ndx_push(&redo_list
, ndx
);
1085 /* Note the fds used for the main socket (which might really be a pipe
1086 * for a local transfer, but we can ignore that). */
1087 void io_set_sock_fds(int f_in
, int f_out
)
1093 void set_io_timeout(int secs
)
1096 allowed_lull
= (io_timeout
+ 1) / 2;
1098 if (!io_timeout
|| allowed_lull
> SELECT_TIMEOUT
)
1099 select_timeout
= SELECT_TIMEOUT
;
1101 select_timeout
= allowed_lull
;
1107 static void check_for_d_option_error(const char *msg
)
1109 static char rsync263_opts
[] = "BCDHIKLPRSTWabceghlnopqrtuvxz";
1114 || strncmp(msg
, REMOTE_OPTION_ERROR
, sizeof REMOTE_OPTION_ERROR
- 1) != 0)
1117 msg
+= sizeof REMOTE_OPTION_ERROR
- 1;
1118 if (*msg
== '-' || (colon
= strchr(msg
, ':')) == NULL
1119 || strncmp(colon
, REMOTE_OPTION_ERROR2
, sizeof REMOTE_OPTION_ERROR2
- 1) != 0)
1122 for ( ; *msg
!= ':'; msg
++) {
1125 else if (*msg
== 'e')
1127 else if (strchr(rsync263_opts
, *msg
) == NULL
)
1132 rprintf(FWARNING
, "*** Try using \"--old-d\" if remote rsync is <= 2.6.3 ***\n");
1136 /* This is used by the generator to limit how many file transfers can
1137 * be active at once when --remove-source-files is specified. Without
1138 * this, sender-side deletions were mostly happening at the end. */
1139 void increment_active_files(int ndx
, int itemizing
, enum logcode code
)
1142 /* TODO: tune these limits? */
1143 int limit
= active_bytecnt
>= 128*1024 ? 10 : 50;
1144 if (active_filecnt
< limit
)
1146 check_for_finished_files(itemizing
, code
, 0);
1147 if (active_filecnt
< limit
)
1149 wait_for_receiver();
1153 active_bytecnt
+= F_LENGTH(cur_flist
->files
[ndx
- cur_flist
->ndx_start
]);
1156 int get_redo_num(void)
1158 return flist_ndx_pop(&redo_list
);
1161 int get_hlink_num(void)
1163 return flist_ndx_pop(&hlink_list
);
1166 /* When we're the receiver and we have a local --files-from list of names
1167 * that needs to be sent over the socket to the sender, we have to do two
1168 * things at the same time: send the sender a list of what files we're
1169 * processing and read the incoming file+info list from the sender. We do
1170 * this by making recv_file_list() call forward_filesfrom_data(), which
1171 * will ensure that we forward data to the sender until we get some data
1172 * for recv_file_list() to use. */
1173 void start_filesfrom_forwarding(int fd
)
1175 if (protocol_version
< 31 && OUT_MULTIPLEXED
) {
1176 /* Older protocols send the files-from data w/o packaging
1177 * it in multiplexed I/O packets, so temporarily switch
1178 * to buffered I/O to match this behavior. */
1179 iobuf
.msg
.pos
= iobuf
.msg
.len
= 0; /* Be extra sure no messages go out. */
1180 ff_reenable_multiplex
= io_end_multiplex_out(MPLX_TO_BUFFERED
);
1184 alloc_xbuf(&ff_xb
, FILESFROM_BUFLEN
);
1187 /* Read a line into the "buf" buffer. */
1188 int read_line(int fd
, char *buf
, size_t bufsiz
, int flags
)
1193 if (flags
& RL_CONVERT
&& iconv_buf
.size
< bufsiz
)
1194 realloc_xbuf(&iconv_buf
, ROUND_UP_1024(bufsiz
) + 1024);
1199 s
= flags
& RL_CONVERT
? iconv_buf
.buf
: buf
;
1203 eob
= s
+ bufsiz
- 1;
1205 /* We avoid read_byte() for files because files can return an EOF. */
1206 if (fd
== iobuf
.in_fd
)
1208 else if (safe_read(fd
, &ch
, 1) == 0)
1210 if (flags
& RL_EOL_NULLS
? ch
== '\0' : (ch
== '\r' || ch
== '\n')) {
1211 /* Skip empty lines if dumping comments. */
1212 if (flags
& RL_DUMP_COMMENTS
&& s
== buf
)
1221 if (flags
& RL_DUMP_COMMENTS
&& (*buf
== '#' || *buf
== ';'))
1225 if (flags
& RL_CONVERT
) {
1227 INIT_XBUF(outbuf
, buf
, 0, bufsiz
);
1229 iconv_buf
.len
= s
- iconv_buf
.buf
;
1230 iconvbufs(ic_recv
, &iconv_buf
, &outbuf
,
1231 ICB_INCLUDE_BAD
| ICB_INCLUDE_INCOMPLETE
| ICB_INIT
);
1232 outbuf
.buf
[outbuf
.len
] = '\0';
1240 void read_args(int f_in
, char *mod_name
, char *buf
, size_t bufsiz
, int rl_nulls
,
1241 char ***argv_p
, int *argc_p
, char **request_p
)
1243 int maxargs
= MAX_ARGS
;
1244 int dot_pos
= 0, argc
= 0, request_len
= 0;
1246 int rl_flags
= (rl_nulls
? RL_EOL_NULLS
: 0);
1249 rl_flags
|= (protect_args
&& ic_recv
!= (iconv_t
)-1 ? RL_CONVERT
: 0);
1252 argv
= new_array(char *, maxargs
);
1253 if (mod_name
&& !protect_args
)
1254 argv
[argc
++] = "rsyncd";
1260 if (read_line(f_in
, buf
, bufsiz
, rl_flags
) == 0)
1263 if (argc
== maxargs
-1) {
1264 maxargs
+= MAX_ARGS
;
1265 argv
= realloc_array(argv
, char *, maxargs
);
1269 if (request_p
&& request_len
< 1024) {
1270 int len
= strlen(buf
);
1272 request_p
[0][request_len
++] = ' ';
1273 *request_p
= realloc_array(*request_p
, char, request_len
+ len
+ 1);
1274 memcpy(*request_p
+ request_len
, buf
, len
+ 1);
1278 glob_expand_module(mod_name
, buf
, &argv
, &argc
, &maxargs
);
1280 glob_expand(buf
, &argv
, &argc
, &maxargs
);
1284 if (*p
== '.' && p
[1] == '\0')
1290 glob_expand(NULL
, NULL
, NULL
, NULL
);
1296 BOOL
io_start_buffering_out(int f_out
)
1298 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2))
1299 rprintf(FINFO
, "[%s] io_start_buffering_out(%d)\n", who_am_i(), f_out
);
1301 if (iobuf
.out
.buf
) {
1302 if (iobuf
.out_fd
== -1)
1303 iobuf
.out_fd
= f_out
;
1305 assert(f_out
== iobuf
.out_fd
);
1309 alloc_xbuf(&iobuf
.out
, ROUND_UP_1024(IO_BUFFER_SIZE
* 2));
1310 iobuf
.out_fd
= f_out
;
1315 BOOL
io_start_buffering_in(int f_in
)
1317 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2))
1318 rprintf(FINFO
, "[%s] io_start_buffering_in(%d)\n", who_am_i(), f_in
);
1321 if (iobuf
.in_fd
== -1)
1324 assert(f_in
== iobuf
.in_fd
);
1328 alloc_xbuf(&iobuf
.in
, ROUND_UP_1024(IO_BUFFER_SIZE
));
1334 void io_end_buffering_in(BOOL free_buffers
)
1336 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2)) {
1337 rprintf(FINFO
, "[%s] io_end_buffering_in(IOBUF_%s_BUFS)\n",
1338 who_am_i(), free_buffers
? "FREE" : "KEEP");
1342 free_xbuf(&iobuf
.in
);
1344 iobuf
.in
.pos
= iobuf
.in
.len
= 0;
1349 void io_end_buffering_out(BOOL free_buffers
)
1351 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2)) {
1352 rprintf(FINFO
, "[%s] io_end_buffering_out(IOBUF_%s_BUFS)\n",
1353 who_am_i(), free_buffers
? "FREE" : "KEEP");
1356 io_flush(FULL_FLUSH
);
1359 free_xbuf(&iobuf
.out
);
1360 free_xbuf(&iobuf
.msg
);
1366 void maybe_flush_socket(int important
)
1368 if (flist_eof
&& iobuf
.out
.buf
&& iobuf
.out
.len
> iobuf
.out_empty_len
1369 && (important
|| time(NULL
) - last_io_out
>= 5))
1370 io_flush(NORMAL_FLUSH
);
1373 /* Older rsync versions used to send either a MSG_NOOP (protocol 30) or a
1374 * raw-data-based keep-alive (protocol 29), both of which implied forwarding of
1375 * the message through the sender. Since the new timeout method does not need
1376 * any forwarding, we just send an empty MSG_DATA message, which works with all
1377 * rsync versions. This avoids any message forwarding, and leaves the raw-data
1378 * stream alone (since we can never be quite sure if that stream is in the
1379 * right state for a keep-alive message). */
1380 void maybe_send_keepalive(time_t now
, int flags
)
1382 if (flags
& MSK_ACTIVE_RECEIVER
)
1383 last_io_in
= now
; /* Fudge things when we're working hard on the files. */
1385 /* Early in the transfer (before the receiver forks) the receiving side doesn't
1386 * care if it hasn't sent data in a while as long as it is receiving data (in
1387 * fact, a pre-3.1.0 rsync would die if we tried to send it a keep alive during
1388 * this time). So, if we're an early-receiving proc, just return and let the
1389 * incoming data determine if we timeout. */
1390 if (!am_sender
&& !am_receiver
&& !am_generator
)
1393 if (now
- last_io_out
>= allowed_lull
) {
1394 /* The receiver is special: it only sends keep-alive messages if it is
1395 * actively receiving data. Otherwise, it lets the generator timeout. */
1396 if (am_receiver
&& now
- last_io_in
>= io_timeout
)
1399 if (!iobuf
.msg
.len
&& iobuf
.out
.len
== iobuf
.out_empty_len
)
1400 send_msg(MSG_DATA
, "", 0, 0);
1401 if (!(flags
& MSK_ALLOW_FLUSH
)) {
1402 /* Let the caller worry about writing out the data. */
1403 } else if (iobuf
.msg
.len
)
1404 perform_io(iobuf
.msg
.size
- iobuf
.msg
.len
+ 1, PIO_NEED_MSGROOM
);
1405 else if (iobuf
.out
.len
> iobuf
.out_empty_len
)
1406 io_flush(NORMAL_FLUSH
);
1410 void start_flist_forward(int ndx
)
1412 write_int(iobuf
.out_fd
, ndx
);
1413 forward_flist_data
= 1;
1416 void stop_flist_forward(void)
1418 forward_flist_data
= 0;
1421 /* Read a message from a multiplexed source. */
1422 static void read_a_msg(void)
1424 char data
[BIGPATHBUFLEN
];
1428 /* This ensures that perform_io() does not try to do any message reading
1429 * until we've read all of the data for this message. We should also
1430 * try to avoid calling things that will cause data to be written via
1431 * perform_io() prior to this being reset to 1. */
1432 iobuf
.in_multiplexed
= -1;
1434 tag
= raw_read_int();
1436 msg_bytes
= tag
& 0xFFFFFF;
1437 tag
= (tag
>> 24) - MPLEX_BASE
;
1439 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 1))
1440 rprintf(FINFO
, "[%s] got msg=%d, len=%ld\n", who_am_i(), (int)tag
, (long)msg_bytes
);
1444 assert(iobuf
.raw_input_ends_before
== 0);
1445 /* Though this does not yet read the data, we do mark where in
1446 * the buffer the msg data will end once it is read. It is
1447 * possible that this points off the end of the buffer, in
1448 * which case the gradual reading of the input stream will
1449 * cause this value to wrap around and eventually become real. */
1451 iobuf
.raw_input_ends_before
= iobuf
.in
.pos
+ msg_bytes
;
1452 iobuf
.in_multiplexed
= 1;
1455 if (msg_bytes
!= sizeof stats
.total_read
|| !am_generator
)
1457 raw_read_buf((char*)&stats
.total_read
, sizeof stats
.total_read
);
1458 iobuf
.in_multiplexed
= 1;
1461 if (msg_bytes
!= 4 || !am_generator
)
1463 val
= raw_read_int();
1464 iobuf
.in_multiplexed
= 1;
1465 got_flist_entry_status(FES_REDO
, val
);
1470 val
= raw_read_int();
1471 iobuf
.in_multiplexed
= 1;
1474 send_msg_int(MSG_IO_ERROR
, val
);
1476 case MSG_IO_TIMEOUT
:
1477 if (msg_bytes
!= 4 || am_server
|| am_generator
)
1479 val
= raw_read_int();
1480 iobuf
.in_multiplexed
= 1;
1481 if (!io_timeout
|| io_timeout
> val
) {
1482 if (INFO_GTE(MISC
, 2))
1483 rprintf(FINFO
, "Setting --timeout=%d to match server\n", val
);
1484 set_io_timeout(val
);
1488 /* Support protocol-30 keep-alive method. */
1491 iobuf
.in_multiplexed
= 1;
1493 maybe_send_keepalive(time(NULL
), MSK_ALLOW_FLUSH
);
1496 if (msg_bytes
>= sizeof data
)
1499 raw_read_buf(data
, msg_bytes
);
1500 iobuf
.in_multiplexed
= 1;
1501 send_msg(MSG_DELETED
, data
, msg_bytes
, 1);
1505 if (ic_recv
!= (iconv_t
)-1) {
1509 int flags
= ICB_INCLUDE_BAD
| ICB_INIT
;
1511 INIT_CONST_XBUF(outbuf
, data
);
1512 INIT_XBUF(inbuf
, ibuf
, 0, (size_t)-1);
1515 size_t len
= msg_bytes
> sizeof ibuf
- inbuf
.len
? sizeof ibuf
- inbuf
.len
: msg_bytes
;
1516 raw_read_buf(ibuf
+ inbuf
.len
, len
);
1519 if (!(msg_bytes
-= len
) && !ibuf
[inbuf
.len
-1])
1520 inbuf
.len
--, add_null
= 1;
1521 if (iconvbufs(ic_send
, &inbuf
, &outbuf
, flags
) < 0) {
1524 /* Buffer ended with an incomplete char, so move the
1525 * bytes to the start of the buffer and continue. */
1526 memmove(ibuf
, ibuf
+ inbuf
.pos
, inbuf
.len
);
1531 if (outbuf
.len
== outbuf
.size
)
1533 outbuf
.buf
[outbuf
.len
++] = '\0';
1535 msg_bytes
= outbuf
.len
;
1538 raw_read_buf(data
, msg_bytes
);
1539 iobuf
.in_multiplexed
= 1;
1540 /* A directory name was sent with the trailing null */
1541 if (msg_bytes
> 0 && !data
[msg_bytes
-1])
1542 log_delete(data
, S_IFDIR
);
1544 data
[msg_bytes
] = '\0';
1545 log_delete(data
, S_IFREG
);
1549 if (msg_bytes
!= 4) {
1551 rprintf(FERROR
, "invalid multi-message %d:%lu [%s%s]\n",
1552 tag
, (unsigned long)msg_bytes
, who_am_i(),
1553 inc_recurse
? "/inc" : "");
1554 exit_cleanup(RERR_STREAMIO
);
1556 val
= raw_read_int();
1557 iobuf
.in_multiplexed
= 1;
1559 got_flist_entry_status(FES_SUCCESS
, val
);
1561 successful_send(val
);
1566 val
= raw_read_int();
1567 iobuf
.in_multiplexed
= 1;
1569 got_flist_entry_status(FES_NO_SEND
, val
);
1571 send_msg_int(MSG_NO_SEND
, val
);
1573 case MSG_ERROR_SOCKET
:
1574 case MSG_ERROR_UTF8
:
1579 if (tag
== MSG_ERROR_SOCKET
)
1584 case MSG_ERROR_XFER
:
1586 if (msg_bytes
>= sizeof data
) {
1589 "multiplexing overflow %d:%lu [%s%s]\n",
1590 tag
, (unsigned long)msg_bytes
, who_am_i(),
1591 inc_recurse
? "/inc" : "");
1592 exit_cleanup(RERR_STREAMIO
);
1594 raw_read_buf(data
, msg_bytes
);
1595 /* We don't set in_multiplexed value back to 1 before writing this message
1596 * because the write might loop back and read yet another message, over and
1597 * over again, while waiting for room to put the message in the msg buffer. */
1598 rwrite((enum logcode
)tag
, data
, msg_bytes
, !am_generator
);
1599 iobuf
.in_multiplexed
= 1;
1600 if (first_message
) {
1601 if (list_only
&& !am_sender
&& tag
== 1 && msg_bytes
< sizeof data
) {
1602 data
[msg_bytes
] = '\0';
1603 check_for_d_option_error(data
);
1608 case MSG_ERROR_EXIT
:
1610 val
= raw_read_int();
1611 else if (msg_bytes
== 0)
1615 iobuf
.in_multiplexed
= 1;
1616 if (DEBUG_GTE(EXIT
, 3))
1617 rprintf(FINFO
, "[%s] got MSG_ERROR_EXIT with %ld bytes\n", who_am_i(), (long)msg_bytes
);
1618 if (msg_bytes
== 0) {
1619 if (!am_sender
&& !am_generator
) {
1620 if (DEBUG_GTE(EXIT
, 3)) {
1621 rprintf(FINFO
, "[%s] sending MSG_ERROR_EXIT (len 0)\n",
1624 send_msg(MSG_ERROR_EXIT
, "", 0, 0);
1625 io_flush(FULL_FLUSH
);
1627 } else if (protocol_version
>= 31) {
1628 if (am_generator
|| am_receiver
) {
1629 if (DEBUG_GTE(EXIT
, 3)) {
1630 rprintf(FINFO
, "[%s] sending MSG_ERROR_EXIT with exit_code %d\n",
1633 send_msg_int(MSG_ERROR_EXIT
, val
);
1635 if (DEBUG_GTE(EXIT
, 3)) {
1636 rprintf(FINFO
, "[%s] sending MSG_ERROR_EXIT (len 0)\n",
1639 send_msg(MSG_ERROR_EXIT
, "", 0, 0);
1642 /* Send a negative linenum so that we don't end up
1643 * with a duplicate exit message. */
1644 _exit_cleanup(val
, __FILE__
, 0 - __LINE__
);
1646 rprintf(FERROR
, "unexpected tag %d [%s%s]\n",
1647 tag
, who_am_i(), inc_recurse
? "/inc" : "");
1648 exit_cleanup(RERR_STREAMIO
);
1651 assert(iobuf
.in_multiplexed
> 0);
1654 static void drain_multiplex_messages(void)
1656 while (IN_MULTIPLEXED_AND_READY
&& iobuf
.in
.len
) {
1657 if (iobuf
.raw_input_ends_before
) {
1658 size_t raw_len
= iobuf
.raw_input_ends_before
- iobuf
.in
.pos
;
1659 iobuf
.raw_input_ends_before
= 0;
1660 if (raw_len
>= iobuf
.in
.len
) {
1664 iobuf
.in
.len
-= raw_len
;
1665 if ((iobuf
.in
.pos
+= raw_len
) >= iobuf
.in
.size
)
1666 iobuf
.in
.pos
-= iobuf
.in
.size
;
1672 void wait_for_receiver(void)
1674 if (!iobuf
.raw_input_ends_before
)
1677 if (iobuf
.raw_input_ends_before
) {
1678 int ndx
= read_int(iobuf
.in_fd
);
1683 if (DEBUG_GTE(FLIST
, 3))
1684 rprintf(FINFO
, "[%s] flist_eof=1\n", who_am_i());
1690 exit_cleanup(RERR_STREAMIO
);
1693 struct file_list
*flist
;
1694 flist_receiving_enabled
= False
;
1695 if (DEBUG_GTE(FLIST
, 2)) {
1696 rprintf(FINFO
, "[%s] receiving flist for dir %d\n",
1699 flist
= recv_file_list(iobuf
.in_fd
, ndx
);
1700 flist
->parent_ndx
= ndx
;
1701 #ifdef SUPPORT_HARD_LINKS
1702 if (preserve_hard_links
)
1703 match_hard_links(flist
);
1705 flist_receiving_enabled
= True
;
1710 unsigned short read_shortint(int f
)
1714 return (UVAL(b
, 1) << 8) + UVAL(b
, 0);
1717 int32
read_int(int f
)
1724 #if SIZEOF_INT32 > 4
1725 if (num
& (int32
)0x80000000)
1726 num
|= ~(int32
)0xffffffff;
1731 int32
read_varint(int f
)
1742 extra
= int_byte_extra
[ch
/ 4];
1744 uchar bit
= ((uchar
)1<<(8-extra
));
1745 if (extra
>= (int)sizeof u
.b
) {
1746 rprintf(FERROR
, "Overflow in read_varint()\n");
1747 exit_cleanup(RERR_STREAMIO
);
1749 read_buf(f
, u
.b
, extra
);
1750 u
.b
[extra
] = ch
& (bit
-1);
1753 #if CAREFUL_ALIGNMENT
1756 #if SIZEOF_INT32 > 4
1757 if (u
.x
& (int32
)0x80000000)
1758 u
.x
|= ~(int32
)0xffffffff;
1763 int64
read_varlong(int f
, uchar min_bytes
)
1772 #if SIZEOF_INT64 < 8
1777 read_buf(f
, b2
, min_bytes
);
1778 memcpy(u
.b
, b2
+1, min_bytes
-1);
1779 extra
= int_byte_extra
[CVAL(b2
, 0) / 4];
1781 uchar bit
= ((uchar
)1<<(8-extra
));
1782 if (min_bytes
+ extra
> (int)sizeof u
.b
) {
1783 rprintf(FERROR
, "Overflow in read_varlong()\n");
1784 exit_cleanup(RERR_STREAMIO
);
1786 read_buf(f
, u
.b
+ min_bytes
- 1, extra
);
1787 u
.b
[min_bytes
+ extra
- 1] = CVAL(b2
, 0) & (bit
-1);
1788 #if SIZEOF_INT64 < 8
1789 if (min_bytes
+ extra
> 5 || u
.b
[4] || CVAL(u
.b
,3) & 0x80) {
1790 rprintf(FERROR
, "Integer overflow: attempted 64-bit offset\n");
1791 exit_cleanup(RERR_UNSUPPORTED
);
1795 u
.b
[min_bytes
+ extra
- 1] = CVAL(b2
, 0);
1796 #if SIZEOF_INT64 < 8
1798 #elif CAREFUL_ALIGNMENT
1799 u
.x
= IVAL64(u
.b
,0);
1804 int64
read_longint(int f
)
1806 #if SIZEOF_INT64 >= 8
1809 int32 num
= read_int(f
);
1811 if (num
!= (int32
)0xffffffff)
1814 #if SIZEOF_INT64 < 8
1815 rprintf(FERROR
, "Integer overflow: attempted 64-bit offset\n");
1816 exit_cleanup(RERR_UNSUPPORTED
);
1819 return IVAL(b
,0) | (((int64
)IVAL(b
,4))<<32);
1823 void read_buf(int f
, char *buf
, size_t len
)
1825 if (f
!= iobuf
.in_fd
) {
1826 if (safe_read(f
, buf
, len
) != len
)
1827 whine_about_eof(False
); /* Doesn't return. */
1831 if (!IN_MULTIPLEXED
) {
1832 raw_read_buf(buf
, len
);
1833 total_data_read
+= len
;
1834 if (forward_flist_data
)
1835 write_buf(iobuf
.out_fd
, buf
, len
);
1837 if (f
== write_batch_monitor_in
)
1838 safe_write(batch_fd
, buf
, len
);
1845 while (!iobuf
.raw_input_ends_before
)
1848 siz
= MIN(len
, iobuf
.raw_input_ends_before
- iobuf
.in
.pos
);
1849 if (siz
>= iobuf
.in
.size
)
1850 siz
= iobuf
.in
.size
;
1851 raw_read_buf(buf
, siz
);
1852 total_data_read
+= siz
;
1854 if (forward_flist_data
)
1855 write_buf(iobuf
.out_fd
, buf
, siz
);
1857 if (f
== write_batch_monitor_in
)
1858 safe_write(batch_fd
, buf
, siz
);
1860 if ((len
-= siz
) == 0)
1866 void read_sbuf(int f
, char *buf
, size_t len
)
1868 read_buf(f
, buf
, len
);
1872 uchar
read_byte(int f
)
1875 read_buf(f
, (char*)&c
, 1);
1879 int read_vstring(int f
, char *buf
, int bufsize
)
1881 int len
= read_byte(f
);
1884 len
= (len
& ~0x80) * 0x100 + read_byte(f
);
1886 if (len
>= bufsize
) {
1887 rprintf(FERROR
, "over-long vstring received (%d > %d)\n",
1893 read_buf(f
, buf
, len
);
1898 /* Populate a sum_struct with values from the socket. This is
1899 * called by both the sender and the receiver. */
1900 void read_sum_head(int f
, struct sum_struct
*sum
)
1902 int32 max_blength
= protocol_version
< 30 ? OLD_MAX_BLOCK_SIZE
: MAX_BLOCK_SIZE
;
1903 sum
->count
= read_int(f
);
1904 if (sum
->count
< 0) {
1905 rprintf(FERROR
, "Invalid checksum count %ld [%s]\n",
1906 (long)sum
->count
, who_am_i());
1907 exit_cleanup(RERR_PROTOCOL
);
1909 sum
->blength
= read_int(f
);
1910 if (sum
->blength
< 0 || sum
->blength
> max_blength
) {
1911 rprintf(FERROR
, "Invalid block length %ld [%s]\n",
1912 (long)sum
->blength
, who_am_i());
1913 exit_cleanup(RERR_PROTOCOL
);
1915 sum
->s2length
= protocol_version
< 27 ? csum_length
: (int)read_int(f
);
1916 if (sum
->s2length
< 0 || sum
->s2length
> MAX_DIGEST_LEN
) {
1917 rprintf(FERROR
, "Invalid checksum length %d [%s]\n",
1918 sum
->s2length
, who_am_i());
1919 exit_cleanup(RERR_PROTOCOL
);
1921 sum
->remainder
= read_int(f
);
1922 if (sum
->remainder
< 0 || sum
->remainder
> sum
->blength
) {
1923 rprintf(FERROR
, "Invalid remainder length %ld [%s]\n",
1924 (long)sum
->remainder
, who_am_i());
1925 exit_cleanup(RERR_PROTOCOL
);
1929 /* Send the values from a sum_struct over the socket. Set sum to
1930 * NULL if there are no checksums to send. This is called by both
1931 * the generator and the sender. */
1932 void write_sum_head(int f
, struct sum_struct
*sum
)
1934 static struct sum_struct null_sum
;
1939 write_int(f
, sum
->count
);
1940 write_int(f
, sum
->blength
);
1941 if (protocol_version
>= 27)
1942 write_int(f
, sum
->s2length
);
1943 write_int(f
, sum
->remainder
);
1946 /* Sleep after writing to limit I/O bandwidth usage.
1948 * @todo Rather than sleeping after each write, it might be better to
1949 * use some kind of averaging. The current algorithm seems to always
1950 * use a bit less bandwidth than specified, because it doesn't make up
1951 * for slow periods. But arguably this is a feature. In addition, we
1952 * ought to take the time used to write the data into account.
1954 * During some phases of big transfers (file FOO is uptodate) this is
1955 * called with a small bytes_written every time. As the kernel has to
1956 * round small waits up to guarantee that we actually wait at least the
1957 * requested number of microseconds, this can become grossly inaccurate.
1958 * We therefore keep track of the bytes we've written over time and only
1959 * sleep when the accumulated delay is at least 1 tenth of a second. */
1960 static void sleep_for_bwlimit(int bytes_written
)
1962 static struct timeval prior_tv
;
1963 static long total_written
= 0;
1964 struct timeval tv
, start_tv
;
1965 long elapsed_usec
, sleep_usec
;
1967 #define ONE_SEC 1000000L /* # of microseconds in a second */
1969 total_written
+= bytes_written
;
1971 gettimeofday(&start_tv
, NULL
);
1972 if (prior_tv
.tv_sec
) {
1973 elapsed_usec
= (start_tv
.tv_sec
- prior_tv
.tv_sec
) * ONE_SEC
1974 + (start_tv
.tv_usec
- prior_tv
.tv_usec
);
1975 total_written
-= (int64
)elapsed_usec
* bwlimit
/ (ONE_SEC
/1024);
1976 if (total_written
< 0)
1980 sleep_usec
= total_written
* (ONE_SEC
/1024) / bwlimit
;
1981 if (sleep_usec
< ONE_SEC
/ 10) {
1982 prior_tv
= start_tv
;
1986 tv
.tv_sec
= sleep_usec
/ ONE_SEC
;
1987 tv
.tv_usec
= sleep_usec
% ONE_SEC
;
1988 select(0, NULL
, NULL
, NULL
, &tv
);
1990 gettimeofday(&prior_tv
, NULL
);
1991 elapsed_usec
= (prior_tv
.tv_sec
- start_tv
.tv_sec
) * ONE_SEC
1992 + (prior_tv
.tv_usec
- start_tv
.tv_usec
);
1993 total_written
= (sleep_usec
- elapsed_usec
) * bwlimit
/ (ONE_SEC
/1024);
1996 void io_flush(int flush_type
)
1998 if (iobuf
.out
.len
> iobuf
.out_empty_len
) {
1999 if (flush_type
== FULL_FLUSH
) /* flush everything in the output buffers */
2000 perform_io(iobuf
.out
.size
- iobuf
.out_empty_len
, PIO_NEED_OUTROOM
);
2001 else if (flush_type
== NORMAL_FLUSH
) /* flush at least 1 byte */
2002 perform_io(iobuf
.out
.size
- iobuf
.out
.len
+ 1, PIO_NEED_OUTROOM
);
2003 /* MSG_FLUSH: flush iobuf.msg only */
2006 perform_io(iobuf
.msg
.size
, PIO_NEED_MSGROOM
);
2009 void write_shortint(int f
, unsigned short x
)
2013 b
[1] = (char)(x
>> 8);
2017 void write_int(int f
, int32 x
)
2024 void write_varint(int f
, int32 x
)
2032 for (cnt
= 4; cnt
> 1 && b
[cnt
] == 0; cnt
--) {}
2033 bit
= ((uchar
)1<<(7-cnt
+1));
2035 if (CVAL(b
, cnt
) >= bit
) {
2039 *b
= b
[cnt
] | ~(bit
*2-1);
2043 write_buf(f
, b
, cnt
);
2046 void write_varlong(int f
, int64 x
, uchar min_bytes
)
2052 #if SIZEOF_INT64 >= 8
2056 if (x
<= 0x7FFFFFFF && x
>= 0)
2057 memset(b
+ 5, 0, 4);
2059 rprintf(FERROR
, "Integer overflow: attempted 64-bit offset\n");
2060 exit_cleanup(RERR_UNSUPPORTED
);
2064 while (cnt
> min_bytes
&& b
[cnt
] == 0)
2066 bit
= ((uchar
)1<<(7-cnt
+min_bytes
));
2067 if (CVAL(b
, cnt
) >= bit
) {
2070 } else if (cnt
> min_bytes
)
2071 *b
= b
[cnt
] | ~(bit
*2-1);
2075 write_buf(f
, b
, cnt
);
2079 * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
2080 * 64-bit types on this platform.
2082 void write_longint(int f
, int64 x
)
2084 char b
[12], * const s
= b
+4;
2087 if (x
<= 0x7FFFFFFF && x
>= 0) {
2092 #if SIZEOF_INT64 < 8
2093 rprintf(FERROR
, "Integer overflow: attempted 64-bit offset\n");
2094 exit_cleanup(RERR_UNSUPPORTED
);
2097 SIVAL(s
, 4, x
>> 32);
2098 write_buf(f
, b
, 12);
2102 void write_bigbuf(int f
, const char *buf
, size_t len
)
2104 size_t half_max
= (iobuf
.out
.size
- iobuf
.out_empty_len
) / 2;
2106 while (len
> half_max
+ 1024) {
2107 write_buf(f
, buf
, half_max
);
2112 write_buf(f
, buf
, len
);
2115 void write_buf(int f
, const char *buf
, size_t len
)
2119 if (f
!= iobuf
.out_fd
) {
2120 safe_write(f
, buf
, len
);
2124 if (iobuf
.out
.len
+ len
> iobuf
.out
.size
)
2125 perform_io(len
, PIO_NEED_OUTROOM
);
2127 pos
= iobuf
.out
.pos
+ iobuf
.out
.len
; /* Must be set after any flushing. */
2128 if (pos
>= iobuf
.out
.size
)
2129 pos
-= iobuf
.out
.size
;
2131 /* Handle a split copy if we wrap around the end of the circular buffer. */
2132 if (pos
>= iobuf
.out
.pos
&& (siz
= iobuf
.out
.size
- pos
) < len
) {
2133 memcpy(iobuf
.out
.buf
+ pos
, buf
, siz
);
2134 memcpy(iobuf
.out
.buf
, buf
+ siz
, len
- siz
);
2136 memcpy(iobuf
.out
.buf
+ pos
, buf
, len
);
2138 iobuf
.out
.len
+= len
;
2139 total_data_written
+= len
;
2142 if (f
== write_batch_monitor_out
)
2143 safe_write(batch_fd
, buf
, len
);
2146 /* Write a string to the connection */
2147 void write_sbuf(int f
, const char *buf
)
2149 write_buf(f
, buf
, strlen(buf
));
2152 void write_byte(int f
, uchar c
)
2154 write_buf(f
, (char *)&c
, 1);
2157 void write_vstring(int f
, const char *str
, int len
)
2159 uchar lenbuf
[3], *lb
= lenbuf
;
2164 "attempting to send over-long vstring (%d > %d)\n",
2166 exit_cleanup(RERR_PROTOCOL
);
2168 *lb
++ = len
/ 0x100 + 0x80;
2172 write_buf(f
, (char*)lenbuf
, lb
- lenbuf
+ 1);
2174 write_buf(f
, str
, len
);
2177 /* Send a file-list index using a byte-reduction method. */
2178 void write_ndx(int f
, int32 ndx
)
2180 static int32 prev_positive
= -1, prev_negative
= 1;
2181 int32 diff
, cnt
= 0;
2184 if (protocol_version
< 30 || read_batch
) {
2189 /* Send NDX_DONE as a single-byte 0 with no side effects. Send
2190 * negative nums as a positive after sending a leading 0xFF. */
2192 diff
= ndx
- prev_positive
;
2193 prev_positive
= ndx
;
2194 } else if (ndx
== NDX_DONE
) {
2199 b
[cnt
++] = (char)0xFF;
2201 diff
= ndx
- prev_negative
;
2202 prev_negative
= ndx
;
2205 /* A diff of 1 - 253 is sent as a one-byte diff; a diff of 254 - 32767
2206 * or 0 is sent as a 0xFE + a two-byte diff; otherwise we send 0xFE
2207 * & all 4 bytes of the (non-negative) num with the high-bit set. */
2208 if (diff
< 0xFE && diff
> 0)
2209 b
[cnt
++] = (char)diff
;
2210 else if (diff
< 0 || diff
> 0x7FFF) {
2211 b
[cnt
++] = (char)0xFE;
2212 b
[cnt
++] = (char)((ndx
>> 24) | 0x80);
2213 b
[cnt
++] = (char)ndx
;
2214 b
[cnt
++] = (char)(ndx
>> 8);
2215 b
[cnt
++] = (char)(ndx
>> 16);
2217 b
[cnt
++] = (char)0xFE;
2218 b
[cnt
++] = (char)(diff
>> 8);
2219 b
[cnt
++] = (char)diff
;
2221 write_buf(f
, b
, cnt
);
2224 /* Receive a file-list index using a byte-reduction method. */
2225 int32
read_ndx(int f
)
2227 static int32 prev_positive
= -1, prev_negative
= 1;
2228 int32
*prev_ptr
, num
;
2231 if (protocol_version
< 30)
2235 if (CVAL(b
, 0) == 0xFF) {
2237 prev_ptr
= &prev_negative
;
2238 } else if (CVAL(b
, 0) == 0)
2241 prev_ptr
= &prev_positive
;
2242 if (CVAL(b
, 0) == 0xFE) {
2244 if (CVAL(b
, 0) & 0x80) {
2245 b
[3] = CVAL(b
, 0) & ~0x80;
2247 read_buf(f
, b
+1, 2);
2250 num
= (UVAL(b
,0)<<8) + UVAL(b
,1) + *prev_ptr
;
2252 num
= UVAL(b
, 0) + *prev_ptr
;
2254 if (prev_ptr
== &prev_negative
)
2259 /* Read a line of up to bufsiz-1 characters into buf. Strips
2260 * the (required) trailing newline and all carriage returns.
2261 * Returns 1 for success; 0 for I/O error or truncation. */
2262 int read_line_old(int fd
, char *buf
, size_t bufsiz
, int eof_ok
)
2264 assert(fd
!= iobuf
.in_fd
);
2265 bufsiz
--; /* leave room for the null */
2266 while (bufsiz
> 0) {
2267 if (safe_read(fd
, buf
, 1) == 0) {
2285 void io_printf(int fd
, const char *format
, ...)
2288 char buf
[BIGPATHBUFLEN
];
2291 va_start(ap
, format
);
2292 len
= vsnprintf(buf
, sizeof buf
, format
, ap
);
2296 exit_cleanup(RERR_PROTOCOL
);
2298 if (len
>= (int)sizeof buf
) {
2299 rprintf(FERROR
, "io_printf() was too long for the buffer.\n");
2300 exit_cleanup(RERR_PROTOCOL
);
2303 write_sbuf(fd
, buf
);
2306 /* Setup for multiplexing a MSG_* stream with the data stream. */
2307 void io_start_multiplex_out(int fd
)
2309 io_flush(FULL_FLUSH
);
2311 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2))
2312 rprintf(FINFO
, "[%s] io_start_multiplex_out(%d)\n", who_am_i(), fd
);
2315 alloc_xbuf(&iobuf
.msg
, ROUND_UP_1024(IO_BUFFER_SIZE
));
2317 iobuf
.out_empty_len
= 4; /* See also OUT_MULTIPLEXED */
2318 io_start_buffering_out(fd
);
2319 got_kill_signal
= 0;
2321 iobuf
.raw_data_header_pos
= iobuf
.out
.pos
+ iobuf
.out
.len
;
2325 /* Setup for multiplexing a MSG_* stream with the data stream. */
2326 void io_start_multiplex_in(int fd
)
2328 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2))
2329 rprintf(FINFO
, "[%s] io_start_multiplex_in(%d)\n", who_am_i(), fd
);
2331 iobuf
.in_multiplexed
= 1; /* See also IN_MULTIPLEXED */
2332 io_start_buffering_in(fd
);
2335 int io_end_multiplex_in(int mode
)
2337 int ret
= iobuf
.in_multiplexed
? iobuf
.in_fd
: -1;
2339 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2))
2340 rprintf(FINFO
, "[%s] io_end_multiplex_in(mode=%d)\n", who_am_i(), mode
);
2342 iobuf
.in_multiplexed
= 0;
2343 if (mode
== MPLX_SWITCHING
)
2344 iobuf
.raw_input_ends_before
= 0;
2346 assert(iobuf
.raw_input_ends_before
== 0);
2347 if (mode
!= MPLX_TO_BUFFERED
)
2348 io_end_buffering_in(mode
);
2353 int io_end_multiplex_out(int mode
)
2355 int ret
= iobuf
.out_empty_len
? iobuf
.out_fd
: -1;
2357 if (msgs2stderr
== 1 && DEBUG_GTE(IO
, 2))
2358 rprintf(FINFO
, "[%s] io_end_multiplex_out(mode=%d)\n", who_am_i(), mode
);
2360 if (mode
!= MPLX_TO_BUFFERED
)
2361 io_end_buffering_out(mode
);
2363 io_flush(FULL_FLUSH
);
2366 iobuf
.out_empty_len
= 0;
2367 if (got_kill_signal
> 0) /* Just in case... */
2368 handle_kill_signal(False
);
2369 got_kill_signal
= -1;
2374 void start_write_batch(int fd
)
2376 /* Some communication has already taken place, but we don't
2377 * enable batch writing until here so that we can write a
2378 * canonical record of the communication even though the
2379 * actual communication so far depends on whether a daemon
2381 write_int(batch_fd
, protocol_version
);
2382 if (protocol_version
>= 30)
2383 write_varint(batch_fd
, compat_flags
);
2384 write_int(batch_fd
, checksum_seed
);
2387 write_batch_monitor_out
= fd
;
2389 write_batch_monitor_in
= fd
;
2392 void stop_write_batch(void)
2394 write_batch_monitor_out
= -1;
2395 write_batch_monitor_in
= -1;