io.c

   1 /* -*- c-file-style: "linux" -*-
   2  *
   3  * Copyright (C) 1996-2001 by Andrew Tridgell
   4  * Copyright (C) Paul Mackerras 1996
   5  * Copyright (C) 2001, 2002 by Martin Pool <mbp@samba.org>
   6  *
   7  * This program is free software; you can redistribute it and/or modify
   8  * it under the terms of the GNU General Public License as published by
   9  * the Free Software Foundation; either version 2 of the License, or
  10  * (at your option) any later version.
  11  *
  12  * This program is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License
  18  * along with this program; if not, write to the Free Software
  19  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22 /**
  23  * @file io.c
  24  *
  25  * Socket and pipe IO utilities used in rsync.
  26  *
  27  * rsync provides its own multiplexing system, which is used to send
  28  * stderr and stdout over a single socket.  We need this because
  29  * stdout normally carries the binary data stream, and stderr all our
  30  * error messages.
  31  *
  32  * For historical reasons this is off during the start of the
  33  * connection, but it's switched on quite early using
  34  * io_start_multiplex_out() and io_start_multiplex_in().
  35  **/
  36
  37 #include "rsync.h"
  38
  39 /** If no timeout is specified then use a 60 second select timeout */
  40 #define SELECT_TIMEOUT 60
  41
  42 static int io_multiplexing_out;
  43 static int io_multiplexing_in;
  44 static int multiplex_in_fd;
  45 static int multiplex_out_fd;
  46 static time_t last_io;
  47 static int no_flush;
  48
  49 extern int bwlimit;
  50 extern int verbose;
  51 extern int io_timeout;
  52 extern struct stats stats;
  53
  54
  55 const char * const phase_unknown = "unknown";
  56
  57 /**
  58  * The connection might be dropped at some point; perhaps because the
  59  * remote instance crashed.  Just giving the offset on the stream is
  60  * not very helpful.  So instead we try to make io_phase_name point to
  61  * something useful.
  62  *
  63  * For buffered/multiplexed IO these names will be somewhat
  64  * approximate; perhaps for ease of support we would rather make the
  65  * buffer always flush when a single application-level IO finishes.
  66  *
  67  * @todo Perhaps we want some simple stack functionality, but there's
  68  * no need to overdo it.
  69  **/
  70 const char *io_write_phase = phase_unknown;
  71 const char *io_read_phase = phase_unknown;
  72
  73 /** Ignore EOF errors while reading a module listing if the remote
  74     version is 24 or less. */
  75 int kludge_around_eof = False;
  76
  77
  78 static int io_error_fd = -1;
  79
  80 static void read_loop(int fd, char *buf, size_t len);
  81
  82 static void check_timeout(void)
  83 {
  84         extern int am_server, am_daemon;
  85         time_t t;
  86
  87         err_list_push();
  88
  89         if (!io_timeout) return;
  90
  91         if (!last_io) {
  92                 last_io = time(NULL);
  93                 return;
  94         }
  95
  96         t = time(NULL);
  97
  98         if (last_io && io_timeout && (t-last_io) >= io_timeout) {
  99                 if (!am_server && !am_daemon) {
 100                         rprintf(FERROR,"io timeout after %d seconds - exiting\n",
 101                                 (int)(t-last_io));
 102                 }
 103                 exit_cleanup(RERR_TIMEOUT);
 104         }
 105 }
 106
 107 /** Setup the fd used to propogate errors */
 108 void io_set_error_fd(int fd)
 109 {
 110         io_error_fd = fd;
 111 }
 112
 113 /** Read some data from the error fd and write it to the write log code */
 114 static void read_error_fd(void)
 115 {
 116         char buf[200];
 117         size_t n;
 118         int fd = io_error_fd;
 119         int tag, len;
 120
 121         /* io_error_fd is temporarily disabled -- is this meant to
 122          * prevent indefinite recursion? */
 123         io_error_fd = -1;
 124
 125         read_loop(fd, buf, 4);
 126         tag = IVAL(buf, 0);
 127
 128         len = tag & 0xFFFFFF;
 129         tag = tag >> 24;
 130         tag -= MPLEX_BASE;
 131
 132         while (len) {
 133                 n = len;
 134                 if (n > (sizeof(buf)-1))
 135                         n = sizeof(buf)-1;
 136                 read_loop(fd, buf, n);
 137                 rwrite((enum logcode)tag, buf, n);
 138                 len -= n;
 139         }
 140
 141         io_error_fd = fd;
 142 }
 143
 144
 145 /**
 146  * It's almost always an error to get an EOF when we're trying to read
 147  * from the network, because the protocol is self-terminating.
 148  *
 149  * However, there is one unfortunate cases where it is not, which is
 150  * rsync <2.4.6 sending a list of modules on a server, since the list
 151  * is terminated by closing the socket. So, for the section of the
 152  * program where that is a problem (start_socket_client),
 153  * kludge_around_eof is True and we just exit.
 154  */
 155 static void whine_about_eof (void)
 156 {
 157         if (kludge_around_eof)
 158                 exit_cleanup (0);
 159         else {
 160                 rprintf (FERROR,
 161                          "%s: connection unexpectedly closed "
 162                          "(%.0f bytes read so far)\n",
 163                          RSYNC_NAME, (double)stats.total_read);
 164
 165                 exit_cleanup (RERR_STREAMIO);
 166         }
 167 }
 168
 169
 170 static void die_from_readerr (int err)
 171 {
 172         /* this prevents us trying to write errors on a dead socket */
 173         io_multiplexing_close();
 174
 175         rprintf(FERROR, "%s: read error: %s\n",
 176                 RSYNC_NAME, strerror (err));
 177         exit_cleanup(RERR_STREAMIO);
 178 }
 179
 180
 181 /**
 182  * Read from a socket with IO timeout. return the number of bytes
 183  * read. If no bytes can be read then exit, never return a number <= 0.
 184  *
 185  * TODO: If the remote shell connection fails, then current versions
 186  * actually report an "unexpected EOF" error here.  Since it's a
 187  * fairly common mistake to try to use rsh when ssh is required, we
 188  * should trap that: if we fail to read any data at all, we should
 189  * give a better explanation.  We can tell whether the connection has
 190  * started by looking e.g. at whether the remote version is known yet.
 191  */
 192 static int read_timeout (int fd, char *buf, size_t len)
 193 {
 194         int n, ret=0;
 195
 196         io_flush();
 197
 198         while (ret == 0) {
 199                 /* until we manage to read *something* */
 200                 fd_set fds;
 201                 struct timeval tv;
 202                 int fd_count = fd+1;
 203                 int count;
 204
 205                 FD_ZERO(&fds);
 206                 FD_SET(fd, &fds);
 207                 if (io_error_fd != -1) {
 208                         FD_SET(io_error_fd, &fds);
 209                         if (io_error_fd > fd) fd_count = io_error_fd+1;
 210                 }
 211
 212                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 213                 tv.tv_usec = 0;
 214
 215                 errno = 0;
 216
 217                 count = select(fd_count, &fds, NULL, NULL, &tv);
 218
 219                 if (count == 0) {
 220                         check_timeout();
 221                 }
 222
 223                 if (count <= 0) {
 224                         if (errno == EBADF) {
 225                                 exit_cleanup(RERR_SOCKETIO);
 226                         }
 227                         continue;
 228                 }
 229
 230                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &fds)) {
 231                         read_error_fd();
 232                 }
 233
 234                 if (!FD_ISSET(fd, &fds)) continue;
 235
 236                 n = read(fd, buf, len);
 237
 238                 if (n > 0) {
 239                         buf += n;
 240                         len -= n;
 241                         ret += n;
 242                         if (io_timeout)
 243                                 last_io = time(NULL);
 244                         continue;
 245                 } else if (n == 0) {
 246                         whine_about_eof ();
 247                         return -1; /* doesn't return */
 248                 } else if (n == -1) {
 249                         if (errno == EINTR || errno == EWOULDBLOCK ||
 250                             errno == EAGAIN)
 251                                 continue;
 252                         else
 253                                 die_from_readerr (errno);
 254                 }
 255         }
 256
 257         return ret;
 258 }
 259
 260
 261
 262
 263 /**
 264  * Continue trying to read len bytes - don't return until len has been
 265  * read.
 266  **/
 267 static void read_loop (int fd, char *buf, size_t len)
 268 {
 269         while (len) {
 270                 int n = read_timeout(fd, buf, len);
 271
 272                 buf += n;
 273                 len -= n;
 274         }
 275 }
 276
 277
 278 /**
 279  * Read from the file descriptor handling multiplexing - return number
 280  * of bytes read.
 281  *
 282  * Never returns <= 0.
 283  */
 284 static int read_unbuffered(int fd, char *buf, size_t len)
 285 {
 286         static size_t remaining;
 287         int tag, ret = 0;
 288         char line[1024];
 289
 290         if (!io_multiplexing_in || fd != multiplex_in_fd)
 291                 return read_timeout(fd, buf, len);
 292
 293         while (ret == 0) {
 294                 if (remaining) {
 295                         len = MIN(len, remaining);
 296                         read_loop(fd, buf, len);
 297                         remaining -= len;
 298                         ret = len;
 299                         continue;
 300                 }
 301
 302                 read_loop(fd, line, 4);
 303                 tag = IVAL(line, 0);
 304
 305                 remaining = tag & 0xFFFFFF;
 306                 tag = tag >> 24;
 307
 308                 if (tag == MPLEX_BASE)
 309                         continue;
 310
 311                 tag -= MPLEX_BASE;
 312
 313                 if (tag != FERROR && tag != FINFO) {
 314                         rprintf(FERROR, "unexpected tag %d\n", tag);
 315                         exit_cleanup(RERR_STREAMIO);
 316                 }
 317
 318                 if (remaining > sizeof(line) - 1) {
 319                         rprintf(FERROR, "multiplexing overflow %d\n\n",
 320                                 remaining);
 321                         exit_cleanup(RERR_STREAMIO);
 322                 }
 323
 324                 read_loop(fd, line, remaining);
 325                 line[remaining] = 0;
 326
 327                 rprintf((enum logcode) tag, "%s", line);
 328                 remaining = 0;
 329         }
 330
 331         return ret;
 332 }
 333
 334
 335
 336 /**
 337  * Do a buffered read from @p fd.  Don't return until all @p n bytes
 338  * have been read.  If all @p n can't be read then exit with an
 339  * error.
 340  **/
 341 static void readfd (int fd, char *buffer, size_t N)
 342 {
 343         int  ret;
 344         size_t total=0;
 345
 346         while (total < N) {
 347                 io_flush();
 348
 349                 ret = read_unbuffered (fd, buffer + total, N-total);
 350                 total += ret;
 351         }
 352
 353         stats.total_read += total;
 354 }
 355
 356
 357 int32 read_int(int f)
 358 {
 359         char b[4];
 360         int32 ret;
 361
 362         readfd(f,b,4);
 363         ret = IVAL(b,0);
 364         if (ret == (int32)0xffffffff) return -1;
 365         return ret;
 366 }
 367
 368 int64 read_longint(int f)
 369 {
 370         extern int remote_version;
 371         int64 ret;
 372         char b[8];
 373         ret = read_int(f);
 374
 375         if ((int32)ret != (int32)0xffffffff) {
 376                 return ret;
 377         }
 378
 379 #ifdef NO_INT64
 380         rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
 381         exit_cleanup(RERR_UNSUPPORTED);
 382 #else
 383         if (remote_version >= 16) {
 384                 readfd(f,b,8);
 385                 ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
 386         }
 387 #endif
 388
 389         return ret;
 390 }
 391
 392 void read_buf(int f,char *buf,size_t len)
 393 {
 394         readfd(f,buf,len);
 395 }
 396
 397 void read_sbuf(int f,char *buf,size_t len)
 398 {
 399         read_buf (f,buf,len);
 400         buf[len] = 0;
 401 }
 402
 403 unsigned char read_byte(int f)
 404 {
 405         unsigned char c;
 406         read_buf (f, (char *)&c, 1);
 407         return c;
 408 }
 409
 410
 411 /**
 412  * Sleep after writing to limit I/O bandwidth usage.
 413  *
 414  * @todo Rather than sleeping after each write, it might be better to
 415  * use some kind of averaging.  The current algorithm seems to always
 416  * use a bit less bandwidth than specified, because it doesn't make up
 417  * for slow periods.  But arguably this is a feature.  In addition, we
 418  * ought to take the time used to write the data into account.
 419  **/
 420 static void sleep_for_bwlimit(int bytes_written)
 421 {
 422         struct timeval tv;
 423
 424         if (!bwlimit)
 425                 return;
 426
 427         assert(bytes_written > 0);
 428         assert(bwlimit > 0);
 429
 430         tv.tv_usec = bytes_written * 1000 / bwlimit;
 431         tv.tv_sec  = tv.tv_usec / 1000000;
 432         tv.tv_usec = tv.tv_usec % 1000000;
 433
 434         select(0, NULL, NULL, NULL, &tv);
 435 }
 436
 437
 438 /**
 439  * Write len bytes to the file descriptor @p fd.
 440  *
 441  * This function underlies the multiplexing system.  The body of the
 442  * application never calls this function directly.
 443  **/
 444 static void writefd_unbuffered(int fd,char *buf,size_t len)
 445 {
 446         size_t total = 0;
 447         fd_set w_fds, r_fds;
 448         int fd_count, count;
 449         struct timeval tv;
 450
 451         err_list_push();
 452
 453         no_flush++;
 454
 455         while (total < len) {
 456                 FD_ZERO(&w_fds);
 457                 FD_ZERO(&r_fds);
 458                 FD_SET(fd,&w_fds);
 459                 fd_count = fd;
 460
 461                 if (io_error_fd != -1) {
 462                         FD_SET(io_error_fd,&r_fds);
 463                         if (io_error_fd > fd_count)
 464                                 fd_count = io_error_fd;
 465                 }
 466
 467                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 468                 tv.tv_usec = 0;
 469
 470                 errno = 0;
 471
 472                 count = select(fd_count+1,
 473                                io_error_fd != -1?&r_fds:NULL,
 474                                &w_fds,NULL,
 475                                &tv);
 476
 477                 if (count == 0) {
 478                         check_timeout();
 479                 }
 480
 481                 if (count <= 0) {
 482                         if (errno == EBADF) {
 483                                 exit_cleanup(RERR_SOCKETIO);
 484                         }
 485                         continue;
 486                 }
 487
 488                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &r_fds)) {
 489                         read_error_fd();
 490                 }
 491
 492                 if (FD_ISSET(fd, &w_fds)) {
 493                         int ret;
 494                         size_t n = len-total;
 495                         ret = write(fd,buf+total,n);
 496
 497                         if (ret == -1 && errno == EINTR) {
 498                                 continue;
 499                         }
 500
 501                         if (ret == -1 &&
 502                             (errno == EWOULDBLOCK || errno == EAGAIN)) {
 503                                 msleep(1);
 504                                 continue;
 505                         }
 506
 507                         if (ret <= 0) {
 508                                 /* Don't try to write errors back
 509                                  * across the stream */
 510                                 io_multiplexing_close();
 511                                 rprintf(FERROR, RSYNC_NAME
 512                                         ": writefd_unbuffered failed to write %ld bytes: phase \"%s\": %s\n",
 513                                         (long) len, io_write_phase,
 514                                         strerror(errno));
 515                                 exit_cleanup(RERR_STREAMIO);
 516                         }
 517
 518                         sleep_for_bwlimit(ret);
 519
 520                         total += ret;
 521
 522                         if (io_timeout)
 523                                 last_io = time(NULL);
 524                 }
 525         }
 526
 527         no_flush--;
 528 }
 529
 530
 531 static char *io_buffer;
 532 static int io_buffer_count;
 533
 534 void io_start_buffering(int fd)
 535 {
 536         if (io_buffer) return;
 537         multiplex_out_fd = fd;
 538         io_buffer = (char *)malloc(IO_BUFFER_SIZE);
 539         if (!io_buffer) out_of_memory("writefd");
 540         io_buffer_count = 0;
 541 }
 542
 543 /**
 544  * Write an message to a multiplexed stream. If this fails then rsync
 545  * exits.
 546  **/
 547 static void mplex_write(int fd, enum logcode code, char *buf, size_t len)
 548 {
 549         char buffer[4096];
 550         size_t n = len;
 551
 552         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
 553
 554         if (n > (sizeof(buffer)-4)) {
 555                 n = sizeof(buffer)-4;
 556         }
 557
 558         memcpy(&buffer[4], buf, n);
 559         writefd_unbuffered(fd, buffer, n+4);
 560
 561         len -= n;
 562         buf += n;
 563
 564         if (len) {
 565                 writefd_unbuffered(fd, buf, len);
 566         }
 567 }
 568
 569
 570 void io_flush(void)
 571 {
 572         int fd = multiplex_out_fd;
 573
 574         err_list_push();
 575
 576         if (!io_buffer_count || no_flush) return;
 577
 578         if (io_multiplexing_out) {
 579                 mplex_write(fd, FNONE, io_buffer, io_buffer_count);
 580         } else {
 581                 writefd_unbuffered(fd, io_buffer, io_buffer_count);
 582         }
 583         io_buffer_count = 0;
 584 }
 585
 586
 587 void io_end_buffering(void)
 588 {
 589         io_flush();
 590         if (!io_multiplexing_out) {
 591                 free(io_buffer);
 592                 io_buffer = NULL;
 593         }
 594 }
 595
 596 static void writefd(int fd,char *buf,size_t len)
 597 {
 598         stats.total_written += len;
 599
 600         err_list_push();
 601
 602         if (!io_buffer || fd != multiplex_out_fd) {
 603                 writefd_unbuffered(fd, buf, len);
 604                 return;
 605         }
 606
 607         while (len) {
 608                 int n = MIN((int) len, IO_BUFFER_SIZE-io_buffer_count);
 609                 if (n > 0) {
 610                         memcpy(io_buffer+io_buffer_count, buf, n);
 611                         buf += n;
 612                         len -= n;
 613                         io_buffer_count += n;
 614                 }
 615
 616                 if (io_buffer_count == IO_BUFFER_SIZE) io_flush();
 617         }
 618 }
 619
 620
 621 void write_int(int f,int32 x)
 622 {
 623         char b[4];
 624         SIVAL(b,0,x);
 625         writefd(f,b,4);
 626 }
 627
 628
 629 void write_int_named(int f, int32 x, const char *phase)
 630 {
 631         io_write_phase = phase;
 632         write_int(f, x);
 633         io_write_phase = phase_unknown;
 634 }
 635
 636
 637 /*
 638  * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
 639  * 64-bit types on this platform.
 640  */
 641 void write_longint(int f, int64 x)
 642 {
 643         extern int remote_version;
 644         char b[8];
 645
 646         if (remote_version < 16 || x <= 0x7FFFFFFF) {
 647                 write_int(f, (int)x);
 648                 return;
 649         }
 650
 651         write_int(f, (int32)0xFFFFFFFF);
 652         SIVAL(b,0,(x&0xFFFFFFFF));
 653         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
 654
 655         writefd(f,b,8);
 656 }
 657
 658 void write_buf(int f,char *buf,size_t len)
 659 {
 660         writefd(f,buf,len);
 661 }
 662
 663 /** Write a string to the connection */
 664 static void write_sbuf(int f,char *buf)
 665 {
 666         write_buf(f, buf, strlen(buf));
 667 }
 668
 669
 670 void write_byte(int f,unsigned char c)
 671 {
 672         write_buf(f,(char *)&c,1);
 673 }
 674
 675
 676
 677 /**
 678  * Read a line of up to @p maxlen characters into @p buf.  Does not
 679  * contain a trailing newline or carriage return.
 680  *
 681  * @return 1 for success; 0 for io error or truncation.
 682  **/
 683 int read_line(int f, char *buf, size_t maxlen)
 684 {
 685         while (maxlen) {
 686                 buf[0] = 0;
 687                 read_buf(f, buf, 1);
 688                 if (buf[0] == 0)
 689                         return 0;
 690                 if (buf[0] == '\n') {
 691                         buf[0] = 0;
 692                         break;
 693                 }
 694                 if (buf[0] != '\r') {
 695                         buf++;
 696                         maxlen--;
 697                 }
 698         }
 699         if (maxlen == 0) {
 700                 *buf = 0;
 701                 return 0;
 702         }
 703
 704         return 1;
 705 }
 706
 707
 708 void io_printf(int fd, const char *format, ...)
 709 {
 710         va_list ap;
 711         char buf[1024];
 712         int len;
 713
 714         va_start(ap, format);
 715         len = vsnprintf(buf, sizeof(buf), format, ap);
 716         va_end(ap);
 717
 718         if (len < 0) exit_cleanup(RERR_STREAMIO);
 719
 720         write_sbuf(fd, buf);
 721 }
 722
 723
 724 /** Setup for multiplexing an error stream with the data stream */
 725 void io_start_multiplex_out(int fd)
 726 {
 727         multiplex_out_fd = fd;
 728         io_flush();
 729         io_start_buffering(fd);
 730         io_multiplexing_out = 1;
 731 }
 732
 733 /** Setup for multiplexing an error stream with the data stream */
 734 void io_start_multiplex_in(int fd)
 735 {
 736         multiplex_in_fd = fd;
 737         io_flush();
 738         io_multiplexing_in = 1;
 739 }
 740
 741 /** Write an message to the multiplexed error stream */
 742 int io_multiplex_write(enum logcode code, char *buf, size_t len)
 743 {
 744         if (!io_multiplexing_out) return 0;
 745
 746         io_flush();
 747         stats.total_written += (len+4);
 748         mplex_write(multiplex_out_fd, code, buf, len);
 749         return 1;
 750 }
 751
 752 /** Stop output multiplexing */
 753 void io_multiplexing_close(void)
 754 {
 755         io_multiplexing_out = 0;
 756 }
 757