- Made flist_find() and receive_file_entry() static functions.
[rsync.git] / io.c
1 /* -*- c-file-style: "linux" -*-
2  *
3  * Copyright (C) 1996-2001 by Andrew Tridgell
4  * Copyright (C) Paul Mackerras 1996
5  * Copyright (C) 2001, 2002 by Martin Pool <mbp@samba.org>
6  *
7  * This program is free software; you can redistribute it and/or modify
8  * it under the terms of the GNU General Public License as published by
9  * the Free Software Foundation; either version 2 of the License, or
10  * (at your option) any later version.
11  *
12  * This program is distributed in the hope that it will be useful,
13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15  * GNU General Public License for more details.
16  *
17  * You should have received a copy of the GNU General Public License
18  * along with this program; if not, write to the Free Software
19  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
20  */
21
22 /**
23  * @file io.c
24  *
25  * Socket and pipe I/O utilities used in rsync.
26  *
27  * rsync provides its own multiplexing system, which is used to send
28  * stderr and stdout over a single socket.  We need this because
29  * stdout normally carries the binary data stream, and stderr all our
30  * error messages.
31  *
32  * For historical reasons this is off during the start of the
33  * connection, but it's switched on quite early using
34  * io_start_multiplex_out() and io_start_multiplex_in().
35  **/
36
37 #include "rsync.h"
38
39 /** If no timeout is specified then use a 60 second select timeout */
40 #define SELECT_TIMEOUT 60
41
42 extern int bwlimit;
43 extern size_t bwlimit_writemax;
44 extern int verbose;
45 extern int io_timeout;
46 extern int am_server;
47 extern int am_daemon;
48 extern int am_sender;
49 extern int am_generator;
50 extern int eol_nulls;
51 extern int csum_length;
52 extern int checksum_seed;
53 extern int protocol_version;
54 extern char *remote_filesfrom_file;
55 extern struct stats stats;
56
57 const char phase_unknown[] = "unknown";
58 int select_timeout = SELECT_TIMEOUT;
59 int batch_fd = -1;
60 int batch_gen_fd = -1;
61
62 /**
63  * The connection might be dropped at some point; perhaps because the
64  * remote instance crashed.  Just giving the offset on the stream is
65  * not very helpful.  So instead we try to make io_phase_name point to
66  * something useful.
67  *
68  * For buffered/multiplexed I/O these names will be somewhat
69  * approximate; perhaps for ease of support we would rather make the
70  * buffer always flush when a single application-level I/O finishes.
71  *
72  * @todo Perhaps we want some simple stack functionality, but there's
73  * no need to overdo it.
74  **/
75 const char *io_write_phase = phase_unknown;
76 const char *io_read_phase = phase_unknown;
77
78 /** Ignore EOF errors while reading a module listing if the remote
79     version is 24 or less. */
80 int kludge_around_eof = False;
81
82 int msg_fd_in = -1;
83 int msg_fd_out = -1;
84
85 static int io_multiplexing_out;
86 static int io_multiplexing_in;
87 static int sock_f_in = -1;
88 static int sock_f_out = -1;
89 static time_t last_io;
90 static int no_flush;
91
92 static int write_batch_monitor_in = -1;
93 static int write_batch_monitor_out = -1;
94
95 static int io_filesfrom_f_in = -1;
96 static int io_filesfrom_f_out = -1;
97 static char io_filesfrom_buf[2048];
98 static char *io_filesfrom_bp;
99 static char io_filesfrom_lastchar;
100 static int io_filesfrom_buflen;
101
102 static void read_loop(int fd, char *buf, size_t len);
103
104 struct redo_list {
105         struct redo_list *next;
106         int num;
107 };
108
109 static struct redo_list *redo_list_head;
110 static struct redo_list *redo_list_tail;
111
112 struct msg_list {
113         struct msg_list *next;
114         char *buf;
115         int len;
116 };
117
118 static struct msg_list *msg_list_head;
119 static struct msg_list *msg_list_tail;
120
121 static void redo_list_add(int num)
122 {
123         struct redo_list *rl;
124
125         if (!(rl = new(struct redo_list)))
126                 exit_cleanup(RERR_MALLOC);
127         rl->next = NULL;
128         rl->num = num;
129         if (redo_list_tail)
130                 redo_list_tail->next = rl;
131         else
132                 redo_list_head = rl;
133         redo_list_tail = rl;
134 }
135
136 static void check_timeout(void)
137 {
138         time_t t;
139
140         if (!io_timeout)
141                 return;
142
143         if (!last_io) {
144                 last_io = time(NULL);
145                 return;
146         }
147
148         t = time(NULL);
149
150         if (t - last_io >= io_timeout) {
151                 if (!am_server && !am_daemon) {
152                         rprintf(FERROR, "io timeout after %d seconds -- exiting\n",
153                                 (int)(t-last_io));
154                 }
155                 exit_cleanup(RERR_TIMEOUT);
156         }
157 }
158
159 /* Note the fds used for the main socket (which might really be a pipe
160  * for a local transfer, but we can ignore that). */
161 void io_set_sock_fds(int f_in, int f_out)
162 {
163         sock_f_in = f_in;
164         sock_f_out = f_out;
165 }
166
167 /* Setup the fd used to receive MSG_* messages.  Only needed during the
168  * early stages of being a local sender (up through the sending of the
169  * file list) or when we're the generator (to fetch the messages from
170  * the receiver). */
171 void set_msg_fd_in(int fd)
172 {
173         msg_fd_in = fd;
174 }
175
176 /* Setup the fd used to send our MSG_* messages.  Only needed when
177  * we're the receiver (to send our messages to the generator). */
178 void set_msg_fd_out(int fd)
179 {
180         msg_fd_out = fd;
181         set_nonblocking(msg_fd_out);
182 }
183
184 /* Add a message to the pending MSG_* list. */
185 static void msg_list_add(int code, char *buf, int len)
186 {
187         struct msg_list *ml;
188
189         if (!(ml = new(struct msg_list)))
190                 exit_cleanup(RERR_MALLOC);
191         ml->next = NULL;
192         if (!(ml->buf = new_array(char, len+4)))
193                 exit_cleanup(RERR_MALLOC);
194         SIVAL(ml->buf, 0, ((code+MPLEX_BASE)<<24) | len);
195         memcpy(ml->buf+4, buf, len);
196         ml->len = len+4;
197         if (msg_list_tail)
198                 msg_list_tail->next = ml;
199         else
200                 msg_list_head = ml;
201         msg_list_tail = ml;
202 }
203
204 void send_msg(enum msgcode code, char *buf, int len)
205 {
206         msg_list_add(code, buf, len);
207         msg_list_push(NORMAL_FLUSH);
208 }
209
210 /* Read a message from the MSG_* fd and handle it.  This is called either
211  * during the early stages of being a local sender (up through the sending
212  * of the file list) or when we're the generator (to fetch the messages
213  * from the receiver). */
214 static void read_msg_fd(void)
215 {
216         char buf[2048];
217         size_t n;
218         int fd = msg_fd_in;
219         int tag, len;
220
221         /* Temporarily disable msg_fd_in.  This is needed to avoid looping back
222          * to this routine from read_timeout() and writefd_unbuffered(). */
223         msg_fd_in = -1;
224
225         read_loop(fd, buf, 4);
226         tag = IVAL(buf, 0);
227
228         len = tag & 0xFFFFFF;
229         tag = (tag >> 24) - MPLEX_BASE;
230
231         switch (tag) {
232         case MSG_DONE:
233                 if (len != 0 || !am_generator) {
234                         rprintf(FERROR, "invalid message %d:%d\n", tag, len);
235                         exit_cleanup(RERR_STREAMIO);
236                 }
237                 redo_list_add(-1);
238                 break;
239         case MSG_REDO:
240                 if (len != 4 || !am_generator) {
241                         rprintf(FERROR, "invalid message %d:%d\n", tag, len);
242                         exit_cleanup(RERR_STREAMIO);
243                 }
244                 read_loop(fd, buf, 4);
245                 redo_list_add(IVAL(buf,0));
246                 break;
247         case MSG_INFO:
248         case MSG_ERROR:
249         case MSG_LOG:
250                 while (len) {
251                         n = len;
252                         if (n >= sizeof buf)
253                                 n = sizeof buf - 1;
254                         read_loop(fd, buf, n);
255                         rwrite((enum logcode)tag, buf, n);
256                         len -= n;
257                 }
258                 break;
259         default:
260                 rprintf(FERROR, "unknown message %d:%d\n", tag, len);
261                 exit_cleanup(RERR_STREAMIO);
262         }
263
264         msg_fd_in = fd;
265 }
266
267 /* Try to push messages off the list onto the wire.  If we leave with more
268  * to do, return 0.  On error, return -1.  If everything flushed, return 1.
269  * This is only active in the receiver. */
270 int msg_list_push(int flush_it_all)
271 {
272         static int written = 0;
273         struct timeval tv;
274         fd_set fds;
275
276         if (msg_fd_out < 0)
277                 return -1;
278
279         while (msg_list_head) {
280                 struct msg_list *ml = msg_list_head;
281                 int n = write(msg_fd_out, ml->buf + written, ml->len - written);
282                 if (n < 0) {
283                         if (errno == EINTR)
284                                 continue;
285                         if (errno != EWOULDBLOCK && errno != EAGAIN)
286                                 return -1;
287                         if (!flush_it_all)
288                                 return 0;
289                         FD_ZERO(&fds);
290                         FD_SET(msg_fd_out, &fds);
291                         tv.tv_sec = select_timeout;
292                         tv.tv_usec = 0;
293                         if (!select(msg_fd_out+1, NULL, &fds, NULL, &tv))
294                                 check_timeout();
295                 } else if ((written += n) == ml->len) {
296                         free(ml->buf);
297                         msg_list_head = ml->next;
298                         if (!msg_list_head)
299                                 msg_list_tail = NULL;
300                         free(ml);
301                         written = 0;
302                 }
303         }
304         return 1;
305 }
306
307 int get_redo_num(void)
308 {
309         struct redo_list *next;
310         int num;
311
312         while (!redo_list_head)
313                 read_msg_fd();
314
315         num = redo_list_head->num;
316         next = redo_list_head->next;
317         free(redo_list_head);
318         redo_list_head = next;
319         if (!next)
320                 redo_list_tail = NULL;
321
322         return num;
323 }
324
325 /**
326  * When we're the receiver and we have a local --files-from list of names
327  * that needs to be sent over the socket to the sender, we have to do two
328  * things at the same time: send the sender a list of what files we're
329  * processing and read the incoming file+info list from the sender.  We do
330  * this by augmenting the read_timeout() function to copy this data.  It
331  * uses the io_filesfrom_buf to read a block of data from f_in (when it is
332  * ready, since it might be a pipe) and then blast it out f_out (when it
333  * is ready to receive more data).
334  */
335 void io_set_filesfrom_fds(int f_in, int f_out)
336 {
337         io_filesfrom_f_in = f_in;
338         io_filesfrom_f_out = f_out;
339         io_filesfrom_bp = io_filesfrom_buf;
340         io_filesfrom_lastchar = '\0';
341         io_filesfrom_buflen = 0;
342 }
343
344 /**
345  * It's almost always an error to get an EOF when we're trying to read
346  * from the network, because the protocol is self-terminating.
347  *
348  * However, there is one unfortunate cases where it is not, which is
349  * rsync <2.4.6 sending a list of modules on a server, since the list
350  * is terminated by closing the socket. So, for the section of the
351  * program where that is a problem (start_socket_client),
352  * kludge_around_eof is True and we just exit.
353  */
354 static void whine_about_eof(int fd)
355 {
356         if (kludge_around_eof && fd == sock_f_in)
357                 exit_cleanup(0);
358
359         rprintf(FERROR, RSYNC_NAME ": connection unexpectedly closed "
360                 "(%.0f bytes received so far) [%s]\n",
361                 (double)stats.total_read, who_am_i());
362
363         exit_cleanup(RERR_STREAMIO);
364 }
365
366
367 /**
368  * Read from a socket with I/O timeout. return the number of bytes
369  * read. If no bytes can be read then exit, never return a number <= 0.
370  *
371  * TODO: If the remote shell connection fails, then current versions
372  * actually report an "unexpected EOF" error here.  Since it's a
373  * fairly common mistake to try to use rsh when ssh is required, we
374  * should trap that: if we fail to read any data at all, we should
375  * give a better explanation.  We can tell whether the connection has
376  * started by looking e.g. at whether the remote version is known yet.
377  */
378 static int read_timeout(int fd, char *buf, size_t len)
379 {
380         int n, ret = 0;
381
382         io_flush(NORMAL_FLUSH);
383
384         while (ret == 0) {
385                 /* until we manage to read *something* */
386                 fd_set r_fds, w_fds;
387                 struct timeval tv;
388                 int maxfd = fd;
389                 int count;
390
391                 FD_ZERO(&r_fds);
392                 FD_ZERO(&w_fds);
393                 FD_SET(fd, &r_fds);
394                 if (msg_fd_in >= 0) {
395                         FD_SET(msg_fd_in, &r_fds);
396                         if (msg_fd_in > maxfd)
397                                 maxfd = msg_fd_in;
398                 } else if (msg_list_head) {
399                         FD_SET(msg_fd_out, &w_fds);
400                         if (msg_fd_out > maxfd)
401                                 maxfd = msg_fd_out;
402                 }
403                 if (io_filesfrom_f_out >= 0) {
404                         int new_fd;
405                         if (io_filesfrom_buflen == 0) {
406                                 if (io_filesfrom_f_in >= 0) {
407                                         FD_SET(io_filesfrom_f_in, &r_fds);
408                                         new_fd = io_filesfrom_f_in;
409                                 } else {
410                                         io_filesfrom_f_out = -1;
411                                         new_fd = -1;
412                                 }
413                         } else {
414                                 FD_SET(io_filesfrom_f_out, &w_fds);
415                                 new_fd = io_filesfrom_f_out;
416                         }
417                         if (new_fd > maxfd)
418                                 maxfd = new_fd;
419                 }
420
421                 tv.tv_sec = select_timeout;
422                 tv.tv_usec = 0;
423
424                 errno = 0;
425
426                 count = select(maxfd + 1, &r_fds, &w_fds, NULL, &tv);
427
428                 if (count <= 0) {
429                         if (errno == EBADF)
430                                 exit_cleanup(RERR_SOCKETIO);
431                         check_timeout();
432                         continue;
433                 }
434
435                 if (msg_fd_in >= 0 && FD_ISSET(msg_fd_in, &r_fds))
436                         read_msg_fd();
437                 else if (msg_list_head && FD_ISSET(msg_fd_out, &w_fds))
438                         msg_list_push(NORMAL_FLUSH);
439
440                 if (io_filesfrom_f_out >= 0) {
441                         if (io_filesfrom_buflen) {
442                                 if (FD_ISSET(io_filesfrom_f_out, &w_fds)) {
443                                         int l = write(io_filesfrom_f_out,
444                                                       io_filesfrom_bp,
445                                                       io_filesfrom_buflen);
446                                         if (l > 0) {
447                                                 if (!(io_filesfrom_buflen -= l))
448                                                         io_filesfrom_bp = io_filesfrom_buf;
449                                                 else
450                                                         io_filesfrom_bp += l;
451                                         } else {
452                                                 /* XXX should we complain? */
453                                                 io_filesfrom_f_out = -1;
454                                         }
455                                 }
456                         } else if (io_filesfrom_f_in >= 0) {
457                                 if (FD_ISSET(io_filesfrom_f_in, &r_fds)) {
458                                         int l = read(io_filesfrom_f_in,
459                                                      io_filesfrom_buf,
460                                                      sizeof io_filesfrom_buf);
461                                         if (l <= 0) {
462                                                 /* Send end-of-file marker */
463                                                 io_filesfrom_buf[0] = '\0';
464                                                 io_filesfrom_buf[1] = '\0';
465                                                 io_filesfrom_buflen = io_filesfrom_lastchar? 2 : 1;
466                                                 io_filesfrom_f_in = -1;
467                                         } else {
468                                                 if (!eol_nulls) {
469                                                         char *s = io_filesfrom_buf + l;
470                                                         /* Transform CR and/or LF into '\0' */
471                                                         while (s-- > io_filesfrom_buf) {
472                                                                 if (*s == '\n' || *s == '\r')
473                                                                         *s = '\0';
474                                                         }
475                                                 }
476                                                 if (!io_filesfrom_lastchar) {
477                                                         /* Last buf ended with a '\0', so don't
478                                                          * let this buf start with one. */
479                                                         while (l && !*io_filesfrom_bp)
480                                                                 io_filesfrom_bp++, l--;
481                                                 }
482                                                 if (!l)
483                                                         io_filesfrom_bp = io_filesfrom_buf;
484                                                 else {
485                                                         char *f = io_filesfrom_bp;
486                                                         char *t = f;
487                                                         char *eob = f + l;
488                                                         /* Eliminate any multi-'\0' runs. */
489                                                         while (f != eob) {
490                                                                 if (!(*t++ = *f++)) {
491                                                                         while (f != eob && !*f)
492                                                                                 f++, l--;
493                                                                 }
494                                                         }
495                                                         io_filesfrom_lastchar = f[-1];
496                                                 }
497                                                 io_filesfrom_buflen = l;
498                                         }
499                                 }
500                         }
501                 }
502
503                 if (!FD_ISSET(fd, &r_fds))
504                         continue;
505
506                 n = read(fd, buf, len);
507
508                 if (n <= 0) {
509                         if (n == 0)
510                                 whine_about_eof(fd); /* Doesn't return. */
511                         if (errno == EINTR || errno == EWOULDBLOCK
512                             || errno == EAGAIN)
513                                 continue;
514
515                         /* Don't write errors on a dead socket. */
516                         if (fd == sock_f_in)
517                                 close_multiplexing_out();
518                         rsyserr(FERROR, errno, "read error");
519                         exit_cleanup(RERR_STREAMIO);
520                 }
521
522                 buf += n;
523                 len -= n;
524                 ret += n;
525
526                 if (io_timeout && fd == sock_f_in)
527                         last_io = time(NULL);
528         }
529
530         return ret;
531 }
532
533 /**
534  * Read a line into the "fname" buffer (which must be at least MAXPATHLEN
535  * characters long).
536  */
537 int read_filesfrom_line(int fd, char *fname)
538 {
539         char ch, *s, *eob = fname + MAXPATHLEN - 1;
540         int cnt;
541         int reading_remotely = remote_filesfrom_file != NULL;
542         int nulls = eol_nulls || reading_remotely;
543
544   start:
545         s = fname;
546         while (1) {
547                 cnt = read(fd, &ch, 1);
548                 if (cnt < 0 && (errno == EWOULDBLOCK
549                   || errno == EINTR || errno == EAGAIN)) {
550                         struct timeval tv;
551                         fd_set fds;
552                         FD_ZERO(&fds);
553                         FD_SET(fd, &fds);
554                         tv.tv_sec = select_timeout;
555                         tv.tv_usec = 0;
556                         if (!select(fd+1, &fds, NULL, NULL, &tv))
557                                 check_timeout();
558                         continue;
559                 }
560                 if (cnt != 1)
561                         break;
562                 if (nulls? !ch : (ch == '\r' || ch == '\n')) {
563                         /* Skip empty lines if reading locally. */
564                         if (!reading_remotely && s == fname)
565                                 continue;
566                         break;
567                 }
568                 if (s < eob)
569                         *s++ = ch;
570         }
571         *s = '\0';
572
573         /* Dump comments. */
574         if (*fname == '#' || *fname == ';')
575                 goto start;
576
577         return s - fname;
578 }
579
580
581 static char *iobuf_out;
582 static int iobuf_out_cnt;
583
584 void io_start_buffering_out(void)
585 {
586         if (iobuf_out)
587                 return;
588         if (!(iobuf_out = new_array(char, IO_BUFFER_SIZE)))
589                 out_of_memory("io_start_buffering_out");
590         iobuf_out_cnt = 0;
591 }
592
593
594 static char *iobuf_in;
595 static size_t iobuf_in_siz;
596
597 void io_start_buffering_in(void)
598 {
599         if (iobuf_in)
600                 return;
601         iobuf_in_siz = 2 * IO_BUFFER_SIZE;
602         if (!(iobuf_in = new_array(char, iobuf_in_siz)))
603                 out_of_memory("io_start_buffering_in");
604 }
605
606
607 void io_end_buffering(void)
608 {
609         io_flush(NORMAL_FLUSH);
610         if (!io_multiplexing_out) {
611                 free(iobuf_out);
612                 iobuf_out = NULL;
613         }
614 }
615
616
617 /**
618  * Continue trying to read len bytes - don't return until len has been
619  * read.
620  **/
621 static void read_loop(int fd, char *buf, size_t len)
622 {
623         while (len) {
624                 int n = read_timeout(fd, buf, len);
625
626                 buf += n;
627                 len -= n;
628         }
629 }
630
631
632 /**
633  * Read from the file descriptor handling multiplexing - return number
634  * of bytes read.
635  *
636  * Never returns <= 0.
637  */
638 static int readfd_unbuffered(int fd, char *buf, size_t len)
639 {
640         static size_t remaining;
641         static size_t iobuf_in_ndx;
642         int tag, ret = 0;
643         char line[1024];
644
645         if (!iobuf_in || fd != sock_f_in)
646                 return read_timeout(fd, buf, len);
647
648         if (!io_multiplexing_in && remaining == 0) {
649                 remaining = read_timeout(fd, iobuf_in, iobuf_in_siz);
650                 iobuf_in_ndx = 0;
651         }
652
653         while (ret == 0) {
654                 if (remaining) {
655                         len = MIN(len, remaining);
656                         memcpy(buf, iobuf_in + iobuf_in_ndx, len);
657                         iobuf_in_ndx += len;
658                         remaining -= len;
659                         ret = len;
660                         break;
661                 }
662
663                 read_loop(fd, line, 4);
664                 tag = IVAL(line, 0);
665
666                 remaining = tag & 0xFFFFFF;
667                 tag = (tag >> 24) - MPLEX_BASE;
668
669                 switch (tag) {
670                 case MSG_DATA:
671                         if (remaining > iobuf_in_siz) {
672                                 if (!(iobuf_in = realloc_array(iobuf_in, char,
673                                                                remaining)))
674                                         out_of_memory("readfd_unbuffered");
675                                 iobuf_in_siz = remaining;
676                         }
677                         read_loop(fd, iobuf_in, remaining);
678                         iobuf_in_ndx = 0;
679                         break;
680                 case MSG_INFO:
681                 case MSG_ERROR:
682                         if (remaining >= sizeof line) {
683                                 rprintf(FERROR,
684                                         "[%s] multiplexing overflow %d:%ld\n\n",
685                                         who_am_i(), tag, (long)remaining);
686                                 exit_cleanup(RERR_STREAMIO);
687                         }
688                         read_loop(fd, line, remaining);
689                         rwrite((enum logcode)tag, line, remaining);
690                         remaining = 0;
691                         break;
692                 default:
693                         rprintf(FERROR, "[%s] unexpected tag %d\n",
694                                 who_am_i(), tag);
695                         exit_cleanup(RERR_STREAMIO);
696                 }
697         }
698
699         if (remaining == 0)
700                 io_flush(NORMAL_FLUSH);
701
702         return ret;
703 }
704
705
706
707 /**
708  * Do a buffered read from @p fd.  Don't return until all @p n bytes
709  * have been read.  If all @p n can't be read then exit with an
710  * error.
711  **/
712 static void readfd(int fd, char *buffer, size_t N)
713 {
714         int  ret;
715         size_t total = 0;
716
717         while (total < N) {
718                 ret = readfd_unbuffered(fd, buffer + total, N-total);
719                 total += ret;
720         }
721
722         if (fd == write_batch_monitor_in) {
723                 if ((size_t)write(batch_fd, buffer, total) != total)
724                         exit_cleanup(RERR_FILEIO);
725         }
726
727         if (fd == sock_f_in)
728                 stats.total_read += total;
729 }
730
731
732 int32 read_int(int f)
733 {
734         char b[4];
735         int32 ret;
736
737         readfd(f,b,4);
738         ret = IVAL(b,0);
739         if (ret == (int32)0xffffffff)
740                 return -1;
741         return ret;
742 }
743
744 int64 read_longint(int f)
745 {
746         int64 ret;
747         char b[8];
748         ret = read_int(f);
749
750         if ((int32)ret != (int32)0xffffffff)
751                 return ret;
752
753 #if SIZEOF_INT64 < 8
754         rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
755         exit_cleanup(RERR_UNSUPPORTED);
756 #else
757         readfd(f,b,8);
758         ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
759 #endif
760
761         return ret;
762 }
763
764 void read_buf(int f,char *buf,size_t len)
765 {
766         readfd(f,buf,len);
767 }
768
769 void read_sbuf(int f,char *buf,size_t len)
770 {
771         readfd(f, buf, len);
772         buf[len] = 0;
773 }
774
775 unsigned char read_byte(int f)
776 {
777         unsigned char c;
778         readfd(f, (char *)&c, 1);
779         return c;
780 }
781
782 /* Populate a sum_struct with values from the socket.  This is
783  * called by both the sender and the receiver. */
784 void read_sum_head(int f, struct sum_struct *sum)
785 {
786         sum->count = read_int(f);
787         sum->blength = read_int(f);
788         if (sum->blength < 0 || sum->blength > MAX_BLOCK_SIZE) {
789                 rprintf(FERROR, "Invalid block length %ld\n",
790                     (long)sum->blength);
791                 exit_cleanup(RERR_PROTOCOL);
792         }
793         sum->s2length = protocol_version < 27 ? csum_length : (int)read_int(f);
794         if (sum->s2length < 0 || sum->s2length > MD4_SUM_LENGTH) {
795                 rprintf(FERROR, "Invalid checksum length %d\n", sum->s2length);
796                 exit_cleanup(RERR_PROTOCOL);
797         }
798         sum->remainder = read_int(f);
799         if (sum->remainder < 0 || sum->remainder > sum->blength) {
800                 rprintf(FERROR, "Invalid remainder length %ld\n",
801                     (long)sum->remainder);
802                 exit_cleanup(RERR_PROTOCOL);
803         }
804 }
805
806 /* Send the values from a sum_struct over the socket.  Set sum to
807  * NULL if there are no checksums to send.  This is called by both
808  * the generator and the sender. */
809 void write_sum_head(int f, struct sum_struct *sum)
810 {
811         static struct sum_struct null_sum;
812
813         if (sum == NULL)
814                 sum = &null_sum;
815
816         write_int(f, sum->count);
817         write_int(f, sum->blength);
818         if (protocol_version >= 27)
819                 write_int(f, sum->s2length);
820         write_int(f, sum->remainder);
821 }
822
823
824 /**
825  * Sleep after writing to limit I/O bandwidth usage.
826  *
827  * @todo Rather than sleeping after each write, it might be better to
828  * use some kind of averaging.  The current algorithm seems to always
829  * use a bit less bandwidth than specified, because it doesn't make up
830  * for slow periods.  But arguably this is a feature.  In addition, we
831  * ought to take the time used to write the data into account.
832  *
833  * During some phases of big transfers (file FOO is uptodate) this is
834  * called with a small bytes_written every time.  As the kernel has to
835  * round small waits up to guarantee that we actually wait at least the
836  * requested number of microseconds, this can become grossly inaccurate.
837  * We therefore keep track of the bytes we've written over time and only
838  * sleep when the accumulated delay is at least 1 tenth of a second.
839  **/
840 static void sleep_for_bwlimit(int bytes_written)
841 {
842         static struct timeval prior_tv;
843         static long total_written = 0; 
844         struct timeval tv, start_tv;
845         long elapsed_usec, sleep_usec;
846
847 #define ONE_SEC 1000000L /* # of microseconds in a second */
848
849         if (!bwlimit)
850                 return;
851
852         total_written += bytes_written; 
853
854         gettimeofday(&start_tv, NULL);
855         if (prior_tv.tv_sec) {
856                 elapsed_usec = (start_tv.tv_sec - prior_tv.tv_sec) * ONE_SEC
857                              + (start_tv.tv_usec - prior_tv.tv_usec);
858                 total_written -= elapsed_usec * bwlimit / (ONE_SEC/1024);
859                 if (total_written < 0)
860                         total_written = 0;
861         }
862
863         sleep_usec = total_written * (ONE_SEC/1024) / bwlimit;
864         if (sleep_usec < ONE_SEC / 10) {
865                 prior_tv = start_tv;
866                 return;
867         }
868
869         tv.tv_sec  = sleep_usec / ONE_SEC;
870         tv.tv_usec = sleep_usec % ONE_SEC;
871         select(0, NULL, NULL, NULL, &tv);
872
873         gettimeofday(&prior_tv, NULL);
874         elapsed_usec = (prior_tv.tv_sec - start_tv.tv_sec) * ONE_SEC
875                      + (prior_tv.tv_usec - start_tv.tv_usec);
876         total_written = (sleep_usec - elapsed_usec) * bwlimit / (ONE_SEC/1024);
877 }
878
879
880 /* Write len bytes to the file descriptor fd, looping as necessary to get
881  * the job done and also (in the generator) reading any data on msg_fd_in
882  * (to avoid deadlock).
883  *
884  * This function underlies the multiplexing system.  The body of the
885  * application never calls this function directly. */
886 static void writefd_unbuffered(int fd,char *buf,size_t len)
887 {
888         size_t n, total = 0;
889         fd_set w_fds, r_fds;
890         int maxfd, count, ret;
891         struct timeval tv;
892
893         no_flush++;
894
895         while (total < len) {
896                 FD_ZERO(&w_fds);
897                 FD_SET(fd,&w_fds);
898                 maxfd = fd;
899
900                 if (msg_fd_in >= 0) {
901                         FD_ZERO(&r_fds);
902                         FD_SET(msg_fd_in,&r_fds);
903                         if (msg_fd_in > maxfd)
904                                 maxfd = msg_fd_in;
905                 }
906                 if (fd != sock_f_out && iobuf_out_cnt && no_flush == 1) {
907                         FD_SET(sock_f_out, &w_fds);
908                         if (sock_f_out > maxfd)
909                                 maxfd = sock_f_out;
910                 }
911
912                 tv.tv_sec = select_timeout;
913                 tv.tv_usec = 0;
914
915                 errno = 0;
916                 count = select(maxfd + 1, msg_fd_in >= 0 ? &r_fds : NULL,
917                                &w_fds, NULL, &tv);
918
919                 if (count <= 0) {
920                         if (count < 0 && errno == EBADF)
921                                 exit_cleanup(RERR_SOCKETIO);
922                         check_timeout();
923                         continue;
924                 }
925
926                 if (msg_fd_in >= 0 && FD_ISSET(msg_fd_in, &r_fds))
927                         read_msg_fd();
928
929                 if (!FD_ISSET(fd, &w_fds)) {
930                         if (fd != sock_f_out && iobuf_out_cnt) {
931                                 no_flush--;
932                                 io_flush(NORMAL_FLUSH);
933                                 no_flush++;
934                         }
935                         continue;
936                 }
937
938                 n = len - total;
939                 if (bwlimit && n > bwlimit_writemax)
940                         n = bwlimit_writemax;
941                 ret = write(fd, buf + total, n);
942
943                 if (ret <= 0) {
944                         if (ret < 0) {
945                                 if (errno == EINTR)
946                                         continue;
947                                 if (errno == EWOULDBLOCK || errno == EAGAIN) {
948                                         msleep(1);
949                                         continue;
950                                 }
951                         }
952
953                         /* Don't try to write errors back across the stream. */
954                         if (fd == sock_f_out)
955                                 close_multiplexing_out();
956                         rsyserr(FERROR, errno,
957                                 "writefd_unbuffered failed to write %ld bytes: phase \"%s\" [%s]",
958                                 (long)len, io_write_phase, who_am_i());
959                         /* If the other side is sending us error messages, try
960                          * to grab any messages they sent before they died. */
961                         while (fd == sock_f_out && io_multiplexing_in) {
962                                 io_timeout = select_timeout = 30;
963                                 readfd_unbuffered(sock_f_in, io_filesfrom_buf,
964                                                   sizeof io_filesfrom_buf);
965                         }
966                         exit_cleanup(RERR_STREAMIO);
967                 }
968
969                 total += ret;
970
971                 if (fd == sock_f_out) {
972                         if (io_timeout)
973                                 last_io = time(NULL);
974                         sleep_for_bwlimit(ret);
975                 }
976         }
977
978         no_flush--;
979 }
980
981
982 /**
983  * Write an message to a multiplexed stream. If this fails then rsync
984  * exits.
985  **/
986 static void mplex_write(enum msgcode code, char *buf, size_t len)
987 {
988         char buffer[4096];
989         size_t n = len;
990
991         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
992
993         if (n > sizeof buffer - 4)
994                 n = sizeof buffer - 4;
995
996         memcpy(&buffer[4], buf, n);
997         writefd_unbuffered(sock_f_out, buffer, n+4);
998
999         len -= n;
1000         buf += n;
1001
1002         if (len)
1003                 writefd_unbuffered(sock_f_out, buf, len);
1004 }
1005
1006
1007 void io_flush(int flush_it_all)
1008 {
1009         msg_list_push(flush_it_all);
1010
1011         if (!iobuf_out_cnt || no_flush)
1012                 return;
1013
1014         if (io_multiplexing_out)
1015                 mplex_write(MSG_DATA, iobuf_out, iobuf_out_cnt);
1016         else
1017                 writefd_unbuffered(sock_f_out, iobuf_out, iobuf_out_cnt);
1018         iobuf_out_cnt = 0;
1019 }
1020
1021
1022 static void writefd(int fd,char *buf,size_t len)
1023 {
1024         if (fd == msg_fd_out) {
1025                 rprintf(FERROR, "Internal error: wrong write used in receiver.\n");
1026                 exit_cleanup(RERR_PROTOCOL);
1027         }
1028
1029         if (fd == sock_f_out)
1030                 stats.total_written += len;
1031
1032         if (fd == write_batch_monitor_out) {
1033                 if ((size_t)write(batch_fd, buf, len) != len)
1034                         exit_cleanup(RERR_FILEIO);
1035         }
1036
1037         if (!iobuf_out || fd != sock_f_out) {
1038                 writefd_unbuffered(fd, buf, len);
1039                 return;
1040         }
1041
1042         while (len) {
1043                 int n = MIN((int)len, IO_BUFFER_SIZE - iobuf_out_cnt);
1044                 if (n > 0) {
1045                         memcpy(iobuf_out+iobuf_out_cnt, buf, n);
1046                         buf += n;
1047                         len -= n;
1048                         iobuf_out_cnt += n;
1049                 }
1050
1051                 if (iobuf_out_cnt == IO_BUFFER_SIZE)
1052                         io_flush(NORMAL_FLUSH);
1053         }
1054 }
1055
1056
1057 void write_int(int f,int32 x)
1058 {
1059         char b[4];
1060         SIVAL(b,0,x);
1061         writefd(f,b,4);
1062 }
1063
1064
1065 void write_int_named(int f, int32 x, const char *phase)
1066 {
1067         io_write_phase = phase;
1068         write_int(f, x);
1069         io_write_phase = phase_unknown;
1070 }
1071
1072
1073 /*
1074  * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
1075  * 64-bit types on this platform.
1076  */
1077 void write_longint(int f, int64 x)
1078 {
1079         char b[8];
1080
1081         if (x <= 0x7FFFFFFF) {
1082                 write_int(f, (int)x);
1083                 return;
1084         }
1085
1086 #if SIZEOF_INT64 < 8
1087         rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
1088         exit_cleanup(RERR_UNSUPPORTED);
1089 #else
1090         write_int(f, (int32)0xFFFFFFFF);
1091         SIVAL(b,0,(x&0xFFFFFFFF));
1092         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
1093
1094         writefd(f,b,8);
1095 #endif
1096 }
1097
1098 void write_buf(int f,char *buf,size_t len)
1099 {
1100         writefd(f,buf,len);
1101 }
1102
1103 /** Write a string to the connection */
1104 void write_sbuf(int f, char *buf)
1105 {
1106         writefd(f, buf, strlen(buf));
1107 }
1108
1109 void write_byte(int f,unsigned char c)
1110 {
1111         writefd(f, (char *)&c, 1);
1112 }
1113
1114
1115
1116 /**
1117  * Read a line of up to @p maxlen characters into @p buf (not counting
1118  * the trailing null).  Strips the (required) trailing newline and all
1119  * carriage returns.
1120  *
1121  * @return 1 for success; 0 for I/O error or truncation.
1122  **/
1123 int read_line(int f, char *buf, size_t maxlen)
1124 {
1125         while (maxlen) {
1126                 buf[0] = 0;
1127                 read_buf(f, buf, 1);
1128                 if (buf[0] == 0)
1129                         return 0;
1130                 if (buf[0] == '\n')
1131                         break;
1132                 if (buf[0] != '\r') {
1133                         buf++;
1134                         maxlen--;
1135                 }
1136         }
1137         *buf = '\0';
1138         return maxlen > 0;
1139 }
1140
1141
1142 void io_printf(int fd, const char *format, ...)
1143 {
1144         va_list ap;
1145         char buf[1024];
1146         int len;
1147
1148         va_start(ap, format);
1149         len = vsnprintf(buf, sizeof buf, format, ap);
1150         va_end(ap);
1151
1152         if (len < 0)
1153                 exit_cleanup(RERR_STREAMIO);
1154
1155         write_sbuf(fd, buf);
1156 }
1157
1158
1159 /** Setup for multiplexing a MSG_* stream with the data stream. */
1160 void io_start_multiplex_out(void)
1161 {
1162         io_flush(NORMAL_FLUSH);
1163         io_start_buffering_out();
1164         io_multiplexing_out = 1;
1165 }
1166
1167 /** Setup for multiplexing a MSG_* stream with the data stream. */
1168 void io_start_multiplex_in(void)
1169 {
1170         io_flush(NORMAL_FLUSH);
1171         io_start_buffering_in();
1172         io_multiplexing_in = 1;
1173 }
1174
1175 /** Write an message to the multiplexed data stream. */
1176 int io_multiplex_write(enum msgcode code, char *buf, size_t len)
1177 {
1178         if (!io_multiplexing_out)
1179                 return 0;
1180
1181         io_flush(NORMAL_FLUSH);
1182         stats.total_written += (len+4);
1183         mplex_write(code, buf, len);
1184         return 1;
1185 }
1186
1187 void close_multiplexing_in(void)
1188 {
1189         io_multiplexing_in = 0;
1190 }
1191
1192 /** Stop output multiplexing. */
1193 void close_multiplexing_out(void)
1194 {
1195         io_multiplexing_out = 0;
1196 }
1197
1198 void start_write_batch(int fd)
1199 {
1200         write_stream_flags(batch_fd);
1201
1202         /* Some communication has already taken place, but we don't
1203          * enable batch writing until here so that we can write a
1204          * canonical record of the communication even though the
1205          * actual communication so far depends on whether a daemon
1206          * is involved. */
1207         write_int(batch_fd, protocol_version);
1208         write_int(batch_fd, checksum_seed);
1209
1210         if (am_sender)
1211                 write_batch_monitor_out = fd;
1212         else
1213                 write_batch_monitor_in = fd;
1214 }
1215
1216 void stop_write_batch(void)
1217 {
1218         write_batch_monitor_out = -1;
1219         write_batch_monitor_in = -1;
1220 }