Refactor model into separate package
This commit is contained in:
Jakob Borg
101
model/blockqueue.go
Normal file
101
model/blockqueue.go
Normal file
@@ -0,0 +1,101 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/calmh/syncthing/scanner"
|
||||
)
|
||||
|
||||
type bqAdd struct {
|
||||
file scanner.File
|
||||
have []scanner.Block
|
||||
need []scanner.Block
|
||||
}
|
||||
|
||||
type bqBlock struct {
|
||||
file scanner.File
|
||||
block scanner.Block // get this block from the network
|
||||
copy []scanner.Block // copy these blocks from the old version of the file
|
||||
last bool
|
||||
}
|
||||
|
||||
type blockQueue struct {
|
||||
inbox chan bqAdd
|
||||
outbox chan bqBlock
|
||||
|
||||
queued []bqBlock
|
||||
qlen uint32
|
||||
}
|
||||
|
||||
func newBlockQueue() *blockQueue {
|
||||
q := &blockQueue{
|
||||
inbox: make(chan bqAdd),
|
||||
outbox: make(chan bqBlock),
|
||||
}
|
||||
go q.run()
|
||||
return q
|
||||
}
|
||||
|
||||
func (q *blockQueue) addBlock(a bqAdd) {
|
||||
// If we already have it queued, return
|
||||
for _, b := range q.queued {
|
||||
if b.file.Name == a.file.Name {
|
||||
return
|
||||
}
|
||||
}
|
||||
if len(a.have) > 0 {
|
||||
// First queue a copy operation
|
||||
q.queued = append(q.queued, bqBlock{
|
||||
file: a.file,
|
||||
copy: a.have,
|
||||
})
|
||||
}
|
||||
// Queue the needed blocks individually
|
||||
l := len(a.need)
|
||||
for i, b := range a.need {
|
||||
q.queued = append(q.queued, bqBlock{
|
||||
file: a.file,
|
||||
block: b,
|
||||
last: i == l-1,
|
||||
})
|
||||
}
|
||||
|
||||
if l == 0 {
|
||||
// If we didn't have anything to fetch, queue an empty block with the "last" flag set to close the file.
|
||||
q.queued = append(q.queued, bqBlock{
|
||||
file: a.file,
|
||||
last: true,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (q *blockQueue) run() {
|
||||
for {
|
||||
if len(q.queued) == 0 {
|
||||
q.addBlock(<-q.inbox)
|
||||
} else {
|
||||
next := q.queued[0]
|
||||
select {
|
||||
case a := <-q.inbox:
|
||||
q.addBlock(a)
|
||||
case q.outbox <- next:
|
||||
q.queued = q.queued[1:]
|
||||
}
|
||||
}
|
||||
atomic.StoreUint32(&q.qlen, uint32(len(q.queued)))
|
||||
}
|
||||
}
|
||||
|
||||
func (q *blockQueue) put(a bqAdd) {
|
||||
q.inbox <- a
|
||||
}
|
||||
|
||||
func (q *blockQueue) get() bqBlock {
|
||||
return <-q.outbox
|
||||
}
|
||||
|
||||
func (q *blockQueue) empty() bool {
|
||||
var l uint32
|
||||
atomic.LoadUint32(&l)
|
||||
return l == 0
|
||||
}
|
||||
13
model/debug.go
Normal file
13
model/debug.go
Normal file
@@ -0,0 +1,13 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"github.com/calmh/syncthing/logger"
|
||||
)
|
||||
|
||||
var (
|
||||
debug = strings.Contains(os.Getenv("STTRACE"), "model") || os.Getenv("STTRACE") == "all"
|
||||
l = logger.DefaultLogger
|
||||
)
|
||||
743
model/model.go
Normal file
743
model/model.go
Normal file
@@ -0,0 +1,743 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"compress/gzip"
|
||||
"crypto/sha1"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/calmh/syncthing/buffers"
|
||||
"github.com/calmh/syncthing/cid"
|
||||
"github.com/calmh/syncthing/config"
|
||||
"github.com/calmh/syncthing/files"
|
||||
"github.com/calmh/syncthing/lamport"
|
||||
"github.com/calmh/syncthing/protocol"
|
||||
"github.com/calmh/syncthing/scanner"
|
||||
)
|
||||
|
||||
type repoState int
|
||||
|
||||
const (
|
||||
RepoIdle repoState = iota
|
||||
RepoScanning
|
||||
RepoSyncing
|
||||
RepoCleaning
|
||||
)
|
||||
|
||||
type Model struct {
|
||||
indexDir string
|
||||
cfg *config.Configuration
|
||||
|
||||
clientName string
|
||||
clientVersion string
|
||||
|
||||
repoDirs map[string]string // repo -> dir
|
||||
repoFiles map[string]*files.Set // repo -> files
|
||||
repoNodes map[string][]string // repo -> nodeIDs
|
||||
nodeRepos map[string][]string // nodeID -> repos
|
||||
repoState map[string]repoState // repo -> state
|
||||
suppressor map[string]*suppressor // repo -> suppressor
|
||||
rmut sync.RWMutex // protects the above
|
||||
|
||||
cm *cid.Map
|
||||
|
||||
protoConn map[string]protocol.Connection
|
||||
rawConn map[string]io.Closer
|
||||
nodeVer map[string]string
|
||||
pmut sync.RWMutex // protects protoConn and rawConn
|
||||
|
||||
sup suppressor
|
||||
|
||||
addedRepo bool
|
||||
started bool
|
||||
}
|
||||
|
||||
var (
|
||||
ErrNoSuchFile = errors.New("no such file")
|
||||
ErrInvalid = errors.New("file is invalid")
|
||||
)
|
||||
|
||||
// NewModel creates and starts a new model. The model starts in read-only mode,
|
||||
// where it sends index information to connected peers and responds to requests
|
||||
// for file data without altering the local repository in any way.
|
||||
func NewModel(indexDir string, cfg *config.Configuration, clientName, clientVersion string) *Model {
|
||||
m := &Model{
|
||||
indexDir: indexDir,
|
||||
cfg: cfg,
|
||||
clientName: clientName,
|
||||
clientVersion: clientVersion,
|
||||
repoDirs: make(map[string]string),
|
||||
repoFiles: make(map[string]*files.Set),
|
||||
repoNodes: make(map[string][]string),
|
||||
nodeRepos: make(map[string][]string),
|
||||
repoState: make(map[string]repoState),
|
||||
suppressor: make(map[string]*suppressor),
|
||||
cm: cid.NewMap(),
|
||||
protoConn: make(map[string]protocol.Connection),
|
||||
rawConn: make(map[string]io.Closer),
|
||||
nodeVer: make(map[string]string),
|
||||
sup: suppressor{threshold: int64(cfg.Options.MaxChangeKbps)},
|
||||
}
|
||||
|
||||
go m.broadcastIndexLoop()
|
||||
return m
|
||||
}
|
||||
|
||||
// StartRW starts read/write processing on the current model. When in
|
||||
// read/write mode the model will attempt to keep in sync with the cluster by
|
||||
// pulling needed files from peer nodes.
|
||||
func (m *Model) StartRepoRW(repo string, threads int) {
|
||||
m.rmut.RLock()
|
||||
defer m.rmut.RUnlock()
|
||||
|
||||
if dir, ok := m.repoDirs[repo]; !ok {
|
||||
panic("cannot start without repo")
|
||||
} else {
|
||||
newPuller(repo, dir, m, threads, m.cfg)
|
||||
}
|
||||
}
|
||||
|
||||
// StartRO starts read only processing on the current model. When in
|
||||
// read only mode the model will announce files to the cluster but not
|
||||
// pull in any external changes.
|
||||
func (m *Model) StartRepoRO(repo string) {
|
||||
m.StartRepoRW(repo, 0) // zero threads => read only
|
||||
}
|
||||
|
||||
type ConnectionInfo struct {
|
||||
protocol.Statistics
|
||||
Address string
|
||||
ClientVersion string
|
||||
Completion int
|
||||
}
|
||||
|
||||
// ConnectionStats returns a map with connection statistics for each connected node.
|
||||
func (m *Model) ConnectionStats() map[string]ConnectionInfo {
|
||||
type remoteAddrer interface {
|
||||
RemoteAddr() net.Addr
|
||||
}
|
||||
|
||||
m.pmut.RLock()
|
||||
m.rmut.RLock()
|
||||
|
||||
var res = make(map[string]ConnectionInfo)
|
||||
for node, conn := range m.protoConn {
|
||||
ci := ConnectionInfo{
|
||||
Statistics: conn.Statistics(),
|
||||
ClientVersion: m.nodeVer[node],
|
||||
}
|
||||
if nc, ok := m.rawConn[node].(remoteAddrer); ok {
|
||||
ci.Address = nc.RemoteAddr().String()
|
||||
}
|
||||
|
||||
var tot int64
|
||||
var have int64
|
||||
|
||||
for _, repo := range m.nodeRepos[node] {
|
||||
for _, f := range m.repoFiles[repo].Global() {
|
||||
if f.Flags&protocol.FlagDeleted == 0 {
|
||||
tot += f.Size
|
||||
have += f.Size
|
||||
}
|
||||
}
|
||||
|
||||
for _, f := range m.repoFiles[repo].Need(m.cm.Get(node)) {
|
||||
if f.Flags&protocol.FlagDeleted == 0 {
|
||||
have -= f.Size
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ci.Completion = 100
|
||||
if tot != 0 {
|
||||
ci.Completion = int(100 * have / tot)
|
||||
}
|
||||
|
||||
res[node] = ci
|
||||
}
|
||||
|
||||
m.rmut.RUnlock()
|
||||
m.pmut.RUnlock()
|
||||
|
||||
return res
|
||||
}
|
||||
|
||||
func sizeOf(fs []scanner.File) (files, deleted int, bytes int64) {
|
||||
for _, f := range fs {
|
||||
if f.Flags&protocol.FlagDeleted == 0 {
|
||||
files++
|
||||
bytes += f.Size
|
||||
} else {
|
||||
deleted++
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// GlobalSize returns the number of files, deleted files and total bytes for all
|
||||
// files in the global model.
|
||||
func (m *Model) GlobalSize(repo string) (files, deleted int, bytes int64) {
|
||||
m.rmut.RLock()
|
||||
defer m.rmut.RUnlock()
|
||||
if rf, ok := m.repoFiles[repo]; ok {
|
||||
return sizeOf(rf.Global())
|
||||
}
|
||||
return 0, 0, 0
|
||||
}
|
||||
|
||||
// LocalSize returns the number of files, deleted files and total bytes for all
|
||||
// files in the local repository.
|
||||
func (m *Model) LocalSize(repo string) (files, deleted int, bytes int64) {
|
||||
m.rmut.RLock()
|
||||
defer m.rmut.RUnlock()
|
||||
if rf, ok := m.repoFiles[repo]; ok {
|
||||
return sizeOf(rf.Have(cid.LocalID))
|
||||
}
|
||||
return 0, 0, 0
|
||||
}
|
||||
|
||||
// NeedFiles returns the list of currently needed files and the total size.
|
||||
func (m *Model) NeedSize(repo string) (files int, bytes int64) {
|
||||
var nf = m.NeedFilesRepo(repo)
|
||||
|
||||
for _, f := range nf {
|
||||
bytes += f.Size
|
||||
}
|
||||
|
||||
return len(nf), bytes
|
||||
}
|
||||
|
||||
// NeedFiles returns the list of currently needed files and the total size.
|
||||
func (m *Model) NeedFilesRepo(repo string) []scanner.File {
|
||||
m.rmut.RLock()
|
||||
defer m.rmut.RUnlock()
|
||||
if rf, ok := m.repoFiles[repo]; ok {
|
||||
return rf.Need(cid.LocalID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Index is called when a new node is connected and we receive their full index.
|
||||
// Implements the protocol.Model interface.
|
||||
func (m *Model) Index(nodeID string, repo string, fs []protocol.FileInfo) {
|
||||
if debug {
|
||||
l.Debugf("IDX(in): %s / %q: %d files", nodeID, repo, len(fs))
|
||||
}
|
||||
|
||||
var files = make([]scanner.File, len(fs))
|
||||
for i := range fs {
|
||||
lamport.Default.Tick(fs[i].Version)
|
||||
files[i] = fileFromFileInfo(fs[i])
|
||||
}
|
||||
|
||||
id := m.cm.Get(nodeID)
|
||||
m.rmut.RLock()
|
||||
if r, ok := m.repoFiles[repo]; ok {
|
||||
r.Replace(id, files)
|
||||
} else {
|
||||
l.Warnf("Index from %s for nonexistant repo %q; dropping", nodeID, repo)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
}
|
||||
|
||||
// IndexUpdate is called for incremental updates to connected nodes' indexes.
|
||||
// Implements the protocol.Model interface.
|
||||
func (m *Model) IndexUpdate(nodeID string, repo string, fs []protocol.FileInfo) {
|
||||
if debug {
|
||||
l.Debugf("IDXUP(in): %s / %q: %d files", nodeID, repo, len(fs))
|
||||
}
|
||||
|
||||
var files = make([]scanner.File, len(fs))
|
||||
for i := range fs {
|
||||
lamport.Default.Tick(fs[i].Version)
|
||||
files[i] = fileFromFileInfo(fs[i])
|
||||
}
|
||||
|
||||
id := m.cm.Get(nodeID)
|
||||
m.rmut.RLock()
|
||||
if r, ok := m.repoFiles[repo]; ok {
|
||||
r.Update(id, files)
|
||||
} else {
|
||||
l.Warnf("Index update from %s for nonexistant repo %q; dropping", nodeID, repo)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
}
|
||||
|
||||
func (m *Model) ClusterConfig(nodeID string, config protocol.ClusterConfigMessage) {
|
||||
compErr := compareClusterConfig(m.clusterConfig(nodeID), config)
|
||||
if debug {
|
||||
l.Debugf("ClusterConfig: %s: %#v", nodeID, config)
|
||||
l.Debugf(" ... compare: %s: %v", nodeID, compErr)
|
||||
}
|
||||
|
||||
if compErr != nil {
|
||||
l.Warnf("%s: %v", nodeID, compErr)
|
||||
m.Close(nodeID, compErr)
|
||||
}
|
||||
|
||||
m.pmut.Lock()
|
||||
if config.ClientName == "syncthing" {
|
||||
m.nodeVer[nodeID] = config.ClientVersion
|
||||
} else {
|
||||
m.nodeVer[nodeID] = config.ClientName + " " + config.ClientVersion
|
||||
}
|
||||
m.pmut.Unlock()
|
||||
}
|
||||
|
||||
// Close removes the peer from the model and closes the underlying connection if possible.
|
||||
// Implements the protocol.Model interface.
|
||||
func (m *Model) Close(node string, err error) {
|
||||
if debug {
|
||||
l.Debugf("%s: %v", node, err)
|
||||
}
|
||||
|
||||
if err != io.EOF {
|
||||
l.Warnf("Connection to %s closed: %v", node, err)
|
||||
} else if _, ok := err.(ClusterConfigMismatch); ok {
|
||||
l.Warnf("Connection to %s closed: %v", node, err)
|
||||
}
|
||||
|
||||
cid := m.cm.Get(node)
|
||||
m.rmut.RLock()
|
||||
for _, repo := range m.nodeRepos[node] {
|
||||
m.repoFiles[repo].Replace(cid, nil)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
m.cm.Clear(node)
|
||||
|
||||
m.pmut.Lock()
|
||||
conn, ok := m.rawConn[node]
|
||||
if ok {
|
||||
conn.Close()
|
||||
}
|
||||
delete(m.protoConn, node)
|
||||
delete(m.rawConn, node)
|
||||
delete(m.nodeVer, node)
|
||||
m.pmut.Unlock()
|
||||
}
|
||||
|
||||
// Request returns the specified data segment by reading it from local disk.
|
||||
// Implements the protocol.Model interface.
|
||||
func (m *Model) Request(nodeID, repo, name string, offset int64, size int) ([]byte, error) {
|
||||
// Verify that the requested file exists in the local model.
|
||||
m.rmut.RLock()
|
||||
r, ok := m.repoFiles[repo]
|
||||
m.rmut.RUnlock()
|
||||
|
||||
if !ok {
|
||||
l.Warnf("Request from %s for file %s in nonexistent repo %q", nodeID, name, repo)
|
||||
return nil, ErrNoSuchFile
|
||||
}
|
||||
|
||||
lf := r.Get(cid.LocalID, name)
|
||||
if lf.Suppressed || lf.Flags&protocol.FlagDeleted != 0 {
|
||||
return nil, ErrInvalid
|
||||
}
|
||||
|
||||
if offset > lf.Size {
|
||||
if debug {
|
||||
l.Debugf("REQ(in; nonexistent): %s: %q o=%d s=%d", nodeID, name, offset, size)
|
||||
}
|
||||
return nil, ErrNoSuchFile
|
||||
}
|
||||
|
||||
if debug && nodeID != "<local>" {
|
||||
l.Debugf("REQ(in): %s: %q / %q o=%d s=%d", nodeID, repo, name, offset, size)
|
||||
}
|
||||
m.rmut.RLock()
|
||||
fn := filepath.Join(m.repoDirs[repo], name)
|
||||
m.rmut.RUnlock()
|
||||
fd, err := os.Open(fn) // XXX: Inefficient, should cache fd?
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer fd.Close()
|
||||
|
||||
buf := buffers.Get(int(size))
|
||||
_, err = fd.ReadAt(buf, offset)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return buf, nil
|
||||
}
|
||||
|
||||
// ReplaceLocal replaces the local repository index with the given list of files.
|
||||
func (m *Model) ReplaceLocal(repo string, fs []scanner.File) {
|
||||
m.rmut.RLock()
|
||||
m.repoFiles[repo].ReplaceWithDelete(cid.LocalID, fs)
|
||||
m.rmut.RUnlock()
|
||||
}
|
||||
|
||||
func (m *Model) SeedLocal(repo string, fs []protocol.FileInfo) {
|
||||
var sfs = make([]scanner.File, len(fs))
|
||||
for i := 0; i < len(fs); i++ {
|
||||
lamport.Default.Tick(fs[i].Version)
|
||||
sfs[i] = fileFromFileInfo(fs[i])
|
||||
}
|
||||
|
||||
m.rmut.RLock()
|
||||
m.repoFiles[repo].Replace(cid.LocalID, sfs)
|
||||
m.rmut.RUnlock()
|
||||
}
|
||||
|
||||
func (m *Model) CurrentRepoFile(repo string, file string) scanner.File {
|
||||
m.rmut.RLock()
|
||||
f := m.repoFiles[repo].Get(cid.LocalID, file)
|
||||
m.rmut.RUnlock()
|
||||
return f
|
||||
}
|
||||
|
||||
func (m *Model) CurrentGlobalFile(repo string, file string) scanner.File {
|
||||
m.rmut.RLock()
|
||||
f := m.repoFiles[repo].GetGlobal(file)
|
||||
m.rmut.RUnlock()
|
||||
return f
|
||||
}
|
||||
|
||||
type cFiler struct {
|
||||
m *Model
|
||||
r string
|
||||
}
|
||||
|
||||
// Implements scanner.CurrentFiler
|
||||
func (cf cFiler) CurrentFile(file string) scanner.File {
|
||||
return cf.m.CurrentRepoFile(cf.r, file)
|
||||
}
|
||||
|
||||
// ConnectedTo returns true if we are connected to the named node.
|
||||
func (m *Model) ConnectedTo(nodeID string) bool {
|
||||
m.pmut.RLock()
|
||||
_, ok := m.protoConn[nodeID]
|
||||
m.pmut.RUnlock()
|
||||
return ok
|
||||
}
|
||||
|
||||
// AddConnection adds a new peer connection to the model. An initial index will
|
||||
// be sent to the connected peer, thereafter index updates whenever the local
|
||||
// repository changes.
|
||||
func (m *Model) AddConnection(rawConn io.Closer, protoConn protocol.Connection) {
|
||||
nodeID := protoConn.ID()
|
||||
m.pmut.Lock()
|
||||
if _, ok := m.protoConn[nodeID]; ok {
|
||||
panic("add existing node")
|
||||
}
|
||||
m.protoConn[nodeID] = protoConn
|
||||
if _, ok := m.rawConn[nodeID]; ok {
|
||||
panic("add existing node")
|
||||
}
|
||||
m.rawConn[nodeID] = rawConn
|
||||
m.pmut.Unlock()
|
||||
|
||||
cm := m.clusterConfig(nodeID)
|
||||
protoConn.ClusterConfig(cm)
|
||||
|
||||
var idxToSend = make(map[string][]protocol.FileInfo)
|
||||
|
||||
m.rmut.RLock()
|
||||
for _, repo := range m.nodeRepos[nodeID] {
|
||||
idxToSend[repo] = m.protocolIndex(repo)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
|
||||
go func() {
|
||||
for repo, idx := range idxToSend {
|
||||
if debug {
|
||||
l.Debugf("IDX(out/initial): %s: %q: %d files", nodeID, repo, len(idx))
|
||||
}
|
||||
protoConn.Index(repo, idx)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// protocolIndex returns the current local index in protocol data types.
|
||||
func (m *Model) protocolIndex(repo string) []protocol.FileInfo {
|
||||
var index []protocol.FileInfo
|
||||
|
||||
fs := m.repoFiles[repo].Have(cid.LocalID)
|
||||
|
||||
for _, f := range fs {
|
||||
mf := fileInfoFromFile(f)
|
||||
if debug {
|
||||
var flagComment string
|
||||
if mf.Flags&protocol.FlagDeleted != 0 {
|
||||
flagComment = " (deleted)"
|
||||
}
|
||||
l.Debugf("IDX(out): %q/%q m=%d f=%o%s v=%d (%d blocks)", repo, mf.Name, mf.Modified, mf.Flags, flagComment, mf.Version, len(mf.Blocks))
|
||||
}
|
||||
index = append(index, mf)
|
||||
}
|
||||
|
||||
return index
|
||||
}
|
||||
|
||||
func (m *Model) updateLocal(repo string, f scanner.File) {
|
||||
m.rmut.RLock()
|
||||
m.repoFiles[repo].Update(cid.LocalID, []scanner.File{f})
|
||||
m.rmut.RUnlock()
|
||||
}
|
||||
|
||||
func (m *Model) requestGlobal(nodeID, repo, name string, offset int64, size int, hash []byte) ([]byte, error) {
|
||||
m.pmut.RLock()
|
||||
nc, ok := m.protoConn[nodeID]
|
||||
m.pmut.RUnlock()
|
||||
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("requestGlobal: no such node: %s", nodeID)
|
||||
}
|
||||
|
||||
if debug {
|
||||
l.Debugf("REQ(out): %s: %q / %q o=%d s=%d h=%x", nodeID, repo, name, offset, size, hash)
|
||||
}
|
||||
|
||||
return nc.Request(repo, name, offset, size)
|
||||
}
|
||||
|
||||
func (m *Model) broadcastIndexLoop() {
|
||||
var lastChange = map[string]uint64{}
|
||||
for {
|
||||
time.Sleep(5 * time.Second)
|
||||
|
||||
m.pmut.RLock()
|
||||
m.rmut.RLock()
|
||||
|
||||
for repo, fs := range m.repoFiles {
|
||||
c := fs.Changes(cid.LocalID)
|
||||
if c == lastChange[repo] {
|
||||
continue
|
||||
}
|
||||
lastChange[repo] = c
|
||||
|
||||
idx := m.protocolIndex(repo)
|
||||
m.saveIndex(repo, m.indexDir, idx)
|
||||
|
||||
var indexWg sync.WaitGroup
|
||||
for _, nodeID := range m.repoNodes[repo] {
|
||||
if conn, ok := m.protoConn[nodeID]; ok {
|
||||
indexWg.Add(1)
|
||||
if debug {
|
||||
l.Debugf("IDX(out/loop): %s: %d files", nodeID, len(idx))
|
||||
}
|
||||
go func() {
|
||||
conn.Index(repo, idx)
|
||||
indexWg.Done()
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
indexWg.Wait()
|
||||
}
|
||||
|
||||
m.rmut.RUnlock()
|
||||
m.pmut.RUnlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Model) AddRepo(id, dir string, nodes []config.NodeConfiguration) {
|
||||
if m.started {
|
||||
panic("cannot add repo to started model")
|
||||
}
|
||||
if len(id) == 0 {
|
||||
panic("cannot add empty repo id")
|
||||
}
|
||||
|
||||
m.rmut.Lock()
|
||||
m.repoDirs[id] = dir
|
||||
m.repoFiles[id] = files.NewSet()
|
||||
m.suppressor[id] = &suppressor{threshold: int64(m.cfg.Options.MaxChangeKbps)}
|
||||
|
||||
m.repoNodes[id] = make([]string, len(nodes))
|
||||
for i, node := range nodes {
|
||||
m.repoNodes[id][i] = node.NodeID
|
||||
m.nodeRepos[node.NodeID] = append(m.nodeRepos[node.NodeID], id)
|
||||
}
|
||||
|
||||
m.addedRepo = true
|
||||
m.rmut.Unlock()
|
||||
}
|
||||
|
||||
func (m *Model) ScanRepos() {
|
||||
m.rmut.RLock()
|
||||
var repos = make([]string, 0, len(m.repoDirs))
|
||||
for repo := range m.repoDirs {
|
||||
repos = append(repos, repo)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(len(repos))
|
||||
for _, repo := range repos {
|
||||
repo := repo
|
||||
go func() {
|
||||
m.ScanRepo(repo)
|
||||
wg.Done()
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func (m *Model) CleanRepos() {
|
||||
m.rmut.RLock()
|
||||
var dirs = make([]string, 0, len(m.repoDirs))
|
||||
for _, dir := range m.repoDirs {
|
||||
dirs = append(dirs, dir)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(len(dirs))
|
||||
for _, dir := range dirs {
|
||||
w := &scanner.Walker{
|
||||
Dir: dir,
|
||||
TempNamer: defTempNamer,
|
||||
}
|
||||
go func() {
|
||||
w.CleanTempFiles()
|
||||
wg.Done()
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func (m *Model) ScanRepo(repo string) error {
|
||||
m.rmut.RLock()
|
||||
w := &scanner.Walker{
|
||||
Dir: m.repoDirs[repo],
|
||||
IgnoreFile: ".stignore",
|
||||
BlockSize: scanner.StandardBlockSize,
|
||||
TempNamer: defTempNamer,
|
||||
Suppressor: m.suppressor[repo],
|
||||
CurrentFiler: cFiler{m, repo},
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
m.setState(repo, RepoScanning)
|
||||
fs, _, err := w.Walk()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.ReplaceLocal(repo, fs)
|
||||
m.setState(repo, RepoIdle)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Model) SaveIndexes(dir string) {
|
||||
m.rmut.RLock()
|
||||
for repo := range m.repoDirs {
|
||||
fs := m.protocolIndex(repo)
|
||||
m.saveIndex(repo, dir, fs)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
}
|
||||
|
||||
func (m *Model) LoadIndexes(dir string) {
|
||||
m.rmut.RLock()
|
||||
for repo := range m.repoDirs {
|
||||
fs := m.loadIndex(repo, dir)
|
||||
m.SeedLocal(repo, fs)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
}
|
||||
|
||||
func (m *Model) saveIndex(repo string, dir string, fs []protocol.FileInfo) {
|
||||
id := fmt.Sprintf("%x", sha1.Sum([]byte(m.repoDirs[repo])))
|
||||
name := id + ".idx.gz"
|
||||
name = filepath.Join(dir, name)
|
||||
|
||||
idxf, err := os.Create(name + ".tmp")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
gzw := gzip.NewWriter(idxf)
|
||||
|
||||
protocol.IndexMessage{
|
||||
Repository: repo,
|
||||
Files: fs,
|
||||
}.EncodeXDR(gzw)
|
||||
gzw.Close()
|
||||
idxf.Close()
|
||||
|
||||
Rename(name+".tmp", name)
|
||||
}
|
||||
|
||||
func (m *Model) loadIndex(repo string, dir string) []protocol.FileInfo {
|
||||
id := fmt.Sprintf("%x", sha1.Sum([]byte(m.repoDirs[repo])))
|
||||
name := id + ".idx.gz"
|
||||
name = filepath.Join(dir, name)
|
||||
|
||||
idxf, err := os.Open(name)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
defer idxf.Close()
|
||||
|
||||
gzr, err := gzip.NewReader(idxf)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
defer gzr.Close()
|
||||
|
||||
var im protocol.IndexMessage
|
||||
err = im.DecodeXDR(gzr)
|
||||
if err != nil || im.Repository != repo {
|
||||
return nil
|
||||
}
|
||||
|
||||
return im.Files
|
||||
}
|
||||
|
||||
// clusterConfig returns a ClusterConfigMessage that is correct for the given peer node
|
||||
func (m *Model) clusterConfig(node string) protocol.ClusterConfigMessage {
|
||||
cm := protocol.ClusterConfigMessage{
|
||||
ClientName: m.clientName,
|
||||
ClientVersion: m.clientVersion,
|
||||
}
|
||||
|
||||
m.rmut.RLock()
|
||||
for _, repo := range m.nodeRepos[node] {
|
||||
cr := protocol.Repository{
|
||||
ID: repo,
|
||||
}
|
||||
for _, node := range m.repoNodes[repo] {
|
||||
// TODO: Set read only bit when relevant
|
||||
cr.Nodes = append(cr.Nodes, protocol.Node{
|
||||
ID: node,
|
||||
Flags: protocol.FlagShareTrusted,
|
||||
})
|
||||
}
|
||||
cm.Repositories = append(cm.Repositories, cr)
|
||||
}
|
||||
m.rmut.RUnlock()
|
||||
|
||||
return cm
|
||||
}
|
||||
|
||||
func (m *Model) setState(repo string, state repoState) {
|
||||
m.rmut.Lock()
|
||||
m.repoState[repo] = state
|
||||
m.rmut.Unlock()
|
||||
}
|
||||
|
||||
func (m *Model) State(repo string) string {
|
||||
m.rmut.RLock()
|
||||
state := m.repoState[repo]
|
||||
m.rmut.RUnlock()
|
||||
switch state {
|
||||
case RepoIdle:
|
||||
return "idle"
|
||||
case RepoScanning:
|
||||
return "scanning"
|
||||
case RepoCleaning:
|
||||
return "cleaning"
|
||||
case RepoSyncing:
|
||||
return "syncing"
|
||||
default:
|
||||
return "unknown"
|
||||
}
|
||||
}
|
||||
242
model/model_test.go
Normal file
242
model/model_test.go
Normal file
@@ -0,0 +1,242 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/calmh/syncthing/cid"
|
||||
"github.com/calmh/syncthing/protocol"
|
||||
"github.com/calmh/syncthing/scanner"
|
||||
)
|
||||
|
||||
var testDataExpected = map[string]scanner.File{
|
||||
"foo": scanner.File{
|
||||
Name: "foo",
|
||||
Flags: 0,
|
||||
Modified: 0,
|
||||
Size: 7,
|
||||
Blocks: []scanner.Block{{Offset: 0x0, Size: 0x7, Hash: []uint8{0xae, 0xc0, 0x70, 0x64, 0x5f, 0xe5, 0x3e, 0xe3, 0xb3, 0x76, 0x30, 0x59, 0x37, 0x61, 0x34, 0xf0, 0x58, 0xcc, 0x33, 0x72, 0x47, 0xc9, 0x78, 0xad, 0xd1, 0x78, 0xb6, 0xcc, 0xdf, 0xb0, 0x1, 0x9f}}},
|
||||
},
|
||||
"empty": scanner.File{
|
||||
Name: "empty",
|
||||
Flags: 0,
|
||||
Modified: 0,
|
||||
Size: 0,
|
||||
Blocks: []scanner.Block{{Offset: 0x0, Size: 0x0, Hash: []uint8{0xe3, 0xb0, 0xc4, 0x42, 0x98, 0xfc, 0x1c, 0x14, 0x9a, 0xfb, 0xf4, 0xc8, 0x99, 0x6f, 0xb9, 0x24, 0x27, 0xae, 0x41, 0xe4, 0x64, 0x9b, 0x93, 0x4c, 0xa4, 0x95, 0x99, 0x1b, 0x78, 0x52, 0xb8, 0x55}}},
|
||||
},
|
||||
"bar": scanner.File{
|
||||
Name: "bar",
|
||||
Flags: 0,
|
||||
Modified: 0,
|
||||
Size: 10,
|
||||
Blocks: []scanner.Block{{Offset: 0x0, Size: 0xa, Hash: []uint8{0x2f, 0x72, 0xcc, 0x11, 0xa6, 0xfc, 0xd0, 0x27, 0x1e, 0xce, 0xf8, 0xc6, 0x10, 0x56, 0xee, 0x1e, 0xb1, 0x24, 0x3b, 0xe3, 0x80, 0x5b, 0xf9, 0xa9, 0xdf, 0x98, 0xf9, 0x2f, 0x76, 0x36, 0xb0, 0x5c}}},
|
||||
},
|
||||
}
|
||||
|
||||
func init() {
|
||||
// Fix expected test data to match reality
|
||||
for n, f := range testDataExpected {
|
||||
fi, _ := os.Stat("testdata/" + n)
|
||||
f.Flags = uint32(fi.Mode())
|
||||
f.Modified = fi.ModTime().Unix()
|
||||
testDataExpected[n] = f
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequest(t *testing.T) {
|
||||
m := NewModel(1e6)
|
||||
m.AddRepo("default", "testdata", nil)
|
||||
m.ScanRepo("default")
|
||||
|
||||
bs, err := m.Request("some node", "default", "foo", 0, 6)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if bytes.Compare(bs, []byte("foobar")) != 0 {
|
||||
t.Errorf("Incorrect data from request: %q", string(bs))
|
||||
}
|
||||
|
||||
bs, err = m.Request("some node", "default", "../walk.go", 0, 6)
|
||||
if err == nil {
|
||||
t.Error("Unexpected nil error on insecure file read")
|
||||
}
|
||||
if bs != nil {
|
||||
t.Errorf("Unexpected non nil data on insecure file read: %q", string(bs))
|
||||
}
|
||||
}
|
||||
|
||||
func genFiles(n int) []protocol.FileInfo {
|
||||
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),
|
||||
Modified: t,
|
||||
Blocks: []protocol.BlockInfo{{100, []byte("some hash bytes")}},
|
||||
}
|
||||
}
|
||||
|
||||
return files
|
||||
}
|
||||
|
||||
func BenchmarkIndex10000(b *testing.B) {
|
||||
m := NewModel(1e6)
|
||||
m.AddRepo("default", "testdata", nil)
|
||||
m.ScanRepo("default")
|
||||
files := genFiles(10000)
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
m.Index("42", "default", files)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkIndex00100(b *testing.B) {
|
||||
m := NewModel(1e6)
|
||||
m.AddRepo("default", "testdata", nil)
|
||||
m.ScanRepo("default")
|
||||
files := genFiles(100)
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
m.Index("42", "default", files)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkIndexUpdate10000f10000(b *testing.B) {
|
||||
m := NewModel(1e6)
|
||||
m.AddRepo("default", "testdata", nil)
|
||||
m.ScanRepo("default")
|
||||
files := genFiles(10000)
|
||||
m.Index("42", "default", files)
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
m.IndexUpdate("42", "default", files)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkIndexUpdate10000f00100(b *testing.B) {
|
||||
m := NewModel(1e6)
|
||||
m.AddRepo("default", "testdata", nil)
|
||||
m.ScanRepo("default")
|
||||
files := genFiles(10000)
|
||||
m.Index("42", "default", files)
|
||||
|
||||
ufiles := genFiles(100)
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
m.IndexUpdate("42", "default", ufiles)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkIndexUpdate10000f00001(b *testing.B) {
|
||||
m := NewModel(1e6)
|
||||
m.AddRepo("default", "testdata", nil)
|
||||
m.ScanRepo("default")
|
||||
files := genFiles(10000)
|
||||
m.Index("42", "default", files)
|
||||
|
||||
ufiles := genFiles(1)
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
m.IndexUpdate("42", "default", ufiles)
|
||||
}
|
||||
}
|
||||
|
||||
type FakeConnection struct {
|
||||
id string
|
||||
requestData []byte
|
||||
}
|
||||
|
||||
func (FakeConnection) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f FakeConnection) ID() string {
|
||||
return string(f.id)
|
||||
}
|
||||
|
||||
func (f FakeConnection) Option(string) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (FakeConnection) Index(string, []protocol.FileInfo) {}
|
||||
|
||||
func (f FakeConnection) Request(repo, name string, offset int64, size int) ([]byte, error) {
|
||||
return f.requestData, nil
|
||||
}
|
||||
|
||||
func (FakeConnection) ClusterConfig(protocol.ClusterConfigMessage) {}
|
||||
|
||||
func (FakeConnection) Ping() bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func (FakeConnection) Statistics() protocol.Statistics {
|
||||
return protocol.Statistics{}
|
||||
}
|
||||
|
||||
func BenchmarkRequest(b *testing.B) {
|
||||
m := NewModel(1e6)
|
||||
m.AddRepo("default", "testdata", nil)
|
||||
m.ScanRepo("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),
|
||||
Modified: t,
|
||||
Blocks: []protocol.BlockInfo{{100, []byte("some hash bytes")}},
|
||||
}
|
||||
}
|
||||
|
||||
fc := FakeConnection{
|
||||
id: "42",
|
||||
requestData: []byte("some data to return"),
|
||||
}
|
||||
m.AddConnection(fc, fc)
|
||||
m.Index("42", "default", files)
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
data, err := m.requestGlobal("42", "default", files[i%n].Name, 0, 32, nil)
|
||||
if err != nil {
|
||||
b.Error(err)
|
||||
}
|
||||
if data == nil {
|
||||
b.Error("nil data")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestActivityMap(t *testing.T) {
|
||||
cm := cid.NewMap()
|
||||
fooID := cm.Get("foo")
|
||||
if fooID == 0 {
|
||||
t.Fatal("ID cannot be zero")
|
||||
}
|
||||
barID := cm.Get("bar")
|
||||
if barID == 0 {
|
||||
t.Fatal("ID cannot be zero")
|
||||
}
|
||||
|
||||
m := make(activityMap)
|
||||
if node := m.leastBusyNode(1<<fooID, cm); node != "foo" {
|
||||
t.Errorf("Incorrect least busy node %q", node)
|
||||
}
|
||||
if node := m.leastBusyNode(1<<barID, cm); node != "bar" {
|
||||
t.Errorf("Incorrect least busy node %q", node)
|
||||
}
|
||||
if node := m.leastBusyNode(1<<fooID|1<<barID, cm); node != "foo" {
|
||||
t.Errorf("Incorrect least busy node %q", node)
|
||||
}
|
||||
if node := m.leastBusyNode(1<<fooID|1<<barID, cm); node != "bar" {
|
||||
t.Errorf("Incorrect least busy node %q", node)
|
||||
}
|
||||
}
|
||||
577
model/puller.go
Normal file
577
model/puller.go
Normal file
@@ -0,0 +1,577 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"github.com/calmh/syncthing/buffers"
|
||||
"github.com/calmh/syncthing/cid"
|
||||
"github.com/calmh/syncthing/config"
|
||||
"github.com/calmh/syncthing/protocol"
|
||||
"github.com/calmh/syncthing/scanner"
|
||||
)
|
||||
|
||||
type requestResult struct {
|
||||
node string
|
||||
file scanner.File
|
||||
filepath string // full filepath name
|
||||
offset int64
|
||||
data []byte
|
||||
err error
|
||||
}
|
||||
|
||||
type openFile struct {
|
||||
filepath string // full filepath name
|
||||
temp string // temporary filename
|
||||
availability uint64 // availability bitset
|
||||
file *os.File
|
||||
err error // error when opening or writing to file, all following operations are cancelled
|
||||
outstanding int // number of requests we still have outstanding
|
||||
done bool // we have sent all requests for this file
|
||||
}
|
||||
|
||||
type activityMap map[string]int
|
||||
|
||||
func (m activityMap) leastBusyNode(availability uint64, cm *cid.Map) string {
|
||||
var low int = 2<<30 - 1
|
||||
var selected string
|
||||
for _, node := range cm.Names() {
|
||||
id := cm.Get(node)
|
||||
if id == cid.LocalID {
|
||||
continue
|
||||
}
|
||||
usage := m[node]
|
||||
if availability&(1<<id) != 0 {
|
||||
if usage < low {
|
||||
low = usage
|
||||
selected = node
|
||||
}
|
||||
}
|
||||
}
|
||||
m[selected]++
|
||||
return selected
|
||||
}
|
||||
|
||||
func (m activityMap) decrease(node string) {
|
||||
m[node]--
|
||||
}
|
||||
|
||||
var errNoNode = errors.New("no available source node")
|
||||
|
||||
type puller struct {
|
||||
cfg *config.Configuration
|
||||
repo string
|
||||
dir string
|
||||
bq *blockQueue
|
||||
model *Model
|
||||
oustandingPerNode activityMap
|
||||
openFiles map[string]openFile
|
||||
requestSlots chan bool
|
||||
blocks chan bqBlock
|
||||
requestResults chan requestResult
|
||||
}
|
||||
|
||||
func newPuller(repo, dir string, model *Model, slots int, cfg *config.Configuration) *puller {
|
||||
p := &puller{
|
||||
cfg: cfg,
|
||||
repo: repo,
|
||||
dir: dir,
|
||||
bq: newBlockQueue(),
|
||||
model: model,
|
||||
oustandingPerNode: make(activityMap),
|
||||
openFiles: make(map[string]openFile),
|
||||
requestSlots: make(chan bool, slots),
|
||||
blocks: make(chan bqBlock),
|
||||
requestResults: make(chan requestResult),
|
||||
}
|
||||
|
||||
if slots > 0 {
|
||||
// Read/write
|
||||
for i := 0; i < slots; i++ {
|
||||
p.requestSlots <- true
|
||||
}
|
||||
if debug {
|
||||
l.Debugf("starting puller; repo %q dir %q slots %d", repo, dir, slots)
|
||||
}
|
||||
go p.run()
|
||||
} else {
|
||||
// Read only
|
||||
if debug {
|
||||
l.Debugf("starting puller; repo %q dir %q (read only)", repo, dir)
|
||||
}
|
||||
go p.runRO()
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func (p *puller) run() {
|
||||
go func() {
|
||||
// fill blocks queue when there are free slots
|
||||
for {
|
||||
<-p.requestSlots
|
||||
b := p.bq.get()
|
||||
if debug {
|
||||
l.Debugf("filler: queueing %q / %q offset %d copy %d", p.repo, b.file.Name, b.block.Offset, len(b.copy))
|
||||
}
|
||||
p.blocks <- b
|
||||
}
|
||||
}()
|
||||
|
||||
walkTicker := time.Tick(time.Duration(p.cfg.Options.RescanIntervalS) * time.Second)
|
||||
timeout := time.Tick(5 * time.Second)
|
||||
changed := true
|
||||
|
||||
for {
|
||||
// Run the pulling loop as long as there are blocks to fetch
|
||||
pull:
|
||||
for {
|
||||
select {
|
||||
case res := <-p.requestResults:
|
||||
p.model.setState(p.repo, RepoSyncing)
|
||||
changed = true
|
||||
p.requestSlots <- true
|
||||
p.handleRequestResult(res)
|
||||
|
||||
case b := <-p.blocks:
|
||||
p.model.setState(p.repo, RepoSyncing)
|
||||
changed = true
|
||||
if p.handleBlock(b) {
|
||||
// Block was fully handled, free up the slot
|
||||
p.requestSlots <- true
|
||||
}
|
||||
|
||||
case <-timeout:
|
||||
if len(p.openFiles) == 0 && p.bq.empty() {
|
||||
// Nothing more to do for the moment
|
||||
break pull
|
||||
}
|
||||
if debug {
|
||||
l.Debugf("%q: idle but have %d open files", p.repo, len(p.openFiles))
|
||||
i := 5
|
||||
for _, f := range p.openFiles {
|
||||
l.Debugf(" %v", f)
|
||||
i--
|
||||
if i == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if changed {
|
||||
p.model.setState(p.repo, RepoCleaning)
|
||||
p.fixupDirectories()
|
||||
changed = false
|
||||
}
|
||||
|
||||
p.model.setState(p.repo, RepoIdle)
|
||||
|
||||
// Do a rescan if it's time for it
|
||||
select {
|
||||
case <-walkTicker:
|
||||
if debug {
|
||||
l.Debugf("%q: time for rescan", p.repo)
|
||||
}
|
||||
err := p.model.ScanRepo(p.repo)
|
||||
if err != nil {
|
||||
invalidateRepo(p.cfg, p.repo, err)
|
||||
return
|
||||
}
|
||||
|
||||
default:
|
||||
}
|
||||
|
||||
// Queue more blocks to fetch, if any
|
||||
p.queueNeededBlocks()
|
||||
}
|
||||
}
|
||||
|
||||
func (p *puller) runRO() {
|
||||
walkTicker := time.Tick(time.Duration(p.cfg.Options.RescanIntervalS) * time.Second)
|
||||
|
||||
for _ = range walkTicker {
|
||||
if debug {
|
||||
l.Debugf("%q: time for rescan", p.repo)
|
||||
}
|
||||
err := p.model.ScanRepo(p.repo)
|
||||
if err != nil {
|
||||
invalidateRepo(p.cfg, p.repo, err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *puller) fixupDirectories() {
|
||||
var deleteDirs []string
|
||||
filepath.Walk(p.dir, func(path string, info os.FileInfo, err error) error {
|
||||
if !info.IsDir() {
|
||||
return nil
|
||||
}
|
||||
|
||||
rn, err := filepath.Rel(p.dir, path)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if rn == "." {
|
||||
return nil
|
||||
}
|
||||
|
||||
cur := p.model.CurrentGlobalFile(p.repo, rn)
|
||||
if cur.Name != rn {
|
||||
// No matching dir in current list; weird
|
||||
return nil
|
||||
}
|
||||
|
||||
if cur.Flags&protocol.FlagDeleted != 0 {
|
||||
if debug {
|
||||
l.Debugf("queue delete dir: %v", cur)
|
||||
}
|
||||
|
||||
// We queue the directories to delete since we walk the
|
||||
// tree in depth first order and need to remove the
|
||||
// directories in the opposite order.
|
||||
|
||||
deleteDirs = append(deleteDirs, path)
|
||||
return nil
|
||||
}
|
||||
|
||||
if cur.Flags&uint32(os.ModePerm) != uint32(info.Mode()&os.ModePerm) {
|
||||
os.Chmod(path, os.FileMode(cur.Flags)&os.ModePerm)
|
||||
if debug {
|
||||
l.Debugf("restored dir flags: %o -> %v", info.Mode()&os.ModePerm, cur)
|
||||
}
|
||||
}
|
||||
|
||||
if cur.Modified != info.ModTime().Unix() {
|
||||
t := time.Unix(cur.Modified, 0)
|
||||
os.Chtimes(path, t, t)
|
||||
if debug {
|
||||
l.Debugf("restored dir modtime: %d -> %v", info.ModTime().Unix(), cur)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
// Delete any queued directories
|
||||
for i := len(deleteDirs) - 1; i >= 0; i-- {
|
||||
if debug {
|
||||
l.Debugln("delete dir:", deleteDirs[i])
|
||||
}
|
||||
err := os.Remove(deleteDirs[i])
|
||||
if err != nil {
|
||||
l.Warnln(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *puller) handleRequestResult(res requestResult) {
|
||||
p.oustandingPerNode.decrease(res.node)
|
||||
f := res.file
|
||||
|
||||
of, ok := p.openFiles[f.Name]
|
||||
if !ok || of.err != nil {
|
||||
// no entry in openFiles means there was an error and we've cancelled the operation
|
||||
return
|
||||
}
|
||||
|
||||
_, of.err = of.file.WriteAt(res.data, res.offset)
|
||||
buffers.Put(res.data)
|
||||
|
||||
of.outstanding--
|
||||
p.openFiles[f.Name] = of
|
||||
|
||||
if debug {
|
||||
l.Debugf("pull: wrote %q / %q offset %d outstanding %d done %v", p.repo, f.Name, res.offset, of.outstanding, of.done)
|
||||
}
|
||||
|
||||
if of.done && of.outstanding == 0 {
|
||||
p.closeFile(f)
|
||||
}
|
||||
}
|
||||
|
||||
// handleBlock fulfills the block request by copying, ignoring or fetching
|
||||
// from the network. Returns true if the block was fully handled
|
||||
// synchronously, i.e. if the slot can be reused.
|
||||
func (p *puller) handleBlock(b bqBlock) bool {
|
||||
f := b.file
|
||||
|
||||
// For directories, simply making sure they exist is enough
|
||||
if f.Flags&protocol.FlagDirectory != 0 {
|
||||
path := filepath.Join(p.dir, f.Name)
|
||||
_, err := os.Stat(path)
|
||||
if err != nil && os.IsNotExist(err) {
|
||||
os.MkdirAll(path, 0777)
|
||||
}
|
||||
p.model.updateLocal(p.repo, f)
|
||||
return true
|
||||
}
|
||||
|
||||
of, ok := p.openFiles[f.Name]
|
||||
of.done = b.last
|
||||
|
||||
if !ok {
|
||||
if debug {
|
||||
l.Debugf("pull: %q: opening file %q", p.repo, f.Name)
|
||||
}
|
||||
|
||||
of.availability = uint64(p.model.repoFiles[p.repo].Availability(f.Name))
|
||||
of.filepath = filepath.Join(p.dir, f.Name)
|
||||
of.temp = filepath.Join(p.dir, defTempNamer.TempName(f.Name))
|
||||
|
||||
dirName := filepath.Dir(of.filepath)
|
||||
_, err := os.Stat(dirName)
|
||||
if err != nil {
|
||||
err = os.MkdirAll(dirName, 0777)
|
||||
}
|
||||
if err != nil {
|
||||
l.Debugf("pull: error: %q / %q: %v", p.repo, f.Name, err)
|
||||
}
|
||||
|
||||
of.file, of.err = os.Create(of.temp)
|
||||
if of.err != nil {
|
||||
if debug {
|
||||
l.Debugf("pull: error: %q / %q: %v", p.repo, f.Name, of.err)
|
||||
}
|
||||
if !b.last {
|
||||
p.openFiles[f.Name] = of
|
||||
}
|
||||
return true
|
||||
}
|
||||
defTempNamer.Hide(of.temp)
|
||||
}
|
||||
|
||||
if of.err != nil {
|
||||
// We have already failed this file.
|
||||
if debug {
|
||||
l.Debugf("pull: error: %q / %q has already failed: %v", p.repo, f.Name, of.err)
|
||||
}
|
||||
if b.last {
|
||||
l.Debugf("pull: removing failed file %q / %q", p.repo, f.Name)
|
||||
delete(p.openFiles, f.Name)
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
p.openFiles[f.Name] = of
|
||||
|
||||
switch {
|
||||
case len(b.copy) > 0:
|
||||
p.handleCopyBlock(b)
|
||||
return true
|
||||
|
||||
case b.block.Size > 0:
|
||||
return p.handleRequestBlock(b)
|
||||
|
||||
default:
|
||||
p.handleEmptyBlock(b)
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
func (p *puller) handleCopyBlock(b bqBlock) {
|
||||
// We have blocks to copy from the existing file
|
||||
f := b.file
|
||||
of := p.openFiles[f.Name]
|
||||
|
||||
if debug {
|
||||
l.Debugf("pull: copying %d blocks for %q / %q", len(b.copy), p.repo, f.Name)
|
||||
}
|
||||
|
||||
var exfd *os.File
|
||||
exfd, of.err = os.Open(of.filepath)
|
||||
if of.err != nil {
|
||||
if debug {
|
||||
l.Debugf("pull: error: %q / %q: %v", p.repo, f.Name, of.err)
|
||||
}
|
||||
of.file.Close()
|
||||
of.file = nil
|
||||
|
||||
p.openFiles[f.Name] = of
|
||||
return
|
||||
}
|
||||
defer exfd.Close()
|
||||
|
||||
for _, b := range b.copy {
|
||||
bs := buffers.Get(int(b.Size))
|
||||
_, of.err = exfd.ReadAt(bs, b.Offset)
|
||||
if of.err == nil {
|
||||
_, of.err = of.file.WriteAt(bs, b.Offset)
|
||||
}
|
||||
buffers.Put(bs)
|
||||
if of.err != nil {
|
||||
if debug {
|
||||
l.Debugf("pull: error: %q / %q: %v", p.repo, f.Name, of.err)
|
||||
}
|
||||
exfd.Close()
|
||||
of.file.Close()
|
||||
of.file = nil
|
||||
|
||||
p.openFiles[f.Name] = of
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// handleRequestBlock tries to pull a block from the network. Returns true if
|
||||
// the block could _not_ be fetched (i.e. it was fully handled, matching the
|
||||
// return criteria of handleBlock)
|
||||
func (p *puller) handleRequestBlock(b bqBlock) bool {
|
||||
f := b.file
|
||||
of, ok := p.openFiles[f.Name]
|
||||
if !ok {
|
||||
panic("bug: request for non-open file")
|
||||
}
|
||||
|
||||
node := p.oustandingPerNode.leastBusyNode(of.availability, p.model.cm)
|
||||
if len(node) == 0 {
|
||||
of.err = errNoNode
|
||||
if of.file != nil {
|
||||
of.file.Close()
|
||||
of.file = nil
|
||||
os.Remove(of.temp)
|
||||
}
|
||||
if b.last {
|
||||
delete(p.openFiles, f.Name)
|
||||
} else {
|
||||
p.openFiles[f.Name] = of
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
of.outstanding++
|
||||
p.openFiles[f.Name] = of
|
||||
|
||||
go func(node string, b bqBlock) {
|
||||
if debug {
|
||||
l.Debugf("pull: requesting %q / %q offset %d size %d from %q outstanding %d", p.repo, f.Name, b.block.Offset, b.block.Size, node, of.outstanding)
|
||||
}
|
||||
|
||||
bs, err := p.model.requestGlobal(node, p.repo, f.Name, b.block.Offset, int(b.block.Size), nil)
|
||||
p.requestResults <- requestResult{
|
||||
node: node,
|
||||
file: f,
|
||||
filepath: of.filepath,
|
||||
offset: b.block.Offset,
|
||||
data: bs,
|
||||
err: err,
|
||||
}
|
||||
}(node, b)
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func (p *puller) handleEmptyBlock(b bqBlock) {
|
||||
f := b.file
|
||||
of := p.openFiles[f.Name]
|
||||
|
||||
if b.last {
|
||||
if of.err == nil {
|
||||
of.file.Close()
|
||||
}
|
||||
}
|
||||
|
||||
if f.Flags&protocol.FlagDeleted != 0 {
|
||||
if debug {
|
||||
l.Debugf("pull: delete %q", f.Name)
|
||||
}
|
||||
os.Remove(of.temp)
|
||||
os.Remove(of.filepath)
|
||||
} else {
|
||||
if debug {
|
||||
l.Debugf("pull: no blocks to fetch and nothing to copy for %q / %q", p.repo, f.Name)
|
||||
}
|
||||
t := time.Unix(f.Modified, 0)
|
||||
os.Chtimes(of.temp, t, t)
|
||||
os.Chmod(of.temp, os.FileMode(f.Flags&0777))
|
||||
defTempNamer.Show(of.temp)
|
||||
Rename(of.temp, of.filepath)
|
||||
}
|
||||
delete(p.openFiles, f.Name)
|
||||
p.model.updateLocal(p.repo, f)
|
||||
}
|
||||
|
||||
func (p *puller) queueNeededBlocks() {
|
||||
queued := 0
|
||||
for _, f := range p.model.NeedFilesRepo(p.repo) {
|
||||
lf := p.model.CurrentRepoFile(p.repo, f.Name)
|
||||
have, need := scanner.BlockDiff(lf.Blocks, f.Blocks)
|
||||
if debug {
|
||||
l.Debugf("need:\n local: %v\n global: %v\n haveBlocks: %v\n needBlocks: %v", lf, f, have, need)
|
||||
}
|
||||
queued++
|
||||
p.bq.put(bqAdd{
|
||||
file: f,
|
||||
have: have,
|
||||
need: need,
|
||||
})
|
||||
}
|
||||
if debug && queued > 0 {
|
||||
l.Debugf("%q: queued %d blocks", p.repo, queued)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *puller) closeFile(f scanner.File) {
|
||||
if debug {
|
||||
l.Debugf("pull: closing %q / %q", p.repo, f.Name)
|
||||
}
|
||||
|
||||
of := p.openFiles[f.Name]
|
||||
of.file.Close()
|
||||
defer os.Remove(of.temp)
|
||||
|
||||
delete(p.openFiles, f.Name)
|
||||
|
||||
fd, err := os.Open(of.temp)
|
||||
if err != nil {
|
||||
if debug {
|
||||
l.Debugf("pull: error: %q / %q: %v", p.repo, f.Name, err)
|
||||
}
|
||||
return
|
||||
}
|
||||
hb, _ := scanner.Blocks(fd, scanner.StandardBlockSize)
|
||||
fd.Close()
|
||||
|
||||
if l0, l1 := len(hb), len(f.Blocks); l0 != l1 {
|
||||
if debug {
|
||||
l.Debugf("pull: %q / %q: nblocks %d != %d", p.repo, f.Name, l0, l1)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
for i := range hb {
|
||||
if bytes.Compare(hb[i].Hash, f.Blocks[i].Hash) != 0 {
|
||||
l.Debugf("pull: %q / %q: block %d hash mismatch", p.repo, f.Name, i)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
t := time.Unix(f.Modified, 0)
|
||||
os.Chtimes(of.temp, t, t)
|
||||
os.Chmod(of.temp, os.FileMode(f.Flags&0777))
|
||||
defTempNamer.Show(of.temp)
|
||||
if debug {
|
||||
l.Debugf("pull: rename %q / %q: %q", p.repo, f.Name, of.filepath)
|
||||
}
|
||||
if err := Rename(of.temp, of.filepath); err == nil {
|
||||
p.model.updateLocal(p.repo, f)
|
||||
} else {
|
||||
l.Debugf("pull: error: %q / %q: %v", p.repo, f.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
func invalidateRepo(cfg *config.Configuration, repoID string, err error) {
|
||||
for i := range cfg.Repositories {
|
||||
repo := &cfg.Repositories[i]
|
||||
if repo.ID == repoID {
|
||||
repo.Invalid = err.Error()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
77
model/suppressor.go
Normal file
77
model/suppressor.go
Normal file
@@ -0,0 +1,77 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
MaxChangeHistory = 4
|
||||
)
|
||||
|
||||
type change struct {
|
||||
size int64
|
||||
when time.Time
|
||||
}
|
||||
|
||||
type changeHistory struct {
|
||||
changes []change
|
||||
next int64
|
||||
prevSup bool
|
||||
}
|
||||
|
||||
type suppressor struct {
|
||||
sync.Mutex
|
||||
changes map[string]changeHistory
|
||||
threshold int64 // bytes/s
|
||||
}
|
||||
|
||||
func (h changeHistory) bandwidth(t time.Time) int64 {
|
||||
if len(h.changes) == 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
var t0 = h.changes[0].when
|
||||
if t == t0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
var bw float64
|
||||
for _, c := range h.changes {
|
||||
bw += float64(c.size)
|
||||
}
|
||||
return int64(bw / t.Sub(t0).Seconds())
|
||||
}
|
||||
|
||||
func (h *changeHistory) append(size int64, t time.Time) {
|
||||
c := change{size, t}
|
||||
if len(h.changes) == MaxChangeHistory {
|
||||
h.changes = h.changes[1:MaxChangeHistory]
|
||||
}
|
||||
h.changes = append(h.changes, c)
|
||||
}
|
||||
|
||||
func (s *suppressor) Suppress(name string, fi os.FileInfo) (cur, prev bool) {
|
||||
return s.suppress(name, fi.Size(), time.Now())
|
||||
}
|
||||
|
||||
func (s *suppressor) suppress(name string, size int64, t time.Time) (bool, bool) {
|
||||
s.Lock()
|
||||
|
||||
if s.changes == nil {
|
||||
s.changes = make(map[string]changeHistory)
|
||||
}
|
||||
h := s.changes[name]
|
||||
sup := h.bandwidth(t) > s.threshold
|
||||
prevSup := h.prevSup
|
||||
h.prevSup = sup
|
||||
if !sup {
|
||||
h.append(size, t)
|
||||
}
|
||||
s.changes[name] = h
|
||||
|
||||
s.Unlock()
|
||||
|
||||
return sup, prevSup
|
||||
}
|
||||
113
model/suppressor_test.go
Normal file
113
model/suppressor_test.go
Normal file
@@ -0,0 +1,113 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestSuppressor(t *testing.T) {
|
||||
s := suppressor{threshold: 10000}
|
||||
t0 := time.Now()
|
||||
|
||||
t1 := t0
|
||||
sup, prev := s.suppress("foo", 10000, t1)
|
||||
if sup {
|
||||
t.Fatal("Never suppress first change")
|
||||
}
|
||||
if prev {
|
||||
t.Fatal("Incorrect prev status")
|
||||
}
|
||||
|
||||
// bw is 10000 / 10 = 1000
|
||||
t1 = t0.Add(10 * time.Second)
|
||||
if bw := s.changes["foo"].bandwidth(t1); bw != 1000 {
|
||||
t.Errorf("Incorrect bw %d", bw)
|
||||
}
|
||||
sup, prev = s.suppress("foo", 10000, t1)
|
||||
if sup {
|
||||
t.Fatal("Should still be fine")
|
||||
}
|
||||
if prev {
|
||||
t.Fatal("Incorrect prev status")
|
||||
}
|
||||
|
||||
// bw is (10000 + 10000) / 11 = 1818
|
||||
t1 = t0.Add(11 * time.Second)
|
||||
if bw := s.changes["foo"].bandwidth(t1); bw != 1818 {
|
||||
t.Errorf("Incorrect bw %d", bw)
|
||||
}
|
||||
sup, prev = s.suppress("foo", 100500, t1)
|
||||
if sup {
|
||||
t.Fatal("Should still be fine")
|
||||
}
|
||||
if prev {
|
||||
t.Fatal("Incorrect prev status")
|
||||
}
|
||||
|
||||
// bw is (10000 + 10000 + 100500) / 12 = 10041
|
||||
t1 = t0.Add(12 * time.Second)
|
||||
if bw := s.changes["foo"].bandwidth(t1); bw != 10041 {
|
||||
t.Errorf("Incorrect bw %d", bw)
|
||||
}
|
||||
sup, prev = s.suppress("foo", 10000000, t1) // value will be ignored
|
||||
if !sup {
|
||||
t.Fatal("Should be over threshold")
|
||||
}
|
||||
if prev {
|
||||
t.Fatal("Incorrect prev status")
|
||||
}
|
||||
|
||||
// bw is (10000 + 10000 + 100500) / 15 = 8033
|
||||
t1 = t0.Add(15 * time.Second)
|
||||
if bw := s.changes["foo"].bandwidth(t1); bw != 8033 {
|
||||
t.Errorf("Incorrect bw %d", bw)
|
||||
}
|
||||
sup, prev = s.suppress("foo", 10000000, t1)
|
||||
if sup {
|
||||
t.Fatal("Should be Ok")
|
||||
}
|
||||
if !prev {
|
||||
t.Fatal("Incorrect prev status")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHistory(t *testing.T) {
|
||||
h := changeHistory{}
|
||||
|
||||
t0 := time.Now()
|
||||
h.append(40, t0)
|
||||
|
||||
if l := len(h.changes); l != 1 {
|
||||
t.Errorf("Incorrect history length %d", l)
|
||||
}
|
||||
if s := h.changes[0].size; s != 40 {
|
||||
t.Errorf("Incorrect first record size %d", s)
|
||||
}
|
||||
|
||||
for i := 1; i < MaxChangeHistory; i++ {
|
||||
h.append(int64(40+i), t0.Add(time.Duration(i)*time.Second))
|
||||
}
|
||||
|
||||
if l := len(h.changes); l != MaxChangeHistory {
|
||||
t.Errorf("Incorrect history length %d", l)
|
||||
}
|
||||
if s := h.changes[0].size; s != 40 {
|
||||
t.Errorf("Incorrect first record size %d", s)
|
||||
}
|
||||
if s := h.changes[MaxChangeHistory-1].size; s != 40+MaxChangeHistory-1 {
|
||||
t.Errorf("Incorrect last record size %d", s)
|
||||
}
|
||||
|
||||
h.append(999, t0.Add(time.Duration(999)*time.Second))
|
||||
|
||||
if l := len(h.changes); l != MaxChangeHistory {
|
||||
t.Errorf("Incorrect history length %d", l)
|
||||
}
|
||||
if s := h.changes[0].size; s != 41 {
|
||||
t.Errorf("Incorrect first record size %d", s)
|
||||
}
|
||||
if s := h.changes[MaxChangeHistory-1].size; s != 999 {
|
||||
t.Errorf("Incorrect last record size %d", s)
|
||||
}
|
||||
|
||||
}
|
||||
33
model/tempname.go
Normal file
33
model/tempname.go
Normal file
@@ -0,0 +1,33 @@
|
||||
// +build !windows
|
||||
|
||||
package model
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type tempNamer struct {
|
||||
prefix string
|
||||
}
|
||||
|
||||
var defTempNamer = tempNamer{".syncthing"}
|
||||
|
||||
func (t tempNamer) IsTemporary(name string) bool {
|
||||
return strings.HasPrefix(filepath.Base(name), t.prefix)
|
||||
}
|
||||
|
||||
func (t tempNamer) TempName(name string) string {
|
||||
tdir := filepath.Dir(name)
|
||||
tname := fmt.Sprintf("%s.%s", t.prefix, filepath.Base(name))
|
||||
return filepath.Join(tdir, tname)
|
||||
}
|
||||
|
||||
func (t tempNamer) Hide(path string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t tempNamer) Show(path string) error {
|
||||
return nil
|
||||
}
|
||||
56
model/tempname_windows.go
Normal file
56
model/tempname_windows.go
Normal file
@@ -0,0 +1,56 @@
|
||||
// +build windows
|
||||
|
||||
package model
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
type tempNamer struct {
|
||||
prefix string
|
||||
}
|
||||
|
||||
var defTempNamer = tempNamer{"~syncthing~"}
|
||||
|
||||
func (t tempNamer) IsTemporary(name string) bool {
|
||||
return strings.HasPrefix(filepath.Base(name), t.prefix)
|
||||
}
|
||||
|
||||
func (t tempNamer) TempName(name string) string {
|
||||
tdir := filepath.Dir(name)
|
||||
tname := fmt.Sprintf("%s.%s.tmp", t.prefix, filepath.Base(name))
|
||||
return filepath.Join(tdir, tname)
|
||||
}
|
||||
|
||||
func (t tempNamer) Hide(path string) error {
|
||||
p, err := syscall.UTF16PtrFromString(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
attrs, err := syscall.GetFileAttributes(p)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
attrs |= syscall.FILE_ATTRIBUTE_HIDDEN
|
||||
return syscall.SetFileAttributes(p, attrs)
|
||||
}
|
||||
|
||||
func (t tempNamer) Show(path string) error {
|
||||
p, err := syscall.UTF16PtrFromString(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
attrs, err := syscall.GetFileAttributes(p)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
attrs &^= syscall.FILE_ATTRIBUTE_HIDDEN
|
||||
return syscall.SetFileAttributes(p, attrs)
|
||||
}
|
||||
109
model/util.go
Normal file
109
model/util.go
Normal file
@@ -0,0 +1,109 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
|
||||
"github.com/calmh/syncthing/protocol"
|
||||
"github.com/calmh/syncthing/scanner"
|
||||
)
|
||||
|
||||
func Rename(from, to string) error {
|
||||
if runtime.GOOS == "windows" {
|
||||
err := os.Remove(to)
|
||||
if err != nil && !os.IsNotExist(err) {
|
||||
l.Warnln(err)
|
||||
}
|
||||
}
|
||||
return os.Rename(from, to)
|
||||
}
|
||||
|
||||
func fileFromFileInfo(f protocol.FileInfo) scanner.File {
|
||||
var blocks = make([]scanner.Block, len(f.Blocks))
|
||||
var offset int64
|
||||
for i, b := range f.Blocks {
|
||||
blocks[i] = scanner.Block{
|
||||
Offset: offset,
|
||||
Size: b.Size,
|
||||
Hash: b.Hash,
|
||||
}
|
||||
offset += int64(b.Size)
|
||||
}
|
||||
return scanner.File{
|
||||
// Name is with native separator and normalization
|
||||
Name: filepath.FromSlash(f.Name),
|
||||
Size: offset,
|
||||
Flags: f.Flags &^ protocol.FlagInvalid,
|
||||
Modified: f.Modified,
|
||||
Version: f.Version,
|
||||
Blocks: blocks,
|
||||
Suppressed: f.Flags&protocol.FlagInvalid != 0,
|
||||
}
|
||||
}
|
||||
|
||||
func fileInfoFromFile(f scanner.File) protocol.FileInfo {
|
||||
var blocks = make([]protocol.BlockInfo, len(f.Blocks))
|
||||
for i, b := range f.Blocks {
|
||||
blocks[i] = protocol.BlockInfo{
|
||||
Size: b.Size,
|
||||
Hash: b.Hash,
|
||||
}
|
||||
}
|
||||
pf := protocol.FileInfo{
|
||||
Name: filepath.ToSlash(f.Name),
|
||||
Flags: f.Flags,
|
||||
Modified: f.Modified,
|
||||
Version: f.Version,
|
||||
Blocks: blocks,
|
||||
}
|
||||
if f.Suppressed {
|
||||
pf.Flags |= protocol.FlagInvalid
|
||||
}
|
||||
return pf
|
||||
}
|
||||
|
||||
func cmMap(cm protocol.ClusterConfigMessage) map[string]map[string]uint32 {
|
||||
m := make(map[string]map[string]uint32)
|
||||
for _, repo := range cm.Repositories {
|
||||
m[repo.ID] = make(map[string]uint32)
|
||||
for _, node := range repo.Nodes {
|
||||
m[repo.ID][node.ID] = node.Flags
|
||||
}
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
type ClusterConfigMismatch error
|
||||
|
||||
// compareClusterConfig returns nil for two equivalent configurations,
|
||||
// otherwise a decriptive error
|
||||
func compareClusterConfig(local, remote protocol.ClusterConfigMessage) error {
|
||||
lm := cmMap(local)
|
||||
rm := cmMap(remote)
|
||||
|
||||
for repo, lnodes := range lm {
|
||||
_ = lnodes
|
||||
if rnodes, ok := rm[repo]; ok {
|
||||
for node, lflags := range lnodes {
|
||||
if rflags, ok := rnodes[node]; ok {
|
||||
if lflags&protocol.FlagShareBits != rflags&protocol.FlagShareBits {
|
||||
return ClusterConfigMismatch(fmt.Errorf("remote has different sharing flags for node %q in repository %q", node, repo))
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
return ClusterConfigMismatch(fmt.Errorf("remote is missing repository %q", repo))
|
||||
}
|
||||
}
|
||||
|
||||
for repo := range rm {
|
||||
if _, ok := lm[repo]; !ok {
|
||||
return ClusterConfigMismatch(fmt.Errorf("remote has extra repository %q", repo))
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user