lib/connections: Allow on the fly changes to rate limits (fixes #3846)

Also replaces github.com/juju/ratelimit with golang.org/x/time/rate as
the latter supports changing the rate on the fly.

GitHub-Pull-Request: https://github.com/syncthing/syncthing/pull/3862

lib/connections/limitedreader.go

@@ -1,33 +0,0 @@
-// Copyright (C) 2014 The Syncthing Authors.
-//
-// This Source Code Form is subject to the terms of the Mozilla Public
-// License, v. 2.0. If a copy of the MPL was not distributed with this file,
-// You can obtain one at http://mozilla.org/MPL/2.0/.
-
-package connections
-
-import (
-	"io"
-
-	"github.com/juju/ratelimit"
-)
-
-type LimitedReader struct {
-	reader io.Reader
-	bucket *ratelimit.Bucket
-}
-
-func NewReadLimiter(r io.Reader, b *ratelimit.Bucket) *LimitedReader {
-	return &LimitedReader{
-		reader: r,
-		bucket: b,
-	}
-}
-
-func (r *LimitedReader) Read(buf []byte) (int, error) {
-	n, err := r.reader.Read(buf)
-	if r.bucket != nil {
-		r.bucket.Wait(int64(n))
-	}
-	return n, err
-}

lib/connections/limitedwriter.go

@@ -1,32 +0,0 @@
-// Copyright (C) 2014 The Syncthing Authors.
-//
-// This Source Code Form is subject to the terms of the Mozilla Public
-// License, v. 2.0. If a copy of the MPL was not distributed with this file,
-// You can obtain one at http://mozilla.org/MPL/2.0/.
-
-package connections
-
-import (
-	"io"
-
-	"github.com/juju/ratelimit"
-)
-
-type LimitedWriter struct {
-	writer io.Writer
-	bucket *ratelimit.Bucket
-}
-
-func NewWriteLimiter(w io.Writer, b *ratelimit.Bucket) *LimitedWriter {
-	return &LimitedWriter{
-		writer: w,
-		bucket: b,
-	}
-}
-
-func (w *LimitedWriter) Write(buf []byte) (int, error) {
-	if w.bucket != nil {
-		w.bucket.Wait(int64(len(buf)))
-	}
-	return w.writer.Write(buf)
-}

lib/connections/limiter.go

@@ -0,0 +1,164 @@
+// Copyright (C) 2017 The Syncthing Authors.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this file,
+// You can obtain one at http://mozilla.org/MPL/2.0/.
+
+package connections
+
+import (
+	"fmt"
+	"io"
+	"sync/atomic"
+
+	"github.com/syncthing/syncthing/lib/config"
+	"golang.org/x/net/context"
+	"golang.org/x/time/rate"
+)
+
+// limiter manages a read and write rate limit, reacting to config changes
+// as appropriate.
+type limiter struct {
+	write     *rate.Limiter
+	read      *rate.Limiter
+	limitsLAN atomicBool
+}
+
+const limiterBurstSize = 4 * 128 << 10
+
+func newLimiter(cfg *config.Wrapper) *limiter {
+	l := &limiter{
+		write: rate.NewLimiter(rate.Inf, limiterBurstSize),
+		read:  rate.NewLimiter(rate.Inf, limiterBurstSize),
+	}
+	cfg.Subscribe(l)
+	prev := config.Configuration{Options: config.OptionsConfiguration{MaxRecvKbps: -1, MaxSendKbps: -1}}
+	l.CommitConfiguration(prev, cfg.RawCopy())
+	return l
+}
+
+func (lim *limiter) newReadLimiter(r io.Reader, isLAN bool) io.Reader {
+	return &limitedReader{reader: r, limiter: lim, isLAN: isLAN}
+}
+
+func (lim *limiter) newWriteLimiter(w io.Writer, isLAN bool) io.Writer {
+	return &limitedWriter{writer: w, limiter: lim, isLAN: isLAN}
+}
+
+func (lim *limiter) VerifyConfiguration(from, to config.Configuration) error {
+	return nil
+}
+
+func (lim *limiter) CommitConfiguration(from, to config.Configuration) bool {
+	if from.Options.MaxRecvKbps == to.Options.MaxRecvKbps &&
+		from.Options.MaxSendKbps == to.Options.MaxSendKbps &&
+		from.Options.LimitBandwidthInLan == to.Options.LimitBandwidthInLan {
+		return true
+	}
+
+	// The rate variables are in KiB/s in the config (despite the camel casing
+	// of the name). We multiply by 1024 to get bytes/s.
+
+	if to.Options.MaxRecvKbps <= 0 {
+		lim.read.SetLimit(rate.Inf)
+	} else {
+		lim.read.SetLimit(1024 * rate.Limit(to.Options.MaxRecvKbps))
+	}
+
+	if to.Options.MaxSendKbps < 0 {
+		lim.write.SetLimit(rate.Inf)
+	} else {
+		lim.write.SetLimit(1024 * rate.Limit(to.Options.MaxSendKbps))
+	}
+
+	lim.limitsLAN.set(to.Options.LimitBandwidthInLan)
+
+	sendLimitStr := "is unlimited"
+	recvLimitStr := "is unlimited"
+	if to.Options.MaxSendKbps > 0 {
+		sendLimitStr = fmt.Sprintf("limit is %d KiB/s", to.Options.MaxSendKbps)
+	}
+	if to.Options.MaxRecvKbps > 0 {
+		recvLimitStr = fmt.Sprintf("limit is %d KiB/s", to.Options.MaxRecvKbps)
+	}
+	l.Infof("Send rate %s, receive rate %s", sendLimitStr, recvLimitStr)
+
+	if to.Options.LimitBandwidthInLan {
+		l.Infoln("Rate limits apply to LAN connections")
+	} else {
+		l.Infoln("Rate limits do not apply to LAN connections")
+	}
+
+	return true
+}
+
+func (lim *limiter) String() string {
+	// required by config.Committer interface
+	return "connections.limiter"
+}
+
+// limitedReader is a rate limited io.Reader
+type limitedReader struct {
+	reader  io.Reader
+	limiter *limiter
+	isLAN   bool
+}
+
+func (r *limitedReader) Read(buf []byte) (int, error) {
+	n, err := r.reader.Read(buf)
+	if !r.isLAN || r.limiter.limitsLAN.get() {
+		take(r.limiter.read, n)
+	}
+	return n, err
+}
+
+// limitedWriter is a rate limited io.Writer
+type limitedWriter struct {
+	writer  io.Writer
+	limiter *limiter
+	isLAN   bool
+}
+
+func (w *limitedWriter) Write(buf []byte) (int, error) {
+	if !w.isLAN || w.limiter.limitsLAN.get() {
+		take(w.limiter.write, len(buf))
+	}
+	return w.writer.Write(buf)
+}
+
+// take is a utility function to consume tokens from a rate.Limiter. No call
+// to WaitN can be larger than the limiter burst size so we split it up into
+// several calls when necessary.
+func take(l *rate.Limiter, tokens int) {
+	if tokens < limiterBurstSize {
+		// This is the by far more common case so we get it out of the way
+		// early.
+		l.WaitN(context.TODO(), tokens)
+		return
+	}
+
+	for tokens > 0 {
+		// Consume limiterBurstSize tokens at a time until we're done.
+		if tokens > limiterBurstSize {
+			l.WaitN(context.TODO(), limiterBurstSize)
+			tokens -= limiterBurstSize
+		} else {
+			l.WaitN(context.TODO(), tokens)
+			tokens = 0
+		}
+	}
+}
+
+type atomicBool int32
+
+func (b *atomicBool) set(v bool) {
+	if v {
+		atomic.StoreInt32((*int32)(b), 1)
+	} else {
+		atomic.StoreInt32((*int32)(b), 0)
+	}
+}
+
+func (b *atomicBool) get() bool {
+	return atomic.LoadInt32((*int32)(b)) != 0
+}

lib/connections/service.go

@@ -10,12 +10,10 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
-	"io"
 	"net"
 	"net/url"
 	"time"
 
-	"github.com/juju/ratelimit"
 	"github.com/syncthing/syncthing/lib/config"
 	"github.com/syncthing/syncthing/lib/discover"
 	"github.com/syncthing/syncthing/lib/events"
@@ -29,6 +27,7 @@ import (
 	_ "github.com/syncthing/syncthing/lib/upnp"
 
 	"github.com/thejerf/suture"
+	"golang.org/x/time/rate"
 )
 
 var (
@@ -37,7 +36,7 @@ var (
 )
 
 const (
-	perDeviceWarningRate = 1.0 / (15 * 60) // Once per 15 minutes
+	perDeviceWarningIntv = 15 * time.Minute
 	tlsHandshakeTimeout  = 10 * time.Second
 )
 
@@ -80,8 +79,7 @@ type Service struct {
 	bepProtocolName      string
 	tlsDefaultCommonName string
 	lans                 []*net.IPNet
-	writeRateLimit       *ratelimit.Bucket
-	readRateLimit        *ratelimit.Bucket
+	limiter              *limiter
 	natService           *nat.Service
 	natServiceToken      *suture.ServiceToken
 
@@ -112,6 +110,7 @@ func NewService(cfg *config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *
 		bepProtocolName:      bepProtocolName,
 		tlsDefaultCommonName: tlsDefaultCommonName,
 		lans:                 lans,
+		limiter:              newLimiter(cfg),
 		natService:           nat.NewService(myID, cfg),
 
 		listenersMut:   sync.NewRWMutex(),
@@ -135,17 +134,6 @@ func NewService(cfg *config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *
 	}
 	cfg.Subscribe(service)
 
-	// The rate variables are in KiB/s in the UI (despite the camel casing
-	// of the name). We multiply by 1024 here to get B/s.
-	options := service.cfg.Options()
-	if options.MaxSendKbps > 0 {
-		service.writeRateLimit = ratelimit.NewBucketWithRate(float64(1024*options.MaxSendKbps), int64(5*1024*options.MaxSendKbps))
-	}
-
-	if options.MaxRecvKbps > 0 {
-		service.readRateLimit = ratelimit.NewBucketWithRate(float64(1024*options.MaxRecvKbps), int64(5*1024*options.MaxRecvKbps))
-	}
-
 	// There are several moving parts here; one routine per listening address
 	// (handled in configuration changing) to handle incoming connections,
 	// one routine to periodically attempt outgoing connections, one routine to
@@ -279,20 +267,12 @@ next:
 			continue next
 		}
 
-		// If rate limiting is set, and based on the address we should
-		// limit the connection, then we wrap it in a limiter.
-
-		limit := s.shouldLimit(c.RemoteAddr())
-
-		wr := io.Writer(c)
-		if limit && s.writeRateLimit != nil {
-			wr = NewWriteLimiter(c, s.writeRateLimit)
-		}
-
-		rd := io.Reader(c)
-		if limit && s.readRateLimit != nil {
-			rd = NewReadLimiter(c, s.readRateLimit)
-		}
+		// Wrap the connection in rate limiters. The limiter itself will
+		// keep up with config changes to the rate and whether or not LAN
+		// connections are limited.
+		isLAN := s.isLAN(c.RemoteAddr())
+		wr := s.limiter.newWriteLimiter(c, isLAN)
+		rd := s.limiter.newReadLimiter(c, isLAN)
 
 		name := fmt.Sprintf("%s-%s (%s)", c.LocalAddr(), c.RemoteAddr(), c.Type())
 		protoConn := protocol.NewConnection(remoteID, rd, wr, s.model, name, deviceCfg.Compression)
@@ -434,21 +414,17 @@ func (s *Service) connect() {
 	}
 }
 
-func (s *Service) shouldLimit(addr net.Addr) bool {
-	if s.cfg.Options().LimitBandwidthInLan {
-		return true
-	}
-
+func (s *Service) isLAN(addr net.Addr) bool {
 	tcpaddr, ok := addr.(*net.TCPAddr)
 	if !ok {
-		return true
+		return false
 	}
 	for _, lan := range s.lans {
 		if lan.Contains(tcpaddr.IP) {
-			return false
+			return true
 		}
 	}
-	return !tcpaddr.IP.IsLoopback()
+	return tcpaddr.IP.IsLoopback()
 }
 
 func (s *Service) createListener(factory listenerFactory, uri *url.URL) bool {
@@ -644,7 +620,7 @@ func urlsToStrings(urls []*url.URL) []string {
 	return strings
 }
 
-var warningLimiters = make(map[protocol.DeviceID]*ratelimit.Bucket)
+var warningLimiters = make(map[protocol.DeviceID]*rate.Limiter)
 var warningLimitersMut = sync.NewMutex()
 
 func warningFor(dev protocol.DeviceID, msg string) {
@@ -652,10 +628,10 @@ func warningFor(dev protocol.DeviceID, msg string) {
 	defer warningLimitersMut.Unlock()
 	lim, ok := warningLimiters[dev]
 	if !ok {
-		lim = ratelimit.NewBucketWithRate(perDeviceWarningRate, 1)
+		lim = rate.NewLimiter(rate.Every(perDeviceWarningIntv), 1)
 		warningLimiters[dev] = lim
 	}
-	if lim.TakeAvailable(1) == 1 {
+	if lim.Allow() {
 		l.Warnln(msg)
 	}
 }