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