mirror of
https://github.com/octoleo/syncthing.git
synced 2025-01-03 15:17:25 +00:00
lib/model: Refactor out folder and folderscan types, simplify somewhat
GitHub-Pull-Request: https://github.com/syncthing/syncthing/pull/3007
This commit is contained in:
parent
2467678bd4
commit
236f121c4e
53
lib/model/folder.go
Normal file
53
lib/model/folder.go
Normal file
@ -0,0 +1,53 @@
|
||||
// Copyright (C) 2014 The Syncthing Authors.
|
||||
//
|
||||
// This Source Code Form is subject to the terms of the Mozilla Public
|
||||
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
|
||||
// You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
|
||||
package model
|
||||
|
||||
import "time"
|
||||
|
||||
type folder struct {
|
||||
stateTracker
|
||||
scan folderscan
|
||||
model *Model
|
||||
stop chan struct{}
|
||||
}
|
||||
|
||||
func (f *folder) IndexUpdated() {
|
||||
}
|
||||
|
||||
func (f *folder) DelayScan(next time.Duration) {
|
||||
f.scan.Delay(next)
|
||||
}
|
||||
|
||||
func (f *folder) Scan(subdirs []string) error {
|
||||
return f.scan.Scan(subdirs)
|
||||
}
|
||||
func (f *folder) Stop() {
|
||||
close(f.stop)
|
||||
}
|
||||
|
||||
func (f *folder) Jobs() ([]string, []string) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (f *folder) BringToFront(string) {}
|
||||
|
||||
func (f *folder) scanSubdirsIfHealthy(subDirs []string) error {
|
||||
if err := f.model.CheckFolderHealth(f.folderID); err != nil {
|
||||
l.Infoln("Skipping folder", f.folderID, "scan due to folder error:", err)
|
||||
return err
|
||||
}
|
||||
l.Debugln(f, "Scanning subdirectories")
|
||||
if err := f.model.internalScanFolderSubdirs(f.folderID, subDirs); err != nil {
|
||||
// Potentially sets the error twice, once in the scanner just
|
||||
// by doing a check, and once here, if the error returned is
|
||||
// the same one as returned by CheckFolderHealth, though
|
||||
// duplicate set is handled by setError.
|
||||
f.setError(err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
49
lib/model/folderscan.go
Normal file
49
lib/model/folderscan.go
Normal file
@ -0,0 +1,49 @@
|
||||
// Copyright (C) 2016 The Syncthing Authors.
|
||||
//
|
||||
// This Source Code Form is subject to the terms of the Mozilla Public
|
||||
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
|
||||
// You can obtain one at http://mozilla.org/MPL/2.0/.
|
||||
|
||||
package model
|
||||
|
||||
import (
|
||||
"math/rand"
|
||||
"time"
|
||||
)
|
||||
|
||||
type rescanRequest struct {
|
||||
subdirs []string
|
||||
err chan error
|
||||
}
|
||||
|
||||
// bundle all folder scan activity
|
||||
type folderscan struct {
|
||||
interval time.Duration
|
||||
timer *time.Timer
|
||||
now chan rescanRequest
|
||||
delay chan time.Duration
|
||||
}
|
||||
|
||||
func (s *folderscan) reschedule() {
|
||||
if s.interval == 0 {
|
||||
return
|
||||
}
|
||||
// Sleep a random time between 3/4 and 5/4 of the configured interval.
|
||||
sleepNanos := (s.interval.Nanoseconds()*3 + rand.Int63n(2*s.interval.Nanoseconds())) / 4
|
||||
interval := time.Duration(sleepNanos) * time.Nanosecond
|
||||
l.Debugln(s, "next rescan in", interval)
|
||||
s.timer.Reset(interval)
|
||||
}
|
||||
|
||||
func (s *folderscan) Scan(subdirs []string) error {
|
||||
req := rescanRequest{
|
||||
subdirs: subdirs,
|
||||
err: make(chan error),
|
||||
}
|
||||
s.now <- req
|
||||
return <-req.err
|
||||
}
|
||||
|
||||
func (s *folderscan) Delay(next time.Duration) {
|
||||
s.delay <- next
|
||||
}
|
@ -38,7 +38,7 @@ func (s folderState) String() string {
|
||||
}
|
||||
|
||||
type stateTracker struct {
|
||||
folder string
|
||||
folderID string
|
||||
|
||||
mut sync.Mutex
|
||||
current folderState
|
||||
@ -61,7 +61,7 @@ func (s *stateTracker) setState(newState folderState) {
|
||||
*/
|
||||
|
||||
eventData := map[string]interface{}{
|
||||
"folder": s.folder,
|
||||
"folder": s.folderID,
|
||||
"to": newState.String(),
|
||||
"from": s.current.String(),
|
||||
}
|
||||
@ -92,7 +92,7 @@ func (s *stateTracker) setError(err error) {
|
||||
s.mut.Lock()
|
||||
if s.current != FolderError || s.err.Error() != err.Error() {
|
||||
eventData := map[string]interface{}{
|
||||
"folder": s.folder,
|
||||
"folder": s.folderID,
|
||||
"to": FolderError.String(),
|
||||
"from": s.current.String(),
|
||||
"error": err.Error(),
|
||||
@ -116,7 +116,7 @@ func (s *stateTracker) clearError() {
|
||||
s.mut.Lock()
|
||||
if s.current == FolderError {
|
||||
eventData := map[string]interface{}{
|
||||
"folder": s.folder,
|
||||
"folder": s.folderID,
|
||||
"to": FolderIdle.String(),
|
||||
"from": s.current.String(),
|
||||
}
|
||||
|
@ -172,7 +172,7 @@ func (m *Model) StartFolderRW(folder string) {
|
||||
if ok {
|
||||
panic("cannot start already running folder " + folder)
|
||||
}
|
||||
p := newRWFolder(m, m.shortID, cfg)
|
||||
p := newRWFolder(m, cfg)
|
||||
m.folderRunners[folder] = p
|
||||
|
||||
if len(cfg.Versioning.Type) > 0 {
|
||||
@ -243,7 +243,7 @@ func (m *Model) StartFolderRO(folder string) {
|
||||
if ok {
|
||||
panic("cannot start already running folder " + folder)
|
||||
}
|
||||
s := newROFolder(m, folder, time.Duration(cfg.RescanIntervalS)*time.Second)
|
||||
s := newROFolder(m, cfg)
|
||||
m.folderRunners[folder] = s
|
||||
|
||||
token := m.Add(s)
|
||||
@ -1360,7 +1360,7 @@ func (m *Model) ScanFolderSubs(folder string, subs []string) error {
|
||||
return runner.Scan(subs)
|
||||
}
|
||||
|
||||
func (m *Model) internalScanFolderSubs(folder string, subs []string) error {
|
||||
func (m *Model) internalScanFolderSubdirs(folder string, subs []string) error {
|
||||
for i, sub := range subs {
|
||||
sub = osutil.NativeFilename(sub)
|
||||
if p := filepath.Clean(filepath.Join(folder, sub)); !strings.HasPrefix(p, folder) {
|
||||
|
@ -8,151 +8,88 @@ package model
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"time"
|
||||
|
||||
"github.com/syncthing/syncthing/lib/config"
|
||||
"github.com/syncthing/syncthing/lib/sync"
|
||||
)
|
||||
|
||||
type roFolder struct {
|
||||
stateTracker
|
||||
|
||||
folder string
|
||||
intv time.Duration
|
||||
timer *time.Timer
|
||||
model *Model
|
||||
stop chan struct{}
|
||||
scanNow chan rescanRequest
|
||||
delayScan chan time.Duration
|
||||
folder
|
||||
}
|
||||
|
||||
type rescanRequest struct {
|
||||
subs []string
|
||||
err chan error
|
||||
}
|
||||
|
||||
func newROFolder(model *Model, folder string, interval time.Duration) *roFolder {
|
||||
func newROFolder(model *Model, cfg config.FolderConfiguration) *roFolder {
|
||||
return &roFolder{
|
||||
folder: folder{
|
||||
stateTracker: stateTracker{
|
||||
folder: folder,
|
||||
folderID: cfg.ID,
|
||||
mut: sync.NewMutex(),
|
||||
},
|
||||
folder: folder,
|
||||
intv: interval,
|
||||
scan: folderscan{
|
||||
interval: time.Duration(cfg.RescanIntervalS) * time.Second,
|
||||
timer: time.NewTimer(time.Millisecond),
|
||||
model: model,
|
||||
now: make(chan rescanRequest),
|
||||
delay: make(chan time.Duration),
|
||||
},
|
||||
stop: make(chan struct{}),
|
||||
scanNow: make(chan rescanRequest),
|
||||
delayScan: make(chan time.Duration),
|
||||
model: model,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (s *roFolder) Serve() {
|
||||
l.Debugln(s, "starting")
|
||||
defer l.Debugln(s, "exiting")
|
||||
func (f *roFolder) Serve() {
|
||||
l.Debugln(f, "starting")
|
||||
defer l.Debugln(f, "exiting")
|
||||
|
||||
defer func() {
|
||||
s.timer.Stop()
|
||||
f.scan.timer.Stop()
|
||||
}()
|
||||
|
||||
reschedule := func() {
|
||||
if s.intv == 0 {
|
||||
return
|
||||
}
|
||||
// Sleep a random time between 3/4 and 5/4 of the configured interval.
|
||||
sleepNanos := (s.intv.Nanoseconds()*3 + rand.Int63n(2*s.intv.Nanoseconds())) / 4
|
||||
s.timer.Reset(time.Duration(sleepNanos) * time.Nanosecond)
|
||||
}
|
||||
|
||||
initialScanCompleted := false
|
||||
for {
|
||||
select {
|
||||
case <-s.stop:
|
||||
case <-f.stop:
|
||||
return
|
||||
|
||||
case <-s.timer.C:
|
||||
if err := s.model.CheckFolderHealth(s.folder); err != nil {
|
||||
l.Infoln("Skipping folder", s.folder, "scan due to folder error:", err)
|
||||
reschedule()
|
||||
case <-f.scan.timer.C:
|
||||
if err := f.model.CheckFolderHealth(f.folderID); err != nil {
|
||||
l.Infoln("Skipping folder", f.folderID, "scan due to folder error:", err)
|
||||
f.scan.reschedule()
|
||||
continue
|
||||
}
|
||||
|
||||
l.Debugln(s, "rescan")
|
||||
l.Debugln(f, "rescan")
|
||||
|
||||
if err := s.model.internalScanFolderSubs(s.folder, nil); err != nil {
|
||||
if err := f.model.internalScanFolderSubdirs(f.folderID, nil); err != nil {
|
||||
// Potentially sets the error twice, once in the scanner just
|
||||
// by doing a check, and once here, if the error returned is
|
||||
// the same one as returned by CheckFolderHealth, though
|
||||
// duplicate set is handled by setError.
|
||||
s.setError(err)
|
||||
reschedule()
|
||||
f.setError(err)
|
||||
f.scan.reschedule()
|
||||
continue
|
||||
}
|
||||
|
||||
if !initialScanCompleted {
|
||||
l.Infoln("Completed initial scan (ro) of folder", s.folder)
|
||||
l.Infoln("Completed initial scan (ro) of folder", f.folderID)
|
||||
initialScanCompleted = true
|
||||
}
|
||||
|
||||
if s.intv == 0 {
|
||||
if f.scan.interval == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
reschedule()
|
||||
f.scan.reschedule()
|
||||
|
||||
case req := <-s.scanNow:
|
||||
if err := s.model.CheckFolderHealth(s.folder); err != nil {
|
||||
l.Infoln("Skipping folder", s.folder, "scan due to folder error:", err)
|
||||
req.err <- err
|
||||
continue
|
||||
}
|
||||
case req := <-f.scan.now:
|
||||
req.err <- f.scanSubdirsIfHealthy(req.subdirs)
|
||||
|
||||
l.Debugln(s, "forced rescan")
|
||||
|
||||
if err := s.model.internalScanFolderSubs(s.folder, req.subs); err != nil {
|
||||
// Potentially sets the error twice, once in the scanner just
|
||||
// by doing a check, and once here, if the error returned is
|
||||
// the same one as returned by CheckFolderHealth, though
|
||||
// duplicate set is handled by setError.
|
||||
s.setError(err)
|
||||
req.err <- err
|
||||
continue
|
||||
}
|
||||
|
||||
req.err <- nil
|
||||
|
||||
case next := <-s.delayScan:
|
||||
s.timer.Reset(next)
|
||||
case next := <-f.scan.delay:
|
||||
f.scan.timer.Reset(next)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *roFolder) Stop() {
|
||||
close(s.stop)
|
||||
}
|
||||
|
||||
func (s *roFolder) IndexUpdated() {
|
||||
}
|
||||
|
||||
func (s *roFolder) Scan(subs []string) error {
|
||||
req := rescanRequest{
|
||||
subs: subs,
|
||||
err: make(chan error),
|
||||
}
|
||||
s.scanNow <- req
|
||||
return <-req.err
|
||||
}
|
||||
|
||||
func (s *roFolder) String() string {
|
||||
return fmt.Sprintf("roFolder/%s@%p", s.folder, s)
|
||||
}
|
||||
|
||||
func (s *roFolder) BringToFront(string) {}
|
||||
|
||||
func (s *roFolder) Jobs() ([]string, []string) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (s *roFolder) DelayScan(next time.Duration) {
|
||||
s.delayScan <- next
|
||||
func (f *roFolder) String() string {
|
||||
return fmt.Sprintf("roFolder/%s@%p", f.folderID, f)
|
||||
}
|
||||
|
File diff suppressed because it is too large
Load Diff
@ -67,13 +67,14 @@ func setUpModel(file protocol.FileInfo) *Model {
|
||||
}
|
||||
|
||||
func setUpRwFolder(model *Model) rwFolder {
|
||||
return rwFolder{
|
||||
folder: "default",
|
||||
f := rwFolder{
|
||||
dir: "testdata",
|
||||
model: model,
|
||||
errors: make(map[string]string),
|
||||
errorsMut: sync.NewMutex(),
|
||||
}
|
||||
f.folderID = "default"
|
||||
f.model = model
|
||||
return f
|
||||
}
|
||||
|
||||
// Layout of the files: (indexes from the above array)
|
||||
@ -329,17 +330,17 @@ func TestDeregisterOnFailInCopy(t *testing.T) {
|
||||
m.AddFolder(defaultFolderConfig)
|
||||
|
||||
emitter := NewProgressEmitter(defaultConfig)
|
||||
m.progressEmitter = emitter
|
||||
go emitter.Serve()
|
||||
|
||||
p := rwFolder{
|
||||
folder: "default",
|
||||
dir: "testdata",
|
||||
model: m,
|
||||
queue: newJobQueue(),
|
||||
progressEmitter: emitter,
|
||||
errors: make(map[string]string),
|
||||
errorsMut: sync.NewMutex(),
|
||||
}
|
||||
p.folderID = "default"
|
||||
p.model = m
|
||||
|
||||
// queue.Done should be called by the finisher routine
|
||||
p.queue.Push("filex", 0, 0)
|
||||
@ -373,7 +374,7 @@ func TestDeregisterOnFailInCopy(t *testing.T) {
|
||||
case state := <-finisherBufferChan:
|
||||
// At this point the file should still be registered with both the job
|
||||
// queue, and the progress emitter. Verify this.
|
||||
if p.progressEmitter.lenRegistry() != 1 || p.queue.lenProgress() != 1 || p.queue.lenQueued() != 0 {
|
||||
if p.model.progressEmitter.lenRegistry() != 1 || p.queue.lenProgress() != 1 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Could not find file")
|
||||
}
|
||||
|
||||
@ -388,16 +389,16 @@ func TestDeregisterOnFailInCopy(t *testing.T) {
|
||||
t.Fatal("File not closed?")
|
||||
}
|
||||
|
||||
if p.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
if p.model.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.model.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
}
|
||||
|
||||
// Doing it again should have no effect
|
||||
finisherChan <- state
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
if p.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
if p.model.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.model.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Didn't get anything to the finisher")
|
||||
@ -413,17 +414,17 @@ func TestDeregisterOnFailInPull(t *testing.T) {
|
||||
m.AddFolder(defaultFolderConfig)
|
||||
|
||||
emitter := NewProgressEmitter(defaultConfig)
|
||||
m.progressEmitter = emitter
|
||||
go emitter.Serve()
|
||||
|
||||
p := rwFolder{
|
||||
folder: "default",
|
||||
dir: "testdata",
|
||||
model: m,
|
||||
queue: newJobQueue(),
|
||||
progressEmitter: emitter,
|
||||
errors: make(map[string]string),
|
||||
errorsMut: sync.NewMutex(),
|
||||
}
|
||||
p.folderID = "default"
|
||||
p.model = m
|
||||
|
||||
// queue.Done should be called by the finisher routine
|
||||
p.queue.Push("filex", 0, 0)
|
||||
@ -450,7 +451,7 @@ func TestDeregisterOnFailInPull(t *testing.T) {
|
||||
case state := <-finisherBufferChan:
|
||||
// At this point the file should still be registered with both the job
|
||||
// queue, and the progress emitter. Verify this.
|
||||
if p.progressEmitter.lenRegistry() != 1 || p.queue.lenProgress() != 1 || p.queue.lenQueued() != 0 {
|
||||
if p.model.progressEmitter.lenRegistry() != 1 || p.queue.lenProgress() != 1 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Could not find file")
|
||||
}
|
||||
|
||||
@ -465,16 +466,16 @@ func TestDeregisterOnFailInPull(t *testing.T) {
|
||||
t.Fatal("File not closed?")
|
||||
}
|
||||
|
||||
if p.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
if p.model.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.model.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
}
|
||||
|
||||
// Doing it again should have no effect
|
||||
finisherChan <- state
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
if p.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
if p.model.progressEmitter.lenRegistry() != 0 || p.queue.lenProgress() != 0 || p.queue.lenQueued() != 0 {
|
||||
t.Fatal("Still registered", p.model.progressEmitter.lenRegistry(), p.queue.lenProgress(), p.queue.lenQueued())
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Didn't get anything to the finisher")
|
||||
|
Loading…
Reference in New Issue
Block a user