1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914
|
// Package sync is the implementation of sync/copy/move
package sync
import (
"context"
"fmt"
"path"
"sort"
"sync"
"github.com/ncw/rclone/fs"
"github.com/ncw/rclone/fs/accounting"
"github.com/ncw/rclone/fs/filter"
"github.com/ncw/rclone/fs/fserrors"
"github.com/ncw/rclone/fs/hash"
"github.com/ncw/rclone/fs/march"
"github.com/ncw/rclone/fs/operations"
"github.com/pkg/errors"
)
type syncCopyMove struct {
// parameters
fdst fs.Fs
fsrc fs.Fs
deleteMode fs.DeleteMode // how we are doing deletions
DoMove bool
deleteEmptySrcDirs bool
dir string
// internal state
ctx context.Context // internal context for controlling go-routines
cancel func() // cancel the context
deletersWg sync.WaitGroup // for delete before go routine
deleteFilesCh chan fs.Object // channel to receive deletes if delete before
trackRenames bool // set if we should do server side renames
dstFilesMu sync.Mutex // protect dstFiles
dstFiles map[string]fs.Object // dst files, always filled
srcFiles map[string]fs.Object // src files, only used if deleteBefore
srcFilesChan chan fs.Object // passes src objects
srcFilesResult chan error // error result of src listing
dstFilesResult chan error // error result of dst listing
dstEmptyDirsMu sync.Mutex // protect dstEmptyDirs
dstEmptyDirs map[string]fs.DirEntry // potentially empty directories
srcEmptyDirsMu sync.Mutex // protect srcEmptyDirs
srcEmptyDirs map[string]fs.DirEntry // potentially empty directories
checkerWg sync.WaitGroup // wait for checkers
toBeChecked *pipe // checkers channel
transfersWg sync.WaitGroup // wait for transfers
toBeUploaded *pipe // copiers channel
errorMu sync.Mutex // Mutex covering the errors variables
err error // normal error from copy process
noRetryErr error // error with NoRetry set
fatalErr error // fatal error
commonHash hash.Type // common hash type between src and dst
renameMapMu sync.Mutex // mutex to protect the below
renameMap map[string][]fs.Object // dst files by hash - only used by trackRenames
renamerWg sync.WaitGroup // wait for renamers
toBeRenamed *pipe // renamers channel
trackRenamesWg sync.WaitGroup // wg for background track renames
trackRenamesCh chan fs.Object // objects are pumped in here
renameCheck []fs.Object // accumulate files to check for rename here
backupDir fs.Fs // place to store overwrites/deletes
suffix string // suffix to add to files placed in backupDir
}
func newSyncCopyMove(fdst, fsrc fs.Fs, deleteMode fs.DeleteMode, DoMove bool, deleteEmptySrcDirs bool) (*syncCopyMove, error) {
s := &syncCopyMove{
fdst: fdst,
fsrc: fsrc,
deleteMode: deleteMode,
DoMove: DoMove,
deleteEmptySrcDirs: deleteEmptySrcDirs,
dir: "",
srcFilesChan: make(chan fs.Object, fs.Config.Checkers+fs.Config.Transfers),
srcFilesResult: make(chan error, 1),
dstFilesResult: make(chan error, 1),
dstEmptyDirs: make(map[string]fs.DirEntry),
srcEmptyDirs: make(map[string]fs.DirEntry),
toBeChecked: newPipe(accounting.Stats.SetCheckQueue, fs.Config.MaxBacklog),
toBeUploaded: newPipe(accounting.Stats.SetTransferQueue, fs.Config.MaxBacklog),
deleteFilesCh: make(chan fs.Object, fs.Config.Checkers),
trackRenames: fs.Config.TrackRenames,
commonHash: fsrc.Hashes().Overlap(fdst.Hashes()).GetOne(),
toBeRenamed: newPipe(accounting.Stats.SetRenameQueue, fs.Config.MaxBacklog),
trackRenamesCh: make(chan fs.Object, fs.Config.Checkers),
}
s.ctx, s.cancel = context.WithCancel(context.Background())
if s.trackRenames {
// Don't track renames for remotes without server-side move support.
if !operations.CanServerSideMove(fdst) {
fs.Errorf(fdst, "Ignoring --track-renames as the destination does not support server-side move or copy")
s.trackRenames = false
}
if s.commonHash == hash.None {
fs.Errorf(fdst, "Ignoring --track-renames as the source and destination do not have a common hash")
s.trackRenames = false
}
if s.deleteMode == fs.DeleteModeOff {
fs.Errorf(fdst, "Ignoring --track-renames as it doesn't work with copy or move, only sync")
s.trackRenames = false
}
}
if s.trackRenames {
// track renames needs delete after
if s.deleteMode != fs.DeleteModeOff {
s.deleteMode = fs.DeleteModeAfter
}
}
// Make Fs for --backup-dir if required
if fs.Config.BackupDir != "" {
var err error
s.backupDir, err = fs.NewFs(fs.Config.BackupDir)
if err != nil {
return nil, fserrors.FatalError(errors.Errorf("Failed to make fs for --backup-dir %q: %v", fs.Config.BackupDir, err))
}
if !operations.CanServerSideMove(s.backupDir) {
return nil, fserrors.FatalError(errors.New("can't use --backup-dir on a remote which doesn't support server side move or copy"))
}
if !operations.SameConfig(fdst, s.backupDir) {
return nil, fserrors.FatalError(errors.New("parameter to --backup-dir has to be on the same remote as destination"))
}
if operations.Overlapping(fdst, s.backupDir) {
return nil, fserrors.FatalError(errors.New("destination and parameter to --backup-dir mustn't overlap"))
}
if operations.Overlapping(fsrc, s.backupDir) {
return nil, fserrors.FatalError(errors.New("source and parameter to --backup-dir mustn't overlap"))
}
s.suffix = fs.Config.Suffix
}
return s, nil
}
// Check to see if the context has been cancelled
func (s *syncCopyMove) aborting() bool {
return s.ctx.Err() != nil
}
// This reads the map and pumps it into the channel passed in, closing
// the channel at the end
func (s *syncCopyMove) pumpMapToChan(files map[string]fs.Object, out chan<- fs.Object) {
outer:
for _, o := range files {
if s.aborting() {
break outer
}
select {
case out <- o:
case <-s.ctx.Done():
break outer
}
}
close(out)
s.srcFilesResult <- nil
}
// This checks the types of errors returned while copying files
func (s *syncCopyMove) processError(err error) {
if err == nil {
return
}
s.errorMu.Lock()
defer s.errorMu.Unlock()
switch {
case fserrors.IsFatalError(err):
if !s.aborting() {
fs.Errorf(nil, "Cancelling sync due to fatal error: %v", err)
s.cancel()
}
s.fatalErr = err
case fserrors.IsNoRetryError(err):
s.noRetryErr = err
default:
s.err = err
}
}
// Returns the current error (if any) in the order of prececedence
// fatalErr
// normal error
// noRetryErr
func (s *syncCopyMove) currentError() error {
s.errorMu.Lock()
defer s.errorMu.Unlock()
if s.fatalErr != nil {
return s.fatalErr
}
if s.err != nil {
return s.err
}
return s.noRetryErr
}
// pairChecker reads Objects~s on in send to out if they need transferring.
//
// FIXME potentially doing lots of hashes at once
func (s *syncCopyMove) pairChecker(in *pipe, out *pipe, wg *sync.WaitGroup) {
defer wg.Done()
for {
pair, ok := in.Get(s.ctx)
if !ok {
return
}
src := pair.Src
accounting.Stats.Checking(src.Remote())
// Check to see if can store this
if src.Storable() {
if operations.NeedTransfer(pair.Dst, pair.Src) {
// If files are treated as immutable, fail if destination exists and does not match
if fs.Config.Immutable && pair.Dst != nil {
fs.Errorf(pair.Dst, "Source and destination exist but do not match: immutable file modified")
s.processError(fs.ErrorImmutableModified)
} else {
// If destination already exists, then we must move it into --backup-dir if required
if pair.Dst != nil && s.backupDir != nil {
remoteWithSuffix := pair.Dst.Remote() + s.suffix
overwritten, _ := s.backupDir.NewObject(remoteWithSuffix)
_, err := operations.Move(s.backupDir, overwritten, remoteWithSuffix, pair.Dst)
if err != nil {
s.processError(err)
} else {
// If successful zero out the dst as it is no longer there and copy the file
pair.Dst = nil
ok = out.Put(s.ctx, pair)
if !ok {
return
}
}
} else {
ok = out.Put(s.ctx, pair)
if !ok {
return
}
}
}
} else {
// If moving need to delete the files we don't need to copy
if s.DoMove {
// Delete src if no error on copy
s.processError(operations.DeleteFile(src))
}
}
}
accounting.Stats.DoneChecking(src.Remote())
}
}
// pairRenamer reads Objects~s on in and attempts to rename them,
// otherwise it sends them out if they need transferring.
func (s *syncCopyMove) pairRenamer(in *pipe, out *pipe, wg *sync.WaitGroup) {
defer wg.Done()
for {
pair, ok := in.Get(s.ctx)
if !ok {
return
}
src := pair.Src
if !s.tryRename(src) {
// pass on if not renamed
ok = out.Put(s.ctx, pair)
if !ok {
return
}
}
}
}
// pairCopyOrMove reads Objects on in and moves or copies them.
func (s *syncCopyMove) pairCopyOrMove(in *pipe, fdst fs.Fs, wg *sync.WaitGroup) {
defer wg.Done()
var err error
for {
pair, ok := in.Get(s.ctx)
if !ok {
return
}
src := pair.Src
accounting.Stats.Transferring(src.Remote())
if s.DoMove {
_, err = operations.Move(fdst, pair.Dst, src.Remote(), src)
} else {
_, err = operations.Copy(fdst, pair.Dst, src.Remote(), src)
}
s.processError(err)
accounting.Stats.DoneTransferring(src.Remote(), err == nil)
}
}
// This starts the background checkers.
func (s *syncCopyMove) startCheckers() {
s.checkerWg.Add(fs.Config.Checkers)
for i := 0; i < fs.Config.Checkers; i++ {
go s.pairChecker(s.toBeChecked, s.toBeUploaded, &s.checkerWg)
}
}
// This stops the background checkers
func (s *syncCopyMove) stopCheckers() {
s.toBeChecked.Close()
fs.Infof(s.fdst, "Waiting for checks to finish")
s.checkerWg.Wait()
}
// This starts the background transfers
func (s *syncCopyMove) startTransfers() {
s.transfersWg.Add(fs.Config.Transfers)
for i := 0; i < fs.Config.Transfers; i++ {
go s.pairCopyOrMove(s.toBeUploaded, s.fdst, &s.transfersWg)
}
}
// This stops the background transfers
func (s *syncCopyMove) stopTransfers() {
s.toBeUploaded.Close()
fs.Infof(s.fdst, "Waiting for transfers to finish")
s.transfersWg.Wait()
}
// This starts the background renamers.
func (s *syncCopyMove) startRenamers() {
if !s.trackRenames {
return
}
s.renamerWg.Add(fs.Config.Checkers)
for i := 0; i < fs.Config.Checkers; i++ {
go s.pairRenamer(s.toBeRenamed, s.toBeUploaded, &s.renamerWg)
}
}
// This stops the background renamers
func (s *syncCopyMove) stopRenamers() {
if !s.trackRenames {
return
}
s.toBeRenamed.Close()
fs.Infof(s.fdst, "Waiting for renames to finish")
s.renamerWg.Wait()
}
// This starts the collection of possible renames
func (s *syncCopyMove) startTrackRenames() {
if !s.trackRenames {
return
}
s.trackRenamesWg.Add(1)
go func() {
defer s.trackRenamesWg.Done()
for o := range s.trackRenamesCh {
s.renameCheck = append(s.renameCheck, o)
}
}()
}
// This stops the background rename collection
func (s *syncCopyMove) stopTrackRenames() {
if !s.trackRenames {
return
}
close(s.trackRenamesCh)
s.trackRenamesWg.Wait()
}
// This starts the background deletion of files for --delete-during
func (s *syncCopyMove) startDeleters() {
if s.deleteMode != fs.DeleteModeDuring && s.deleteMode != fs.DeleteModeOnly {
return
}
s.deletersWg.Add(1)
go func() {
defer s.deletersWg.Done()
err := operations.DeleteFilesWithBackupDir(s.deleteFilesCh, s.backupDir)
s.processError(err)
}()
}
// This stops the background deleters
func (s *syncCopyMove) stopDeleters() {
if s.deleteMode != fs.DeleteModeDuring && s.deleteMode != fs.DeleteModeOnly {
return
}
close(s.deleteFilesCh)
s.deletersWg.Wait()
}
// This deletes the files in the dstFiles map. If checkSrcMap is set
// then it checks to see if they exist first in srcFiles the source
// file map, otherwise it unconditionally deletes them. If
// checkSrcMap is clear then it assumes that the any source files that
// have been found have been removed from dstFiles already.
func (s *syncCopyMove) deleteFiles(checkSrcMap bool) error {
if accounting.Stats.Errored() && !fs.Config.IgnoreErrors {
fs.Errorf(s.fdst, "%v", fs.ErrorNotDeleting)
return fs.ErrorNotDeleting
}
// Delete the spare files
toDelete := make(fs.ObjectsChan, fs.Config.Transfers)
go func() {
outer:
for remote, o := range s.dstFiles {
if checkSrcMap {
_, exists := s.srcFiles[remote]
if exists {
continue
}
}
if s.aborting() {
break
}
select {
case <-s.ctx.Done():
break outer
case toDelete <- o:
}
}
close(toDelete)
}()
return operations.DeleteFilesWithBackupDir(toDelete, s.backupDir)
}
// This deletes the empty directories in the slice passed in. It
// ignores any errors deleting directories
func deleteEmptyDirectories(f fs.Fs, entriesMap map[string]fs.DirEntry) error {
if len(entriesMap) == 0 {
return nil
}
if accounting.Stats.Errored() && !fs.Config.IgnoreErrors {
fs.Errorf(f, "%v", fs.ErrorNotDeletingDirs)
return fs.ErrorNotDeletingDirs
}
var entries fs.DirEntries
for _, entry := range entriesMap {
entries = append(entries, entry)
}
// Now delete the empty directories starting from the longest path
sort.Sort(entries)
var errorCount int
var okCount int
for i := len(entries) - 1; i >= 0; i-- {
entry := entries[i]
dir, ok := entry.(fs.Directory)
if ok {
// TryRmdir only deletes empty directories
err := operations.TryRmdir(f, dir.Remote())
if err != nil {
fs.Debugf(fs.LogDirName(f, dir.Remote()), "Failed to Rmdir: %v", err)
errorCount++
} else {
okCount++
}
} else {
fs.Errorf(f, "Not a directory: %v", entry)
}
}
if errorCount > 0 {
fs.Debugf(f, "failed to delete %d directories", errorCount)
}
if okCount > 0 {
fs.Debugf(f, "deleted %d directories", okCount)
}
return nil
}
// This copies the empty directories in the slice passed in and logs
// any errors copying the directories
func copyEmptyDirectories(f fs.Fs, entries map[string]fs.DirEntry) error {
if len(entries) == 0 {
return nil
}
var okCount int
for _, entry := range entries {
dir, ok := entry.(fs.Directory)
if ok {
err := operations.Mkdir(f, dir.Remote())
if err != nil {
fs.Errorf(fs.LogDirName(f, dir.Remote()), "Failed to Mkdir: %v", err)
} else {
okCount++
}
} else {
fs.Errorf(f, "Not a directory: %v", entry)
}
}
if accounting.Stats.Errored() {
fs.Debugf(f, "failed to copy %d directories", accounting.Stats.GetErrors())
}
if okCount > 0 {
fs.Debugf(f, "copied %d directories", okCount)
}
return nil
}
func (s *syncCopyMove) srcParentDirCheck(entry fs.DirEntry) {
// If we are moving files then we don't want to remove directories with files in them
// from the srcEmptyDirs as we are about to move them making the directory empty.
if s.DoMove {
return
}
parentDir := path.Dir(entry.Remote())
if parentDir == "." {
parentDir = ""
}
if _, ok := s.srcEmptyDirs[parentDir]; ok {
delete(s.srcEmptyDirs, parentDir)
}
}
// renameHash makes a string with the size and the hash for rename detection
//
// it may return an empty string in which case no hash could be made
func (s *syncCopyMove) renameHash(obj fs.Object) (hash string) {
var err error
hash, err = obj.Hash(s.commonHash)
if err != nil {
fs.Debugf(obj, "Hash failed: %v", err)
return ""
}
if hash == "" {
return ""
}
return fmt.Sprintf("%d,%s", obj.Size(), hash)
}
// pushRenameMap adds the object with hash to the rename map
func (s *syncCopyMove) pushRenameMap(hash string, obj fs.Object) {
s.renameMapMu.Lock()
s.renameMap[hash] = append(s.renameMap[hash], obj)
s.renameMapMu.Unlock()
}
// popRenameMap finds the object with hash and pop the first match from
// renameMap or returns nil if not found.
func (s *syncCopyMove) popRenameMap(hash string) (dst fs.Object) {
s.renameMapMu.Lock()
dsts, ok := s.renameMap[hash]
if ok && len(dsts) > 0 {
dst, dsts = dsts[0], dsts[1:]
if len(dsts) > 0 {
s.renameMap[hash] = dsts
} else {
delete(s.renameMap, hash)
}
}
s.renameMapMu.Unlock()
return dst
}
// makeRenameMap builds a map of the destination files by hash that
// match sizes in the slice of objects in s.renameCheck
func (s *syncCopyMove) makeRenameMap() {
fs.Infof(s.fdst, "Making map for --track-renames")
// first make a map of possible sizes we need to check
possibleSizes := map[int64]struct{}{}
for _, obj := range s.renameCheck {
possibleSizes[obj.Size()] = struct{}{}
}
// pump all the dstFiles into in
in := make(chan fs.Object, fs.Config.Checkers)
go s.pumpMapToChan(s.dstFiles, in)
// now make a map of size,hash for all dstFiles
s.renameMap = make(map[string][]fs.Object)
var wg sync.WaitGroup
wg.Add(fs.Config.Transfers)
for i := 0; i < fs.Config.Transfers; i++ {
go func() {
defer wg.Done()
for obj := range in {
// only create hash for dst fs.Object if its size could match
if _, found := possibleSizes[obj.Size()]; found {
accounting.Stats.Checking(obj.Remote())
hash := s.renameHash(obj)
if hash != "" {
s.pushRenameMap(hash, obj)
}
accounting.Stats.DoneChecking(obj.Remote())
}
}
}()
}
wg.Wait()
fs.Infof(s.fdst, "Finished making map for --track-renames")
}
// tryRename renames a src object when doing track renames if
// possible, it returns true if the object was renamed.
func (s *syncCopyMove) tryRename(src fs.Object) bool {
accounting.Stats.Checking(src.Remote())
defer accounting.Stats.DoneChecking(src.Remote())
// Calculate the hash of the src object
hash := s.renameHash(src)
if hash == "" {
return false
}
// Get a match on fdst
dst := s.popRenameMap(hash)
if dst == nil {
return false
}
// Find dst object we are about to overwrite if it exists
dstOverwritten, _ := s.fdst.NewObject(src.Remote())
// Rename dst to have name src.Remote()
_, err := operations.Move(s.fdst, dstOverwritten, src.Remote(), dst)
if err != nil {
fs.Debugf(src, "Failed to rename to %q: %v", dst.Remote(), err)
return false
}
// remove file from dstFiles if present
s.dstFilesMu.Lock()
delete(s.dstFiles, dst.Remote())
s.dstFilesMu.Unlock()
fs.Infof(src, "Renamed from %q", dst.Remote())
return true
}
// Syncs fsrc into fdst
//
// If Delete is true then it deletes any files in fdst that aren't in fsrc
//
// If DoMove is true then files will be moved instead of copied
//
// dir is the start directory, "" for root
func (s *syncCopyMove) run() error {
if operations.Same(s.fdst, s.fsrc) {
fs.Errorf(s.fdst, "Nothing to do as source and destination are the same")
return nil
}
// Start background checking and transferring pipeline
s.startCheckers()
s.startRenamers()
s.startTransfers()
s.startDeleters()
s.dstFiles = make(map[string]fs.Object)
s.startTrackRenames()
// set up a march over fdst and fsrc
m := march.New(s.ctx, s.fdst, s.fsrc, s.dir, s)
m.Run()
s.stopTrackRenames()
if s.trackRenames {
// Build the map of the remaining dstFiles by hash
s.makeRenameMap()
// Attempt renames for all the files which don't have a matching dst
for _, src := range s.renameCheck {
ok := s.toBeRenamed.Put(s.ctx, fs.ObjectPair{Src: src, Dst: nil})
if !ok {
break
}
}
}
// Stop background checking and transferring pipeline
s.stopCheckers()
s.stopRenamers()
s.stopTransfers()
s.stopDeleters()
s.processError(copyEmptyDirectories(s.fdst, s.srcEmptyDirs))
// Delete files after
if s.deleteMode == fs.DeleteModeAfter {
if s.currentError() != nil && !fs.Config.IgnoreErrors {
fs.Errorf(s.fdst, "%v", fs.ErrorNotDeleting)
} else {
s.processError(s.deleteFiles(false))
}
}
// Prune empty directories
if s.deleteMode != fs.DeleteModeOff {
if s.currentError() != nil && !fs.Config.IgnoreErrors {
fs.Errorf(s.fdst, "%v", fs.ErrorNotDeletingDirs)
} else {
s.processError(deleteEmptyDirectories(s.fdst, s.dstEmptyDirs))
}
}
// Delete empty fsrc subdirectories
// if DoMove and --delete-empty-src-dirs flag is set
if s.DoMove && s.deleteEmptySrcDirs {
//delete empty subdirectories that were part of the move
s.processError(deleteEmptyDirectories(s.fsrc, s.srcEmptyDirs))
}
// cancel the context to free resources
s.cancel()
return s.currentError()
}
// DstOnly have an object which is in the destination only
func (s *syncCopyMove) DstOnly(dst fs.DirEntry) (recurse bool) {
if s.deleteMode == fs.DeleteModeOff {
return false
}
switch x := dst.(type) {
case fs.Object:
switch s.deleteMode {
case fs.DeleteModeAfter:
// record object as needs deleting
s.dstFilesMu.Lock()
s.dstFiles[x.Remote()] = x
s.dstFilesMu.Unlock()
case fs.DeleteModeDuring, fs.DeleteModeOnly:
select {
case <-s.ctx.Done():
return
case s.deleteFilesCh <- x:
}
default:
panic(fmt.Sprintf("unexpected delete mode %d", s.deleteMode))
}
case fs.Directory:
// Do the same thing to the entire contents of the directory
// Record directory as it is potentially empty and needs deleting
if s.fdst.Features().CanHaveEmptyDirectories {
s.dstEmptyDirsMu.Lock()
s.dstEmptyDirs[dst.Remote()] = dst
s.dstEmptyDirsMu.Unlock()
}
return true
default:
panic("Bad object in DirEntries")
}
return false
}
// SrcOnly have an object which is in the source only
func (s *syncCopyMove) SrcOnly(src fs.DirEntry) (recurse bool) {
if s.deleteMode == fs.DeleteModeOnly {
return false
}
switch x := src.(type) {
case fs.Object:
// If it's a copy operation,
// remove parent directory from srcEmptyDirs
// since it's not really empty
s.srcEmptyDirsMu.Lock()
s.srcParentDirCheck(src)
s.srcEmptyDirsMu.Unlock()
if s.trackRenames {
// Save object to check for a rename later
select {
case <-s.ctx.Done():
return
case s.trackRenamesCh <- x:
}
} else {
// No need to check since doesn't exist
ok := s.toBeUploaded.Put(s.ctx, fs.ObjectPair{Src: x, Dst: nil})
if !ok {
return
}
}
case fs.Directory:
// Do the same thing to the entire contents of the directory
// Record the directory for deletion
s.srcEmptyDirsMu.Lock()
s.srcParentDirCheck(src)
s.srcEmptyDirs[src.Remote()] = src
s.srcEmptyDirsMu.Unlock()
return true
default:
panic("Bad object in DirEntries")
}
return false
}
// Match is called when src and dst are present, so sync src to dst
func (s *syncCopyMove) Match(dst, src fs.DirEntry) (recurse bool) {
switch srcX := src.(type) {
case fs.Object:
s.srcEmptyDirsMu.Lock()
s.srcParentDirCheck(src)
s.srcEmptyDirsMu.Unlock()
if s.deleteMode == fs.DeleteModeOnly {
return false
}
dstX, ok := dst.(fs.Object)
if ok {
ok = s.toBeChecked.Put(s.ctx, fs.ObjectPair{Src: srcX, Dst: dstX})
if !ok {
return false
}
} else {
// FIXME src is file, dst is directory
err := errors.New("can't overwrite directory with file")
fs.Errorf(dst, "%v", err)
s.processError(err)
}
case fs.Directory:
// Do the same thing to the entire contents of the directory
_, ok := dst.(fs.Directory)
if ok {
// Record the src directory for deletion
s.srcEmptyDirsMu.Lock()
s.srcParentDirCheck(src)
s.srcEmptyDirs[src.Remote()] = src
s.srcEmptyDirsMu.Unlock()
return true
}
// FIXME src is dir, dst is file
err := errors.New("can't overwrite file with directory")
fs.Errorf(dst, "%v", err)
s.processError(err)
default:
panic("Bad object in DirEntries")
}
return false
}
// Syncs fsrc into fdst
//
// If Delete is true then it deletes any files in fdst that aren't in fsrc
//
// If DoMove is true then files will be moved instead of copied
//
// dir is the start directory, "" for root
func runSyncCopyMove(fdst, fsrc fs.Fs, deleteMode fs.DeleteMode, DoMove bool, deleteEmptySrcDirs bool) error {
if deleteMode != fs.DeleteModeOff && DoMove {
return fserrors.FatalError(errors.New("can't delete and move at the same time"))
}
// Run an extra pass to delete only
if deleteMode == fs.DeleteModeBefore {
if fs.Config.TrackRenames {
return fserrors.FatalError(errors.New("can't use --delete-before with --track-renames"))
}
// only delete stuff during in this pass
do, err := newSyncCopyMove(fdst, fsrc, fs.DeleteModeOnly, false, deleteEmptySrcDirs)
if err != nil {
return err
}
err = do.run()
if err != nil {
return err
}
// Next pass does a copy only
deleteMode = fs.DeleteModeOff
}
do, err := newSyncCopyMove(fdst, fsrc, deleteMode, DoMove, deleteEmptySrcDirs)
if err != nil {
return err
}
return do.run()
}
// Sync fsrc into fdst
func Sync(fdst, fsrc fs.Fs) error {
return runSyncCopyMove(fdst, fsrc, fs.Config.DeleteMode, false, false)
}
// CopyDir copies fsrc into fdst
func CopyDir(fdst, fsrc fs.Fs) error {
return runSyncCopyMove(fdst, fsrc, fs.DeleteModeOff, false, false)
}
// moveDir moves fsrc into fdst
func moveDir(fdst, fsrc fs.Fs, deleteEmptySrcDirs bool) error {
return runSyncCopyMove(fdst, fsrc, fs.DeleteModeOff, true, deleteEmptySrcDirs)
}
// MoveDir moves fsrc into fdst
func MoveDir(fdst, fsrc fs.Fs, deleteEmptySrcDirs bool) error {
if operations.Same(fdst, fsrc) {
fs.Errorf(fdst, "Nothing to do as source and destination are the same")
return nil
}
// First attempt to use DirMover if exists, same Fs and no filters are active
if fdstDirMove := fdst.Features().DirMove; fdstDirMove != nil && operations.SameConfig(fsrc, fdst) && filter.Active.InActive() {
if fs.Config.DryRun {
fs.Logf(fdst, "Not doing server side directory move as --dry-run")
return nil
}
fs.Debugf(fdst, "Using server side directory move")
err := fdstDirMove(fsrc, "", "")
switch err {
case fs.ErrorCantDirMove, fs.ErrorDirExists:
fs.Infof(fdst, "Server side directory move failed - fallback to file moves: %v", err)
case nil:
fs.Infof(fdst, "Server side directory move succeeded")
return nil
default:
fs.CountError(err)
fs.Errorf(fdst, "Server side directory move failed: %v", err)
return err
}
}
// The two remotes mustn't overlap if we didn't do server side move
if operations.Overlapping(fdst, fsrc) {
err := fs.ErrorCantMoveOverlapping
fs.Errorf(fdst, "%v", err)
return err
}
// Otherwise move the files one by one
return moveDir(fdst, fsrc, deleteEmptySrcDirs)
}
|