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, size_t 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, size_t 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                 int count;
 178
 179                 FD_ZERO(&fds);
 180                 FD_SET(fd, &fds);
 181                 if (io_error_fd != -1) {
 182                         FD_SET(io_error_fd, &fds);
 183                         if (io_error_fd > fd) fd_count = io_error_fd+1;
 184                 }
 185
 186                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 187                 tv.tv_usec = 0;
 188
 189                 errno = 0;
 190
 191                 count = select(fd_count, &fds, NULL, NULL, &tv);
 192
 193                 if (count == 0) {
 194                         check_timeout();
 195                 }
 196
 197                 if (count <= 0) {
 198                         if (errno == EBADF) {
 199                                 exit_cleanup(RERR_SOCKETIO);
 200                         }
 201                         continue;
 202                 }
 203
 204                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &fds)) {
 205                         read_error_fd();
 206                 }
 207
 208                 if (!FD_ISSET(fd, &fds)) continue;
 209
 210                 n = read(fd, buf, len);
 211
 212                 if (n > 0) {
 213                         buf += n;
 214                         len -= n;
 215                         ret += n;
 216                         if (io_timeout)
 217                                 last_io = time(NULL);
 218                         continue;
 219                 } else if (n == 0) {
 220                         whine_about_eof ();
 221                         return -1; /* doesn't return */
 222                 } else if (n == -1) {
 223                         if (errno == EINTR || errno == EWOULDBLOCK ||
 224                             errno == EAGAIN)
 225                                 continue;
 226                         else
 227                                 die_from_readerr (errno);
 228                 }
 229         }
 230
 231         return ret;
 232 }
 233
 234
 235
 236
 237 /*! Continue trying to read len bytes - don't return until len has
 238   been read.   */
 239 static void read_loop (int fd, char *buf, size_t len)
 240 {
 241         while (len) {
 242                 int n = read_timeout(fd, buf, len);
 243
 244                 buf += n;
 245                 len -= n;
 246         }
 247 }
 248
 249
 250 /**
 251  * Read from the file descriptor handling multiplexing - return number
 252  * of bytes read.
 253  *
 254  * Never returns <= 0.
 255  */
 256 static int read_unbuffered(int fd, char *buf, size_t len)
 257 {
 258         static int remaining;
 259         int tag, ret = 0;
 260         char line[1024];
 261
 262         if (!io_multiplexing_in || fd != multiplex_in_fd)
 263                 return read_timeout(fd, buf, len);
 264
 265         while (ret == 0) {
 266                 if (remaining) {
 267                         len = MIN(len, remaining);
 268                         read_loop(fd, buf, len);
 269                         remaining -= len;
 270                         ret = len;
 271                         continue;
 272                 }
 273
 274                 read_loop(fd, line, 4);
 275                 tag = IVAL(line, 0);
 276
 277                 remaining = tag & 0xFFFFFF;
 278                 tag = tag >> 24;
 279
 280                 if (tag == MPLEX_BASE)
 281                         continue;
 282
 283                 tag -= MPLEX_BASE;
 284
 285                 if (tag != FERROR && tag != FINFO) {
 286                         rprintf(FERROR, "unexpected tag %d\n", tag);
 287                         exit_cleanup(RERR_STREAMIO);
 288                 }
 289
 290                 if (remaining > sizeof(line) - 1) {
 291                         rprintf(FERROR, "multiplexing overflow %d\n\n",
 292                                 remaining);
 293                         exit_cleanup(RERR_STREAMIO);
 294                 }
 295
 296                 read_loop(fd, line, remaining);
 297                 line[remaining] = 0;
 298
 299                 rprintf((enum logcode) tag, "%s", line);
 300                 remaining = 0;
 301         }
 302
 303         return ret;
 304 }
 305
 306
 307
 308 /* do a buffered read from fd. don't return until all N bytes
 309    have been read. If all N can't be read then exit with an error */
 310 static void readfd (int fd, char *buffer, size_t N)
 311 {
 312         int  ret;
 313         int total=0;
 314
 315         while (total < N) {
 316                 io_flush();
 317
 318                 ret = read_unbuffered (fd, buffer + total, N-total);
 319                 total += ret;
 320         }
 321
 322         stats.total_read += total;
 323 }
 324
 325
 326 int32 read_int(int f)
 327 {
 328         char b[4];
 329         int32 ret;
 330
 331         readfd(f,b,4);
 332         ret = IVAL(b,0);
 333         if (ret == (int32)0xffffffff) return -1;
 334         return ret;
 335 }
 336
 337 int64 read_longint(int f)
 338 {
 339         extern int remote_version;
 340         int64 ret;
 341         char b[8];
 342         ret = read_int(f);
 343
 344         if ((int32)ret != (int32)0xffffffff) {
 345                 return ret;
 346         }
 347
 348 #ifdef NO_INT64
 349         rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
 350         exit_cleanup(RERR_UNSUPPORTED);
 351 #else
 352         if (remote_version >= 16) {
 353                 readfd(f,b,8);
 354                 ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
 355         }
 356 #endif
 357
 358         return ret;
 359 }
 360
 361 void read_buf(int f,char *buf,size_t len)
 362 {
 363         readfd(f,buf,len);
 364 }
 365
 366 void read_sbuf(int f,char *buf,size_t len)
 367 {
 368         read_buf (f,buf,len);
 369         buf[len] = 0;
 370 }
 371
 372 unsigned char read_byte(int f)
 373 {
 374         unsigned char c;
 375         read_buf (f, (char *)&c, 1);
 376         return c;
 377 }
 378
 379 /* write len bytes to fd */
 380 static void writefd_unbuffered(int fd,char *buf,size_t len)
 381 {
 382         int total = 0;
 383         fd_set w_fds, r_fds;
 384         int fd_count, count;
 385         struct timeval tv;
 386
 387         err_list_push();
 388
 389         no_flush++;
 390
 391         while (total < len) {
 392                 FD_ZERO(&w_fds);
 393                 FD_ZERO(&r_fds);
 394                 FD_SET(fd,&w_fds);
 395                 fd_count = fd;
 396
 397                 if (io_error_fd != -1) {
 398                         FD_SET(io_error_fd,&r_fds);
 399                         if (io_error_fd > fd_count)
 400                                 fd_count = io_error_fd;
 401                 }
 402
 403                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 404                 tv.tv_usec = 0;
 405
 406                 errno = 0;
 407
 408                 count = select(fd_count+1,
 409                                io_error_fd != -1?&r_fds:NULL,
 410                                &w_fds,NULL,
 411                                &tv);
 412
 413                 if (count == 0) {
 414                         check_timeout();
 415                 }
 416
 417                 if (count <= 0) {
 418                         if (errno == EBADF) {
 419                                 exit_cleanup(RERR_SOCKETIO);
 420                         }
 421                         continue;
 422                 }
 423
 424                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &r_fds)) {
 425                         read_error_fd();
 426                 }
 427
 428                 if (FD_ISSET(fd, &w_fds)) {
 429                         int ret, n = len-total;
 430                         ret = write(fd,buf+total,n);
 431
 432                         if (ret == -1 && errno == EINTR) {
 433                                 continue;
 434                         }
 435
 436                         if (ret == -1 &&
 437                             (errno == EWOULDBLOCK || errno == EAGAIN)) {
 438                                 msleep(1);
 439                                 continue;
 440                         }
 441
 442                         if (ret <= 0) {
 443                                 rprintf(FERROR,
 444                                         "error writing %d unbuffered bytes"
 445                                         " - exiting: %s\n", len,
 446                                         strerror(errno));
 447                                 exit_cleanup(RERR_STREAMIO);
 448                         }
 449
 450                         /* Sleep after writing to limit I/O bandwidth */
 451                         if (bwlimit)
 452                         {
 453                             tv.tv_sec = 0;
 454                             tv.tv_usec = ret * 1000 / bwlimit;
 455                             while (tv.tv_usec > 1000000)
 456                             {
 457                                 tv.tv_sec++;
 458                                 tv.tv_usec -= 1000000;
 459                             }
 460                             select(0, NULL, NULL, NULL, &tv);
 461                         }
 462
 463                         total += ret;
 464
 465                         if (io_timeout)
 466                                 last_io = time(NULL);
 467                 }
 468         }
 469
 470         no_flush--;
 471 }
 472
 473
 474 static char *io_buffer;
 475 static int io_buffer_count;
 476
 477 void io_start_buffering(int fd)
 478 {
 479         if (io_buffer) return;
 480         multiplex_out_fd = fd;
 481         io_buffer = (char *)malloc(IO_BUFFER_SIZE);
 482         if (!io_buffer) out_of_memory("writefd");
 483         io_buffer_count = 0;
 484 }
 485
 486 /* write an message to a multiplexed stream. If this fails then rsync
 487    exits */
 488 static void mplex_write(int fd, enum logcode code, char *buf, size_t len)
 489 {
 490         char buffer[4096];
 491         int n = len;
 492
 493         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
 494
 495         if (n > (sizeof(buffer)-4)) {
 496                 n = sizeof(buffer)-4;
 497         }
 498
 499         memcpy(&buffer[4], buf, n);
 500         writefd_unbuffered(fd, buffer, n+4);
 501
 502         len -= n;
 503         buf += n;
 504
 505         if (len) {
 506                 writefd_unbuffered(fd, buf, len);
 507         }
 508 }
 509
 510
 511 void io_flush(void)
 512 {
 513         int fd = multiplex_out_fd;
 514
 515         err_list_push();
 516
 517         if (!io_buffer_count || no_flush) return;
 518
 519         if (io_multiplexing_out) {
 520                 mplex_write(fd, FNONE, io_buffer, io_buffer_count);
 521         } else {
 522                 writefd_unbuffered(fd, io_buffer, io_buffer_count);
 523         }
 524         io_buffer_count = 0;
 525 }
 526
 527
 528 /* XXX: fd is ignored, which seems a little strange. */
 529 void io_end_buffering(int fd)
 530 {
 531         io_flush();
 532         if (!io_multiplexing_out) {
 533                 free(io_buffer);
 534                 io_buffer = NULL;
 535         }
 536 }
 537
 538 static void writefd(int fd,char *buf,size_t len)
 539 {
 540         stats.total_written += len;
 541
 542         err_list_push();
 543
 544         if (!io_buffer || fd != multiplex_out_fd) {
 545                 writefd_unbuffered(fd, buf, len);
 546                 return;
 547         }
 548
 549         while (len) {
 550                 int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count);
 551                 if (n > 0) {
 552                         memcpy(io_buffer+io_buffer_count, buf, n);
 553                         buf += n;
 554                         len -= n;
 555                         io_buffer_count += n;
 556                 }
 557
 558                 if (io_buffer_count == IO_BUFFER_SIZE) io_flush();
 559         }
 560 }
 561
 562
 563 void write_int(int f,int32 x)
 564 {
 565         char b[4];
 566         SIVAL(b,0,x);
 567         writefd(f,b,4);
 568 }
 569
 570
 571 /*
 572  * Note: int64 may actually be a 32-bit type if ./configure couldn't find any
 573  * 64-bit types on this platform.
 574  */
 575 void write_longint(int f, int64 x)
 576 {
 577         extern int remote_version;
 578         char b[8];
 579
 580         if (remote_version < 16 || x <= 0x7FFFFFFF) {
 581                 write_int(f, (int)x);
 582                 return;
 583         }
 584
 585         write_int(f, (int32)0xFFFFFFFF);
 586         SIVAL(b,0,(x&0xFFFFFFFF));
 587         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
 588
 589         writefd(f,b,8);
 590 }
 591
 592 void write_buf(int f,char *buf,size_t len)
 593 {
 594         writefd(f,buf,len);
 595 }
 596
 597 /* write a string to the connection */
 598 static void write_sbuf(int f,char *buf)
 599 {
 600         write_buf(f, buf, strlen(buf));
 601 }
 602
 603
 604 void write_byte(int f,unsigned char c)
 605 {
 606         write_buf(f,(char *)&c,1);
 607 }
 608
 609
 610
 611 int read_line(int f, char *buf, size_t maxlen)
 612 {
 613         while (maxlen) {
 614                 buf[0] = 0;
 615                 read_buf(f, buf, 1);
 616                 if (buf[0] == 0) return 0;
 617                 if (buf[0] == '\n') {
 618                         buf[0] = 0;
 619                         break;
 620                 }
 621                 if (buf[0] != '\r') {
 622                         buf++;
 623                         maxlen--;
 624                 }
 625         }
 626         if (maxlen == 0) {
 627                 *buf = 0;
 628                 return 0;
 629         }
 630
 631         return 1;
 632 }
 633
 634
 635 void io_printf(int fd, const char *format, ...)
 636 {
 637         va_list ap;
 638         char buf[1024];
 639         int len;
 640
 641         va_start(ap, format);
 642         len = vsnprintf(buf, sizeof(buf), format, ap);
 643         va_end(ap);
 644
 645         if (len < 0) exit_cleanup(RERR_STREAMIO);
 646
 647         write_sbuf(fd, buf);
 648 }
 649
 650
 651 /* setup for multiplexing an error stream with the data stream */
 652 void io_start_multiplex_out(int fd)
 653 {
 654         multiplex_out_fd = fd;
 655         io_flush();
 656         io_start_buffering(fd);
 657         io_multiplexing_out = 1;
 658 }
 659
 660 /* setup for multiplexing an error stream with the data stream */
 661 void io_start_multiplex_in(int fd)
 662 {
 663         multiplex_in_fd = fd;
 664         io_flush();
 665         io_multiplexing_in = 1;
 666 }
 667
 668 /* write an message to the multiplexed error stream */
 669 int io_multiplex_write(enum logcode code, char *buf, size_t len)
 670 {
 671         if (!io_multiplexing_out) return 0;
 672
 673         io_flush();
 674         stats.total_written += (len+4);
 675         mplex_write(multiplex_out_fd, code, buf, len);
 676         return 1;
 677 }
 678
 679 /* stop output multiplexing */
 680 void io_multiplexing_close(void)
 681 {
 682         io_multiplexing_out = 0;
 683 }
 684