// Package sync is the implementation of sync/copy/movepackage 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 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	}	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//   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, 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 checkersfunc (s *syncCopyMove) stopCheckers() {	s.toBeChecked.Close()	fs.Infof(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++ {		go s.pairCopyOrMove(s.toBeUploaded, s.fdst, &s.transfersWg)	}} // This stops the background transfersfunc (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 renamersfunc (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 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.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.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 directoriesfunc 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 directoriesfunc 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 := f.Mkdir(dir.Remote())			if err != nil {				fs.Errorf(fs.LogDirName(f, dir.Remote()), "Failed to Mkdir: %v", err)				accounting.Stats.Error(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

Lines 1–500 of 915. Load more