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