discosrv: Remove duplicate logging of limiter cache entries

Still needs implemention in GUI

Godeps/Godeps.json

@@ -27,6 +27,10 @@
 			"ImportPath": "github.com/codegangsta/martini",
 			"Comment": "v0.1-142-g8659df7",
 			"Rev": "8659df7a51aebe6c6120268cd5a8b4c34fa8441a"
+		},
+		{
+			"ImportPath": "github.com/juju/ratelimit",
+			"Rev": "cbaa435c80a9716e086f25d409344b26c4039358"
 		}
 	]
 }

Godeps/_workspace/src/github.com/juju/ratelimit/LICENSE

@@ -0,0 +1,185 @@
+This software is licensed under the LGPLv3, included below.
+
+As a special exception to the GNU Lesser General Public License version 3
+("LGPL3"), the copyright holders of this Library give you permission to
+convey to a third party a Combined Work that links statically or dynamically
+to this Library without providing any Minimal Corresponding Source or
+Minimal Application Code as set out in 4d or providing the installation
+information set out in section 4e, provided that you comply with the other
+provisions of LGPL3 and provided that you meet, for the Application the
+terms and conditions of the license(s) which apply to the Application.
+
+Except as stated in this special exception, the provisions of LGPL3 will
+continue to comply in full to this Library. If you modify this Library, you
+may apply this exception to your version of this Library, but you are not
+obliged to do so. If you do not wish to do so, delete this exception
+statement from your version. This exception does not (and cannot) modify any
+license terms which apply to the Application, with which you must still
+comply.
+
+
+                   GNU LESSER GENERAL PUBLIC LICENSE
+                       Version 3, 29 June 2007
+
+ Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>
+ Everyone is permitted to copy and distribute verbatim copies
+ of this license document, but changing it is not allowed.
+
+
+  This version of the GNU Lesser General Public License incorporates
+the terms and conditions of version 3 of the GNU General Public
+License, supplemented by the additional permissions listed below.
+
+  0. Additional Definitions.
+
+  As used herein, "this License" refers to version 3 of the GNU Lesser
+General Public License, and the "GNU GPL" refers to version 3 of the GNU
+General Public License.
+
+  "The Library" refers to a covered work governed by this License,
+other than an Application or a Combined Work as defined below.
+
+  An "Application" is any work that makes use of an interface provided
+by the Library, but which is not otherwise based on the Library.
+Defining a subclass of a class defined by the Library is deemed a mode
+of using an interface provided by the Library.
+
+  A "Combined Work" is a work produced by combining or linking an
+Application with the Library.  The particular version of the Library
+with which the Combined Work was made is also called the "Linked
+Version".
+
+  The "Minimal Corresponding Source" for a Combined Work means the
+Corresponding Source for the Combined Work, excluding any source code
+for portions of the Combined Work that, considered in isolation, are
+based on the Application, and not on the Linked Version.
+
+  The "Corresponding Application Code" for a Combined Work means the
+object code and/or source code for the Application, including any data
+and utility programs needed for reproducing the Combined Work from the
+Application, but excluding the System Libraries of the Combined Work.
+
+  1. Exception to Section 3 of the GNU GPL.
+
+  You may convey a covered work under sections 3 and 4 of this License
+without being bound by section 3 of the GNU GPL.
+
+  2. Conveying Modified Versions.
+
+  If you modify a copy of the Library, and, in your modifications, a
+facility refers to a function or data to be supplied by an Application
+that uses the facility (other than as an argument passed when the
+facility is invoked), then you may convey a copy of the modified
+version:
+
+   a) under this License, provided that you make a good faith effort to
+   ensure that, in the event an Application does not supply the
+   function or data, the facility still operates, and performs
+   whatever part of its purpose remains meaningful, or
+
+   b) under the GNU GPL, with none of the additional permissions of
+   this License applicable to that copy.
+
+  3. Object Code Incorporating Material from Library Header Files.
+
+  The object code form of an Application may incorporate material from
+a header file that is part of the Library.  You may convey such object
+code under terms of your choice, provided that, if the incorporated
+material is not limited to numerical parameters, data structure
+layouts and accessors, or small macros, inline functions and templates
+(ten or fewer lines in length), you do both of the following:
+
+   a) Give prominent notice with each copy of the object code that the
+   Library is used in it and that the Library and its use are
+   covered by this License.
+
+   b) Accompany the object code with a copy of the GNU GPL and this license
+   document.
+
+  4. Combined Works.
+
+  You may convey a Combined Work under terms of your choice that,
+taken together, effectively do not restrict modification of the
+portions of the Library contained in the Combined Work and reverse
+engineering for debugging such modifications, if you also do each of
+the following:
+
+   a) Give prominent notice with each copy of the Combined Work that
+   the Library is used in it and that the Library and its use are
+   covered by this License.
+
+   b) Accompany the Combined Work with a copy of the GNU GPL and this license
+   document.
+
+   c) For a Combined Work that displays copyright notices during
+   execution, include the copyright notice for the Library among
+   these notices, as well as a reference directing the user to the
+   copies of the GNU GPL and this license document.
+
+   d) Do one of the following:
+
+       0) Convey the Minimal Corresponding Source under the terms of this
+       License, and the Corresponding Application Code in a form
+       suitable for, and under terms that permit, the user to
+       recombine or relink the Application with a modified version of
+       the Linked Version to produce a modified Combined Work, in the
+       manner specified by section 6 of the GNU GPL for conveying
+       Corresponding Source.
+
+       1) Use a suitable shared library mechanism for linking with the
+       Library.  A suitable mechanism is one that (a) uses at run time
+       a copy of the Library already present on the user's computer
+       system, and (b) will operate properly with a modified version
+       of the Library that is interface-compatible with the Linked
+       Version.
+
+   e) Provide Installation Information, but only if you would otherwise
+   be required to provide such information under section 6 of the
+   GNU GPL, and only to the extent that such information is
+   necessary to install and execute a modified version of the
+   Combined Work produced by recombining or relinking the
+   Application with a modified version of the Linked Version. (If
+   you use option 4d0, the Installation Information must accompany
+   the Minimal Corresponding Source and Corresponding Application
+   Code. If you use option 4d1, you must provide the Installation
+   Information in the manner specified by section 6 of the GNU GPL
+   for conveying Corresponding Source.)
+
+  5. Combined Libraries.
+
+  You may place library facilities that are a work based on the
+Library side by side in a single library together with other library
+facilities that are not Applications and are not covered by this
+License, and convey such a combined library under terms of your
+choice, if you do both of the following:
+
+   a) Accompany the combined library with a copy of the same work based
+   on the Library, uncombined with any other library facilities,
+   conveyed under the terms of this License.
+
+   b) Give prominent notice with the combined library that part of it
+   is a work based on the Library, and explaining where to find the
+   accompanying uncombined form of the same work.
+
+  6. Revised Versions of the GNU Lesser General Public License.
+
+  The Free Software Foundation may publish revised and/or new versions
+of the GNU Lesser General Public License from time to time. Such new
+versions will be similar in spirit to the present version, but may
+differ in detail to address new problems or concerns.
+
+  Each version is given a distinguishing version number. If the
+Library as you received it specifies that a certain numbered version
+of the GNU Lesser General Public License "or any later version"
+applies to it, you have the option of following the terms and
+conditions either of that published version or of any later version
+published by the Free Software Foundation. If the Library as you
+received it does not specify a version number of the GNU Lesser
+General Public License, you may choose any version of the GNU Lesser
+General Public License ever published by the Free Software Foundation.
+
+  If the Library as you received it specifies that a proxy can decide
+whether future versions of the GNU Lesser General Public License shall
+apply, that proxy's public statement of acceptance of any version is
+permanent authorization for you to choose that version for the
+Library.

Godeps/_workspace/src/github.com/juju/ratelimit/README.md

@@ -0,0 +1,109 @@
+# ratelimit
+--
+    import "github.com/juju/ratelimit"
+
+The ratelimit package provides an efficient token bucket implementation. See
+http://en.wikipedia.org/wiki/Token_bucket.
+
+## Usage
+
+#### func  Reader
+
+```go
+func Reader(r io.Reader, bucket *Bucket) io.Reader
+```
+Reader returns a reader that is rate limited by the given token bucket. Each
+token in the bucket represents one byte.
+
+#### func  Writer
+
+```go
+func Writer(w io.Writer, bucket *Bucket) io.Writer
+```
+Writer returns a reader that is rate limited by the given token bucket. Each
+token in the bucket represents one byte.
+
+#### type Bucket
+
+```go
+type Bucket struct {
+}
+```
+
+Bucket represents a token bucket that fills at a predetermined rate. Methods on
+Bucket may be called concurrently.
+
+#### func  NewBucket
+
+```go
+func NewBucket(fillInterval time.Duration, capacity int64) *Bucket
+```
+NewBucket returns a new token bucket that fills at the rate of one token every
+fillInterval, up to the given maximum capacity. Both arguments must be positive.
+The bucket is initially full.
+
+#### func  NewBucketWithRate
+
+```go
+func NewBucketWithRate(rate float64, capacity int64) *Bucket
+```
+NewBucketWithRate returns a token bucket that fills the bucket at the rate of
+rate tokens per second up to the given maximum capacity. Because of limited
+clock resolution, at high rates, the actual rate may be up to 1% different from
+the specified rate.
+
+#### func (*Bucket) Rate
+
+```go
+func (tb *Bucket) Rate() float64
+```
+Rate returns the fill rate of the bucket, in tokens per second.
+
+#### func (*Bucket) Take
+
+```go
+func (tb *Bucket) Take(count int64) time.Duration
+```
+Take takes count tokens from the bucket without blocking. It returns the time
+that the caller should wait until the tokens are actually available.
+
+Note that if the request is irrevocable - there is no way to return tokens to
+the bucket once this method commits us to taking them.
+
+#### func (*Bucket) TakeAvailable
+
+```go
+func (tb *Bucket) TakeAvailable(count int64) int64
+```
+TakeAvailable takes up to count immediately available tokens from the bucket. It
+returns the number of tokens removed, or zero if there are no available tokens.
+It does not block.
+
+#### func (*Bucket) TakeMaxDuration
+
+```go
+func (tb *Bucket) TakeMaxDuration(count int64, maxWait time.Duration) (time.Duration, bool)
+```
+TakeMaxDuration is like Take, except that it will only take tokens from the
+bucket if the wait time for the tokens is no greater than maxWait.
+
+If it would take longer than maxWait for the tokens to become available, it does
+nothing and reports false, otherwise it returns the time that the caller should
+wait until the tokens are actually available, and reports true.
+
+#### func (*Bucket) Wait
+
+```go
+func (tb *Bucket) Wait(count int64)
+```
+Wait takes count tokens from the bucket, waiting until they are available.
+
+#### func (*Bucket) WaitMaxDuration
+
+```go
+func (tb *Bucket) WaitMaxDuration(count int64, maxWait time.Duration) bool
+```
+WaitMaxDuration is like Wait except that it will only take tokens from the
+bucket if it needs to wait for no greater than maxWait. It reports whether any
+tokens have been removed from the bucket If no tokens have been removed, it
+returns immediately.

Godeps/_workspace/src/github.com/juju/ratelimit/ratelimit.go

@@ -0,0 +1,227 @@
+// Copyright 2014 Canonical Ltd.
+// Licensed under the LGPLv3 with static-linking exception.
+// See LICENCE file for details.
+
+// The ratelimit package provides an efficient token bucket implementation.
+// See http://en.wikipedia.org/wiki/Token_bucket.
+package ratelimit
+
+import (
+	"strconv"
+	"sync"
+	"time"
+)
+
+// Bucket represents a token bucket that fills at a predetermined rate.
+// Methods on Bucket may be called concurrently.
+type Bucket struct {
+	startTime    time.Time
+	capacity     int64
+	quantum      int64
+	fillInterval time.Duration
+
+	// The mutex guards the fields following it.
+	mu sync.Mutex
+
+	// avail holds the number of available tokens
+	// in the bucket, as of availTick ticks from startTime.
+	// It will be negative when there are consumers
+	// waiting for tokens.
+	avail     int64
+	availTick int64
+}
+
+// NewBucket returns a new token bucket that fills at the
+// rate of one token every fillInterval, up to the given
+// maximum capacity. Both arguments must be
+// positive. The bucket is initially full.
+func NewBucket(fillInterval time.Duration, capacity int64) *Bucket {
+	return newBucketWithQuantum(fillInterval, capacity, 1)
+}
+
+// rateMargin specifes the allowed variance of actual
+// rate from specified rate. 1% seems reasonable.
+const rateMargin = 0.01
+
+// NewBucketWithRate returns a token bucket that fills the bucket
+// at the rate of rate tokens per second up to the given
+// maximum capacity. Because of limited clock resolution,
+// at high rates, the actual rate may be up to 1% different from the
+// specified rate.
+func NewBucketWithRate(rate float64, capacity int64) *Bucket {
+	for quantum := int64(1); quantum < 1<<50; quantum = nextQuantum(quantum) {
+		fillInterval := time.Duration(1e9 * float64(quantum) / rate)
+		if fillInterval <= 0 {
+			continue
+		}
+		tb := newBucketWithQuantum(fillInterval, capacity, quantum)
+		if diff := abs(tb.Rate() - rate); diff/rate <= rateMargin {
+			return tb
+		}
+	}
+	panic("cannot find suitable quantum for " + strconv.FormatFloat(rate, 'g', -1, 64))
+}
+
+// nextQuantum returns the next quantum to try after q.
+// We grow the quantum exponentially, but slowly, so we
+// get a good fit in the lower numbers.
+func nextQuantum(q int64) int64 {
+	q1 := q * 11 / 10
+	if q1 == q {
+		q1++
+	}
+	return q1
+}
+
+// newBucketWithQuantum is similar to NewBucket, but allows
+// the specification of the quantum size - quantum tokens
+// are added every fillInterval. This is so that we can get accurate
+// rates even when we want to add more than one token per ns.
+func newBucketWithQuantum(fillInterval time.Duration, capacity, quantum int64) *Bucket {
+	if fillInterval <= 0 {
+		panic("token bucket fill interval is not > 0")
+	}
+	if capacity <= 0 {
+		panic("token bucket capacity is not > 0")
+	}
+	if quantum <= 0 {
+		panic("token bucket quantum is not > 0")
+	}
+	return &Bucket{
+		startTime:    time.Now(),
+		capacity:     capacity,
+		quantum:      quantum,
+		avail:        capacity,
+		fillInterval: fillInterval,
+	}
+}
+
+// Wait takes count tokens from the bucket, waiting until they are
+// available.
+func (tb *Bucket) Wait(count int64) {
+	if d := tb.Take(count); d > 0 {
+		time.Sleep(d)
+	}
+}
+
+// WaitMaxDuration is like Wait except that it will
+// only take tokens from the bucket if it needs to wait
+// for no greater than maxWait. It reports whether
+// any tokens have been removed from the bucket
+// If no tokens have been removed, it returns immediately.
+func (tb *Bucket) WaitMaxDuration(count int64, maxWait time.Duration) bool {
+	d, ok := tb.TakeMaxDuration(count, maxWait)
+	if d > 0 {
+		time.Sleep(d)
+	}
+	return ok
+}
+
+const infinityDuration time.Duration = 0x7fffffffffffffff
+
+// Take takes count tokens from the bucket without blocking. It returns
+// the time that the caller should wait until the tokens are actually
+// available.
+//
+// Note that if the request is irrevocable - there is no way to return
+// tokens to the bucket once this method commits us to taking them.
+func (tb *Bucket) Take(count int64) time.Duration {
+	d, _ := tb.take(time.Now(), count, infinityDuration)
+	return d
+}
+
+// TakeMaxDuration is like Take, except that
+// it will only take tokens from the bucket if the wait
+// time for the tokens is no greater than maxWait.
+//
+// If it would take longer than maxWait for the tokens
+// to become available, it does nothing and reports false,
+// otherwise it returns the time that the caller should
+// wait until the tokens are actually available, and reports
+// true.
+func (tb *Bucket) TakeMaxDuration(count int64, maxWait time.Duration) (time.Duration, bool) {
+	return tb.take(time.Now(), count, maxWait)
+}
+
+// TakeAvailable takes up to count immediately available tokens from the
+// bucket. It returns the number of tokens removed, or zero if there are
+// no available tokens. It does not block.
+func (tb *Bucket) TakeAvailable(count int64) int64 {
+	return tb.takeAvailable(time.Now(), count)
+}
+
+// takeAvailable is the internal version of TakeAvailable - it takes the
+// current time as an argument to enable easy testing.
+func (tb *Bucket) takeAvailable(now time.Time, count int64) int64 {
+	if count <= 0 {
+		return 0
+	}
+	tb.mu.Lock()
+	defer tb.mu.Unlock()
+
+	tb.adjust(now)
+	if tb.avail <= 0 {
+		return 0
+	}
+	if count > tb.avail {
+		count = tb.avail
+	}
+	tb.avail -= count
+	return count
+}
+
+// Rate returns the fill rate of the bucket, in tokens per second.
+func (tb *Bucket) Rate() float64 {
+	return 1e9 * float64(tb.quantum) / float64(tb.fillInterval)
+}
+
+// take is the internal version of Take - it takes the current time as
+// an argument to enable easy testing.
+func (tb *Bucket) take(now time.Time, count int64, maxWait time.Duration) (time.Duration, bool) {
+	if count <= 0 {
+		return 0, true
+	}
+	tb.mu.Lock()
+	defer tb.mu.Unlock()
+
+	currentTick := tb.adjust(now)
+	avail := tb.avail - count
+	if avail >= 0 {
+		tb.avail = avail
+		return 0, true
+	}
+	// Round up the missing tokens to the nearest multiple
+	// of quantum - the tokens won't be available until
+	// that tick.
+	endTick := currentTick + (-avail+tb.quantum-1)/tb.quantum
+	endTime := tb.startTime.Add(time.Duration(endTick) * tb.fillInterval)
+	waitTime := endTime.Sub(now)
+	if waitTime > maxWait {
+		return 0, false
+	}
+	tb.avail = avail
+	return waitTime, true
+}
+
+// adjust adjusts the current bucket capacity based on the current time.
+// It returns the current tick.
+func (tb *Bucket) adjust(now time.Time) (currentTick int64) {
+	currentTick = int64(now.Sub(tb.startTime) / tb.fillInterval)
+
+	if tb.avail >= tb.capacity {
+		return
+	}
+	tb.avail += (currentTick - tb.availTick) * tb.quantum
+	if tb.avail > tb.capacity {
+		tb.avail = tb.capacity
+	}
+	tb.availTick = currentTick
+	return
+}
+
+func abs(f float64) float64 {
+	if f < 0 {
+		return -f
+	}
+	return f
+}

Godeps/_workspace/src/github.com/juju/ratelimit/ratelimit_test.go

@@ -0,0 +1,328 @@
+// Copyright 2014 Canonical Ltd.
+// Licensed under the LGPLv3 with static-linking exception.
+// See LICENCE file for details.
+
+package ratelimit
+
+import (
+	gc "launchpad.net/gocheck"
+
+	"testing"
+	"time"
+)
+
+func TestPackage(t *testing.T) {
+	gc.TestingT(t)
+}
+
+type rateLimitSuite struct{}
+
+var _ = gc.Suite(rateLimitSuite{})
+
+type takeReq struct {
+	time       time.Duration
+	count      int64
+	expectWait time.Duration
+}
+
+var takeTests = []struct {
+	about        string
+	fillInterval time.Duration
+	capacity     int64
+	reqs         []takeReq
+}{{
+	about:        "serial requests",
+	fillInterval: 250 * time.Millisecond,
+	capacity:     10,
+	reqs: []takeReq{{
+		time:       0,
+		count:      0,
+		expectWait: 0,
+	}, {
+		time:       0,
+		count:      10,
+		expectWait: 0,
+	}, {
+		time:       0,
+		count:      1,
+		expectWait: 250 * time.Millisecond,
+	}, {
+		time:       250 * time.Millisecond,
+		count:      1,
+		expectWait: 250 * time.Millisecond,
+	}},
+}, {
+	about:        "concurrent requests",
+	fillInterval: 250 * time.Millisecond,
+	capacity:     10,
+	reqs: []takeReq{{
+		time:       0,
+		count:      10,
+		expectWait: 0,
+	}, {
+		time:       0,
+		count:      2,
+		expectWait: 500 * time.Millisecond,
+	}, {
+		time:       0,
+		count:      2,
+		expectWait: 1000 * time.Millisecond,
+	}, {
+		time:       0,
+		count:      1,
+		expectWait: 1250 * time.Millisecond,
+	}},
+}, {
+	about:        "more than capacity",
+	fillInterval: 1 * time.Millisecond,
+	capacity:     10,
+	reqs: []takeReq{{
+		time:       0,
+		count:      10,
+		expectWait: 0,
+	}, {
+		time:       20 * time.Millisecond,
+		count:      15,
+		expectWait: 5 * time.Millisecond,
+	}},
+}, {
+	about:        "sub-quantum time",
+	fillInterval: 10 * time.Millisecond,
+	capacity:     10,
+	reqs: []takeReq{{
+		time:       0,
+		count:      10,
+		expectWait: 0,
+	}, {
+		time:       7 * time.Millisecond,
+		count:      1,
+		expectWait: 3 * time.Millisecond,
+	}, {
+		time:       8 * time.Millisecond,
+		count:      1,
+		expectWait: 12 * time.Millisecond,
+	}},
+}, {
+	about:        "within capacity",
+	fillInterval: 10 * time.Millisecond,
+	capacity:     5,
+	reqs: []takeReq{{
+		time:       0,
+		count:      5,
+		expectWait: 0,
+	}, {
+		time:       60 * time.Millisecond,
+		count:      5,
+		expectWait: 0,
+	}, {
+		time:       60 * time.Millisecond,
+		count:      1,
+		expectWait: 10 * time.Millisecond,
+	}, {
+		time:       80 * time.Millisecond,
+		count:      2,
+		expectWait: 10 * time.Millisecond,
+	}},
+}}
+
+func (rateLimitSuite) TestTake(c *gc.C) {
+	for i, test := range takeTests {
+		tb := NewBucket(test.fillInterval, test.capacity)
+		for j, req := range test.reqs {
+			d, ok := tb.take(tb.startTime.Add(req.time), req.count, infinityDuration)
+			c.Assert(ok, gc.Equals, true)
+			if d != req.expectWait {
+				c.Fatalf("test %d.%d, %s, got %v want %v", i, j, test.about, d, req.expectWait)
+			}
+		}
+	}
+}
+
+func (rateLimitSuite) TestTakeMaxDuration(c *gc.C) {
+	for i, test := range takeTests {
+		tb := NewBucket(test.fillInterval, test.capacity)
+		for j, req := range test.reqs {
+			if req.expectWait > 0 {
+				d, ok := tb.take(tb.startTime.Add(req.time), req.count, req.expectWait-1)
+				c.Assert(ok, gc.Equals, false)
+				c.Assert(d, gc.Equals, time.Duration(0))
+			}
+			d, ok := tb.take(tb.startTime.Add(req.time), req.count, req.expectWait)
+			c.Assert(ok, gc.Equals, true)
+			if d != req.expectWait {
+				c.Fatalf("test %d.%d, %s, got %v want %v", i, j, test.about, d, req.expectWait)
+			}
+		}
+	}
+}
+
+type takeAvailableReq struct {
+	time   time.Duration
+	count  int64
+	expect int64
+}
+
+var takeAvailableTests = []struct {
+	about        string
+	fillInterval time.Duration
+	capacity     int64
+	reqs         []takeAvailableReq
+}{{
+	about:        "serial requests",
+	fillInterval: 250 * time.Millisecond,
+	capacity:     10,
+	reqs: []takeAvailableReq{{
+		time:   0,
+		count:  0,
+		expect: 0,
+	}, {
+		time:   0,
+		count:  10,
+		expect: 10,
+	}, {
+		time:   0,
+		count:  1,
+		expect: 0,
+	}, {
+		time:   250 * time.Millisecond,
+		count:  1,
+		expect: 1,
+	}},
+}, {
+	about:        "concurrent requests",
+	fillInterval: 250 * time.Millisecond,
+	capacity:     10,
+	reqs: []takeAvailableReq{{
+		time:   0,
+		count:  5,
+		expect: 5,
+	}, {
+		time:   0,
+		count:  2,
+		expect: 2,
+	}, {
+		time:   0,
+		count:  5,
+		expect: 3,
+	}, {
+		time:   0,
+		count:  1,
+		expect: 0,
+	}},
+}, {
+	about:        "more than capacity",
+	fillInterval: 1 * time.Millisecond,
+	capacity:     10,
+	reqs: []takeAvailableReq{{
+		time:   0,
+		count:  10,
+		expect: 10,
+	}, {
+		time:   20 * time.Millisecond,
+		count:  15,
+		expect: 10,
+	}},
+}, {
+	about:        "within capacity",
+	fillInterval: 10 * time.Millisecond,
+	capacity:     5,
+	reqs: []takeAvailableReq{{
+		time:   0,
+		count:  5,
+		expect: 5,
+	}, {
+		time:   60 * time.Millisecond,
+		count:  5,
+		expect: 5,
+	}, {
+		time:   70 * time.Millisecond,
+		count:  1,
+		expect: 1,
+	}},
+}}
+
+func (rateLimitSuite) TestTakeAvailable(c *gc.C) {
+	for i, test := range takeAvailableTests {
+		tb := NewBucket(test.fillInterval, test.capacity)
+		for j, req := range test.reqs {
+			d := tb.takeAvailable(tb.startTime.Add(req.time), req.count)
+			if d != req.expect {
+				c.Fatalf("test %d.%d, %s, got %v want %v", i, j, test.about, d, req.expect)
+			}
+		}
+	}
+}
+
+func (rateLimitSuite) TestPanics(c *gc.C) {
+	c.Assert(func() { NewBucket(0, 1) }, gc.PanicMatches, "token bucket fill interval is not > 0")
+	c.Assert(func() { NewBucket(-2, 1) }, gc.PanicMatches, "token bucket fill interval is not > 0")
+	c.Assert(func() { NewBucket(1, 0) }, gc.PanicMatches, "token bucket capacity is not > 0")
+	c.Assert(func() { NewBucket(1, -2) }, gc.PanicMatches, "token bucket capacity is not > 0")
+}
+
+func isCloseTo(x, y, tolerance float64) bool {
+	return abs(x-y)/y < tolerance
+}
+
+func (rateLimitSuite) TestRate(c *gc.C) {
+	tb := NewBucket(1, 1)
+	if !isCloseTo(tb.Rate(), 1e9, 0.00001) {
+		c.Fatalf("got %v want 1e9", tb.Rate())
+	}
+	tb = NewBucket(2*time.Second, 1)
+	if !isCloseTo(tb.Rate(), 0.5, 0.00001) {
+		c.Fatalf("got %v want 0.5", tb.Rate())
+	}
+	tb = newBucketWithQuantum(100*time.Millisecond, 1, 5)
+	if !isCloseTo(tb.Rate(), 50, 0.00001) {
+		c.Fatalf("got %v want 50", tb.Rate())
+	}
+}
+
+func checkRate(c *gc.C, rate float64) {
+	tb := NewBucketWithRate(rate, 1<<62)
+	if !isCloseTo(tb.Rate(), rate, rateMargin) {
+		c.Fatalf("got %g want %v", tb.Rate(), rate)
+	}
+	d, ok := tb.take(tb.startTime, 1<<62, infinityDuration)
+	c.Assert(ok, gc.Equals, true)
+	c.Assert(d, gc.Equals, time.Duration(0))
+
+	// Check that the actual rate is as expected by
+	// asking for a not-quite multiple of the bucket's
+	// quantum and checking that the wait time
+	// correct.
+	d, ok = tb.take(tb.startTime, tb.quantum*2-tb.quantum/2, infinityDuration)
+	c.Assert(ok, gc.Equals, true)
+	expectTime := 1e9 * float64(tb.quantum) * 2 / rate
+	if !isCloseTo(float64(d), expectTime, rateMargin) {
+		c.Fatalf("rate %g: got %g want %v", rate, float64(d), expectTime)
+	}
+}
+
+func (rateLimitSuite) TestNewWithRate(c *gc.C) {
+	for rate := float64(1); rate < 1e6; rate += 7 {
+		checkRate(c, rate)
+	}
+	for _, rate := range []float64{
+		1024 * 1024 * 1024,
+		1e-5,
+		0.9e-5,
+		0.5,
+		0.9,
+		0.9e8,
+		3e12,
+		4e18,
+	} {
+		checkRate(c, rate)
+		checkRate(c, rate/3)
+		checkRate(c, rate*1.3)
+	}
+}
+
+func BenchmarkWait(b *testing.B) {
+	tb := NewBucket(1, 16*1024)
+	for i := b.N - 1; i >= 0; i-- {
+		tb.Wait(1)
+	}
+}

Godeps/_workspace/src/github.com/juju/ratelimit/reader.go

@@ -0,0 +1,51 @@
+// Copyright 2014 Canonical Ltd.
+// Licensed under the LGPLv3 with static-linking exception.
+// See LICENCE file for details.
+
+package ratelimit
+
+import "io"
+
+type reader struct {
+	r      io.Reader
+	bucket *Bucket
+}
+
+// Reader returns a reader that is rate limited by
+// the given token bucket. Each token in the bucket
+// represents one byte.
+func Reader(r io.Reader, bucket *Bucket) io.Reader {
+	return &reader{
+		r:      r,
+		bucket: bucket,
+	}
+}
+
+func (r *reader) Read(buf []byte) (int, error) {
+	n, err := r.r.Read(buf)
+	if n <= 0 {
+		return n, err
+	}
+	r.bucket.Wait(int64(n))
+	return n, err
+}
+
+type writer struct {
+	w      io.Writer
+	bucket *Bucket
+}
+
+// Writer returns a reader that is rate limited by
+// the given token bucket. Each token in the bucket
+// represents one byte.
+func Writer(w io.Writer, bucket *Bucket) io.Writer {
+	return &writer{
+		w:      w,
+		bucket: bucket,
+	}
+}
+
+func (w *writer) Write(buf []byte) (int, error) {
+	w.bucket.Wait(int64(len(buf)))
+	return w.w.Write(buf)
+}

build.sh

@@ -86,8 +86,7 @@ case "$1" in
 		prepare
 		test || exit 1
 
-		export GOARM=7
-		for os in darwin-amd64 linux-amd64 linux-arm freebsd-amd64 windows-amd64 ; do
+		for os in darwin-amd64 linux-386 linux-amd64 freebsd-amd64 windows-amd64 ; do
 			export GOOS=${os%-*}
 			export GOARCH=${os#*-}
 
@@ -105,6 +104,18 @@ case "$1" in
 					;;
 			esac
 		done
+
+		export GOOS=linux
+		export GOARCH=arm
+
+		export GOARM=7
+		build
+		tarDist "syncthing-linux-armv7-$version"
+
+		export GOARM=6
+		build
+		tarDist "syncthing-linux-armv6-$version"
+
 		;;
 
 	upload)

cmd/syncthing/gui.go

@@ -38,6 +38,7 @@ func startGUI(addr string, m *Model) {
 
 	router.Post("/rest/config", restPostConfig)
 	router.Post("/rest/restart", restPostRestart)
+	router.Post("/rest/reset", restPostReset)
 	router.Post("/rest/error", restPostError)
 
 	go func() {
@@ -112,7 +113,12 @@ func restGetConfigInSync(w http.ResponseWriter) {
 }
 
 func restPostRestart(req *http.Request) {
-	restart()
+	go restart()
+}
+
+func restPostReset(req *http.Request) {
+	resetRepositories()
+	go restart()
 }
 
 type guiFile scanner.File

cmd/syncthing/limitedwriter.go

@@ -0,0 +1,19 @@
+package main
+
+import (
+	"io"
+
+	"github.com/juju/ratelimit"
+)
+
+type limitedWriter struct {
+	w      io.Writer
+	bucket *ratelimit.Bucket
+}
+
+func (w *limitedWriter) Write(buf []byte) (int, error) {
+	if w.bucket != nil {
+		w.bucket.Wait(int64(len(buf)))
+	}
+	return w.w.Write(buf)
+}

cmd/syncthing/main.go

@@ -4,6 +4,7 @@ import (
 	"crypto/tls"
 	"flag"
 	"fmt"
+	"io"
 	"log"
 	"net"
 	"net/http"
@@ -19,6 +20,7 @@ import (
 	"github.com/calmh/ini"
 	"github.com/calmh/syncthing/discover"
 	"github.com/calmh/syncthing/protocol"
+	"github.com/juju/ratelimit"
 )
 
 const BlockSize = 128 * 1024
@@ -27,13 +29,9 @@ var cfg Configuration
 var Version = "unknown-dev"
 
 var (
-	myID string
-)
-
-var (
-	showVersion bool
-	confDir     string
-	verbose     bool
+	myID       string
+	confDir    string
+	rateBucket *ratelimit.Bucket
 )
 
 const (
@@ -61,9 +59,11 @@ const (
 )
 
 func main() {
+	var reset bool
+	var showVersion 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(&verbose, "v", false, "Be more verbose")
 	flag.Usage = usageFor(flag.CommandLine, usage, extraUsage)
 	flag.Parse()
 
@@ -73,7 +73,7 @@ func main() {
 	}
 
 	if showVersion {
-		fmt.Println(Version)
+		fmt.Printf("syncthing %s (%s %s-%s)\n", Version, runtime.Version(), runtime.GOOS, runtime.GOARCH)
 		os.Exit(0)
 	}
 
@@ -165,6 +165,11 @@ func main() {
 		infof("Edit %s to taste or use the GUI\n", cfgFile)
 	}
 
+	if reset {
+		resetRepositories()
+		os.Exit(0)
+	}
+
 	if profiler := os.Getenv("STPROFILER"); len(profiler) > 0 {
 		go func() {
 			dlog.Println("Starting profiler on", profiler)
@@ -188,11 +193,15 @@ func main() {
 		MinVersion:             tls.VersionTLS12,
 	}
 
-	m := NewModel(cfg.Options.MaxChangeKbps * 1000)
+	// If the write rate should be limited, set up a rate limiter for it.
+	// This will be used on connections created in the connect and listen routines.
+
 	if cfg.Options.MaxSendKbps > 0 {
-		m.LimitRate(cfg.Options.MaxSendKbps)
+		rateBucket = ratelimit.NewBucketWithRate(float64(1000*cfg.Options.MaxSendKbps), int64(5*1000*cfg.Options.MaxSendKbps))
 	}
 
+	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)
@@ -230,9 +239,7 @@ func main() {
 	// Walk the repository and update the local model before establishing any
 	// connections to other nodes.
 
-	if verbose {
-		infoln("Populating repository index")
-	}
+	infoln("Populating repository index")
 	m.LoadIndexes(confDir)
 	m.ScanRepos()
 	m.SaveIndexes(confDir)
@@ -244,29 +251,41 @@ func main() {
 	}
 
 	// Routine to connect out to configured nodes
-	if verbose {
-		infoln("Attempting to connect to other nodes")
-	}
 	disc := discovery()
 	go listenConnect(myID, disc, m, tlsCfg, connOpts)
 
 	// 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 cfg.Options.ReadOnly {
-		if verbose {
-			okln("Ready to synchronize (read only; no external updates accepted)")
-		}
+		okln("Ready to synchronize (read only; no external updates accepted)")
 		m.StartRO()
 	} else {
-		if verbose {
-			okln("Ready to synchronize (read-write)")
-		}
+		okln("Ready to synchronize (read-write)")
 		m.StartRW(cfg.Options.ParallelRequests)
 	}
 
 	select {}
 }
 
+func resetRepositories() {
+	suffix := fmt.Sprintf(".syncthing-reset-%d", time.Now().UnixNano())
+	for _, repo := range cfg.Repositories {
+		if _, err := os.Stat(repo.Directory); err == nil {
+			infof("Reset: Moving %s -> %s", repo.Directory, repo.Directory+suffix)
+			os.Rename(repo.Directory, repo.Directory+suffix)
+		}
+	}
+
+	pat := filepath.Join(confDir, "*.idx.gz")
+	idxs, err := filepath.Glob(pat)
+	if err == nil {
+		for _, idx := range idxs {
+			infof("Reset: Removing %s", idx)
+			os.Remove(idx)
+		}
+	}
+}
+
 func restart() {
 	infoln("Restarting")
 	if os.Getenv("SMF_FMRI") != "" || os.Getenv("STNORESTART") != "" {
@@ -426,7 +445,11 @@ next:
 
 		for _, nodeCfg := range cfg.Repositories[0].Nodes {
 			if nodeCfg.NodeID == remoteID {
-				protoConn := protocol.NewConnection(remoteID, conn, conn, m, connOpts)
+				var wr io.Writer = conn
+				if rateBucket != nil {
+					wr = &limitedWriter{conn, rateBucket}
+				}
+				protoConn := protocol.NewConnection(remoteID, conn, wr, m, connOpts)
 				m.AddConnection(conn, protoConn)
 				continue next
 			}
@@ -444,7 +467,7 @@ func discovery() *discover.Discoverer {
 
 	if !cfg.Options.GlobalAnnEnabled {
 		cfg.Options.GlobalAnnServer = ""
-	} else if verbose {
+	} else {
 		infoln("Sending external discovery announcements")
 	}
 

cmd/syncthing/model.go

@@ -35,8 +35,6 @@ type Model struct {
 
 	sup suppressor
 
-	limitRequestRate chan struct{}
-
 	addedRepo bool
 	started   bool
 }
@@ -65,21 +63,6 @@ func NewModel(maxChangeBw int) *Model {
 	return m
 }
 
-func (m *Model) LimitRate(kbps int) {
-	m.limitRequestRate = make(chan struct{}, kbps)
-	n := kbps/10 + 1
-	go func() {
-		for {
-			time.Sleep(100 * time.Millisecond)
-			for i := 0; i < n; i++ {
-				select {
-				case m.limitRequestRate <- struct{}{}:
-				}
-			}
-		}
-	}()
-}
-
 // 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.
@@ -244,6 +227,14 @@ func (m *Model) NeedFiles() ([]scanner.File, int64) {
 	return nf, bytes
 }
 
+// NeedFiles returns the list of currently needed files and the total size.
+func (m *Model) NeedFilesRepo(repo string) []scanner.File {
+	m.rmut.RLock()
+	nf := m.repoFiles[repo].Need(cid.LocalID)
+	m.rmut.RUnlock()
+	return nf
+}
+
 // Index is called when a new node is connected and we receive their full index.
 // Implements the protocol.Model interface.
 func (m *Model) Index(nodeID string, repo string, fs []protocol.FileInfo) {
@@ -361,12 +352,6 @@ func (m *Model) Request(nodeID, repo, name string, offset int64, size int) ([]by
 		return nil, err
 	}
 
-	if m.limitRequestRate != nil {
-		for s := 0; s < len(buf); s += 1024 {
-			<-m.limitRequestRate
-		}
-	}
-
 	return buf, nil
 }
 
@@ -389,6 +374,20 @@ func (m *Model) SeedLocal(repo string, fs []protocol.FileInfo) {
 	m.rmut.RUnlock()
 }
 
+func (m *Model) CurrentRepoFile(repo string, file string) scanner.File {
+	m.rmut.RLock()
+	f := m.repoFiles[repo].Get(cid.LocalID, file)
+	m.rmut.RUnlock()
+	return f
+}
+
+func (m *Model) CurrentGlobalFile(repo string, file string) scanner.File {
+	m.rmut.RLock()
+	f := m.repoFiles[repo].GetGlobal(file)
+	m.rmut.RUnlock()
+	return f
+}
+
 type cFiler struct {
 	m *Model
 	r string
@@ -396,10 +395,7 @@ type cFiler struct {
 
 // Implements scanner.CurrentFiler
 func (cf cFiler) CurrentFile(file string) scanner.File {
-	cf.m.rmut.RLock()
-	f := cf.m.repoFiles[cf.r].Get(cid.LocalID, file)
-	cf.m.rmut.RUnlock()
-	return f
+	return cf.m.CurrentRepoFile(cf.r, file)
 }
 
 // ConnectedTo returns true if we are connected to the named node.

cmd/syncthing/puller.go

@@ -119,6 +119,7 @@ func (p *puller) run() {
 
 	walkTicker := time.Tick(time.Duration(cfg.Options.RescanIntervalS) * time.Second)
 	timeout := time.Tick(5 * time.Second)
+	changed := true
 
 	for {
 		// Run the pulling loop as long as there are blocks to fetch
@@ -126,16 +127,15 @@ func (p *puller) run() {
 		for {
 			select {
 			case res := <-p.requestResults:
+				changed = true
 				p.requestSlots <- true
 				p.handleRequestResult(res)
 
 			case b := <-p.blocks:
+				changed = true
 				p.handleBlock(b)
 
 			case <-timeout:
-				if debugPull {
-					dlog.Println("timeout")
-				}
 				if len(p.openFiles) == 0 && p.bq.empty() {
 					// Nothing more to do for the moment
 					break pull
@@ -154,6 +154,11 @@ func (p *puller) run() {
 			}
 		}
 
+		if changed {
+			p.fixupDirectories()
+			changed = false
+		}
+
 		// Do a rescan if it's time for it
 		select {
 		case <-walkTicker:
@@ -181,6 +186,72 @@ func (p *puller) runRO() {
 	}
 }
 
+func (p *puller) fixupDirectories() {
+	var deleteDirs []string
+	fn := func(path string, info os.FileInfo, err error) error {
+		if !info.IsDir() {
+			return nil
+		}
+
+		rn, err := filepath.Rel(p.dir, path)
+		if err != nil {
+			return nil
+		}
+
+		if rn == "." {
+			return nil
+		}
+
+		cur := p.model.CurrentGlobalFile(p.repo, rn)
+		if cur.Name != rn {
+			// No matching dir in current list; weird
+			return nil
+		}
+
+		if cur.Flags&protocol.FlagDeleted != 0 {
+			if debugPull {
+				dlog.Printf("queue delete dir: %v", cur)
+			}
+
+			// We queue the directories to delete since we walk the
+			// tree in depth first order and need to remove the
+			// directories in the opposite order.
+
+			deleteDirs = append(deleteDirs, path)
+			return nil
+		}
+
+		if cur.Flags&uint32(os.ModePerm) != uint32(info.Mode()&os.ModePerm) {
+			os.Chmod(path, os.FileMode(cur.Flags)&os.ModePerm)
+			if debugPull {
+				dlog.Printf("restored dir flags: %o -> %v", info.Mode()&os.ModePerm, cur)
+			}
+		}
+
+		if cur.Modified != info.ModTime().Unix() {
+			t := time.Unix(cur.Modified, 0)
+			os.Chtimes(path, t, t)
+			if debugPull {
+				dlog.Printf("restored dir modtime: %d -> %v", info.ModTime().Unix(), cur)
+			}
+		}
+
+		return nil
+	}
+	filepath.Walk(p.dir, fn)
+
+	// Delete any queued directories
+	for i := len(deleteDirs) - 1; i >= 0; i-- {
+		if debugPull {
+			dlog.Println("delete dir:", deleteDirs[i])
+		}
+		err := os.Remove(deleteDirs[i])
+		if err != nil {
+			warnln(err)
+		}
+	}
+}
+
 func (p *puller) handleRequestResult(res requestResult) {
 	p.oustandingPerNode.decrease(res.node)
 	f := res.file
@@ -251,6 +322,18 @@ func (p *puller) handleRequestResult(res requestResult) {
 func (p *puller) handleBlock(b bqBlock) {
 	f := b.file
 
+	// For directories, simply making sure they exist is enough
+	if f.Flags&protocol.FlagDirectory != 0 {
+		path := filepath.Join(p.dir, f.Name)
+		_, err := os.Stat(path)
+		if err != nil && os.IsNotExist(err) {
+			os.MkdirAll(path, 0777)
+		}
+		p.model.updateLocal(p.repo, f)
+		p.requestSlots <- true
+		return
+	}
+
 	of, ok := p.openFiles[f.Name]
 	of.done = b.last
 
@@ -429,13 +512,13 @@ func (p *puller) handleEmptyBlock(b bqBlock) {
 		Rename(of.temp, of.filepath)
 	}
 	delete(p.openFiles, f.Name)
-	p.model.repoFiles[p.repo].Update(cid.LocalID, []scanner.File{f})
+	p.model.updateLocal(p.repo, f)
 }
 
 func (p *puller) queueNeededBlocks() {
 	queued := 0
-	for _, f := range p.model.repoFiles[p.repo].Need(cid.LocalID) {
-		lf := p.model.repoFiles[p.repo].Get(cid.LocalID, f.Name)
+	for _, f := range p.model.NeedFilesRepo(p.repo) {
+		lf := p.model.CurrentRepoFile(p.repo, f.Name)
 		have, need := scanner.BlockDiff(lf.Blocks, f.Blocks)
 		if debugNeed {
 			dlog.Printf("need:\n  local: %v\n  global: %v\n  haveBlocks: %v\n  needBlocks: %v", lf, f, have, need)

discover/cmd/discosrv/main.go

@@ -11,6 +11,8 @@ import (
 	"time"
 
 	"github.com/calmh/syncthing/discover"
+	"github.com/golang/groupcache/lru"
+	"github.com/juju/ratelimit"
 )
 
 type Node struct {
@@ -28,10 +30,12 @@ var (
 	lock     sync.Mutex
 	queries  = 0
 	answered = 0
+	limited  = 0
+	debug    = false
+	limiter  = lru.New(1024)
 )
 
 func main() {
-	var debug bool
 	var listen string
 	var timestamp bool
 
@@ -48,37 +52,25 @@ func main() {
 	addr, _ := net.ResolveUDPAddr("udp", listen)
 	conn, err := net.ListenUDP("udp", addr)
 	if err != nil {
-		panic(err)
+		log.Fatal(err)
 	}
 
-	go func() {
-		for {
-			time.Sleep(600 * time.Second)
-
-			lock.Lock()
-
-			var deleted = 0
-			for id, node := range nodes {
-				if time.Since(node.Updated) > 60*time.Minute {
-					delete(nodes, id)
-					deleted++
-				}
-			}
-			log.Printf("Expired %d nodes; %d nodes in registry; %d queries (%d answered)", deleted, len(nodes), queries, answered)
-			queries = 0
-			answered = 0
-
-			lock.Unlock()
-		}
-	}()
+	go logStats()
 
 	var buf = make([]byte, 1024)
 	for {
 		buf = buf[:cap(buf)]
 		n, addr, err := conn.ReadFromUDP(buf)
-		if err != nil {
-			panic(err)
+
+		if limit(addr) {
+			// Rate limit in effect for source
+			continue
 		}
+
+		if err != nil {
+			log.Fatal(err)
+		}
+
 		if n < 4 {
 			log.Printf("Received short packet (%d bytes)", n)
 			continue
@@ -89,149 +81,219 @@ func main() {
 
 		switch magic {
 		case discover.AnnouncementMagicV1:
-			var pkt discover.AnnounceV1
-			err := pkt.UnmarshalXDR(buf)
-			if err != nil {
-				log.Println("AnnounceV1 Unmarshal:", err)
-				log.Println(hex.Dump(buf))
-				continue
-			}
-			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()
+			handleAnnounceV1(addr, buf)
 
 		case discover.QueryMagicV1:
-			var pkt discover.QueryV1
-			err := pkt.UnmarshalXDR(buf)
-			if err != nil {
-				log.Println("QueryV1 Unmarshal:", err)
-				log.Println(hex.Dump(buf))
-				continue
-			}
-			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()
-			}
+			handleQueryV1(conn, addr, buf)
 
 		case discover.AnnouncementMagicV2:
-			var pkt discover.AnnounceV2
-			err := pkt.UnmarshalXDR(buf)
-			if err != nil {
-				log.Println("AnnounceV2 Unmarshal:", err)
-				log.Println(hex.Dump(buf))
-				continue
-			}
-			if debug {
-				log.Printf("<- %v %#v", addr, pkt)
-			}
-
-			ip := addr.IP.To4()
-			if ip == nil {
-				ip = addr.IP.To16()
-			}
-
-			var addrs []Address
-			for _, addr := range pkt.Addresses {
-				tip := addr.IP
-				if len(tip) == 0 {
-					tip = ip
-				}
-				addrs = append(addrs, Address{
-					IP:   tip,
-					Port: addr.Port,
-				})
-			}
-
-			node := Node{
-				Addresses: addrs,
-				Updated:   time.Now(),
-			}
-
-			lock.Lock()
-			nodes[pkt.NodeID] = node
-			lock.Unlock()
+			handleAnnounceV2(addr, buf)
 
 		case discover.QueryMagicV2:
-			var pkt discover.QueryV2
-			err := pkt.UnmarshalXDR(buf)
-			if err != nil {
-				log.Println("QueryV2 Unmarshal:", err)
-				log.Println(hex.Dump(buf))
-				continue
-			}
-			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.AnnounceV2{
-					Magic:  discover.AnnouncementMagicV2,
-					NodeID: pkt.NodeID,
-				}
-				for _, addr := range node.Addresses {
-					pkt.Addresses = append(pkt.Addresses, discover.Address{IP: addr.IP, Port: addr.Port})
-				}
-				if debug {
-					log.Printf("-> %v %#v", addr, pkt)
-				}
-
-				tb := pkt.MarshalXDR()
-				_, _, err = conn.WriteMsgUDP(tb, nil, addr)
-				if err != nil {
-					log.Println("QueryV2 response write:", err)
-				}
-
-				lock.Lock()
-				answered++
-				lock.Unlock()
-			}
+			handleQueryV2(conn, addr, buf)
 		}
 	}
 }
+
+func limit(addr *net.UDPAddr) bool {
+	key := addr.IP.String()
+
+	lock.Lock()
+	defer lock.Unlock()
+
+	bkt, ok := limiter.Get(key)
+	if ok {
+		bkt := bkt.(*ratelimit.Bucket)
+		if bkt.TakeAvailable(1) != 1 {
+			// Rate limit exceeded; ignore packet
+			if debug {
+				log.Printf("Rate limit exceeded for", key)
+			}
+			limited++
+			return true
+		} else if debug {
+			log.Printf("Rate limit OK for", key)
+		}
+	} else {
+		if debug {
+			log.Printf("New limiter for", key)
+		}
+		// One packet per ten seconds average rate, burst ten packets
+		limiter.Add(key, ratelimit.NewBucket(10*time.Second, 10))
+	}
+
+	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)
+	if err != nil {
+		log.Println("AnnounceV2 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()
+	}
+
+	var addrs []Address
+	for _, addr := range pkt.Addresses {
+		tip := addr.IP
+		if len(tip) == 0 {
+			tip = ip
+		}
+		addrs = append(addrs, Address{
+			IP:   tip,
+			Port: addr.Port,
+		})
+	}
+
+	node := Node{
+		Addresses: addrs,
+		Updated:   time.Now(),
+	}
+
+	lock.Lock()
+	nodes[pkt.NodeID] = node
+	lock.Unlock()
+}
+
+func handleQueryV2(conn *net.UDPConn, addr *net.UDPAddr, buf []byte) {
+	var pkt discover.QueryV2
+	err := pkt.UnmarshalXDR(buf)
+	if err != nil {
+		log.Println("QueryV2 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.AnnounceV2{
+			Magic:  discover.AnnouncementMagicV2,
+			NodeID: pkt.NodeID,
+		}
+		for _, addr := range node.Addresses {
+			pkt.Addresses = append(pkt.Addresses, discover.Address{IP: addr.IP, Port: addr.Port})
+		}
+		if debug {
+			log.Printf("-> %v %#v", addr, pkt)
+		}
+
+		tb := pkt.MarshalXDR()
+		_, _, err = conn.WriteMsgUDP(tb, nil, addr)
+		if err != nil {
+			log.Println("QueryV2 response write:", err)
+		}
+
+		lock.Lock()
+		answered++
+		lock.Unlock()
+	}
+}
+
+func logStats() {
+	for {
+		time.Sleep(600 * time.Second)
+
+		lock.Lock()
+
+		var deleted = 0
+		for id, node := range nodes {
+			if time.Since(node.Updated) > 60*time.Minute {
+				delete(nodes, id)
+				deleted++
+			}
+		}
+
+		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())
+		queries = 0
+		answered = 0
+		limited = 0
+
+		lock.Unlock()
+	}
+}

files/set.go

@@ -112,7 +112,7 @@ func (m *Set) ReplaceWithDelete(id uint, fs []scanner.File) {
 			if _, ok := nf[ck.Name]; !ok {
 				cf := m.files[ck].File
 				if cf.Flags&protocol.FlagDeleted != protocol.FlagDeleted {
-					cf.Flags = protocol.FlagDeleted
+					cf.Flags |= protocol.FlagDeleted
 					cf.Blocks = nil
 					cf.Size = 0
 					cf.Version = lamport.Default.Tick(cf.Version)
@@ -145,9 +145,13 @@ func (m *Set) Need(id uint) []scanner.File {
 	}
 	var fs []scanner.File
 	m.Lock()
+	rkID := m.remoteKey[id]
 	for name, gk := range m.globalKey {
-		if gk.newerThan(m.remoteKey[id][name]) {
-			fs = append(fs, m.files[gk].File)
+		if gk.newerThan(rkID[name]) {
+			if m.files[gk].File.Flags&protocol.FlagDirectory == 0 || // Regular file
+				m.files[gk].File.Flags&(protocol.FlagDirectory|protocol.FlagDeleted) == protocol.FlagDirectory { // Non-deleted directory
+				fs = append(fs, m.files[gk].File)
+			}
 		}
 	}
 	m.Unlock()

files/set_test.go

@@ -77,6 +77,7 @@ func TestLocalDeleted(t *testing.T) {
 		scanner.File{Name: "b", Version: 1000},
 		scanner.File{Name: "c", Version: 1000},
 		scanner.File{Name: "d", Version: 1000},
+		scanner.File{Name: "z", Version: 1000, Flags: protocol.FlagDirectory},
 	}
 
 	m.ReplaceWithDelete(cid.LocalID, local1)
@@ -91,6 +92,7 @@ func TestLocalDeleted(t *testing.T) {
 		scanner.File{Name: "b", Version: 1001, Flags: protocol.FlagDeleted},
 		local1[2],
 		scanner.File{Name: "d", Version: 1002, Flags: protocol.FlagDeleted},
+		scanner.File{Name: "z", Version: 1003, Flags: protocol.FlagDeleted | protocol.FlagDirectory},
 	}
 
 	m.ReplaceWithDelete(cid.LocalID, local2)
@@ -109,8 +111,9 @@ func TestLocalDeleted(t *testing.T) {
 	expectedGlobal2 := []scanner.File{
 		local1[0],
 		scanner.File{Name: "b", Version: 1001, Flags: protocol.FlagDeleted},
-		scanner.File{Name: "c", Version: 1003, Flags: protocol.FlagDeleted},
+		scanner.File{Name: "c", Version: 1004, Flags: protocol.FlagDeleted},
 		scanner.File{Name: "d", Version: 1002, Flags: protocol.FlagDeleted},
+		scanner.File{Name: "z", Version: 1003, Flags: protocol.FlagDeleted | protocol.FlagDirectory},
 	}
 
 	m.ReplaceWithDelete(cid.LocalID, local3)

gui/index.html

@@ -205,6 +205,7 @@ thead tr th {
     <div class="container">
         <p class="navbar-text">{{version}}</p>
         <ul class="nav navbar-nav navbar-right">
+            <li><a class="navbar-link" href="http://discourse.syncthing.net/">Support / Forum</a></li>
             <li><a class="navbar-link" href="https://github.com/calmh/syncthing/releases">Latest Release</a></li>
             <li><a class="navbar-link" href="https://github.com/calmh/syncthing/wiki">Documentation</a></li>
             <li><a class="navbar-link" href="https://github.com/calmh/syncthing/issues">Bugs</a></li>

protocol/protocol.go

@@ -27,8 +27,9 @@ const (
 )
 
 const (
-	FlagDeleted uint32 = 1 << 12
-	FlagInvalid        = 1 << 13
+	FlagDeleted   uint32 = 1 << 12
+	FlagInvalid          = 1 << 13
+	FlagDirectory        = 1 << 14
 )
 
 var (

scanner/walk.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/calmh/syncthing/lamport"
+	"github.com/calmh/syncthing/protocol"
 )
 
 type Walker struct {
@@ -137,7 +138,6 @@ func (w *Walker) loadIgnoreFiles(dir string, ign map[string][]string) filepath.W
 
 func (w *Walker) walkAndHashFiles(res *[]File, ign map[string][]string) filepath.WalkFunc {
 	return func(p string, info os.FileInfo, err error) error {
-
 		if err != nil {
 			if debug {
 				dlog.Println("error:", p, info, err)
@@ -153,7 +153,12 @@ func (w *Walker) walkAndHashFiles(res *[]File, ign map[string][]string) filepath
 			return nil
 		}
 
+		if rn == "." {
+			return nil
+		}
+
 		if w.TempNamer != nil && w.TempNamer.IsTemporary(rn) {
+			// A temporary file
 			if debug {
 				dlog.Println("temporary:", rn)
 			}
@@ -161,13 +166,15 @@ func (w *Walker) walkAndHashFiles(res *[]File, ign map[string][]string) filepath
 		}
 
 		if _, sn := filepath.Split(rn); sn == w.IgnoreFile {
+			// An ignore-file; these are ignored themselves
 			if debug {
 				dlog.Println("ignorefile:", rn)
 			}
 			return nil
 		}
 
-		if rn != "." && w.ignoreFile(ign, rn) {
+		if w.ignoreFile(ign, rn) {
+			// An ignored file
 			if debug {
 				dlog.Println("ignored:", rn)
 			}
@@ -177,10 +184,34 @@ func (w *Walker) walkAndHashFiles(res *[]File, ign map[string][]string) filepath
 			return nil
 		}
 
-		if info.Mode()&os.ModeType == 0 {
+		if info.Mode().IsDir() {
 			if w.CurrentFiler != nil {
 				cf := w.CurrentFiler.CurrentFile(rn)
-				if cf.Modified == info.ModTime().Unix() {
+				if cf.Modified == info.ModTime().Unix() && cf.Flags == uint32(info.Mode()&os.ModePerm|protocol.FlagDirectory) {
+					if debug {
+						dlog.Println("unchanged:", cf)
+					}
+					*res = append(*res, cf)
+				} else {
+					f := File{
+						Name:     rn,
+						Version:  lamport.Default.Tick(0),
+						Flags:    uint32(info.Mode()&os.ModePerm) | protocol.FlagDirectory,
+						Modified: info.ModTime().Unix(),
+					}
+					if debug {
+						dlog.Println("dir:", cf, f)
+					}
+					*res = append(*res, f)
+				}
+				return nil
+			}
+		}
+
+		if info.Mode().IsRegular() {
+			if w.CurrentFiler != nil {
+				cf := w.CurrentFiler.CurrentFile(rn)
+				if cf.Flags&protocol.FlagDeleted == 0 && cf.Modified == info.ModTime().Unix() {
 					if debug {
 						dlog.Println("unchanged:", cf)
 					}