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