2014-07-12 22:45:33 +00:00
// Copyright (C) 2014 Jakob Borg and Contributors (see the CONTRIBUTORS file).
// All rights reserved. Use of this source code is governed by an MIT-style
// license that can be found in the LICENSE file.
2014-06-01 20:50:14 +00:00
2014-05-15 03:26:55 +00:00
package model
2013-12-15 10:43:31 +00:00
import (
2014-01-06 20:31:36 +00:00
"errors"
2013-12-23 17:12:44 +00:00
"fmt"
2014-01-01 02:22:49 +00:00
"io"
2014-01-05 22:54:57 +00:00
"net"
2013-12-15 10:43:31 +00:00
"os"
2014-03-28 13:36:57 +00:00
"path/filepath"
2014-06-26 09:24:38 +00:00
"strconv"
2013-12-15 10:43:31 +00:00
"sync"
"time"
2014-06-21 07:43:12 +00:00
2014-05-15 00:18:09 +00:00
"github.com/calmh/syncthing/config"
2014-07-13 19:07:24 +00:00
"github.com/calmh/syncthing/events"
2014-03-28 13:36:57 +00:00
"github.com/calmh/syncthing/files"
"github.com/calmh/syncthing/lamport"
2013-12-15 10:43:31 +00:00
"github.com/calmh/syncthing/protocol"
2014-03-08 22:02:01 +00:00
"github.com/calmh/syncthing/scanner"
2014-07-06 12:46:48 +00:00
"github.com/syndtr/goleveldb/leveldb"
2013-12-15 10:43:31 +00:00
)
2014-04-14 07:58:17 +00:00
type repoState int
const (
RepoIdle repoState = iota
RepoScanning
RepoSyncing
RepoCleaning
)
2014-07-17 11:38:36 +00:00
func ( s repoState ) String ( ) string {
switch s {
case RepoIdle :
return "idle"
case RepoScanning :
return "scanning"
case RepoCleaning :
return "cleaning"
case RepoSyncing :
return "syncing"
default :
return "unknown"
}
}
2014-05-18 10:56:50 +00:00
// Somewhat arbitrary amount of bytes that we choose to let represent the size
// of an unsynchronized directory entry or a deleted file. We need it to be
// larger than zero so that it's visible that there is some amount of bytes to
// transfer to bring the systems into synchronization.
const zeroEntrySize = 128
2014-07-15 11:04:37 +00:00
// How many files to send in each Index/IndexUpdate message.
const indexBatchSize = 1000
2013-12-15 10:43:31 +00:00
type Model struct {
2014-05-15 03:26:55 +00:00
indexDir string
cfg * config . Configuration
2014-07-06 12:46:48 +00:00
db * leveldb . DB
2014-05-15 03:26:55 +00:00
clientName string
clientVersion string
2014-05-23 12:31:16 +00:00
repoCfgs map [ string ] config . RepositoryConfiguration // repo -> cfg
repoFiles map [ string ] * files . Set // repo -> files
2014-06-29 23:42:03 +00:00
repoNodes map [ string ] [ ] protocol . NodeID // repo -> nodeIDs
nodeRepos map [ protocol . NodeID ] [ ] string // nodeID -> repos
2014-05-23 12:31:16 +00:00
suppressor map [ string ] * suppressor // repo -> suppressor
rmut sync . RWMutex // protects the above
2014-03-29 17:53:48 +00:00
2014-07-17 11:38:36 +00:00
repoState map [ string ] repoState // repo -> state
repoStateChanged map [ string ] time . Time // repo -> time when state changed
smut sync . RWMutex
2014-05-20 16:41:01 +00:00
2014-06-29 23:42:03 +00:00
protoConn map [ protocol . NodeID ] protocol . Connection
rawConn map [ protocol . NodeID ] io . Closer
nodeVer map [ protocol . NodeID ] string
2014-01-18 03:06:44 +00:00
pmut sync . RWMutex // protects protoConn and rawConn
2013-12-30 14:30:29 +00:00
2014-07-15 11:04:37 +00:00
sentLocalVer map [ protocol . NodeID ] map [ string ] uint64
slMut sync . Mutex
2014-01-22 10:38:52 +00:00
sup suppressor
2014-01-12 15:59:35 +00:00
2014-03-29 17:53:48 +00:00
addedRepo bool
started bool
2013-12-15 10:43:31 +00:00
}
2014-01-07 21:44:21 +00:00
var (
ErrNoSuchFile = errors . New ( "no such file" )
ErrInvalid = errors . New ( "file is invalid" )
)
2014-01-06 20:31:36 +00:00
2014-01-06 10:11:18 +00:00
// 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.
2014-07-06 12:46:48 +00:00
func NewModel ( indexDir string , cfg * config . Configuration , clientName , clientVersion string , db * leveldb . DB ) * Model {
2013-12-15 10:43:31 +00:00
m := & Model {
2014-07-17 11:38:36 +00:00
indexDir : indexDir ,
cfg : cfg ,
db : db ,
clientName : clientName ,
clientVersion : clientVersion ,
repoCfgs : make ( map [ string ] config . RepositoryConfiguration ) ,
repoFiles : make ( map [ string ] * files . Set ) ,
repoNodes : make ( map [ string ] [ ] protocol . NodeID ) ,
nodeRepos : make ( map [ protocol . NodeID ] [ ] string ) ,
repoState : make ( map [ string ] repoState ) ,
repoStateChanged : make ( map [ string ] time . Time ) ,
suppressor : make ( map [ string ] * suppressor ) ,
protoConn : make ( map [ protocol . NodeID ] protocol . Connection ) ,
rawConn : make ( map [ protocol . NodeID ] io . Closer ) ,
nodeVer : make ( map [ protocol . NodeID ] string ) ,
sentLocalVer : make ( map [ protocol . NodeID ] map [ string ] uint64 ) ,
sup : suppressor { threshold : int64 ( cfg . Options . MaxChangeKbps ) } ,
2013-12-15 10:43:31 +00:00
}
2014-06-26 09:24:38 +00:00
var timeout = 20 * 60 // seconds
if t := os . Getenv ( "STDEADLOCKTIMEOUT" ) ; len ( t ) > 0 {
it , err := strconv . Atoi ( t )
if err == nil {
timeout = it
}
}
deadlockDetect ( & m . rmut , time . Duration ( timeout ) * time . Second )
deadlockDetect ( & m . smut , time . Duration ( timeout ) * time . Second )
deadlockDetect ( & m . pmut , time . Duration ( timeout ) * time . Second )
2013-12-15 10:43:31 +00:00
return m
}
2014-01-06 10:11:18 +00:00
// 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.
2014-04-08 11:45:18 +00:00
func ( m * Model ) StartRepoRW ( repo string , threads int ) {
2014-05-02 15:14:53 +00:00
m . rmut . RLock ( )
defer m . rmut . RUnlock ( )
2014-03-29 17:53:48 +00:00
2014-05-23 12:31:16 +00:00
if cfg , ok := m . repoCfgs [ repo ] ; ! ok {
2014-03-29 17:53:48 +00:00
panic ( "cannot start without repo" )
2014-04-08 11:45:18 +00:00
} else {
2014-05-23 12:31:16 +00:00
newPuller ( cfg , m , threads , m . cfg )
2014-03-29 17:53:48 +00:00
}
2014-03-28 13:36:57 +00:00
}
2014-01-06 10:11:18 +00:00
2014-03-28 13:36:57 +00:00
// 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.
2014-04-08 11:45:18 +00:00
func ( m * Model ) StartRepoRO ( repo string ) {
m . StartRepoRW ( repo , 0 ) // zero threads => read only
2014-01-20 21:22:27 +00:00
}
2014-01-05 22:54:57 +00:00
type ConnectionInfo struct {
protocol . Statistics
2014-01-23 12:12:45 +00:00
Address string
ClientVersion string
2014-01-05 22:54:57 +00:00
}
2014-01-06 10:11:18 +00:00
// ConnectionStats returns a map with connection statistics for each connected node.
2014-01-05 22:54:57 +00:00
func ( m * Model ) ConnectionStats ( ) map [ string ] ConnectionInfo {
type remoteAddrer interface {
RemoteAddr ( ) net . Addr
}
2014-01-18 03:06:44 +00:00
m . pmut . RLock ( )
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-01-05 15:16:37 +00:00
2014-01-05 22:54:57 +00:00
var res = make ( map [ string ] ConnectionInfo )
2014-01-09 12:58:35 +00:00
for node , conn := range m . protoConn {
2014-01-05 22:54:57 +00:00
ci := ConnectionInfo {
2014-01-23 12:12:45 +00:00
Statistics : conn . Statistics ( ) ,
2014-04-13 13:28:26 +00:00
ClientVersion : m . nodeVer [ node ] ,
2014-01-05 22:54:57 +00:00
}
if nc , ok := m . rawConn [ node ] . ( remoteAddrer ) ; ok {
ci . Address = nc . RemoteAddr ( ) . String ( )
}
2014-02-13 11:41:37 +00:00
2014-06-29 23:42:03 +00:00
res [ node . String ( ) ] = ci
2013-12-30 14:30:29 +00:00
}
2014-01-18 03:06:44 +00:00
2014-03-29 17:53:48 +00:00
m . rmut . RUnlock ( )
2014-01-18 03:06:44 +00:00
m . pmut . RUnlock ( )
2014-03-28 13:36:57 +00:00
2014-05-24 19:34:11 +00:00
in , out := protocol . TotalInOut ( )
res [ "total" ] = ConnectionInfo {
Statistics : protocol . Statistics {
At : time . Now ( ) ,
2014-06-01 19:56:05 +00:00
InBytesTotal : in ,
OutBytesTotal : out ,
2014-05-24 19:34:11 +00:00
} ,
}
2014-01-05 15:16:37 +00:00
return res
2013-12-30 14:30:29 +00:00
}
2014-07-29 09:06:52 +00:00
// Returns the completion status, in percent, for the given node and repo.
func ( m * Model ) Completion ( node protocol . NodeID , repo string ) float64 {
var tot int64
m . repoFiles [ repo ] . WithGlobal ( func ( f protocol . FileInfo ) bool {
if ! protocol . IsDeleted ( f . Flags ) {
var size int64
if protocol . IsDirectory ( f . Flags ) {
size = zeroEntrySize
} else {
size = f . Size ( )
}
tot += size
}
return true
} )
var need int64
m . repoFiles [ repo ] . WithNeed ( node , func ( f protocol . FileInfo ) bool {
if ! protocol . IsDeleted ( f . Flags ) {
var size int64
if protocol . IsDirectory ( f . Flags ) {
size = zeroEntrySize
} else {
size = f . Size ( )
}
need += size
}
return true
} )
return 100 * ( 1 - float64 ( need ) / float64 ( tot ) )
}
2014-07-12 21:06:48 +00:00
func sizeOf ( fs [ ] protocol . FileInfo ) ( files , deleted int , bytes int64 ) {
2014-03-28 13:36:57 +00:00
for _ , f := range fs {
2014-07-06 12:46:48 +00:00
fs , de , by := sizeOfFile ( f )
files += fs
deleted += de
bytes += by
}
return
}
2014-07-12 21:06:48 +00:00
func sizeOfFile ( f protocol . FileInfo ) ( files , deleted int , bytes int64 ) {
2014-07-06 12:46:48 +00:00
if ! protocol . IsDeleted ( f . Flags ) {
files ++
if ! protocol . IsDirectory ( f . Flags ) {
2014-07-12 21:06:48 +00:00
bytes += f . Size ( )
2014-01-05 22:54:57 +00:00
} else {
2014-05-18 10:56:50 +00:00
bytes += zeroEntrySize
2014-01-05 22:54:57 +00:00
}
2014-07-06 12:46:48 +00:00
} else {
deleted ++
bytes += zeroEntrySize
2013-12-30 14:30:29 +00:00
}
2014-01-05 15:16:37 +00:00
return
}
2013-12-30 14:30:29 +00:00
2014-03-28 13:36:57 +00:00
// GlobalSize returns the number of files, deleted files and total bytes for all
// files in the global model.
2014-04-09 20:03:30 +00:00
func ( m * Model ) GlobalSize ( repo string ) ( files , deleted int , bytes int64 ) {
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-04-09 20:03:30 +00:00
defer m . rmut . RUnlock ( )
if rf , ok := m . repoFiles [ repo ] ; ok {
2014-07-12 21:06:48 +00:00
rf . WithGlobal ( func ( f protocol . FileInfo ) bool {
2014-07-06 12:46:48 +00:00
fs , de , by := sizeOfFile ( f )
files += fs
deleted += de
bytes += by
return true
} )
2014-03-29 17:53:48 +00:00
}
2014-07-06 12:46:48 +00:00
return
2014-03-28 13:36:57 +00:00
}
2014-01-06 10:11:18 +00:00
// LocalSize returns the number of files, deleted files and total bytes for all
// files in the local repository.
2014-04-09 20:03:30 +00:00
func ( m * Model ) LocalSize ( repo string ) ( files , deleted int , bytes int64 ) {
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-04-09 20:03:30 +00:00
defer m . rmut . RUnlock ( )
if rf , ok := m . repoFiles [ repo ] ; ok {
2014-07-12 21:06:48 +00:00
rf . WithHave ( protocol . LocalNodeID , func ( f protocol . FileInfo ) bool {
2014-07-06 12:46:48 +00:00
fs , de , by := sizeOfFile ( f )
files += fs
deleted += de
bytes += by
return true
} )
2014-03-29 17:53:48 +00:00
}
2014-07-06 21:15:28 +00:00
return
2014-01-06 05:38:01 +00:00
}
2014-05-19 20:31:28 +00:00
// NeedSize returns the number and total size of currently needed files.
2014-04-09 20:03:30 +00:00
func ( m * Model ) NeedSize ( repo string ) ( files int , bytes int64 ) {
2014-07-15 15:54:00 +00:00
m . rmut . RLock ( )
defer m . rmut . RUnlock ( )
if rf , ok := m . repoFiles [ repo ] ; ok {
rf . WithNeed ( protocol . LocalNodeID , func ( f protocol . FileInfo ) bool {
fs , de , by := sizeOfFile ( f )
files += fs + de
bytes += by
return true
} )
}
return
2013-12-23 17:12:44 +00:00
}
2014-07-06 12:46:48 +00:00
// NeedFiles returns the list of currently needed files
2014-07-12 21:06:48 +00:00
func ( m * Model ) NeedFilesRepo ( repo string ) [ ] protocol . FileInfo {
2014-04-01 21:18:32 +00:00
m . rmut . RLock ( )
2014-04-09 20:03:30 +00:00
defer m . rmut . RUnlock ( )
if rf , ok := m . repoFiles [ repo ] ; ok {
2014-07-15 15:54:00 +00:00
fs := make ( [ ] protocol . FileInfo , 0 , indexBatchSize )
2014-07-12 21:06:48 +00:00
rf . WithNeed ( protocol . LocalNodeID , func ( f protocol . FileInfo ) bool {
2014-07-06 12:46:48 +00:00
fs = append ( fs , f )
2014-07-15 15:54:00 +00:00
return len ( fs ) < indexBatchSize
2014-07-06 12:46:48 +00:00
} )
return fs
2014-04-09 20:03:30 +00:00
}
return nil
2014-04-01 21:18:32 +00:00
}
2013-12-30 14:30:29 +00:00
// Index is called when a new node is connected and we receive their full index.
2014-01-06 10:11:18 +00:00
// Implements the protocol.Model interface.
2014-06-29 23:42:03 +00:00
func ( m * Model ) Index ( nodeID protocol . NodeID , repo string , fs [ ] protocol . FileInfo ) {
2014-05-15 03:26:55 +00:00
if debug {
2014-05-19 20:31:28 +00:00
l . Debugf ( "IDX(in): %s %q: %d files" , nodeID , repo , len ( fs ) )
2014-03-29 17:53:48 +00:00
}
2014-06-06 19:48:29 +00:00
if ! m . repoSharedWith ( repo , nodeID ) {
l . Warnf ( "Unexpected repository ID %q sent from node %q; ensure that the repository exists and that this node is selected under \"Share With\" in the repository configuration." , repo , nodeID )
return
}
2014-07-18 09:41:51 +00:00
for i := range fs {
lamport . Default . Tick ( fs [ i ] . Version )
}
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-07-17 11:38:36 +00:00
r , ok := m . repoFiles [ repo ]
m . rmut . RUnlock ( )
if ok {
2014-07-12 21:06:48 +00:00
r . Replace ( nodeID , fs )
2014-03-29 17:53:48 +00:00
} else {
2014-06-06 19:48:29 +00:00
l . Fatalf ( "Index for nonexistant repo %q" , repo )
2013-12-15 10:43:31 +00:00
}
2014-07-13 19:07:24 +00:00
2014-07-17 11:38:36 +00:00
events . Default . Log ( events . RemoteIndexUpdated , map [ string ] interface { } {
"node" : nodeID . String ( ) ,
"repo" : repo ,
"items" : len ( fs ) ,
"version" : r . LocalVersion ( nodeID ) ,
2014-07-13 19:07:24 +00:00
} )
2013-12-28 13:10:36 +00:00
}
2013-12-30 14:30:29 +00:00
// IndexUpdate is called for incremental updates to connected nodes' indexes.
2014-01-06 10:11:18 +00:00
// Implements the protocol.Model interface.
2014-06-29 23:42:03 +00:00
func ( m * Model ) IndexUpdate ( nodeID protocol . NodeID , repo string , fs [ ] protocol . FileInfo ) {
2014-05-15 03:26:55 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l . Debugf ( "IDXUP(in): %s / %q: %d files" , nodeID , repo , len ( fs ) )
2014-03-29 17:53:48 +00:00
}
2014-06-06 19:48:29 +00:00
if ! m . repoSharedWith ( repo , nodeID ) {
l . Warnf ( "Unexpected repository ID %q sent from node %q; ensure that the repository exists and that this node is selected under \"Share With\" in the repository configuration." , repo , nodeID )
return
}
2014-07-18 09:41:51 +00:00
for i := range fs {
lamport . Default . Tick ( fs [ i ] . Version )
}
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-07-17 11:38:36 +00:00
r , ok := m . repoFiles [ repo ]
m . rmut . RUnlock ( )
if ok {
2014-07-12 21:06:48 +00:00
r . Update ( nodeID , fs )
2014-03-29 17:53:48 +00:00
} else {
2014-06-06 19:48:29 +00:00
l . Fatalf ( "IndexUpdate for nonexistant repo %q" , repo )
2013-12-28 13:10:36 +00:00
}
2014-07-13 19:07:24 +00:00
2014-07-17 11:38:36 +00:00
events . Default . Log ( events . RemoteIndexUpdated , map [ string ] interface { } {
"node" : nodeID . String ( ) ,
"repo" : repo ,
"items" : len ( fs ) ,
"version" : r . LocalVersion ( nodeID ) ,
2014-07-13 19:07:24 +00:00
} )
2014-01-09 09:59:09 +00:00
}
2014-06-29 23:42:03 +00:00
func ( m * Model ) repoSharedWith ( repo string , nodeID protocol . NodeID ) bool {
2014-06-06 19:48:29 +00:00
m . rmut . RLock ( )
defer m . rmut . RUnlock ( )
for _ , nrepo := range m . nodeRepos [ nodeID ] {
if nrepo == repo {
return true
}
}
return false
}
2014-06-29 23:42:03 +00:00
func ( m * Model ) ClusterConfig ( nodeID protocol . NodeID , config protocol . ClusterConfigMessage ) {
2014-04-13 13:28:26 +00:00
compErr := compareClusterConfig ( m . clusterConfig ( nodeID ) , config )
2014-05-15 03:26:55 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l . Debugf ( "ClusterConfig: %s: %#v" , nodeID , config )
l . Debugf ( " ... compare: %s: %v" , nodeID , compErr )
2014-04-13 13:28:26 +00:00
}
if compErr != nil {
2014-05-15 00:08:56 +00:00
l . Warnf ( "%s: %v" , nodeID , compErr )
2014-04-13 13:28:26 +00:00
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 ( )
2014-07-02 18:43:16 +00:00
l . Infof ( ` Node %s client is "%s %s" ` , nodeID , config . ClientName , config . ClientVersion )
2014-04-13 13:28:26 +00:00
}
2014-01-20 21:22:27 +00:00
// Close removes the peer from the model and closes the underlying connection if possible.
2014-01-06 10:11:18 +00:00
// Implements the protocol.Model interface.
2014-06-29 23:42:03 +00:00
func ( m * Model ) Close ( node protocol . NodeID , err error ) {
2014-06-23 13:06:20 +00:00
l . Infof ( "Connection to %s closed: %v" , node , err )
2014-07-13 19:07:24 +00:00
events . Default . Log ( events . NodeDisconnected , map [ string ] string {
"id" : node . String ( ) ,
"error" : err . Error ( ) ,
} )
2014-02-09 22:13:06 +00:00
2014-07-15 11:04:37 +00:00
m . pmut . Lock ( )
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
for _ , repo := range m . nodeRepos [ node ] {
2014-07-06 12:46:48 +00:00
m . repoFiles [ repo ] . Replace ( node , nil )
2014-03-29 17:53:48 +00:00
}
m . rmut . RUnlock ( )
2014-01-20 21:22:27 +00:00
2014-01-01 02:22:49 +00:00
conn , ok := m . rawConn [ node ]
2014-01-01 13:09:17 +00:00
if ok {
conn . Close ( )
2013-12-31 02:21:57 +00:00
}
2014-01-09 12:58:35 +00:00
delete ( m . protoConn , node )
2014-01-01 02:22:49 +00:00
delete ( m . rawConn , node )
2014-04-13 13:28:26 +00:00
delete ( m . nodeVer , node )
2014-01-18 03:06:44 +00:00
m . pmut . Unlock ( )
2013-12-15 10:43:31 +00:00
}
2014-01-06 10:11:18 +00:00
// Request returns the specified data segment by reading it from local disk.
// Implements the protocol.Model interface.
2014-06-29 23:42:03 +00:00
func ( m * Model ) Request ( nodeID protocol . NodeID , repo , name string , offset int64 , size int ) ( [ ] byte , error ) {
2014-03-28 13:36:57 +00:00
// Verify that the requested file exists in the local model.
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
r , ok := m . repoFiles [ repo ]
m . rmut . RUnlock ( )
if ! ok {
2014-05-15 00:08:56 +00:00
l . Warnf ( "Request from %s for file %s in nonexistent repo %q" , nodeID , name , repo )
2014-03-29 17:53:48 +00:00
return nil , ErrNoSuchFile
}
2014-07-06 12:46:48 +00:00
lf := r . Get ( protocol . LocalNodeID , name )
2014-07-12 21:06:48 +00:00
if protocol . IsInvalid ( lf . Flags ) || protocol . IsDeleted ( lf . Flags ) {
2014-05-20 18:26:44 +00:00
if debug {
l . Debugf ( "REQ(in): %s: %q / %q o=%d s=%d; invalid: %v" , nodeID , repo , name , offset , size , lf )
}
2014-05-11 17:54:26 +00:00
return nil , ErrInvalid
2014-01-06 20:31:36 +00:00
}
2014-03-29 17:53:48 +00:00
2014-07-12 21:06:48 +00:00
if offset > lf . Size ( ) {
2014-05-15 03:26:55 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l . Debugf ( "REQ(in; nonexistent): %s: %q o=%d s=%d" , nodeID , name , offset , size )
2014-05-11 17:54:26 +00:00
}
return nil , ErrNoSuchFile
2014-01-07 21:44:21 +00:00
}
2014-01-06 20:31:36 +00:00
2014-07-06 12:46:48 +00:00
if debug && nodeID != protocol . LocalNodeID {
2014-05-15 00:08:56 +00:00
l . Debugf ( "REQ(in): %s: %q / %q o=%d s=%d" , nodeID , repo , name , offset , size )
2013-12-15 10:43:31 +00:00
}
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-05-23 12:31:16 +00:00
fn := filepath . Join ( m . repoCfgs [ repo ] . Directory , name )
2014-03-29 17:53:48 +00:00
m . rmut . RUnlock ( )
2013-12-15 10:43:31 +00:00
fd , err := os . Open ( fn ) // XXX: Inefficient, should cache fd?
if err != nil {
return nil , err
}
defer fd . Close ( )
2014-06-18 21:57:22 +00:00
buf := make ( [ ] byte , size )
2014-01-09 15:35:49 +00:00
_ , err = fd . ReadAt ( buf , offset )
2013-12-15 10:43:31 +00:00
if err != nil {
return nil , err
}
return buf , nil
}
2014-01-06 10:11:18 +00:00
// ReplaceLocal replaces the local repository index with the given list of files.
2014-07-12 21:06:48 +00:00
func ( m * Model ) ReplaceLocal ( repo string , fs [ ] protocol . FileInfo ) {
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-07-06 12:46:48 +00:00
m . repoFiles [ repo ] . ReplaceWithDelete ( protocol . LocalNodeID , fs )
2014-03-29 17:53:48 +00:00
m . rmut . RUnlock ( )
2013-12-15 10:43:31 +00:00
}
2014-07-12 21:06:48 +00:00
func ( m * Model ) CurrentRepoFile ( repo string , file string ) protocol . FileInfo {
2014-04-01 21:18:32 +00:00
m . rmut . RLock ( )
2014-07-06 12:46:48 +00:00
f := m . repoFiles [ repo ] . Get ( protocol . LocalNodeID , file )
2014-04-01 21:18:32 +00:00
m . rmut . RUnlock ( )
return f
}
2014-07-12 21:06:48 +00:00
func ( m * Model ) CurrentGlobalFile ( repo string , file string ) protocol . FileInfo {
2014-04-01 21:18:32 +00:00
m . rmut . RLock ( )
f := m . repoFiles [ repo ] . GetGlobal ( file )
m . rmut . RUnlock ( )
return f
}
2014-03-29 17:53:48 +00:00
type cFiler struct {
m * Model
r string
2014-01-06 10:11:18 +00:00
}
2014-03-16 07:14:55 +00:00
// Implements scanner.CurrentFiler
2014-07-12 21:06:48 +00:00
func ( cf cFiler ) CurrentFile ( file string ) protocol . FileInfo {
2014-04-01 21:18:32 +00:00
return cf . m . CurrentRepoFile ( cf . r , file )
2014-03-16 07:14:55 +00:00
}
2014-01-06 10:11:18 +00:00
// ConnectedTo returns true if we are connected to the named node.
2014-06-29 23:42:03 +00:00
func ( m * Model ) ConnectedTo ( nodeID protocol . NodeID ) bool {
2014-01-18 03:06:44 +00:00
m . pmut . RLock ( )
2014-01-09 12:58:35 +00:00
_ , ok := m . protoConn [ nodeID ]
2014-01-18 03:06:44 +00:00
m . pmut . RUnlock ( )
2014-01-06 10:11:18 +00:00
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.
2014-03-28 13:36:57 +00:00
func ( m * Model ) AddConnection ( rawConn io . Closer , protoConn protocol . Connection ) {
2014-01-09 12:58:35 +00:00
nodeID := protoConn . ID ( )
2014-07-15 11:04:37 +00:00
2014-01-18 03:06:44 +00:00
m . pmut . Lock ( )
2014-03-23 07:45:05 +00:00
if _ , ok := m . protoConn [ nodeID ] ; ok {
panic ( "add existing node" )
}
2014-01-09 12:58:35 +00:00
m . protoConn [ nodeID ] = protoConn
2014-03-23 07:45:05 +00:00
if _ , ok := m . rawConn [ nodeID ] ; ok {
panic ( "add existing node" )
}
2014-01-09 12:58:35 +00:00
m . rawConn [ nodeID ] = rawConn
2014-01-06 10:11:18 +00:00
2014-04-13 13:28:26 +00:00
cm := m . clusterConfig ( nodeID )
protoConn . ClusterConfig ( cm )
2014-05-04 15:18:58 +00:00
m . rmut . RLock ( )
for _ , repo := range m . nodeRepos [ nodeID ] {
2014-07-15 11:04:37 +00:00
fs := m . repoFiles [ repo ]
go sendIndexes ( protoConn , repo , fs )
2014-05-04 15:18:58 +00:00
}
m . rmut . RUnlock ( )
2014-07-15 11:04:37 +00:00
m . pmut . Unlock ( )
}
func sendIndexes ( conn protocol . Connection , repo string , fs * files . Set ) {
nodeID := conn . ID ( )
name := conn . Name ( )
2014-07-30 18:08:04 +00:00
var err error
2014-07-15 11:04:37 +00:00
if debug {
l . Debugf ( "sendIndexes for %s-%s@/%q starting" , nodeID , name , repo )
}
2014-05-04 15:18:58 +00:00
2014-07-15 11:04:37 +00:00
defer func ( ) {
if debug {
l . Debugf ( "sendIndexes for %s-%s@/%q exiting: %v" , nodeID , name , repo , err )
}
} ( )
2014-07-30 18:08:04 +00:00
minLocalVer , err := sendIndexTo ( true , 0 , conn , repo , fs )
2014-07-15 11:04:37 +00:00
for err == nil {
2014-07-30 18:08:04 +00:00
time . Sleep ( 5 * time . Second )
if fs . LocalVersion ( protocol . LocalNodeID ) <= minLocalVer {
continue
2014-07-15 11:04:37 +00:00
}
2014-07-30 18:08:04 +00:00
minLocalVer , err = sendIndexTo ( false , minLocalVer , conn , repo , fs )
}
}
2014-07-15 11:04:37 +00:00
2014-07-30 18:08:04 +00:00
func sendIndexTo ( initial bool , minLocalVer uint64 , conn protocol . Connection , repo string , fs * files . Set ) ( uint64 , error ) {
nodeID := conn . ID ( )
name := conn . Name ( )
batch := make ( [ ] protocol . FileInfo , 0 , indexBatchSize )
maxLocalVer := uint64 ( 0 )
var err error
2014-07-15 11:04:37 +00:00
2014-07-30 18:08:04 +00:00
fs . WithHave ( protocol . LocalNodeID , func ( f protocol . FileInfo ) bool {
if f . LocalVersion <= minLocalVer {
return true
}
2014-07-15 11:04:37 +00:00
2014-07-30 18:08:04 +00:00
if f . LocalVersion > maxLocalVer {
maxLocalVer = f . LocalVersion
}
2014-07-15 11:04:37 +00:00
2014-07-30 18:08:04 +00:00
if len ( batch ) == indexBatchSize {
if initial {
if err = conn . Index ( repo , batch ) ; err != nil {
return false
}
if debug {
l . Debugf ( "sendIndexes for %s-%s/%q: %d files (initial index)" , nodeID , name , repo , len ( batch ) )
}
initial = false
} else {
if err = conn . IndexUpdate ( repo , batch ) ; err != nil {
return false
}
if debug {
l . Debugf ( "sendIndexes for %s-%s/%q: %d files (batched update)" , nodeID , name , repo , len ( batch ) )
}
2014-07-03 10:30:10 +00:00
}
2014-01-06 10:11:18 +00:00
2014-07-30 18:08:04 +00:00
batch = make ( [ ] protocol . FileInfo , 0 , indexBatchSize )
2014-07-15 11:04:37 +00:00
}
2014-07-30 18:08:04 +00:00
batch = append ( batch , f )
return true
} )
if initial && err == nil {
err = conn . Index ( repo , batch )
if debug && err == nil {
l . Debugf ( "sendIndexes for %s-%s/%q: %d files (small initial index)" , nodeID , name , repo , len ( batch ) )
}
} else if len ( batch ) > 0 && err == nil {
err = conn . IndexUpdate ( repo , batch )
if debug && err == nil {
l . Debugf ( "sendIndexes for %s-%s/%q: %d files (last batch)" , nodeID , name , repo , len ( batch ) )
}
2014-07-15 11:04:37 +00:00
}
2014-07-30 18:08:04 +00:00
return maxLocalVer , err
2014-01-06 10:11:18 +00:00
}
2014-07-12 21:06:48 +00:00
func ( m * Model ) updateLocal ( repo string , f protocol . FileInfo ) {
2014-07-15 11:04:37 +00:00
f . LocalVersion = 0
2014-03-29 17:53:48 +00:00
m . rmut . RLock ( )
2014-07-12 21:06:48 +00:00
m . repoFiles [ repo ] . Update ( protocol . LocalNodeID , [ ] protocol . FileInfo { f } )
2014-03-29 17:53:48 +00:00
m . rmut . RUnlock ( )
2014-07-17 11:38:36 +00:00
events . Default . Log ( events . LocalIndexUpdated , map [ string ] interface { } {
"repo" : repo ,
"name" : f . Name ,
"modified" : time . Unix ( f . Modified , 0 ) ,
"flags" : fmt . Sprintf ( "0%o" , f . Flags ) ,
"size" : f . Size ( ) ,
} )
2014-03-28 13:36:57 +00:00
}
2014-06-29 23:42:03 +00:00
func ( m * Model ) requestGlobal ( nodeID protocol . NodeID , repo , name string , offset int64 , size int , hash [ ] byte ) ( [ ] byte , error ) {
2014-01-18 03:06:44 +00:00
m . pmut . RLock ( )
2014-01-09 12:58:35 +00:00
nc , ok := m . protoConn [ nodeID ]
2014-01-18 03:06:44 +00:00
m . pmut . RUnlock ( )
2014-01-06 10:11:18 +00:00
if ! ok {
return nil , fmt . Errorf ( "requestGlobal: no such node: %s" , nodeID )
}
2014-05-15 03:26:55 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l . Debugf ( "REQ(out): %s: %q / %q o=%d s=%d h=%x" , nodeID , repo , name , offset , size , hash )
2014-01-06 10:11:18 +00:00
}
2014-03-29 17:53:48 +00:00
return nc . Request ( repo , name , offset , size )
2014-01-06 10:11:18 +00:00
}
2014-05-23 12:31:16 +00:00
func ( m * Model ) AddRepo ( cfg config . RepositoryConfiguration ) {
2014-03-29 17:53:48 +00:00
if m . started {
panic ( "cannot add repo to started model" )
}
2014-05-23 12:31:16 +00:00
if len ( cfg . ID ) == 0 {
2014-03-29 17:53:48 +00:00
panic ( "cannot add empty repo id" )
}
m . rmut . Lock ( )
2014-05-23 12:31:16 +00:00
m . repoCfgs [ cfg . ID ] = cfg
2014-07-06 12:46:48 +00:00
m . repoFiles [ cfg . ID ] = files . NewSet ( cfg . ID , m . db )
2014-05-23 12:31:16 +00:00
m . suppressor [ cfg . ID ] = & suppressor { threshold : int64 ( m . cfg . Options . MaxChangeKbps ) }
2013-12-15 10:43:31 +00:00
2014-06-29 23:42:03 +00:00
m . repoNodes [ cfg . ID ] = make ( [ ] protocol . NodeID , len ( cfg . Nodes ) )
2014-05-23 12:31:16 +00:00
for i , node := range cfg . Nodes {
m . repoNodes [ cfg . ID ] [ i ] = node . NodeID
m . nodeRepos [ node . NodeID ] = append ( m . nodeRepos [ node . NodeID ] , cfg . ID )
2014-03-29 17:53:48 +00:00
}
2014-01-23 21:20:15 +00:00
2014-03-29 17:53:48 +00:00
m . addedRepo = true
m . rmut . Unlock ( )
}
2014-01-23 21:20:15 +00:00
2014-03-29 17:53:48 +00:00
func ( m * Model ) ScanRepos ( ) {
m . rmut . RLock ( )
2014-05-23 12:31:16 +00:00
var repos = make ( [ ] string , 0 , len ( m . repoCfgs ) )
for repo := range m . repoCfgs {
2014-04-14 07:58:17 +00:00
repos = append ( repos , repo )
2014-03-29 17:53:48 +00:00
}
m . rmut . RUnlock ( )
2014-04-14 07:58:17 +00:00
2014-05-13 23:42:12 +00:00
var wg sync . WaitGroup
wg . Add ( len ( repos ) )
2014-04-14 07:58:17 +00:00
for _ , repo := range repos {
2014-05-13 23:42:12 +00:00
repo := repo
go func ( ) {
2014-05-28 04:55:30 +00:00
err := m . ScanRepo ( repo )
if err != nil {
invalidateRepo ( m . cfg , repo , err )
}
2014-05-13 23:42:12 +00:00
wg . Done ( )
} ( )
2014-04-14 07:58:17 +00:00
}
2014-05-13 23:42:12 +00:00
wg . Wait ( )
2014-03-29 17:53:48 +00:00
}
2013-12-15 10:43:31 +00:00
2014-05-15 03:26:55 +00:00
func ( m * Model ) CleanRepos ( ) {
m . rmut . RLock ( )
2014-05-23 12:31:16 +00:00
var dirs = make ( [ ] string , 0 , len ( m . repoCfgs ) )
for _ , cfg := range m . repoCfgs {
dirs = append ( dirs , cfg . Directory )
2014-05-15 03:26:55 +00:00
}
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 ( )
}
2014-05-04 16:20:25 +00:00
func ( m * Model ) ScanRepo ( repo string ) error {
2014-05-02 15:14:53 +00:00
m . rmut . RLock ( )
2014-07-15 12:27:46 +00:00
fs := m . repoFiles [ repo ]
dir := m . repoCfgs [ repo ] . Directory
2014-03-29 17:53:48 +00:00
w := & scanner . Walker {
2014-07-15 12:27:46 +00:00
Dir : dir ,
2014-04-08 11:45:18 +00:00
IgnoreFile : ".stignore" ,
2014-05-15 03:26:55 +00:00
BlockSize : scanner . StandardBlockSize ,
2014-04-08 11:45:18 +00:00
TempNamer : defTempNamer ,
2014-05-15 03:26:55 +00:00
Suppressor : m . suppressor [ repo ] ,
2014-04-08 11:45:18 +00:00
CurrentFiler : cFiler { m , repo } ,
2014-05-23 12:31:16 +00:00
IgnorePerms : m . repoCfgs [ repo ] . IgnorePerms ,
2014-03-29 17:53:48 +00:00
}
2014-05-02 15:14:53 +00:00
m . rmut . RUnlock ( )
2014-07-15 12:27:46 +00:00
2014-04-14 07:58:17 +00:00
m . setState ( repo , RepoScanning )
2014-07-15 12:27:46 +00:00
fchan , _ , err := w . Walk ( )
2014-05-04 16:20:25 +00:00
if err != nil {
return err
}
2014-07-15 15:54:00 +00:00
batchSize := 100
batch := make ( [ ] protocol . FileInfo , 0 , 00 )
2014-07-15 12:27:46 +00:00
for f := range fchan {
2014-07-29 09:53:45 +00:00
events . Default . Log ( events . LocalIndexUpdated , map [ string ] interface { } {
"repo" : repo ,
"name" : f . Name ,
"modified" : time . Unix ( f . Modified , 0 ) ,
"flags" : fmt . Sprintf ( "0%o" , f . Flags ) ,
"size" : f . Size ( ) ,
} )
2014-07-15 15:54:00 +00:00
if len ( batch ) == batchSize {
2014-07-15 12:27:46 +00:00
fs . Update ( protocol . LocalNodeID , batch )
batch = batch [ : 0 ]
}
batch = append ( batch , f )
}
if len ( batch ) > 0 {
fs . Update ( protocol . LocalNodeID , batch )
}
batch = batch [ : 0 ]
fs . WithHave ( protocol . LocalNodeID , func ( f protocol . FileInfo ) bool {
if ! protocol . IsDeleted ( f . Flags ) {
2014-07-15 15:54:00 +00:00
if len ( batch ) == batchSize {
2014-07-15 12:27:46 +00:00
fs . Update ( protocol . LocalNodeID , batch )
batch = batch [ : 0 ]
}
if _ , err := os . Stat ( filepath . Join ( dir , f . Name ) ) ; err != nil && os . IsNotExist ( err ) {
// File has been deleted
f . Blocks = nil
f . Flags |= protocol . FlagDeleted
f . Version = lamport . Default . Tick ( f . Version )
f . LocalVersion = 0
2014-07-29 09:53:45 +00:00
events . Default . Log ( events . LocalIndexUpdated , map [ string ] interface { } {
"repo" : repo ,
"name" : f . Name ,
"modified" : time . Unix ( f . Modified , 0 ) ,
"flags" : fmt . Sprintf ( "0%o" , f . Flags ) ,
"size" : f . Size ( ) ,
} )
2014-07-15 12:27:46 +00:00
batch = append ( batch , f )
}
}
return true
} )
if len ( batch ) > 0 {
fs . Update ( protocol . LocalNodeID , batch )
}
2014-04-14 07:58:17 +00:00
m . setState ( repo , RepoIdle )
2014-05-04 16:20:25 +00:00
return nil
2014-03-29 17:53:48 +00:00
}
2014-04-13 13:28:26 +00:00
// clusterConfig returns a ClusterConfigMessage that is correct for the given peer node
2014-06-29 23:42:03 +00:00
func ( m * Model ) clusterConfig ( node protocol . NodeID ) protocol . ClusterConfigMessage {
2014-04-13 13:28:26 +00:00
cm := protocol . ClusterConfigMessage {
2014-05-15 03:26:55 +00:00
ClientName : m . clientName ,
ClientVersion : m . clientVersion ,
2014-04-13 13:28:26 +00:00
}
2014-05-02 15:14:53 +00:00
m . rmut . RLock ( )
2014-04-13 13:28:26 +00:00
for _ , repo := range m . nodeRepos [ node ] {
cr := protocol . Repository {
ID : repo ,
2014-01-09 12:58:35 +00:00
}
2014-04-13 13:28:26 +00:00
for _ , node := range m . repoNodes [ repo ] {
// TODO: Set read only bit when relevant
cr . Nodes = append ( cr . Nodes , protocol . Node {
2014-06-29 23:42:03 +00:00
ID : node [ : ] ,
2014-04-13 13:28:26 +00:00
Flags : protocol . FlagShareTrusted ,
} )
2014-01-09 12:58:35 +00:00
}
2014-04-13 13:28:26 +00:00
cm . Repositories = append ( cm . Repositories , cr )
2013-12-30 01:33:57 +00:00
}
2014-05-02 15:14:53 +00:00
m . rmut . RUnlock ( )
2014-04-13 13:28:26 +00:00
return cm
2013-12-30 01:33:57 +00:00
}
2014-04-14 07:58:17 +00:00
func ( m * Model ) setState ( repo string , state repoState ) {
2014-05-20 16:41:01 +00:00
m . smut . Lock ( )
2014-07-17 11:38:36 +00:00
oldState := m . repoState [ repo ]
changed , ok := m . repoStateChanged [ repo ]
if state != oldState {
m . repoState [ repo ] = state
m . repoStateChanged [ repo ] = time . Now ( )
eventData := map [ string ] interface { } {
"repo" : repo ,
"to" : state . String ( ) ,
}
if ok {
eventData [ "duration" ] = time . Since ( changed ) . Seconds ( )
eventData [ "from" ] = oldState . String ( )
}
events . Default . Log ( events . StateChanged , eventData )
}
2014-05-20 16:41:01 +00:00
m . smut . Unlock ( )
2014-04-14 07:58:17 +00:00
}
2014-07-17 11:38:36 +00:00
func ( m * Model ) State ( repo string ) ( string , time . Time ) {
2014-05-20 16:41:01 +00:00
m . smut . RLock ( )
2014-04-14 07:58:17 +00:00
state := m . repoState [ repo ]
2014-07-17 11:38:36 +00:00
changed := m . repoStateChanged [ repo ]
2014-05-20 16:41:01 +00:00
m . smut . RUnlock ( )
2014-07-17 11:38:36 +00:00
return state . String ( ) , changed
2014-04-14 07:58:17 +00:00
}
2014-06-16 08:47:02 +00:00
func ( m * Model ) Override ( repo string ) {
2014-06-23 09:52:13 +00:00
m . rmut . RLock ( )
2014-07-15 15:54:00 +00:00
fs := m . repoFiles [ repo ]
2014-06-23 09:52:13 +00:00
m . rmut . RUnlock ( )
2014-07-15 15:54:00 +00:00
batch := make ( [ ] protocol . FileInfo , 0 , indexBatchSize )
fs . WithNeed ( protocol . LocalNodeID , func ( need protocol . FileInfo ) bool {
if len ( batch ) == indexBatchSize {
fs . Update ( protocol . LocalNodeID , batch )
batch = batch [ : 0 ]
}
have := fs . Get ( protocol . LocalNodeID , need . Name )
if have . Name != need . Name {
2014-06-16 08:47:02 +00:00
// We are missing the file
2014-07-15 15:54:00 +00:00
need . Flags |= protocol . FlagDeleted
need . Blocks = nil
2014-06-16 08:47:02 +00:00
} else {
// We have the file, replace with our version
2014-07-15 15:54:00 +00:00
need = have
2014-06-16 08:47:02 +00:00
}
2014-07-15 15:54:00 +00:00
need . Version = lamport . Default . Tick ( need . Version )
need . LocalVersion = 0
batch = append ( batch , need )
return true
} )
if len ( batch ) > 0 {
fs . Update ( protocol . LocalNodeID , batch )
2014-06-16 08:47:02 +00:00
}
}
2014-06-19 22:27:54 +00:00
// Version returns the change version for the given repository. This is
// guaranteed to increment if the contents of the local or global repository
// has changed.
2014-07-15 11:04:37 +00:00
func ( m * Model ) LocalVersion ( repo string ) uint64 {
2014-06-19 22:27:54 +00:00
m . rmut . Lock ( )
2014-07-15 15:54:00 +00:00
defer m . rmut . Unlock ( )
fs , ok := m . repoFiles [ repo ]
if ! ok {
return 0
}
ver := fs . LocalVersion ( protocol . LocalNodeID )
2014-06-19 22:27:54 +00:00
for _ , n := range m . repoNodes [ repo ] {
2014-07-15 15:54:00 +00:00
ver += fs . LocalVersion ( n )
2014-06-19 22:27:54 +00:00
}
return ver
}