io.c

   1 /*
   2    Copyright (C) Andrew Tridgell 1996
   3    Copyright (C) Paul Mackerras 1996
   4
   5    This program is free software; you can redistribute it and/or modify
   6    it under the terms of the GNU General Public License as published by
   7    the Free Software Foundation; either version 2 of the License, or
   8    (at your option) any later version.
   9
  10    This program is distributed in the hope that it will be useful,
  11    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13    GNU General Public License for more details.
  14
  15    You should have received a copy of the GNU General Public License
  16    along with this program; if not, write to the Free Software
  17    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  18 */
  19
  20 /*
  21   socket and pipe IO utilities used in rsync
  22
  23   tridge, June 1996
  24   */
  25 #include "rsync.h"
  26
  27 /* if no timeout is specified then use a 60 second select timeout */
  28 #define SELECT_TIMEOUT 60
  29
  30 extern int bwlimit;
  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 eof_error=1;
  38 extern int verbose;
  39 extern int io_timeout;
  40 extern struct stats stats;
  41
  42 static int io_error_fd = -1;
  43
  44 static void read_loop(int fd, char *buf, int len);
  45
  46 static void check_timeout(void)
  47 {
  48         extern int am_server, am_daemon;
  49         time_t t;
  50
  51         if (!io_timeout) return;
  52
  53         if (!last_io) {
  54                 last_io = time(NULL);
  55                 return;
  56         }
  57
  58         t = time(NULL);
  59
  60         if (last_io && io_timeout && (t-last_io) >= io_timeout) {
  61                 if (!am_server && !am_daemon) {
  62                         rprintf(FERROR,"io timeout after %d second - exiting\n",
  63                                 (int)(t-last_io));
  64                 }
  65                 exit_cleanup(RERR_TIMEOUT);
  66         }
  67 }
  68
  69 /* setup the fd used to propogate errors */
  70 void io_set_error_fd(int fd)
  71 {
  72         io_error_fd = fd;
  73 }
  74
  75 /* read some data from the error fd and write it to the write log code */
  76 static void read_error_fd(void)
  77 {
  78         char buf[200];
  79         int n;
  80         int fd = io_error_fd;
  81         int tag, len;
  82
  83         io_error_fd = -1;
  84
  85         read_loop(fd, buf, 4);
  86         tag = IVAL(buf, 0);
  87
  88         len = tag & 0xFFFFFF;
  89         tag = tag >> 24;
  90         tag -= MPLEX_BASE;
  91
  92         while (len) {
  93                 n = len;
  94                 if (n > (sizeof(buf)-1)) n = sizeof(buf)-1;
  95                 read_loop(fd, buf, n);
  96                 rwrite((enum logcode)tag, buf, n);
  97                 len -= n;
  98         }
  99
 100         io_error_fd = fd;
 101 }
 102
 103
 104 static int no_flush;
 105
 106 /* read from a socket with IO timeout. return the number of
 107    bytes read. If no bytes can be read then exit, never return
 108    a number <= 0 */
 109 static int read_timeout(int fd, char *buf, int len)
 110 {
 111         int n, ret=0;
 112
 113         io_flush();
 114
 115         while (ret == 0) {
 116                 fd_set fds;
 117                 struct timeval tv;
 118                 int fd_count = fd+1;
 119
 120                 FD_ZERO(&fds);
 121                 FD_SET(fd, &fds);
 122                 if (io_error_fd != -1) {
 123                         FD_SET(io_error_fd, &fds);
 124                         if (io_error_fd > fd) fd_count = io_error_fd+1;
 125                 }
 126
 127                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 128                 tv.tv_usec = 0;
 129
 130                 errno = 0;
 131
 132                 if (select(fd_count, &fds, NULL, NULL, &tv) < 1) {
 133                         if (errno == EBADF) {
 134                                 exit_cleanup(RERR_SOCKETIO);
 135                         }
 136                         check_timeout();
 137                         continue;
 138                 }
 139
 140                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &fds)) {
 141                         read_error_fd();
 142                 }
 143
 144                 if (!FD_ISSET(fd, &fds)) continue;
 145
 146                 n = read(fd, buf, len);
 147
 148                 if (n > 0) {
 149                         buf += n;
 150                         len -= n;
 151                         ret += n;
 152                         if (io_timeout)
 153                                 last_io = time(NULL);
 154                         continue;
 155                 }
 156
 157                 if (n == -1 && errno == EINTR) {
 158                         continue;
 159                 }
 160
 161                 if (n == -1 &&
 162                     (errno == EWOULDBLOCK || errno == EAGAIN)) {
 163                         continue;
 164                 }
 165
 166
 167                 if (n == 0) {
 168                         if (eof_error) {
 169                                 rprintf(FERROR,"unexpected EOF in read_timeout\n");
 170                         }
 171                         exit_cleanup(RERR_STREAMIO);
 172                 }
 173
 174                 /* this prevents us trying to write errors on a dead socket */
 175                 io_multiplexing_close();
 176
 177                 rprintf(FERROR,"read error: %s\n", strerror(errno));
 178                 exit_cleanup(RERR_STREAMIO);
 179         }
 180
 181         return ret;
 182 }
 183
 184 /* continue trying to read len bytes - don't return until len
 185    has been read */
 186 static void read_loop(int fd, char *buf, int len)
 187 {
 188         while (len) {
 189                 int n = read_timeout(fd, buf, len);
 190
 191                 buf += n;
 192                 len -= n;
 193         }
 194 }
 195
 196 /* read from the file descriptor handling multiplexing -
 197    return number of bytes read
 198    never return <= 0 */
 199 static int read_unbuffered(int fd, char *buf, int len)
 200 {
 201         static int remaining;
 202         int tag, ret=0;
 203         char line[1024];
 204
 205         if (!io_multiplexing_in || fd != multiplex_in_fd)
 206                 return read_timeout(fd, buf, len);
 207
 208         while (ret == 0) {
 209                 if (remaining) {
 210                         len = MIN(len, remaining);
 211                         read_loop(fd, buf, len);
 212                         remaining -= len;
 213                         ret = len;
 214                         continue;
 215                 }
 216
 217                 read_loop(fd, line, 4);
 218                 tag = IVAL(line, 0);
 219
 220                 remaining = tag & 0xFFFFFF;
 221                 tag = tag >> 24;
 222
 223                 if (tag == MPLEX_BASE) continue;
 224
 225                 tag -= MPLEX_BASE;
 226
 227                 if (tag != FERROR && tag != FINFO) {
 228                         rprintf(FERROR,"unexpected tag %d\n", tag);
 229                         exit_cleanup(RERR_STREAMIO);
 230                 }
 231
 232                 if (remaining > sizeof(line)-1) {
 233                         rprintf(FERROR,"multiplexing overflow %d\n\n",
 234                                 remaining);
 235                         exit_cleanup(RERR_STREAMIO);
 236                 }
 237
 238                 read_loop(fd, line, remaining);
 239                 line[remaining] = 0;
 240
 241                 rprintf((enum logcode)tag,"%s", line);
 242                 remaining = 0;
 243         }
 244
 245         return ret;
 246 }
 247
 248
 249 /* do a buffered read from fd. don't return until all N bytes
 250    have been read. If all N can't be read then exit with an error */
 251 static void readfd(int fd,char *buffer,int N)
 252 {
 253         int  ret;
 254         int total=0;
 255
 256         while (total < N) {
 257                 io_flush();
 258
 259                 ret = read_unbuffered(fd,buffer + total,N-total);
 260                 total += ret;
 261         }
 262
 263         stats.total_read += total;
 264 }
 265
 266
 267 int32 read_int(int f)
 268 {
 269         char b[4];
 270         int32 ret;
 271
 272         readfd(f,b,4);
 273         ret = IVAL(b,0);
 274         if (ret == (int32)0xffffffff) return -1;
 275         return ret;
 276 }
 277
 278 int64 read_longint(int f)
 279 {
 280         extern int remote_version;
 281         int64 ret;
 282         char b[8];
 283         ret = read_int(f);
 284
 285         if ((int32)ret != (int32)0xffffffff) {
 286                 return ret;
 287         }
 288
 289 #ifdef NO_INT64
 290         rprintf(FERROR,"Integer overflow - attempted 64 bit offset\n");
 291         exit_cleanup(RERR_UNSUPPORTED);
 292 #else
 293         if (remote_version >= 16) {
 294                 readfd(f,b,8);
 295                 ret = IVAL(b,0) | (((int64)IVAL(b,4))<<32);
 296         }
 297 #endif
 298
 299         return ret;
 300 }
 301
 302 void read_buf(int f,char *buf,int len)
 303 {
 304         readfd(f,buf,len);
 305 }
 306
 307 void read_sbuf(int f,char *buf,int len)
 308 {
 309         read_buf(f,buf,len);
 310         buf[len] = 0;
 311 }
 312
 313 unsigned char read_byte(int f)
 314 {
 315         unsigned char c;
 316         read_buf(f,(char *)&c,1);
 317         return c;
 318 }
 319
 320 /* write len bytes to fd */
 321 static void writefd_unbuffered(int fd,char *buf,int len)
 322 {
 323         int total = 0;
 324         fd_set w_fds, r_fds;
 325         int fd_count, count;
 326         struct timeval tv;
 327
 328         no_flush++;
 329
 330         while (total < len) {
 331                 FD_ZERO(&w_fds);
 332                 FD_ZERO(&r_fds);
 333                 FD_SET(fd,&w_fds);
 334                 fd_count = fd;
 335
 336                 if (io_error_fd != -1) {
 337                         FD_SET(io_error_fd,&r_fds);
 338                         if (io_error_fd > fd_count)
 339                                 fd_count = io_error_fd;
 340                 }
 341
 342                 tv.tv_sec = io_timeout?io_timeout:SELECT_TIMEOUT;
 343                 tv.tv_usec = 0;
 344
 345                 errno = 0;
 346
 347                 count = select(fd_count+1,
 348                                io_error_fd != -1?&r_fds:NULL,
 349                                &w_fds,NULL,
 350                                &tv);
 351
 352                 if (count <= 0) {
 353                         if (errno == EBADF) {
 354                                 exit_cleanup(RERR_SOCKETIO);
 355                         }
 356                         check_timeout();
 357                         continue;
 358                 }
 359
 360                 if (io_error_fd != -1 && FD_ISSET(io_error_fd, &r_fds)) {
 361                         read_error_fd();
 362                 }
 363
 364                 if (FD_ISSET(fd, &w_fds)) {
 365                         int ret, n = len-total;
 366
 367                         ret = write(fd,buf+total,n);
 368
 369                         if (ret == -1 && errno == EINTR) {
 370                                 continue;
 371                         }
 372
 373                         if (ret == -1 &&
 374                             (errno == EWOULDBLOCK || errno == EAGAIN)) {
 375                                 msleep(1);
 376                                 continue;
 377                         }
 378
 379                         if (ret <= 0) {
 380                                 rprintf(FERROR,"erroring writing %d bytes - exiting\n", len);
 381                                 exit_cleanup(RERR_STREAMIO);
 382                         }
 383
 384                         /* Sleep after writing to limit I/O bandwidth */
 385                         if (bwlimit)
 386                         {
 387                             tv.tv_sec = 0;
 388                             tv.tv_usec = ret * 1000 / bwlimit;
 389                             while (tv.tv_usec > 1000000)
 390                             {
 391                                 tv.tv_sec++;
 392                                 tv.tv_usec -= 1000000;
 393                             }
 394                             select(0, NULL, NULL, NULL, &tv);
 395                         }
 396
 397                         total += ret;
 398
 399                         if (io_timeout)
 400                                 last_io = time(NULL);
 401                 }
 402         }
 403
 404         no_flush--;
 405 }
 406
 407
 408 static char *io_buffer;
 409 static int io_buffer_count;
 410
 411 void io_start_buffering(int fd)
 412 {
 413         if (io_buffer) return;
 414         multiplex_out_fd = fd;
 415         io_buffer = (char *)malloc(IO_BUFFER_SIZE);
 416         if (!io_buffer) out_of_memory("writefd");
 417         io_buffer_count = 0;
 418 }
 419
 420 /* write an message to a multiplexed stream. If this fails then rsync
 421    exits */
 422 static void mplex_write(int fd, enum logcode code, char *buf, int len)
 423 {
 424         char buffer[4096];
 425         int n = len;
 426
 427         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
 428
 429         if (n > (sizeof(buffer)-4)) {
 430                 n = sizeof(buffer)-4;
 431         }
 432
 433         memcpy(&buffer[4], buf, n);
 434         writefd_unbuffered(fd, buffer, n+4);
 435
 436         len -= n;
 437         buf += n;
 438
 439         if (len) {
 440                 writefd_unbuffered(fd, buf, len);
 441         }
 442 }
 443
 444
 445 void io_flush(void)
 446 {
 447         int fd = multiplex_out_fd;
 448         if (!io_buffer_count || no_flush) return;
 449
 450         if (io_multiplexing_out) {
 451                 mplex_write(fd, FNONE, io_buffer, io_buffer_count);
 452         } else {
 453                 writefd_unbuffered(fd, io_buffer, io_buffer_count);
 454         }
 455         io_buffer_count = 0;
 456 }
 457
 458
 459 /* XXX: fd is ignored, which seems a little strange. */
 460 void io_end_buffering(int fd)
 461 {
 462         io_flush();
 463         if (!io_multiplexing_out) {
 464                 free(io_buffer);
 465                 io_buffer = NULL;
 466         }
 467 }
 468
 469 /* some OSes have a bug where an exit causes the pending writes on
 470    a socket to be flushed. Do an explicit shutdown to try to prevent this */
 471 void io_shutdown(void)
 472 {
 473         if (multiplex_out_fd != -1) close(multiplex_out_fd);
 474         if (io_error_fd != -1) close(io_error_fd);
 475         multiplex_out_fd = -1;
 476         io_error_fd = -1;
 477 }
 478
 479
 480 static void writefd(int fd,char *buf,int len)
 481 {
 482         stats.total_written += len;
 483
 484         if (!io_buffer || fd != multiplex_out_fd) {
 485                 writefd_unbuffered(fd, buf, len);
 486                 return;
 487         }
 488
 489         while (len) {
 490                 int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count);
 491                 if (n > 0) {
 492                         memcpy(io_buffer+io_buffer_count, buf, n);
 493                         buf += n;
 494                         len -= n;
 495                         io_buffer_count += n;
 496                 }
 497
 498                 if (io_buffer_count == IO_BUFFER_SIZE) io_flush();
 499         }
 500 }
 501
 502
 503 void write_int(int f,int32 x)
 504 {
 505         char b[4];
 506         SIVAL(b,0,x);
 507         writefd(f,b,4);
 508 }
 509
 510 void write_longint(int f, int64 x)
 511 {
 512         extern int remote_version;
 513         char b[8];
 514
 515         if (remote_version < 16 || x <= 0x7FFFFFFF) {
 516                 write_int(f, (int)x);
 517                 return;
 518         }
 519
 520         write_int(f, (int32)0xFFFFFFFF);
 521         SIVAL(b,0,(x&0xFFFFFFFF));
 522         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
 523
 524         writefd(f,b,8);
 525 }
 526
 527 void write_buf(int f,char *buf,int len)
 528 {
 529         writefd(f,buf,len);
 530 }
 531
 532 /* write a string to the connection */
 533 static void write_sbuf(int f,char *buf)
 534 {
 535         write_buf(f, buf, strlen(buf));
 536 }
 537
 538
 539 void write_byte(int f,unsigned char c)
 540 {
 541         write_buf(f,(char *)&c,1);
 542 }
 543
 544 int read_line(int f, char *buf, int maxlen)
 545 {
 546         eof_error = 0;
 547
 548         while (maxlen) {
 549                 buf[0] = 0;
 550                 read_buf(f, buf, 1);
 551                 if (buf[0] == 0) return 0;
 552                 if (buf[0] == '\n') {
 553                         buf[0] = 0;
 554                         break;
 555                 }
 556                 if (buf[0] != '\r') {
 557                         buf++;
 558                         maxlen--;
 559                 }
 560         }
 561         if (maxlen == 0) {
 562                 *buf = 0;
 563                 return 0;
 564         }
 565
 566         eof_error = 1;
 567
 568         return 1;
 569 }
 570
 571
 572 void io_printf(int fd, const char *format, ...)
 573 {
 574         va_list ap;
 575         char buf[1024];
 576         int len;
 577
 578         va_start(ap, format);
 579         len = vslprintf(buf, sizeof(buf), format, ap);
 580         va_end(ap);
 581
 582         if (len < 0) exit_cleanup(RERR_STREAMIO);
 583
 584         write_sbuf(fd, buf);
 585 }
 586
 587
 588 /* setup for multiplexing an error stream with the data stream */
 589 void io_start_multiplex_out(int fd)
 590 {
 591         multiplex_out_fd = fd;
 592         io_flush();
 593         io_start_buffering(fd);
 594         io_multiplexing_out = 1;
 595 }
 596
 597 /* setup for multiplexing an error stream with the data stream */
 598 void io_start_multiplex_in(int fd)
 599 {
 600         multiplex_in_fd = fd;
 601         io_flush();
 602         io_multiplexing_in = 1;
 603 }
 604
 605 /* write an message to the multiplexed error stream */
 606 int io_multiplex_write(enum logcode code, char *buf, int len)
 607 {
 608         if (!io_multiplexing_out) return 0;
 609
 610         io_flush();
 611         stats.total_written += (len+4);
 612         mplex_write(multiplex_out_fd, code, buf, len);
 613         return 1;
 614 }
 615
 616 /* write a message to the special error fd */
 617 int io_error_write(int f, enum logcode code, char *buf, int len)
 618 {
 619         if (f == -1) return 0;
 620         mplex_write(f, code, buf, len);
 621         return 1;
 622 }
 623
 624 /* stop output multiplexing */
 625 void io_multiplexing_close(void)
 626 {
 627         io_multiplexing_out = 0;
 628 }
 629