1 /* $Id: uploader.c,v 1.23 2019/08/26 22:22:14 benno Exp $ */ 2 /* 3 * Copyright (c) 2019 Kristaps Dzonsons <kristaps@bsd.lv> 4 * Copyright (c) 2019 Florian Obser <florian@openbsd.org> 5 * 6 * Permission to use, copy, modify, and distribute this software for any 7 * purpose with or without fee is hereby granted, provided that the above 8 * copyright notice and this permission notice appear in all copies. 9 * 10 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES 11 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF 12 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR 13 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES 14 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN 15 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF 16 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. 17 */ 18 #include <sys/mman.h> 19 #include <sys/stat.h> 20 21 #include <assert.h> 22 #include <errno.h> 23 #include <fcntl.h> 24 #include <inttypes.h> 25 #include <math.h> 26 #include <stdio.h> 27 #include <stdlib.h> 28 #include <string.h> 29 #include <time.h> 30 #include <unistd.h> 31 32 #include "extern.h" 33 34 enum uploadst { 35 UPLOAD_FIND_NEXT = 0, /* find next to upload to sender */ 36 UPLOAD_WRITE_LOCAL, /* wait to write to sender */ 37 UPLOAD_READ_LOCAL, /* wait to read from local file */ 38 UPLOAD_FINISHED /* nothing more to do in phase */ 39 }; 40 41 /* 42 * Used to keep track of data flowing from the receiver to the sender. 43 * This is managed by the receiver process. 44 */ 45 struct upload { 46 enum uploadst state; 47 char *buf; /* if not NULL, pending upload */ 48 size_t bufsz; /* size of buf */ 49 size_t bufmax; /* maximum size of buf */ 50 size_t bufpos; /* position in buf */ 51 size_t idx; /* current transfer index */ 52 mode_t oumask; /* umask for creating files */ 53 char *root; /* destination directory path */ 54 int rootfd; /* destination directory */ 55 size_t csumlen; /* checksum length */ 56 int fdout; /* write descriptor to sender */ 57 const struct flist *fl; /* file list */ 58 size_t flsz; /* size of file list */ 59 int *newdir; /* non-zero if mkdir'd */ 60 }; 61 62 /* 63 * Log a directory by emitting the file and a trailing slash, just to 64 * show the operator that we're a directory. 65 */ 66 static void 67 log_dir(struct sess *sess, const struct flist *f) 68 { 69 size_t sz; 70 71 if (sess->opts->server) 72 return; 73 sz = strlen(f->path); 74 assert(sz > 0); 75 LOG1("%s%s", f->path, (f->path[sz - 1] == '/') ? "" : "/"); 76 } 77 78 /* 79 * Log a link by emitting the file and the target, just to show the 80 * operator that we're a link. 81 */ 82 static void 83 log_link(struct sess *sess, const struct flist *f) 84 { 85 86 if (!sess->opts->server) 87 LOG1("%s -> %s", f->path, f->link); 88 } 89 90 /* 91 * Simply log the filename. 92 */ 93 static void 94 log_file(struct sess *sess, const struct flist *f) 95 { 96 97 if (!sess->opts->server) 98 LOG1("%s", f->path); 99 } 100 101 /* 102 * Prepare the overall block set's metadata. 103 * We always have at least one block. 104 * The block size is an important part of the algorithm. 105 * I use the same heuristic as the reference rsync, but implemented in a 106 * bit more of a straightforward way. 107 * In general, the individual block length is the rounded square root of 108 * the total file size. 109 * The minimum block length is 700. 110 */ 111 static void 112 init_blkset(struct blkset *p, off_t sz) 113 { 114 double v; 115 116 if (sz >= (BLOCK_SIZE_MIN * BLOCK_SIZE_MIN)) { 117 /* Simple rounded-up integer square root. */ 118 119 v = sqrt(sz); 120 p->len = ceil(v); 121 122 /* 123 * Always be a multiple of eight. 124 * There's no reason to do this, but rsync does. 125 */ 126 127 if ((p->len % 8) > 0) 128 p->len += 8 - (p->len % 8); 129 } else 130 p->len = BLOCK_SIZE_MIN; 131 132 p->size = sz; 133 if ((p->blksz = sz / p->len) == 0) 134 p->rem = sz; 135 else 136 p->rem = sz % p->len; 137 138 /* If we have a remainder, then we need an extra block. */ 139 140 if (p->rem) 141 p->blksz++; 142 } 143 144 /* 145 * For each block, prepare the block's metadata. 146 * We use the mapped "map" file to set our checksums. 147 */ 148 static void 149 init_blk(struct blk *p, const struct blkset *set, off_t offs, 150 size_t idx, const void *map, const struct sess *sess) 151 { 152 153 assert(map != MAP_FAILED); 154 155 /* Block length inherits for all but the last. */ 156 157 p->idx = idx; 158 p->len = idx < set->blksz - 1 ? set->len : set->rem; 159 p->offs = offs; 160 161 p->chksum_short = hash_fast(map, p->len); 162 hash_slow(map, p->len, p->chksum_long, sess); 163 } 164 165 /* 166 * Handle a symbolic link. 167 * If we encounter directories existing in the symbolic link's place, 168 * then try to unlink the directory. 169 * Otherwise, simply overwrite with the symbolic link by renaming. 170 * Return <0 on failure 0 on success. 171 */ 172 static int 173 pre_link(struct upload *p, struct sess *sess) 174 { 175 struct stat st; 176 const struct flist *f; 177 int rc, newlink = 0, updatelink = 0; 178 char *b, *temp = NULL; 179 180 f = &p->fl[p->idx]; 181 assert(S_ISLNK(f->st.mode)); 182 183 if (!sess->opts->preserve_links) { 184 WARNX("%s: ignoring symlink", f->path); 185 return 0; 186 } else if (sess->opts->dry_run) { 187 log_link(sess, f); 188 return 0; 189 } 190 191 /* 192 * See if the symlink already exists. 193 * If it's a directory, then try to unlink the directory prior 194 * to overwriting with a symbolic link. 195 * If it's a non-directory, we just overwrite it. 196 */ 197 198 assert(p->rootfd != -1); 199 rc = fstatat(p->rootfd, f->path, &st, AT_SYMLINK_NOFOLLOW); 200 if (rc != -1 && !S_ISLNK(st.st_mode)) { 201 if (S_ISDIR(st.st_mode) && 202 unlinkat(p->rootfd, f->path, AT_REMOVEDIR) == -1) { 203 ERR("%s: unlinkat", f->path); 204 return -1; 205 } 206 rc = -1; 207 } else if (rc == -1 && errno != ENOENT) { 208 ERR("%s: fstatat", f->path); 209 return -1; 210 } 211 212 /* 213 * If the symbolic link already exists, then make sure that it 214 * points to the correct place. 215 */ 216 217 if (rc != -1) { 218 b = symlinkat_read(p->rootfd, f->path); 219 if (b == NULL) { 220 ERRX1("symlinkat_read"); 221 return -1; 222 } 223 if (strcmp(f->link, b)) { 224 free(b); 225 b = NULL; 226 LOG3("%s: updating symlink: %s", f->path, f->link); 227 updatelink = 1; 228 } 229 free(b); 230 b = NULL; 231 } 232 233 /* 234 * Create the temporary file as a symbolic link, then rename the 235 * temporary file as the real one, overwriting anything there. 236 */ 237 238 if (rc == -1 || updatelink) { 239 LOG3("%s: creating symlink: %s", f->path, f->link); 240 if (mktemplate(&temp, f->path, sess->opts->recursive) == -1) { 241 ERRX1("mktemplate"); 242 return -1; 243 } 244 if (mkstemplinkat(f->link, p->rootfd, temp) == NULL) { 245 ERR("mkstemplinkat"); 246 free(temp); 247 return -1; 248 } 249 newlink = 1; 250 } 251 252 rsync_set_metadata_at(sess, newlink, 253 p->rootfd, f, newlink ? temp : f->path); 254 255 if (newlink) { 256 if (renameat(p->rootfd, temp, p->rootfd, f->path) == -1) { 257 ERR("%s: renameat %s", temp, f->path); 258 (void)unlinkat(p->rootfd, temp, 0); 259 free(temp); 260 return -1; 261 } 262 free(temp); 263 } 264 265 log_link(sess, f); 266 return 0; 267 } 268 269 /* 270 * See pre_link(), but for devices. 271 * FIXME: this is very similar to the other pre_xxx() functions. 272 * Return <0 on failure 0 on success. 273 */ 274 static int 275 pre_dev(struct upload *p, struct sess *sess) 276 { 277 struct stat st; 278 const struct flist *f; 279 int rc, newdev = 0, updatedev = 0; 280 char *temp = NULL; 281 282 f = &p->fl[p->idx]; 283 assert(S_ISBLK(f->st.mode) || S_ISCHR(f->st.mode)); 284 285 if (!sess->opts->devices || getuid() != 0) { 286 WARNX("skipping non-regular file %s", f->path); 287 return 0; 288 } else if (sess->opts->dry_run) { 289 log_file(sess, f); 290 return 0; 291 } 292 293 /* 294 * See if the dev already exists. 295 * If a non-device exists in its place, we'll replace that. 296 * If it replaces a directory, remove the directory first. 297 */ 298 299 assert(p->rootfd != -1); 300 rc = fstatat(p->rootfd, f->path, &st, AT_SYMLINK_NOFOLLOW); 301 302 if (rc != -1 && !(S_ISBLK(st.st_mode) || S_ISCHR(st.st_mode))) { 303 if (S_ISDIR(st.st_mode) && 304 unlinkat(p->rootfd, f->path, AT_REMOVEDIR) == -1) { 305 ERR("%s: unlinkat", f->path); 306 return -1; 307 } 308 rc = -1; 309 } else if (rc == -1 && errno != ENOENT) { 310 ERR("%s: fstatat", f->path); 311 return -1; 312 } 313 314 /* Make sure existing device is of the correct type. */ 315 316 if (rc != -1) { 317 if ((f->st.mode & (S_IFCHR|S_IFBLK)) != 318 (st.st_mode & (S_IFCHR|S_IFBLK)) || 319 f->st.rdev != st.st_rdev) { 320 LOG3("%s: updating device", f->path); 321 updatedev = 1; 322 } 323 } 324 325 if (rc == -1 || updatedev) { 326 newdev = 1; 327 if (mktemplate(&temp, f->path, sess->opts->recursive) == -1) { 328 ERRX1("mktemplate"); 329 return -1; 330 } 331 if (mkstempnodat(p->rootfd, temp, 332 f->st.mode & (S_IFCHR|S_IFBLK), f->st.rdev) == NULL) { 333 ERR("mkstempnodat"); 334 free(temp); 335 return -1; 336 } 337 } 338 339 rsync_set_metadata_at(sess, newdev, 340 p->rootfd, f, newdev ? temp : f->path); 341 342 if (newdev) { 343 if (renameat(p->rootfd, temp, p->rootfd, f->path) == -1) { 344 ERR("%s: renameat %s", temp, f->path); 345 (void)unlinkat(p->rootfd, temp, 0); 346 free(temp); 347 return -1; 348 } 349 free(temp); 350 } 351 352 log_file(sess, f); 353 return 0; 354 } 355 356 /* 357 * See pre_link(), but for FIFOs. 358 * FIXME: this is very similar to the other pre_xxx() functions. 359 * Return <0 on failure 0 on success. 360 */ 361 static int 362 pre_fifo(struct upload *p, struct sess *sess) 363 { 364 struct stat st; 365 const struct flist *f; 366 int rc, newfifo = 0; 367 char *temp = NULL; 368 369 f = &p->fl[p->idx]; 370 assert(S_ISFIFO(f->st.mode)); 371 372 if (!sess->opts->specials) { 373 WARNX("skipping non-regular file %s", f->path); 374 return 0; 375 } else if (sess->opts->dry_run) { 376 log_file(sess, f); 377 return 0; 378 } 379 380 /* 381 * See if the fifo already exists. 382 * If it exists as a non-FIFO, unlink it (if a directory) then 383 * mark it from replacement. 384 */ 385 386 assert(p->rootfd != -1); 387 rc = fstatat(p->rootfd, f->path, &st, AT_SYMLINK_NOFOLLOW); 388 389 if (rc != -1 && !S_ISFIFO(st.st_mode)) { 390 if (S_ISDIR(st.st_mode) && 391 unlinkat(p->rootfd, f->path, AT_REMOVEDIR) == -1) { 392 ERR("%s: unlinkat", f->path); 393 return -1; 394 } 395 rc = -1; 396 } else if (rc == -1 && errno != ENOENT) { 397 ERR("%s: fstatat", f->path); 398 return -1; 399 } 400 401 if (rc == -1) { 402 newfifo = 1; 403 if (mktemplate(&temp, f->path, sess->opts->recursive) == -1) { 404 ERRX1("mktemplate"); 405 return -1; 406 } 407 if (mkstempfifoat(p->rootfd, temp) == NULL) { 408 ERR("mkstempfifoat"); 409 free(temp); 410 return -1; 411 } 412 } 413 414 rsync_set_metadata_at(sess, newfifo, 415 p->rootfd, f, newfifo ? temp : f->path); 416 417 if (newfifo) { 418 if (renameat(p->rootfd, temp, p->rootfd, f->path) == -1) { 419 ERR("%s: renameat %s", temp, f->path); 420 (void)unlinkat(p->rootfd, temp, 0); 421 free(temp); 422 return -1; 423 } 424 free(temp); 425 } 426 427 log_file(sess, f); 428 return 0; 429 } 430 431 /* 432 * See pre_link(), but for socket files. 433 * FIXME: this is very similar to the other pre_xxx() functions. 434 * Return <0 on failure 0 on success. 435 */ 436 static int 437 pre_sock(struct upload *p, struct sess *sess) 438 { 439 struct stat st; 440 const struct flist *f; 441 int rc, newsock = 0; 442 char *temp = NULL; 443 444 f = &p->fl[p->idx]; 445 assert(S_ISSOCK(f->st.mode)); 446 447 if (!sess->opts->specials) { 448 WARNX("skipping non-regular file %s", f->path); 449 return 0; 450 } else if (sess->opts->dry_run) { 451 log_file(sess, f); 452 return 0; 453 } 454 455 /* 456 * See if the fifo already exists. 457 * If it exists as a non-FIFO, unlink it (if a directory) then 458 * mark it from replacement. 459 */ 460 461 assert(p->rootfd != -1); 462 rc = fstatat(p->rootfd, f->path, &st, AT_SYMLINK_NOFOLLOW); 463 464 if (rc != -1 && !S_ISSOCK(st.st_mode)) { 465 if (S_ISDIR(st.st_mode) && 466 unlinkat(p->rootfd, f->path, AT_REMOVEDIR) == -1) { 467 ERR("%s: unlinkat", f->path); 468 return -1; 469 } 470 rc = -1; 471 } else if (rc == -1 && errno != ENOENT) { 472 ERR("%s: fstatat", f->path); 473 return -1; 474 } 475 476 if (rc == -1) { 477 newsock = 1; 478 if (mktemplate(&temp, f->path, sess->opts->recursive) == -1) { 479 ERRX1("mktemplate"); 480 return -1; 481 } 482 if (mkstempsock(p->root, temp) == NULL) { 483 ERR("mkstempsock"); 484 free(temp); 485 return -1; 486 } 487 } 488 489 rsync_set_metadata_at(sess, newsock, 490 p->rootfd, f, newsock ? temp : f->path); 491 492 if (newsock) { 493 if (renameat(p->rootfd, temp, p->rootfd, f->path) == -1) { 494 ERR("%s: renameat %s", temp, f->path); 495 (void)unlinkat(p->rootfd, temp, 0); 496 free(temp); 497 return -1; 498 } 499 free(temp); 500 } 501 502 log_file(sess, f); 503 return 0; 504 } 505 506 /* 507 * If not found, create the destination directory in prefix order. 508 * Create directories using the existing umask. 509 * Return <0 on failure 0 on success. 510 */ 511 static int 512 pre_dir(const struct upload *p, struct sess *sess) 513 { 514 struct stat st; 515 int rc; 516 const struct flist *f; 517 518 f = &p->fl[p->idx]; 519 assert(S_ISDIR(f->st.mode)); 520 521 if (!sess->opts->recursive) { 522 WARNX("%s: ignoring directory", f->path); 523 return 0; 524 } else if (sess->opts->dry_run) { 525 log_dir(sess, f); 526 return 0; 527 } 528 529 assert(p->rootfd != -1); 530 rc = fstatat(p->rootfd, f->path, &st, AT_SYMLINK_NOFOLLOW); 531 532 if (rc == -1 && errno != ENOENT) { 533 ERR("%s: fstatat", f->path); 534 return -1; 535 } else if (rc != -1 && !S_ISDIR(st.st_mode)) { 536 ERRX("%s: not a directory", f->path); 537 return -1; 538 } else if (rc != -1) { 539 /* 540 * FIXME: we should fchmod the permissions here as well, 541 * as we may locally have shut down writing into the 542 * directory and that doesn't work. 543 */ 544 LOG3("%s: updating directory", f->path); 545 return 0; 546 } 547 548 /* 549 * We want to make the directory with default permissions (using 550 * our old umask, which we've since unset), then adjust 551 * permissions (assuming preserve_perms or new) afterward in 552 * case it's u-w or something. 553 */ 554 555 LOG3("%s: creating directory", f->path); 556 if (mkdirat(p->rootfd, f->path, 0777 & ~p->oumask) == -1) { 557 ERR("%s: mkdirat", f->path); 558 return -1; 559 } 560 561 p->newdir[p->idx] = 1; 562 log_dir(sess, f); 563 return 0; 564 } 565 566 /* 567 * Process the directory time and mode for "idx" in the file list. 568 * Returns zero on failure, non-zero on success. 569 */ 570 static int 571 post_dir(struct sess *sess, const struct upload *u, size_t idx) 572 { 573 struct timespec tv[2]; 574 int rc; 575 struct stat st; 576 const struct flist *f; 577 578 f = &u->fl[idx]; 579 assert(S_ISDIR(f->st.mode)); 580 581 /* We already warned about the directory in pre_process_dir(). */ 582 583 if (!sess->opts->recursive) 584 return 1; 585 else if (sess->opts->dry_run) 586 return 1; 587 588 if (fstatat(u->rootfd, f->path, &st, AT_SYMLINK_NOFOLLOW) == -1) { 589 ERR("%s: fstatat", f->path); 590 return 0; 591 } else if (!S_ISDIR(st.st_mode)) { 592 WARNX("%s: not a directory", f->path); 593 return 0; 594 } 595 596 /* 597 * Update the modification time if we're a new directory *or* if 598 * we're preserving times and the time has changed. 599 * FIXME: run rsync_set_metadata()? 600 */ 601 602 if (u->newdir[idx] || 603 (sess->opts->preserve_times && 604 st.st_mtime != f->st.mtime)) { 605 tv[0].tv_sec = time(NULL); 606 tv[0].tv_nsec = 0; 607 tv[1].tv_sec = f->st.mtime; 608 tv[1].tv_nsec = 0; 609 rc = utimensat(u->rootfd, f->path, tv, 0); 610 if (rc == -1) { 611 ERR("%s: utimensat", f->path); 612 return 0; 613 } 614 LOG4("%s: updated date", f->path); 615 } 616 617 /* 618 * Update the mode if we're a new directory *or* if we're 619 * preserving modes and it has changed. 620 */ 621 622 if (u->newdir[idx] || 623 (sess->opts->preserve_perms && st.st_mode != f->st.mode)) { 624 rc = fchmodat(u->rootfd, f->path, f->st.mode, 0); 625 if (rc == -1) { 626 ERR("%s: fchmodat", f->path); 627 return 0; 628 } 629 LOG4("%s: updated mode", f->path); 630 } 631 632 return 1; 633 } 634 635 /* 636 * Try to open the file at the current index. 637 * If the file does not exist, returns with success. 638 * Return <0 on failure, 0 on success w/nothing to be done, >0 on 639 * success and the file needs attention. 640 */ 641 static int 642 pre_file(const struct upload *p, int *filefd, struct sess *sess) 643 { 644 const struct flist *f; 645 646 f = &p->fl[p->idx]; 647 assert(S_ISREG(f->st.mode)); 648 649 if (sess->opts->dry_run) { 650 log_file(sess, f); 651 if (!io_write_int(sess, p->fdout, p->idx)) { 652 ERRX1("io_write_int"); 653 return -1; 654 } 655 return 0; 656 } 657 658 /* 659 * For non dry-run cases, we'll write the acknowledgement later 660 * in the rsync_uploader() function because we need to wait for 661 * the open() call to complete. 662 * If the call to openat() fails with ENOENT, there's a 663 * fast-path between here and the write function, so we won't do 664 * any blocking between now and then. 665 */ 666 667 *filefd = openat(p->rootfd, f->path, 668 O_RDONLY | O_NOFOLLOW | O_NONBLOCK, 0); 669 if (*filefd != -1 || errno == ENOENT) 670 return 1; 671 ERR("%s: openat", f->path); 672 return -1; 673 } 674 675 /* 676 * Allocate an uploader object in the correct state to start. 677 * Returns NULL on failure or the pointer otherwise. 678 * On success, upload_free() must be called with the allocated pointer. 679 */ 680 struct upload * 681 upload_alloc(const char *root, int rootfd, int fdout, 682 size_t clen, const struct flist *fl, size_t flsz, mode_t msk) 683 { 684 struct upload *p; 685 686 if ((p = calloc(1, sizeof(struct upload))) == NULL) { 687 ERR("calloc"); 688 return NULL; 689 } 690 691 p->state = UPLOAD_FIND_NEXT; 692 p->oumask = msk; 693 p->root = strdup(root); 694 if (p->root == NULL) { 695 ERR("strdup"); 696 free(p); 697 return NULL; 698 } 699 p->rootfd = rootfd; 700 p->csumlen = clen; 701 p->fdout = fdout; 702 p->fl = fl; 703 p->flsz = flsz; 704 p->newdir = calloc(flsz, sizeof(int)); 705 if (p->newdir == NULL) { 706 ERR("calloc"); 707 free(p->root); 708 free(p); 709 return NULL; 710 } 711 return p; 712 } 713 714 /* 715 * Perform all cleanups and free. 716 * Passing a NULL to this function is ok. 717 */ 718 void 719 upload_free(struct upload *p) 720 { 721 722 if (p == NULL) 723 return; 724 free(p->root); 725 free(p->newdir); 726 free(p->buf); 727 free(p); 728 } 729 730 /* 731 * Iterates through all available files and conditionally gets the file 732 * ready for processing to check whether it's up to date. 733 * If not up to date or empty, sends file information to the sender. 734 * If returns 0, we've processed all files there are to process. 735 * If returns >0, we're waiting for POLLIN or POLLOUT data. 736 * Otherwise returns <0, which is an error. 737 */ 738 int 739 rsync_uploader(struct upload *u, int *fileinfd, 740 struct sess *sess, int *fileoutfd) 741 { 742 struct blkset blk; 743 struct stat st; 744 void *mbuf, *bufp; 745 ssize_t msz; 746 size_t i, pos, sz; 747 off_t offs; 748 int c; 749 const struct flist *f; 750 751 /* This should never get called. */ 752 753 assert(u->state != UPLOAD_FINISHED); 754 755 /* 756 * If we have an upload in progress, then keep writing until the 757 * buffer has been fully written. 758 * We must only have the output file descriptor working and also 759 * have a valid buffer to write. 760 */ 761 762 if (u->state == UPLOAD_WRITE_LOCAL) { 763 assert(u->buf != NULL); 764 assert(*fileoutfd != -1); 765 assert(*fileinfd == -1); 766 767 /* 768 * Unfortunately, we need to chunk these: if we're 769 * the server side of things, then we're multiplexing 770 * output and need to wrap this in chunks. 771 * This is a major deficiency of rsync. 772 * FIXME: add a "fast-path" mode that simply dumps out 773 * the buffer non-blocking if we're not mplexing. 774 */ 775 776 if (u->bufpos < u->bufsz) { 777 sz = MAX_CHUNK < (u->bufsz - u->bufpos) ? 778 MAX_CHUNK : (u->bufsz - u->bufpos); 779 c = io_write_buf(sess, u->fdout, 780 u->buf + u->bufpos, sz); 781 if (c == 0) { 782 ERRX1("io_write_nonblocking"); 783 return -1; 784 } 785 u->bufpos += sz; 786 if (u->bufpos < u->bufsz) 787 return 1; 788 } 789 790 /* 791 * Let the UPLOAD_FIND_NEXT state handle things if we 792 * finish, as we'll need to write a POLLOUT message and 793 * not have a writable descriptor yet. 794 */ 795 796 u->state = UPLOAD_FIND_NEXT; 797 u->idx++; 798 return 1; 799 } 800 801 /* 802 * If we invoke the uploader without a file currently open, then 803 * we iterate through til the next available regular file and 804 * start the opening process. 805 * This means we must have the output file descriptor working. 806 */ 807 808 if (u->state == UPLOAD_FIND_NEXT) { 809 assert(*fileinfd == -1); 810 assert(*fileoutfd != -1); 811 812 for ( ; u->idx < u->flsz; u->idx++) { 813 if (S_ISDIR(u->fl[u->idx].st.mode)) 814 c = pre_dir(u, sess); 815 else if (S_ISLNK(u->fl[u->idx].st.mode)) 816 c = pre_link(u, sess); 817 else if (S_ISREG(u->fl[u->idx].st.mode)) 818 c = pre_file(u, fileinfd, sess); 819 else if (S_ISBLK(u->fl[u->idx].st.mode) || 820 S_ISCHR(u->fl[u->idx].st.mode)) 821 c = pre_dev(u, sess); 822 else if (S_ISFIFO(u->fl[u->idx].st.mode)) 823 c = pre_fifo(u, sess); 824 else if (S_ISSOCK(u->fl[u->idx].st.mode)) 825 c = pre_sock(u, sess); 826 else 827 c = 0; 828 829 if (c < 0) 830 return -1; 831 else if (c > 0) 832 break; 833 } 834 835 /* 836 * Whether we've finished writing files or not, we 837 * disable polling on the output channel. 838 */ 839 840 *fileoutfd = -1; 841 if (u->idx == u->flsz) { 842 assert(*fileinfd == -1); 843 if (!io_write_int(sess, u->fdout, -1)) { 844 ERRX1("io_write_int"); 845 return -1; 846 } 847 u->state = UPLOAD_FINISHED; 848 LOG4("uploader: finished"); 849 return 0; 850 } 851 852 /* Go back to the event loop, if necessary. */ 853 854 u->state = (*fileinfd == -1) ? 855 UPLOAD_WRITE_LOCAL : UPLOAD_READ_LOCAL; 856 if (u->state == UPLOAD_READ_LOCAL) 857 return 1; 858 } 859 860 /* 861 * If an input file is open, stat it and see if it's already up 862 * to date, in which case close it and go to the next one. 863 * Either way, we don't have a write channel open. 864 */ 865 866 if (u->state == UPLOAD_READ_LOCAL) { 867 assert(*fileinfd != -1); 868 assert(*fileoutfd == -1); 869 f = &u->fl[u->idx]; 870 871 if (fstat(*fileinfd, &st) == -1) { 872 ERR("%s: fstat", f->path); 873 close(*fileinfd); 874 *fileinfd = -1; 875 return -1; 876 } else if (!S_ISREG(st.st_mode)) { 877 ERRX("%s: not regular", f->path); 878 close(*fileinfd); 879 *fileinfd = -1; 880 return -1; 881 } 882 883 if (st.st_size == f->st.size && 884 st.st_mtime == f->st.mtime) { 885 LOG3("%s: skipping: up to date", f->path); 886 if (!rsync_set_metadata 887 (sess, 0, *fileinfd, f, f->path)) { 888 ERRX1("rsync_set_metadata"); 889 close(*fileinfd); 890 *fileinfd = -1; 891 return -1; 892 } 893 close(*fileinfd); 894 *fileinfd = -1; 895 *fileoutfd = u->fdout; 896 u->state = UPLOAD_FIND_NEXT; 897 u->idx++; 898 return 1; 899 } 900 901 /* Fallthrough... */ 902 903 u->state = UPLOAD_WRITE_LOCAL; 904 } 905 906 /* Initialies our blocks. */ 907 908 assert(u->state == UPLOAD_WRITE_LOCAL); 909 memset(&blk, 0, sizeof(struct blkset)); 910 blk.csum = u->csumlen; 911 912 if (*fileinfd != -1 && st.st_size > 0) { 913 init_blkset(&blk, st.st_size); 914 assert(blk.blksz); 915 916 blk.blks = calloc(blk.blksz, sizeof(struct blk)); 917 if (blk.blks == NULL) { 918 ERR("calloc"); 919 close(*fileinfd); 920 *fileinfd = -1; 921 return -1; 922 } 923 924 if ((mbuf = calloc(1, blk.len)) == NULL) { 925 ERR("calloc"); 926 close(*fileinfd); 927 *fileinfd = -1; 928 return -1; 929 } 930 931 offs = 0; 932 i = 0; 933 do { 934 msz = pread(*fileinfd, mbuf, blk.len, offs); 935 if (msz < 0) { 936 ERR("pread"); 937 close(*fileinfd); 938 *fileinfd = -1; 939 return -1; 940 } 941 if ((size_t)msz != blk.len && (size_t)msz != blk.rem) { 942 /* short read, try again */ 943 continue; 944 } 945 init_blk(&blk.blks[i], &blk, offs, i, mbuf, sess); 946 offs += blk.len; 947 LOG3( 948 "i=%ld, offs=%lld, msz=%ld, blk.len=%lu, blk.rem=%lu", 949 i, offs, msz, blk.len, blk.rem); 950 i++; 951 } while (i < blk.blksz); 952 953 close(*fileinfd); 954 *fileinfd = -1; 955 LOG3("%s: mapped %jd B with %zu blocks", 956 u->fl[u->idx].path, (intmax_t)blk.size, 957 blk.blksz); 958 } else { 959 if (*fileinfd != -1) { 960 close(*fileinfd); 961 *fileinfd = -1; 962 } 963 blk.len = MAX_CHUNK; /* Doesn't matter. */ 964 LOG3("%s: not mapped", u->fl[u->idx].path); 965 } 966 967 assert(*fileinfd == -1); 968 969 /* Make sure the block metadata buffer is big enough. */ 970 971 u->bufsz = 972 sizeof(int32_t) + /* identifier */ 973 sizeof(int32_t) + /* block count */ 974 sizeof(int32_t) + /* block length */ 975 sizeof(int32_t) + /* checksum length */ 976 sizeof(int32_t) + /* block remainder */ 977 blk.blksz * 978 (sizeof(int32_t) + /* short checksum */ 979 blk.csum); /* long checksum */ 980 981 if (u->bufsz > u->bufmax) { 982 if ((bufp = realloc(u->buf, u->bufsz)) == NULL) { 983 ERR("realloc"); 984 return -1; 985 } 986 u->buf = bufp; 987 u->bufmax = u->bufsz; 988 } 989 990 u->bufpos = pos = 0; 991 io_buffer_int(u->buf, &pos, u->bufsz, u->idx); 992 io_buffer_int(u->buf, &pos, u->bufsz, blk.blksz); 993 io_buffer_int(u->buf, &pos, u->bufsz, blk.len); 994 io_buffer_int(u->buf, &pos, u->bufsz, blk.csum); 995 io_buffer_int(u->buf, &pos, u->bufsz, blk.rem); 996 for (i = 0; i < blk.blksz; i++) { 997 io_buffer_int(u->buf, &pos, u->bufsz, 998 blk.blks[i].chksum_short); 999 io_buffer_buf(u->buf, &pos, u->bufsz, 1000 blk.blks[i].chksum_long, blk.csum); 1001 } 1002 assert(pos == u->bufsz); 1003 1004 /* Reenable the output poller and clean up. */ 1005 1006 *fileoutfd = u->fdout; 1007 free(blk.blks); 1008 return 1; 1009 } 1010 1011 /* 1012 * Fix up the directory permissions and times post-order. 1013 * We can't fix up directory permissions in place because the server may 1014 * want us to have overly-tight permissions---say, those that don't 1015 * allow writing into the directory. 1016 * We also need to do our directory times post-order because making 1017 * files within the directory will change modification times. 1018 * Returns zero on failure, non-zero on success. 1019 */ 1020 int 1021 rsync_uploader_tail(struct upload *u, struct sess *sess) 1022 { 1023 size_t i; 1024 1025 1026 if (!sess->opts->preserve_times && 1027 !sess->opts->preserve_perms) 1028 return 1; 1029 1030 LOG2("fixing up directory times and permissions"); 1031 1032 for (i = 0; i < u->flsz; i++) 1033 if (S_ISDIR(u->fl[i].st.mode)) 1034 if (!post_dir(sess, u, i)) 1035 return 0; 1036 1037 return 1; 1038 } 1039