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