// Package sync is the implementation of sync/copy/movepackage sync import (	"context"	"fmt"	"path"	"sort"	"strings"	"sync"	"time" 	"github.com/pkg/errors"	"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	ctx                    context.Context        // internal context for controlling go-routines	cancel                 func()                 // cancel the 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   // stratgies 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} type trackRenamesStrategy byte const (	trackRenamesStrategyHash trackRenamesStrategy = 1 << iota	trackRenamesStrategyModtime) func (strategy trackRenamesStrategy) hash() bool {	return (strategy & trackRenamesStrategyHash) != 0} func (strategy trackRenamesStrategy) modTime() bool {	return (strategy & trackRenamesStrategyModtime) != 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.Overlapping(fdst, fsrc) {		return nil, fserrors.FatalError(fs.ErrorOverlapping)	}	s := &syncCopyMove{		fdst:                   fdst,		fsrc:                   fsrc,		deleteMode:             deleteMode,		DoMove:                 DoMove,		copyEmptySrcDirs:       copyEmptySrcDirs,		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),		noTraverse:             fs.Config.NoTraverse,		noCheckDest:            fs.Config.NoCheckDest,		noUnicodeNormalization: fs.Config.NoUnicodeNormalization,		deleteFilesCh:          make(chan fs.Object, fs.Config.Checkers),		trackRenames:           fs.Config.TrackRenames,		commonHash:             fsrc.Hashes().Overlap(fdst.Hashes()).GetOne(),		modifyWindow:           fs.GetModifyWindow(fsrc, fdst),		trackRenamesCh:         make(chan fs.Object, fs.Config.Checkers),		checkFirst:             fs.Config.CheckFirst,	}	backlog := fs.Config.MaxBacklog	if s.checkFirst {		fs.Infof(s.fdst, "Running all checks before starting transfers")		backlog = -1	}	var err error	s.toBeChecked, err = newPipe(fs.Config.OrderBy, accounting.Stats(ctx).SetCheckQueue, backlog)	if err != nil {		return nil, err	}	s.toBeUploaded, err = newPipe(fs.Config.OrderBy, accounting.Stats(ctx).SetTransferQueue, backlog)	if err != nil {		return nil, err	}	s.toBeRenamed, err = newPipe(fs.Config.OrderBy, accounting.Stats(ctx).SetRenameQueue, backlog)	if err != nil {		return nil, err	}	// If a max session duration has been defined add a deadline to the context	if fs.Config.MaxDuration > 0 {		endTime := time.Now().Add(fs.Config.MaxDuration)		fs.Infof(s.fdst, "Transfer session deadline: %s", endTime.Format("2006/01/02 15:04:05"))		s.ctx, s.cancel = context.WithDeadline(ctx, endTime)	} else {		s.ctx, s.cancel = context.WithCancel(ctx)	}	if s.noTraverse && s.deleteMode != fs.DeleteModeOff {		fs.Errorf(nil, "Ignoring --no-traverse with sync")		s.noTraverse = false	}	s.trackRenamesStrategy, err = parseTrackRenamesStrategy(fs.Config.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 fs.Config.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 {		// 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 fs.Config.BackupDir != "" || fs.Config.Suffix != "" {		var err error		s.backupDir, err = operations.BackupDir(fdst, fsrc, "")		if err != nil {			return nil, err		}	}	if fs.Config.CompareDest != "" {		var err error		s.compareCopyDest, err = operations.GetCompareDest()		if err != nil {			return nil, err		}	} else if fs.Config.CopyDest != "" {		var err error		s.compareCopyDest, err = operations.GetCopyDest(fdst)		if err != nil {			return nil, err		}	}	return s, nil} // Check to see if the context has been cancelledfunc (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 endfunc (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 filesfunc (s *syncCopyMove) processError(err error) {	if err == nil {		return	}	if err == context.DeadlineExceeded {		err = fserrors.NoRetryError(err)	}	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 precedence//   fatalErr//   normal error//   noRetryErrfunc (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 oncefunc (s *syncCopyMove) pairChecker(in *pipe, out *pipe, fraction int, wg *sync.WaitGroup) {	defer wg.Done()	for {		pair, ok := in.GetMax(s.ctx, 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 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 {						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					s.processError(operations.DeleteFile(s.ctx, src))				}			}		}		tr.Done(err)	}} // 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, fraction int, wg *sync.WaitGroup) {	defer wg.Done()	for {		pair, ok := in.GetMax(s.ctx, 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.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.ctx, 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(fs.Config.Checkers)	for i := 0; i < fs.Config.Checkers; i++ {		fraction := (100 * i) / fs.Config.Checkers		go s.pairChecker(s.toBeChecked, s.toBeUploaded, fraction, &s.checkerWg)	}} // This stops the background checkersfunc (s *syncCopyMove) stopCheckers() {	s.toBeChecked.Close()	fs.Debugf(s.fdst, "Waiting for checks to finish")	s.checkerWg.Wait()} // This starts the background transfersfunc (s *syncCopyMove) startTransfers() {	s.transfersWg.Add(fs.Config.Transfers)	for i := 0; i < fs.Config.Transfers; i++ {		fraction := (100 * i) / fs.Config.Transfers		go s.pairCopyOrMove(s.ctx, s.toBeUploaded, s.fdst, fraction, &s.transfersWg)	}} // This stops the background transfersfunc (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(fs.Config.Checkers)	for i := 0; i < fs.Config.Checkers; i++ {		fraction := (100 * i) / fs.Config.Checkers		go s.pairRenamer(s.toBeRenamed, s.toBeUploaded, fraction, &s.renamerWg)	}} // This stops the background renamersfunc (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 renamesfunc (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 collectionfunc (s *syncCopyMove) stopTrackRenames() {	if !s.trackRenames {		return	}	close(s.trackRenamesCh)	s.trackRenamesWg.Wait()} // This starts the background deletion of files for --delete-duringfunc (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 deletersfunc (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() && !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():

Lines 1–500 of 1089. Load more