Compare commits

...

42 Commits

Author SHA1 Message Date
Jakob Borg
6807d9bd4c Fix upgrade non-support on Windows 2014-05-02 20:19:21 +02:00
Jakob Borg
699ecc7140 Some places should use RLock instead of Lock (ref #169) 2014-05-02 17:15:04 +02:00
Jakob Borg
b374ec9355 Save temporary in correct dir during upgrade 2014-05-02 17:04:45 +02:00
Jakob Borg
9659d021cb Merge pull request #170 from andrew-d/patch-1
Fix typo in header name
2014-05-02 16:59:08 +02:00
Jakob Borg
a4ad9eb134 Add andrew-d 2014-05-02 16:58:55 +02:00
Andrew
a455258a62 Fix typo in header name 2014-05-02 01:26:12 -07:00
Jakob Borg
0ae342673a Update saved dependencies 2014-05-02 10:05:48 +02:00
Jakob Borg
33d75a264d Built in upgrade functionality 2014-05-02 10:01:09 +02:00
Jakob Borg
89dc5bb951 Windows doesn't have SysProcAttr 2014-05-02 08:57:34 +02:00
Jakob Borg
45403917de Minor cleanup in discovery 2014-05-02 08:53:19 +02:00
Jakob Borg
ed476271a6 Start xdg-open in new process group (fixes #164) 2014-05-02 08:53:05 +02:00
Jakob Borg
1e92c47960 Don't bother starting without GUI (fixes #156) 2014-04-30 22:52:38 +02:00
Jakob Borg
4f2fe07ae4 Show node ID in regular text not disabled control (fixes #162) 2014-04-30 22:42:39 +02:00
Jakob Borg
aff3cd01c5 Don't show Offline badge when global disco is disabled (fixes #167) 2014-04-30 22:17:43 +02:00
Jakob Borg
ac74ee1468 Don't redirect to absolute URL (fixes #166) 2014-04-30 22:10:13 +02:00
Jakob Borg
0d55cf4be5 Don't use absolute URL for rest calls (fixes #166) 2014-04-30 22:02:34 +02:00
Jakob Borg
5399a25532 Getting started 2014-04-30 16:13:29 +02:00
Jakob Borg
ae882c93c9 Links to discourse 2014-04-30 15:14:42 +02:00
Jakob Borg
f398ca77c1 Better trace output from mc 2014-04-30 15:13:54 +02:00
Jakob Borg
dcd7d278aa Handle and indicate duplicate repo ID:s (fixes #153) 2014-04-27 21:53:27 +02:00
Jakob Borg
89f5f3bf9a Fix small data races 2014-04-27 21:33:57 +02:00
Jakob Borg
76ef42ee07 No drone.io badge 2014-04-27 13:37:53 +02:00
Jakob Borg
92c1ce57a6 Fix protocol close test 2014-04-27 13:25:35 +02:00
Jakob Borg
116f232f5a Streamline error handling and locking 2014-04-27 13:10:50 +02:00
Jakob Borg
ef81a36654 Extract method closeFile 2014-04-27 12:14:53 +02:00
Jakob Borg
9fd2724d73 Simplify requestSlots filling 2014-04-27 12:06:11 +02:00
Jakob Borg
07d49b61d0 Debug utility to print index file 2014-04-25 08:28:56 +02:00
Jakob Borg
0c4e6ae7de Safety: don't start if repo dir is missing (ref #154) 2014-04-24 10:27:43 +02:00
Jakob Borg
65ec129dfb Only create default config if it is actually missing (fixes #139) 2014-04-23 10:28:36 +02:00
Jakob Borg
3e4d628f54 Handle non-word characters in repo name (fixes #152) 2014-04-23 10:04:25 +02:00
Jakob Borg
71684bfa45 Use a more lenient cluster config check (fixes #148) 2014-04-22 16:42:25 +02:00
Jakob Borg
e73b7e0398 Show properly formatted time (fixes #149) 2014-04-22 15:59:16 +02:00
Jakob Borg
35ebdc76ff Hide temporary files on Windows (fixes #146) 2014-04-22 14:27:31 +02:00
Jakob Borg
90d0896848 Change default config directory (fixes #145) 2014-04-22 14:27:09 +02:00
Jakob Borg
5528db9693 Fix config test (hostname check) 2014-04-22 12:06:32 +02:00
Jakob Borg
aa78fbb09d Don't offer to delete this node (fixes #144) 2014-04-22 12:01:09 +02:00
Jakob Borg
d53b193e09 Ensure sensible node config on load (fixes #143) 2014-04-22 11:46:08 +02:00
Jakob Borg
e0e16c371f Don't include test utils in testing 2014-04-22 08:27:00 +02:00
Jakob Borg
53cd877899 More portable hostname 2014-04-22 08:25:40 +02:00
Jakob Borg
1207223f3d Report rates over the wire, not uncompressed 2014-04-21 12:49:47 +02:00
Jakob Borg
39be6932b5 discosrv: Better statistics 2014-04-19 23:14:56 +02:00
Jakob Borg
44a194d226 discosrv: Remove deprecated v1 support 2014-04-19 23:02:14 +02:00
45 changed files with 1243 additions and 597 deletions

View File

@@ -1,6 +1,6 @@
Please do contribute! If you want to contribute but are unsure where to
start, the [Contributions Needed
page](https://github.com/calmh/syncthing/wiki/Contributions-Needed)
topic](http://discourse.syncthing.net/t/contributions-needed/49)
lists areas in need of attention.
## Licensing
@@ -15,7 +15,8 @@ will ensure that you are added to the CONTRIBUTORS file.
## Building
[See the wiki](https://github.com/calmh/syncthing/wiki/Building)
[See the
documentation](http://discourse.syncthing.net/t/building-syncthing/44)
## Branches
@@ -46,7 +47,7 @@ Yes please!
## Documentation
[Hack it here](https://github.com/calmh/syncthing/wiki)
[Over here!](http://discourse.syncthing.net/category/documentation)
## License

View File

@@ -1,4 +1,5 @@
Aaron Bieber <qbit@deftly.net>
Andrew Dunham <andrew@du.nham.ca>
Brandon Philips <brandon@ifup.org>
James Patterson <jamespatterson@operamail.com>
Philippe Schommers <philippe@schommers.be>

5
Godeps/Godeps.json generated
View File

@@ -8,6 +8,11 @@
"./discover/cmd/discosrv"
],
"Deps": [
{
"ImportPath": "bitbucket.org/kardianos/osext",
"Comment": "null-9",
"Rev": "364fb577de68fb646c4cb39cc0e09c887ee16376"
},
{
"ImportPath": "code.google.com/p/go.crypto/bcrypt",
"Comment": "null-185",

View File

@@ -0,0 +1,20 @@
Copyright (c) 2012 Daniel Theophanes
This software is provided 'as-is', without any express or implied
warranty. In no event will the authors be held liable for any damages
arising from the use of this software.
Permission is granted to anyone to use this software for any purpose,
including commercial applications, and to alter it and redistribute it
freely, subject to the following restrictions:
1. The origin of this software must not be misrepresented; you must not
claim that you wrote the original software. If you use this software
in a product, an acknowledgment in the product documentation would be
appreciated but is not required.
2. Altered source versions must be plainly marked as such, and must not be
misrepresented as being the original software.
3. This notice may not be removed or altered from any source
distribution.

View File

@@ -0,0 +1,32 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Extensions to the standard "os" package.
package osext
import "path/filepath"
// Executable returns an absolute path that can be used to
// re-invoke the current program.
// It may not be valid after the current program exits.
func Executable() (string, error) {
p, err := executable()
return filepath.Clean(p), err
}
// Returns same path as Executable, returns just the folder
// path. Excludes the executable name.
func ExecutableFolder() (string, error) {
p, err := Executable()
if err != nil {
return "", err
}
folder, _ := filepath.Split(p)
return folder, nil
}
// Depricated. Same as Executable().
func GetExePath() (exePath string, err error) {
return Executable()
}

View File

@@ -0,0 +1,16 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package osext
import "syscall"
func executable() (string, error) {
f, err := Open("/proc/" + itoa(Getpid()) + "/text")
if err != nil {
return "", err
}
defer f.Close()
return syscall.Fd2path(int(f.Fd()))
}

View File

@@ -0,0 +1,25 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// +build linux netbsd openbsd
package osext
import (
"errors"
"os"
"runtime"
)
func executable() (string, error) {
switch runtime.GOOS {
case "linux":
return os.Readlink("/proc/self/exe")
case "netbsd":
return os.Readlink("/proc/curproc/exe")
case "openbsd":
return os.Readlink("/proc/curproc/file")
}
return "", errors.New("ExecPath not implemented for " + runtime.GOOS)
}

View File

@@ -0,0 +1,82 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// +build darwin freebsd
package osext
import (
"os"
"path/filepath"
"runtime"
"syscall"
"unsafe"
)
var startUpcwd, getwdError = os.Getwd()
func executable() (string, error) {
var mib [4]int32
switch runtime.GOOS {
case "freebsd":
mib = [4]int32{1 /* CTL_KERN */, 14 /* KERN_PROC */, 12 /* KERN_PROC_PATHNAME */, -1}
case "darwin":
mib = [4]int32{1 /* CTL_KERN */, 38 /* KERN_PROCARGS */, int32(os.Getpid()), -1}
}
n := uintptr(0)
// get length
_, _, err := syscall.Syscall6(syscall.SYS___SYSCTL, uintptr(unsafe.Pointer(&mib[0])), 4, 0, uintptr(unsafe.Pointer(&n)), 0, 0)
if err != 0 {
return "", err
}
if n == 0 { // shouldn't happen
return "", nil
}
buf := make([]byte, n)
_, _, err = syscall.Syscall6(syscall.SYS___SYSCTL, uintptr(unsafe.Pointer(&mib[0])), 4, uintptr(unsafe.Pointer(&buf[0])), uintptr(unsafe.Pointer(&n)), 0, 0)
if err != 0 {
return "", err
}
if n == 0 { // shouldn't happen
return "", nil
}
for i, v := range buf {
if v == 0 {
buf = buf[:i]
break
}
}
var strpath string
if buf[0] != '/' {
var e error
if strpath, e = getAbs(buf); e != nil {
return strpath, e
}
} else {
strpath = string(buf)
}
// darwin KERN_PROCARGS may return the path to a symlink rather than the
// actual executable
if runtime.GOOS == "darwin" {
if strpath, err := filepath.EvalSymlinks(strpath); err != nil {
return strpath, err
}
}
return strpath, nil
}
func getAbs(buf []byte) (string, error) {
if getwdError != nil {
return string(buf), getwdError
} else {
if buf[0] == '.' {
buf = buf[1:]
}
if startUpcwd[len(startUpcwd)-1] != '/' && buf[0] != '/' {
return startUpcwd + "/" + string(buf), nil
}
return startUpcwd + string(buf), nil
}
}

View File

@@ -0,0 +1,79 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// +build darwin linux freebsd netbsd windows
package osext
import (
"fmt"
"os"
oexec "os/exec"
"path/filepath"
"runtime"
"testing"
)
const execPath_EnvVar = "OSTEST_OUTPUT_EXECPATH"
func TestExecPath(t *testing.T) {
ep, err := Executable()
if err != nil {
t.Fatalf("ExecPath failed: %v", err)
}
// we want fn to be of the form "dir/prog"
dir := filepath.Dir(filepath.Dir(ep))
fn, err := filepath.Rel(dir, ep)
if err != nil {
t.Fatalf("filepath.Rel: %v", err)
}
cmd := &oexec.Cmd{}
// make child start with a relative program path
cmd.Dir = dir
cmd.Path = fn
// forge argv[0] for child, so that we can verify we could correctly
// get real path of the executable without influenced by argv[0].
cmd.Args = []string{"-", "-test.run=XXXX"}
cmd.Env = []string{fmt.Sprintf("%s=1", execPath_EnvVar)}
out, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("exec(self) failed: %v", err)
}
outs := string(out)
if !filepath.IsAbs(outs) {
t.Fatalf("Child returned %q, want an absolute path", out)
}
if !sameFile(outs, ep) {
t.Fatalf("Child returned %q, not the same file as %q", out, ep)
}
}
func sameFile(fn1, fn2 string) bool {
fi1, err := os.Stat(fn1)
if err != nil {
return false
}
fi2, err := os.Stat(fn2)
if err != nil {
return false
}
return os.SameFile(fi1, fi2)
}
func init() {
if e := os.Getenv(execPath_EnvVar); e != "" {
// first chdir to another path
dir := "/"
if runtime.GOOS == "windows" {
dir = filepath.VolumeName(".")
}
os.Chdir(dir)
if ep, err := Executable(); err != nil {
fmt.Fprint(os.Stderr, "ERROR: ", err)
} else {
fmt.Fprint(os.Stderr, ep)
}
os.Exit(0)
}
}

View File

@@ -0,0 +1,34 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package osext
import (
"syscall"
"unicode/utf16"
"unsafe"
)
var (
kernel = syscall.MustLoadDLL("kernel32.dll")
getModuleFileNameProc = kernel.MustFindProc("GetModuleFileNameW")
)
// GetModuleFileName() with hModule = NULL
func executable() (exePath string, err error) {
return getModuleFileName()
}
func getModuleFileName() (string, error) {
var n uint32
b := make([]uint16, syscall.MAX_PATH)
size := uint32(len(b))
r0, _, e1 := getModuleFileNameProc.Call(0, uintptr(unsafe.Pointer(&b[0])), uintptr(size))
n = uint32(r0)
if n == 0 {
return "", e1
}
return string(utf16.Decode(b[0:n])), nil
}

View File

@@ -1,4 +1,4 @@
syncthing [![Build Status](https://drone.io/github.com/calmh/syncthing/status.png)](https://drone.io/github.com/calmh/syncthing/latest)
syncthing
=========
This is the `syncthing` project. The following are the project goals:
@@ -25,6 +25,11 @@ making sure large swarms of selfish agents behave and somehow work
towards a common goal. Here we have a much smaller swarm of cooperative
agents and a simpler approach will suffice.
Getting Started
---------------
Take a look at the [getting started guide](http://discourse.syncthing.net/t/getting-started/46).
Signed Releases
---------------
@@ -35,8 +40,9 @@ normal release bundle as `syncthing.asc` or `syncthing.exe.asc`.
Documentation
=============
The syncthing documentation is kept on the
[GitHub Wiki](https://github.com/calmh/syncthing/wiki).
The [syncthing
documentation](http://discourse.syncthing.net/category/documentation) is
on the discourse site.
License
=======

View File

File diff suppressed because one or more lines are too long

View File

@@ -6,7 +6,8 @@ distFiles=(README.md LICENSE) # apart from the binary itself
version=$(git describe --always --dirty)
date=$(date +%s)
user=$(whoami)
host=$(hostname -s)
host=$(hostname)
host=${host%%.*}
ldflags="-w -X main.Version $version -X main.BuildStamp $date -X main.BuildUser $user -X main.BuildHost $host"
build() {

52
cmd/stpidx/main.go Normal file
View File

@@ -0,0 +1,52 @@
package main
import (
"compress/gzip"
"flag"
"log"
"os"
"github.com/calmh/syncthing/protocol"
)
func main() {
log.SetFlags(0)
log.SetOutput(os.Stdout)
showBlocks := flag.Bool("b", false, "Show blocks")
flag.Parse()
name := flag.Arg(0)
idxf, err := os.Open(name)
if err != nil {
log.Fatal(err)
}
defer idxf.Close()
gzr, err := gzip.NewReader(idxf)
if err != nil {
log.Fatal(err)
}
defer gzr.Close()
var im protocol.IndexMessage
err = im.DecodeXDR(gzr)
if err != nil {
log.Fatal(err)
}
log.Printf("Repo: %q, Files: %d", im.Repository, len(im.Files))
for _, file := range im.Files {
del := file.Flags&protocol.FlagDeleted != 0
inv := file.Flags&protocol.FlagInvalid != 0
dir := file.Flags&protocol.FlagDirectory != 0
prm := file.Flags & 0777
log.Printf("File: %q, Del: %v, Inv: %v, Dir: %v, Perm: 0%03o, Modified: %d, Blocks: %d",
file.Name, del, inv, dir, prm, file.Modified, len(file.Blocks))
if *showBlocks {
for _, block := range file.Blocks {
log.Printf(" Size: %6d, Hash: %x", block.Size, block.Hash)
}
}
}
}

View File

@@ -1,6 +1,10 @@
package main
import "github.com/calmh/syncthing/scanner"
import (
"sync/atomic"
"github.com/calmh/syncthing/scanner"
)
type bqAdd struct {
file scanner.File
@@ -20,6 +24,7 @@ type blockQueue struct {
outbox chan bqBlock
queued []bqBlock
qlen uint32
}
func newBlockQueue() *blockQueue {
@@ -77,6 +82,7 @@ func (q *blockQueue) run() {
q.queued = q.queued[1:]
}
}
atomic.StoreUint32(&q.qlen, uint32(len(q.queued)))
}
}
@@ -89,6 +95,7 @@ func (q *blockQueue) get() bqBlock {
}
func (q *blockQueue) empty() bool {
// There is a race condition here. We're only mostly sure the queue is empty if the expression below is true.
return len(q.queued) == 0 && len(q.inbox) == 0 && len(q.outbox) == 0
var l uint32
atomic.LoadUint32(&l)
return l == 0
}

View File

@@ -3,6 +3,7 @@ package main
import (
"encoding/xml"
"io"
"os"
"reflect"
"sort"
"strconv"
@@ -24,6 +25,7 @@ type RepositoryConfiguration struct {
Directory string `xml:"directory,attr"`
Nodes []NodeConfiguration `xml:"node"`
ReadOnly bool `xml:"ro,attr"`
Invalid string `xml:"-"` // Set at runtime when there is an error, not saved
nodeIDs []string
}
@@ -153,7 +155,7 @@ func uniqueStrings(ss []string) []string {
return us
}
func readConfigXML(rd io.Reader) (Configuration, error) {
func readConfigXML(rd io.Reader, myID string) (Configuration, error) {
var cfg Configuration
setDefaults(&cfg)
@@ -169,23 +171,30 @@ func readConfigXML(rd io.Reader) (Configuration, error) {
cfg.Options.ListenAddress = uniqueStrings(cfg.Options.ListenAddress)
var seenRepos = map[string]bool{}
// Check for missing or duplicate repository ID:s
var seenRepos = map[string]*RepositoryConfiguration{}
for i := range cfg.Repositories {
if cfg.Repositories[i].ID == "" {
cfg.Repositories[i].ID = "default"
repo := &cfg.Repositories[i]
if repo.ID == "" {
repo.ID = "default"
}
id := cfg.Repositories[i].ID
if seenRepos[id] {
panic("duplicate repository ID " + id)
if seen, ok := seenRepos[repo.ID]; ok {
seen.Invalid = "duplicate repository ID"
repo.Invalid = "duplicate repository ID"
warnf("Multiple repositories with ID %q; disabling", repo.ID)
} else {
seenRepos[repo.ID] = repo
}
seenRepos[id] = true
}
// Upgrade to v2 configuration if appropriate
if cfg.Version == 1 {
convertV1V2(&cfg)
}
// Hash old cleartext passwords
if len(cfg.GUI.Password) > 0 && cfg.GUI.Password[0] != '$' {
hash, err := bcrypt.GenerateFromPassword([]byte(cfg.GUI.Password), 0)
if err != nil {
@@ -195,6 +204,20 @@ func readConfigXML(rd io.Reader) (Configuration, error) {
}
}
// Ensure this node is present in all relevant places
cfg.Nodes = ensureNodePresent(cfg.Nodes, myID)
for i := range cfg.Repositories {
cfg.Repositories[i].Nodes = ensureNodePresent(cfg.Repositories[i].Nodes, myID)
}
// An empty address list is equivalent to a single "dynamic" entry
for i := range cfg.Nodes {
n := &cfg.Nodes[i]
if len(n.Addresses) == 0 || len(n.Addresses) == 1 && n.Addresses[0] == "" {
n.Addresses = []string{"dynamic"}
}
}
return cfg, err
}
@@ -241,7 +264,7 @@ func (l NodeConfigurationList) Len() int {
return len(l)
}
func cleanNodeList(nodes []NodeConfiguration, myID string) []NodeConfiguration {
func ensureNodePresent(nodes []NodeConfiguration, myID string) []NodeConfiguration {
var myIDExists bool
for _, node := range nodes {
if node.NodeID == myID {
@@ -251,10 +274,10 @@ func cleanNodeList(nodes []NodeConfiguration, myID string) []NodeConfiguration {
}
if !myIDExists {
name, _ := os.Hostname()
nodes = append(nodes, NodeConfiguration{
NodeID: myID,
Addresses: []string{"dynamic"},
Name: "",
NodeID: myID,
Name: name,
})
}

View File

@@ -3,6 +3,7 @@ package main
import (
"bytes"
"io"
"os"
"reflect"
"testing"
)
@@ -22,7 +23,7 @@ func TestDefaultValues(t *testing.T) {
UPnPEnabled: true,
}
cfg, err := readConfigXML(bytes.NewReader(nil))
cfg, err := readConfigXML(bytes.NewReader(nil), "nodeID")
if err != io.EOF {
t.Error(err)
}
@@ -65,7 +66,7 @@ func TestNodeConfig(t *testing.T) {
`)
for i, data := range [][]byte{v1data, v2data} {
cfg, err := readConfigXML(bytes.NewReader(data))
cfg, err := readConfigXML(bytes.NewReader(data), "node1")
if err != nil {
t.Error(err)
}
@@ -120,7 +121,7 @@ func TestNoListenAddress(t *testing.T) {
</configuration>
`)
cfg, err := readConfigXML(bytes.NewReader(data))
cfg, err := readConfigXML(bytes.NewReader(data), "nodeID")
if err != nil {
t.Error(err)
}
@@ -169,7 +170,7 @@ func TestOverriddenValues(t *testing.T) {
UPnPEnabled: false,
}
cfg, err := readConfigXML(bytes.NewReader(data))
cfg, err := readConfigXML(bytes.NewReader(data), "nodeID")
if err != nil {
t.Error(err)
}
@@ -178,3 +179,48 @@ func TestOverriddenValues(t *testing.T) {
t.Errorf("Overridden config differs;\n E: %#v\n A: %#v", expected, cfg.Options)
}
}
func TestNodeAddresses(t *testing.T) {
data := []byte(`
<configuration version="2">
<node id="n1">
<address>dynamic</address>
</node>
<node id="n2">
<address></address>
</node>
<node id="n3">
</node>
</configuration>
`)
name, _ := os.Hostname()
expected := []NodeConfiguration{
{
NodeID: "n1",
Addresses: []string{"dynamic"},
},
{
NodeID: "n2",
Addresses: []string{"dynamic"},
},
{
NodeID: "n3",
Addresses: []string{"dynamic"},
},
{
NodeID: "n4",
Name: name, // Set when auto created
Addresses: []string{"dynamic"},
},
}
cfg, err := readConfigXML(bytes.NewReader(data), "n4")
if err != nil {
t.Error(err)
}
if !reflect.DeepEqual(cfg.Nodes, expected) {
t.Errorf("Nodes differ;\n E: %#v\n A: %#v", expected, cfg.Nodes)
}
}

View File

@@ -5,7 +5,9 @@ import (
"encoding/base64"
"encoding/json"
"io/ioutil"
"log"
"math/rand"
"net"
"net/http"
"runtime"
"sync"
@@ -25,21 +27,27 @@ var (
configInSync = true
guiErrors = []guiError{}
guiErrorsMut sync.Mutex
static = embeddedStatic()
staticFunc = static.(func(http.ResponseWriter, *http.Request, *log.Logger))
)
const (
unchangedPassword = "--password-unchanged--"
)
func startGUI(cfg GUIConfiguration, m *Model) {
func startGUI(cfg GUIConfiguration, m *Model) error {
l, err := net.Listen("tcp", cfg.Address)
if err != nil {
return err
}
router := martini.NewRouter()
router.Get("/", getRoot)
router.Get("/rest/version", restGetVersion)
router.Get("/rest/model/:repo", restGetModel)
router.Get("/rest/model", restGetModel)
router.Get("/rest/connections", restGetConnections)
router.Get("/rest/config", restGetConfig)
router.Get("/rest/config/sync", restGetConfigInSync)
router.Get("/rest/need/:repo", restGetNeed)
router.Get("/rest/system", restGetSystem)
router.Get("/rest/errors", restGetErrors)
@@ -49,25 +57,24 @@ func startGUI(cfg GUIConfiguration, m *Model) {
router.Post("/rest/error", restPostError)
router.Post("/rest/error/clear", restClearErrors)
go func() {
mr := martini.New()
if len(cfg.User) > 0 && len(cfg.Password) > 0 {
mr.Use(basic(cfg.User, cfg.Password))
}
mr.Use(embeddedStatic())
mr.Use(martini.Recovery())
mr.Use(restMiddleware)
mr.Action(router.Handle)
mr.Map(m)
err := http.ListenAndServe(cfg.Address, mr)
if err != nil {
warnln("GUI not possible:", err)
}
}()
mr := martini.New()
if len(cfg.User) > 0 && len(cfg.Password) > 0 {
mr.Use(basic(cfg.User, cfg.Password))
}
mr.Use(static)
mr.Use(martini.Recovery())
mr.Use(restMiddleware)
mr.Action(router.Handle)
mr.Map(m)
go http.Serve(l, mr)
return nil
}
func getRoot(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, "/index.html", 302)
r.URL.Path = "/index.html"
staticFunc(w, r, nil)
}
func restMiddleware(w http.ResponseWriter, r *http.Request) {
@@ -80,10 +87,18 @@ func restGetVersion() string {
return Version
}
func restGetModel(m *Model, w http.ResponseWriter, params martini.Params) {
var repo = params["repo"]
func restGetModel(m *Model, w http.ResponseWriter, r *http.Request) {
var qs = r.URL.Query()
var repo = qs.Get("repo")
var res = make(map[string]interface{})
for _, cr := range cfg.Repositories {
if cr.ID == repo {
res["invalid"] = cr.Invalid
break
}
}
globalFiles, globalDeleted, globalBytes := m.GlobalSize(repo)
res["globalFiles"], res["globalDeleted"], res["globalBytes"] = globalFiles, globalDeleted, globalBytes
@@ -168,17 +183,6 @@ func (f guiFile) MarshalJSON() ([]byte, error) {
})
}
func restGetNeed(m *Model, w http.ResponseWriter, params martini.Params) {
repo := params["repo"]
files := m.NeedFilesRepo(repo)
gfs := make([]guiFile, len(files))
for i, f := range files {
gfs[i] = guiFile(f)
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(gfs)
}
var cpuUsagePercent [10]float64 // The last ten seconds
var cpuUsageLock sync.RWMutex
@@ -191,7 +195,7 @@ func restGetSystem(w http.ResponseWriter) {
res["goroutines"] = runtime.NumGoroutine()
res["alloc"] = m.Alloc
res["sys"] = m.Sys
if discoverer != nil {
if cfg.Options.GlobalAnnEnabled && discoverer != nil {
res["extAnnounceOK"] = discoverer.ExtAnnounceOK()
}
cpuUsageLock.RLock()

View File

@@ -32,7 +32,7 @@ func embeddedStatic() interface{} {
if len(mtype) != 0 {
res.Header().Set("Content-Type", mtype)
}
res.Header().Set("Content-Size", fmt.Sprintf("%d", len(bs)))
res.Header().Set("Content-Length", fmt.Sprintf("%d", len(bs)))
res.Header().Set("Last-Modified", modt)
res.Write(bs)

View File

@@ -83,9 +83,11 @@ const (
func main() {
var reset bool
var showVersion bool
var doUpgrade bool
flag.StringVar(&confDir, "home", getDefaultConfDir(), "Set configuration directory")
flag.BoolVar(&reset, "reset", false, "Prepare to resync from cluster")
flag.BoolVar(&showVersion, "version", false, "Show version")
flag.BoolVar(&doUpgrade, "upgrade", false, "Perform upgrade")
flag.Usage = usageFor(flag.CommandLine, usage, extraUsage)
flag.Parse()
@@ -99,6 +101,14 @@ func main() {
return
}
if doUpgrade {
err := upgrade()
if err != nil {
fatalln(err)
}
return
}
if len(os.Getenv("GOGC")) == 0 {
debug.SetGCPercent(25)
}
@@ -109,6 +119,22 @@ func main() {
confDir = expandTilde(confDir)
if _, err := os.Stat(confDir); err != nil && confDir == getDefaultConfDir() {
// We are supposed to use the default configuration directory. It
// doesn't exist. In the past our default has been ~/.syncthing, so if
// that directory exists we move it to the new default location and
// continue. We don't much care if this fails at this point, we will
// be checking that later.
oldDefault := expandTilde("~/.syncthing")
if _, err := os.Stat(oldDefault); err == nil {
os.MkdirAll(filepath.Dir(confDir), 0700)
if err := os.Rename(oldDefault, confDir); err == nil {
infoln("Moved config dir", oldDefault, "to", confDir)
}
}
}
// Ensure that our home directory exists and that we have a certificate and key.
ensureDir(confDir, 0700)
@@ -137,22 +163,22 @@ func main() {
cf, err := os.Open(cfgFile)
if err == nil {
// Read config.xml
cfg, err = readConfigXML(cf)
cfg, err = readConfigXML(cf, myID)
if err != nil {
fatalln(err)
}
cf.Close()
}
if len(cfg.Repositories) == 0 {
} else {
infoln("No config file; starting with empty defaults")
name, _ := os.Hostname()
defaultRepo := filepath.Join(getHomeDir(), "Sync")
ensureDir(defaultRepo, 0755)
cfg, err = readConfigXML(nil)
cfg, err = readConfigXML(nil, myID)
cfg.Repositories = []RepositoryConfiguration{
{
ID: "default",
Directory: filepath.Join(getHomeDir(), "Sync"),
Directory: defaultRepo,
Nodes: []NodeConfiguration{{NodeID: myID}},
},
}
@@ -205,18 +231,19 @@ func main() {
m := NewModel(cfg.Options.MaxChangeKbps * 1000)
for i := range cfg.Repositories {
cfg.Repositories[i].Nodes = cleanNodeList(cfg.Repositories[i].Nodes, myID)
dir := expandTilde(cfg.Repositories[i].Directory)
ensureDir(dir, -1)
m.AddRepo(cfg.Repositories[i].ID, dir, cfg.Repositories[i].Nodes)
for _, repo := range cfg.Repositories {
if repo.Invalid != "" {
continue
}
dir := expandTilde(repo.Directory)
m.AddRepo(repo.ID, dir, repo.Nodes)
}
// GUI
if cfg.GUI.Enabled && cfg.GUI.Address != "" {
addr, err := net.ResolveTCPAddr("tcp", cfg.GUI.Address)
if err != nil {
warnf("Cannot start GUI on %q: %v", cfg.GUI.Address, err)
fatalf("Cannot start GUI on %q: %v", cfg.GUI.Address, err)
} else {
var hostOpen, hostShow string
switch {
@@ -232,7 +259,10 @@ func main() {
}
infof("Starting web GUI on http://%s:%d/", hostShow, addr.Port)
startGUI(cfg.GUI, m)
err := startGUI(cfg.GUI, m)
if err != nil {
fatalln("Cannot start GUI:", err)
}
if cfg.Options.StartBrowser && len(os.Getenv("STRESTART")) == 0 {
openURL(fmt.Sprintf("http://%s:%d", hostOpen, addr.Port))
}
@@ -244,6 +274,29 @@ func main() {
infoln("Populating repository index")
m.LoadIndexes(confDir)
for _, repo := range cfg.Repositories {
if repo.Invalid != "" {
continue
}
dir := expandTilde(repo.Directory)
// Safety check. If the cached index contains files but the repository
// doesn't exist, we have a problem. We would assume that all files
// have been deleted which might not be the case, so abort instead.
if files, _, _ := m.LocalSize(repo.ID); files > 0 {
if fi, err := os.Stat(dir); err != nil || !fi.IsDir() {
warnf("Configured repository %q has index but directory %q is missing; not starting.", repo.ID, repo.Directory)
fatalf("Ensure that directory is present or remove repository from configuration.")
}
}
// Ensure that repository directories exist for newly configured repositories.
ensureDir(dir, -1)
}
m.ScanRepos()
m.SaveIndexes(confDir)
@@ -262,6 +315,10 @@ func main() {
go listenConnect(myID, m, tlsCfg)
for _, repo := range cfg.Repositories {
if repo.Invalid != "" {
continue
}
// Routine to pull blocks from other nodes to synchronize the local
// repository. Does not run when we are in read only (publish only) mode.
if repo.ReadOnly {
@@ -561,39 +618,47 @@ func ensureDir(dir string, mode int) {
}
}
func getDefaultConfDir() string {
switch runtime.GOOS {
case "windows":
return filepath.Join(os.Getenv("AppData"), "Syncthing")
case "darwin":
return expandTilde("~/Library/Application Support/Syncthing")
default:
if xdgCfg := os.Getenv("XDG_CONFIG_HOME"); xdgCfg != "" {
return filepath.Join(xdgCfg, "syncthing")
} else {
return expandTilde("~/.config/syncthing")
}
}
}
func expandTilde(p string) string {
if runtime.GOOS == "windows" {
if runtime.GOOS == "windows" || !strings.HasPrefix(p, "~/") {
return p
}
if strings.HasPrefix(p, "~/") {
return strings.Replace(p, "~", getUnixHomeDir(), 1)
}
return p
}
func getUnixHomeDir() string {
home := os.Getenv("HOME")
if home == "" {
fatalln("No home directory?")
}
return home
return filepath.Join(getHomeDir(), p[2:])
}
func getHomeDir() string {
if runtime.GOOS == "windows" {
home := os.Getenv("HOMEDRIVE") + os.Getenv("HOMEPATH")
if home == "" {
home = os.Getenv("USERPROFILE")
}
return home
}
return getUnixHomeDir()
}
var home string
func getDefaultConfDir() string {
if runtime.GOOS == "windows" {
return filepath.Join(os.Getenv("AppData"), "syncthing")
switch runtime.GOOS {
case "windows":
home = filepath.Join(os.Getenv("HomeDrive"), os.Getenv("HomePath"))
if home == "" {
home = os.Getenv("UserProfile")
}
default:
home = os.Getenv("HOME")
}
return expandTilde("~/.syncthing")
if home == "" {
fatalln("No home directory found - set $HOME (or the platform equivalent).")
}
return home
}

View File

@@ -80,8 +80,8 @@ func NewModel(maxChangeBw int) *Model {
// 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.Lock()
defer m.rmut.Unlock()
m.rmut.RLock()
defer m.rmut.RUnlock()
if dir, ok := m.repoDirs[repo]; !ok {
panic("cannot start without repo")
@@ -423,14 +423,16 @@ func (m *Model) AddConnection(rawConn io.Closer, protoConn protocol.Connection)
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() {
m.rmut.RLock()
repos := m.nodeRepos[nodeID]
m.rmut.RUnlock()
for _, repo := range repos {
m.rmut.RLock()
idx := m.protocolIndex(repo)
m.rmut.RUnlock()
for repo, idx := range idxToSend {
if debugNet {
dlog.Printf("IDX(out/initial): %s: %q: %d files", nodeID, repo, len(idx))
}
@@ -559,7 +561,7 @@ func (m *Model) ScanRepos() {
func (m *Model) ScanRepo(repo string) {
sup := &suppressor{threshold: int64(cfg.Options.MaxChangeKbps)}
m.rmut.Lock()
m.rmut.RLock()
w := &scanner.Walker{
Dir: m.repoDirs[repo],
IgnoreFile: ".stignore",
@@ -568,7 +570,7 @@ func (m *Model) ScanRepo(repo string) {
Suppressor: sup,
CurrentFiler: cFiler{m, repo},
}
m.rmut.Unlock()
m.rmut.RUnlock()
m.setState(repo, RepoScanning)
fs, _ := w.Walk()
m.ReplaceLocal(repo, fs)
@@ -648,7 +650,7 @@ func (m *Model) clusterConfig(node string) protocol.ClusterConfigMessage {
ClientVersion: Version,
}
m.rmut.Lock()
m.rmut.RLock()
for _, repo := range m.nodeRepos[node] {
cr := protocol.Repository{
ID: repo,
@@ -662,7 +664,7 @@ func (m *Model) clusterConfig(node string) protocol.ClusterConfigMessage {
}
cm.Repositories = append(cm.Repositories, cr)
}
m.rmut.Unlock()
m.rmut.RUnlock()
return cm
}
@@ -674,9 +676,9 @@ func (m *Model) setState(repo string, state repoState) {
}
func (m *Model) State(repo string) string {
m.rmut.Lock()
m.rmut.RLock()
state := m.repoState[repo]
m.rmut.Unlock()
m.rmut.RUnlock()
switch state {
case RepoIdle:
return "idle"

View File

@@ -1,34 +0,0 @@
/*
Copyright 2011 Google Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package main
import (
"os/exec"
"runtime"
)
func openURL(url string) error {
if runtime.GOOS == "windows" {
return exec.Command("cmd.exe", "/C", "start "+url).Run()
}
if runtime.GOOS == "darwin" {
return exec.Command("open", url).Run()
}
return exec.Command("xdg-open", url).Run()
}

View File

@@ -0,0 +1,23 @@
// +build !windows
package main
import (
"os/exec"
"runtime"
"syscall"
)
func openURL(url string) error {
switch runtime.GOOS {
case "darwin":
return exec.Command("open", url).Run()
default:
cmd := exec.Command("xdg-open", url)
cmd.SysProcAttr = &syscall.SysProcAttr{
Setpgid: true,
}
return cmd.Run()
}
}

View File

@@ -0,0 +1,9 @@
// +build windows
package main
import "os/exec"
func openURL(url string) error {
return exec.Command("cmd.exe", "/C", "start "+url).Run()
}

View File

@@ -135,7 +135,10 @@ func (p *puller) run() {
case b := <-p.blocks:
p.model.setState(p.repo, RepoSyncing)
changed = true
p.handleBlock(b)
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() {
@@ -193,7 +196,7 @@ func (p *puller) runRO() {
func (p *puller) fixupDirectories() {
var deleteDirs []string
fn := func(path string, info os.FileInfo, err error) error {
filepath.Walk(p.dir, func(path string, info os.FileInfo, err error) error {
if !info.IsDir() {
return nil
}
@@ -242,8 +245,7 @@ func (p *puller) fixupDirectories() {
}
return nil
}
filepath.Walk(p.dir, fn)
})
// Delete any queued directories
for i := len(deleteDirs) - 1; i >= 0; i-- {
@@ -278,53 +280,14 @@ func (p *puller) handleRequestResult(res requestResult) {
}
if of.done && of.outstanding == 0 {
if debugPull {
dlog.Printf("pull: closing %q / %q", p.repo, 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 debugPull {
dlog.Printf("pull: error: %q / %q: %v", p.repo, f.Name, err)
}
return
}
hb, _ := scanner.Blocks(fd, BlockSize)
fd.Close()
if l0, l1 := len(hb), len(f.Blocks); l0 != l1 {
if debugPull {
dlog.Printf("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 {
dlog.Printf("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))
if debugPull {
dlog.Printf("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 {
dlog.Printf("pull: error: %q / %q: %v", p.repo, f.Name, err)
}
p.closeFile(f)
}
}
func (p *puller) handleBlock(b bqBlock) {
// 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
@@ -335,8 +298,7 @@ func (p *puller) handleBlock(b bqBlock) {
os.MkdirAll(path, 0777)
}
p.model.updateLocal(p.repo, f)
p.requestSlots <- true
return
return true
}
of, ok := p.openFiles[f.Name]
@@ -368,9 +330,9 @@ func (p *puller) handleBlock(b bqBlock) {
if !b.last {
p.openFiles[f.Name] = of
}
p.requestSlots <- true
return
return true
}
defTempNamer.Hide(of.temp)
}
if of.err != nil {
@@ -383,8 +345,7 @@ func (p *puller) handleBlock(b bqBlock) {
delete(p.openFiles, f.Name)
}
p.requestSlots <- true
return
return true
}
p.openFiles[f.Name] = of
@@ -392,15 +353,14 @@ func (p *puller) handleBlock(b bqBlock) {
switch {
case len(b.copy) > 0:
p.handleCopyBlock(b)
p.requestSlots <- true
return true
case b.block.Size > 0:
p.handleRequestBlock(b)
// Request slot gets freed in <-p.blocks case
return p.handleRequestBlock(b)
default:
p.handleEmptyBlock(b)
p.requestSlots <- true
return true
}
}
@@ -448,11 +408,15 @@ func (p *puller) handleCopyBlock(b bqBlock) {
}
}
func (p *puller) handleRequestBlock(b bqBlock) {
// We have a block to get from the network
// 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 := p.openFiles[f.Name]
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 {
@@ -467,8 +431,7 @@ func (p *puller) handleRequestBlock(b bqBlock) {
} else {
p.openFiles[f.Name] = of
}
p.requestSlots <- true
return
return true
}
of.outstanding++
@@ -489,6 +452,8 @@ func (p *puller) handleRequestBlock(b bqBlock) {
err: err,
}
}(node, b)
return false
}
func (p *puller) handleEmptyBlock(b bqBlock) {
@@ -514,6 +479,7 @@ func (p *puller) handleEmptyBlock(b bqBlock) {
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)
@@ -539,3 +505,52 @@ func (p *puller) queueNeededBlocks() {
dlog.Printf("%q: queued %d blocks", p.repo, queued)
}
}
func (p *puller) closeFile(f scanner.File) {
if debugPull {
dlog.Printf("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 debugPull {
dlog.Printf("pull: error: %q / %q: %v", p.repo, f.Name, err)
}
return
}
hb, _ := scanner.Blocks(fd, BlockSize)
fd.Close()
if l0, l1 := len(hb), len(f.Blocks); l0 != l1 {
if debugPull {
dlog.Printf("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 {
dlog.Printf("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 debugPull {
dlog.Printf("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 {
dlog.Printf("pull: error: %q / %q: %v", p.repo, f.Name, err)
}
}

View File

@@ -1,3 +1,5 @@
// +build !windows
package main
import (
@@ -21,3 +23,11 @@ func (t tempNamer) TempName(name string) string {
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
}

View File

@@ -0,0 +1,56 @@
// +build windows
package main
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)
}

View File

@@ -0,0 +1,146 @@
// +build !windows
package main
import (
"archive/tar"
"compress/gzip"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"net/http"
"os"
"path"
"path/filepath"
"runtime"
"strings"
"bitbucket.org/kardianos/osext"
)
type githubRelease struct {
Tag string `json:"tag_name"`
Prelease bool `json:"prerelease"`
Assets []githubAsset `json:"assets"`
}
type githubAsset struct {
URL string `json:"url"`
Name string `json:"name"`
}
func upgrade() error {
path, err := osext.Executable()
if err != nil {
return err
}
resp, err := http.Get("https://api.github.com/repos/calmh/syncthing/releases?per_page=1")
if err != nil {
return err
}
var rels []githubRelease
json.NewDecoder(resp.Body).Decode(&rels)
resp.Body.Close()
if len(rels) != 1 {
return fmt.Errorf("Unexpected number of releases: %d", len(rels))
}
rel := rels[0]
if rel.Tag > Version {
infof("Attempting upgrade to %s...", rel.Tag)
} else if rel.Tag == Version {
okf("Already running the latest version, %s. Not upgrading.", Version)
return nil
} else {
okf("Current version %s is newer than latest release %s. Not upgrading.", Version, rel.Tag)
return nil
}
expectedRelease := fmt.Sprintf("syncthing-%s-%s-%s.", runtime.GOOS, runtime.GOARCH, rel.Tag)
for _, asset := range rel.Assets {
if strings.HasPrefix(asset.Name, expectedRelease) {
if strings.HasSuffix(asset.Name, ".tar.gz") {
infof("Downloading %s...", asset.Name)
fname, err := readTarGZ(asset.URL, filepath.Dir(path))
if err != nil {
return err
}
old := path + "." + Version
err = os.Rename(path, old)
if err != nil {
return err
}
err = os.Rename(fname, path)
if err != nil {
return err
}
okf("Upgraded %q to %s.", path, rel.Tag)
okf("Previous version saved in %q.", old)
return nil
}
}
}
return nil
}
func readTarGZ(url string, dir string) (string, error) {
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return "", err
}
req.Header.Add("Accept", "application/octet-stream")
resp, err := http.DefaultClient.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
gr, err := gzip.NewReader(resp.Body)
if err != nil {
return "", err
}
tr := tar.NewReader(gr)
if err != nil {
return "", err
}
// Iterate through the files in the archive.
for {
hdr, err := tr.Next()
if err == io.EOF {
// end of tar archive
break
}
if err != nil {
return "", err
}
if path.Base(hdr.Name) == "syncthing" {
of, err := ioutil.TempFile(dir, "syncthing")
if err != nil {
return "", err
}
io.Copy(of, tr)
err = of.Close()
if err != nil {
os.Remove(of.Name())
return "", err
}
os.Chmod(of.Name(), os.FileMode(hdr.Mode))
return of.Name(), nil
}
}
return "", fmt.Errorf("No upgrade found")
}

View File

@@ -0,0 +1,9 @@
// +build windows
package main
import "errors"
func upgrade() error {
return errors.New("Upgrade currently unsupported on Windows")
}

View File

@@ -117,8 +117,6 @@ func compareClusterConfig(local, remote protocol.ClusterConfigMessage) error {
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 node %q in repository %q", node, repo))
}
}
} else {
@@ -126,14 +124,8 @@ func compareClusterConfig(local, remote protocol.ClusterConfigMessage) error {
}
}
for repo, rnodes := range rm {
if lnodes, ok := lm[repo]; ok {
for node := range rnodes {
if _, ok := lnodes[node]; !ok {
return ClusterConfigMismatch(fmt.Errorf("remote has extra node %q in repository %q", node, repo))
}
}
} else {
for repo := range rm {
if _, ok := lm[repo]; !ok {
return ClusterConfigMismatch(fmt.Errorf("remote has extra repository %q", repo))
}

View File

@@ -103,7 +103,7 @@ var testcases = []struct {
{ID: "bar"},
},
},
err: `remote is missing node "a" in repository "foo"`,
err: "",
},
{
@@ -130,7 +130,7 @@ var testcases = []struct {
{ID: "bar"},
},
},
err: `remote has extra node "b" in repository "foo"`,
err: "",
},
{

View File

@@ -4,6 +4,7 @@ import (
"encoding/binary"
"encoding/hex"
"flag"
"fmt"
"log"
"net"
"os"
@@ -26,22 +27,28 @@ type Address struct {
}
var (
nodes = make(map[string]Node)
lock sync.Mutex
queries = 0
answered = 0
limited = 0
debug = false
limiter = lru.New(1024)
nodes = make(map[string]Node)
lock sync.Mutex
queries = 0
announces = 0
answered = 0
limited = 0
unknowns = 0
debug = false
limiter = lru.New(1024)
)
func main() {
var listen string
var timestamp bool
var statsIntv int
var statsFile string
flag.StringVar(&listen, "listen", ":22025", "Listen address")
flag.BoolVar(&debug, "debug", false, "Enable debug output")
flag.BoolVar(&timestamp, "timestamp", true, "Timestamp the log output")
flag.IntVar(&statsIntv, "stats-intv", 0, "Statistics output interval (s)")
flag.StringVar(&statsFile, "stats-file", "/var/log/discosrv.stats", "Statistics file name")
flag.Parse()
log.SetOutput(os.Stdout)
@@ -55,7 +62,9 @@ func main() {
log.Fatal(err)
}
go logStats()
if statsIntv > 0 {
go logStats(statsFile, statsIntv)
}
var buf = make([]byte, 1024)
for {
@@ -80,17 +89,16 @@ func main() {
magic := binary.BigEndian.Uint32(buf)
switch magic {
case discover.AnnouncementMagicV1:
handleAnnounceV1(addr, buf)
case discover.QueryMagicV1:
handleQueryV1(conn, addr, buf)
case discover.AnnouncementMagicV2:
handleAnnounceV2(addr, buf)
case discover.QueryMagicV2:
handleQueryV2(conn, addr, buf)
default:
lock.Lock()
unknowns++
lock.Unlock()
}
}
}
@@ -123,75 +131,6 @@ func limit(addr *net.UDPAddr) bool {
return false
}
func handleAnnounceV1(addr *net.UDPAddr, buf []byte) {
var pkt discover.AnnounceV1
err := pkt.UnmarshalXDR(buf)
if err != nil {
log.Println("AnnounceV1 Unmarshal:", err)
log.Println(hex.Dump(buf))
return
}
if debug {
log.Printf("<- %v %#v", addr, pkt)
}
ip := addr.IP.To4()
if ip == nil {
ip = addr.IP.To16()
}
node := Node{
Addresses: []Address{{
IP: ip,
Port: pkt.Port,
}},
Updated: time.Now(),
}
lock.Lock()
nodes[pkt.NodeID] = node
lock.Unlock()
}
func handleQueryV1(conn *net.UDPConn, addr *net.UDPAddr, buf []byte) {
var pkt discover.QueryV1
err := pkt.UnmarshalXDR(buf)
if err != nil {
log.Println("QueryV1 Unmarshal:", err)
log.Println(hex.Dump(buf))
return
}
if debug {
log.Printf("<- %v %#v", addr, pkt)
}
lock.Lock()
node, ok := nodes[pkt.NodeID]
queries++
lock.Unlock()
if ok && len(node.Addresses) > 0 {
pkt := discover.AnnounceV1{
Magic: discover.AnnouncementMagicV1,
NodeID: pkt.NodeID,
Port: node.Addresses[0].Port,
IP: node.Addresses[0].IP,
}
if debug {
log.Printf("-> %v %#v", addr, pkt)
}
tb := pkt.MarshalXDR()
_, _, err = conn.WriteMsgUDP(tb, nil, addr)
if err != nil {
log.Println("QueryV1 response write:", err)
}
lock.Lock()
answered++
lock.Unlock()
}
}
func handleAnnounceV2(addr *net.UDPAddr, buf []byte) {
var pkt discover.AnnounceV2
err := pkt.UnmarshalXDR(buf)
@@ -204,6 +143,10 @@ func handleAnnounceV2(addr *net.UDPAddr, buf []byte) {
log.Printf("<- %v %#v", addr, pkt)
}
lock.Lock()
announces++
lock.Unlock()
ip := addr.IP.To4()
if ip == nil {
ip = addr.IP.To16()
@@ -272,9 +215,21 @@ func handleQueryV2(conn *net.UDPConn, addr *net.UDPAddr, buf []byte) {
}
}
func logStats() {
func next(intv int) time.Time {
d := time.Duration(intv) * time.Second
t0 := time.Now()
t1 := t0.Add(d).Truncate(d)
time.Sleep(t1.Sub(t0))
return t1
}
func logStats(file string, intv int) {
f, err := os.OpenFile(file, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
if err != nil {
log.Fatal(err)
}
for {
time.Sleep(600 * time.Second)
t := next(intv)
lock.Lock()
@@ -286,11 +241,15 @@ func logStats() {
}
}
log.Printf("Expired %d nodes; %d nodes in registry; %d queries (%d answered)", deleted, len(nodes), queries, answered)
log.Printf("Limited %d queries; %d entries in limiter cache", limited, limiter.Len())
fmt.Fprintf(f, "%d Nr:%d Ne:%d Qt:%d Qa:%d A:%d U:%d Lq:%d Lc:%d\n",
t.Unix(), len(nodes), deleted, queries, answered, announces, unknowns, limited, limiter.Len())
f.Sync()
queries = 0
announces = 0
answered = 0
limited = 0
unknowns = 0
lock.Unlock()
}

View File

@@ -6,7 +6,6 @@ import (
"fmt"
"log"
"net"
"strings"
"sync"
"time"
@@ -76,6 +75,20 @@ func (d *Discoverer) ExtAnnounceOK() bool {
return d.extAnnounceOK
}
func (d *Discoverer) Lookup(node string) []string {
d.registryLock.Lock()
addr, ok := d.registry[node]
d.registryLock.Unlock()
if ok {
return addr
} else if len(d.extServer) != 0 {
// We might want to cache this, but not permanently so it needs some intelligence
return d.externalLookup(node)
}
return nil
}
func (d *Discoverer) announcementPkt() []byte {
var addrs []Address
for _, astr := range d.listenAddrs {
@@ -203,7 +216,7 @@ func (d *Discoverer) recvAnnouncements() {
for _, a := range pkt.Addresses {
var nodeAddr string
if len(a.IP) > 0 {
nodeAddr = fmt.Sprintf("%s:%d", ipStr(a.IP), a.Port)
nodeAddr = fmt.Sprintf("%s:%d", net.IP(a.IP), a.Port)
} else {
ua := addr.(*net.UDPAddr)
ua.Port = int(a.Port)
@@ -285,39 +298,8 @@ func (d *Discoverer) externalLookup(node string) []string {
var addrs []string
for _, a := range pkt.Addresses {
var nodeAddr string
if len(a.IP) > 0 {
nodeAddr = fmt.Sprintf("%s:%d", ipStr(a.IP), a.Port)
}
nodeAddr := fmt.Sprintf("%s:%d", net.IP(a.IP), a.Port)
addrs = append(addrs, nodeAddr)
}
return addrs
}
func (d *Discoverer) Lookup(node string) []string {
d.registryLock.Lock()
addr, ok := d.registry[node]
d.registryLock.Unlock()
if ok {
return addr
} else if len(d.extServer) != 0 {
// We might want to cache this, but not permanently so it needs some intelligence
return d.externalLookup(node)
}
return nil
}
func ipStr(ip []byte) string {
var f = "%d"
var s = "."
if len(ip) > 4 {
f = "%x"
s = ":"
}
var ss = make([]string, len(ip))
for i := range ip {
ss[i] = fmt.Sprintf(f, ip[i])
}
return strings.Join(ss, s)
}

View File

@@ -1,22 +1,5 @@
package discover
const (
AnnouncementMagicV1 = 0x20121025
QueryMagicV1 = 0x19760309
)
type QueryV1 struct {
Magic uint32
NodeID string // max:64
}
type AnnounceV1 struct {
Magic uint32
Port uint16
NodeID string // max:64
IP []byte // max:16
}
const (
AnnouncementMagicV2 = 0x029E4C77
QueryMagicV2 = 0x23D63A9A

View File

@@ -7,89 +7,6 @@ import (
"github.com/calmh/syncthing/xdr"
)
func (o QueryV1) EncodeXDR(w io.Writer) (int, error) {
var xw = xdr.NewWriter(w)
return o.encodeXDR(xw)
}
func (o QueryV1) MarshalXDR() []byte {
var buf bytes.Buffer
var xw = xdr.NewWriter(&buf)
o.encodeXDR(xw)
return buf.Bytes()
}
func (o QueryV1) encodeXDR(xw *xdr.Writer) (int, error) {
xw.WriteUint32(o.Magic)
if len(o.NodeID) > 64 {
return xw.Tot(), xdr.ErrElementSizeExceeded
}
xw.WriteString(o.NodeID)
return xw.Tot(), xw.Error()
}
func (o *QueryV1) DecodeXDR(r io.Reader) error {
xr := xdr.NewReader(r)
return o.decodeXDR(xr)
}
func (o *QueryV1) UnmarshalXDR(bs []byte) error {
var buf = bytes.NewBuffer(bs)
var xr = xdr.NewReader(buf)
return o.decodeXDR(xr)
}
func (o *QueryV1) decodeXDR(xr *xdr.Reader) error {
o.Magic = xr.ReadUint32()
o.NodeID = xr.ReadStringMax(64)
return xr.Error()
}
func (o AnnounceV1) EncodeXDR(w io.Writer) (int, error) {
var xw = xdr.NewWriter(w)
return o.encodeXDR(xw)
}
func (o AnnounceV1) MarshalXDR() []byte {
var buf bytes.Buffer
var xw = xdr.NewWriter(&buf)
o.encodeXDR(xw)
return buf.Bytes()
}
func (o AnnounceV1) encodeXDR(xw *xdr.Writer) (int, error) {
xw.WriteUint32(o.Magic)
xw.WriteUint16(o.Port)
if len(o.NodeID) > 64 {
return xw.Tot(), xdr.ErrElementSizeExceeded
}
xw.WriteString(o.NodeID)
if len(o.IP) > 16 {
return xw.Tot(), xdr.ErrElementSizeExceeded
}
xw.WriteBytes(o.IP)
return xw.Tot(), xw.Error()
}
func (o *AnnounceV1) DecodeXDR(r io.Reader) error {
xr := xdr.NewReader(r)
return o.decodeXDR(xr)
}
func (o *AnnounceV1) UnmarshalXDR(bs []byte) error {
var buf = bytes.NewBuffer(bs)
var xr = xdr.NewReader(buf)
return o.decodeXDR(xr)
}
func (o *AnnounceV1) decodeXDR(xr *xdr.Reader) error {
o.Magic = xr.ReadUint32()
o.Port = xr.ReadUint16()
o.NodeID = xr.ReadStringMax(64)
o.IP = xr.ReadBytesMax(16)
return xr.Error()
}
func (o QueryV2) EncodeXDR(w io.Writer) (int, error) {
var xw = xdr.NewWriter(w)
return o.encodeXDR(xw)

View File

@@ -108,8 +108,8 @@ func (m *Set) Need(id uint) []scanner.File {
if debug {
dlog.Printf("Need(%d)", id)
}
var fs = make([]scanner.File, 0, len(m.globalKey)/2) // Just a guess, but avoids too many reallocations
m.Lock()
var fs = make([]scanner.File, 0, len(m.globalKey)/2) // Just a guess, but avoids too many reallocations
rkID := m.remoteKey[id]
for gk, gf := range m.files {
if !gf.Global {
@@ -145,8 +145,8 @@ func (m *Set) Global() []scanner.File {
if debug {
dlog.Printf("Global()")
}
var fs = make([]scanner.File, 0, len(m.globalKey))
m.Lock()
var fs = make([]scanner.File, 0, len(m.globalKey))
for _, file := range m.files {
if file.Global {
fs = append(fs, file.File)

View File

@@ -4,6 +4,7 @@
'use strict';
var syncthing = angular.module('syncthing', []);
var urlbase = 'rest';
syncthing.controller('SyncthingCtrl', function ($scope, $http) {
var prevDate = 0;
@@ -75,18 +76,18 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
}
$scope.refresh = function () {
$http.get('/rest/system').success(function (data) {
$http.get(urlbase + '/system').success(function (data) {
getSucceeded();
$scope.system = data;
}).error(function () {
getFailed();
});
$scope.repos.forEach(function (repo) {
$http.get('/rest/model/' + repo.ID).success(function (data) {
$http.get(urlbase + '/model?repo=' + encodeURIComponent(repo.ID)).success(function (data) {
$scope.model[repo.ID] = data;
});
});
$http.get('/rest/connections').success(function (data) {
$http.get(urlbase + '/connections').success(function (data) {
var now = Date.now(),
td = (now - prevDate) / 1000,
id;
@@ -111,7 +112,7 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
}
$scope.connections = data;
});
$http.get('/rest/errors').success(function (data) {
$http.get(urlbase + '/errors').success(function (data) {
$scope.errors = data;
});
};
@@ -121,6 +122,10 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
return 'Unknown';
}
if ($scope.model[repo].invalid !== '') {
return 'Stopped';
}
var state = '' + $scope.model[repo].state;
state = state[0].toUpperCase() + state.substr(1);
@@ -136,6 +141,10 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
return 'text-info';
}
if ($scope.model[repo].invalid !== '') {
return 'text-warning';
}
var state = '' + $scope.model[repo].state;
if (state == 'idle') {
return 'text-success';
@@ -238,20 +247,21 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
$scope.saveSettings = function () {
$scope.configInSync = false;
$scope.config.Options.ListenAddress = $scope.config.Options.ListenStr.split(',').map(function (x) { return x.trim(); });
$http.post('/rest/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
$http.post(urlbase + '/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
$('#settings').modal("hide");
};
$scope.restart = function () {
restarting = true;
$('#restarting').modal('show');
$http.post('/rest/restart');
$http.post(urlbase + '/restart');
$scope.configInSync = true;
};
$scope.editNode = function (nodeCfg) {
$scope.currentNode = $.extend({}, nodeCfg);
$scope.editingExisting = true;
$scope.editingSelf = (nodeCfg.NodeID == $scope.myID);
$scope.currentNode.AddressesStr = nodeCfg.Addresses.join(', ');
$('#editNode').modal({backdrop: 'static', keyboard: true});
};
@@ -259,6 +269,7 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
$scope.addNode = function () {
$scope.currentNode = {AddressesStr: 'dynamic'};
$scope.editingExisting = false;
$scope.editingSelf = false;
$('#editNode').modal({backdrop: 'static', keyboard: true});
};
@@ -280,7 +291,7 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
}
$scope.configInSync = false;
$http.post('/rest/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
$http.post(urlbase + '/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
};
$scope.saveNode = function () {
@@ -307,7 +318,7 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
$scope.nodes.sort(nodeCompare);
$scope.config.Nodes = $scope.nodes;
$http.post('/rest/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
$http.post(urlbase + '/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
};
$scope.otherNodes = function () {
@@ -335,7 +346,7 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
$scope.clearErrors = function () {
$scope.seenError = $scope.errors[$scope.errors.length - 1].Time;
$http.post('/rest/error/clear');
$http.post(urlbase + '/error/clear');
};
$scope.friendlyNodes = function (str) {
@@ -391,7 +402,7 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
$scope.config.Repositories = $scope.repos;
$http.post('/rest/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
$http.post(urlbase + '/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
};
$scope.deleteRepo = function () {
@@ -407,19 +418,19 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
$scope.config.Repositories = $scope.repos;
$scope.configInSync = false;
$http.post('/rest/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
$http.post(urlbase + '/config', JSON.stringify($scope.config), {headers: {'Content-Type': 'application/json'}});
};
$http.get('/rest/version').success(function (data) {
$http.get(urlbase + '/version').success(function (data) {
$scope.version = data;
});
$http.get('/rest/system').success(function (data) {
$http.get(urlbase + '/system').success(function (data) {
$scope.system = data;
$scope.myID = data.myID;
});
$http.get('/rest/config').success(function (data) {
$http.get(urlbase + '/config').success(function (data) {
$scope.config = data;
$scope.config.Options.ListenStr = $scope.config.Options.ListenAddress.join(', ');
@@ -432,7 +443,7 @@ syncthing.controller('SyncthingCtrl', function ($scope, $http) {
$scope.refresh();
});
$http.get('/rest/config/sync').success(function (data) {
$http.get(urlbase + '/config/sync').success(function (data) {
$scope.configInSync = data.configInSync;
});

View File

@@ -134,6 +134,7 @@
<ul class="list-unstyled" ng-repeat="repo in repos">
<li>
<span class="text-monospace">{{repo.Directory}}</span>
<span ng-if="repo.Invalid" class="label label-danger">Invalid: {{repo.Invalid}}</span>
<ul class="list-no-bullet">
<li>
<div class="li-column" title="Repository ID">
@@ -278,7 +279,7 @@
<div class="panel panel-warning">
<div class="panel-heading"><h3 class="panel-title">Notice</h3></div>
<div class="panel-body">
<p ng-repeat="err in errorList()"><small>{{err.Time | date:"hh:mm:ss.sss"}}:</small> {{friendlyNodes(err.Error)}}</p>
<p ng-repeat="err in errorList()"><small>{{err.Time | date:"H:mm:ss"}}:</small> {{friendlyNodes(err.Error)}}</p>
</div>
<div class="panel-footer">
<button type="button" class="pull-right btn btn-sm btn-default" ng-click="clearErrors()">OK</button>
@@ -360,8 +361,9 @@
<form role="form">
<div class="form-group">
<label for="nodeID">Node ID</label>
<input ng-disabled="editingExisting" id="nodeID" class="form-control" type="text" ng-model="currentNode.NodeID"></input>
<p class="help-block">The node ID can be found in the logs or in the "Add Node" dialog on the other node.</p>
<input ng-if="!editingExisting" id="nodeID" class="form-control" type="text" ng-model="currentNode.NodeID"></input>
<div ng-if="editingExisting" class="well well-sm">{{currentNode.NodeID}}</div>
<p class="help-block">The node ID can be found in the "Add Node" dialog on the other node.</p>
</div>
<div class="form-group">
<label for="name">Name</label>
@@ -382,7 +384,7 @@
<div class="modal-footer">
<button type="button" class="btn btn-primary" ng-click="saveNode()">Save</button>
<button type="button" class="btn btn-default" data-dismiss="modal">Close</button>
<button ng-if="editingExisting" type="button" class="btn btn-danger pull-left" ng-click="deleteNode()">Delete</button>
<button ng-if="editingExisting && !editingSelf" type="button" class="btn btn-danger pull-left" ng-click="deleteNode()">Delete</button>
</div>
</div>
</div>

View File

@@ -1,3 +1,5 @@
// +build ignore
package main
import (

View File

@@ -1,3 +1,5 @@
// +build ignore
package main
import (

View File

@@ -1,3 +1,5 @@
// +build ignore
package main
import (

View File

@@ -45,6 +45,9 @@ func (b *Beacon) run() {
if err != nil {
log.Fatal(err)
}
if debug {
dlog.Printf("trying %d interfaces", len(intfs))
}
for _, intf := range intfs {
intf := intf
@@ -55,10 +58,13 @@ func (b *Beacon) run() {
conn, err := net.ListenMulticastUDP("udp4", &intf, group)
if err != nil {
if debug {
dlog.Printf("listen for multicast group on %q: %v", intf.Name, err)
dlog.Printf("failed to listen for multicast group on %q: %v", intf.Name, err)
}
} else {
b.conns = append(b.conns, conn)
if debug {
dlog.Printf("listening for multicast group on %q", intf.Name)
}
}
}
@@ -72,6 +78,9 @@ func (b *Beacon) run() {
dlog.Println(err)
return
}
if debug {
dlog.Printf("recv %d bytes from %s on %v", n, addr, conn)
}
b.outbox <- recv{bs[:n], addr}
}
}()
@@ -85,6 +94,9 @@ func (b *Beacon) run() {
dlog.Println(err)
return
}
if debug {
dlog.Printf("sent %d bytes to %s on %v", len(bs), group, conn)
}
}
}
}()

36
protocol/counting.go Normal file
View File

@@ -0,0 +1,36 @@
package protocol
import (
"io"
"sync/atomic"
)
type countingReader struct {
io.Reader
tot uint64
}
func (c *countingReader) Read(bs []byte) (int, error) {
n, err := c.Reader.Read(bs)
atomic.AddUint64(&c.tot, uint64(n))
return n, err
}
func (c *countingReader) Tot() uint64 {
return atomic.LoadUint64(&c.tot)
}
type countingWriter struct {
io.Writer
tot uint64
}
func (c *countingWriter) Write(bs []byte) (int, error) {
n, err := c.Writer.Write(bs)
atomic.AddUint64(&c.tot, uint64(n))
return n, err
}
func (c *countingWriter) Tot() uint64 {
return atomic.LoadUint64(&c.tot)
}

View File

@@ -64,22 +64,26 @@ type Connection interface {
}
type rawConnection struct {
sync.RWMutex
id string
receiver Model
reader io.ReadCloser
cr *countingReader
xr *xdr.Reader
writer io.WriteCloser
cw *countingWriter
wb *bufio.Writer
xw *xdr.Writer
wmut sync.Mutex
close chan error
closed chan struct{}
id string
receiver Model
reader io.ReadCloser
xr *xdr.Reader
writer io.WriteCloser
wb *bufio.Writer
xw *xdr.Writer
closed chan struct{}
awaiting map[int]chan asyncResult
nextID int
indexSent map[string]map[string][2]int64
hasSentIndex bool
hasRecvdIndex bool
imut sync.Mutex
}
type asyncResult struct {
@@ -93,8 +97,11 @@ const (
)
func NewConnection(nodeID string, reader io.Reader, writer io.Writer, receiver Model) Connection {
flrd := flate.NewReader(reader)
flwr, err := flate.NewWriter(writer, flate.BestSpeed)
cr := &countingReader{Reader: reader}
cw := &countingWriter{Writer: writer}
flrd := flate.NewReader(cr)
flwr, err := flate.NewWriter(cw, flate.BestSpeed)
if err != nil {
panic(err)
}
@@ -104,15 +111,19 @@ func NewConnection(nodeID string, reader io.Reader, writer io.Writer, receiver M
id: nodeID,
receiver: nativeModel{receiver},
reader: flrd,
cr: cr,
xr: xdr.NewReader(flrd),
writer: flwr,
cw: cw,
wb: wb,
xw: xdr.NewWriter(wb),
close: make(chan error),
closed: make(chan struct{}),
awaiting: make(map[int]chan asyncResult),
indexSent: make(map[string]map[string][2]int64),
}
go c.closer()
go c.readerLoop()
go c.pingerLoop()
@@ -125,11 +136,11 @@ func (c *rawConnection) ID() string {
// Index writes the list of file information to the connected peer node
func (c *rawConnection) Index(repo string, idx []FileInfo) {
c.Lock()
if c.isClosed() {
c.Unlock()
return
}
c.imut.Lock()
var msgType int
if c.indexSent[repo] == nil {
// This is the first time we send an index.
@@ -152,45 +163,48 @@ func (c *rawConnection) Index(repo string, idx []FileInfo) {
idx = diff
}
header{0, c.nextID, msgType}.encodeXDR(c.xw)
_, err := IndexMessage{repo, idx}.encodeXDR(c.xw)
if err == nil {
err = c.flush()
}
id := c.nextID
c.nextID = (c.nextID + 1) & 0xfff
c.hasSentIndex = true
c.Unlock()
c.imut.Unlock()
c.wmut.Lock()
header{0, id, msgType}.encodeXDR(c.xw)
IndexMessage{repo, idx}.encodeXDR(c.xw)
err := c.flush()
c.wmut.Unlock()
if err != nil {
c.close(err)
c.close <- err
return
}
}
// Request returns the bytes for the specified block after fetching them from the connected peer.
func (c *rawConnection) Request(repo string, name string, offset int64, size int) ([]byte, error) {
c.Lock()
if c.isClosed() {
c.Unlock()
return nil, ErrClosed
}
c.imut.Lock()
id := c.nextID
c.nextID = (c.nextID + 1) & 0xfff
rc := make(chan asyncResult)
if _, ok := c.awaiting[c.nextID]; ok {
if _, ok := c.awaiting[id]; ok {
panic("id taken")
}
c.awaiting[c.nextID] = rc
header{0, c.nextID, messageTypeRequest}.encodeXDR(c.xw)
_, err := RequestMessage{repo, name, uint64(offset), uint32(size)}.encodeXDR(c.xw)
if err == nil {
err = c.flush()
}
c.awaiting[id] = rc
c.imut.Unlock()
c.wmut.Lock()
header{0, id, messageTypeRequest}.encodeXDR(c.xw)
RequestMessage{repo, name, uint64(offset), uint32(size)}.encodeXDR(c.xw)
err := c.flush()
c.wmut.Unlock()
if err != nil {
c.Unlock()
c.close(err)
c.close <- err
return nil, err
}
c.nextID = (c.nextID + 1) & 0xfff
c.Unlock()
res, ok := <-rc
if !ok {
@@ -201,46 +215,47 @@ func (c *rawConnection) Request(repo string, name string, offset int64, size int
// ClusterConfig send the cluster configuration message to the peer and returns any error
func (c *rawConnection) ClusterConfig(config ClusterConfigMessage) {
c.Lock()
defer c.Unlock()
if c.isClosed() {
return
}
header{0, c.nextID, messageTypeClusterConfig}.encodeXDR(c.xw)
c.imut.Lock()
id := c.nextID
c.nextID = (c.nextID + 1) & 0xfff
c.imut.Unlock()
c.wmut.Lock()
header{0, id, messageTypeClusterConfig}.encodeXDR(c.xw)
config.encodeXDR(c.xw)
err := c.flush()
c.wmut.Unlock()
_, err := config.encodeXDR(c.xw)
if err == nil {
err = c.flush()
}
if err != nil {
c.close(err)
c.close <- err
}
}
func (c *rawConnection) ping() bool {
c.Lock()
if c.isClosed() {
c.Unlock()
return false
}
rc := make(chan asyncResult, 1)
c.awaiting[c.nextID] = rc
header{0, c.nextID, messageTypePing}.encodeXDR(c.xw)
err := c.flush()
if err != nil {
c.Unlock()
c.close(err)
return false
} else if c.xw.Error() != nil {
c.Unlock()
c.close(c.xw.Error())
return false
}
c.imut.Lock()
id := c.nextID
c.nextID = (c.nextID + 1) & 0xfff
c.Unlock()
rc := make(chan asyncResult, 1)
c.awaiting[id] = rc
c.imut.Unlock()
c.wmut.Lock()
header{0, id, messageTypePing}.encodeXDR(c.xw)
err := c.flush()
c.wmut.Unlock()
if err != nil {
c.close <- err
return false
}
res, ok := <-rc
return ok && res.err == nil
@@ -251,21 +266,24 @@ type flusher interface {
}
func (c *rawConnection) flush() error {
c.wb.Flush()
if err := c.xw.Error(); err != nil {
return err
}
if err := c.wb.Flush(); err != nil {
return err
}
if f, ok := c.writer.(flusher); ok {
return f.Flush()
}
return nil
}
func (c *rawConnection) close(err error) {
c.Lock()
select {
case <-c.closed:
c.Unlock()
return
default:
}
func (c *rawConnection) closer() {
err := <-c.close
close(c.closed)
for _, ch := range c.awaiting {
close(ch)
@@ -273,7 +291,6 @@ func (c *rawConnection) close(err error) {
c.awaiting = nil
c.writer.Close()
c.reader.Close()
c.Unlock()
c.receiver.Close(c.id, err)
}
@@ -292,12 +309,12 @@ loop:
for !c.isClosed() {
var hdr header
hdr.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
if err := c.xr.Error(); err != nil {
c.close <- err
break loop
}
if hdr.version != 0 {
c.close(fmt.Errorf("protocol error: %s: unknown message version %#x", c.id, hdr.version))
c.close <- fmt.Errorf("protocol error: %s: unknown message version %#x", c.id, hdr.version)
break loop
}
@@ -305,8 +322,8 @@ loop:
case messageTypeIndex:
var im IndexMessage
im.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
if err := c.xr.Error(); err != nil {
c.close <- err
break loop
} else {
@@ -319,15 +336,12 @@ loop:
go c.receiver.Index(c.id, im.Repository, im.Files)
}
c.Lock()
c.hasRecvdIndex = true
c.Unlock()
case messageTypeIndexUpdate:
var im IndexMessage
im.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
if err := c.xr.Error(); err != nil {
c.close <- err
break loop
} else {
go c.receiver.IndexUpdate(c.id, im.Repository, im.Files)
@@ -336,8 +350,8 @@ loop:
case messageTypeRequest:
var req RequestMessage
req.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
if err := c.xr.Error(); err != nil {
c.close <- err
break loop
}
go c.processRequest(hdr.msgID, req)
@@ -345,16 +359,16 @@ loop:
case messageTypeResponse:
data := c.xr.ReadBytesMax(256 * 1024) // Sufficiently larger than max expected block size
if c.xr.Error() != nil {
c.close(c.xr.Error())
if err := c.xr.Error(); err != nil {
c.close <- err
break loop
}
go func(hdr header, err error) {
c.Lock()
c.imut.Lock()
rc, ok := c.awaiting[hdr.msgID]
delete(c.awaiting, hdr.msgID)
c.Unlock()
c.imut.Unlock()
if ok {
rc <- asyncResult{data, err}
@@ -363,44 +377,41 @@ loop:
}(hdr, c.xr.Error())
case messageTypePing:
c.Lock()
c.wmut.Lock()
header{0, hdr.msgID, messageTypePong}.encodeXDR(c.xw)
err := c.flush()
c.Unlock()
c.wmut.Unlock()
if err != nil {
c.close(err)
break loop
} else if c.xw.Error() != nil {
c.close(c.xw.Error())
c.close <- err
break loop
}
case messageTypePong:
c.RLock()
c.imut.Lock()
rc, ok := c.awaiting[hdr.msgID]
c.RUnlock()
if ok {
rc <- asyncResult{}
close(rc)
go func() {
rc <- asyncResult{}
close(rc)
}()
c.Lock()
delete(c.awaiting, hdr.msgID)
c.Unlock()
}
c.imut.Unlock()
case messageTypeClusterConfig:
var cm ClusterConfigMessage
cm.decodeXDR(c.xr)
if c.xr.Error() != nil {
c.close(c.xr.Error())
if err := c.xr.Error(); err != nil {
c.close <- err
break loop
} else {
go c.receiver.ClusterConfig(c.id, cm)
}
default:
c.close(fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType))
c.close <- fmt.Errorf("protocol error: %s: unknown message type %#x", c.id, hdr.msgType)
break loop
}
}
@@ -409,17 +420,16 @@ loop:
func (c *rawConnection) processRequest(msgID int, req RequestMessage) {
data, _ := c.receiver.Request(c.id, req.Repository, req.Name, int64(req.Offset), int(req.Size))
c.Lock()
c.wmut.Lock()
header{0, msgID, messageTypeResponse}.encodeXDR(c.xw)
_, err := c.xw.WriteBytes(data)
if err == nil {
err = c.flush()
}
c.Unlock()
c.xw.WriteBytes(data)
err := c.flush()
c.wmut.Unlock()
buffers.Put(data)
if err != nil {
c.close(err)
c.close <- err
}
}
@@ -429,22 +439,16 @@ func (c *rawConnection) pingerLoop() {
for {
select {
case <-ticker:
c.RLock()
ready := c.hasRecvdIndex && c.hasSentIndex
c.RUnlock()
if ready {
go func() {
rc <- c.ping()
}()
select {
case ok := <-rc:
if !ok {
c.close(fmt.Errorf("ping failure"))
}
case <-time.After(pingTimeout):
c.close(fmt.Errorf("ping timeout"))
go func() {
rc <- c.ping()
}()
select {
case ok := <-rc:
if !ok {
c.close <- fmt.Errorf("ping failure")
}
case <-time.After(pingTimeout):
c.close <- fmt.Errorf("ping timeout")
}
case <-c.closed:
return
@@ -461,7 +465,7 @@ type Statistics struct {
func (c *rawConnection) Statistics() Statistics {
return Statistics{
At: time.Now(),
InBytesTotal: int(c.xr.Tot()),
OutBytesTotal: int(c.xw.Tot()),
InBytesTotal: int(c.cr.Tot()),
OutBytesTotal: int(c.cw.Tot()),
}
}

View File

@@ -5,6 +5,7 @@ import (
"io"
"testing"
"testing/quick"
"time"
)
func TestHeaderFunctions(t *testing.T) {
@@ -172,7 +173,13 @@ func TestClose(t *testing.T) {
c0 := NewConnection("c0", ar, bw, m0).(wireFormatConnection).next.(*rawConnection)
NewConnection("c1", br, aw, m1)
c0.close(nil)
c0.close <- nil
select {
case <-c0.closed:
case <-time.After(1 * time.Second):
t.Fatal("Did not close within a second")
}
if !c0.isClosed() {
t.Fatal("Connection should be closed")