]> Sergey Matveev's repositories - btrtrc.git/blobdiff - client.go
Drop peer request alloc reservations when peer is closed
[btrtrc.git] / client.go
index df746ecd3369644da89fa24b8f927c006ba50c2f..3c42fda4bc489fa5afa361f246da8d6bbe82e109 100644 (file)
--- a/client.go
+++ b/client.go
@@ -16,37 +16,40 @@ import (
        "net/netip"
        "sort"
        "strconv"
-       "strings"
        "time"
 
+       "github.com/anacrolix/torrent/internal/panicif"
+
+       "github.com/anacrolix/chansync"
        "github.com/anacrolix/chansync/events"
        "github.com/anacrolix/dht/v2"
        "github.com/anacrolix/dht/v2/krpc"
-       "github.com/anacrolix/generics"
+       . "github.com/anacrolix/generics"
+       g "github.com/anacrolix/generics"
        "github.com/anacrolix/log"
        "github.com/anacrolix/missinggo/perf"
        "github.com/anacrolix/missinggo/v2"
        "github.com/anacrolix/missinggo/v2/bitmap"
        "github.com/anacrolix/missinggo/v2/pproffd"
        "github.com/anacrolix/sync"
-       request_strategy "github.com/anacrolix/torrent/request-strategy"
        "github.com/davecgh/go-spew/spew"
        "github.com/dustin/go-humanize"
-       "github.com/google/btree"
+       gbtree "github.com/google/btree"
        "github.com/pion/datachannel"
        "golang.org/x/time/rate"
 
-       "github.com/anacrolix/chansync"
-       . "github.com/anacrolix/generics"
-
        "github.com/anacrolix/torrent/bencode"
+       "github.com/anacrolix/torrent/internal/check"
        "github.com/anacrolix/torrent/internal/limiter"
        "github.com/anacrolix/torrent/iplist"
        "github.com/anacrolix/torrent/metainfo"
        "github.com/anacrolix/torrent/mse"
        pp "github.com/anacrolix/torrent/peer_protocol"
+       utHolepunch "github.com/anacrolix/torrent/peer_protocol/ut-holepunch"
+       request_strategy "github.com/anacrolix/torrent/request-strategy"
        "github.com/anacrolix/torrent/storage"
        "github.com/anacrolix/torrent/tracker"
+       "github.com/anacrolix/torrent/types/infohash"
        "github.com/anacrolix/torrent/webtorrent"
 )
 
@@ -55,7 +58,7 @@ import (
 type Client struct {
        // An aggregate of stats over all connections. First in struct to ensure 64-bit alignment of
        // fields. See #262.
-       stats ConnStats
+       connStats ConnStats
 
        _mu    lockWithDeferreds
        event  sync.Cond
@@ -88,6 +91,10 @@ type Client struct {
 
        activeAnnounceLimiter limiter.Instance
        httpClient            *http.Client
+
+       undialableWithoutHolepunch                                   map[netip.AddrPort]struct{}
+       undialableWithoutHolepunchDialAttemptedAfterHolepunchConnect map[netip.AddrPort]struct{}
+       dialableOnlyAfterHolepunch                                   map[netip.AddrPort]struct{}
 }
 
 type ipStr string
@@ -147,7 +154,7 @@ func (cl *Client) WriteStatus(_w io.Writer) {
                fmt.Fprintf(w, "%s DHT server at %s:\n", s.Addr().Network(), s.Addr().String())
                writeDhtServerStatus(w, s)
        })
-       spew.Fdump(w, &cl.stats)
+       dumpStats(w, cl.statsLocked())
        torrentsSlice := cl.torrentsAsSlice()
        fmt.Fprintf(w, "# Torrents: %d\n", len(torrentsSlice))
        fmt.Fprintln(w)
@@ -166,8 +173,8 @@ func (cl *Client) WriteStatus(_w io.Writer) {
                                w,
                                "%f%% of %d bytes (%s)",
                                100*(1-float64(t.bytesMissingLocked())/float64(t.info.TotalLength())),
-                               *t.length,
-                               humanize.Bytes(uint64(*t.length)))
+                               t.length(),
+                               humanize.Bytes(uint64(t.length())))
                } else {
                        w.WriteString("<missing metainfo>")
                }
@@ -195,7 +202,7 @@ func (cl *Client) announceKey() int32 {
 // Initializes a bare minimum Client. *Client and *ClientConfig must not be nil.
 func (cl *Client) init(cfg *ClientConfig) {
        cl.config = cfg
-       generics.MakeMap(&cl.dopplegangerAddrs)
+       g.MakeMap(&cl.dopplegangerAddrs)
        cl.torrents = make(map[metainfo.Hash]*Torrent)
        cl.dialRateLimiter = rate.NewLimiter(10, 10)
        cl.activeAnnounceLimiter.SlotsPerKey = 2
@@ -203,7 +210,8 @@ func (cl *Client) init(cfg *ClientConfig) {
        cl.ipBlockList = cfg.IPBlocklist
        cl.httpClient = &http.Client{
                Transport: &http.Transport{
-                       Proxy: cfg.HTTPProxy,
+                       Proxy:       cfg.HTTPProxy,
+                       DialContext: cfg.HTTPDialContext,
                        // I think this value was observed from some webseeds. It seems reasonable to extend it
                        // to other uses of HTTP from the client.
                        MaxConnsPerHost: 10,
@@ -297,7 +305,9 @@ func NewClient(cfg *ClientConfig) (cl *Client, err error) {
                        }
                        return t.announceRequest(event), nil
                },
-               Proxy: cl.config.HTTPProxy,
+               Proxy:                      cl.config.HTTPProxy,
+               WebsocketTrackerHttpHeader: cl.config.WebsocketTrackerHttpHeader,
+               DialContext:                cl.config.TrackerDialContext,
                OnConn: func(dc datachannel.ReadWriteCloser, dcc webtorrent.DataChannelContext) {
                        cl.lock()
                        defer cl.unlock()
@@ -404,13 +414,7 @@ func (cl *Client) NewAnacrolixDhtServer(conn net.PacketConn) (s *dht.Server, err
        }
        s, err = dht.NewServer(&cfg)
        if err == nil {
-               go func() {
-                       ts, err := s.Bootstrap()
-                       if err != nil {
-                               logger.Levelf(log.Warning, "error bootstrapping dht: %s", err)
-                       }
-                       logger.Levelf(log.Debug, "completed bootstrap: %+v", ts)
-               }()
+               go s.TableMaintainer()
        }
        return
 }
@@ -462,7 +466,7 @@ func (cl *Client) wantConns() bool {
                return true
        }
        for _, t := range cl.torrents {
-               if t.wantConns() {
+               if t.wantIncomingConns() {
                        return true
                }
        }
@@ -551,8 +555,16 @@ func (cl *Client) incomingConnection(nc net.Conn) {
        if tc, ok := nc.(*net.TCPConn); ok {
                tc.SetLinger(0)
        }
-       c := cl.newConnection(nc, false, nc.RemoteAddr(), nc.RemoteAddr().Network(),
-               regularNetConnPeerConnConnString(nc))
+       remoteAddr, _ := tryIpPortFromNetAddr(nc.RemoteAddr())
+       c := cl.newConnection(
+               nc,
+               newConnectionOpts{
+                       outgoing:        false,
+                       remoteAddr:      nc.RemoteAddr(),
+                       localPublicAddr: cl.publicAddr(remoteAddr.IP),
+                       network:         nc.RemoteAddr().Network(),
+                       connString:      regularNetConnPeerConnConnString(nc),
+               })
        defer func() {
                cl.lock()
                defer cl.unlock()
@@ -564,8 +576,8 @@ func (cl *Client) incomingConnection(nc net.Conn) {
 
 // Returns a handle to the given torrent, if it's present in the client.
 func (cl *Client) Torrent(ih metainfo.Hash) (t *Torrent, ok bool) {
-       cl.lock()
-       defer cl.unlock()
+       cl.rLock()
+       defer cl.rUnlock()
        t, ok = cl.torrents[ih]
        return
 }
@@ -640,7 +652,7 @@ func DialFirst(ctx context.Context, addr string, dialers []Dialer) (res DialResu
                        res = <-resCh
                }
        }()
-       // There are still incompleted dials.
+       // There are still uncompleted dials.
        go func() {
                for ; left > 0; left-- {
                        conn := (<-resCh).Conn
@@ -657,8 +669,12 @@ func DialFirst(ctx context.Context, addr string, dialers []Dialer) (res DialResu
 
 func dialFromSocket(ctx context.Context, s Dialer, addr string) net.Conn {
        c, err := s.Dial(ctx, addr)
+       if err != nil {
+               log.Levelf(log.Debug, "error dialing %q: %v", addr, err)
+       }
        // This is a bit optimistic, but it looks non-trivial to thread this through the proxy code. Set
-       // it now in case we close the connection forthwith.
+       // it now in case we close the connection forthwith. Note this is also done in the TCP dialer
+       // code to increase the chance it's done.
        if tc, ok := c.(*net.TCPConn); ok {
                tc.SetLinger(0)
        }
@@ -666,34 +682,40 @@ func dialFromSocket(ctx context.Context, s Dialer, addr string) net.Conn {
        return c
 }
 
-func forgettableDialError(err error) bool {
-       return strings.Contains(err.Error(), "no suitable address found")
-}
-
-func (cl *Client) noLongerHalfOpen(t *Torrent, addr string) {
-       if _, ok := t.halfOpen[addr]; !ok {
-               panic("invariant broken")
+func (cl *Client) noLongerHalfOpen(t *Torrent, addr string, attemptKey outgoingConnAttemptKey) {
+       path := t.getHalfOpenPath(addr, attemptKey)
+       if !path.Exists() {
+               panic("should exist")
        }
-       delete(t.halfOpen, addr)
+       path.Delete()
        cl.numHalfOpen--
+       if cl.numHalfOpen < 0 {
+               panic("should not be possible")
+       }
        for _, t := range cl.torrents {
                t.openNewConns()
        }
 }
 
-// Performs initiator handshakes and returns a connection. Returns nil *connection if no connection
+func (cl *Client) countHalfOpenFromTorrents() (count int) {
+       for _, t := range cl.torrents {
+               count += t.numHalfOpenAttempts()
+       }
+       return
+}
+
+// Performs initiator handshakes and returns a connection. Returns nil *PeerConn if no connection
 // for valid reasons.
 func (cl *Client) initiateProtocolHandshakes(
        ctx context.Context,
        nc net.Conn,
        t *Torrent,
-       outgoing, encryptHeader bool,
-       remoteAddr PeerRemoteAddr,
-       network, connString string,
+       encryptHeader bool,
+       newConnOpts newConnectionOpts,
 ) (
        c *PeerConn, err error,
 ) {
-       c = cl.newConnection(nc, outgoing, remoteAddr, network, connString)
+       c = cl.newConnection(nc, newConnOpts)
        c.headerEncrypted = encryptHeader
        ctx, cancel := context.WithTimeout(ctx, cl.config.HandshakesTimeout)
        defer cancel()
@@ -709,8 +731,77 @@ func (cl *Client) initiateProtocolHandshakes(
        return
 }
 
+func (cl *Client) waitForRendezvousConnect(ctx context.Context, rz *utHolepunchRendezvous) error {
+       for {
+               switch {
+               case rz.gotConnect.IsSet():
+                       return nil
+               case len(rz.relays) == 0:
+                       return errors.New("all relays failed")
+               case ctx.Err() != nil:
+                       return context.Cause(ctx)
+               }
+               relayCond := rz.relayCond.Signaled()
+               cl.unlock()
+               select {
+               case <-rz.gotConnect.Done():
+               case <-relayCond:
+               case <-ctx.Done():
+               }
+               cl.lock()
+       }
+}
+
 // Returns nil connection and nil error if no connection could be established for valid reasons.
-func (cl *Client) establishOutgoingConnEx(t *Torrent, addr PeerRemoteAddr, obfuscatedHeader bool) (*PeerConn, error) {
+func (cl *Client) initiateRendezvousConnect(
+       t *Torrent, holepunchAddr netip.AddrPort,
+) (ok bool, err error) {
+       cl.lock()
+       defer cl.unlock()
+       rz, err := t.startHolepunchRendezvous(holepunchAddr)
+       if err != nil {
+               return
+       }
+       if rz == nil {
+               return
+       }
+       ok = true
+       ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+       defer cancel()
+       err = cl.waitForRendezvousConnect(ctx, rz)
+       delete(t.utHolepunchRendezvous, holepunchAddr)
+       if err != nil {
+               err = fmt.Errorf("waiting for rendezvous connect signal: %w", err)
+       }
+       return
+}
+
+// Returns nil connection and nil error if no connection could be established for valid reasons.
+func (cl *Client) establishOutgoingConnEx(
+       opts outgoingConnOpts,
+       obfuscatedHeader bool,
+) (
+       _ *PeerConn, err error,
+) {
+       t := opts.t
+       addr := opts.addr
+       holepunchAddr, err := addrPortFromPeerRemoteAddr(addr)
+       var sentRendezvous bool
+       if err == nil {
+               if !opts.skipHolepunchRendezvous {
+                       sentRendezvous, err = cl.initiateRendezvousConnect(t, holepunchAddr)
+                       if err != nil {
+                               err = fmt.Errorf("initiating rendezvous connect: %w", err)
+                       }
+               }
+       }
+       gotHolepunchConnect := (err == nil && sentRendezvous) || opts.receivedHolepunchConnect
+       if opts.requireRendezvous && !sentRendezvous {
+               return nil, err
+       }
+       if err != nil {
+               t.logger.Print(err)
+       }
        dialCtx, cancel := context.WithTimeout(context.Background(), func() time.Duration {
                cl.rLock()
                defer cl.rUnlock()
@@ -719,13 +810,49 @@ func (cl *Client) establishOutgoingConnEx(t *Torrent, addr PeerRemoteAddr, obfus
        defer cancel()
        dr := cl.dialFirst(dialCtx, addr.String())
        nc := dr.Conn
+       cl.lock()
+       if gotHolepunchConnect && g.MapContains(cl.undialableWithoutHolepunch, holepunchAddr) {
+               g.MakeMapIfNilAndSet(
+                       &cl.undialableWithoutHolepunchDialAttemptedAfterHolepunchConnect,
+                       holepunchAddr,
+                       struct{}{},
+               )
+       }
+       cl.unlock()
        if nc == nil {
+               if !sentRendezvous && !gotHolepunchConnect {
+                       cl.lock()
+                       g.MakeMapIfNilAndSet(&cl.undialableWithoutHolepunch, holepunchAddr, struct{}{})
+                       cl.unlock()
+               }
                if dialCtx.Err() != nil {
                        return nil, fmt.Errorf("dialing: %w", dialCtx.Err())
                }
                return nil, errors.New("dial failed")
        }
-       c, err := cl.initiateProtocolHandshakes(context.Background(), nc, t, true, obfuscatedHeader, addr, dr.Dialer.DialerNetwork(), regularNetConnPeerConnConnString(nc))
+       if gotHolepunchConnect {
+               panicif.False(holepunchAddr.IsValid())
+               cl.lock()
+               if g.MapContains(cl.undialableWithoutHolepunchDialAttemptedAfterHolepunchConnect, holepunchAddr) {
+                       g.MakeMapIfNilAndSet(
+                               &cl.dialableOnlyAfterHolepunch,
+                               holepunchAddr,
+                               struct{}{},
+                       )
+               }
+               cl.unlock()
+       }
+       addrIpPort, _ := tryIpPortFromNetAddr(addr)
+       c, err := cl.initiateProtocolHandshakes(
+               context.Background(), nc, t, obfuscatedHeader,
+               newConnectionOpts{
+                       outgoing:   true,
+                       remoteAddr: addr,
+                       // It would be possible to retrieve a public IP from the dialer used here?
+                       localPublicAddr: cl.publicAddr(addrIpPort.IP),
+                       network:         dr.Dialer.DialerNetwork(),
+                       connString:      regularNetConnPeerConnConnString(nc),
+               })
        if err != nil {
                nc.Close()
        }
@@ -734,10 +861,10 @@ func (cl *Client) establishOutgoingConnEx(t *Torrent, addr PeerRemoteAddr, obfus
 
 // Returns nil connection and nil error if no connection could be established
 // for valid reasons.
-func (cl *Client) establishOutgoingConn(t *Torrent, addr PeerRemoteAddr) (c *PeerConn, err error) {
+func (cl *Client) establishOutgoingConn(opts outgoingConnOpts) (c *PeerConn, err error) {
        torrent.Add("establish outgoing connection", 1)
        obfuscatedHeaderFirst := cl.config.HeaderObfuscationPolicy.Preferred
-       c, err = cl.establishOutgoingConnEx(t, addr, obfuscatedHeaderFirst)
+       c, err = cl.establishOutgoingConnEx(opts, obfuscatedHeaderFirst)
        if err == nil {
                torrent.Add("initiated conn with preferred header obfuscation", 1)
                return
@@ -749,7 +876,7 @@ func (cl *Client) establishOutgoingConn(t *Torrent, addr PeerRemoteAddr) (c *Pee
                return
        }
        // Try again with encryption if we didn't earlier, or without if we did.
-       c, err = cl.establishOutgoingConnEx(t, addr, !obfuscatedHeaderFirst)
+       c, err = cl.establishOutgoingConnEx(opts, !obfuscatedHeaderFirst)
        if err == nil {
                torrent.Add("initiated conn with fallback header obfuscation", 1)
        }
@@ -757,11 +884,27 @@ func (cl *Client) establishOutgoingConn(t *Torrent, addr PeerRemoteAddr) (c *Pee
        return
 }
 
+type outgoingConnOpts struct {
+       t    *Torrent
+       addr PeerRemoteAddr
+       // Don't attempt to connect unless a connect message is received after initiating a rendezvous.
+       requireRendezvous bool
+       // Don't send rendezvous requests to eligible relays.
+       skipHolepunchRendezvous bool
+       // Outgoing connection attempt is in response to holepunch connect message.
+       receivedHolepunchConnect bool
+}
+
 // Called to dial out and run a connection. The addr we're given is already
 // considered half-open.
-func (cl *Client) outgoingConnection(t *Torrent, addr PeerRemoteAddr, ps PeerSource, trusted bool) {
+func (cl *Client) outgoingConnection(
+       opts outgoingConnOpts,
+       ps PeerSource,
+       trusted bool,
+       attemptKey outgoingConnAttemptKey,
+) {
        cl.dialRateLimiter.Wait(context.Background())
-       c, err := cl.establishOutgoingConn(t, addr)
+       c, err := cl.establishOutgoingConn(opts)
        if err == nil {
                c.conn.SetWriteDeadline(time.Time{})
        }
@@ -769,17 +912,17 @@ func (cl *Client) outgoingConnection(t *Torrent, addr PeerRemoteAddr, ps PeerSou
        defer cl.unlock()
        // Don't release lock between here and addPeerConn, unless it's for
        // failure.
-       cl.noLongerHalfOpen(t, addr.String())
+       cl.noLongerHalfOpen(opts.t, opts.addr.String(), attemptKey)
        if err != nil {
                if cl.config.Debug {
-                       cl.logger.Printf("error establishing outgoing connection to %v: %v", addr, err)
+                       cl.logger.Levelf(log.Debug, "error establishing outgoing connection to %v: %v", opts.addr, err)
                }
                return
        }
        defer c.close()
        c.Discovery = ps
        c.trusted = trusted
-       t.runHandshookConnLoggingErr(c)
+       opts.t.runHandshookConnLoggingErr(c)
 }
 
 // The port number for incoming peer connections. 0 if the client isn't listening.
@@ -973,7 +1116,7 @@ func (cl *Client) runHandshookConn(c *PeerConn, t *Torrent) error {
                return fmt.Errorf("adding connection: %w", err)
        }
        defer t.dropConnection(c)
-       c.startWriter()
+       c.startMessageWriter()
        cl.sendInitialMessages(c, t)
        c.initUpdateRequestsTimer()
        err := c.mainReadLoop()
@@ -983,15 +1126,15 @@ func (cl *Client) runHandshookConn(c *PeerConn, t *Torrent) error {
        return nil
 }
 
-const check = false
-
 func (p *Peer) initUpdateRequestsTimer() {
-       if check {
+       if check.Enabled {
                if p.updateRequestsTimer != nil {
                        panic(p.updateRequestsTimer)
                }
        }
-       p.updateRequestsTimer = time.AfterFunc(math.MaxInt64, p.updateRequestsTimerFunc)
+       if enableUpdateRequestsTimer {
+               p.updateRequestsTimer = time.AfterFunc(math.MaxInt64, p.updateRequestsTimerFunc)
+       }
 }
 
 const peerUpdateRequestsTimerReason = "updateRequestsTimer"
@@ -1018,7 +1161,7 @@ func (c *Peer) updateRequestsTimerFunc() {
 // Maximum pending requests we allow peers to send us. If peer requests are buffered on read, this
 // instructs the amount of memory that might be used to cache pending writes. Assuming 512KiB
 // (1<<19) cached for sending, for 16KiB (1<<14) chunks.
-const localClientReqq = 1 << 5
+const localClientReqq = 1024
 
 // See the order given in Transmission's tr_peerMsgsNew.
 func (cl *Client) sendInitialMessages(conn *PeerConn, torrent *Torrent) {
@@ -1029,7 +1172,8 @@ func (cl *Client) sendInitialMessages(conn *PeerConn, torrent *Torrent) {
                        ExtendedPayload: func() []byte {
                                msg := pp.ExtendedHandshakeMessage{
                                        M: map[pp.ExtensionName]pp.ExtensionNumber{
-                                               pp.ExtensionNameMetadata: metadataExtendedId,
+                                               pp.ExtensionNameMetadata:  metadataExtendedId,
+                                               utHolepunch.ExtensionName: utHolepunchExtendedId,
                                        },
                                        V:            cl.config.ExtendedHandshakeClientVersion,
                                        Reqq:         localClientReqq,
@@ -1037,8 +1181,7 @@ func (cl *Client) sendInitialMessages(conn *PeerConn, torrent *Torrent) {
                                        Encryption:   cl.config.HeaderObfuscationPolicy.Preferred || !cl.config.HeaderObfuscationPolicy.RequirePreferred,
                                        Port:         cl.incomingPeerPort(),
                                        MetadataSize: torrent.metadataSize(),
-                                       // TODO: We can figured these out specific to the socket
-                                       // used.
+                                       // TODO: We can figure these out specific to the socket used.
                                        Ipv4: pp.CompactIp(cl.config.PublicIp4.To4()),
                                        Ipv6: cl.config.PublicIp6.To16(),
                                }
@@ -1136,8 +1279,9 @@ func (cl *Client) badPeerAddr(addr PeerRemoteAddr) bool {
        return false
 }
 
+// Returns whether the IP address and port are considered "bad".
 func (cl *Client) badPeerIPPort(ip net.IP, port int) bool {
-       if port == 0 {
+       if port == 0 || ip == nil {
                return true
        }
        if cl.dopplegangerAddr(net.JoinHostPort(ip.String(), strconv.FormatInt(int64(port), 10))) {
@@ -1176,7 +1320,7 @@ func (cl *Client) newTorrentOpt(opts AddTorrentOpts) (t *Torrent) {
                cl:       cl,
                infoHash: opts.InfoHash,
                peers: prioritizedPeers{
-                       om: btree.New(32),
+                       om: gbtree.New(32),
                        getPrio: func(p PeerInfo) peerPriority {
                                ipPort := p.addr()
                                return bep40PriorityIgnoreError(cl.publicAddr(ipPort.IP), ipPort)
@@ -1184,8 +1328,6 @@ func (cl *Client) newTorrentOpt(opts AddTorrentOpts) (t *Torrent) {
                },
                conns: make(map[*PeerConn]struct{}, 2*cl.config.EstablishedConnsPerTorrent),
 
-               halfOpen: make(map[string]PeerInfo),
-
                storageOpener:       storageClient,
                maxEstablishedConns: cl.config.EstablishedConnsPerTorrent,
 
@@ -1198,7 +1340,7 @@ func (cl *Client) newTorrentOpt(opts AddTorrentOpts) (t *Torrent) {
        t.smartBanCache.Hash = sha1.Sum
        t.smartBanCache.Init()
        t.networkingEnabled.Set()
-       t.logger = cl.logger.WithContextValue(t).WithNames("torrent", t.infoHash.HexString())
+       t.logger = cl.logger.WithContextValue(t).WithNames("torrent", t.infoHash.HexString()).WithDefaultLevel(log.Debug)
        t.sourcesLogger = t.logger.WithNames("sources")
        if opts.ChunkSize == 0 {
                opts.ChunkSize = defaultChunkSize
@@ -1245,9 +1387,8 @@ func (cl *Client) AddTorrentInfoHashWithStorage(infoHash metainfo.Hash, specStor
        return
 }
 
-// Adds a torrent by InfoHash with a custom Storage implementation.
-// If the torrent already exists then this Storage is ignored and the
-// existing torrent returned with `new` set to `false`
+// Adds a torrent by InfoHash with a custom Storage implementation. If the torrent already exists
+// then this Storage is ignored and the existing torrent returned with `new` set to `false`.
 func (cl *Client) AddTorrentOpt(opts AddTorrentOpts) (t *Torrent, new bool) {
        infoHash := opts.InfoHash
        cl.lock()
@@ -1265,6 +1406,7 @@ func (cl *Client) AddTorrentOpt(opts AddTorrentOpts) (t *Torrent, new bool) {
                }
        })
        cl.torrents[infoHash] = t
+       t.setInfoBytesLocked(opts.InfoBytes)
        cl.clearAcceptLimits()
        t.updateWantPeersEvent()
        // Tickle Client.waitAccept, new torrent may want conns.
@@ -1273,9 +1415,10 @@ func (cl *Client) AddTorrentOpt(opts AddTorrentOpts) (t *Torrent, new bool) {
 }
 
 type AddTorrentOpts struct {
-       InfoHash  InfoHash
+       InfoHash  infohash.T
        Storage   storage.ClientImpl
        ChunkSize pp.Integer
+       InfoBytes []byte
 }
 
 // Add or merge a torrent spec. Returns new if the torrent wasn't already in the client. See also
@@ -1299,13 +1442,6 @@ func (cl *Client) AddTorrentSpec(spec *TorrentSpec) (t *Torrent, new bool, err e
        return
 }
 
-type stringAddr string
-
-var _ net.Addr = stringAddr("")
-
-func (stringAddr) Network() string   { return "" }
-func (me stringAddr) String() string { return string(me) }
-
 // The trackers will be merged with the existing ones. If the Info isn't yet known, it will be set.
 // spec.DisallowDataDownload/Upload will be read and applied
 // The display name is replaced if the new spec provides one. Note that any `Storage` is ignored.
@@ -1326,11 +1462,11 @@ func (t *Torrent) MergeSpec(spec *TorrentSpec) error {
        defer cl.unlock()
        t.initialPieceCheckDisabled = spec.DisableInitialPieceCheck
        for _, url := range spec.Webseeds {
-               t.addWebSeed(url, spec.DefaultWebseedEscapePath)
+               t.addWebSeed(url)
        }
        for _, peerAddr := range spec.PeerAddrs {
                t.addPeer(PeerInfo{
-                       Addr:    stringAddr(peerAddr),
+                       Addr:    StringAddr(peerAddr),
                        Source:  PeerSourceDirect,
                        Trusted: true,
                })
@@ -1384,8 +1520,8 @@ func (cl *Client) WaitAll() bool {
 
 // Returns handles to all the torrents loaded in the Client.
 func (cl *Client) Torrents() []*Torrent {
-       cl.lock()
-       defer cl.unlock()
+       cl.rLock()
+       defer cl.rUnlock()
        return cl.torrentsAsSlice()
 }
 
@@ -1453,7 +1589,7 @@ func (cl *Client) banPeerIP(ip net.IP) {
        if !ok {
                panic(ip)
        }
-       generics.MakeMapIfNilAndSet(&cl.badPeerIPs, ipAddr, struct{}{})
+       g.MakeMapIfNilAndSet(&cl.badPeerIPs, ipAddr, struct{}{})
        for _, t := range cl.torrents {
                t.iterPeers(func(p *Peer) {
                        if p.remoteIp().Equal(ip) {
@@ -1465,39 +1601,54 @@ func (cl *Client) banPeerIP(ip net.IP) {
        }
 }
 
-func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr PeerRemoteAddr, network, connString string) (c *PeerConn) {
-       if network == "" {
-               panic(remoteAddr)
+type newConnectionOpts struct {
+       outgoing        bool
+       remoteAddr      PeerRemoteAddr
+       localPublicAddr peerLocalPublicAddr
+       network         string
+       connString      string
+}
+
+func (cl *Client) newConnection(nc net.Conn, opts newConnectionOpts) (c *PeerConn) {
+       if opts.network == "" {
+               panic(opts.remoteAddr)
        }
        c = &PeerConn{
                Peer: Peer{
-                       outgoing:        outgoing,
+                       outgoing:        opts.outgoing,
                        choking:         true,
                        peerChoking:     true,
                        PeerMaxRequests: 250,
 
-                       RemoteAddr: remoteAddr,
-                       Network:    network,
-                       callbacks:  &cl.config.Callbacks,
+                       RemoteAddr:      opts.remoteAddr,
+                       localPublicAddr: opts.localPublicAddr,
+                       Network:         opts.network,
+                       callbacks:       &cl.config.Callbacks,
                },
-               connString: connString,
+               connString: opts.connString,
                conn:       nc,
        }
+       c.peerRequestDataAllocLimiter.Max = cl.config.MaxAllocPeerRequestDataPerConn
+       c.initRequestState()
        // TODO: Need to be much more explicit about this, including allowing non-IP bannable addresses.
-       if remoteAddr != nil {
-               netipAddrPort, err := netip.ParseAddrPort(remoteAddr.String())
+       if opts.remoteAddr != nil {
+               netipAddrPort, err := netip.ParseAddrPort(opts.remoteAddr.String())
                if err == nil {
                        c.bannableAddr = Some(netipAddrPort.Addr())
                }
        }
        c.peerImpl = c
-       c.logger = cl.logger.WithDefaultLevel(log.Warning).WithContextValue(c)
+       c.logger = cl.logger.WithDefaultLevel(log.Warning)
        c.setRW(connStatsReadWriter{nc, c})
        c.r = &rateLimitedReader{
                l: cl.config.DownloadRateLimiter,
                r: c.r,
        }
-       c.logger.WithDefaultLevel(log.Debug).Printf("initialized with remote %v over network %v (outgoing=%t)", remoteAddr, network, outgoing)
+       c.logger.Levelf(
+               log.Debug,
+               "new PeerConn %p [Client %p remoteAddr %v network %v outgoing %t]",
+               c, cl, opts.remoteAddr, opts.network, opts.outgoing,
+       )
        for _, f := range cl.config.Callbacks.NewPeer {
                f(&c.Peer)
        }
@@ -1586,6 +1737,16 @@ func (cl *Client) ListenAddrs() (ret []net.Addr) {
        return
 }
 
+func (cl *Client) PublicIPs() (ips []net.IP) {
+       if ip := cl.config.PublicIp4; ip != nil {
+               ips = append(ips, ip)
+       }
+       if ip := cl.config.PublicIp6; ip != nil {
+               ips = append(ips, ip)
+       }
+       return
+}
+
 func (cl *Client) onBadAccept(addr PeerRemoteAddr) {
        ipa, ok := tryIpPortFromNetAddr(addr)
        if !ok {
@@ -1653,8 +1814,24 @@ func (cl *Client) String() string {
        return fmt.Sprintf("<%[1]T %[1]p>", cl)
 }
 
-// Returns connection-level aggregate stats at the Client level. See the comment on
+// Returns connection-level aggregate connStats at the Client level. See the comment on
 // TorrentStats.ConnStats.
 func (cl *Client) ConnStats() ConnStats {
-       return cl.stats.Copy()
+       return cl.connStats.Copy()
+}
+
+func (cl *Client) Stats() ClientStats {
+       cl.rLock()
+       defer cl.rUnlock()
+       return cl.statsLocked()
+}
+
+func (cl *Client) statsLocked() (stats ClientStats) {
+       stats.ConnStats = cl.connStats.Copy()
+       stats.ActiveHalfOpenAttempts = cl.numHalfOpen
+       stats.NumPeersUndialableWithoutHolepunchDialedAfterHolepunchConnect =
+               len(cl.undialableWithoutHolepunchDialAttemptedAfterHolepunchConnect)
+       stats.NumPeersDialableOnlyAfterHolepunch =
+               len(cl.dialableOnlyAfterHolepunch)
+       return
 }