"crypto/rand"
"crypto/sha1"
"encoding/binary"
+ "encoding/hex"
"errors"
"expvar"
"fmt"
"net/netip"
"sort"
"strconv"
- "strings"
"time"
+ "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"
)
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
torrents map[InfoHash]*Torrent
pieceRequestOrder map[interface{}]*request_strategy.PieceRequestOrder
- acceptLimiter map[ipStr]int
- dialRateLimiter *rate.Limiter
- numHalfOpen int
+ acceptLimiter map[ipStr]int
+ numHalfOpen int
websocketTrackers websocketTrackers
activeAnnounceLimiter limiter.Instance
httpClient *http.Client
+
+ clientHolepunchAddrSets
}
type ipStr string
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)
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>")
}
// 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
cl.event.L = cl.locker()
cl.ipBlockList = cfg.IPBlocklist
cl.httpClient = &http.Client{
- Transport: &http.Transport{
- Proxy: cfg.HTTPProxy,
+ Transport: cfg.WebTransport,
+ }
+ if cl.httpClient.Transport == nil {
+ cl.httpClient.Transport = &http.Transport{
+ 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,
- },
+ }
}
}
}
return t.announceRequest(event), nil
},
- Proxy: cl.config.HTTPProxy,
+ Proxy: cl.config.HTTPProxy,
+ WebsocketTrackerHttpHeader: cl.config.WebsocketTrackerHttpHeader,
+ ICEServers: cl.config.ICEServers,
+ DialContext: cl.config.TrackerDialContext,
OnConn: func(dc datachannel.ReadWriteCloser, dcc webtorrent.DataChannelContext) {
cl.lock()
defer cl.unlock()
}
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
}
return true
}
for _, t := range cl.torrents {
- if t.wantConns() {
+ if t.wantIncomingConns() {
return true
}
}
for {
conn, err := l.Accept()
torrent.Add("client listener accepts", 1)
+ if err == nil {
+ holepunchAddr, holepunchErr := addrPortFromPeerRemoteAddr(conn.RemoteAddr())
+ if holepunchErr == nil {
+ cl.lock()
+ if g.MapContains(cl.undialableWithoutHolepunch, holepunchAddr) {
+ setAdd(&cl.accepted, holepunchAddr)
+ }
+ if g.MapContains(
+ cl.undialableWithoutHolepunchDialedAfterHolepunchConnect,
+ holepunchAddr,
+ ) {
+ setAdd(&cl.probablyOnlyConnectedDueToHolepunch, holepunchAddr)
+ }
+ cl.unlock()
+ }
+ }
conn = pproffd.WrapNetConn(conn)
cl.rLock()
closed := cl.closed.IsSet()
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()
// 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
}
// Returns a connection over UTP or TCP, whichever is first to connect.
func DialFirst(ctx context.Context, addr string, dialers []Dialer) (res DialResult) {
- {
- t := perf.NewTimer(perf.CallerName(0))
- defer func() {
- if res.Conn == nil {
- t.Mark(fmt.Sprintf("returned no conn (context: %v)", ctx.Err()))
- } else {
- t.Mark("returned conn over " + res.Dialer.DialerNetwork())
- }
- }()
+ pool := dialPool{
+ addr: addr,
}
- ctx, cancel := context.WithCancel(ctx)
- // As soon as we return one connection, cancel the others.
- defer cancel()
- left := 0
- resCh := make(chan DialResult, left)
+ defer pool.startDrainer()
for _, _s := range dialers {
- left++
- s := _s
- go func() {
- resCh <- DialResult{
- dialFromSocket(ctx, s, addr),
- s,
- }
- }()
+ pool.add(ctx, _s)
}
- // Wait for a successful connection.
- func() {
- defer perf.ScopeTimer()()
- for ; left > 0 && res.Conn == nil; left-- {
- res = <-resCh
- }
- }()
- // There are still incompleted dials.
- go func() {
- for ; left > 0; left-- {
- conn := (<-resCh).Conn
- if conn != nil {
- conn.Close()
- }
- }
- }()
- if res.Conn != nil {
- go torrent.Add(fmt.Sprintf("network dialed first: %s", res.Conn.RemoteAddr().Network()), 1)
- }
- return res
+ return pool.getFirst()
}
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)
}
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()
return
}
-// 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) {
- dialCtx, cancel := context.WithTimeout(context.Background(), func() time.Duration {
- cl.rLock()
- defer cl.rUnlock()
- return t.dialTimeout()
- }())
- defer cancel()
- dr := cl.dialFirst(dialCtx, addr.String())
+func doProtocolHandshakeOnDialResult(
+ t *Torrent,
+ obfuscatedHeader bool,
+ addr PeerRemoteAddr,
+ dr DialResult,
+) (
+ c *PeerConn, err error,
+) {
+ cl := t.cl
nc := dr.Conn
- if nc == nil {
- 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))
+ 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()
}
return c, err
}
-// 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) {
+// Returns nil connection and nil error if no connection could be established for valid reasons.
+func (cl *Client) dialAndCompleteHandshake(opts outgoingConnOpts) (c *PeerConn, err error) {
+ // It would be better if dial rate limiting could be tested when considering to open connections
+ // instead. Doing it here means if the limit is low, and the half-open limit is high, we could
+ // end up with lots of outgoing connection attempts pending that were initiated on stale data.
+ {
+ dialReservation := cl.config.DialRateLimiter.Reserve()
+ if !opts.receivedHolepunchConnect {
+ if !dialReservation.OK() {
+ err = errors.New("can't make dial limit reservation")
+ return
+ }
+ time.Sleep(dialReservation.Delay())
+ }
+ }
torrent.Add("establish outgoing connection", 1)
- obfuscatedHeaderFirst := cl.config.HeaderObfuscationPolicy.Preferred
- c, err = cl.establishOutgoingConnEx(t, addr, obfuscatedHeaderFirst)
+ addr := opts.peerInfo.Addr
+ dialPool := dialPool{
+ resCh: make(chan DialResult),
+ addr: addr.String(),
+ }
+ defer dialPool.startDrainer()
+ dialTimeout := opts.t.getDialTimeoutUnlocked()
+ {
+ ctx, cancel := context.WithTimeout(context.Background(), dialTimeout)
+ defer cancel()
+ for _, d := range cl.dialers {
+ dialPool.add(ctx, d)
+ }
+ }
+ holepunchAddr, holepunchAddrErr := addrPortFromPeerRemoteAddr(addr)
+ headerObfuscationPolicy := opts.HeaderObfuscationPolicy
+ obfuscatedHeaderFirst := headerObfuscationPolicy.Preferred
+ firstDialResult := dialPool.getFirst()
+ if firstDialResult.Conn == nil {
+ // No dialers worked. Try to initiate a holepunching rendezvous.
+ if holepunchAddrErr == nil {
+ cl.lock()
+ if !opts.receivedHolepunchConnect {
+ g.MakeMapIfNilAndSet(&cl.undialableWithoutHolepunch, holepunchAddr, struct{}{})
+ }
+ if !opts.skipHolepunchRendezvous {
+ opts.t.trySendHolepunchRendezvous(holepunchAddr)
+ }
+ cl.unlock()
+ }
+ err = fmt.Errorf("all initial dials failed")
+ return
+ }
+ if opts.receivedHolepunchConnect && holepunchAddrErr == nil {
+ cl.lock()
+ if g.MapContains(cl.undialableWithoutHolepunch, holepunchAddr) {
+ g.MakeMapIfNilAndSet(&cl.dialableOnlyAfterHolepunch, holepunchAddr, struct{}{})
+ }
+ g.MakeMapIfNil(&cl.dialedSuccessfullyAfterHolepunchConnect)
+ g.MapInsert(cl.dialedSuccessfullyAfterHolepunchConnect, holepunchAddr, struct{}{})
+ cl.unlock()
+ }
+ c, err = doProtocolHandshakeOnDialResult(
+ opts.t,
+ obfuscatedHeaderFirst,
+ addr,
+ firstDialResult,
+ )
if err == nil {
torrent.Add("initiated conn with preferred header obfuscation", 1)
return
}
- // cl.logger.Printf("error establishing connection to %s (obfuscatedHeader=%t): %v", addr, obfuscatedHeaderFirst, err)
- if cl.config.HeaderObfuscationPolicy.RequirePreferred {
- // We should have just tried with the preferred header obfuscation. If it was required,
- // there's nothing else to try.
+ c.logger.Levelf(
+ log.Debug,
+ "error doing protocol handshake with header obfuscation %v",
+ obfuscatedHeaderFirst,
+ )
+ firstDialResult.Conn.Close()
+ // We should have just tried with the preferred header obfuscation. If it was required, there's nothing else to try.
+ if headerObfuscationPolicy.RequirePreferred {
return
}
- // Try again with encryption if we didn't earlier, or without if we did.
- c, err = cl.establishOutgoingConnEx(t, addr, !obfuscatedHeaderFirst)
+ // Reuse the dialer that returned already but failed to handshake.
+ {
+ ctx, cancel := context.WithTimeout(context.Background(), dialTimeout)
+ defer cancel()
+ dialPool.add(ctx, firstDialResult.Dialer)
+ }
+ secondDialResult := dialPool.getFirst()
+ if secondDialResult.Conn == nil {
+ return
+ }
+ c, err = doProtocolHandshakeOnDialResult(
+ opts.t,
+ !obfuscatedHeaderFirst,
+ addr,
+ secondDialResult,
+ )
if err == nil {
torrent.Add("initiated conn with fallback header obfuscation", 1)
+ return
}
- // cl.logger.Printf("error establishing fallback connection to %v: %v", addr, err)
+ c.logger.Levelf(
+ log.Debug,
+ "error doing protocol handshake with header obfuscation %v",
+ !obfuscatedHeaderFirst,
+ )
+ secondDialResult.Conn.Close()
return
}
+type outgoingConnOpts struct {
+ peerInfo PeerInfo
+ t *Torrent
+ // 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
+ HeaderObfuscationPolicy HeaderObfuscationPolicy
+}
+
// 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) {
- cl.dialRateLimiter.Wait(context.Background())
- c, err := cl.establishOutgoingConn(t, addr)
+func (cl *Client) outgoingConnection(
+ opts outgoingConnOpts,
+ attemptKey outgoingConnAttemptKey,
+) {
+ c, err := cl.dialAndCompleteHandshake(opts)
if err == nil {
c.conn.SetWriteDeadline(time.Time{})
}
cl.lock()
defer cl.unlock()
- // Don't release lock between here and addPeerConn, unless it's for
- // failure.
- cl.noLongerHalfOpen(t, addr.String())
+ // Don't release lock between here and addPeerConn, unless it's for failure.
+ cl.noLongerHalfOpen(opts.t, opts.peerInfo.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.peerInfo.Addr,
+ err,
+ )
}
return
}
defer c.close()
- c.Discovery = ps
- c.trusted = trusted
- t.runHandshookConnLoggingErr(c)
+ c.Discovery = opts.peerInfo.Source
+ c.trusted = opts.peerInfo.Trusted
+ opts.t.runHandshookConnLoggingErr(c)
}
// The port number for incoming peer connections. 0 if the client isn't listening.
if err != nil {
return
}
- successfulPeerWireProtocolHandshakePeerReservedBytes.Add(res.PeerExtensionBits.String(), 1)
+ successfulPeerWireProtocolHandshakePeerReservedBytes.Add(
+ hex.EncodeToString(res.PeerExtensionBits[:]), 1)
ret = res.Hash
c.PeerExtensionBytes = res.PeerExtensionBits
c.PeerID = res.PeerID
return fmt.Errorf("adding connection: %w", err)
}
defer t.dropConnection(c)
- c.startWriter()
+ c.startMessageWriter()
cl.sendInitialMessages(c, t)
c.initUpdateRequestsTimer()
err := c.mainReadLoop()
return nil
}
-const check = false
-
func (p *Peer) initUpdateRequestsTimer() {
- if check {
+ if check.Enabled {
if p.updateRequestsTimer != nil {
panic(p.updateRequestsTimer)
}
// 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) {
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,
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(),
}
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)
},
conns: make(map[*PeerConn]struct{}, 2*cl.config.EstablishedConnsPerTorrent),
- halfOpen: make(map[string]PeerInfo),
-
storageOpener: storageClient,
maxEstablishedConns: cl.config.EstablishedConnsPerTorrent,
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.WithDefaultLevel(log.Debug)
t.sourcesLogger = t.logger.WithNames("sources")
if opts.ChunkSize == 0 {
opts.ChunkSize = defaultChunkSize
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()
}
})
cl.torrents[infoHash] = t
+ t.setInfoBytesLocked(opts.InfoBytes)
cl.clearAcceptLimits()
t.updateWantPeersEvent()
// Tickle Client.waitAccept, new torrent may want conns.
}
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
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.
}
for _, peerAddr := range spec.PeerAddrs {
t.addPeer(PeerInfo{
- Addr: stringAddr(peerAddr),
+ Addr: StringAddr(peerAddr),
Source: PeerSourceDirect,
Trusted: true,
})
// 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()
}
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) {
}
}
-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.logger = c.logger.WithContextText(fmt.Sprintf("%T %p", c, c))
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,
+ "inited with remoteAddr %v network %v outgoing %t",
+ opts.remoteAddr, opts.network, opts.outgoing,
+ )
for _, f := range cl.config.Callbacks.NewPeer {
f(&c.Peer)
}
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 {
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()
}