rclone/fs/sync/sync.go

1159 lines
35 KiB
Go
Raw Normal View History

// Package sync is the implementation of sync/copy/move
package sync
import (
"context"
"errors"
"fmt"
"path"
"sort"
"strings"
"sync"
"time"
"github.com/rclone/rclone/fs"
"github.com/rclone/rclone/fs/accounting"
"github.com/rclone/rclone/fs/filter"
"github.com/rclone/rclone/fs/fserrors"
"github.com/rclone/rclone/fs/hash"
"github.com/rclone/rclone/fs/march"
"github.com/rclone/rclone/fs/operations"
)
type syncCopyMove struct {
// parameters
fdst fs.Fs
fsrc fs.Fs
deleteMode fs.DeleteMode // how we are doing deletions
DoMove bool
copyEmptySrcDirs bool
deleteEmptySrcDirs bool
dir string
// internal state
ci *fs.ConfigInfo // global config
fi *filter.Filter // filter config
ctx context.Context // internal context for controlling go-routines
cancel func() // cancel the context
inCtx context.Context // internal context for controlling march
inCancel func() // cancel the march context
noTraverse bool // if set don't traverse the dst
noCheckDest bool // if set transfer all objects regardless without checking dst
noUnicodeNormalization bool // don't normalize unicode characters in filenames
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
trackRenamesStrategy trackRenamesStrategy // strategies used for tracking 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
modifyWindow time.Duration // modify window between fsrc, fdst
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
compareCopyDest []fs.Fs // place to check for files to server side copy
backupDir fs.Fs // place to store overwrites/deletes
checkFirst bool // if set run all the checkers before starting transfers
maxDurationEndTime time.Time // end time if --max-duration is set
}
type trackRenamesStrategy byte
const (
trackRenamesStrategyHash trackRenamesStrategy = 1 << iota
trackRenamesStrategyModtime
trackRenamesStrategyLeaf
)
func (strategy trackRenamesStrategy) hash() bool {
return (strategy & trackRenamesStrategyHash) != 0
}
func (strategy trackRenamesStrategy) modTime() bool {
return (strategy & trackRenamesStrategyModtime) != 0
}
func (strategy trackRenamesStrategy) leaf() bool {
return (strategy & trackRenamesStrategyLeaf) != 0
}
func newSyncCopyMove(ctx context.Context, fdst, fsrc fs.Fs, deleteMode fs.DeleteMode, DoMove bool, deleteEmptySrcDirs bool, copyEmptySrcDirs bool) (*syncCopyMove, error) {
if (deleteMode != fs.DeleteModeOff || DoMove) && operations.OverlappingFilterCheck(ctx, fdst, fsrc) {
return nil, fserrors.FatalError(fs.ErrorOverlapping)
}
ci := fs.GetConfig(ctx)
fi := filter.GetConfig(ctx)
s := &syncCopyMove{
ci: ci,
fi: fi,
fdst: fdst,
fsrc: fsrc,
deleteMode: deleteMode,
DoMove: DoMove,
copyEmptySrcDirs: copyEmptySrcDirs,
deleteEmptySrcDirs: deleteEmptySrcDirs,
dir: "",
srcFilesChan: make(chan fs.Object, ci.Checkers+ci.Transfers),
srcFilesResult: make(chan error, 1),
dstFilesResult: make(chan error, 1),
dstEmptyDirs: make(map[string]fs.DirEntry),
srcEmptyDirs: make(map[string]fs.DirEntry),
noTraverse: ci.NoTraverse,
noCheckDest: ci.NoCheckDest,
noUnicodeNormalization: ci.NoUnicodeNormalization,
deleteFilesCh: make(chan fs.Object, ci.Checkers),
trackRenames: ci.TrackRenames,
commonHash: fsrc.Hashes().Overlap(fdst.Hashes()).GetOne(),
modifyWindow: fs.GetModifyWindow(ctx, fsrc, fdst),
trackRenamesCh: make(chan fs.Object, ci.Checkers),
checkFirst: ci.CheckFirst,
}
backlog := ci.MaxBacklog
if s.checkFirst {
fs.Infof(s.fdst, "Running all checks before starting transfers")
backlog = -1
}
var err error
s.toBeChecked, err = newPipe(ci.OrderBy, accounting.Stats(ctx).SetCheckQueue, backlog)
if err != nil {
return nil, err
}
s.toBeUploaded, err = newPipe(ci.OrderBy, accounting.Stats(ctx).SetTransferQueue, backlog)
if err != nil {
return nil, err
}
s.toBeRenamed, err = newPipe(ci.OrderBy, accounting.Stats(ctx).SetRenameQueue, backlog)
if err != nil {
return nil, err
}
if ci.MaxDuration > 0 {
s.maxDurationEndTime = time.Now().Add(ci.MaxDuration)
fs.Infof(s.fdst, "Transfer session %v deadline: %s", ci.CutoffMode, s.maxDurationEndTime.Format("2006/01/02 15:04:05"))
}
// If a max session duration has been defined add a deadline
// to the main context if cutoff mode is hard. This will cut
// the transfers off.
if !s.maxDurationEndTime.IsZero() && ci.CutoffMode == fs.CutoffModeHard {
s.ctx, s.cancel = context.WithDeadline(ctx, s.maxDurationEndTime)
} else {
s.ctx, s.cancel = context.WithCancel(ctx)
}
// Input context - cancel this for graceful stop.
//
// If a max session duration has been defined add a deadline
// to the input context if cutoff mode is graceful or soft.
// This won't stop the transfers but will cut the
// list/check/transfer pipelines.
if !s.maxDurationEndTime.IsZero() && ci.CutoffMode != fs.CutoffModeHard {
s.inCtx, s.inCancel = context.WithDeadline(s.ctx, s.maxDurationEndTime)
} else {
s.inCtx, s.inCancel = context.WithCancel(s.ctx)
}
if s.noTraverse && s.deleteMode != fs.DeleteModeOff {
if !fi.HaveFilesFrom() {
fs.Errorf(nil, "Ignoring --no-traverse with sync")
}
s.noTraverse = false
}
s.trackRenamesStrategy, err = parseTrackRenamesStrategy(ci.TrackRenamesStrategy)
if err != nil {
return nil, err
}
if s.noCheckDest {
if s.deleteMode != fs.DeleteModeOff {
return nil, errors.New("can't use --no-check-dest with sync: use copy instead")
}
if ci.Immutable {
return nil, errors.New("can't use --no-check-dest with --immutable")
}
if s.backupDir != nil {
return nil, errors.New("can't use --no-check-dest with --backup-dir")
}
}
if s.trackRenames {
2017-01-11 04:03:55 +08:00
// 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.trackRenamesStrategy.hash() && 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.trackRenamesStrategy.modTime() && s.modifyWindow == fs.ModTimeNotSupported {
fs.Errorf(fdst, "Ignoring --track-renames as either the source or destination do not support modtime")
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
}
if s.noTraverse {
fs.Errorf(nil, "Ignoring --no-traverse with --track-renames")
s.noTraverse = false
}
}
// Make Fs for --backup-dir if required
if ci.BackupDir != "" || ci.Suffix != "" {
var err error
s.backupDir, err = operations.BackupDir(ctx, fdst, fsrc, "")
if err != nil {
return nil, err
}
}
if len(ci.CompareDest) > 0 {
var err error
s.compareCopyDest, err = operations.GetCompareDest(ctx)
if err != nil {
return nil, err
}
} else if len(ci.CopyDest) > 0 {
var err error
s.compareCopyDest, err = operations.GetCopyDest(ctx, fdst)
if err != nil {
return nil, err
}
}
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
}
if err == context.DeadlineExceeded {
err = fserrors.NoRetryError(err)
} else if err == accounting.ErrorMaxTransferLimitReachedGraceful {
if s.inCtx.Err() == nil {
fs.Logf(nil, "%v - stopping transfers", err)
// Cancel the march and stop the pipes
s.inCancel()
}
} else if err == context.Canceled && s.inCtx.Err() != nil {
// Ignore context Canceled if we have called s.inCancel()
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
}
}
2019-04-30 20:06:24 +08:00
// Returns the current error (if any) in the order of precedence
// 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
2020-03-14 05:12:22 +08:00
func (s *syncCopyMove) pairChecker(in *pipe, out *pipe, fraction int, wg *sync.WaitGroup) {
defer wg.Done()
for {
pair, ok := in.GetMax(s.inCtx, fraction)
if !ok {
return
}
src := pair.Src
var err error
tr := accounting.Stats(s.ctx).NewCheckingTransfer(src)
// Check to see if can store this
if src.Storable() {
NoNeedTransfer, err := operations.CompareOrCopyDest(s.ctx, s.fdst, pair.Dst, pair.Src, s.compareCopyDest, s.backupDir)
if err != nil {
s.processError(err)
}
if !NoNeedTransfer && operations.NeedTransfer(s.ctx, pair.Dst, pair.Src) {
// If files are treated as immutable, fail if destination exists and does not match
if s.ci.Immutable && pair.Dst != nil {
err := fs.CountError(fserrors.NoRetryError(fs.ErrorImmutableModified))
fs.Errorf(pair.Dst, "Source and destination exist but do not match: %v", err)
s.processError(err)
} else {
// If destination already exists, then we must move it into --backup-dir if required
if pair.Dst != nil && s.backupDir != nil {
err := operations.MoveBackupDir(s.ctx, s.backupDir, 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
if operations.SameObject(src, pair.Dst) {
fs.Logf(src, "Not removing source file as it is the same file as the destination")
} else if s.ci.IgnoreExisting {
fs.Debugf(src, "Not removing source file as destination file exists and --ignore-existing is set")
} else {
s.processError(operations.DeleteFile(s.ctx, src))
}
}
}
}
tr.Done(s.ctx, err)
}
}
// pairRenamer reads Objects~s on in and attempts to rename them,
// otherwise it sends them out if they need transferring.
2020-03-14 05:12:22 +08:00
func (s *syncCopyMove) pairRenamer(in *pipe, out *pipe, fraction int, wg *sync.WaitGroup) {
defer wg.Done()
for {
pair, ok := in.GetMax(s.inCtx, fraction)
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.
2020-03-14 05:12:22 +08:00
func (s *syncCopyMove) pairCopyOrMove(ctx context.Context, in *pipe, fdst fs.Fs, fraction int, wg *sync.WaitGroup) {
defer wg.Done()
var err error
for {
pair, ok := in.GetMax(s.inCtx, fraction)
if !ok {
return
}
src := pair.Src
if s.DoMove {
_, err = operations.Move(ctx, fdst, pair.Dst, src.Remote(), src)
} else {
_, err = operations.Copy(ctx, fdst, pair.Dst, src.Remote(), src)
}
s.processError(err)
}
}
// This starts the background checkers.
func (s *syncCopyMove) startCheckers() {
s.checkerWg.Add(s.ci.Checkers)
for i := 0; i < s.ci.Checkers; i++ {
fraction := (100 * i) / s.ci.Checkers
2020-03-14 05:12:22 +08:00
go s.pairChecker(s.toBeChecked, s.toBeUploaded, fraction, &s.checkerWg)
}
}
// This stops the background checkers
func (s *syncCopyMove) stopCheckers() {
s.toBeChecked.Close()
fs.Debugf(s.fdst, "Waiting for checks to finish")
s.checkerWg.Wait()
}
// This starts the background transfers
func (s *syncCopyMove) startTransfers() {
s.transfersWg.Add(s.ci.Transfers)
for i := 0; i < s.ci.Transfers; i++ {
fraction := (100 * i) / s.ci.Transfers
2020-03-14 05:12:22 +08:00
go s.pairCopyOrMove(s.ctx, s.toBeUploaded, s.fdst, fraction, &s.transfersWg)
}
}
// This stops the background transfers
func (s *syncCopyMove) stopTransfers() {
s.toBeUploaded.Close()
fs.Debugf(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(s.ci.Checkers)
for i := 0; i < s.ci.Checkers; i++ {
fraction := (100 * i) / s.ci.Checkers
2020-03-14 05:12:22 +08:00
go s.pairRenamer(s.toBeRenamed, s.toBeUploaded, fraction, &s.renamerWg)
}
}
// This stops the background renamers
func (s *syncCopyMove) stopRenamers() {
if !s.trackRenames {
return
}
s.toBeRenamed.Close()
fs.Debugf(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.ctx, 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(s.ctx).Errored() && !s.ci.IgnoreErrors {
fs.Errorf(s.fdst, "%v", fs.ErrorNotDeleting)
return fs.ErrorNotDeleting
}
// Delete the spare files
toDelete := make(fs.ObjectsChan, s.ci.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(s.ctx, toDelete, s.backupDir)
}
// This deletes the empty directories in the slice passed in. It
// ignores any errors deleting directories
func (s *syncCopyMove) deleteEmptyDirectories(ctx context.Context, f fs.Fs, entriesMap map[string]fs.DirEntry) error {
if len(entriesMap) == 0 {
return nil
}
if accounting.Stats(ctx).Errored() && !s.ci.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(ctx, 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(ctx context.Context, 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(ctx, 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(ctx).Errored() {
fs.Debugf(f, "failed to copy %d directories", accounting.Stats(ctx).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 = ""
}
delete(s.srcEmptyDirs, parentDir)
}
// parseTrackRenamesStrategy turns a config string into a trackRenamesStrategy
func parseTrackRenamesStrategy(strategies string) (strategy trackRenamesStrategy, err error) {
if len(strategies) == 0 {
return strategy, nil
}
for _, s := range strings.Split(strategies, ",") {
switch s {
case "hash":
strategy |= trackRenamesStrategyHash
case "modtime":
strategy |= trackRenamesStrategyModtime
case "leaf":
strategy |= trackRenamesStrategyLeaf
case "size":
// ignore
default:
return strategy, fmt.Errorf("unknown track renames strategy %q", s)
}
}
return strategy, nil
}
// renameID makes a string with the size and the other identifiers of the requested rename strategies
//
// it may return an empty string in which case no hash could be made
func (s *syncCopyMove) renameID(obj fs.Object, renamesStrategy trackRenamesStrategy, precision time.Duration) string {
var builder strings.Builder
fmt.Fprintf(&builder, "%d", obj.Size())
if renamesStrategy.hash() {
var err error
hash, err := obj.Hash(s.ctx, s.commonHash)
if err != nil {
fs.Debugf(obj, "Hash failed: %v", err)
return ""
}
if hash == "" {
return ""
}
builder.WriteRune(',')
builder.WriteString(hash)
}
// for renamesStrategy.modTime() we don't add to the hash but we check the times in
// popRenameMap
if renamesStrategy.leaf() {
builder.WriteRune(',')
builder.WriteString(path.Base(obj.Remote()))
}
return builder.String()
}
// 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, src fs.Object) (dst fs.Object) {
s.renameMapMu.Lock()
defer s.renameMapMu.Unlock()
dsts, ok := s.renameMap[hash]
if ok && len(dsts) > 0 {
// Element to remove
i := 0
// If using track renames strategy modtime then we need to check the modtimes here
if s.trackRenamesStrategy.modTime() {
i = -1
srcModTime := src.ModTime(s.ctx)
for j, dst := range dsts {
dstModTime := dst.ModTime(s.ctx)
dt := dstModTime.Sub(srcModTime)
if dt < s.modifyWindow && dt > -s.modifyWindow {
i = j
break
}
}
// If nothing matched then return nil
if i < 0 {
return nil
}
}
// Remove the entry and return it
dst = dsts[i]
dsts = append(dsts[:i], dsts[i+1:]...)
if len(dsts) > 0 {
s.renameMap[hash] = dsts
} else {
delete(s.renameMap, hash)
}
}
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, s.ci.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(s.ci.Transfers)
for i := 0; i < s.ci.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 {
tr := accounting.Stats(s.ctx).NewCheckingTransfer(obj)
hash := s.renameID(obj, s.trackRenamesStrategy, s.modifyWindow)
if hash != "" {
s.pushRenameMap(hash, obj)
}
tr.Done(s.ctx, nil)
}
}
}()
}
wg.Wait()
fs.Infof(s.fdst, "Finished making map for --track-renames")
}
// tryRename renames an src object when doing track renames if
// possible, it returns true if the object was renamed.
func (s *syncCopyMove) tryRename(src fs.Object) bool {
// Calculate the hash of the src object
hash := s.renameID(src, s.trackRenamesStrategy, fs.GetModifyWindow(s.ctx, s.fsrc, s.fdst))
if hash == "" {
return false
}
// Get a match on fdst
dst := s.popRenameMap(hash, src)
if dst == nil {
return false
}
// Find dst object we are about to overwrite if it exists
dstOverwritten, _ := s.fdst.NewObject(s.ctx, src.Remote())
// Rename dst to have name src.Remote()
_, err := operations.Move(s.ctx, 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
}
// errorMaxDurationReached defines error when transfer duration is reached
// Used for checking on exit and matching to correct exit code.
var errorMaxDurationReached = fserrors.FatalError(errors.New("max transfer duration reached as set by --max-duration"))
// 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()
if !s.checkFirst {
s.startTransfers()
}
s.startDeleters()
s.dstFiles = make(map[string]fs.Object)
s.startTrackRenames()
// set up a march over fdst and fsrc
m := &march.March{
Ctx: s.inCtx,
Fdst: s.fdst,
Fsrc: s.fsrc,
Dir: s.dir,
NoTraverse: s.noTraverse,
Callback: s,
DstIncludeAll: s.fi.Opt.DeleteExcluded,
NoCheckDest: s.noCheckDest,
NoUnicodeNormalization: s.noUnicodeNormalization,
}
s.processError(m.Run(s.ctx))
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()
if s.checkFirst {
fs.Infof(s.fdst, "Checks finished, now starting transfers")
s.startTransfers()
}
s.stopRenamers()
s.stopTransfers()
s.stopDeleters()
if s.copyEmptySrcDirs {
s.processError(copyEmptyDirectories(s.ctx, s.fdst, s.srcEmptyDirs))
}
// Delete files after
if s.deleteMode == fs.DeleteModeAfter {
if s.currentError() != nil && !s.ci.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 && !s.ci.IgnoreErrors {
fs.Errorf(s.fdst, "%v", fs.ErrorNotDeletingDirs)
} else {
s.processError(s.deleteEmptyDirectories(s.ctx, 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(s.deleteEmptyDirectories(s.ctx, s.fsrc, s.srcEmptyDirs))
}
// Read the error out of the contexts if there is one
s.processError(s.ctx.Err())
s.processError(s.inCtx.Err())
// If the duration was exceeded then add a Fatal Error so we don't retry
if !s.maxDurationEndTime.IsZero() && time.Since(s.maxDurationEndTime) > 0 {
fs.Errorf(s.fdst, "%v", errorMaxDurationReached)
s.processError(errorMaxDurationReached)
}
// Print nothing to transfer message if there were no transfers and no errors
if s.deleteMode != fs.DeleteModeOnly && accounting.Stats(s.ctx).GetTransfers() == 0 && s.currentError() == nil {
fs.Infof(nil, "There was nothing to transfer")
}
// cancel the contexts to free resources
s.inCancel()
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 {
// Check CompareDest && CopyDest
NoNeedTransfer, err := operations.CompareOrCopyDest(s.ctx, s.fdst, nil, x, s.compareCopyDest, s.backupDir)
if err != nil {
s.processError(err)
}
if !NoNeedTransfer {
// 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(ctx context.Context, 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 {
// Only record matched (src & dst) empty dirs when performing move
if s.DoMove {
// 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(ctx context.Context, fdst, fsrc fs.Fs, deleteMode fs.DeleteMode, DoMove bool, deleteEmptySrcDirs bool, copyEmptySrcDirs bool) error {
ci := fs.GetConfig(ctx)
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 ci.TrackRenames {
return fserrors.FatalError(errors.New("can't use --delete-before with --track-renames"))
}
// only delete stuff during in this pass
do, err := newSyncCopyMove(ctx, fdst, fsrc, fs.DeleteModeOnly, false, deleteEmptySrcDirs, copyEmptySrcDirs)
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(ctx, fdst, fsrc, deleteMode, DoMove, deleteEmptySrcDirs, copyEmptySrcDirs)
if err != nil {
return err
}
return do.run()
}
// Sync fsrc into fdst
func Sync(ctx context.Context, fdst, fsrc fs.Fs, copyEmptySrcDirs bool) error {
ci := fs.GetConfig(ctx)
return runSyncCopyMove(ctx, fdst, fsrc, ci.DeleteMode, false, false, copyEmptySrcDirs)
}
// CopyDir copies fsrc into fdst
func CopyDir(ctx context.Context, fdst, fsrc fs.Fs, copyEmptySrcDirs bool) error {
return runSyncCopyMove(ctx, fdst, fsrc, fs.DeleteModeOff, false, false, copyEmptySrcDirs)
}
// moveDir moves fsrc into fdst
func moveDir(ctx context.Context, fdst, fsrc fs.Fs, deleteEmptySrcDirs bool, copyEmptySrcDirs bool) error {
return runSyncCopyMove(ctx, fdst, fsrc, fs.DeleteModeOff, true, deleteEmptySrcDirs, copyEmptySrcDirs)
}
// MoveDir moves fsrc into fdst
func MoveDir(ctx context.Context, fdst, fsrc fs.Fs, deleteEmptySrcDirs bool, copyEmptySrcDirs bool) error {
fi := filter.GetConfig(ctx)
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) && fi.InActive() {
if operations.SkipDestructive(ctx, fdst, "server-side directory move") {
return nil
}
fs.Debugf(fdst, "Using server-side directory move")
err := fdstDirMove(ctx, 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:
err = fs.CountError(err)
fs.Errorf(fdst, "Server side directory move failed: %v", err)
return err
}
}
// Otherwise move the files one by one
return moveDir(ctx, fdst, fsrc, deleteEmptySrcDirs, copyEmptySrcDirs)
}