Output an error before we die if we get a wacko message from the receiver.
[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 IO 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 static int io_multiplexing_out;
43 static int io_multiplexing_in;
44 static int multiplex_in_fd = -1;
45 static int multiplex_out_fd = -1;
46 static time_t last_io;
47 static int no_flush;
48
49 extern int bwlimit;
50 extern int verbose;
51 extern int io_timeout;
52 extern int am_server;
53 extern int am_daemon;
54 extern int am_sender;
55 extern struct stats stats;
56
57
58 const char phase_unknown[] = "unknown";
59
60 /**
61  * The connection might be dropped at some point; perhaps because the
62  * remote instance crashed.  Just giving the offset on the stream is
63  * not very helpful.  So instead we try to make io_phase_name point to
64  * something useful.
65  *
66  * For buffered/multiplexed IO these names will be somewhat
67  * approximate; perhaps for ease of support we would rather make the
68  * buffer always flush when a single application-level IO finishes.
69  *
70  * @todo Perhaps we want some simple stack functionality, but there's
71  * no need to overdo it.
72  **/
73 const char *io_write_phase = phase_unknown;
74 const char *io_read_phase = phase_unknown;
75
76 /** Ignore EOF errors while reading a module listing if the remote
77     version is 24 or less. */
78 int kludge_around_eof = False;
79
80 int msg_fd_in = -1;
81 int msg_fd_out = -1;
82
83 static int io_filesfrom_f_in = -1;
84 static int io_filesfrom_f_out = -1;
85 static char io_filesfrom_buf[2048];
86 static char *io_filesfrom_bp;
87 static char io_filesfrom_lastchar;
88 static int io_filesfrom_buflen;
89
90 static void read_loop(int fd, char *buf, size_t len);
91
92 struct redo_list {
93         struct redo_list *next;
94         int num;
95 };
96
97 static struct redo_list *redo_list_head;
98 static struct redo_list *redo_list_tail;
99
100 struct msg_list {
101         struct msg_list *next;
102         char *buf;
103         int len;
104 };
105
106 static struct msg_list *msg_list_head;
107 static struct msg_list *msg_list_tail;
108
109 static void redo_list_add(int num)
110 {
111         struct redo_list *rl;
112
113         if (!(rl = new(struct redo_list)))
114                 exit_cleanup(RERR_MALLOC);
115         rl->next = NULL;
116         rl->num = num;
117         if (redo_list_tail)
118                 redo_list_tail->next = rl;
119         else
120                 redo_list_head = rl;
121         redo_list_tail = rl;
122 }
123
124 static void check_timeout(void)
125 {
126         time_t t;
127
128         if (!io_timeout)
129                 return;
130
131         if (!last_io) {
132                 last_io = time(NULL);
133                 return;
134         }
135
136         t = time(NULL);
137
138         if (last_io && io_timeout && (t-last_io) >= io_timeout) {
139                 if (!am_server && !am_daemon) {
140                         rprintf(FERROR,"io timeout after %d seconds - exiting\n", 
141                                 (int)(t-last_io));
142                 }
143                 exit_cleanup(RERR_TIMEOUT);
144         }
145 }
146
147 /** Setup the fd used to receive MSG_* messages.  Only needed when
148  * we're the generator because the sender and receiver both use the
149  * multiplexed IO setup. */
150 void set_msg_fd_in(int fd)
151 {
152         msg_fd_in = fd;
153 }
154
155 /** Setup the fd used to send our MSG_* messages.  Only needed when
156  * we're the receiver because the generator and the sender both use
157  * the multiplexed IO setup. */
158 void set_msg_fd_out(int fd)
159 {
160         msg_fd_out = fd;
161         set_nonblocking(msg_fd_out);
162 }
163
164 /* Add a message to the pending MSG_* list. */
165 static void msg_list_add(int code, char *buf, int len)
166 {
167         struct msg_list *ml;
168
169         if (!(ml = new(struct msg_list)))
170                 exit_cleanup(RERR_MALLOC);
171         ml->next = NULL;
172         if (!(ml->buf = new_array(char, len+4)))
173                 exit_cleanup(RERR_MALLOC);
174         SIVAL(ml->buf, 0, ((code+MPLEX_BASE)<<24) | len);
175         memcpy(ml->buf+4, buf, len);
176         ml->len = len+4;
177         if (msg_list_tail)
178                 msg_list_tail->next = ml;
179         else
180                 msg_list_head = ml;
181         msg_list_tail = ml;
182 }
183
184 void send_msg(enum msgcode code, char *buf, int len)
185 {
186         msg_list_add(code, buf, len);
187         msg_list_push(NORMAL_FLUSH);
188 }
189
190 /** Read a message from the MSG_* fd and dispatch it.  This is only
191  * called by the generator. */
192 static void read_msg_fd(void)
193 {
194         char buf[200];
195         size_t n;
196         int fd = msg_fd_in;
197         int tag, len;
198
199         /* Temporarily disable msg_fd_in.  This is needed because we
200          * may call a write routine that could try to call us back. */
201         msg_fd_in = -1;
202
203         read_loop(fd, buf, 4);
204         tag = IVAL(buf, 0);
205
206         len = tag & 0xFFFFFF;
207         tag = (tag >> 24) - MPLEX_BASE;
208
209         switch (tag) {
210         case MSG_DONE:
211                 if (len != 0) {
212                         rprintf(FERROR, "invalid message %d:%d\n", tag, len);
213                         exit_cleanup(RERR_STREAMIO);
214                 }
215                 redo_list_add(-1);
216                 break;
217         case MSG_REDO:
218                 if (len != 4) {
219                         rprintf(FERROR, "invalid message %d:%d\n", tag, len);
220                         exit_cleanup(RERR_STREAMIO);
221                 }
222                 read_loop(fd, buf, 4);
223                 redo_list_add(IVAL(buf,0));
224                 break;
225         case MSG_INFO:
226         case MSG_ERROR:
227         case MSG_LOG:
228                 while (len) {
229                         n = len;
230                         if (n >= sizeof buf)
231                                 n = sizeof buf - 1;
232                         read_loop(fd, buf, n);
233                         rwrite((enum logcode)tag, buf, n);
234                         len -= n;
235                 }
236                 break;
237         default:
238                 rprintf(FERROR, "unknown message %d:%d\n", tag, len);
239                 exit_cleanup(RERR_STREAMIO);
240         }
241
242         msg_fd_in = fd;
243 }
244
245 /* Try to push messages off the list onto the wire.  If we leave with more
246  * to do, return 0.  On error, return -1.  If everything flushed, return 1.
247  * This is only called by the receiver. */
248 int msg_list_push(int flush_it_all)
249 {
250         static int written = 0;
251         struct timeval tv;
252         fd_set fds;
253
254         if (msg_fd_out < 0)
255                 return -1;
256
257         while (msg_list_head) {
258                 struct msg_list *ml = msg_list_head;
259                 int n = write(msg_fd_out, ml->buf + written, ml->len - written);
260                 if (n < 0) {
261                         if (errno == EINTR)
262                                 continue;
263                         if (errno != EWOULDBLOCK && errno != EAGAIN)
264                                 return -1;
265                         if (!flush_it_all)
266                                 return 0;
267                         FD_ZERO(&fds);
268                         FD_SET(msg_fd_out, &fds);
269                         tv.tv_sec = io_timeout ? io_timeout : SELECT_TIMEOUT;
270                         tv.tv_usec = 0;
271                         if (!select(msg_fd_out+1, NULL, &fds, NULL, &tv))
272                                 check_timeout();
273                 } else if ((written += n) == ml->len) {
274                         free(ml->buf);
275                         msg_list_head = ml->next;
276                         if (!msg_list_head)
277                                 msg_list_tail = NULL;
278                         free(ml);
279                         written = 0;
280                 }
281         }
282         return 1;
283 }
284
285 int get_redo_num(void)
286 {
287         struct redo_list *next;
288         int num;
289
290         while (!redo_list_head)
291                 read_msg_fd();
292
293         num = redo_list_head->num;
294         next = redo_list_head->next;
295         free(redo_list_head);
296         redo_list_head = next;
297         if (!next)
298                 redo_list_tail = NULL;
299
300         return num;
301 }
302
303 /**
304  * When we're the receiver and we have a local --files-from list of names
305  * that needs to be sent over the socket to the sender, we have to do two
306  * things at the same time: send the sender a list of what files we're
307  * processing and read the incoming file+info list from the sender.  We do
308  * this by augmenting the read_timeout() function to copy this data.  It
309  * uses the io_filesfrom_buf to read a block of data from f_in (when it is
310  * ready, since it might be a pipe) and then blast it out f_out (when it
311  * is ready to receive more data).
312  */
313 void io_set_filesfrom_fds(int f_in, int f_out)
314 {
315         io_filesfrom_f_in = f_in;
316         io_filesfrom_f_out = f_out;
317         io_filesfrom_bp = io_filesfrom_buf;
318         io_filesfrom_lastchar = '\0';
319         io_filesfrom_buflen = 0;
320 }
321
322 /**
323  * It's almost always an error to get an EOF when we're trying to read
324  * from the network, because the protocol is self-terminating.
325  *
326  * However, there is one unfortunate cases where it is not, which is
327  * rsync <2.4.6 sending a list of modules on a server, since the list
328  * is terminated by closing the socket. So, for the section of the
329  * program where that is a problem (start_socket_client),
330  * kludge_around_eof is True and we just exit.
331  */
332 static void whine_about_eof(void)
333 {
334         if (kludge_around_eof)
335                 exit_cleanup(0);
336         else {
337                 rprintf(FERROR,
338                         "%s: connection unexpectedly closed "
339                         "(%.0f bytes read so far)\n",
340                         RSYNC_NAME, (double)stats.total_read);
341
342                 exit_cleanup(RERR_STREAMIO);
343         }
344 }
345
346
347 static void die_from_readerr(int err)
348 {
349         /* this prevents us trying to write errors on a dead socket */
350         io_multiplexing_close();
351
352         rprintf(FERROR, "%s: read error: %s\n",
353                 RSYNC_NAME, strerror(err));
354         exit_cleanup(RERR_STREAMIO);
355 }
356
357
358 /**
359  * Read from a socket with IO timeout. return the number of bytes
360  * read. If no bytes can be read then exit, never return a number <= 0.
361  *
362  * TODO: If the remote shell connection fails, then current versions
363  * actually report an "unexpected EOF" error here.  Since it's a
364  * fairly common mistake to try to use rsh when ssh is required, we
365  * should trap that: if we fail to read any data at all, we should
366  * give a better explanation.  We can tell whether the connection has
367  * started by looking e.g. at whether the remote version is known yet.
368  */
369 static int read_timeout(int fd, char *buf, size_t len)
370 {
371         int n, ret=0;
372
373         io_flush(NORMAL_FLUSH);
374
375         while (ret == 0) {
376                 /* until we manage to read *something* */
377                 fd_set r_fds, w_fds;
378                 struct timeval tv;
379                 int fd_count = fd+1;
380                 int count;
381
382                 FD_ZERO(&r_fds);
383                 FD_SET(fd, &r_fds);
384                 if (msg_fd_in >= 0) {
385                         FD_SET(msg_fd_in, &r_fds);
386                         if (msg_fd_in >= fd_count)
387                                 fd_count = msg_fd_in+1;
388                 }
389                 if (io_filesfrom_f_out >= 0) {
390                         int new_fd;
391                         if (io_filesfrom_buflen == 0) {
392                                 if (io_filesfrom_f_in >= 0) {
393                                         FD_SET(io_filesfrom_f_in, &r_fds);
394                                         new_fd = io_filesfrom_f_in;
395                                 } else {
396                                         io_filesfrom_f_out = -1;
397                                         new_fd = -1;
398                                 }
399                         } else {
400                                 FD_ZERO(&w_fds);
401                                 FD_SET(io_filesfrom_f_out, &w_fds);
402                                 new_fd = io_filesfrom_f_out;
403                         }
404                         if (new_fd >= fd_count)
405                                 fd_count = new_fd+1;
406                 }
407
408                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
409                 tv.tv_usec = 0;
410
411                 errno = 0;
412
413                 count = select(fd_count, &r_fds,
414                                io_filesfrom_buflen? &w_fds : NULL,
415                                NULL, &tv);
416
417                 if (count == 0) {
418                         msg_list_push(NORMAL_FLUSH);
419                         check_timeout();
420                 }
421
422                 if (count <= 0) {
423                         if (errno == EBADF) {
424                                 exit_cleanup(RERR_SOCKETIO);
425                         }
426                         continue;
427                 }
428
429                 if (msg_fd_in >= 0 && FD_ISSET(msg_fd_in, &r_fds))
430                         read_msg_fd();
431
432                 if (io_filesfrom_f_out >= 0) {
433                         if (io_filesfrom_buflen) {
434                                 if (FD_ISSET(io_filesfrom_f_out, &w_fds)) {
435                                         int l = write(io_filesfrom_f_out,
436                                                       io_filesfrom_bp,
437                                                       io_filesfrom_buflen);
438                                         if (l > 0) {
439                                                 if (!(io_filesfrom_buflen -= l))
440                                                         io_filesfrom_bp = io_filesfrom_buf;
441                                                 else
442                                                         io_filesfrom_bp += l;
443                                         } else {
444                                                 /* XXX should we complain? */
445                                                 io_filesfrom_f_out = -1;
446                                         }
447                                 }
448                         } else if (io_filesfrom_f_in >= 0) {
449                                 if (FD_ISSET(io_filesfrom_f_in, &r_fds)) {
450                                         int l = read(io_filesfrom_f_in,
451                                                      io_filesfrom_buf,
452                                                      sizeof io_filesfrom_buf);
453                                         if (l <= 0) {
454                                                 /* Send end-of-file marker */
455                                                 io_filesfrom_buf[0] = '\0';
456                                                 io_filesfrom_buf[1] = '\0';
457                                                 io_filesfrom_buflen = io_filesfrom_lastchar? 2 : 1;
458                                                 io_filesfrom_f_in = -1;
459                                         } else {
460                                                 extern int eol_nulls;
461                                                 if (!eol_nulls) {
462                                                         char *s = io_filesfrom_buf + l;
463                                                         /* Transform CR and/or LF into '\0' */
464                                                         while (s-- > io_filesfrom_buf) {
465                                                                 if (*s == '\n' || *s == '\r')
466                                                                         *s = '\0';
467                                                         }
468                                                 }
469                                                 if (!io_filesfrom_lastchar) {
470                                                         /* Last buf ended with a '\0', so don't
471                                                          * let this buf start with one. */
472                                                         while (l && !*io_filesfrom_bp)
473                                                                 io_filesfrom_bp++, l--;
474                                                 }
475                                                 if (!l)
476                                                         io_filesfrom_bp = io_filesfrom_buf;
477                                                 else {
478                                                         char *f = io_filesfrom_bp;
479                                                         char *t = f;
480                                                         char *eob = f + l;
481                                                         /* Eliminate any multi-'\0' runs. */
482                                                         while (f != eob) {
483                                                                 if (!(*t++ = *f++)) {
484                                                                         while (f != eob && !*f)
485                                                                                 f++, l--;
486                                                                 }
487                                                         }
488                                                         io_filesfrom_lastchar = f[-1];
489                                                 }
490                                                 io_filesfrom_buflen = l;
491                                         }
492                                 }
493                         }
494                 }
495
496                 if (!FD_ISSET(fd, &r_fds)) continue;
497
498                 n = read(fd, buf, len);
499
500                 if (n > 0) {
501                         buf += n;
502                         len -= n;
503                         ret += n;
504                         if (io_timeout)
505                                 last_io = time(NULL);
506                         continue;
507                 } else if (n == 0) {
508                         whine_about_eof();
509                         return -1; /* doesn't return */
510                 } else if (n < 0) {
511                         if (errno == EINTR || errno == EWOULDBLOCK ||
512                             errno == EAGAIN) 
513                                 continue;
514                         die_from_readerr(errno);
515                 }
516         }
517
518         return ret;
519 }
520
521 /**
522  * Read a line into the "fname" buffer (which must be at least MAXPATHLEN
523  * characters long).
524  */
525 int read_filesfrom_line(int fd, char *fname)
526 {
527         char ch, *s, *eob = fname + MAXPATHLEN - 1;
528         int cnt;
529         extern int io_timeout;
530         extern int eol_nulls;
531         extern char *remote_filesfrom_file;
532         int reading_remotely = remote_filesfrom_file != NULL;
533         int nulls = eol_nulls || reading_remotely;
534
535   start:
536         s = fname;
537         while (1) {
538                 cnt = read(fd, &ch, 1);
539                 if (cnt < 0 && (errno == EWOULDBLOCK
540                   || errno == EINTR || errno == EAGAIN)) {
541                         struct timeval tv;
542                         fd_set fds;
543                         FD_ZERO(&fds);
544                         FD_SET(fd, &fds);
545                         tv.tv_sec = io_timeout? io_timeout : SELECT_TIMEOUT;
546                         tv.tv_usec = 0;
547                         if (!select(fd+1, &fds, NULL, NULL, &tv))
548                                 check_timeout();
549                         continue;
550                 }
551                 if (cnt != 1)
552                         break;
553                 if (nulls? !ch : (ch == '\r' || ch == '\n')) {
554                         /* Skip empty lines if reading locally. */
555                         if (!reading_remotely && s == fname)
556                                 continue;
557                         break;
558                 }
559                 if (s < eob)
560                         *s++ = ch;
561         }
562         *s = '\0';
563
564         /* Dump comments. */
565         if (*fname == '#' || *fname == ';')
566                 goto start;
567
568         return s - fname;
569 }
570
571
572 /**
573  * Continue trying to read len bytes - don't return until len has been
574  * read.
575  **/
576 static void read_loop(int fd, char *buf, size_t len)
577 {
578         while (len) {
579                 int n = read_timeout(fd, buf, len);
580
581                 buf += n;
582                 len -= n;
583         }
584 }
585
586
587 /**
588  * Read from the file descriptor handling multiplexing - return number
589  * of bytes read.
590  * 
591  * Never returns <= 0. 
592  */
593 static int read_unbuffered(int fd, char *buf, size_t len)
594 {
595         static size_t remaining;
596         int tag, ret = 0;
597         char line[1024];
598         static char *buffer;
599         static size_t bufferIdx = 0;
600         static size_t bufferSz;
601
602         if (fd != multiplex_in_fd)
603                 return read_timeout(fd, buf, len);
604
605         if (!io_multiplexing_in && remaining == 0) {
606                 if (!buffer) {
607                         bufferSz = 2 * IO_BUFFER_SIZE;
608                         buffer   = new_array(char, bufferSz);
609                         if (!buffer) out_of_memory("read_unbuffered");
610                 }
611                 remaining = read_timeout(fd, buffer, bufferSz);
612                 bufferIdx = 0;
613         }
614
615         while (ret == 0) {
616                 if (remaining) {
617                         len = MIN(len, remaining);
618                         memcpy(buf, buffer + bufferIdx, len);
619                         bufferIdx += len;
620                         remaining -= len;
621                         ret = len;
622                         break;
623                 }
624
625                 read_loop(fd, line, 4);
626                 tag = IVAL(line, 0);
627
628                 remaining = tag & 0xFFFFFF;
629                 tag = (tag >> 24) - MPLEX_BASE;
630
631                 switch (tag) {
632                 case MSG_DATA:
633                         if (!buffer || remaining > bufferSz) {
634                                 buffer = realloc_array(buffer, char, remaining);
635                                 if (!buffer) out_of_memory("read_unbuffered");
636                                 bufferSz = remaining;
637                         }
638                         read_loop(fd, buffer, remaining);
639                         bufferIdx = 0;
640                         break;
641                 case MSG_INFO:
642                 case MSG_ERROR:
643                         if (remaining >= sizeof line) {
644                                 rprintf(FERROR, "multiplexing overflow %d:%ld\n\n",
645                                         tag, (long)remaining);
646                                 exit_cleanup(RERR_STREAMIO);
647                         }
648                         read_loop(fd, line, remaining);
649                         rwrite((enum logcode)tag, line, remaining);
650                         remaining = 0;
651                         break;
652                 default:
653                         rprintf(FERROR, "unexpected tag %d\n", tag);
654                         exit_cleanup(RERR_STREAMIO);
655                 }
656         }
657
658         if (remaining == 0)
659                 io_flush(NORMAL_FLUSH);
660
661         return ret;
662 }
663
664
665
666 /**
667  * Do a buffered read from @p fd.  Don't return until all @p n bytes
668  * have been read.  If all @p n can't be read then exit with an
669  * error.
670  **/
671 static void readfd(int fd, char *buffer, size_t N)
672 {
673         int  ret;
674         size_t total=0;  
675
676         while (total < N) {
677                 ret = read_unbuffered(fd, buffer + total, N-total);
678                 total += ret;
679         }
680
681         stats.total_read += total;
682 }
683
684
685 int32 read_int(int f)
686 {
687         char b[4];
688         int32 ret;
689
690         readfd(f,b,4);
691         ret = IVAL(b,0);
692         if (ret == (int32)0xffffffff) return -1;
693         return ret;
694 }
695
696 int64 read_longint(int f)
697 {
698         int64 ret;
699         char b[8];
700         ret = read_int(f);
701
702         if ((int32)ret != (int32)0xffffffff) {
703                 return ret;
704         }
705
706 #ifdef NO_INT64
707         rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
708         exit_cleanup(RERR_UNSUPPORTED);
709 #else
710         readfd(f,b,8);
711         ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
712 #endif
713
714         return ret;
715 }
716
717 void read_buf(int f,char *buf,size_t len)
718 {
719         readfd(f,buf,len);
720 }
721
722 void read_sbuf(int f,char *buf,size_t len)
723 {
724         read_buf(f,buf,len);
725         buf[len] = 0;
726 }
727
728 unsigned char read_byte(int f)
729 {
730         unsigned char c;
731         read_buf(f, (char *)&c, 1);
732         return c;
733 }
734
735
736 /**
737  * Sleep after writing to limit I/O bandwidth usage.
738  *
739  * @todo Rather than sleeping after each write, it might be better to
740  * use some kind of averaging.  The current algorithm seems to always
741  * use a bit less bandwidth than specified, because it doesn't make up
742  * for slow periods.  But arguably this is a feature.  In addition, we
743  * ought to take the time used to write the data into account.
744  **/
745 static void sleep_for_bwlimit(int bytes_written)
746 {
747         struct timeval tv;
748
749         if (!bwlimit)
750                 return;
751
752         assert(bytes_written > 0);
753         assert(bwlimit > 0);
754
755         tv.tv_usec = bytes_written * 1000 / bwlimit;
756         tv.tv_sec  = tv.tv_usec / 1000000;
757         tv.tv_usec = tv.tv_usec % 1000000;
758
759         select(0, NULL, NULL, NULL, &tv);
760 }
761
762
763 /**
764  * Write len bytes to the file descriptor @p fd.
765  *
766  * This function underlies the multiplexing system.  The body of the
767  * application never calls this function directly.
768  **/
769 static void writefd_unbuffered(int fd,char *buf,size_t len)
770 {
771         size_t total = 0;
772         fd_set w_fds, r_fds;
773         int fd_count, count;
774         struct timeval tv;
775
776         msg_list_push(NORMAL_FLUSH);
777
778         no_flush++;
779
780         while (total < len) {
781                 FD_ZERO(&w_fds);
782                 FD_SET(fd,&w_fds);
783                 fd_count = fd;
784
785                 if (msg_fd_in >= 0) {
786                         FD_ZERO(&r_fds);
787                         FD_SET(msg_fd_in,&r_fds);
788                         if (msg_fd_in > fd_count) 
789                                 fd_count = msg_fd_in;
790                 }
791
792                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
793                 tv.tv_usec = 0;
794
795                 errno = 0;
796                 count = select(fd_count+1, msg_fd_in >= 0 ? &r_fds : NULL,
797                                &w_fds, NULL, &tv);
798
799                 if (count == 0) {
800                         msg_list_push(NORMAL_FLUSH);
801                         check_timeout();
802                 }
803
804                 if (count <= 0) {
805                         if (errno == EBADF) {
806                                 exit_cleanup(RERR_SOCKETIO);
807                         }
808                         continue;
809                 }
810
811                 if (msg_fd_in >= 0 && FD_ISSET(msg_fd_in, &r_fds))
812                         read_msg_fd();
813
814                 if (FD_ISSET(fd, &w_fds)) {
815                         int ret;
816                         size_t n = len-total;
817                         ret = write(fd,buf+total,n);
818
819                         if (ret < 0) {
820                                 if (errno == EINTR)
821                                         continue;
822                                 if (errno == EWOULDBLOCK || errno == EAGAIN) {
823                                         msleep(1);
824                                         continue;
825                                 }
826                         }
827
828                         if (ret <= 0) {
829                                 /* Don't try to write errors back
830                                  * across the stream */
831                                 io_multiplexing_close();
832                                 rprintf(FERROR, RSYNC_NAME
833                                         ": writefd_unbuffered failed to write %ld bytes: phase \"%s\": %s\n",
834                                         (long) len, io_write_phase, 
835                                         strerror(errno));
836                                 exit_cleanup(RERR_STREAMIO);
837                         }
838
839                         sleep_for_bwlimit(ret);
840  
841                         total += ret;
842
843                         if (io_timeout)
844                                 last_io = time(NULL);
845                 }
846         }
847
848         no_flush--;
849 }
850
851
852 static char *io_buffer;
853 static int io_buffer_count;
854
855 void io_start_buffering_out(int fd)
856 {
857         if (io_buffer) return;
858         multiplex_out_fd = fd;
859         io_buffer = new_array(char, IO_BUFFER_SIZE);
860         if (!io_buffer) out_of_memory("writefd");
861         io_buffer_count = 0;
862 }
863
864 void io_start_buffering_in(int fd)
865 {
866         multiplex_in_fd = fd;
867 }
868
869 /**
870  * Write an message to a multiplexed stream. If this fails then rsync
871  * exits.
872  **/
873 static void mplex_write(int fd, enum msgcode code, char *buf, size_t len)
874 {
875         char buffer[4096];
876         size_t n = len;
877
878         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
879
880         if (n > (sizeof buffer - 4)) {
881                 n = sizeof buffer - 4;
882         }
883
884         memcpy(&buffer[4], buf, n);
885         writefd_unbuffered(fd, buffer, n+4);
886
887         len -= n;
888         buf += n;
889
890         if (len) {
891                 writefd_unbuffered(fd, buf, len);
892         }
893 }
894
895
896 void io_flush(int flush_it_all)
897 {
898         int fd = multiplex_out_fd;
899         
900         msg_list_push(flush_it_all);
901
902         if (!io_buffer_count || no_flush)
903                 return;
904
905         if (io_multiplexing_out)
906                 mplex_write(fd, MSG_DATA, io_buffer, io_buffer_count);
907         else
908                 writefd_unbuffered(fd, io_buffer, io_buffer_count);
909         io_buffer_count = 0;
910 }
911
912
913 void io_end_buffering(void)
914 {
915         io_flush(NORMAL_FLUSH);
916         if (!io_multiplexing_out) {
917                 free(io_buffer);
918                 io_buffer = NULL;
919         }
920 }
921
922 static void writefd(int fd,char *buf,size_t len)
923 {
924         stats.total_written += len;
925
926         msg_list_push(NORMAL_FLUSH);
927
928         if (!io_buffer || fd != multiplex_out_fd) {
929                 writefd_unbuffered(fd, buf, len);
930                 return;
931         }
932
933         while (len) {
934                 int n = MIN((int) len, IO_BUFFER_SIZE-io_buffer_count);
935                 if (n > 0) {
936                         memcpy(io_buffer+io_buffer_count, buf, n);
937                         buf += n;
938                         len -= n;
939                         io_buffer_count += n;
940                 }
941
942                 if (io_buffer_count == IO_BUFFER_SIZE)
943                         io_flush(NORMAL_FLUSH);
944         }
945 }
946
947
948 void write_int(int f,int32 x)
949 {
950         char b[4];
951         SIVAL(b,0,x);
952         writefd(f,b,4);
953 }
954
955
956 void write_int_named(int f, int32 x, const char *phase)
957 {
958         io_write_phase = phase;
959         write_int(f, x);
960         io_write_phase = phase_unknown;
961 }
962
963
964 /*
965  * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
966  * 64-bit types on this platform.
967  */
968 void write_longint(int f, int64 x)
969 {
970         char b[8];
971
972         if (x <= 0x7FFFFFFF) {
973                 write_int(f, (int)x);
974                 return;
975         }
976
977 #ifdef NO_INT64
978         rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
979         exit_cleanup(RERR_UNSUPPORTED);
980 #else
981         write_int(f, (int32)0xFFFFFFFF);
982         SIVAL(b,0,(x&0xFFFFFFFF));
983         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
984
985         writefd(f,b,8);
986 #endif
987 }
988
989 void write_buf(int f,char *buf,size_t len)
990 {
991         writefd(f,buf,len);
992 }
993
994 /** Write a string to the connection */
995 static void write_sbuf(int f,char *buf)
996 {
997         write_buf(f, buf, strlen(buf));
998 }
999
1000
1001 void write_byte(int f,unsigned char c)
1002 {
1003         write_buf(f,(char *)&c,1);
1004 }
1005
1006
1007
1008 /**
1009  * Read a line of up to @p maxlen characters into @p buf.  Does not
1010  * contain a trailing newline or carriage return.
1011  *
1012  * @return 1 for success; 0 for io error or truncation.
1013  **/
1014 int read_line(int f, char *buf, size_t maxlen)
1015 {
1016         while (maxlen) {
1017                 buf[0] = 0;
1018                 read_buf(f, buf, 1);
1019                 if (buf[0] == 0)
1020                         return 0;
1021                 if (buf[0] == '\n') {
1022                         buf[0] = 0;
1023                         break;
1024                 }
1025                 if (buf[0] != '\r') {
1026                         buf++;
1027                         maxlen--;
1028                 }
1029         }
1030         if (maxlen == 0) {
1031                 *buf = 0;
1032                 return 0;
1033         }
1034
1035         return 1;
1036 }
1037
1038
1039 void io_printf(int fd, const char *format, ...)
1040 {
1041         va_list ap;  
1042         char buf[1024];
1043         int len;
1044
1045         va_start(ap, format);
1046         len = vsnprintf(buf, sizeof buf, format, ap);
1047         va_end(ap);
1048
1049         if (len < 0) exit_cleanup(RERR_STREAMIO);
1050
1051         write_sbuf(fd, buf);
1052 }
1053
1054
1055 /** Setup for multiplexing a MSG_* stream with the data stream. */
1056 void io_start_multiplex_out(int fd)
1057 {
1058         multiplex_out_fd = fd;
1059         io_flush(NORMAL_FLUSH);
1060         io_start_buffering_out(fd);
1061         io_multiplexing_out = 1;
1062 }
1063
1064 /** Setup for multiplexing a MSG_* stream with the data stream. */
1065 void io_start_multiplex_in(int fd)
1066 {
1067         multiplex_in_fd = fd;
1068         io_flush(NORMAL_FLUSH);
1069         io_multiplexing_in = 1;
1070 }
1071
1072 /** Write an message to the multiplexed data stream. */
1073 int io_multiplex_write(enum msgcode code, char *buf, size_t len)
1074 {
1075         if (!io_multiplexing_out) return 0;
1076
1077         io_flush(NORMAL_FLUSH);
1078         stats.total_written += (len+4);
1079         mplex_write(multiplex_out_fd, code, buf, len);
1080         return 1;
1081 }
1082
1083 /** Stop output multiplexing. */
1084 void io_multiplexing_close(void)
1085 {
1086         io_multiplexing_out = 0;
1087 }
1088