* cmd/syncthing, lib/gui: Separate gui into own package (ref #4085) * fix tests * Don't use main as interface name (make old go happy) * gui->api * don't leak state via locations and use in-tree config * let api (un-)subscribe to config * interface naming and exporting * lib/ur * fix tests and lib/foldersummary * shorter URVersion and ur debug fix * review * model.JsonCompletion(FolderCompletion) -> FolderCompletion.Map() * rename debug facility https -> api * folder summaries in model * disassociate unrelated constants * fix merge fail * missing id assignement
This commit is contained in:
Simon Frei
committed by
Audrius Butkevicius
Audrius Butkevicius
parent
d4e81fff8a
commit
b50039a920
312
lib/model/folder_summary.go
Normal file
312
lib/model/folder_summary.go
Normal file
@@ -0,0 +1,312 @@
|
||||
// Copyright (C) 2015 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 https://mozilla.org/MPL/2.0/.
|
||||
|
||||
package model
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/syncthing/syncthing/lib/config"
|
||||
"github.com/syncthing/syncthing/lib/events"
|
||||
"github.com/syncthing/syncthing/lib/protocol"
|
||||
"github.com/syncthing/syncthing/lib/sync"
|
||||
"github.com/thejerf/suture"
|
||||
)
|
||||
|
||||
const minSummaryInterval = time.Minute
|
||||
|
||||
type FolderSummaryService interface {
|
||||
suture.Service
|
||||
Summary(folder string) (map[string]interface{}, error)
|
||||
OnEventRequest()
|
||||
}
|
||||
|
||||
// The folderSummaryService adds summary information events (FolderSummary and
|
||||
// FolderCompletion) into the event stream at certain intervals.
|
||||
type folderSummaryService struct {
|
||||
*suture.Supervisor
|
||||
|
||||
cfg config.Wrapper
|
||||
model Model
|
||||
id protocol.DeviceID
|
||||
stop chan struct{}
|
||||
immediate chan string
|
||||
|
||||
// For keeping track of folders to recalculate for
|
||||
foldersMut sync.Mutex
|
||||
folders map[string]struct{}
|
||||
|
||||
// For keeping track of when the last event request on the API was
|
||||
lastEventReq time.Time
|
||||
lastEventReqMut sync.Mutex
|
||||
}
|
||||
|
||||
func NewFolderSummaryService(cfg config.Wrapper, m Model, id protocol.DeviceID) FolderSummaryService {
|
||||
service := &folderSummaryService{
|
||||
Supervisor: suture.New("folderSummaryService", suture.Spec{
|
||||
PassThroughPanics: true,
|
||||
}),
|
||||
cfg: cfg,
|
||||
model: m,
|
||||
id: id,
|
||||
stop: make(chan struct{}),
|
||||
immediate: make(chan string),
|
||||
folders: make(map[string]struct{}),
|
||||
foldersMut: sync.NewMutex(),
|
||||
lastEventReqMut: sync.NewMutex(),
|
||||
}
|
||||
|
||||
service.Add(serviceFunc(service.listenForUpdates))
|
||||
service.Add(serviceFunc(service.calculateSummaries))
|
||||
|
||||
return service
|
||||
}
|
||||
|
||||
func (c *folderSummaryService) Stop() {
|
||||
c.Supervisor.Stop()
|
||||
close(c.stop)
|
||||
}
|
||||
|
||||
func (c *folderSummaryService) String() string {
|
||||
return fmt.Sprintf("FolderSummaryService@%p", c)
|
||||
}
|
||||
|
||||
func (c *folderSummaryService) Summary(folder string) (map[string]interface{}, error) {
|
||||
var res = make(map[string]interface{})
|
||||
|
||||
errors, err := c.model.FolderErrors(folder)
|
||||
if err != nil && err != ErrFolderPaused {
|
||||
// Stats from the db can still be obtained if the folder is just paused
|
||||
return nil, err
|
||||
}
|
||||
res["errors"] = len(errors)
|
||||
res["pullErrors"] = len(errors) // deprecated
|
||||
|
||||
res["invalid"] = "" // Deprecated, retains external API for now
|
||||
|
||||
global := c.model.GlobalSize(folder)
|
||||
res["globalFiles"], res["globalDirectories"], res["globalSymlinks"], res["globalDeleted"], res["globalBytes"], res["globalTotalItems"] = global.Files, global.Directories, global.Symlinks, global.Deleted, global.Bytes, global.TotalItems()
|
||||
|
||||
local := c.model.LocalSize(folder)
|
||||
res["localFiles"], res["localDirectories"], res["localSymlinks"], res["localDeleted"], res["localBytes"], res["localTotalItems"] = local.Files, local.Directories, local.Symlinks, local.Deleted, local.Bytes, local.TotalItems()
|
||||
|
||||
need := c.model.NeedSize(folder)
|
||||
res["needFiles"], res["needDirectories"], res["needSymlinks"], res["needDeletes"], res["needBytes"], res["needTotalItems"] = need.Files, need.Directories, need.Symlinks, need.Deleted, need.Bytes, need.TotalItems()
|
||||
|
||||
if c.cfg.Folders()[folder].Type == config.FolderTypeReceiveOnly {
|
||||
// Add statistics for things that have changed locally in a receive
|
||||
// only folder.
|
||||
ro := c.model.ReceiveOnlyChangedSize(folder)
|
||||
res["receiveOnlyChangedFiles"] = ro.Files
|
||||
res["receiveOnlyChangedDirectories"] = ro.Directories
|
||||
res["receiveOnlyChangedSymlinks"] = ro.Symlinks
|
||||
res["receiveOnlyChangedDeletes"] = ro.Deleted
|
||||
res["receiveOnlyChangedBytes"] = ro.Bytes
|
||||
res["receiveOnlyTotalItems"] = ro.TotalItems()
|
||||
}
|
||||
|
||||
res["inSyncFiles"], res["inSyncBytes"] = global.Files-need.Files, global.Bytes-need.Bytes
|
||||
|
||||
res["state"], res["stateChanged"], err = c.model.State(folder)
|
||||
if err != nil {
|
||||
res["error"] = err.Error()
|
||||
}
|
||||
|
||||
ourSeq, _ := c.model.CurrentSequence(folder)
|
||||
remoteSeq, _ := c.model.RemoteSequence(folder)
|
||||
|
||||
res["version"] = ourSeq + remoteSeq // legacy
|
||||
res["sequence"] = ourSeq + remoteSeq // new name
|
||||
|
||||
ignorePatterns, _, _ := c.model.GetIgnores(folder)
|
||||
res["ignorePatterns"] = false
|
||||
for _, line := range ignorePatterns {
|
||||
if len(line) > 0 && !strings.HasPrefix(line, "//") {
|
||||
res["ignorePatterns"] = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
err = c.model.WatchError(folder)
|
||||
if err != nil {
|
||||
res["watchError"] = err.Error()
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (c *folderSummaryService) OnEventRequest() {
|
||||
c.lastEventReqMut.Lock()
|
||||
c.lastEventReq = time.Now()
|
||||
c.lastEventReqMut.Unlock()
|
||||
}
|
||||
|
||||
// listenForUpdates subscribes to the event bus and makes note of folders that
|
||||
// need their data recalculated.
|
||||
func (c *folderSummaryService) listenForUpdates() {
|
||||
sub := events.Default.Subscribe(events.LocalIndexUpdated | events.RemoteIndexUpdated | events.StateChanged | events.RemoteDownloadProgress | events.DeviceConnected | events.FolderWatchStateChanged)
|
||||
defer events.Default.Unsubscribe(sub)
|
||||
|
||||
for {
|
||||
// This loop needs to be fast so we don't miss too many events.
|
||||
|
||||
select {
|
||||
case ev := <-sub.C():
|
||||
if ev.Type == events.DeviceConnected {
|
||||
// When a device connects we schedule a refresh of all
|
||||
// folders shared with that device.
|
||||
|
||||
data := ev.Data.(map[string]string)
|
||||
deviceID, _ := protocol.DeviceIDFromString(data["id"])
|
||||
|
||||
c.foldersMut.Lock()
|
||||
nextFolder:
|
||||
for _, folder := range c.cfg.Folders() {
|
||||
for _, dev := range folder.Devices {
|
||||
if dev.DeviceID == deviceID {
|
||||
c.folders[folder.ID] = struct{}{}
|
||||
continue nextFolder
|
||||
}
|
||||
}
|
||||
}
|
||||
c.foldersMut.Unlock()
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
// The other events all have a "folder" attribute that they
|
||||
// affect. Whenever the local or remote index is updated for a
|
||||
// given folder we make a note of it.
|
||||
|
||||
data := ev.Data.(map[string]interface{})
|
||||
folder := data["folder"].(string)
|
||||
|
||||
switch ev.Type {
|
||||
case events.StateChanged:
|
||||
if data["to"].(string) == "idle" && data["from"].(string) == "syncing" {
|
||||
// The folder changed to idle from syncing. We should do an
|
||||
// immediate refresh to update the GUI. The send to
|
||||
// c.immediate must be nonblocking so that we can continue
|
||||
// handling events.
|
||||
|
||||
c.foldersMut.Lock()
|
||||
select {
|
||||
case c.immediate <- folder:
|
||||
delete(c.folders, folder)
|
||||
default:
|
||||
c.folders[folder] = struct{}{}
|
||||
}
|
||||
c.foldersMut.Unlock()
|
||||
}
|
||||
|
||||
default:
|
||||
// This folder needs to be refreshed whenever we do the next
|
||||
// refresh.
|
||||
|
||||
c.foldersMut.Lock()
|
||||
c.folders[folder] = struct{}{}
|
||||
c.foldersMut.Unlock()
|
||||
}
|
||||
|
||||
case <-c.stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// calculateSummaries periodically recalculates folder summaries and
|
||||
// completion percentage, and sends the results on the event bus.
|
||||
func (c *folderSummaryService) calculateSummaries() {
|
||||
const pumpInterval = 2 * time.Second
|
||||
pump := time.NewTimer(pumpInterval)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-pump.C:
|
||||
t0 := time.Now()
|
||||
for _, folder := range c.foldersToHandle() {
|
||||
c.sendSummary(folder)
|
||||
}
|
||||
|
||||
// We don't want to spend all our time calculating summaries. Lets
|
||||
// set an arbitrary limit at not spending more than about 30% of
|
||||
// our time here...
|
||||
wait := 2*time.Since(t0) + pumpInterval
|
||||
pump.Reset(wait)
|
||||
|
||||
case folder := <-c.immediate:
|
||||
c.sendSummary(folder)
|
||||
|
||||
case <-c.stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// foldersToHandle returns the list of folders needing a summary update, and
|
||||
// clears the list.
|
||||
func (c *folderSummaryService) foldersToHandle() []string {
|
||||
// We only recalculate summaries if someone is listening to events
|
||||
// (a request to /rest/events has been made within the last
|
||||
// pingEventInterval).
|
||||
|
||||
c.lastEventReqMut.Lock()
|
||||
last := c.lastEventReq
|
||||
c.lastEventReqMut.Unlock()
|
||||
if time.Since(last) > minSummaryInterval {
|
||||
return nil
|
||||
}
|
||||
|
||||
c.foldersMut.Lock()
|
||||
res := make([]string, 0, len(c.folders))
|
||||
for folder := range c.folders {
|
||||
res = append(res, folder)
|
||||
delete(c.folders, folder)
|
||||
}
|
||||
c.foldersMut.Unlock()
|
||||
return res
|
||||
}
|
||||
|
||||
// sendSummary send the summary events for a single folder
|
||||
func (c *folderSummaryService) sendSummary(folder string) {
|
||||
// The folder summary contains how many bytes, files etc
|
||||
// are in the folder and how in sync we are.
|
||||
data, err := c.Summary(folder)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
events.Default.Log(events.FolderSummary, map[string]interface{}{
|
||||
"folder": folder,
|
||||
"summary": data,
|
||||
})
|
||||
|
||||
for _, devCfg := range c.cfg.Folders()[folder].Devices {
|
||||
if devCfg.DeviceID.Equals(c.id) {
|
||||
// We already know about ourselves.
|
||||
continue
|
||||
}
|
||||
if _, ok := c.model.Connection(devCfg.DeviceID); !ok {
|
||||
// We're not interested in disconnected devices.
|
||||
continue
|
||||
}
|
||||
|
||||
// Get completion percentage of this folder for the
|
||||
// remote device.
|
||||
comp := c.model.Completion(devCfg.DeviceID, folder).Map()
|
||||
comp["folder"] = folder
|
||||
comp["device"] = devCfg.DeviceID.String()
|
||||
events.Default.Log(events.FolderCompletion, comp)
|
||||
}
|
||||
}
|
||||
|
||||
// serviceFunc wraps a function to create a suture.Service without stop
|
||||
// functionality.
|
||||
type serviceFunc func()
|
||||
|
||||
func (f serviceFunc) Serve() { f() }
|
||||
func (f serviceFunc) Stop() {}
|
||||
Reference in New Issue
Block a user