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                                 continue;
 376                         }
 377
 378                         if (ret <= 0) {
 379                                 rprintf(FERROR,"erroring writing %d bytes - exiting\n", len);
 380                                 exit_cleanup(RERR_STREAMIO);
 381                         }
 382
 383                         /* Sleep after writing to limit I/O bandwidth */
 384                         if (bwlimit)
 385                         {
 386                             tv.tv_sec = 0;
 387                             tv.tv_usec = ret * 1000 / bwlimit;
 388                             while (tv.tv_usec > 1000000)
 389                             {
 390                                 tv.tv_sec++;
 391                                 tv.tv_usec -= 1000000;
 392                             }
 393                             select(0, NULL, NULL, NULL, &tv);
 394                         }
 395
 396                         total += ret;
 397
 398                         if (io_timeout)
 399                                 last_io = time(NULL);
 400                 }
 401         }
 402
 403         no_flush--;
 404 }
 405
 406
 407 static char *io_buffer;
 408 static int io_buffer_count;
 409
 410 void io_start_buffering(int fd)
 411 {
 412         if (io_buffer) return;
 413         multiplex_out_fd = fd;
 414         io_buffer = (char *)malloc(IO_BUFFER_SIZE);
 415         if (!io_buffer) out_of_memory("writefd");
 416         io_buffer_count = 0;
 417 }
 418
 419 /* write an message to a multiplexed stream. If this fails then rsync
 420    exits */
 421 static void mplex_write(int fd, enum logcode code, char *buf, int len)
 422 {
 423         char buffer[4096];
 424         int n = len;
 425
 426         SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
 427
 428         if (n > (sizeof(buffer)-4)) {
 429                 n = sizeof(buffer)-4;
 430         }
 431
 432         memcpy(&buffer[4], buf, n);
 433         writefd_unbuffered(fd, buffer, n+4);
 434
 435         len -= n;
 436         buf += n;
 437
 438         if (len) {
 439                 writefd_unbuffered(fd, buf, len);
 440         }
 441 }
 442
 443
 444 void io_flush(void)
 445 {
 446         int fd = multiplex_out_fd;
 447         if (!io_buffer_count || no_flush) return;
 448
 449         if (io_multiplexing_out) {
 450                 mplex_write(fd, FNONE, io_buffer, io_buffer_count);
 451         } else {
 452                 writefd_unbuffered(fd, io_buffer, io_buffer_count);
 453         }
 454         io_buffer_count = 0;
 455 }
 456
 457 void io_end_buffering(int fd)
 458 {
 459         io_flush();
 460         if (!io_multiplexing_out) {
 461                 free(io_buffer);
 462                 io_buffer = NULL;
 463         }
 464 }
 465
 466 static void writefd(int fd,char *buf,int len)
 467 {
 468         stats.total_written += len;
 469
 470         if (!io_buffer || fd != multiplex_out_fd) {
 471                 writefd_unbuffered(fd, buf, len);
 472                 return;
 473         }
 474
 475         while (len) {
 476                 int n = MIN(len, IO_BUFFER_SIZE-io_buffer_count);
 477                 if (n > 0) {
 478                         memcpy(io_buffer+io_buffer_count, buf, n);
 479                         buf += n;
 480                         len -= n;
 481                         io_buffer_count += n;
 482                 }
 483
 484                 if (io_buffer_count == IO_BUFFER_SIZE) io_flush();
 485         }
 486 }
 487
 488
 489 void write_int(int f,int32 x)
 490 {
 491         char b[4];
 492         SIVAL(b,0,x);
 493         writefd(f,b,4);
 494 }
 495
 496 void write_longint(int f, int64 x)
 497 {
 498         extern int remote_version;
 499         char b[8];
 500
 501         if (remote_version < 16 || x <= 0x7FFFFFFF) {
 502                 write_int(f, (int)x);
 503                 return;
 504         }
 505
 506         write_int(f, (int32)0xFFFFFFFF);
 507         SIVAL(b,0,(x&0xFFFFFFFF));
 508         SIVAL(b,4,((x>>32)&0xFFFFFFFF));
 509
 510         writefd(f,b,8);
 511 }
 512
 513 void write_buf(int f,char *buf,int len)
 514 {
 515         writefd(f,buf,len);
 516 }
 517
 518 /* write a string to the connection */
 519 static void write_sbuf(int f,char *buf)
 520 {
 521         write_buf(f, buf, strlen(buf));
 522 }
 523
 524
 525 void write_byte(int f,unsigned char c)
 526 {
 527         write_buf(f,(char *)&c,1);
 528 }
 529
 530 int read_line(int f, char *buf, int maxlen)
 531 {
 532         eof_error = 0;
 533
 534         while (maxlen) {
 535                 buf[0] = 0;
 536                 read_buf(f, buf, 1);
 537                 if (buf[0] == 0) return 0;
 538                 if (buf[0] == '\n') {
 539                         buf[0] = 0;
 540                         break;
 541                 }
 542                 if (buf[0] != '\r') {
 543                         buf++;
 544                         maxlen--;
 545                 }
 546         }
 547         if (maxlen == 0) {
 548                 *buf = 0;
 549                 return 0;
 550         }
 551
 552         eof_error = 1;
 553
 554         return 1;
 555 }
 556
 557
 558 void io_printf(int fd, const char *format, ...)
 559 {
 560         va_list ap;
 561         char buf[1024];
 562         int len;
 563
 564         va_start(ap, format);
 565         len = vslprintf(buf, sizeof(buf), format, ap);
 566         va_end(ap);
 567
 568         if (len < 0) exit_cleanup(RERR_STREAMIO);
 569
 570         write_sbuf(fd, buf);
 571 }
 572
 573
 574 /* setup for multiplexing an error stream with the data stream */
 575 void io_start_multiplex_out(int fd)
 576 {
 577         multiplex_out_fd = fd;
 578         io_flush();
 579         io_start_buffering(fd);
 580         io_multiplexing_out = 1;
 581 }
 582
 583 /* setup for multiplexing an error stream with the data stream */
 584 void io_start_multiplex_in(int fd)
 585 {
 586         multiplex_in_fd = fd;
 587         io_flush();
 588         io_multiplexing_in = 1;
 589 }
 590
 591 /* write an message to the multiplexed error stream */
 592 int io_multiplex_write(enum logcode code, char *buf, int len)
 593 {
 594         if (!io_multiplexing_out) return 0;
 595
 596         io_flush();
 597         stats.total_written += (len+4);
 598         mplex_write(multiplex_out_fd, code, buf, len);
 599         return 1;
 600 }
 601
 602 /* write a message to the special error fd */
 603 int io_error_write(int f, enum logcode code, char *buf, int len)
 604 {
 605         if (f == -1) return 0;
 606         mplex_write(f, code, buf, len);
 607         return 1;
 608 }
 609
 610 /* stop output multiplexing */
 611 void io_multiplexing_close(void)
 612 {
 613         io_multiplexing_out = 0;
 614 }
 615