2014-06-01 22:50:14 +02:00
// Copyright (C) 2014 Jakob Borg and other contributors. All rights reserved.
// Use of this source code is governed by an MIT-style license that can be
// found in the LICENSE file.
2014-05-15 05:26:55 +02:00
package model
2013-12-15 11:43:31 +01:00
import (
2014-03-29 18:53:48 +01:00
"compress/gzip"
2014-01-06 11:11:18 +01:00
"crypto/sha1"
2014-01-06 21:31:36 +01:00
"errors"
2013-12-23 18:12:44 +01:00
"fmt"
2014-01-01 03:22:49 +01:00
"io"
2014-01-05 23:54:57 +01:00
"net"
2013-12-15 11:43:31 +01:00
"os"
2014-03-28 14:36:57 +01:00
"path/filepath"
2013-12-15 11:43:31 +01:00
"sync"
"time"
2014-06-21 09:43:12 +02:00
2014-03-28 14:36:57 +01:00
"github.com/calmh/syncthing/cid"
2014-05-15 02:18:09 +02:00
"github.com/calmh/syncthing/config"
2014-03-28 14:36:57 +01:00
"github.com/calmh/syncthing/files"
"github.com/calmh/syncthing/lamport"
2014-05-25 20:49:08 +02:00
"github.com/calmh/syncthing/osutil"
2013-12-15 11:43:31 +01:00
"github.com/calmh/syncthing/protocol"
2014-03-08 23:02:01 +01:00
"github.com/calmh/syncthing/scanner"
2013-12-15 11:43:31 +01:00
)
2014-04-14 09:58:17 +02:00
type repoState int
const (
RepoIdle repoState = iota
RepoScanning
RepoSyncing
RepoCleaning
)
2014-05-18 12:56:50 +02: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
2013-12-15 11:43:31 +01:00
type Model struct {
2014-05-15 05:26:55 +02:00
indexDir string
cfg * config . Configuration
clientName string
clientVersion string
2014-05-23 14:31:16 +02:00
repoCfgs map [ string ] config . RepositoryConfiguration // repo -> cfg
repoFiles map [ string ] * files . Set // repo -> files
repoNodes map [ string ] [ ] string // repo -> nodeIDs
nodeRepos map [ string ] [ ] string // nodeID -> repos
suppressor map [ string ] * suppressor // repo -> suppressor
rmut sync . RWMutex // protects the above
2014-03-29 18:53:48 +01:00
2014-05-20 18:41:01 +02:00
repoState map [ string ] repoState // repo -> state
smut sync . RWMutex
2014-03-29 18:53:48 +01:00
cm * cid . Map
2013-12-30 15:30:29 +01:00
2014-03-28 14:36:57 +01:00
protoConn map [ string ] protocol . Connection
2014-01-09 13:58:35 +01:00
rawConn map [ string ] io . Closer
2014-04-13 15:28:26 +02:00
nodeVer map [ string ] string
2014-01-18 04:06:44 +01:00
pmut sync . RWMutex // protects protoConn and rawConn
2013-12-30 15:30:29 +01:00
2014-01-22 11:38:52 +01:00
sup suppressor
2014-01-12 16:59:35 +01:00
2014-03-29 18:53:48 +01:00
addedRepo bool
started bool
2013-12-15 11:43:31 +01:00
}
2014-01-07 22:44:21 +01:00
var (
ErrNoSuchFile = errors . New ( "no such file" )
ErrInvalid = errors . New ( "file is invalid" )
)
2014-01-06 21:31:36 +01:00
2014-01-06 11:11:18 +01: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-05-15 05:26:55 +02:00
func NewModel ( indexDir string , cfg * config . Configuration , clientName , clientVersion string ) * Model {
2013-12-15 11:43:31 +01:00
m := & Model {
2014-05-15 05:26:55 +02:00
indexDir : indexDir ,
cfg : cfg ,
clientName : clientName ,
clientVersion : clientVersion ,
2014-05-23 14:31:16 +02:00
repoCfgs : make ( map [ string ] config . RepositoryConfiguration ) ,
2014-05-15 05:26:55 +02:00
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 ) } ,
2013-12-15 11:43:31 +01:00
}
2014-06-21 09:43:12 +02:00
deadlockDetect ( & m . rmut , 60 * time . Second )
deadlockDetect ( & m . smut , 60 * time . Second )
deadlockDetect ( & m . pmut , 60 * time . Second )
2013-12-24 21:21:03 +01:00
go m . broadcastIndexLoop ( )
2013-12-15 11:43:31 +01:00
return m
}
2014-01-06 11:11:18 +01: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 13:45:18 +02:00
func ( m * Model ) StartRepoRW ( repo string , threads int ) {
2014-05-02 17:14:53 +02:00
m . rmut . RLock ( )
defer m . rmut . RUnlock ( )
2014-03-29 18:53:48 +01:00
2014-05-23 14:31:16 +02:00
if cfg , ok := m . repoCfgs [ repo ] ; ! ok {
2014-03-29 18:53:48 +01:00
panic ( "cannot start without repo" )
2014-04-08 13:45:18 +02:00
} else {
2014-05-23 14:31:16 +02:00
newPuller ( cfg , m , threads , m . cfg )
2014-03-29 18:53:48 +01:00
}
2014-03-28 14:36:57 +01:00
}
2014-01-06 11:11:18 +01:00
2014-03-28 14:36:57 +01: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 13:45:18 +02:00
func ( m * Model ) StartRepoRO ( repo string ) {
m . StartRepoRW ( repo , 0 ) // zero threads => read only
2014-01-20 22:22:27 +01:00
}
2014-01-05 23:54:57 +01:00
type ConnectionInfo struct {
protocol . Statistics
2014-01-23 13:12:45 +01:00
Address string
ClientVersion string
2014-02-13 12:41:37 +01:00
Completion int
2014-01-05 23:54:57 +01:00
}
2014-01-06 11:11:18 +01:00
// ConnectionStats returns a map with connection statistics for each connected node.
2014-01-05 23:54:57 +01:00
func ( m * Model ) ConnectionStats ( ) map [ string ] ConnectionInfo {
type remoteAddrer interface {
RemoteAddr ( ) net . Addr
}
2014-01-18 04:06:44 +01:00
m . pmut . RLock ( )
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
2014-01-05 16:16:37 +01:00
2014-01-05 23:54:57 +01:00
var res = make ( map [ string ] ConnectionInfo )
2014-01-09 13:58:35 +01:00
for node , conn := range m . protoConn {
2014-01-05 23:54:57 +01:00
ci := ConnectionInfo {
2014-01-23 13:12:45 +01:00
Statistics : conn . Statistics ( ) ,
2014-04-13 15:28:26 +02:00
ClientVersion : m . nodeVer [ node ] ,
2014-01-05 23:54:57 +01:00
}
if nc , ok := m . rawConn [ node ] . ( remoteAddrer ) ; ok {
ci . Address = nc . RemoteAddr ( ) . String ( )
}
2014-02-13 12:41:37 +01:00
2014-03-29 18:53:48 +01:00
var tot int64
var have int64
for _ , repo := range m . nodeRepos [ node ] {
for _ , f := range m . repoFiles [ repo ] . Global ( ) {
2014-05-23 12:53:26 +02:00
if ! protocol . IsDeleted ( f . Flags ) {
2014-05-18 12:56:50 +02:00
size := f . Size
2014-05-23 12:53:26 +02:00
if protocol . IsDirectory ( f . Flags ) {
2014-05-18 12:56:50 +02:00
size = zeroEntrySize
}
tot += size
have += size
2014-03-29 18:53:48 +01:00
}
}
for _ , f := range m . repoFiles [ repo ] . Need ( m . cm . Get ( node ) ) {
2014-05-23 12:53:26 +02:00
if ! protocol . IsDeleted ( f . Flags ) {
2014-05-18 12:56:50 +02:00
size := f . Size
2014-05-23 12:53:26 +02:00
if protocol . IsDirectory ( f . Flags ) {
2014-05-18 12:56:50 +02:00
size = zeroEntrySize
}
have -= size
2014-03-29 18:53:48 +01:00
}
2014-02-13 12:41:37 +01:00
}
}
2014-03-03 08:47:52 +01:00
ci . Completion = 100
if tot != 0 {
ci . Completion = int ( 100 * have / tot )
}
2014-02-13 12:41:37 +01:00
2014-01-05 23:54:57 +01:00
res [ node ] = ci
2013-12-30 15:30:29 +01:00
}
2014-01-18 04:06:44 +01:00
2014-03-29 18:53:48 +01:00
m . rmut . RUnlock ( )
2014-01-18 04:06:44 +01:00
m . pmut . RUnlock ( )
2014-03-28 14:36:57 +01:00
2014-05-24 21:34:11 +02:00
in , out := protocol . TotalInOut ( )
res [ "total" ] = ConnectionInfo {
Statistics : protocol . Statistics {
At : time . Now ( ) ,
2014-06-01 21:56:05 +02:00
InBytesTotal : in ,
OutBytesTotal : out ,
2014-05-24 21:34:11 +02:00
} ,
}
2014-01-05 16:16:37 +01:00
return res
2013-12-30 15:30:29 +01:00
}
2014-03-28 14:36:57 +01:00
func sizeOf ( fs [ ] scanner . File ) ( files , deleted int , bytes int64 ) {
for _ , f := range fs {
2014-05-23 12:53:26 +02:00
if ! protocol . IsDeleted ( f . Flags ) {
2014-01-05 23:54:57 +01:00
files ++
2014-05-23 12:53:26 +02:00
if ! protocol . IsDirectory ( f . Flags ) {
2014-05-18 12:56:50 +02:00
bytes += f . Size
} else {
bytes += zeroEntrySize
}
2014-01-05 23:54:57 +01:00
} else {
deleted ++
2014-05-18 12:56:50 +02:00
bytes += zeroEntrySize
2014-01-05 23:54:57 +01:00
}
2013-12-30 15:30:29 +01:00
}
2014-01-05 16:16:37 +01:00
return
}
2013-12-30 15:30:29 +01:00
2014-03-28 14:36:57 +01:00
// GlobalSize returns the number of files, deleted files and total bytes for all
// files in the global model.
2014-04-09 22:03:30 +02:00
func ( m * Model ) GlobalSize ( repo string ) ( files , deleted int , bytes int64 ) {
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
2014-04-09 22:03:30 +02:00
defer m . rmut . RUnlock ( )
if rf , ok := m . repoFiles [ repo ] ; ok {
return sizeOf ( rf . Global ( ) )
2014-03-29 18:53:48 +01:00
}
2014-04-09 22:03:30 +02:00
return 0 , 0 , 0
2014-03-28 14:36:57 +01:00
}
2014-01-06 11:11:18 +01:00
// LocalSize returns the number of files, deleted files and total bytes for all
// files in the local repository.
2014-04-09 22:03:30 +02:00
func ( m * Model ) LocalSize ( repo string ) ( files , deleted int , bytes int64 ) {
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
2014-04-09 22:03:30 +02:00
defer m . rmut . RUnlock ( )
if rf , ok := m . repoFiles [ repo ] ; ok {
return sizeOf ( rf . Have ( cid . LocalID ) )
2014-03-29 18:53:48 +01:00
}
2014-04-09 22:03:30 +02:00
return 0 , 0 , 0
2014-01-06 06:38:01 +01:00
}
2014-05-19 22:31:28 +02:00
// NeedSize returns the number and total size of currently needed files.
2014-04-09 22:03:30 +02:00
func ( m * Model ) NeedSize ( repo string ) ( files int , bytes int64 ) {
2014-05-18 12:56:50 +02:00
f , d , b := sizeOf ( m . NeedFilesRepo ( repo ) )
return f + d , b
2013-12-23 18:12:44 +01:00
}
2014-04-01 23:18:32 +02:00
// NeedFiles returns the list of currently needed files and the total size.
func ( m * Model ) NeedFilesRepo ( repo string ) [ ] scanner . File {
m . rmut . RLock ( )
2014-04-09 22:03:30 +02:00
defer m . rmut . RUnlock ( )
if rf , ok := m . repoFiles [ repo ] ; ok {
2014-06-07 05:10:15 +02:00
f := rf . Need ( cid . LocalID )
if r := m . repoCfgs [ repo ] . FileRanker ( ) ; r != nil {
files . SortBy ( r ) . Sort ( f )
}
return f
2014-04-09 22:03:30 +02:00
}
return nil
2014-04-01 23:18:32 +02:00
}
2013-12-30 15:30:29 +01:00
// Index is called when a new node is connected and we receive their full index.
2014-01-06 11:11:18 +01:00
// Implements the protocol.Model interface.
2014-03-29 18:53:48 +01:00
func ( m * Model ) Index ( nodeID string , repo string , fs [ ] protocol . FileInfo ) {
2014-05-15 05:26:55 +02:00
if debug {
2014-05-19 22:31:28 +02:00
l . Debugf ( "IDX(in): %s %q: %d files" , nodeID , repo , len ( fs ) )
2014-03-29 18:53:48 +01:00
}
2014-06-06 21:48:29 +02: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-03-08 23:02:01 +01:00
var files = make ( [ ] scanner . File , len ( fs ) )
2014-01-23 22:20:15 +01:00
for i := range fs {
2014-05-19 22:31:28 +02:00
f := fs [ i ]
lamport . Default . Tick ( f . Version )
if debug {
var flagComment string
2014-05-23 12:53:26 +02:00
if protocol . IsDeleted ( f . Flags ) {
2014-05-19 22:31:28 +02:00
flagComment = " (deleted)"
}
l . Debugf ( "IDX(in): %s %q/%q m=%d f=%o%s v=%d (%d blocks)" , nodeID , repo , f . Name , f . Modified , f . Flags , flagComment , f . Version , len ( f . Blocks ) )
}
files [ i ] = fileFromFileInfo ( f )
2014-01-23 22:20:15 +01:00
}
2014-03-29 18:53:48 +01:00
id := m . cm . Get ( nodeID )
m . rmut . RLock ( )
if r , ok := m . repoFiles [ repo ] ; ok {
r . Replace ( id , files )
} else {
2014-06-06 21:48:29 +02:00
l . Fatalf ( "Index for nonexistant repo %q" , repo )
2013-12-15 11:43:31 +01:00
}
2014-03-29 18:53:48 +01:00
m . rmut . RUnlock ( )
2013-12-28 14:10:36 +01:00
}
2013-12-30 15:30:29 +01:00
// IndexUpdate is called for incremental updates to connected nodes' indexes.
2014-01-06 11:11:18 +01:00
// Implements the protocol.Model interface.
2014-03-29 18:53:48 +01:00
func ( m * Model ) IndexUpdate ( nodeID string , repo string , fs [ ] protocol . FileInfo ) {
2014-05-15 05:26:55 +02:00
if debug {
2014-05-15 02:08:56 +02:00
l . Debugf ( "IDXUP(in): %s / %q: %d files" , nodeID , repo , len ( fs ) )
2014-03-29 18:53:48 +01:00
}
2014-06-06 21:48:29 +02: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-03-08 23:02:01 +01:00
var files = make ( [ ] scanner . File , len ( fs ) )
2014-01-23 22:20:15 +01:00
for i := range fs {
2014-05-19 22:31:28 +02:00
f := fs [ i ]
lamport . Default . Tick ( f . Version )
if debug {
var flagComment string
2014-05-23 12:53:26 +02:00
if protocol . IsDeleted ( f . Flags ) {
2014-05-19 22:31:28 +02:00
flagComment = " (deleted)"
}
l . Debugf ( "IDXUP(in): %s %q/%q m=%d f=%o%s v=%d (%d blocks)" , nodeID , repo , f . Name , f . Modified , f . Flags , flagComment , f . Version , len ( f . Blocks ) )
}
files [ i ] = fileFromFileInfo ( f )
2014-01-23 22:20:15 +01:00
}
2014-03-28 14:36:57 +01:00
id := m . cm . Get ( nodeID )
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
if r , ok := m . repoFiles [ repo ] ; ok {
r . Update ( id , files )
} else {
2014-06-06 21:48:29 +02:00
l . Fatalf ( "IndexUpdate for nonexistant repo %q" , repo )
2013-12-28 14:10:36 +01:00
}
2014-03-29 18:53:48 +01:00
m . rmut . RUnlock ( )
2014-01-09 10:59:09 +01:00
}
2014-06-06 21:48:29 +02:00
func ( m * Model ) repoSharedWith ( repo , nodeID string ) bool {
m . rmut . RLock ( )
defer m . rmut . RUnlock ( )
for _ , nrepo := range m . nodeRepos [ nodeID ] {
if nrepo == repo {
return true
}
}
return false
}
2014-04-13 15:28:26 +02:00
func ( m * Model ) ClusterConfig ( nodeID string , config protocol . ClusterConfigMessage ) {
compErr := compareClusterConfig ( m . clusterConfig ( nodeID ) , config )
2014-05-15 05:26:55 +02:00
if debug {
2014-05-15 02:08:56 +02:00
l . Debugf ( "ClusterConfig: %s: %#v" , nodeID , config )
l . Debugf ( " ... compare: %s: %v" , nodeID , compErr )
2014-04-13 15:28:26 +02:00
}
if compErr != nil {
2014-05-15 02:08:56 +02:00
l . Warnf ( "%s: %v" , nodeID , compErr )
2014-04-13 15:28:26 +02: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-01-20 22:22:27 +01:00
// Close removes the peer from the model and closes the underlying connection if possible.
2014-01-06 11:11:18 +01:00
// Implements the protocol.Model interface.
2013-12-31 03:21:57 +01:00
func ( m * Model ) Close ( node string , err error ) {
2014-05-15 05:26:55 +02:00
if debug {
2014-05-15 02:08:56 +02:00
l . Debugf ( "%s: %v" , node , err )
2014-02-09 23:13:06 +01:00
}
2014-04-13 15:28:26 +02:00
if err != io . EOF {
2014-05-15 02:08:56 +02:00
l . Warnf ( "Connection to %s closed: %v" , node , err )
2014-04-13 15:28:26 +02:00
} else if _ , ok := err . ( ClusterConfigMismatch ) ; ok {
2014-05-15 02:08:56 +02:00
l . Warnf ( "Connection to %s closed: %v" , node , err )
2014-02-09 23:13:06 +01:00
}
2014-03-28 14:36:57 +01:00
cid := m . cm . Get ( node )
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
for _ , repo := range m . nodeRepos [ node ] {
m . repoFiles [ repo ] . Replace ( cid , nil )
}
m . rmut . RUnlock ( )
2014-03-28 14:36:57 +01:00
m . cm . Clear ( node )
2014-01-20 22:22:27 +01:00
2014-01-18 04:06:44 +01:00
m . pmut . Lock ( )
2014-01-01 03:22:49 +01:00
conn , ok := m . rawConn [ node ]
2014-01-01 14:09:17 +01:00
if ok {
conn . Close ( )
2013-12-31 03:21:57 +01:00
}
2014-01-09 13:58:35 +01:00
delete ( m . protoConn , node )
2014-01-01 03:22:49 +01:00
delete ( m . rawConn , node )
2014-04-13 15:28:26 +02:00
delete ( m . nodeVer , node )
2014-01-18 04:06:44 +01:00
m . pmut . Unlock ( )
2013-12-15 11:43:31 +01:00
}
2014-01-06 11:11:18 +01:00
// Request returns the specified data segment by reading it from local disk.
// Implements the protocol.Model interface.
2014-02-20 17:40:15 +01:00
func ( m * Model ) Request ( nodeID , repo , name string , offset int64 , size int ) ( [ ] byte , error ) {
2014-03-28 14:36:57 +01:00
// Verify that the requested file exists in the local model.
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
r , ok := m . repoFiles [ repo ]
m . rmut . RUnlock ( )
if ! ok {
2014-05-15 02:08:56 +02:00
l . Warnf ( "Request from %s for file %s in nonexistent repo %q" , nodeID , name , repo )
2014-03-29 18:53:48 +01:00
return nil , ErrNoSuchFile
}
lf := r . Get ( cid . LocalID , name )
2014-05-23 12:53:26 +02:00
if lf . Suppressed || protocol . IsDeleted ( lf . Flags ) {
2014-05-20 20:26:44 +02: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 19:54:26 +02:00
return nil , ErrInvalid
2014-01-06 21:31:36 +01:00
}
2014-03-29 18:53:48 +01:00
2014-05-11 19:54:26 +02:00
if offset > lf . Size {
2014-05-15 05:26:55 +02:00
if debug {
2014-05-15 02:08:56 +02:00
l . Debugf ( "REQ(in; nonexistent): %s: %q o=%d s=%d" , nodeID , name , offset , size )
2014-05-11 19:54:26 +02:00
}
return nil , ErrNoSuchFile
2014-01-07 22:44:21 +01:00
}
2014-01-06 21:31:36 +01:00
2014-05-15 05:26:55 +02:00
if debug && nodeID != "<local>" {
2014-05-15 02:08:56 +02:00
l . Debugf ( "REQ(in): %s: %q / %q o=%d s=%d" , nodeID , repo , name , offset , size )
2013-12-15 11:43:31 +01:00
}
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
2014-05-23 14:31:16 +02:00
fn := filepath . Join ( m . repoCfgs [ repo ] . Directory , name )
2014-03-29 18:53:48 +01:00
m . rmut . RUnlock ( )
2013-12-15 11:43:31 +01:00
fd , err := os . Open ( fn ) // XXX: Inefficient, should cache fd?
if err != nil {
return nil , err
}
defer fd . Close ( )
2014-06-18 23:57:22 +02:00
buf := make ( [ ] byte , size )
2014-01-09 16:35:49 +01:00
_ , err = fd . ReadAt ( buf , offset )
2013-12-15 11:43:31 +01:00
if err != nil {
return nil , err
}
return buf , nil
}
2014-01-06 11:11:18 +01:00
// ReplaceLocal replaces the local repository index with the given list of files.
2014-03-29 18:53:48 +01:00
func ( m * Model ) ReplaceLocal ( repo string , fs [ ] scanner . File ) {
m . rmut . RLock ( )
m . repoFiles [ repo ] . ReplaceWithDelete ( cid . LocalID , fs )
m . rmut . RUnlock ( )
2013-12-15 11:43:31 +01:00
}
2014-03-29 18:53:48 +01:00
func ( m * Model ) SeedLocal ( repo string , fs [ ] protocol . FileInfo ) {
2014-03-28 14:36:57 +01:00
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 ] )
2014-05-20 20:26:44 +02:00
sfs [ i ] . Suppressed = false // we might have saved an index with files that were suppressed; the should not be on startup
2014-01-06 11:11:18 +01:00
}
2014-03-29 18:53:48 +01:00
m . rmut . RLock ( )
m . repoFiles [ repo ] . Replace ( cid . LocalID , sfs )
m . rmut . RUnlock ( )
}
2014-04-01 23:18:32 +02:00
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
}
2014-03-29 18:53:48 +01:00
type cFiler struct {
m * Model
r string
2014-01-06 11:11:18 +01:00
}
2014-03-16 08:14:55 +01:00
// Implements scanner.CurrentFiler
2014-03-29 18:53:48 +01:00
func ( cf cFiler ) CurrentFile ( file string ) scanner . File {
2014-04-01 23:18:32 +02:00
return cf . m . CurrentRepoFile ( cf . r , file )
2014-03-16 08:14:55 +01:00
}
2014-01-06 11:11:18 +01:00
// ConnectedTo returns true if we are connected to the named node.
func ( m * Model ) ConnectedTo ( nodeID string ) bool {
2014-01-18 04:06:44 +01:00
m . pmut . RLock ( )
2014-01-09 13:58:35 +01:00
_ , ok := m . protoConn [ nodeID ]
2014-01-18 04:06:44 +01:00
m . pmut . RUnlock ( )
2014-01-06 11:11:18 +01: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 14:36:57 +01:00
func ( m * Model ) AddConnection ( rawConn io . Closer , protoConn protocol . Connection ) {
2014-01-09 13:58:35 +01:00
nodeID := protoConn . ID ( )
2014-01-18 04:06:44 +01:00
m . pmut . Lock ( )
2014-03-23 08:45:05 +01:00
if _ , ok := m . protoConn [ nodeID ] ; ok {
panic ( "add existing node" )
}
2014-01-09 13:58:35 +01:00
m . protoConn [ nodeID ] = protoConn
2014-03-23 08:45:05 +01:00
if _ , ok := m . rawConn [ nodeID ] ; ok {
panic ( "add existing node" )
}
2014-01-09 13:58:35 +01:00
m . rawConn [ nodeID ] = rawConn
2014-01-18 04:06:44 +01:00
m . pmut . Unlock ( )
2014-01-06 11:11:18 +01:00
2014-04-13 15:28:26 +02:00
cm := m . clusterConfig ( nodeID )
protoConn . ClusterConfig ( cm )
2014-05-04 17:18:58 +02:00
var idxToSend = make ( map [ string ] [ ] protocol . FileInfo )
m . rmut . RLock ( )
for _ , repo := range m . nodeRepos [ nodeID ] {
idxToSend [ repo ] = m . protocolIndex ( repo )
}
m . rmut . RUnlock ( )
2014-01-06 11:11:18 +01:00
go func ( ) {
2014-05-04 17:18:58 +02:00
for repo , idx := range idxToSend {
2014-05-15 05:26:55 +02:00
if debug {
2014-05-15 02:08:56 +02:00
l . Debugf ( "IDX(out/initial): %s: %q: %d files" , nodeID , repo , len ( idx ) )
2014-03-29 18:53:48 +01:00
}
protoConn . Index ( repo , idx )
2014-03-09 09:15:36 +01:00
}
2014-01-06 11:11:18 +01:00
} ( )
}
2014-04-14 13:01:21 +02:00
// protocolIndex returns the current local index in protocol data types.
func ( m * Model ) protocolIndex ( repo string ) [ ] protocol . FileInfo {
2014-01-06 11:11:18 +01:00
var index [ ] protocol . FileInfo
2014-01-18 04:06:44 +01:00
2014-03-29 18:53:48 +01:00
fs := m . repoFiles [ repo ] . Have ( cid . LocalID )
2014-01-18 04:06:44 +01:00
2014-03-28 14:36:57 +01:00
for _ , f := range fs {
2014-01-06 11:11:18 +01:00
mf := fileInfoFromFile ( f )
2014-05-15 05:26:55 +02:00
if debug {
2014-01-06 11:11:18 +01:00
var flagComment string
2014-05-23 12:53:26 +02:00
if protocol . IsDeleted ( mf . Flags ) {
2014-01-06 11:11:18 +01:00
flagComment = " (deleted)"
}
2014-05-15 02:08:56 +02:00
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 ) )
2014-01-06 11:11:18 +01:00
}
index = append ( index , mf )
}
2014-01-18 04:06:44 +01:00
2014-01-06 11:11:18 +01:00
return index
}
2014-03-29 18:53:48 +01:00
func ( m * Model ) updateLocal ( repo string , f scanner . File ) {
m . rmut . RLock ( )
m . repoFiles [ repo ] . Update ( cid . LocalID , [ ] scanner . File { f } )
m . rmut . RUnlock ( )
2014-03-28 14:36:57 +01:00
}
2014-03-29 18:53:48 +01:00
func ( m * Model ) requestGlobal ( nodeID , repo , name string , offset int64 , size int , hash [ ] byte ) ( [ ] byte , error ) {
2014-01-18 04:06:44 +01:00
m . pmut . RLock ( )
2014-01-09 13:58:35 +01:00
nc , ok := m . protoConn [ nodeID ]
2014-01-18 04:06:44 +01:00
m . pmut . RUnlock ( )
2014-01-06 11:11:18 +01:00
if ! ok {
return nil , fmt . Errorf ( "requestGlobal: no such node: %s" , nodeID )
}
2014-05-15 05:26:55 +02:00
if debug {
2014-05-15 02:08:56 +02:00
l . Debugf ( "REQ(out): %s: %q / %q o=%d s=%d h=%x" , nodeID , repo , name , offset , size , hash )
2014-01-06 11:11:18 +01:00
}
2014-03-29 18:53:48 +01:00
return nc . Request ( repo , name , offset , size )
2014-01-06 11:11:18 +01:00
}
2013-12-24 21:21:03 +01:00
func ( m * Model ) broadcastIndexLoop ( ) {
2014-03-29 18:53:48 +01:00
var lastChange = map [ string ] uint64 { }
2013-12-24 21:21:03 +01:00
for {
2014-03-28 14:36:57 +01:00
time . Sleep ( 5 * time . Second )
2014-01-18 04:06:44 +01:00
2014-03-29 18:53:48 +01:00
m . pmut . RLock ( )
m . rmut . RLock ( )
2014-05-20 18:41:01 +02:00
var indexWg sync . WaitGroup
2014-03-29 18:53:48 +01:00
for repo , fs := range m . repoFiles {
2014-05-20 18:41:01 +02:00
repo := repo
2014-03-29 18:53:48 +01:00
c := fs . Changes ( cid . LocalID )
if c == lastChange [ repo ] {
continue
}
lastChange [ repo ] = c
2014-04-14 13:01:21 +02:00
idx := m . protocolIndex ( repo )
2014-05-20 18:41:01 +02:00
indexWg . Add ( 1 )
go func ( ) {
2014-05-31 23:32:47 +02:00
err := m . saveIndex ( repo , m . indexDir , idx )
if err != nil {
2014-06-04 10:53:27 +02:00
l . Infof ( "Saving index for %q: %v" , repo , err )
2014-05-31 23:32:47 +02:00
}
2014-05-20 18:41:01 +02:00
indexWg . Done ( )
} ( )
2014-03-29 18:53:48 +01:00
for _ , nodeID := range m . repoNodes [ repo ] {
2014-05-20 18:41:01 +02:00
nodeID := nodeID
2014-03-29 18:53:48 +01:00
if conn , ok := m . protoConn [ nodeID ] ; ok {
indexWg . Add ( 1 )
2014-05-15 05:26:55 +02:00
if debug {
2014-05-15 02:08:56 +02:00
l . Debugf ( "IDX(out/loop): %s: %d files" , nodeID , len ( idx ) )
2014-03-29 18:53:48 +01:00
}
go func ( ) {
conn . Index ( repo , idx )
indexWg . Done ( )
} ( )
}
}
2013-12-15 11:43:31 +01:00
}
2014-01-09 16:35:49 +01:00
2014-03-29 18:53:48 +01:00
m . rmut . RUnlock ( )
m . pmut . RUnlock ( )
2014-05-20 18:41:01 +02:00
indexWg . Wait ( )
2014-03-29 18:53:48 +01:00
}
}
2014-05-23 14:31:16 +02:00
func ( m * Model ) AddRepo ( cfg config . RepositoryConfiguration ) {
2014-03-29 18:53:48 +01:00
if m . started {
panic ( "cannot add repo to started model" )
}
2014-05-23 14:31:16 +02:00
if len ( cfg . ID ) == 0 {
2014-03-29 18:53:48 +01:00
panic ( "cannot add empty repo id" )
}
m . rmut . Lock ( )
2014-05-23 14:31:16 +02:00
m . repoCfgs [ cfg . ID ] = cfg
m . repoFiles [ cfg . ID ] = files . NewSet ( )
m . suppressor [ cfg . ID ] = & suppressor { threshold : int64 ( m . cfg . Options . MaxChangeKbps ) }
2013-12-15 11:43:31 +01:00
2014-05-23 14:31:16 +02:00
m . repoNodes [ cfg . ID ] = make ( [ ] string , len ( cfg . Nodes ) )
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 18:53:48 +01:00
}
2014-01-23 22:20:15 +01:00
2014-03-29 18:53:48 +01:00
m . addedRepo = true
m . rmut . Unlock ( )
}
2014-01-23 22:20:15 +01:00
2014-03-29 18:53:48 +01:00
func ( m * Model ) ScanRepos ( ) {
m . rmut . RLock ( )
2014-05-23 14:31:16 +02:00
var repos = make ( [ ] string , 0 , len ( m . repoCfgs ) )
for repo := range m . repoCfgs {
2014-04-14 09:58:17 +02:00
repos = append ( repos , repo )
2014-03-29 18:53:48 +01:00
}
m . rmut . RUnlock ( )
2014-04-14 09:58:17 +02:00
2014-05-14 01:42:12 +02:00
var wg sync . WaitGroup
wg . Add ( len ( repos ) )
2014-04-14 09:58:17 +02:00
for _ , repo := range repos {
2014-05-14 01:42:12 +02:00
repo := repo
go func ( ) {
2014-05-28 06:55:30 +02:00
err := m . ScanRepo ( repo )
if err != nil {
invalidateRepo ( m . cfg , repo , err )
}
2014-05-14 01:42:12 +02:00
wg . Done ( )
} ( )
2014-04-14 09:58:17 +02:00
}
2014-05-14 01:42:12 +02:00
wg . Wait ( )
2014-03-29 18:53:48 +01:00
}
2013-12-15 11:43:31 +01:00
2014-05-15 05:26:55 +02:00
func ( m * Model ) CleanRepos ( ) {
m . rmut . RLock ( )
2014-05-23 14:31:16 +02:00
var dirs = make ( [ ] string , 0 , len ( m . repoCfgs ) )
for _ , cfg := range m . repoCfgs {
dirs = append ( dirs , cfg . Directory )
2014-05-15 05:26:55 +02: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 18:20:25 +02:00
func ( m * Model ) ScanRepo ( repo string ) error {
2014-05-02 17:14:53 +02:00
m . rmut . RLock ( )
2014-03-29 18:53:48 +01:00
w := & scanner . Walker {
2014-05-23 14:31:16 +02:00
Dir : m . repoCfgs [ repo ] . Directory ,
2014-04-08 13:45:18 +02:00
IgnoreFile : ".stignore" ,
2014-05-15 05:26:55 +02:00
BlockSize : scanner . StandardBlockSize ,
2014-04-08 13:45:18 +02:00
TempNamer : defTempNamer ,
2014-05-15 05:26:55 +02:00
Suppressor : m . suppressor [ repo ] ,
2014-04-08 13:45:18 +02:00
CurrentFiler : cFiler { m , repo } ,
2014-05-23 14:31:16 +02:00
IgnorePerms : m . repoCfgs [ repo ] . IgnorePerms ,
2014-03-29 18:53:48 +01:00
}
2014-05-02 17:14:53 +02:00
m . rmut . RUnlock ( )
2014-04-14 09:58:17 +02:00
m . setState ( repo , RepoScanning )
2014-05-04 18:20:25 +02:00
fs , _ , err := w . Walk ( )
if err != nil {
return err
}
2014-03-29 18:53:48 +01:00
m . ReplaceLocal ( repo , fs )
2014-04-14 09:58:17 +02:00
m . setState ( repo , RepoIdle )
2014-05-04 18:20:25 +02:00
return nil
2014-03-29 18:53:48 +01:00
}
func ( m * Model ) SaveIndexes ( dir string ) {
m . rmut . RLock ( )
2014-05-23 14:31:16 +02:00
for repo := range m . repoCfgs {
2014-04-14 13:01:21 +02:00
fs := m . protocolIndex ( repo )
2014-05-31 23:32:47 +02:00
err := m . saveIndex ( repo , dir , fs )
if err != nil {
2014-06-04 10:53:27 +02:00
l . Infof ( "Saving index for %q: %v" , repo , err )
2014-05-31 23:32:47 +02:00
}
2014-03-29 18:53:48 +01:00
}
m . rmut . RUnlock ( )
}
func ( m * Model ) LoadIndexes ( dir string ) {
m . rmut . RLock ( )
2014-05-23 14:31:16 +02:00
for repo := range m . repoCfgs {
2014-03-29 18:53:48 +01:00
fs := m . loadIndex ( repo , dir )
m . SeedLocal ( repo , fs )
}
m . rmut . RUnlock ( )
}
2014-05-31 23:32:47 +02:00
func ( m * Model ) saveIndex ( repo string , dir string , fs [ ] protocol . FileInfo ) error {
2014-05-23 14:31:16 +02:00
id := fmt . Sprintf ( "%x" , sha1 . Sum ( [ ] byte ( m . repoCfgs [ repo ] . Directory ) ) )
2014-03-29 18:53:48 +01:00
name := id + ".idx.gz"
name = filepath . Join ( dir , name )
2014-06-04 13:43:59 +02:00
tmp := fmt . Sprintf ( "%s.tmp.%d" , name , time . Now ( ) . UnixNano ( ) )
idxf , err := os . OpenFile ( tmp , os . O_CREATE | os . O_EXCL | os . O_WRONLY , 0644 )
2014-03-29 18:53:48 +01:00
if err != nil {
2014-05-31 23:32:47 +02:00
return err
2014-03-29 18:53:48 +01:00
}
2014-06-04 13:43:59 +02:00
defer os . Remove ( tmp )
2014-01-18 04:06:44 +01:00
2014-03-29 18:53:48 +01:00
gzw := gzip . NewWriter ( idxf )
2014-05-31 23:32:47 +02:00
n , err := protocol . IndexMessage {
2014-03-29 18:53:48 +01:00
Repository : repo ,
Files : fs ,
} . EncodeXDR ( gzw )
2014-05-31 23:32:47 +02:00
if err != nil {
gzw . Close ( )
idxf . Close ( )
return err
}
err = gzw . Close ( )
if err != nil {
return err
}
err = idxf . Close ( )
if err != nil {
return err
}
if debug {
l . Debugln ( "wrote index," , n , "bytes uncompressed" )
}
2014-03-29 18:53:48 +01:00
2014-06-04 13:43:59 +02:00
return osutil . Rename ( tmp , name )
2014-03-29 18:53:48 +01:00
}
func ( m * Model ) loadIndex ( repo string , dir string ) [ ] protocol . FileInfo {
2014-05-23 14:31:16 +02:00
id := fmt . Sprintf ( "%x" , sha1 . Sum ( [ ] byte ( m . repoCfgs [ repo ] . Directory ) ) )
2014-03-29 18:53:48 +01:00
name := id + ".idx.gz"
name = filepath . Join ( dir , name )
idxf , err := os . Open ( name )
if err != nil {
return nil
2014-01-09 16:35:49 +01:00
}
2014-03-29 18:53:48 +01:00
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
2014-01-09 16:35:49 +01:00
}
2014-04-13 15:28:26 +02:00
// clusterConfig returns a ClusterConfigMessage that is correct for the given peer node
func ( m * Model ) clusterConfig ( node string ) protocol . ClusterConfigMessage {
cm := protocol . ClusterConfigMessage {
2014-05-15 05:26:55 +02:00
ClientName : m . clientName ,
ClientVersion : m . clientVersion ,
2014-04-13 15:28:26 +02:00
}
2014-05-02 17:14:53 +02:00
m . rmut . RLock ( )
2014-04-13 15:28:26 +02:00
for _ , repo := range m . nodeRepos [ node ] {
cr := protocol . Repository {
ID : repo ,
2014-01-09 13:58:35 +01:00
}
2014-04-13 15:28:26 +02:00
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 ,
} )
2014-01-09 13:58:35 +01:00
}
2014-04-13 15:28:26 +02:00
cm . Repositories = append ( cm . Repositories , cr )
2013-12-30 02:33:57 +01:00
}
2014-05-02 17:14:53 +02:00
m . rmut . RUnlock ( )
2014-04-13 15:28:26 +02:00
return cm
2013-12-30 02:33:57 +01:00
}
2014-04-14 09:58:17 +02:00
func ( m * Model ) setState ( repo string , state repoState ) {
2014-05-20 18:41:01 +02:00
m . smut . Lock ( )
2014-04-14 09:58:17 +02:00
m . repoState [ repo ] = state
2014-05-20 18:41:01 +02:00
m . smut . Unlock ( )
2014-04-14 09:58:17 +02:00
}
func ( m * Model ) State ( repo string ) string {
2014-05-20 18:41:01 +02:00
m . smut . RLock ( )
2014-04-14 09:58:17 +02:00
state := m . repoState [ repo ]
2014-05-20 18:41:01 +02:00
m . smut . RUnlock ( )
2014-04-14 09:58:17 +02:00
switch state {
case RepoIdle :
return "idle"
case RepoScanning :
return "scanning"
case RepoCleaning :
return "cleaning"
case RepoSyncing :
return "syncing"
default :
return "unknown"
}
}
2014-06-16 10:47:02 +02:00
func ( m * Model ) Override ( repo string ) {
fs := m . NeedFilesRepo ( repo )
m . rmut . Lock ( )
r := m . repoFiles [ repo ]
for i := range fs {
f := & fs [ i ]
h := r . Get ( cid . LocalID , f . Name )
if h . Name != f . Name {
// We are missing the file
f . Flags |= protocol . FlagDeleted
f . Blocks = nil
} else {
// We have the file, replace with our version
* f = h
}
f . Version = lamport . Default . Tick ( f . Version )
}
m . rmut . Unlock ( )
r . Update ( cid . LocalID , fs )
}
2014-06-20 00:27:54 +02: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.
func ( m * Model ) Version ( repo string ) uint64 {
var ver uint64
m . rmut . Lock ( )
for _ , n := range m . repoNodes [ repo ] {
ver += m . repoFiles [ repo ] . Changes ( m . cm . Get ( n ) )
}
m . rmut . Unlock ( )
return ver
}