perso/syncthing-arm
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

lib/model, lib/config: Support "live" device removal, folder unsharing and folder configuration changes

Browse Source 
Furthermore:
1. Cleans configs received, migrates them as we receive them.
2. Clears indexes of devices we no longer share the folder with

GitHub-Pull-Request: https://github.com/syncthing/syncthing/pull/3478
This commit is contained in:
Audrius Butkevicius
2016-08-07 16:21:59 +00:00
parent fbe42c156d
commit af3b6f9c83
9 changed files with 352 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
lib/model/model.go
View File

@@ -93,12 +93,11 @@ type Model struct {
folderStatRefs map[string]*stats.FolderStatisticsReference // folder -> statsRef
fmut sync.RWMutex // protects the above
conn map[protocol.DeviceID]connections.Connection
helloMessages map[protocol.DeviceID]protocol.HelloResult
deviceClusterConf map[protocol.DeviceID]protocol.ClusterConfig
devicePaused map[protocol.DeviceID]bool
deviceDownloads map[protocol.DeviceID]*deviceDownloadState
pmut sync.RWMutex // protects the above
conn map[protocol.DeviceID]connections.Connection
helloMessages map[protocol.DeviceID]protocol.HelloResult
devicePaused map[protocol.DeviceID]bool
deviceDownloads map[protocol.DeviceID]*deviceDownloadState
pmut sync.RWMutex // protects the above
}
type folderFactory func(*Model, config.FolderConfiguration, versioner.Versioner, *fs.MtimeFS) service
@@ -154,7 +153,6 @@ func NewModel(cfg *config.Wrapper, id protocol.DeviceID, deviceName, clientName,
folderStatRefs: make(map[string]*stats.FolderStatisticsReference),
conn: make(map[protocol.DeviceID]connections.Connection),
helloMessages: make(map[protocol.DeviceID]protocol.HelloResult),
deviceClusterConf: make(map[protocol.DeviceID]protocol.ClusterConfig),
devicePaused: make(map[protocol.DeviceID]bool),
deviceDownloads: make(map[protocol.DeviceID]*deviceDownloadState),
fmut: sync.NewRWMutex(),
@@ -163,6 +161,7 @@ func NewModel(cfg *config.Wrapper, id protocol.DeviceID, deviceName, clientName,
if cfg.Options().ProgressUpdateIntervalS > -1 {
go m.progressEmitter.Serve()
}
cfg.Subscribe(m)
return m
}
@@ -179,6 +178,13 @@ func (m *Model) StartDeadlockDetector(timeout time.Duration) {
// StartFolder constructs the folder service and starts it.
func (m *Model) StartFolder(folder string) {
m.fmut.Lock()
folderType := m.startFolderLocked(folder)
m.fmut.Unlock()
l.Infoln("Ready to synchronize", folder, fmt.Sprintf("(%s)", folderType))
}
func (m *Model) startFolderLocked(folder string) config.FolderType {
cfg, ok := m.folderCfgs[folder]
if !ok {
panic("cannot start nonexistent folder " + folder)
@@ -195,6 +201,17 @@ func (m *Model) StartFolder(folder string) {
}
fs := m.folderFiles[folder]
// Find any devices for which we hold the index in the db, but the folder
// is not shared, and drop it.
expected := mapDevices(cfg.DeviceIDs())
for _, available := range fs.ListDevices() {
if _, ok := expected[available]; !ok {
l.Debugln("dropping", folder, "state for", available)
fs.Replace(available, nil)
}
}
v, ok := fs.Sequence(protocol.LocalDeviceID), true
indexHasFiles := ok && v > 0
if !indexHasFiles {
@@ -238,9 +255,8 @@ func (m *Model) StartFolder(folder string) {
token := m.Add(p)
m.folderRunnerTokens[folder] = append(m.folderRunnerTokens[folder], token)
m.fmut.Unlock()
l.Infoln("Ready to synchronize", folder, fmt.Sprintf("(%s)", cfg.Type))
return cfg.Type
}
func (m *Model) warnAboutOverwritingProtectedFiles(folder string) {
@@ -271,10 +287,46 @@ func (m *Model) warnAboutOverwritingProtectedFiles(folder string) {
}
}
func (m *Model) AddFolder(cfg config.FolderConfiguration) {
if len(cfg.ID) == 0 {
panic("cannot add empty folder id")
}
m.fmut.Lock()
m.addFolderLocked(cfg)
m.fmut.Unlock()
}
func (m *Model) addFolderLocked(cfg config.FolderConfiguration) {
m.folderCfgs[cfg.ID] = cfg
m.folderFiles[cfg.ID] = db.NewFileSet(cfg.ID, m.db)
m.folderDevices[cfg.ID] = make([]protocol.DeviceID, len(cfg.Devices))
for i, device := range cfg.Devices {
m.folderDevices[cfg.ID][i] = device.DeviceID
m.deviceFolders[device.DeviceID] = append(m.deviceFolders[device.DeviceID], cfg.ID)
}
ignores := ignore.New(m.cacheIgnoredFiles)
if err := ignores.Load(filepath.Join(cfg.Path(), ".stignore")); err != nil && !os.IsNotExist(err) {
l.Warnln("Loading ignores:", err)
}
m.folderIgnores[cfg.ID] = ignores
}
func (m *Model) RemoveFolder(folder string) {
m.fmut.Lock()
m.pmut.Lock()
m.tearDownFolderLocked(folder)
// Remove it from the database
db.DropFolder(m.db, folder)
m.pmut.Unlock()
m.fmut.Unlock()
}
func (m *Model) tearDownFolderLocked(folder string) {
// Stop the services running for this folder
for _, id := range m.folderRunnerTokens[folder] {
m.Remove(id)
@@ -298,12 +350,24 @@ func (m *Model) RemoveFolder(folder string) {
for dev, folders := range m.deviceFolders {
m.deviceFolders[dev] = stringSliceWithout(folders, folder)
}
}
// Remove it from the database
db.DropFolder(m.db, folder)
func (m *Model) RestartFolder(cfg config.FolderConfiguration) {
if len(cfg.ID) == 0 {
panic("cannot add empty folder id")
}
m.fmut.Lock()
m.pmut.Lock()
m.tearDownFolderLocked(cfg.ID)
m.addFolderLocked(cfg)
folderType := m.startFolderLocked(cfg.ID)
m.pmut.Unlock()
m.fmut.Unlock()
l.Infoln("Restarted folder", cfg.ID, fmt.Sprintf("(%s)", folderType))
}
type ConnectionInfo struct {
@@ -640,10 +704,10 @@ func (m *Model) IndexUpdate(deviceID protocol.DeviceID, folder string, fs []prot
func (m *Model) folderSharedWith(folder string, deviceID protocol.DeviceID) bool {
m.fmut.RLock()
defer m.fmut.RUnlock()
return m.folderSharedWithUnlocked(folder, deviceID)
return m.folderSharedWithLocked(folder, deviceID)
}
func (m *Model) folderSharedWithUnlocked(folder string, deviceID protocol.DeviceID) bool {
func (m *Model) folderSharedWithLocked(folder string, deviceID protocol.DeviceID) bool {
for _, nfolder := range m.deviceFolders[deviceID] {
if nfolder == folder {
return true
@@ -671,7 +735,7 @@ func (m *Model) ClusterConfig(deviceID protocol.DeviceID, cm protocol.ClusterCon
m.fmut.Lock()
for _, folder := range cm.Folders {
if !m.folderSharedWithUnlocked(folder.ID, deviceID) {
if !m.folderSharedWithLocked(folder.ID, deviceID) {
events.Default.Log(events.FolderRejected, map[string]string{
"folder": folder.ID,
"folderLabel": folder.Label,
@@ -865,7 +929,6 @@ func (m *Model) Close(device protocol.DeviceID, err error) {
}
delete(m.conn, device)
delete(m.helloMessages, device)
delete(m.deviceClusterConf, device)
delete(m.deviceDownloads, device)
m.pmut.Unlock()
}
@@ -1427,30 +1490,6 @@ func (m *Model) requestGlobal(deviceID protocol.DeviceID, folder, name string, o
return nc.Request(folder, name, offset, size, hash, fromTemporary)
}
func (m *Model) AddFolder(cfg config.FolderConfiguration) {
if len(cfg.ID) == 0 {
panic("cannot add empty folder id")
}
m.fmut.Lock()
m.folderCfgs[cfg.ID] = cfg
m.folderFiles[cfg.ID] = db.NewFileSet(cfg.ID, m.db)
m.folderDevices[cfg.ID] = make([]protocol.DeviceID, len(cfg.Devices))
for i, device := range cfg.Devices {
m.folderDevices[cfg.ID][i] = device.DeviceID
m.deviceFolders[device.DeviceID] = append(m.deviceFolders[device.DeviceID], cfg.ID)
}
ignores := ignore.New(m.cacheIgnoredFiles)
if err := ignores.Load(filepath.Join(cfg.Path(), ".stignore")); err != nil && !os.IsNotExist(err) {
l.Warnln("Loading ignores:", err)
}
m.folderIgnores[cfg.ID] = ignores
m.fmut.Unlock()
}
func (m *Model) ScanFolders() map[string]error {
m.fmut.RLock()
folders := make([]string, 0, len(m.folderCfgs))
@@ -2154,62 +2193,24 @@ func (m *Model) CommitConfiguration(from, to config.Configuration) bool {
continue
}
// This folder exists on both sides. Compare the device lists, as we
// can handle adding a device (but not currently removing one).
// This folder exists on both sides. Settings might have changed.
// Check if anything differs, apart from the label.
toCfgCopy := toCfg
fromCfgCopy := fromCfg
fromCfgCopy.Label = ""
toCfgCopy.Label = ""
fromDevs := mapDevices(fromCfg.DeviceIDs())
toDevs := mapDevices(toCfg.DeviceIDs())
for dev := range fromDevs {
if _, ok := toDevs[dev]; !ok {
// A device was removed. Requires restart.
l.Debugln(m, "requires restart, removing device", dev, "from folder", folderID)
return false
}
}
for dev := range toDevs {
if _, ok := fromDevs[dev]; !ok {
// A device was added. Handle it!
m.fmut.Lock()
m.pmut.Lock()
m.folderCfgs[folderID] = toCfg
m.folderDevices[folderID] = append(m.folderDevices[folderID], dev)
m.deviceFolders[dev] = append(m.deviceFolders[dev], folderID)
// If we already have a connection to this device, we should
// disconnect it so that we start sharing the folder with it.
// We close the underlying connection and let the normal error
// handling kick in to clean up and reconnect.
if conn, ok := m.conn[dev]; ok {
closeRawConn(conn)
}
m.pmut.Unlock()
m.fmut.Unlock()
}
}
// Check if anything else differs, apart from the device list and label.
fromCfg.Devices = nil
toCfg.Devices = nil
fromCfg.Label = ""
toCfg.Label = ""
if !reflect.DeepEqual(fromCfg, toCfg) {
l.Debugln(m, "requires restart, folder", folderID, "configuration differs")
return false
if !reflect.DeepEqual(fromCfgCopy, toCfgCopy) {
m.RestartFolder(toCfg)
}
}
// Removing a device requires restart
toDevs := mapDeviceCfgs(from.Devices)
for _, dev := range from.Devices {
if _, ok := toDevs[dev.DeviceID]; !ok {
l.Debugln(m, "requires restart, device", dev.DeviceID, "was removed")
return false
}
}
// Removing a device. We actually don't need to do anything.
// Because folder config has changed (since the device lists do not match)
// Folders for that had device got "restarted", which involves killing
// connections to all devices that we were sharing the folder with.
// At some point model.Close() will get called for that device which will
// clean residue device state that is not part of any folder.
// Some options don't require restart as those components handle it fine
// by themselves.
@@ -2252,16 +2253,6 @@ func mapDevices(devices []protocol.DeviceID) map[protocol.DeviceID]struct{} {
return m
}
// mapDeviceCfgs returns a map of device ID to nothing for the given slice of
// device configurations.
func mapDeviceCfgs(devices []config.DeviceConfiguration) map[protocol.DeviceID]struct{} {
m := make(map[protocol.DeviceID]struct{}, len(devices))
for _, dev := range devices {
m[dev.DeviceID] = struct{}{}
}
return m
}
func symlinkInvalid(folder string, fi db.FileIntf) bool {
if !symlinks.Supported && fi.IsSymlink() && !fi.IsInvalid() && !fi.IsDeleted() {
symlinkWarning.Do(func() {

187
lib/model/model_test.go
View File

@@ -146,6 +146,7 @@ func genFiles(n int) []protocol.FileInfo {
files[i] = protocol.FileInfo{
Name: fmt.Sprintf("file%d", i),
ModifiedS: t,
Sequence: int64(i + 1),
Blocks: []protocol.BlockInfo{{Offset: 0, Size: 100, Hash: []byte("some hash bytes")}},
}
}
@@ -280,15 +281,7 @@ func BenchmarkRequest(b *testing.B) {
m.ScanFolder("default")
const n = 1000
files := make([]protocol.FileInfo, n)
t := time.Now().Unix()
for i := 0; i < n; i++ {
files[i] = protocol.FileInfo{
Name: fmt.Sprintf("file%d", i),
ModifiedS: t,
Blocks: []protocol.BlockInfo{{Offset: 0, Size: 100, Hash: []byte("some hash bytes")}},
}
}
files := genFiles(n)
fc := &FakeConnection{
id: device1,
@@ -1482,6 +1475,175 @@ func TestIssue2782(t *testing.T) {
}
}
func TestIndexesForUnknownDevicesDropped(t *testing.T) {
dbi := db.OpenMemory()
files := db.NewFileSet("default", dbi)
files.Replace(device1, genFiles(1))
files.Replace(device2, genFiles(1))
if len(files.ListDevices()) != 2 {
t.Error("expected two devices")
}
m := NewModel(defaultConfig, protocol.LocalDeviceID, "device", "syncthing", "dev", dbi, nil)
m.AddFolder(defaultFolderConfig)
m.StartFolder("default")
// Remote sequence is cached, hence need to recreated.
files = db.NewFileSet("default", dbi)
if len(files.ListDevices()) != 1 {
t.Error("Expected one device")
}
}
func TestSharedWithClearedOnDisconnect(t *testing.T) {
dbi := db.OpenMemory()
fcfg := config.NewFolderConfiguration("default", "testdata")
fcfg.Devices = []config.FolderDeviceConfiguration{
{DeviceID: device1},
{DeviceID: device2},
}
cfg := config.Configuration{
Folders: []config.FolderConfiguration{fcfg},
Devices: []config.DeviceConfiguration{
config.NewDeviceConfiguration(device1, "device1"),
config.NewDeviceConfiguration(device2, "device2"),
},
Options: config.OptionsConfiguration{
// Don't remove temporaries directly on startup
KeepTemporariesH: 1,
},
}
wcfg := config.Wrap("/tmp/test", cfg)
d2c := &fakeConn{}
m := NewModel(wcfg, protocol.LocalDeviceID, "device", "syncthing", "dev", dbi, nil)
m.AddFolder(fcfg)
m.StartFolder(fcfg.ID)
m.ServeBackground()
m.AddConnection(connections.Connection{
IntermediateConnection: connections.IntermediateConnection{
Conn: tls.Client(&fakeConn{}, nil),
Type: "foo",
Priority: 10,
},
Connection: &FakeConnection{
id: device1,
},
}, protocol.HelloResult{})
m.AddConnection(connections.Connection{
IntermediateConnection: connections.IntermediateConnection{
Conn: tls.Client(d2c, nil),
Type: "foo",
Priority: 10,
},
Connection: &FakeConnection{
id: device2,
},
}, protocol.HelloResult{})
m.ClusterConfig(device1, protocol.ClusterConfig{
Folders: []protocol.Folder{
{
ID: "default",
Devices: []protocol.Device{
{ID: device1[:]},
{ID: device2[:]},
},
},
},
})
m.ClusterConfig(device2, protocol.ClusterConfig{
Folders: []protocol.Folder{
{
ID: "default",
Devices: []protocol.Device{
{ID: device1[:]},
{ID: device2[:]},
},
},
},
})
if !m.folderSharedWith("default", device1) {
t.Error("not shared with device1")
}
if !m.folderSharedWith("default", device2) {
t.Error("not shared with device2")
}
if d2c.closed {
t.Error("conn already closed")
}
cfg = cfg.Copy()
cfg.Devices = cfg.Devices[:1]
if err := wcfg.Replace(cfg); err != nil {
t.Error(err)
}
time.Sleep(100 * time.Millisecond) // Committer notification happens in a separate routine
if !m.folderSharedWith("default", device1) {
t.Error("not shared with device1")
}
if m.folderSharedWith("default", device2) { // checks m.deviceFolders
t.Error("shared with device2")
}
if !d2c.closed {
t.Error("connection not closed")
}
if _, ok := wcfg.Devices()[device2]; ok {
t.Error("device still in config")
}
fdevs, ok := m.folderDevices["default"]
if !ok {
t.Error("folder missing?")
}
for _, id := range fdevs {
if id == device2 {
t.Error("still there")
}
}
if _, ok := m.conn[device2]; !ok {
t.Error("conn missing early")
}
if _, ok := m.helloMessages[device2]; !ok {
t.Error("hello missing early")
}
if _, ok := m.deviceDownloads[device2]; !ok {
t.Error("downloads missing early")
}
m.Close(device2, fmt.Errorf("foo"))
if _, ok := m.conn[device2]; ok {
t.Error("conn not missing")
}
if _, ok := m.helloMessages[device2]; ok {
t.Error("hello not missing")
}
if _, ok := m.deviceDownloads[device2]; ok {
t.Error("downloads not missing")
}
}
type fakeAddr struct{}
func (fakeAddr) Network() string {
@@ -1492,9 +1654,12 @@ func (fakeAddr) String() string {
return "address"
}
type fakeConn struct{}
type fakeConn struct {
closed bool
}
func (fakeConn) Close() error {
func (c *fakeConn) Close() error {
c.closed = true
return nil
}