<commit_before>// Package sync is the implementation of sync/copy/move
package sync

import (
	"context"
	"fmt"
	"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   []fs.DirEntry          // potentially empty directories
	srcEmptyDirsMu sync.Mutex             // protect srcEmptyDirs
	srcEmptyDirs   []fs.DirEntry          // potentially empty directories
	checkerWg      sync.WaitGroup         // wait for checkers
	toBeChecked    fs.ObjectPairChan      // checkers channel
	transfersWg    sync.WaitGroup         // wait for transfers
	toBeUploaded   fs.ObjectPairChan      // 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    fs.ObjectPairChan      // 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),
		toBeChecked:        make(fs.ObjectPairChan, fs.Config.Transfers),
		toBeUploaded:       make(fs.ObjectPairChan, fs.Config.Transfers),
		deleteFilesCh:      make(chan fs.Object, fs.Config.Checkers),
		trackRenames:       fs.Config.TrackRenames,
		commonHash:         fsrc.Hashes().Overlap(fdst.Hashes()).GetOne(),
		toBeRenamed:        make(fs.ObjectPairChan, fs.Config.Transfers),
		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 {
	select {
	case <-s.ctx.Done():
		return true
	default:
	}
	return false
}

// 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() {
			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<commit_msg>sync: log when we abandon the sync due to a fatal error<commit_after>// Package sync is the implementation of sync/copy/move
package sync

import (
	"context"
	"fmt"
	"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   []fs.DirEntry          // potentially empty directories
	srcEmptyDirsMu sync.Mutex             // protect srcEmptyDirs
	srcEmptyDirs   []fs.DirEntry          // potentially empty directories
	checkerWg      sync.WaitGroup         // wait for checkers
	toBeChecked    fs.ObjectPairChan      // checkers channel
	transfersWg    sync.WaitGroup         // wait for transfers
	toBeUploaded   fs.ObjectPairChan      // 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    fs.ObjectPairChan      // 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),
		toBeChecked:        make(fs.ObjectPairChan, fs.Config.Transfers),
		toBeUploaded:       make(fs.ObjectPairChan, fs.Config.Transfers),
		deleteFilesCh:      make(chan fs.Object, fs.Config.Checkers),
		trackRenames:       fs.Config.TrackRenames,
		commonHash:         fsrc.Hashes().Overlap(fdst.Hashes()).GetOne(),
		toBeRenamed:        make(fs.ObjectPairChan, fs.Config.Transfers),
		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 {
	select {
	case <-s.ctx.Done():
		return true
	default:
	}
	return false
}

// 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
{'fraction_non_alphanumeric': 0.09608667759640745, 'fraction_numerical': 0.00028512367239290045, 'mean_word_length': 2.5600101497081957, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '755216', 'n_tokens_mistral': 4658, 'n_tokens_neox': 4345, 'n_words': 1277}