b21dff427ae2b10e328a56537038639d9c08a209
[rsync.git] / io.c
1 /*
2  * Socket and pipe I/O utilities used in rsync.
3  *
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-2008 Wayne Davison
8  *
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.
13  *
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.
18  *
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.
21  */
22
23 /* Rsync provides its own multiplexing system, which is used to send
24  * stderr and stdout over a single socket.
25  *
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(). */
29
30 #include "rsync.h"
31 #include "ifuncs.h"
32
33 /** If no timeout is specified then use a 60 second select timeout */
34 #define SELECT_TIMEOUT 60
35
36 extern int bwlimit;
37 extern size_t bwlimit_writemax;
38 extern int io_timeout;
39 extern int allowed_lull;
40 extern int am_server;
41 extern int am_daemon;
42 extern int am_sender;
43 extern int am_generator;
44 extern int inc_recurse;
45 extern int io_error;
46 extern int eol_nulls;
47 extern int flist_eof;
48 extern int list_only;
49 extern int read_batch;
50 extern int csum_length;
51 extern int protect_args;
52 extern int checksum_seed;
53 extern int protocol_version;
54 extern int remove_source_files;
55 extern int preserve_hard_links;
56 extern struct stats stats;
57 extern struct file_list *cur_flist;
58 #ifdef ICONV_OPTION
59 extern int filesfrom_convert;
60 extern iconv_t ic_send, ic_recv;
61 #endif
62
63 const char phase_unknown[] = "unknown";
64 int ignore_timeout = 0;
65 int batch_fd = -1;
66 int msgdone_cnt = 0;
67
68 /* Ignore an EOF error if non-zero. See whine_about_eof(). */
69 int kluge_around_eof = 0;
70
71 int msg_fd_in = -1;
72 int msg_fd_out = -1;
73 int sock_f_in = -1;
74 int sock_f_out = -1;
75
76 static int iobuf_f_in = -1;
77 static char *iobuf_in;
78 static size_t iobuf_in_siz;
79 static size_t iobuf_in_ndx;
80 static size_t iobuf_in_remaining;
81
82 static int iobuf_f_out = -1;
83 static char *iobuf_out;
84 static int iobuf_out_cnt;
85
86 int flist_forward_from = -1;
87
88 static int io_multiplexing_out;
89 static int io_multiplexing_in;
90 static time_t last_io_in;
91 static time_t last_io_out;
92 static int no_flush;
93
94 static int write_batch_monitor_in = -1;
95 static int write_batch_monitor_out = -1;
96
97 static int io_filesfrom_f_in = -1;
98 static int io_filesfrom_f_out = -1;
99 static xbuf ff_buf = EMPTY_XBUF;
100 static char ff_lastchar;
101 #ifdef ICONV_OPTION
102 static xbuf iconv_buf = EMPTY_XBUF;
103 #endif
104 static int defer_forwarding_messages = 0, defer_forwarding_keep = 0;
105 static int select_timeout = SELECT_TIMEOUT;
106 static int active_filecnt = 0;
107 static OFF_T active_bytecnt = 0;
108 static int first_message = 1;
109
110 static char int_byte_extra[64] = {
111         0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, /* (00 - 3F)/4 */
112         0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, /* (40 - 7F)/4 */
113         1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, /* (80 - BF)/4 */
114         2, 2, 2, 2, 2, 2, 2, 2, 3, 3, 3, 3, 4, 4, 5, 6, /* (C0 - FF)/4 */
115 };
116
117 #define REMOTE_OPTION_ERROR "rsync: on remote machine: -"
118 #define REMOTE_OPTION_ERROR2 ": unknown option"
119
120 enum festatus { FES_SUCCESS, FES_REDO, FES_NO_SEND };
121
122 static void readfd(int fd, char *buffer, size_t N);
123 static void writefd(int fd, const char *buf, size_t len);
124 static void writefd_unbuffered(int fd, const char *buf, size_t len);
125 static void mplex_write(int fd, enum msgcode code, const char *buf, size_t len, int convert);
126
127 struct flist_ndx_item {
128         struct flist_ndx_item *next;
129         int ndx;
130 };
131
132 struct flist_ndx_list {
133         struct flist_ndx_item *head, *tail;
134 };
135
136 static struct flist_ndx_list redo_list, hlink_list;
137
138 struct msg_list_item {
139         struct msg_list_item *next;
140         char convert;
141         char buf[1];
142 };
143
144 struct msg_list {
145         struct msg_list_item *head, *tail;
146 };
147
148 static struct msg_list msg_queue;
149
150 static void flist_ndx_push(struct flist_ndx_list *lp, int ndx)
151 {
152         struct flist_ndx_item *item;
153
154         if (!(item = new(struct flist_ndx_item)))
155                 out_of_memory("flist_ndx_push");
156         item->next = NULL;
157         item->ndx = ndx;
158         if (lp->tail)
159                 lp->tail->next = item;
160         else
161                 lp->head = item;
162         lp->tail = item;
163 }
164
165 static int flist_ndx_pop(struct flist_ndx_list *lp)
166 {
167         struct flist_ndx_item *next;
168         int ndx;
169
170         if (!lp->head)
171                 return -1;
172
173         ndx = lp->head->ndx;
174         next = lp->head->next;
175         free(lp->head);
176         lp->head = next;
177         if (!next)
178                 lp->tail = NULL;
179
180         return ndx;
181 }
182
183 static void got_flist_entry_status(enum festatus status, const char *buf)
184 {
185         int ndx = IVAL(buf, 0);
186         struct file_list *flist = flist_for_ndx(ndx);
187
188         assert(flist != NULL);
189
190         if (remove_source_files) {
191                 active_filecnt--;
192                 active_bytecnt -= F_LENGTH(flist->files[ndx - flist->ndx_start]);
193         }
194
195         if (inc_recurse)
196                 flist->in_progress--;
197
198         switch (status) {
199         case FES_SUCCESS:
200                 if (remove_source_files)
201                         send_msg(MSG_SUCCESS, buf, 4, 0);
202                 if (preserve_hard_links) {
203                         struct file_struct *file = flist->files[ndx - flist->ndx_start];
204                         if (F_IS_HLINKED(file)) {
205                                 flist_ndx_push(&hlink_list, ndx);
206                                 flist->in_progress++;
207                         }
208                 }
209                 break;
210         case FES_REDO:
211                 if (inc_recurse)
212                         flist->to_redo++;
213                 flist_ndx_push(&redo_list, ndx);
214                 break;
215         case FES_NO_SEND:
216                 break;
217         }
218 }
219
220 static void check_timeout(void)
221 {
222         time_t t;
223
224         if (!io_timeout || ignore_timeout)
225                 return;
226
227         if (!last_io_in) {
228                 last_io_in = time(NULL);
229                 return;
230         }
231
232         t = time(NULL);
233
234         if (t - last_io_in >= io_timeout) {
235                 if (!am_server && !am_daemon) {
236                         rprintf(FERROR, "io timeout after %d seconds -- exiting\n",
237                                 (int)(t-last_io_in));
238                 }
239                 exit_cleanup(RERR_TIMEOUT);
240         }
241 }
242
243 /* Note the fds used for the main socket (which might really be a pipe
244  * for a local transfer, but we can ignore that). */
245 void io_set_sock_fds(int f_in, int f_out)
246 {
247         sock_f_in = f_in;
248         sock_f_out = f_out;
249 }
250
251 void set_io_timeout(int secs)
252 {
253         io_timeout = secs;
254
255         if (!io_timeout || io_timeout > SELECT_TIMEOUT)
256                 select_timeout = SELECT_TIMEOUT;
257         else
258                 select_timeout = io_timeout;
259
260         allowed_lull = read_batch ? 0 : (io_timeout + 1) / 2;
261 }
262
263 /* Setup the fd used to receive MSG_* messages.  Only needed during the
264  * early stages of being a local sender (up through the sending of the
265  * file list) or when we're the generator (to fetch the messages from
266  * the receiver). */
267 void set_msg_fd_in(int fd)
268 {
269         msg_fd_in = fd;
270 }
271
272 /* Setup the fd used to send our MSG_* messages.  Only needed when
273  * we're the receiver (to send our messages to the generator). */
274 void set_msg_fd_out(int fd)
275 {
276         msg_fd_out = fd;
277         set_nonblocking(msg_fd_out);
278 }
279
280 /* Add a message to the pending MSG_* list. */
281 static void msg_list_add(struct msg_list *lst, int code, const char *buf, int len, int convert)
282 {
283         struct msg_list_item *m;
284         int sz = len + 4 + sizeof m[0] - 1;
285
286         if (!(m = (struct msg_list_item *)new_array(char, sz)))
287                 out_of_memory("msg_list_add");
288         m->next = NULL;
289         m->convert = convert;
290         SIVAL(m->buf, 0, ((code+MPLEX_BASE)<<24) | len);
291         memcpy(m->buf + 4, buf, len);
292         if (lst->tail)
293                 lst->tail->next = m;
294         else
295                 lst->head = m;
296         lst->tail = m;
297 }
298
299 static void msg_flush(void)
300 {
301         if (am_generator) {
302                 while (msg_queue.head && io_multiplexing_out) {
303                         struct msg_list_item *m = msg_queue.head;
304                         int len = IVAL(m->buf, 0) & 0xFFFFFF;
305                         int tag = *((uchar*)m->buf+3) - MPLEX_BASE;
306                         if (!(msg_queue.head = m->next))
307                                 msg_queue.tail = NULL;
308                         stats.total_written += len + 4;
309                         defer_forwarding_messages++;
310                         mplex_write(sock_f_out, tag, m->buf + 4, len, m->convert);
311                         defer_forwarding_messages--;
312                         free(m);
313                 }
314         } else {
315                 while (msg_queue.head) {
316                         struct msg_list_item *m = msg_queue.head;
317                         int len = IVAL(m->buf, 0) & 0xFFFFFF;
318                         int tag = *((uchar*)m->buf+3) - MPLEX_BASE;
319                         if (!(msg_queue.head = m->next))
320                                 msg_queue.tail = NULL;
321                         defer_forwarding_messages++;
322                         mplex_write(msg_fd_out, tag, m->buf + 4, len, m->convert);
323                         defer_forwarding_messages--;
324                         free(m);
325                 }
326         }
327 }
328
329 static void check_for_d_option_error(const char *msg)
330 {
331         static char rsync263_opts[] = "BCDHIKLPRSTWabceghlnopqrtuvxz";
332         char *colon;
333         int saw_d = 0;
334
335         if (*msg != 'r'
336          || strncmp(msg, REMOTE_OPTION_ERROR, sizeof REMOTE_OPTION_ERROR - 1) != 0)
337                 return;
338
339         msg += sizeof REMOTE_OPTION_ERROR - 1;
340         if (*msg == '-' || (colon = strchr(msg, ':')) == NULL
341          || strncmp(colon, REMOTE_OPTION_ERROR2, sizeof REMOTE_OPTION_ERROR2 - 1) != 0)
342                 return;
343
344         for ( ; *msg != ':'; msg++) {
345                 if (*msg == 'd')
346                         saw_d = 1;
347                 else if (*msg == 'e')
348                         break;
349                 else if (strchr(rsync263_opts, *msg) == NULL)
350                         return;
351         }
352
353         if (saw_d) {
354                 rprintf(FWARNING,
355                     "*** Try using \"--old-d\" if remote rsync is <= 2.6.3 ***\n");
356         }
357 }
358
359 /* Read a message from the MSG_* fd and handle it.  This is called either
360  * during the early stages of being a local sender (up through the sending
361  * of the file list) or when we're the generator (to fetch the messages
362  * from the receiver). */
363 static void read_msg_fd(void)
364 {
365         char buf[2048];
366         size_t n;
367         struct file_list *flist;
368         int fd = msg_fd_in;
369         int tag, len;
370
371         /* Temporarily disable msg_fd_in.  This is needed to avoid looping back
372          * to this routine from writefd_unbuffered(). */
373         no_flush++;
374         msg_fd_in = -1;
375         defer_forwarding_messages++;
376
377         readfd(fd, buf, 4);
378         tag = IVAL(buf, 0);
379
380         len = tag & 0xFFFFFF;
381         tag = (tag >> 24) - MPLEX_BASE;
382
383         switch (tag) {
384         case MSG_DONE:
385                 if (len < 0 || len > 1 || !am_generator) {
386                   invalid_msg:
387                         rprintf(FERROR, "invalid message %d:%d [%s%s]\n",
388                                 tag, len, who_am_i(),
389                                 inc_recurse ? "/inc" : "");
390                         exit_cleanup(RERR_STREAMIO);
391                 }
392                 if (len) {
393                         readfd(fd, buf, len);
394                         stats.total_read = read_varlong(fd, 3);
395                 }
396                 msgdone_cnt++;
397                 break;
398         case MSG_REDO:
399                 if (len != 4 || !am_generator)
400                         goto invalid_msg;
401                 readfd(fd, buf, 4);
402                 got_flist_entry_status(FES_REDO, buf);
403                 break;
404         case MSG_FLIST:
405                 if (len != 4 || !am_generator || !inc_recurse)
406                         goto invalid_msg;
407                 readfd(fd, buf, 4);
408                 /* Read extra file list from receiver. */
409                 assert(iobuf_in != NULL);
410                 assert(iobuf_f_in == fd);
411                 if (DEBUG_GTE(FLIST, 2)) {
412                         rprintf(FINFO, "[%s] receiving flist for dir %d\n",
413                                 who_am_i(), IVAL(buf,0));
414                 }
415                 flist = recv_file_list(fd);
416                 flist->parent_ndx = IVAL(buf,0);
417 #ifdef SUPPORT_HARD_LINKS
418                 if (preserve_hard_links)
419                         match_hard_links(flist);
420 #endif
421                 break;
422         case MSG_FLIST_EOF:
423                 if (len != 0 || !am_generator || !inc_recurse)
424                         goto invalid_msg;
425                 flist_eof = 1;
426                 break;
427         case MSG_IO_ERROR:
428                 if (len != 4)
429                         goto invalid_msg;
430                 readfd(fd, buf, len);
431                 io_error |= IVAL(buf, 0);
432                 break;
433         case MSG_DELETED:
434                 if (len >= (int)sizeof buf || !am_generator)
435                         goto invalid_msg;
436                 readfd(fd, buf, len);
437                 send_msg(MSG_DELETED, buf, len, 1);
438                 break;
439         case MSG_SUCCESS:
440                 if (len != 4 || !am_generator)
441                         goto invalid_msg;
442                 readfd(fd, buf, 4);
443                 got_flist_entry_status(FES_SUCCESS, buf);
444                 break;
445         case MSG_NO_SEND:
446                 if (len != 4 || !am_generator)
447                         goto invalid_msg;
448                 readfd(fd, buf, 4);
449                 got_flist_entry_status(FES_NO_SEND, buf);
450                 break;
451         case MSG_ERROR_SOCKET:
452         case MSG_CLIENT:
453                 if (!am_generator)
454                         goto invalid_msg;
455                 if (tag == MSG_ERROR_SOCKET)
456                         io_end_multiplex_out();
457                 /* FALL THROUGH */
458         case MSG_INFO:
459         case MSG_ERROR:
460         case MSG_ERROR_XFER:
461         case MSG_WARNING:
462         case MSG_LOG:
463                 while (len) {
464                         n = len;
465                         if (n >= sizeof buf)
466                                 n = sizeof buf - 1;
467                         readfd(fd, buf, n);
468                         rwrite((enum logcode)tag, buf, n, !am_generator);
469                         len -= n;
470                 }
471                 break;
472         default:
473                 rprintf(FERROR, "unknown message %d:%d [%s]\n",
474                         tag, len, who_am_i());
475                 exit_cleanup(RERR_STREAMIO);
476         }
477
478         no_flush--;
479         msg_fd_in = fd;
480         if (!--defer_forwarding_messages && !no_flush)
481                 msg_flush();
482 }
483
484 /* This is used by the generator to limit how many file transfers can
485  * be active at once when --remove-source-files is specified.  Without
486  * this, sender-side deletions were mostly happening at the end. */
487 void increment_active_files(int ndx, int itemizing, enum logcode code)
488 {
489         /* TODO: tune these limits? */
490         while (active_filecnt >= (active_bytecnt >= 128*1024 ? 10 : 50)) {
491                 check_for_finished_files(itemizing, code, 0);
492                 if (iobuf_out_cnt)
493                         io_flush(NORMAL_FLUSH);
494                 else
495                         read_msg_fd();
496         }
497
498         active_filecnt++;
499         active_bytecnt += F_LENGTH(cur_flist->files[ndx - cur_flist->ndx_start]);
500 }
501
502 /* Write an message to a multiplexed stream. If this fails, rsync exits. */
503 static void mplex_write(int fd, enum msgcode code, const char *buf, size_t len, int convert)
504 {
505         char buffer[BIGPATHBUFLEN]; /* Oversized for use by iconv code. */
506         size_t n = len;
507
508 #ifdef ICONV_OPTION
509         /* We need to convert buf before doing anything else so that we
510          * can include the (converted) byte length in the message header. */
511         if (convert && ic_send != (iconv_t)-1) {
512                 xbuf outbuf, inbuf;
513
514                 INIT_XBUF(outbuf, buffer + 4, 0, sizeof buffer - 4);
515                 INIT_XBUF(inbuf, (char*)buf, len, -1);
516
517                 iconvbufs(ic_send, &inbuf, &outbuf,
518                           ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE);
519                 if (inbuf.len > 0) {
520                         rprintf(FERROR, "overflowed conversion buffer in mplex_write");
521                         exit_cleanup(RERR_UNSUPPORTED);
522                 }
523
524                 n = len = outbuf.len;
525         } else
526 #endif
527         if (n > 1024 - 4) /* BIGPATHBUFLEN can handle 1024 bytes */
528                 n = 0;    /* We'd rather do 2 writes than too much memcpy(). */
529         else
530                 memcpy(buffer + 4, buf, n);
531
532         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
533
534         defer_forwarding_keep = 1; /* defer_forwarding_messages++ on return */
535         writefd_unbuffered(fd, buffer, n+4);
536         defer_forwarding_keep = 0;
537
538         if (len > n)
539                 writefd_unbuffered(fd, buf+n, len-n);
540
541         if (!--defer_forwarding_messages && !no_flush)
542                 msg_flush();
543 }
544
545 int send_msg(enum msgcode code, const char *buf, int len, int convert)
546 {
547         if (msg_fd_out < 0) {
548                 if (!defer_forwarding_messages)
549                         return io_multiplex_write(code, buf, len, convert);
550                 if (!io_multiplexing_out)
551                         return 0;
552                 msg_list_add(&msg_queue, code, buf, len, convert);
553                 return 1;
554         }
555         if (flist_forward_from >= 0)
556                 msg_list_add(&msg_queue, code, buf, len, convert);
557         else
558                 mplex_write(msg_fd_out, code, buf, len, convert);
559         return 1;
560 }
561
562 void send_msg_int(enum msgcode code, int num)
563 {
564         char numbuf[4];
565         SIVAL(numbuf, 0, num);
566         send_msg(code, numbuf, 4, 0);
567 }
568
569 void wait_for_receiver(void)
570 {
571         if (iobuf_out_cnt)
572                 io_flush(NORMAL_FLUSH);
573         else
574                 read_msg_fd();
575 }
576
577 int get_redo_num(void)
578 {
579         return flist_ndx_pop(&redo_list);
580 }
581
582 int get_hlink_num(void)
583 {
584         return flist_ndx_pop(&hlink_list);
585 }
586
587 /**
588  * When we're the receiver and we have a local --files-from list of names
589  * that needs to be sent over the socket to the sender, we have to do two
590  * things at the same time: send the sender a list of what files we're
591  * processing and read the incoming file+info list from the sender.  We do
592  * this by augmenting the read_timeout() function to copy this data.  It
593  * uses ff_buf to read a block of data from f_in (when it is ready, since
594  * it might be a pipe) and then blast it out f_out (when it is ready to
595  * receive more data).
596  */
597 void io_set_filesfrom_fds(int f_in, int f_out)
598 {
599         io_filesfrom_f_in = f_in;
600         io_filesfrom_f_out = f_out;
601         alloc_xbuf(&ff_buf, 2048);
602 #ifdef ICONV_OPTION
603         if (protect_args)
604                 alloc_xbuf(&iconv_buf, 1024);
605 #endif
606 }
607
608 /* It's almost always an error to get an EOF when we're trying to read from the
609  * network, because the protocol is (for the most part) self-terminating.
610  *
611  * There is one case for the receiver when it is at the end of the transfer
612  * (hanging around reading any keep-alive packets that might come its way): if
613  * the sender dies before the generator's kill-signal comes through, we can end
614  * up here needing to loop until the kill-signal arrives.  In this situation,
615  * kluge_around_eof will be < 0.
616  *
617  * There is another case for older protocol versions (< 24) where the module
618  * listing was not terminated, so we must ignore an EOF error in that case and
619  * exit.  In this situation, kluge_around_eof will be > 0. */
620 static void whine_about_eof(int fd)
621 {
622         if (kluge_around_eof && fd == sock_f_in) {
623                 int i;
624                 if (kluge_around_eof > 0)
625                         exit_cleanup(0);
626                 /* If we're still here after 10 seconds, exit with an error. */
627                 for (i = 10*1000/20; i--; )
628                         msleep(20);
629         }
630
631         rprintf(FERROR, RSYNC_NAME ": connection unexpectedly closed "
632                 "(%.0f bytes received so far) [%s]\n",
633                 (double)stats.total_read, who_am_i());
634
635         exit_cleanup(RERR_STREAMIO);
636 }
637
638 /**
639  * Read from a socket with I/O timeout. return the number of bytes
640  * read. If no bytes can be read then exit, never return a number <= 0.
641  *
642  * TODO: If the remote shell connection fails, then current versions
643  * actually report an "unexpected EOF" error here.  Since it's a
644  * fairly common mistake to try to use rsh when ssh is required, we
645  * should trap that: if we fail to read any data at all, we should
646  * give a better explanation.  We can tell whether the connection has
647  * started by looking e.g. at whether the remote version is known yet.
648  */
649 static int read_timeout(int fd, char *buf, size_t len)
650 {
651         int n, cnt = 0;
652
653         io_flush(FULL_FLUSH);
654
655         while (cnt == 0) {
656                 /* until we manage to read *something* */
657                 fd_set r_fds, w_fds;
658                 struct timeval tv;
659                 int maxfd = fd;
660                 int count;
661
662                 FD_ZERO(&r_fds);
663                 FD_ZERO(&w_fds);
664                 FD_SET(fd, &r_fds);
665                 if (io_filesfrom_f_out >= 0) {
666                         int new_fd;
667                         if (ff_buf.len == 0) {
668                                 if (io_filesfrom_f_in >= 0) {
669                                         FD_SET(io_filesfrom_f_in, &r_fds);
670                                         new_fd = io_filesfrom_f_in;
671                                 } else {
672                                         io_filesfrom_f_out = -1;
673                                         new_fd = -1;
674                                 }
675                         } else {
676                                 FD_SET(io_filesfrom_f_out, &w_fds);
677                                 new_fd = io_filesfrom_f_out;
678                         }
679                         if (new_fd > maxfd)
680                                 maxfd = new_fd;
681                 }
682
683                 tv.tv_sec = select_timeout;
684                 tv.tv_usec = 0;
685
686                 errno = 0;
687
688                 count = select(maxfd + 1, &r_fds, &w_fds, NULL, &tv);
689
690                 if (count <= 0) {
691                         if (errno == EBADF) {
692                                 defer_forwarding_messages = 0;
693                                 exit_cleanup(RERR_SOCKETIO);
694                         }
695                         check_timeout();
696                         continue;
697                 }
698
699                 if (io_filesfrom_f_out >= 0) {
700                         if (ff_buf.len) {
701                                 if (FD_ISSET(io_filesfrom_f_out, &w_fds)) {
702                                         int l = write(io_filesfrom_f_out,
703                                                       ff_buf.buf + ff_buf.pos,
704                                                       ff_buf.len);
705                                         if (l > 0) {
706                                                 if (!(ff_buf.len -= l))
707                                                         ff_buf.pos = 0;
708                                                 else
709                                                         ff_buf.pos += l;
710                                         } else if (errno != EINTR) {
711                                                 /* XXX should we complain? */
712                                                 io_filesfrom_f_out = -1;
713                                         }
714                                 }
715                         } else if (io_filesfrom_f_in >= 0) {
716                                 if (FD_ISSET(io_filesfrom_f_in, &r_fds)) {
717 #ifdef ICONV_OPTION
718                                         xbuf *ibuf = filesfrom_convert ? &iconv_buf : &ff_buf;
719 #else
720                                         xbuf *ibuf = &ff_buf;
721 #endif
722                                         int l = read(io_filesfrom_f_in, ibuf->buf, ibuf->size);
723                                         if (l <= 0) {
724                                                 if (l == 0 || errno != EINTR) {
725                                                         /* Send end-of-file marker */
726                                                         memcpy(ff_buf.buf, "\0\0", 2);
727                                                         ff_buf.len = ff_lastchar? 2 : 1;
728                                                         ff_buf.pos = 0;
729                                                         io_filesfrom_f_in = -1;
730                                                 }
731                                         } else {
732 #ifdef ICONV_OPTION
733                                                 if (filesfrom_convert) {
734                                                         iconv_buf.pos = 0;
735                                                         iconv_buf.len = l;
736                                                         iconvbufs(ic_send, &iconv_buf, &ff_buf,
737                                                             ICB_EXPAND_OUT|ICB_INCLUDE_BAD|ICB_INCLUDE_INCOMPLETE);
738                                                         l = ff_buf.len;
739                                                 }
740 #endif
741                                                 if (!eol_nulls) {
742                                                         char *s = ff_buf.buf + l;
743                                                         /* Transform CR and/or LF into '\0' */
744                                                         while (s-- > ff_buf.buf) {
745                                                                 if (*s == '\n' || *s == '\r')
746                                                                         *s = '\0';
747                                                         }
748                                                 }
749                                                 if (!ff_lastchar) {
750                                                         /* Last buf ended with a '\0', so don't
751                                                          * let this buf start with one. */
752                                                         while (l && ff_buf.buf[ff_buf.pos] == '\0')
753                                                                 ff_buf.pos++, l--;
754                                                 }
755                                                 if (!l)
756                                                         ff_buf.pos = 0;
757                                                 else {
758                                                         char *f = ff_buf.buf + ff_buf.pos;
759                                                         char *t = f;
760                                                         char *eob = f + l;
761                                                         /* Eliminate any multi-'\0' runs. */
762                                                         while (f != eob) {
763                                                                 if (!(*t++ = *f++)) {
764                                                                         while (f != eob && !*f)
765                                                                                 f++, l--;
766                                                                 }
767                                                         }
768                                                         ff_lastchar = f[-1];
769                                                 }
770                                                 ff_buf.len = l;
771                                         }
772                                 }
773                         }
774                 }
775
776                 if (!FD_ISSET(fd, &r_fds))
777                         continue;
778
779                 n = read(fd, buf, len);
780
781                 if (n <= 0) {
782                         if (n == 0)
783                                 whine_about_eof(fd); /* Doesn't return. */
784                         if (errno == EINTR || errno == EWOULDBLOCK
785                             || errno == EAGAIN)
786                                 continue;
787
788                         /* Don't write errors on a dead socket. */
789                         if (fd == sock_f_in) {
790                                 io_end_multiplex_out();
791                                 rsyserr(FERROR_SOCKET, errno, "read error");
792                         } else
793                                 rsyserr(FERROR, errno, "read error");
794                         exit_cleanup(RERR_STREAMIO);
795                 }
796
797                 buf += n;
798                 len -= n;
799                 cnt += n;
800
801                 if (fd == sock_f_in && io_timeout)
802                         last_io_in = time(NULL);
803         }
804
805         return cnt;
806 }
807
808 /* Read a line into the "buf" buffer. */
809 int read_line(int fd, char *buf, size_t bufsiz, int flags)
810 {
811         char ch, *s, *eob;
812         int cnt;
813
814 #ifdef ICONV_OPTION
815         if (flags & RL_CONVERT && iconv_buf.size < bufsiz)
816                 realloc_xbuf(&iconv_buf, bufsiz + 1024);
817 #endif
818
819   start:
820 #ifdef ICONV_OPTION
821         s = flags & RL_CONVERT ? iconv_buf.buf : buf;
822 #else
823         s = buf;
824 #endif
825         eob = s + bufsiz - 1;
826         while (1) {
827                 cnt = read(fd, &ch, 1);
828                 if (cnt < 0 && (errno == EWOULDBLOCK
829                   || errno == EINTR || errno == EAGAIN)) {
830                         struct timeval tv;
831                         fd_set r_fds, e_fds;
832                         FD_ZERO(&r_fds);
833                         FD_SET(fd, &r_fds);
834                         FD_ZERO(&e_fds);
835                         FD_SET(fd, &e_fds);
836                         tv.tv_sec = select_timeout;
837                         tv.tv_usec = 0;
838                         if (!select(fd+1, &r_fds, NULL, &e_fds, &tv))
839                                 check_timeout();
840                         /*if (FD_ISSET(fd, &e_fds))
841                                 rprintf(FINFO, "select exception on fd %d\n", fd); */
842                         continue;
843                 }
844                 if (cnt != 1)
845                         break;
846                 if (flags & RL_EOL_NULLS ? ch == '\0' : (ch == '\r' || ch == '\n')) {
847                         /* Skip empty lines if dumping comments. */
848                         if (flags & RL_DUMP_COMMENTS && s == buf)
849                                 continue;
850                         break;
851                 }
852                 if (s < eob)
853                         *s++ = ch;
854         }
855         *s = '\0';
856
857         if (flags & RL_DUMP_COMMENTS && (*buf == '#' || *buf == ';'))
858                 goto start;
859
860 #ifdef ICONV_OPTION
861         if (flags & RL_CONVERT) {
862                 xbuf outbuf;
863                 INIT_XBUF(outbuf, buf, 0, bufsiz);
864                 iconv_buf.pos = 0;
865                 iconv_buf.len = s - iconv_buf.buf;
866                 iconvbufs(ic_recv, &iconv_buf, &outbuf,
867                           ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE);
868                 outbuf.buf[outbuf.len] = '\0';
869                 return outbuf.len;
870         }
871 #endif
872
873         return s - buf;
874 }
875
876 void read_args(int f_in, char *mod_name, char *buf, size_t bufsiz, int rl_nulls,
877                char ***argv_p, int *argc_p, char **request_p)
878 {
879         int maxargs = MAX_ARGS;
880         int dot_pos = 0;
881         int argc = 0;
882         char **argv, *p;
883         int rl_flags = (rl_nulls ? RL_EOL_NULLS : 0);
884
885 #ifdef ICONV_OPTION
886         rl_flags |= (protect_args && ic_recv != (iconv_t)-1 ? RL_CONVERT : 0);
887 #endif
888
889         if (!(argv = new_array(char *, maxargs)))
890                 out_of_memory("read_args");
891         if (mod_name && !protect_args)
892                 argv[argc++] = "rsyncd";
893
894         while (1) {
895                 if (read_line(f_in, buf, bufsiz, rl_flags) == 0)
896                         break;
897
898                 if (argc == maxargs-1) {
899                         maxargs += MAX_ARGS;
900                         if (!(argv = realloc_array(argv, char *, maxargs)))
901                                 out_of_memory("read_args");
902                 }
903
904                 if (dot_pos) {
905                         if (request_p) {
906                                 *request_p = strdup(buf);
907                                 request_p = NULL;
908                         }
909                         if (mod_name)
910                                 glob_expand_module(mod_name, buf, &argv, &argc, &maxargs);
911                         else
912                                 glob_expand(buf, &argv, &argc, &maxargs);
913                 } else {
914                         if (!(p = strdup(buf)))
915                                 out_of_memory("read_args");
916                         argv[argc++] = p;
917                         if (*p == '.' && p[1] == '\0')
918                                 dot_pos = argc;
919                 }
920         }
921         argv[argc] = NULL;
922
923         glob_expand(NULL, NULL, NULL, NULL);
924
925         *argc_p = argc;
926         *argv_p = argv;
927 }
928
929 int io_start_buffering_out(int f_out)
930 {
931         if (iobuf_out) {
932                 assert(f_out == iobuf_f_out);
933                 return 0;
934         }
935         if (!(iobuf_out = new_array(char, IO_BUFFER_SIZE)))
936                 out_of_memory("io_start_buffering_out");
937         iobuf_out_cnt = 0;
938         iobuf_f_out = f_out;
939         return 1;
940 }
941
942 int io_start_buffering_in(int f_in)
943 {
944         if (iobuf_in) {
945                 assert(f_in == iobuf_f_in);
946                 return 0;
947         }
948         iobuf_in_siz = 2 * IO_BUFFER_SIZE;
949         if (!(iobuf_in = new_array(char, iobuf_in_siz)))
950                 out_of_memory("io_start_buffering_in");
951         iobuf_f_in = f_in;
952         return 1;
953 }
954
955 void io_end_buffering_in(void)
956 {
957         if (!iobuf_in)
958                 return;
959         free(iobuf_in);
960         iobuf_in = NULL;
961         iobuf_in_ndx = 0;
962         iobuf_in_remaining = 0;
963         iobuf_f_in = -1;
964 }
965
966 void io_end_buffering_out(void)
967 {
968         if (!iobuf_out)
969                 return;
970         io_flush(FULL_FLUSH);
971         free(iobuf_out);
972         iobuf_out = NULL;
973         iobuf_f_out = -1;
974 }
975
976 void maybe_flush_socket(int important)
977 {
978         if (iobuf_out && iobuf_out_cnt
979          && (important || time(NULL) - last_io_out >= 5))
980                 io_flush(NORMAL_FLUSH);
981 }
982
983 void maybe_send_keepalive(void)
984 {
985         if (time(NULL) - last_io_out >= allowed_lull) {
986                 if (!iobuf_out || !iobuf_out_cnt) {
987                         if (protocol_version < 29)
988                                 return; /* there's nothing we can do */
989                         if (protocol_version >= 30)
990                                 send_msg(MSG_NOOP, "", 0, 0);
991                         else {
992                                 write_int(sock_f_out, cur_flist->used);
993                                 write_shortint(sock_f_out, ITEM_IS_NEW);
994                         }
995                 }
996                 if (iobuf_out)
997                         io_flush(NORMAL_FLUSH);
998         }
999 }
1000
1001 void start_flist_forward(int f_in)
1002 {
1003         assert(iobuf_out != NULL);
1004         assert(iobuf_f_out == msg_fd_out);
1005         flist_forward_from = f_in;
1006 }
1007
1008 void stop_flist_forward()
1009 {
1010         flist_forward_from = -1;
1011         io_flush(FULL_FLUSH);
1012 }
1013
1014 /**
1015  * Continue trying to read len bytes - don't return until len has been
1016  * read.
1017  **/
1018 static void read_loop(int fd, char *buf, size_t len)
1019 {
1020         while (len) {
1021                 int n = read_timeout(fd, buf, len);
1022
1023                 buf += n;
1024                 len -= n;
1025         }
1026 }
1027
1028 /**
1029  * Read from the file descriptor handling multiplexing - return number
1030  * of bytes read.
1031  *
1032  * Never returns <= 0.
1033  */
1034 static int readfd_unbuffered(int fd, char *buf, size_t len)
1035 {
1036         size_t msg_bytes;
1037         int tag, cnt = 0;
1038         char line[BIGPATHBUFLEN];
1039
1040         if (!iobuf_in || fd != iobuf_f_in)
1041                 return read_timeout(fd, buf, len);
1042
1043         if (!io_multiplexing_in && iobuf_in_remaining == 0) {
1044                 iobuf_in_remaining = read_timeout(fd, iobuf_in, iobuf_in_siz);
1045                 iobuf_in_ndx = 0;
1046         }
1047
1048         while (cnt == 0) {
1049                 if (iobuf_in_remaining) {
1050                         len = MIN(len, iobuf_in_remaining);
1051                         memcpy(buf, iobuf_in + iobuf_in_ndx, len);
1052                         iobuf_in_ndx += len;
1053                         iobuf_in_remaining -= len;
1054                         cnt = len;
1055                         break;
1056                 }
1057
1058                 read_loop(fd, line, 4);
1059                 tag = IVAL(line, 0);
1060
1061                 msg_bytes = tag & 0xFFFFFF;
1062                 tag = (tag >> 24) - MPLEX_BASE;
1063
1064                 switch (tag) {
1065                 case MSG_DATA:
1066                         if (msg_bytes > iobuf_in_siz) {
1067                                 if (!(iobuf_in = realloc_array(iobuf_in, char,
1068                                                                msg_bytes)))
1069                                         out_of_memory("readfd_unbuffered");
1070                                 iobuf_in_siz = msg_bytes;
1071                         }
1072                         read_loop(fd, iobuf_in, msg_bytes);
1073                         iobuf_in_remaining = msg_bytes;
1074                         iobuf_in_ndx = 0;
1075                         break;
1076                 case MSG_NOOP:
1077                         if (am_sender)
1078                                 maybe_send_keepalive();
1079                         break;
1080                 case MSG_IO_ERROR:
1081                         if (msg_bytes != 4)
1082                                 goto invalid_msg;
1083                         read_loop(fd, line, msg_bytes);
1084                         send_msg_int(MSG_IO_ERROR, IVAL(line, 0));
1085                         io_error |= IVAL(line, 0);
1086                         break;
1087                 case MSG_DELETED:
1088                         if (msg_bytes >= sizeof line)
1089                                 goto overflow;
1090 #ifdef ICONV_OPTION
1091                         if (ic_recv != (iconv_t)-1) {
1092                                 xbuf outbuf, inbuf;
1093                                 char ibuf[512];
1094                                 int add_null = 0;
1095                                 int pos = 0;
1096
1097                                 INIT_CONST_XBUF(outbuf, line);
1098                                 INIT_XBUF(inbuf, ibuf, 0, -1);
1099
1100                                 while (msg_bytes) {
1101                                         inbuf.len = msg_bytes > sizeof ibuf
1102                                                   ? sizeof ibuf : msg_bytes;
1103                                         read_loop(fd, inbuf.buf, inbuf.len);
1104                                         if (!(msg_bytes -= inbuf.len)
1105                                          && !ibuf[inbuf.len-1])
1106                                                 inbuf.len--, add_null = 1;
1107                                         if (iconvbufs(ic_send, &inbuf, &outbuf,
1108                                             ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE) < 0)
1109                                                 goto overflow;
1110                                         pos = -1;
1111                                 }
1112                                 if (add_null) {
1113                                         if (outbuf.len == outbuf.size)
1114                                                 goto overflow;
1115                                         outbuf.buf[outbuf.len++] = '\0';
1116                                 }
1117                                 msg_bytes = outbuf.len;
1118                         } else
1119 #endif
1120                                 read_loop(fd, line, msg_bytes);
1121                         /* A directory name was sent with the trailing null */
1122                         if (msg_bytes > 0 && !line[msg_bytes-1])
1123                                 log_delete(line, S_IFDIR);
1124                         else {
1125                                 line[msg_bytes] = '\0';
1126                                 log_delete(line, S_IFREG);
1127                         }
1128                         break;
1129                 case MSG_SUCCESS:
1130                         if (msg_bytes != 4) {
1131                           invalid_msg:
1132                                 rprintf(FERROR, "invalid multi-message %d:%ld [%s]\n",
1133                                         tag, (long)msg_bytes, who_am_i());
1134                                 exit_cleanup(RERR_STREAMIO);
1135                         }
1136                         read_loop(fd, line, msg_bytes);
1137                         successful_send(IVAL(line, 0));
1138                         break;
1139                 case MSG_NO_SEND:
1140                         if (msg_bytes != 4)
1141                                 goto invalid_msg;
1142                         read_loop(fd, line, msg_bytes);
1143                         send_msg_int(MSG_NO_SEND, IVAL(line, 0));
1144                         break;
1145                 case MSG_INFO:
1146                 case MSG_ERROR:
1147                 case MSG_ERROR_XFER:
1148                 case MSG_WARNING:
1149                         if (msg_bytes >= sizeof line) {
1150                             overflow:
1151                                 rprintf(FERROR,
1152                                         "multiplexing overflow %d:%ld [%s]\n",
1153                                         tag, (long)msg_bytes, who_am_i());
1154                                 exit_cleanup(RERR_STREAMIO);
1155                         }
1156                         read_loop(fd, line, msg_bytes);
1157                         rwrite((enum logcode)tag, line, msg_bytes, 1);
1158                         if (first_message) {
1159                                 if (list_only && !am_sender && tag == 1) {
1160                                         line[msg_bytes] = '\0';
1161                                         check_for_d_option_error(line);
1162                                 }
1163                                 first_message = 0;
1164                         }
1165                         break;
1166                 default:
1167                         rprintf(FERROR, "unexpected tag %d [%s]\n",
1168                                 tag, who_am_i());
1169                         exit_cleanup(RERR_STREAMIO);
1170                 }
1171         }
1172
1173         if (iobuf_in_remaining == 0)
1174                 io_flush(NORMAL_FLUSH);
1175
1176         return cnt;
1177 }
1178
1179 /* Do a buffered read from fd.  Don't return until all N bytes have
1180  * been read.  If all N can't be read then exit with an error. */
1181 static void readfd(int fd, char *buffer, size_t N)
1182 {
1183         int  cnt;
1184         size_t total = 0;
1185
1186         while (total < N) {
1187                 cnt = readfd_unbuffered(fd, buffer + total, N-total);
1188                 total += cnt;
1189         }
1190
1191         if (fd == write_batch_monitor_in) {
1192                 if ((size_t)write(batch_fd, buffer, total) != total)
1193                         exit_cleanup(RERR_FILEIO);
1194         }
1195
1196         if (fd == flist_forward_from)
1197                 writefd(iobuf_f_out, buffer, total);
1198
1199         if (fd == sock_f_in)
1200                 stats.total_read += total;
1201 }
1202
1203 unsigned short read_shortint(int f)
1204 {
1205         char b[2];
1206         readfd(f, b, 2);
1207         return (UVAL(b, 1) << 8) + UVAL(b, 0);
1208 }
1209
1210 int32 read_int(int f)
1211 {
1212         char b[4];
1213         int32 num;
1214
1215         readfd(f, b, 4);
1216         num = IVAL(b, 0);
1217 #if SIZEOF_INT32 > 4
1218         if (num & (int32)0x80000000)
1219                 num |= ~(int32)0xffffffff;
1220 #endif
1221         return num;
1222 }
1223
1224 int32 read_varint(int f)
1225 {
1226         union {
1227             char b[5];
1228             int32 x;
1229         } u;
1230         uchar ch;
1231         int extra;
1232
1233         u.x = 0;
1234         readfd(f, (char*)&ch, 1);
1235         extra = int_byte_extra[ch / 4];
1236         if (extra) {
1237                 uchar bit = ((uchar)1<<(8-extra));
1238                 if (extra >= (int)sizeof u.b) {
1239                         rprintf(FERROR, "Overflow in read_varint()\n");
1240                         exit_cleanup(RERR_STREAMIO);
1241                 }
1242                 readfd(f, u.b, extra);
1243                 u.b[extra] = ch & (bit-1);
1244         } else
1245                 u.b[0] = ch;
1246 #if CAREFUL_ALIGNMENT
1247         u.x = IVAL(u.b,0);
1248 #endif
1249 #if SIZEOF_INT32 > 4
1250         if (u.x & (int32)0x80000000)
1251                 u.x |= ~(int32)0xffffffff;
1252 #endif
1253         return u.x;
1254 }
1255
1256 int64 read_varlong(int f, uchar min_bytes)
1257 {
1258         union {
1259             char b[9];
1260             int64 x;
1261         } u;
1262         char b2[8];
1263         int extra;
1264
1265 #if SIZEOF_INT64 < 8
1266         memset(u.b, 0, 8);
1267 #else
1268         u.x = 0;
1269 #endif
1270         readfd(f, b2, min_bytes);
1271         memcpy(u.b, b2+1, min_bytes-1);
1272         extra = int_byte_extra[CVAL(b2, 0) / 4];
1273         if (extra) {
1274                 uchar bit = ((uchar)1<<(8-extra));
1275                 if (min_bytes + extra > (int)sizeof u.b) {
1276                         rprintf(FERROR, "Overflow in read_varlong()\n");
1277                         exit_cleanup(RERR_STREAMIO);
1278                 }
1279                 readfd(f, u.b + min_bytes - 1, extra);
1280                 u.b[min_bytes + extra - 1] = CVAL(b2, 0) & (bit-1);
1281 #if SIZEOF_INT64 < 8
1282                 if (min_bytes + extra > 5 || u.b[4] || CVAL(u.b,3) & 0x80) {
1283                         rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
1284                         exit_cleanup(RERR_UNSUPPORTED);
1285                 }
1286 #endif
1287         } else
1288                 u.b[min_bytes + extra - 1] = CVAL(b2, 0);
1289 #if SIZEOF_INT64 < 8
1290         u.x = IVAL(u.b,0);
1291 #elif CAREFUL_ALIGNMENT
1292         u.x = IVAL(u.b,0) | (((int64)IVAL(u.b,4))<<32);
1293 #endif
1294         return u.x;
1295 }
1296
1297 int64 read_longint(int f)
1298 {
1299 #if SIZEOF_INT64 >= 8
1300         char b[9];
1301 #endif
1302         int32 num = read_int(f);
1303
1304         if (num != (int32)0xffffffff)
1305                 return num;
1306
1307 #if SIZEOF_INT64 < 8
1308         rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
1309         exit_cleanup(RERR_UNSUPPORTED);
1310 #else
1311         readfd(f, b, 8);
1312         return IVAL(b,0) | (((int64)IVAL(b,4))<<32);
1313 #endif
1314 }
1315
1316 void read_buf(int f, char *buf, size_t len)
1317 {
1318         readfd(f,buf,len);
1319 }
1320
1321 void read_sbuf(int f, char *buf, size_t len)
1322 {
1323         readfd(f, buf, len);
1324         buf[len] = '\0';
1325 }
1326
1327 uchar read_byte(int f)
1328 {
1329         uchar c;
1330         readfd(f, (char *)&c, 1);
1331         return c;
1332 }
1333
1334 int read_vstring(int f, char *buf, int bufsize)
1335 {
1336         int len = read_byte(f);
1337
1338         if (len & 0x80)
1339                 len = (len & ~0x80) * 0x100 + read_byte(f);
1340
1341         if (len >= bufsize) {
1342                 rprintf(FERROR, "over-long vstring received (%d > %d)\n",
1343                         len, bufsize - 1);
1344                 return -1;
1345         }
1346
1347         if (len)
1348                 readfd(f, buf, len);
1349         buf[len] = '\0';
1350         return len;
1351 }
1352
1353 /* Populate a sum_struct with values from the socket.  This is
1354  * called by both the sender and the receiver. */
1355 void read_sum_head(int f, struct sum_struct *sum)
1356 {
1357         int32 max_blength = protocol_version < 30 ? OLD_MAX_BLOCK_SIZE : MAX_BLOCK_SIZE;
1358         sum->count = read_int(f);
1359         if (sum->count < 0) {
1360                 rprintf(FERROR, "Invalid checksum count %ld [%s]\n",
1361                         (long)sum->count, who_am_i());
1362                 exit_cleanup(RERR_PROTOCOL);
1363         }
1364         sum->blength = read_int(f);
1365         if (sum->blength < 0 || sum->blength > max_blength) {
1366                 rprintf(FERROR, "Invalid block length %ld [%s]\n",
1367                         (long)sum->blength, who_am_i());
1368                 exit_cleanup(RERR_PROTOCOL);
1369         }
1370         sum->s2length = protocol_version < 27 ? csum_length : (int)read_int(f);
1371         if (sum->s2length < 0 || sum->s2length > MAX_DIGEST_LEN) {
1372                 rprintf(FERROR, "Invalid checksum length %d [%s]\n",
1373                         sum->s2length, who_am_i());
1374                 exit_cleanup(RERR_PROTOCOL);
1375         }
1376         sum->remainder = read_int(f);
1377         if (sum->remainder < 0 || sum->remainder > sum->blength) {
1378                 rprintf(FERROR, "Invalid remainder length %ld [%s]\n",
1379                         (long)sum->remainder, who_am_i());
1380                 exit_cleanup(RERR_PROTOCOL);
1381         }
1382 }
1383
1384 /* Send the values from a sum_struct over the socket.  Set sum to
1385  * NULL if there are no checksums to send.  This is called by both
1386  * the generator and the sender. */
1387 void write_sum_head(int f, struct sum_struct *sum)
1388 {
1389         static struct sum_struct null_sum;
1390
1391         if (sum == NULL)
1392                 sum = &null_sum;
1393
1394         write_int(f, sum->count);
1395         write_int(f, sum->blength);
1396         if (protocol_version >= 27)
1397                 write_int(f, sum->s2length);
1398         write_int(f, sum->remainder);
1399 }
1400
1401 /**
1402  * Sleep after writing to limit I/O bandwidth usage.
1403  *
1404  * @todo Rather than sleeping after each write, it might be better to
1405  * use some kind of averaging.  The current algorithm seems to always
1406  * use a bit less bandwidth than specified, because it doesn't make up
1407  * for slow periods.  But arguably this is a feature.  In addition, we
1408  * ought to take the time used to write the data into account.
1409  *
1410  * During some phases of big transfers (file FOO is uptodate) this is
1411  * called with a small bytes_written every time.  As the kernel has to
1412  * round small waits up to guarantee that we actually wait at least the
1413  * requested number of microseconds, this can become grossly inaccurate.
1414  * We therefore keep track of the bytes we've written over time and only
1415  * sleep when the accumulated delay is at least 1 tenth of a second.
1416  **/
1417 static void sleep_for_bwlimit(int bytes_written)
1418 {
1419         static struct timeval prior_tv;
1420         static long total_written = 0;
1421         struct timeval tv, start_tv;
1422         long elapsed_usec, sleep_usec;
1423
1424 #define ONE_SEC 1000000L /* # of microseconds in a second */
1425
1426         if (!bwlimit_writemax)
1427                 return;
1428
1429         total_written += bytes_written;
1430
1431         gettimeofday(&start_tv, NULL);
1432         if (prior_tv.tv_sec) {
1433                 elapsed_usec = (start_tv.tv_sec - prior_tv.tv_sec) * ONE_SEC
1434                              + (start_tv.tv_usec - prior_tv.tv_usec);
1435                 total_written -= elapsed_usec * bwlimit / (ONE_SEC/1024);
1436                 if (total_written < 0)
1437                         total_written = 0;
1438         }
1439
1440         sleep_usec = total_written * (ONE_SEC/1024) / bwlimit;
1441         if (sleep_usec < ONE_SEC / 10) {
1442                 prior_tv = start_tv;
1443                 return;
1444         }
1445
1446         tv.tv_sec  = sleep_usec / ONE_SEC;
1447         tv.tv_usec = sleep_usec % ONE_SEC;
1448         select(0, NULL, NULL, NULL, &tv);
1449
1450         gettimeofday(&prior_tv, NULL);
1451         elapsed_usec = (prior_tv.tv_sec - start_tv.tv_sec) * ONE_SEC
1452                      + (prior_tv.tv_usec - start_tv.tv_usec);
1453         total_written = (sleep_usec - elapsed_usec) * bwlimit / (ONE_SEC/1024);
1454 }
1455
1456 /* Write len bytes to the file descriptor fd, looping as necessary to get
1457  * the job done and also (in certain circumstances) reading any data on
1458  * msg_fd_in to avoid deadlock.
1459  *
1460  * This function underlies the multiplexing system.  The body of the
1461  * application never calls this function directly. */
1462 static void writefd_unbuffered(int fd, const char *buf, size_t len)
1463 {
1464         size_t n, total = 0;
1465         fd_set w_fds, r_fds, e_fds;
1466         int maxfd, count, cnt, using_r_fds;
1467         int defer_inc = 0;
1468         struct timeval tv;
1469
1470         if (no_flush++)
1471                 defer_forwarding_messages++, defer_inc++;
1472
1473         while (total < len) {
1474                 FD_ZERO(&w_fds);
1475                 FD_SET(fd, &w_fds);
1476                 FD_ZERO(&e_fds);
1477                 FD_SET(fd, &e_fds);
1478                 maxfd = fd;
1479
1480                 if (msg_fd_in >= 0) {
1481                         FD_ZERO(&r_fds);
1482                         FD_SET(msg_fd_in, &r_fds);
1483                         if (msg_fd_in > maxfd)
1484                                 maxfd = msg_fd_in;
1485                         using_r_fds = 1;
1486                 } else
1487                         using_r_fds = 0;
1488
1489                 tv.tv_sec = select_timeout;
1490                 tv.tv_usec = 0;
1491
1492                 errno = 0;
1493                 count = select(maxfd + 1, using_r_fds ? &r_fds : NULL,
1494                                &w_fds, &e_fds, &tv);
1495
1496                 if (count <= 0) {
1497                         if (count < 0 && errno == EBADF)
1498                                 exit_cleanup(RERR_SOCKETIO);
1499                         check_timeout();
1500                         continue;
1501                 }
1502
1503                 /*if (FD_ISSET(fd, &e_fds))
1504                         rprintf(FINFO, "select exception on fd %d\n", fd); */
1505
1506                 if (using_r_fds && FD_ISSET(msg_fd_in, &r_fds))
1507                         read_msg_fd();
1508
1509                 if (!FD_ISSET(fd, &w_fds))
1510                         continue;
1511
1512                 n = len - total;
1513                 if (bwlimit_writemax && n > bwlimit_writemax)
1514                         n = bwlimit_writemax;
1515                 cnt = write(fd, buf + total, n);
1516
1517                 if (cnt <= 0) {
1518                         if (cnt < 0) {
1519                                 if (errno == EINTR)
1520                                         continue;
1521                                 if (errno == EWOULDBLOCK || errno == EAGAIN) {
1522                                         msleep(1);
1523                                         continue;
1524                                 }
1525                         }
1526
1527                         /* Don't try to write errors back across the stream. */
1528                         if (fd == sock_f_out)
1529                                 io_end_multiplex_out();
1530                         /* Don't try to write errors down a failing msg pipe. */
1531                         if (am_server && fd == msg_fd_out)
1532                                 exit_cleanup(RERR_STREAMIO);
1533                         rsyserr(FERROR, errno,
1534                                 "writefd_unbuffered failed to write %ld bytes [%s]",
1535                                 (long)len, who_am_i());
1536                         /* If the other side is sending us error messages, try
1537                          * to grab any messages they sent before they died. */
1538                         while (!am_server && fd == sock_f_out && io_multiplexing_in) {
1539                                 char buf[1024];
1540                                 set_io_timeout(30);
1541                                 ignore_timeout = 0;
1542                                 readfd_unbuffered(sock_f_in, buf, sizeof buf);
1543                         }
1544                         exit_cleanup(RERR_STREAMIO);
1545                 }
1546
1547                 total += cnt;
1548                 defer_forwarding_messages++, defer_inc++;
1549
1550                 if (fd == sock_f_out) {
1551                         if (io_timeout || am_generator)
1552                                 last_io_out = time(NULL);
1553                         sleep_for_bwlimit(cnt);
1554                 }
1555         }
1556
1557         no_flush--;
1558         defer_inc -= defer_forwarding_keep;
1559         if (!(defer_forwarding_messages -= defer_inc) && !no_flush)
1560                 msg_flush();
1561 }
1562
1563 void io_flush(int flush_it_all)
1564 {
1565         if (!iobuf_out_cnt || no_flush)
1566                 return;
1567
1568         if (io_multiplexing_out)
1569                 mplex_write(sock_f_out, MSG_DATA, iobuf_out, iobuf_out_cnt, 0);
1570         else
1571                 writefd_unbuffered(iobuf_f_out, iobuf_out, iobuf_out_cnt);
1572         iobuf_out_cnt = 0;
1573
1574         if (flush_it_all && !defer_forwarding_messages)
1575                 msg_flush();
1576 }
1577
1578 static void writefd(int fd, const char *buf, size_t len)
1579 {
1580         if (fd == sock_f_out)
1581                 stats.total_written += len;
1582
1583         if (fd == write_batch_monitor_out) {
1584                 if ((size_t)write(batch_fd, buf, len) != len)
1585                         exit_cleanup(RERR_FILEIO);
1586         }
1587
1588         if (!iobuf_out || fd != iobuf_f_out) {
1589                 writefd_unbuffered(fd, buf, len);
1590                 return;
1591         }
1592
1593         while (len) {
1594                 int n = MIN((int)len, IO_BUFFER_SIZE - iobuf_out_cnt);
1595                 if (n > 0) {
1596                         memcpy(iobuf_out+iobuf_out_cnt, buf, n);
1597                         buf += n;
1598                         len -= n;
1599                         iobuf_out_cnt += n;
1600                 }
1601
1602                 if (iobuf_out_cnt == IO_BUFFER_SIZE)
1603                         io_flush(NORMAL_FLUSH);
1604         }
1605 }
1606
1607 void write_shortint(int f, unsigned short x)
1608 {
1609         char b[2];
1610         b[0] = (char)x;
1611         b[1] = (char)(x >> 8);
1612         writefd(f, b, 2);
1613 }
1614
1615 void write_int(int f, int32 x)
1616 {
1617         char b[4];
1618         SIVAL(b, 0, x);
1619         writefd(f, b, 4);
1620 }
1621
1622 void write_varint(int f, int32 x)
1623 {
1624         char b[5];
1625         uchar bit;
1626         int cnt = 4;
1627
1628         SIVAL(b, 1, x);
1629
1630         while (cnt > 1 && b[cnt] == 0)
1631                 cnt--;
1632         bit = ((uchar)1<<(7-cnt+1));
1633         if (CVAL(b, cnt) >= bit) {
1634                 cnt++;
1635                 *b = ~(bit-1);
1636         } else if (cnt > 1)
1637                 *b = b[cnt] | ~(bit*2-1);
1638         else
1639                 *b = b[cnt];
1640
1641         writefd(f, b, cnt);
1642 }
1643
1644 void write_varlong(int f, int64 x, uchar min_bytes)
1645 {
1646         char b[9];
1647         uchar bit;
1648         int cnt = 8;
1649
1650         SIVAL(b, 1, x);
1651 #if SIZEOF_INT64 >= 8
1652         SIVAL(b, 5, x >> 32);
1653 #else
1654         if (x <= 0x7FFFFFFF && x >= 0)
1655                 memset(b + 5, 0, 4);
1656         else {
1657                 rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
1658                 exit_cleanup(RERR_UNSUPPORTED);
1659         }
1660 #endif
1661
1662         while (cnt > min_bytes && b[cnt] == 0)
1663                 cnt--;
1664         bit = ((uchar)1<<(7-cnt+min_bytes));
1665         if (CVAL(b, cnt) >= bit) {
1666                 cnt++;
1667                 *b = ~(bit-1);
1668         } else if (cnt > min_bytes)
1669                 *b = b[cnt] | ~(bit*2-1);
1670         else
1671                 *b = b[cnt];
1672
1673         writefd(f, b, cnt);
1674 }
1675
1676 /*
1677  * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
1678  * 64-bit types on this platform.
1679  */
1680 void write_longint(int f, int64 x)
1681 {
1682         char b[12], * const s = b+4;
1683
1684         SIVAL(s, 0, x);
1685         if (x <= 0x7FFFFFFF && x >= 0) {
1686                 writefd(f, s, 4);
1687                 return;
1688         }
1689
1690 #if SIZEOF_INT64 < 8
1691         rprintf(FERROR, "Integer overflow: attempted 64-bit offset\n");
1692         exit_cleanup(RERR_UNSUPPORTED);
1693 #else
1694         memset(b, 0xFF, 4);
1695         SIVAL(s, 4, x >> 32);
1696         writefd(f, b, 12);
1697 #endif
1698 }
1699
1700 void write_buf(int f, const char *buf, size_t len)
1701 {
1702         writefd(f,buf,len);
1703 }
1704
1705 /** Write a string to the connection */
1706 void write_sbuf(int f, const char *buf)
1707 {
1708         writefd(f, buf, strlen(buf));
1709 }
1710
1711 void write_byte(int f, uchar c)
1712 {
1713         writefd(f, (char *)&c, 1);
1714 }
1715
1716 void write_vstring(int f, const char *str, int len)
1717 {
1718         uchar lenbuf[3], *lb = lenbuf;
1719
1720         if (len > 0x7F) {
1721                 if (len > 0x7FFF) {
1722                         rprintf(FERROR,
1723                                 "attempting to send over-long vstring (%d > %d)\n",
1724                                 len, 0x7FFF);
1725                         exit_cleanup(RERR_PROTOCOL);
1726                 }
1727                 *lb++ = len / 0x100 + 0x80;
1728         }
1729         *lb = len;
1730
1731         writefd(f, (char*)lenbuf, lb - lenbuf + 1);
1732         if (len)
1733                 writefd(f, str, len);
1734 }
1735
1736 /* Send a file-list index using a byte-reduction method. */
1737 void write_ndx(int f, int32 ndx)
1738 {
1739         static int32 prev_positive = -1, prev_negative = 1;
1740         int32 diff, cnt = 0;
1741         char b[6];
1742
1743         if (protocol_version < 30 || read_batch) {
1744                 write_int(f, ndx);
1745                 return;
1746         }
1747
1748         /* Send NDX_DONE as a single-byte 0 with no side effects.  Send
1749          * negative nums as a positive after sending a leading 0xFF. */
1750         if (ndx >= 0) {
1751                 diff = ndx - prev_positive;
1752                 prev_positive = ndx;
1753         } else if (ndx == NDX_DONE) {
1754                 *b = 0;
1755                 writefd(f, b, 1);
1756                 return;
1757         } else {
1758                 b[cnt++] = (char)0xFF;
1759                 ndx = -ndx;
1760                 diff = ndx - prev_negative;
1761                 prev_negative = ndx;
1762         }
1763
1764         /* A diff of 1 - 253 is sent as a one-byte diff; a diff of 254 - 32767
1765          * or 0 is sent as a 0xFE + a two-byte diff; otherwise we send 0xFE
1766          * & all 4 bytes of the (non-negative) num with the high-bit set. */
1767         if (diff < 0xFE && diff > 0)
1768                 b[cnt++] = (char)diff;
1769         else if (diff < 0 || diff > 0x7FFF) {
1770                 b[cnt++] = (char)0xFE;
1771                 b[cnt++] = (char)((ndx >> 24) | 0x80);
1772                 b[cnt++] = (char)ndx;
1773                 b[cnt++] = (char)(ndx >> 8);
1774                 b[cnt++] = (char)(ndx >> 16);
1775         } else {
1776                 b[cnt++] = (char)0xFE;
1777                 b[cnt++] = (char)(diff >> 8);
1778                 b[cnt++] = (char)diff;
1779         }
1780         writefd(f, b, cnt);
1781 }
1782
1783 /* Receive a file-list index using a byte-reduction method. */
1784 int32 read_ndx(int f)
1785 {
1786         static int32 prev_positive = -1, prev_negative = 1;
1787         int32 *prev_ptr, num;
1788         char b[4];
1789
1790         if (protocol_version < 30)
1791                 return read_int(f);
1792
1793         readfd(f, b, 1);
1794         if (CVAL(b, 0) == 0xFF) {
1795                 readfd(f, b, 1);
1796                 prev_ptr = &prev_negative;
1797         } else if (CVAL(b, 0) == 0)
1798                 return NDX_DONE;
1799         else
1800                 prev_ptr = &prev_positive;
1801         if (CVAL(b, 0) == 0xFE) {
1802                 readfd(f, b, 2);
1803                 if (CVAL(b, 0) & 0x80) {
1804                         b[3] = CVAL(b, 0) & ~0x80;
1805                         b[0] = b[1];
1806                         readfd(f, b+1, 2);
1807                         num = IVAL(b, 0);
1808                 } else
1809                         num = (UVAL(b,0)<<8) + UVAL(b,1) + *prev_ptr;
1810         } else
1811                 num = UVAL(b, 0) + *prev_ptr;
1812         *prev_ptr = num;
1813         if (prev_ptr == &prev_negative)
1814                 num = -num;
1815         return num;
1816 }
1817
1818 /* Read a line of up to bufsiz-1 characters into buf.  Strips
1819  * the (required) trailing newline and all carriage returns.
1820  * Returns 1 for success; 0 for I/O error or truncation. */
1821 int read_line_old(int f, char *buf, size_t bufsiz)
1822 {
1823         bufsiz--; /* leave room for the null */
1824         while (bufsiz > 0) {
1825                 buf[0] = 0;
1826                 read_buf(f, buf, 1);
1827                 if (buf[0] == 0)
1828                         return 0;
1829                 if (buf[0] == '\n')
1830                         break;
1831                 if (buf[0] != '\r') {
1832                         buf++;
1833                         bufsiz--;
1834                 }
1835         }
1836         *buf = '\0';
1837         return bufsiz > 0;
1838 }
1839
1840 void io_printf(int fd, const char *format, ...)
1841 {
1842         va_list ap;
1843         char buf[BIGPATHBUFLEN];
1844         int len;
1845
1846         va_start(ap, format);
1847         len = vsnprintf(buf, sizeof buf, format, ap);
1848         va_end(ap);
1849
1850         if (len < 0)
1851                 exit_cleanup(RERR_STREAMIO);
1852
1853         if (len > (int)sizeof buf) {
1854                 rprintf(FERROR, "io_printf() was too long for the buffer.\n");
1855                 exit_cleanup(RERR_STREAMIO);
1856         }
1857
1858         write_sbuf(fd, buf);
1859 }
1860
1861 /** Setup for multiplexing a MSG_* stream with the data stream. */
1862 void io_start_multiplex_out(void)
1863 {
1864         io_flush(NORMAL_FLUSH);
1865         io_start_buffering_out(sock_f_out);
1866         io_multiplexing_out = 1;
1867 }
1868
1869 /** Setup for multiplexing a MSG_* stream with the data stream. */
1870 void io_start_multiplex_in(void)
1871 {
1872         io_flush(NORMAL_FLUSH);
1873         io_start_buffering_in(sock_f_in);
1874         io_multiplexing_in = 1;
1875 }
1876
1877 /** Write an message to the multiplexed data stream. */
1878 int io_multiplex_write(enum msgcode code, const char *buf, size_t len, int convert)
1879 {
1880         if (!io_multiplexing_out)
1881                 return 0;
1882         io_flush(NORMAL_FLUSH);
1883         stats.total_written += (len+4);
1884         mplex_write(sock_f_out, code, buf, len, convert);
1885         return 1;
1886 }
1887
1888 void io_end_multiplex_in(void)
1889 {
1890         io_multiplexing_in = 0;
1891         io_end_buffering_in();
1892 }
1893
1894 /** Stop output multiplexing. */
1895 void io_end_multiplex_out(void)
1896 {
1897         io_multiplexing_out = 0;
1898         io_end_buffering_out();
1899 }
1900
1901 void start_write_batch(int fd)
1902 {
1903         /* Some communication has already taken place, but we don't
1904          * enable batch writing until here so that we can write a
1905          * canonical record of the communication even though the
1906          * actual communication so far depends on whether a daemon
1907          * is involved. */
1908         write_int(batch_fd, protocol_version);
1909         if (protocol_version >= 30)
1910                 write_byte(batch_fd, inc_recurse);
1911         write_int(batch_fd, checksum_seed);
1912
1913         if (am_sender)
1914                 write_batch_monitor_out = fd;
1915         else
1916                 write_batch_monitor_in = fd;
1917 }
1918
1919 void stop_write_batch(void)
1920 {
1921         write_batch_monitor_out = -1;
1922         write_batch_monitor_in = -1;
1923 }