felixfoertsch/syncthing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

lib: Removal global connection registry (#8254)

Browse Source
This commit is contained in:
Simon Frei
2022-04-09 16:04:56 +02:00
committed by GitHub
parent e30898ddb3
commit b947056e62
17 changed files with 118 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/connections/connections_test.go
+4 -2
View File
@@ -21,6 +21,7 @@ import (
"github.com/thejerf/suture/v4"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/nat"
"github.com/syncthing/syncthing/lib/protocol"
@@ -414,7 +415,7 @@ func withConnectionPair(b *testing.B, connUri string, h func(client, server inte
}
natSvc := nat.NewService(deviceId, wcfg)
conns := make(chan internalConn, 1)
listenSvc := lf.New(uri, wcfg, tlsCfg, conns, natSvc)
listenSvc := lf.New(uri, wcfg, tlsCfg, conns, natSvc, registry.New())
supervisor.Add(listenSvc)
var addr *url.URL
@@ -433,7 +434,8 @@ func withConnectionPair(b *testing.B, connUri string, h func(client, server inte
if err != nil {
b.Fatal(err)
}
dialer := df.New(cfg.Options, tlsCfg)
// Purposely using a different registry: Don't want to reuse port between dialer and listener on the same device
dialer := df.New(cfg.Options, tlsCfg, registry.New())
// Relays might take some time to register the device, so dial multiple times
clientConn, err := dialer.Dial(ctx, deviceId, addr)
lib/connections/quic_dial.go
+10 -6
View File
@@ -41,6 +41,7 @@ func init() {
type quicDialer struct {
commonDialer
registry *registry.Registry
}
func (d *quicDialer) Dial(ctx context.Context, _ protocol.DeviceID, uri *url.URL) (internalConn, error) {
@@ -58,7 +59,7 @@ func (d *quicDialer) Dial(ctx context.Context, _ protocol.DeviceID, uri *url.URL
// Given we always pass the connection to quic, it assumes it's a remote connection it never closes it,
// So our wrapper around it needs to close it, but it only needs to close it if it's not the listening connection.
var createdConn net.PacketConn
listenConn := registry.Get(uri.Scheme, packetConnUnspecified)
listenConn := d.registry.Get(uri.Scheme, packetConnUnspecified)
if listenConn != nil {
conn = listenConn.(net.PacketConn)
} else {
@@ -96,7 +97,7 @@ func (d *quicDialer) Dial(ctx context.Context, _ protocol.DeviceID, uri *url.URL
type quicDialerFactory struct{}
func (quicDialerFactory) New(opts config.OptionsConfiguration, tlsCfg *tls.Config) genericDialer {
func (quicDialerFactory) New(opts config.OptionsConfiguration, tlsCfg *tls.Config, registry *registry.Registry) genericDialer {
// So the idea is that we should probably try dialing every 20 seconds.
// However it would still be nice if this was adjustable/proportional to ReconnectIntervalS
// But prevent something silly like 1/3 = 0 etc.
@@ -104,10 +105,13 @@ func (quicDialerFactory) New(opts config.OptionsConfiguration, tlsCfg *tls.Confi
if quicInterval < 10 {
quicInterval = 10
}
return &quicDialer{commonDialer{
reconnectInterval: time.Duration(quicInterval) * time.Second,
tlsCfg: tlsCfg,
}}
return &quicDialer{
commonDialer: commonDialer{
reconnectInterval: time.Duration(quicInterval) * time.Second,
tlsCfg: tlsCfg,
},
registry: registry,
}
}
func (quicDialerFactory) Priority() int {
lib/connections/quic_listen.go
+15 -13
View File
@@ -40,11 +40,12 @@ type quicListener struct {
onAddressesChangedNotifier
uri *url.URL
cfg config.Wrapper
tlsCfg *tls.Config
conns chan internalConn
factory listenerFactory
uri *url.URL
cfg config.Wrapper
tlsCfg *tls.Config
conns chan internalConn
factory listenerFactory
registry *registry.Registry
address *url.URL
laddr net.Addr
@@ -100,8 +101,8 @@ func (t *quicListener) serve(ctx context.Context) error {
go svc.Serve(ctx)
registry.Register(t.uri.Scheme, conn)
defer registry.Unregister(t.uri.Scheme, conn)
t.registry.Register(t.uri.Scheme, conn)
defer t.registry.Unregister(t.uri.Scheme, conn)
listener, err := quic.Listen(conn, t.tlsCfg, quicConfig)
if err != nil {
@@ -217,13 +218,14 @@ func (f *quicListenerFactory) Valid(config.Configuration) error {
return nil
}
func (f *quicListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service) genericListener {
func (f *quicListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service, registry *registry.Registry) genericListener {
l := &quicListener{
uri: fixupPort(uri, config.DefaultQUICPort),
cfg: cfg,
tlsCfg: tlsCfg,
conns: conns,
factory: f,
uri: fixupPort(uri, config.DefaultQUICPort),
cfg: cfg,
tlsCfg: tlsCfg,
conns: conns,
factory: f,
registry: registry,
}
l.ServiceWithError = svcutil.AsService(l.serve, l.String())
l.nat.Store(stun.NATUnknown)
lib/connections/registry/registry.go
-16
View File
@@ -15,10 +15,6 @@ import (
"github.com/syncthing/syncthing/lib/sync"
)
var (
Default = New()
)
type Registry struct {
mut sync.Mutex
available map[string][]interface{}
@@ -85,15 +81,3 @@ func (r *Registry) Get(scheme string, preferred func(interface{}) bool) interfac
}
return best
}
func Register(scheme string, item interface{}) {
Default.Register(scheme, item)
}
func Unregister(scheme string, item interface{}) {
Default.Unregister(scheme, item)
}
func Get(scheme string, preferred func(interface{}) bool) interface{} {
return Default.Get(scheme, preferred)
}
lib/connections/relay_dial.go
+2 -1
View File
@@ -13,6 +13,7 @@ import (
"time"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/dialer"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/relay/client"
@@ -68,7 +69,7 @@ func (d *relayDialer) Dial(ctx context.Context, id protocol.DeviceID, uri *url.U
type relayDialerFactory struct{}
func (relayDialerFactory) New(opts config.OptionsConfiguration, tlsCfg *tls.Config) genericDialer {
func (relayDialerFactory) New(opts config.OptionsConfiguration, tlsCfg *tls.Config, _ *registry.Registry) genericDialer {
return &relayDialer{commonDialer{
trafficClass: opts.TrafficClass,
reconnectInterval: time.Duration(opts.RelayReconnectIntervalM) * time.Minute,
lib/connections/relay_listen.go
+2 -1
View File
@@ -16,6 +16,7 @@ import (
"github.com/pkg/errors"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/dialer"
"github.com/syncthing/syncthing/lib/nat"
"github.com/syncthing/syncthing/lib/relay/client"
@@ -177,7 +178,7 @@ func (t *relayListener) NATType() string {
type relayListenerFactory struct{}
func (f *relayListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service) genericListener {
func (f *relayListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service, _ *registry.Registry) genericListener {
t := &relayListener{
uri: uri,
cfg: cfg,
lib/connections/service.go
+6 -3
View File
@@ -23,6 +23,7 @@ import (
"time"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/discover"
"github.com/syncthing/syncthing/lib/events"
"github.com/syncthing/syncthing/lib/nat"
@@ -160,6 +161,7 @@ type service struct {
limiter *limiter
natService *nat.Service
evLogger events.Logger
registry *registry.Registry
dialNow chan struct{}
dialNowDevices map[protocol.DeviceID]struct{}
@@ -170,7 +172,7 @@ type service struct {
listenerTokens map[string]suture.ServiceToken
}
func NewService(cfg config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *tls.Config, discoverer discover.Finder, bepProtocolName string, tlsDefaultCommonName string, evLogger events.Logger) Service {
func NewService(cfg config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *tls.Config, discoverer discover.Finder, bepProtocolName string, tlsDefaultCommonName string, evLogger events.Logger, registry *registry.Registry) Service {
spec := svcutil.SpecWithInfoLogger(l)
service := &service{
Supervisor: suture.New("connections.Service", spec),
@@ -187,6 +189,7 @@ func NewService(cfg config.Wrapper, myID protocol.DeviceID, mdl Model, tlsCfg *t
limiter: newLimiter(myID, cfg),
natService: nat.NewService(myID, cfg),
evLogger: evLogger,
registry: registry,
dialNowDevicesMut: sync.NewMutex(),
dialNow: make(chan struct{}, 1),
@@ -655,7 +658,7 @@ func (s *service) resolveDialTargets(ctx context.Context, now time.Time, cfg con
continue
}
dialer := dialerFactory.New(s.cfg.Options(), s.tlsCfg)
dialer := dialerFactory.New(s.cfg.Options(), s.tlsCfg, s.registry)
nextDialAt.set(deviceID, addr, now.Add(dialer.RedialFrequency()))
// For LAN addresses, increase the priority so that we
@@ -755,7 +758,7 @@ func (s *service) createListener(factory listenerFactory, uri *url.URL) bool {
l.Debugln("Starting listener", uri)
listener := factory.New(uri, s.cfg, s.tlsCfg, s.conns, s.natService)
listener := factory.New(uri, s.cfg, s.tlsCfg, s.conns, s.natService, s.registry)
listener.OnAddressesChanged(s.logListenAddressesChangedEvent)
// Retrying a listener many times in rapid succession is unlikely to help,
lib/connections/structs.go
+3 -2
View File
@@ -16,6 +16,7 @@ import (
"time"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/nat"
"github.com/syncthing/syncthing/lib/protocol"
"github.com/syncthing/syncthing/lib/stats"
@@ -139,7 +140,7 @@ func (c internalConn) String() string {
}
type dialerFactory interface {
New(config.OptionsConfiguration, *tls.Config) genericDialer
New(config.OptionsConfiguration, *tls.Config, *registry.Registry) genericDialer
Priority() int
AlwaysWAN() bool
Valid(config.Configuration) error
@@ -162,7 +163,7 @@ type genericDialer interface {
}
type listenerFactory interface {
New(*url.URL, config.Wrapper, *tls.Config, chan internalConn, *nat.Service) genericListener
New(*url.URL, config.Wrapper, *tls.Config, chan internalConn, *nat.Service, *registry.Registry) genericListener
Valid(config.Configuration) error
}
lib/connections/tcp_dial.go
+12 -7
View File
@@ -13,6 +13,7 @@ import (
"time"
"github.com/syncthing/syncthing/lib/config"
"github.com/syncthing/syncthing/lib/connections/registry"
"github.com/syncthing/syncthing/lib/dialer"
"github.com/syncthing/syncthing/lib/protocol"
)
@@ -28,6 +29,7 @@ func init() {
type tcpDialer struct {
commonDialer
registry *registry.Registry
}
func (d *tcpDialer) Dial(ctx context.Context, _ protocol.DeviceID, uri *url.URL) (internalConn, error) {
@@ -35,7 +37,7 @@ func (d *tcpDialer) Dial(ctx context.Context, _ protocol.DeviceID, uri *url.URL)
timeoutCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
conn, err := dialer.DialContextReusePort(timeoutCtx, uri.Scheme, uri.Host)
conn, err := dialer.DialContextReusePortFunc(d.registry)(timeoutCtx, uri.Scheme, uri.Host)
if err != nil {
return internalConn{}, err
}
@@ -62,12 +64,15 @@ func (d *tcpDialer) Dial(ctx context.Context, _ protocol.DeviceID, uri *url.URL)
type tcpDialerFactory struct{}
func (tcpDialerFactory) New(opts config.OptionsConfiguration, tlsCfg *tls.Config) genericDialer {
return &tcpDialer{commonDialer{
trafficClass: opts.TrafficClass,
reconnectInterval: time.Duration(opts.ReconnectIntervalS) * time.Second,
tlsCfg: tlsCfg,
}}
func (tcpDialerFactory) New(opts config.OptionsConfiguration, tlsCfg *tls.Config, registry *registry.Registry) genericDialer {
return &tcpDialer{
commonDialer: commonDialer{
trafficClass: opts.TrafficClass,
reconnectInterval: time.Duration(opts.ReconnectIntervalS) * time.Second,
tlsCfg: tlsCfg,
},
registry: registry,
}
}
func (tcpDialerFactory) Priority() int {
lib/connections/tcp_listen.go
+10 -8
View File
@@ -32,11 +32,12 @@ type tcpListener struct {
svcutil.ServiceWithError
onAddressesChangedNotifier
uri *url.URL
cfg config.Wrapper
tlsCfg *tls.Config
conns chan internalConn
factory listenerFactory
uri *url.URL
cfg config.Wrapper
tlsCfg *tls.Config
conns chan internalConn
factory listenerFactory
registry *registry.Registry
natService *nat.Service
mapping *nat.Mapping
@@ -69,8 +70,8 @@ func (t *tcpListener) serve(ctx context.Context) error {
t.notifyAddressesChanged(t)
defer t.clearAddresses(t)
registry.Register(t.uri.Scheme, tcaddr)
defer registry.Unregister(t.uri.Scheme, tcaddr)
t.registry.Register(t.uri.Scheme, tcaddr)
defer t.registry.Unregister(t.uri.Scheme, tcaddr)
l.Infof("TCP listener (%v) starting", tcaddr)
defer l.Infof("TCP listener (%v) shutting down", tcaddr)
@@ -213,7 +214,7 @@ func (t *tcpListener) NATType() string {
type tcpListenerFactory struct{}
func (f *tcpListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service) genericListener {
func (f *tcpListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.Config, conns chan internalConn, natService *nat.Service, registry *registry.Registry) genericListener {
l := &tcpListener{
uri: fixupPort(uri, config.DefaultTCPPort),
cfg: cfg,
@@ -221,6 +222,7 @@ func (f *tcpListenerFactory) New(uri *url.URL, cfg config.Wrapper, tlsCfg *tls.C
conns: conns,
natService: natService,
factory: f,
registry: registry,
}
l.ServiceWithError = svcutil.AsService(l.serve, l.String())
return l