felixfoertsch/syncthing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

lib/model, lib/versioner: Drive version cleanup from scanner (fixes #6313) (#6693)

Browse Source 
This change adds a separate config for the cleanup interval, and runs that cleanup from the main folder loop.
This commit is contained in:
Jakob Borg
2020-07-14 10:48:50 +02:00
committed by GitHub
parent 245b4b98c4
commit aedc2d788f
15 changed files with 178 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/model/folder.go
+53 -14
View File
@@ -29,6 +29,7 @@ import (
"github.com/syncthing/syncthing/lib/stats"
"github.com/syncthing/syncthing/lib/sync"
"github.com/syncthing/syncthing/lib/util"
"github.com/syncthing/syncthing/lib/versioner"
"github.com/syncthing/syncthing/lib/watchaggregator"
"github.com/thejerf/suture"
@@ -49,12 +50,14 @@ type folder struct {
ignores *ignore.Matcher
ctx context.Context
scanInterval time.Duration
scanTimer *time.Timer
scanDelay chan time.Duration
initialScanFinished chan struct{}
scanErrors []FileError
scanErrorsMut sync.Mutex
scanInterval time.Duration
scanTimer *time.Timer
scanDelay chan time.Duration
initialScanFinished chan struct{}
scanErrors []FileError
scanErrorsMut sync.Mutex
versionCleanupInterval time.Duration
versionCleanupTimer *time.Timer
pullScheduled chan struct{}
pullPause time.Duration
@@ -72,7 +75,8 @@ type folder struct {
watchErr error
watchMut sync.Mutex
puller puller
puller puller
versioner versioner.Versioner
}
type syncRequest struct {
@@ -81,10 +85,10 @@ type syncRequest struct {
}
type puller interface {
pull() bool // true when successfull and should not be retried
pull() bool // true when successful and should not be retried
}
func newFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, evLogger events.Logger, ioLimiter *byteSemaphore) folder {
func newFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, evLogger events.Logger, ioLimiter *byteSemaphore, ver versioner.Versioner) folder {
f := folder{
stateTracker: newStateTracker(cfg.ID, evLogger),
FolderConfiguration: cfg,
@@ -96,11 +100,13 @@ func newFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg conf
fset: fset,
ignores: ignores,
scanInterval: time.Duration(cfg.RescanIntervalS) * time.Second,
scanTimer: time.NewTimer(time.Millisecond), // The first scan should be done immediately.
scanDelay: make(chan time.Duration),
initialScanFinished: make(chan struct{}),
scanErrorsMut: sync.NewMutex(),
scanInterval: time.Duration(cfg.RescanIntervalS) * time.Second,
scanTimer: time.NewTimer(0), // The first scan should be done immediately.
scanDelay: make(chan time.Duration),
initialScanFinished: make(chan struct{}),
scanErrorsMut: sync.NewMutex(),
versionCleanupInterval: time.Duration(cfg.Versioning.CleanupIntervalS) * time.Second,
versionCleanupTimer: time.NewTimer(time.Duration(cfg.Versioning.CleanupIntervalS) * time.Second),
pullScheduled: make(chan struct{}, 1), // This needs to be 1-buffered so that we queue a pull if we're busy when it comes.
@@ -113,6 +119,8 @@ func newFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg conf
watchCancel: func() {},
restartWatchChan: make(chan struct{}, 1),
watchMut: sync.NewMutex(),
versioner: ver,
}
f.pullPause = f.pullBasePause()
f.pullFailTimer = time.NewTimer(0)
@@ -131,6 +139,7 @@ func (f *folder) serve(ctx context.Context) {
defer func() {
f.scanTimer.Stop()
f.versionCleanupTimer.Stop()
f.setState(FolderIdle)
}()
@@ -138,6 +147,14 @@ func (f *folder) serve(ctx context.Context) {
f.startWatch()
}
// If we're configured to not do version cleanup, or we don't have a
// versioner, cancel and drain that timer now.
if f.versionCleanupInterval == 0 || f.versioner == nil {
if !f.versionCleanupTimer.Stop() {
<-f.versionCleanupTimer.C
}
}
initialCompleted := f.initialScanFinished
for {
@@ -181,6 +198,10 @@ func (f *folder) serve(ctx context.Context) {
case <-f.restartWatchChan:
l.Debugln(f, "Restart watcher")
f.restartWatch()
case <-f.versionCleanupTimer.C:
l.Debugln(f, "Doing version cleanup")
f.versionCleanupTimerFired()
}
}
}
@@ -701,6 +722,24 @@ func (f *folder) scanTimerFired() {
f.Reschedule()
}
func (f *folder) versionCleanupTimerFired() {
f.setState(FolderCleanWaiting)
defer f.setState(FolderIdle)
if err := f.ioLimiter.takeWithContext(f.ctx, 1); err != nil {
return
}
defer f.ioLimiter.give(1)
f.setState(FolderCleaning)
if err := f.versioner.Clean(f.ctx); err != nil {
l.Infoln("Failed to clean versions in %s: %v", f.Description(), err)
}
f.versionCleanupTimer.Reset(f.versionCleanupInterval)
}
func (f *folder) WatchError() error {
f.watchMut.Lock()
defer f.watchMut.Unlock()
lib/model/folder_sendonly.go
+1 -1
View File
@@ -27,7 +27,7 @@ type sendOnlyFolder struct {
func newSendOnlyFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, _ versioner.Versioner, _ fs.Filesystem, evLogger events.Logger, ioLimiter *byteSemaphore) service {
f := &sendOnlyFolder{
folder: newFolder(model, fset, ignores, cfg, evLogger, ioLimiter),
folder: newFolder(model, fset, ignores, cfg, evLogger, ioLimiter, nil),
}
f.folder.puller = f
f.folder.Service = util.AsService(f.serve, f.String())
lib/model/folder_sendrecv.go
+2 -4
View File
@@ -123,8 +123,7 @@ type dbUpdateJob struct {
type sendReceiveFolder struct {
folder
fs fs.Filesystem
versioner versioner.Versioner
fs fs.Filesystem
queue *jobQueue
blockPullReorderer blockPullReorderer
@@ -137,9 +136,8 @@ type sendReceiveFolder struct {
func newSendReceiveFolder(model *model, fset *db.FileSet, ignores *ignore.Matcher, cfg config.FolderConfiguration, ver versioner.Versioner, fs fs.Filesystem, evLogger events.Logger, ioLimiter *byteSemaphore) service {
f := &sendReceiveFolder{
folder: newFolder(model, fset, ignores, cfg, evLogger, ioLimiter),
folder: newFolder(model, fset, ignores, cfg, evLogger, ioLimiter, ver),
fs: fs,
versioner: ver,
queue: newJobQueue(),
blockPullReorderer: newBlockPullReorderer(cfg.BlockPullOrder, model.id, cfg.DeviceIDs()),
writeLimiter: newByteSemaphore(cfg.MaxConcurrentWrites),
lib/model/folderstate.go
+6
View File
@@ -22,6 +22,8 @@ const (
FolderSyncWaiting
FolderSyncPreparing
FolderSyncing
FolderCleaning
FolderCleanWaiting
FolderError
)
@@ -39,6 +41,10 @@ func (s folderState) String() string {
return "sync-preparing"
case FolderSyncing:
return "syncing"
case FolderCleaning:
return "cleaning"
case FolderCleanWaiting:
return "clean-waiting"
case FolderError:
return "error"
default:
lib/model/model.go
+22 -30
View File
@@ -133,15 +133,15 @@ type model struct {
folderIOLimiter *byteSemaphore
// fields protected by fmut
fmut sync.RWMutex
folderCfgs map[string]config.FolderConfiguration // folder -> cfg
folderFiles map[string]*db.FileSet // folder -> files
deviceStatRefs map[protocol.DeviceID]*stats.DeviceStatisticsReference // deviceID -> statsRef
folderIgnores map[string]*ignore.Matcher // folder -> matcher object
folderRunners map[string]service // folder -> puller or scanner
folderRunnerTokens map[string][]suture.ServiceToken // folder -> tokens for puller or scanner
folderRestartMuts syncMutexMap // folder -> restart mutex
folderVersioners map[string]versioner.Versioner // folder -> versioner (may be nil)
fmut sync.RWMutex
folderCfgs map[string]config.FolderConfiguration // folder -> cfg
folderFiles map[string]*db.FileSet // folder -> files
deviceStatRefs map[protocol.DeviceID]*stats.DeviceStatisticsReference // deviceID -> statsRef
folderIgnores map[string]*ignore.Matcher // folder -> matcher object
folderRunners map[string]service // folder -> puller or scanner
folderRunnerToken map[string]suture.ServiceToken // folder -> token for folder runner
folderRestartMuts syncMutexMap // folder -> restart mutex
folderVersioners map[string]versioner.Versioner // folder -> versioner (may be nil)
// fields protected by pmut
pmut sync.RWMutex
@@ -207,14 +207,14 @@ func NewModel(cfg config.Wrapper, id protocol.DeviceID, clientName, clientVersio
folderIOLimiter: newByteSemaphore(cfg.Options().MaxFolderConcurrency()),
// fields protected by fmut
fmut: sync.NewRWMutex(),
folderCfgs: make(map[string]config.FolderConfiguration),
folderFiles: make(map[string]*db.FileSet),
deviceStatRefs: make(map[protocol.DeviceID]*stats.DeviceStatisticsReference),
folderIgnores: make(map[string]*ignore.Matcher),
folderRunners: make(map[string]service),
folderRunnerTokens: make(map[string][]suture.ServiceToken),
folderVersioners: make(map[string]versioner.Versioner),
fmut: sync.NewRWMutex(),
folderCfgs: make(map[string]config.FolderConfiguration),
folderFiles: make(map[string]*db.FileSet),
deviceStatRefs: make(map[protocol.DeviceID]*stats.DeviceStatisticsReference),
folderIgnores: make(map[string]*ignore.Matcher),
folderRunners: make(map[string]service),
folderRunnerToken: make(map[string]suture.ServiceToken),
folderVersioners: make(map[string]versioner.Versioner),
// fields protected by pmut
pmut: sync.NewRWMutex(),
@@ -346,13 +346,6 @@ func (m *model) addAndStartFolderLockedWithIgnores(cfg config.FolderConfiguratio
if err != nil {
panic(fmt.Errorf("creating versioner: %w", err))
}
if service, ok := ver.(suture.Service); ok {
// The versioner implements the suture.Service interface, so
// expects to be run in the background in addition to being called
// when files are going to be archived.
token := m.Add(service)
m.folderRunnerTokens[folder] = append(m.folderRunnerTokens[folder], token)
}
}
m.folderVersioners[folder] = ver
@@ -362,8 +355,7 @@ func (m *model) addAndStartFolderLockedWithIgnores(cfg config.FolderConfiguratio
m.warnAboutOverwritingProtectedFiles(cfg, ignores)
token := m.Add(p)
m.folderRunnerTokens[folder] = append(m.folderRunnerTokens[folder], token)
m.folderRunnerToken[folder] = m.Add(p)
l.Infof("Ready to synchronize %s (%s)", cfg.Description(), cfg.Type)
}
@@ -430,11 +422,11 @@ func (m *model) stopFolder(cfg config.FolderConfiguration, err error) {
// Stop the services running for this folder and wait for them to finish
// stopping to prevent races on restart.
m.fmut.RLock()
tokens := m.folderRunnerTokens[cfg.ID]
token, ok := m.folderRunnerToken[cfg.ID]
m.fmut.RUnlock()
for _, id := range tokens {
m.RemoveAndWait(id, 0)
if ok {
m.RemoveAndWait(token, 0)
}
// Wait for connections to stop to ensure that no more calls to methods
@@ -449,7 +441,7 @@ func (m *model) cleanupFolderLocked(cfg config.FolderConfiguration) {
delete(m.folderFiles, cfg.ID)
delete(m.folderIgnores, cfg.ID)
delete(m.folderRunners, cfg.ID)
delete(m.folderRunnerTokens, cfg.ID)
delete(m.folderRunnerToken, cfg.ID)
delete(m.folderVersioners, cfg.ID)
}