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

all: Use Go 1.21, new QUIC API (#9040)

Browse Source
This commit is contained in:
Jakob Borg
2023-08-21 15:25:52 +02:00
committed by GitHub
parent c40dae315b
commit cbf0e31f69
8 changed files with 139 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/connections/quic_dial.go
+6 -10
View File
@@ -52,27 +52,23 @@ func (d *quicDialer) Dial(ctx context.Context, _ protocol.DeviceID, uri *url.URL
return internalConn{}, err
}
var conn net.PacketConn
// We need to track who created the conn.
// 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.
// If we created the conn we need to close it at the end. If we got a
// Transport from the registry we have no conn to close.
var createdConn net.PacketConn
listenConn := d.registry.Get(uri.Scheme, packetConnUnspecified)
if listenConn != nil {
conn = listenConn.(net.PacketConn)
} else {
transport, _ := d.registry.Get(uri.Scheme, transportConnUnspecified).(*quic.Transport)
if transport == nil {
if packetConn, err := net.ListenPacket("udp", ":0"); err != nil {
return internalConn{}, err
} else {
conn = packetConn
createdConn = packetConn
transport = &quic.Transport{Conn: packetConn}
}
}
ctx, cancel := context.WithTimeout(ctx, quicOperationTimeout)
defer cancel()
session, err := quic.DialContext(ctx, conn, addr, uri.Host, d.tlsCfg, quicConfig)
session, err := transport.Dial(ctx, addr, d.tlsCfg, quicConfig)
if err != nil {
if createdConn != nil {
_ = createdConn.Close()
lib/connections/quic_listen.go
+11 -6
View File
@@ -95,17 +95,22 @@ func (t *quicListener) serve(ctx context.Context) error {
l.Infoln("Listen (BEP/quic):", err)
return err
}
defer func() { _ = udpConn.Close() }()
defer udpConn.Close()
svc, conn := stun.New(t.cfg, t, udpConn)
defer conn.Close()
tracer := &writeTrackingTracer{}
quicTransport := &quic.Transport{
Conn: udpConn,
Tracer: tracer,
}
defer quicTransport.Close()
svc := stun.New(t.cfg, t, &transportPacketConn{tran: quicTransport}, tracer)
go svc.Serve(ctx)
t.registry.Register(t.uri.Scheme, conn)
defer t.registry.Unregister(t.uri.Scheme, conn)
t.registry.Register(t.uri.Scheme, quicTransport)
defer t.registry.Unregister(t.uri.Scheme, quicTransport)
listener, err := quic.Listen(conn, t.tlsCfg, quicConfig)
listener, err := quicTransport.Listen(t.tlsCfg, quicConfig)
if err != nil {
l.Infoln("Listen (BEP/quic):", err)
return err
lib/connections/quic_misc.go
+72 -6
View File
@@ -10,20 +10,22 @@
package connections
import (
"context"
"crypto/tls"
"net"
"net/url"
"sync/atomic"
"time"
"github.com/quic-go/quic-go"
"github.com/quic-go/quic-go/logging"
"github.com/syncthing/syncthing/lib/osutil"
)
var quicConfig = &quic.Config{
ConnectionIDLength: 4,
MaxIdleTimeout: 30 * time.Second,
KeepAlivePeriod: 15 * time.Second,
MaxIdleTimeout: 30 * time.Second,
KeepAlivePeriod: 15 * time.Second,
}
func quicNetwork(uri *url.URL) string {
@@ -61,11 +63,75 @@ func (q *quicTlsConn) Close() error {
}
func (q *quicTlsConn) ConnectionState() tls.ConnectionState {
return q.Connection.ConnectionState().TLS.ConnectionState
return q.Connection.ConnectionState().TLS
}
func packetConnUnspecified(conn interface{}) bool {
addr := conn.(net.PacketConn).LocalAddr()
func transportConnUnspecified(conn any) bool {
tran, ok := conn.(*quic.Transport)
if !ok {
return false
}
addr := tran.Conn.LocalAddr()
ip, err := osutil.IPFromAddr(addr)
return err == nil && ip.IsUnspecified()
}
type writeTrackingTracer struct {
lastWrite atomic.Int64 // unix nanos
}
func (t *writeTrackingTracer) SentPacket(net.Addr, *logging.Header, logging.ByteCount, []logging.Frame) {
t.lastWrite.Store(time.Now().UnixNano())
}
func (t *writeTrackingTracer) SentVersionNegotiationPacket(_ net.Addr, dest, src logging.ArbitraryLenConnectionID, _ []quic.VersionNumber) {
t.lastWrite.Store(time.Now().UnixNano())
}
func (t *writeTrackingTracer) DroppedPacket(net.Addr, logging.PacketType, logging.ByteCount, logging.PacketDropReason) {
}
func (t *writeTrackingTracer) LastWrite() time.Time {
return time.Unix(0, t.lastWrite.Load())
}
// A transportPacketConn is a net.PacketConn that uses a quic.Transport.
type transportPacketConn struct {
tran *quic.Transport
readDeadline atomic.Value // time.Time
}
func (t *transportPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
ctx := context.Background()
if deadline, ok := t.readDeadline.Load().(time.Time); ok && !deadline.IsZero() {
var cancel context.CancelFunc
ctx, cancel = context.WithDeadline(ctx, deadline)
defer cancel()
}
return t.tran.ReadNonQUICPacket(ctx, p)
}
func (t *transportPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
return t.tran.WriteTo(p, addr)
}
func (t *transportPacketConn) Close() error {
return errUnsupported
}
func (t *transportPacketConn) LocalAddr() net.Addr {
return t.tran.Conn.LocalAddr()
}
func (t *transportPacketConn) SetDeadline(deadline time.Time) error {
return t.SetReadDeadline(deadline)
}
func (t *transportPacketConn) SetReadDeadline(deadline time.Time) error {
t.readDeadline.Store(deadline)
return nil
}
func (t *transportPacketConn) SetWriteDeadline(_ time.Time) error {
return nil // yolo
}
lib/stun/stun.go
+19 -63
View File
@@ -9,10 +9,8 @@ package stun
import (
"context"
"net"
"sync/atomic"
"time"
"github.com/AudriusButkevicius/pfilter"
"github.com/ccding/go-stun/stun"
"github.com/syncthing/syncthing/lib/config"
@@ -21,8 +19,10 @@ import (
const stunRetryInterval = 5 * time.Minute
type Host = stun.Host
type NATType = stun.NATType
type (
Host = stun.Host
NATType = stun.NATType
)
// NAT types.
@@ -38,38 +38,6 @@ const (
NATSymmetricUDPFirewall = stun.NATSymmetricUDPFirewall
)
type writeTrackingUdpConn struct {
// Needs to be UDPConn not PacketConn, as pfilter checks for WriteMsgUDP/ReadMsgUDP
// and even if we embed UDPConn here, in place of a PacketConn, seems the interface
// check fails.
*net.UDPConn
lastWrite atomic.Int64
}
func (c *writeTrackingUdpConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
c.lastWrite.Store(time.Now().Unix())
return c.UDPConn.WriteTo(p, addr)
}
func (c *writeTrackingUdpConn) WriteMsgUDP(b, oob []byte, addr *net.UDPAddr) (n, oobn int, err error) {
c.lastWrite.Store(time.Now().Unix())
return c.UDPConn.WriteMsgUDP(b, oob, addr)
}
func (c *writeTrackingUdpConn) WriteToUDP(b []byte, addr *net.UDPAddr) (int, error) {
c.lastWrite.Store(time.Now().Unix())
return c.UDPConn.WriteToUDP(b, addr)
}
func (c *writeTrackingUdpConn) Write(b []byte) (int, error) {
c.lastWrite.Store(time.Now().Unix())
return c.UDPConn.Write(b)
}
func (c *writeTrackingUdpConn) getLastWrite() time.Time {
return time.Unix(c.lastWrite.Load(), 0)
}
type Subscriber interface {
OnNATTypeChanged(natType NATType)
OnExternalAddressChanged(address *Host, via string)
@@ -79,30 +47,21 @@ type Service struct {
name string
cfg config.Wrapper
subscriber Subscriber
stunConn net.PacketConn
client *stun.Client
writeTrackingUdpConn *writeTrackingUdpConn
lastWriter LastWriter
natType NATType
addr *Host
}
func New(cfg config.Wrapper, subscriber Subscriber, conn *net.UDPConn) (*Service, net.PacketConn) {
// Wrap the original connection to track writes on it
writeTrackingUdpConn := &writeTrackingUdpConn{UDPConn: conn}
// Wrap it in a filter and split it up, so that stun packets arrive on stun conn, others arrive on the data conn
filterConn := pfilter.NewPacketFilter(writeTrackingUdpConn)
otherDataConn := filterConn.NewConn(otherDataPriority, nil)
stunConn := filterConn.NewConn(stunFilterPriority, &stunFilter{
ids: make(map[string]time.Time),
})
filterConn.Start()
type LastWriter interface {
LastWrite() time.Time
}
func New(cfg config.Wrapper, subscriber Subscriber, conn net.PacketConn, lastWriter LastWriter) *Service {
// Construct the client to use the stun conn
client := stun.NewClientWithConnection(stunConn)
client := stun.NewClientWithConnection(conn)
client.SetSoftwareName("") // Explicitly unset this, seems to freak some servers out.
// Return the service and the other conn to the client
@@ -117,15 +76,14 @@ func New(cfg config.Wrapper, subscriber Subscriber, conn *net.UDPConn) (*Service
cfg: cfg,
subscriber: subscriber,
stunConn: stunConn,
client: client,
writeTrackingUdpConn: writeTrackingUdpConn,
lastWriter: lastWriter,
natType: NATUnknown,
addr: nil,
}
return s, otherDataConn
return s
}
func (s *Service) Serve(ctx context.Context) error {
@@ -134,13 +92,6 @@ func (s *Service) Serve(ctx context.Context) error {
s.setExternalAddress(nil, "")
}()
// Closing s.stunConn unblocks operations that use the connection
// (Discover, Keepalive) and might otherwise block us from returning.
go func() {
<-ctx.Done()
_ = s.stunConn.Close()
}()
timer := time.NewTimer(time.Millisecond)
for {
@@ -244,6 +195,7 @@ func (s *Service) stunKeepAlive(ctx context.Context, addr string, extAddr *Host)
l.Debugf("%s starting stun keepalive via %s, next sleep %s", s, addr, nextSleep)
var ourLastWrite time.Time
for {
if areDifferent(s.addr, extAddr) {
// If the port has changed (addresses are not equal but the hosts are equal),
@@ -264,7 +216,10 @@ func (s *Service) stunKeepAlive(ctx context.Context, addr string, extAddr *Host)
}
// Adjust the keepalives to fire only nextSleep after last write.
lastWrite := s.writeTrackingUdpConn.getLastWrite()
lastWrite := ourLastWrite
if quicLastWrite := s.lastWriter.LastWrite(); quicLastWrite.After(lastWrite) {
lastWrite = quicLastWrite
}
minSleep := time.Duration(s.cfg.Options().StunKeepaliveMinS) * time.Second
if nextSleep < minSleep {
nextSleep = minSleep
@@ -293,7 +248,7 @@ func (s *Service) stunKeepAlive(ctx context.Context, addr string, extAddr *Host)
}
// Check if any writes happened while we were sleeping, if they did, sleep again
lastWrite = s.writeTrackingUdpConn.getLastWrite()
lastWrite = s.lastWriter.LastWrite()
if gap := time.Since(lastWrite); gap < nextSleep {
l.Debugf("%s stun last write gap less than next sleep: %s < %s. Will try later", s, gap, nextSleep)
goto tryLater
@@ -306,6 +261,7 @@ func (s *Service) stunKeepAlive(ctx context.Context, addr string, extAddr *Host)
l.Debugf("%s stun keepalive on %s: %s (%v)", s, addr, err, extAddr)
return
}
ourLastWrite = time.Now()
}
}