X-Git-Url: http://www.git.stargrave.org/?a=blobdiff_plain;f=pex.go;h=a0a5f49f5babfb54a2ca11b532d7a017babf4bb8;hb=HEAD;hp=a808af2bc0f9bc6c3724b5d8c4663e03182d67e6;hpb=d8d71addfc59245ec47859e497deedb00b6c02d3;p=btrtrc.git diff --git a/pex.go b/pex.go index a808af2b..a0a5f49f 100644 --- a/pex.go +++ b/pex.go @@ -2,10 +2,10 @@ package torrent import ( "net" + "net/netip" "sync" "time" - "github.com/anacrolix/dht/v2/krpc" pp "github.com/anacrolix/torrent/peer_protocol" ) @@ -26,15 +26,16 @@ const ( // represents a single connection (t=pexAdd) or disconnection (t=pexDrop) event type pexEvent struct { t pexEventType - addr PeerRemoteAddr + addr netip.AddrPort f pp.PexPeerFlags + next *pexEvent // event feed list } // facilitates efficient de-duplication while generating PEX messages type pexMsgFactory struct { msg pp.PexMsg - added map[addrKey]struct{} - dropped map[addrKey]struct{} + added map[netip.AddrPort]struct{} + dropped map[netip.AddrPort]struct{} } func (me *pexMsgFactory) DeltaLen() int { @@ -43,24 +44,11 @@ func (me *pexMsgFactory) DeltaLen() int { int64(len(me.dropped)))) } -type addrKey string +type addrKey = netip.AddrPort // Returns the key to use to identify a given addr in the factory. -func (me *pexMsgFactory) addrKey(addr PeerRemoteAddr) addrKey { - return addrKey(addr.String()) -} - -func addrEqual(a, b *krpc.NodeAddr) bool { - return a.IP.Equal(b.IP) && a.Port == b.Port -} - -func addrIndex(v []krpc.NodeAddr, a *krpc.NodeAddr) int { - for i := range v { - if addrEqual(&v[i], a) { - return i - } - } - return -1 +func (me *pexMsgFactory) addrKey(addr netip.AddrPort) addrKey { + return addr } // Returns whether the entry was added (we can check if we're cancelling out another entry and so @@ -73,15 +61,12 @@ func (me *pexMsgFactory) add(e pexEvent) { if me.added == nil { me.added = make(map[addrKey]struct{}, pexMaxDelta) } - addr, ok := nodeAddr(e.addr) - if !ok { - return - } + addr := krpcNodeAddrFromAddrPort(e.addr) m := &me.msg switch { case addr.IP.To4() != nil: if _, ok := me.dropped[key]; ok { - if i := addrIndex(m.Dropped.NodeAddrs(), &addr); i >= 0 { + if i := m.Dropped.Index(addr); i >= 0 { m.Dropped = append(m.Dropped[:i], m.Dropped[i+1:]...) } delete(me.dropped, key) @@ -91,7 +76,7 @@ func (me *pexMsgFactory) add(e pexEvent) { m.AddedFlags = append(m.AddedFlags, e.f) case len(addr.IP) == net.IPv6len: if _, ok := me.dropped[key]; ok { - if i := addrIndex(m.Dropped6.NodeAddrs(), &addr); i >= 0 { + if i := m.Dropped6.Index(addr); i >= 0 { m.Dropped6 = append(m.Dropped6[:i], m.Dropped6[i+1:]...) } delete(me.dropped, key) @@ -108,10 +93,7 @@ func (me *pexMsgFactory) add(e pexEvent) { // Returns whether the entry was added (we can check if we're cancelling out another entry and so // won't hit the limit consuming this event). func (me *pexMsgFactory) drop(e pexEvent) { - addr, ok := nodeAddr(e.addr) - if !ok { - return - } + addr := krpcNodeAddrFromAddrPort(e.addr) key := me.addrKey(e.addr) if me.dropped == nil { me.dropped = make(map[addrKey]struct{}, pexMaxDelta) @@ -123,7 +105,7 @@ func (me *pexMsgFactory) drop(e pexEvent) { switch { case addr.IP.To4() != nil: if _, ok := me.added[key]; ok { - if i := addrIndex(m.Added.NodeAddrs(), &addr); i >= 0 { + if i := m.Added.Index(addr); i >= 0 { m.Added = append(m.Added[:i], m.Added[i+1:]...) m.AddedFlags = append(m.AddedFlags[:i], m.AddedFlags[i+1:]...) } @@ -133,7 +115,7 @@ func (me *pexMsgFactory) drop(e pexEvent) { m.Dropped = append(m.Dropped, addr) case len(addr.IP) == net.IPv6len: if _, ok := me.added[key]; ok { - if i := addrIndex(m.Added6.NodeAddrs(), &addr); i >= 0 { + if i := m.Added6.Index(addr); i >= 0 { m.Added6 = append(m.Added6[:i], m.Added6[i+1:]...) m.Added6Flags = append(m.Added6Flags[:i], m.Added6Flags[i+1:]...) } @@ -145,7 +127,7 @@ func (me *pexMsgFactory) drop(e pexEvent) { me.dropped[key] = struct{}{} } -func (me *pexMsgFactory) addEvent(event pexEvent) { +func (me *pexMsgFactory) append(event pexEvent) { switch event.t { case pexAdd: me.add(event) @@ -156,58 +138,58 @@ func (me *pexMsgFactory) addEvent(event pexEvent) { } } -func (me *pexMsgFactory) PexMsg() pp.PexMsg { - return me.msg -} - -// Convert an arbitrary torrent peer Addr into one that can be represented by the compact addr -// format. -func nodeAddr(addr PeerRemoteAddr) (krpc.NodeAddr, bool) { - ipport, _ := tryIpPortFromNetAddr(addr) - ok := ipport.IP != nil - return krpc.NodeAddr{IP: shortestIP(ipport.IP), Port: ipport.Port}, ok -} - -// mainly for the krpc marshallers -func shortestIP(ip net.IP) net.IP { - if ip4 := ip.To4(); ip4 != nil { - return ip4 - } - return ip +func (me *pexMsgFactory) PexMsg() *pp.PexMsg { + return &me.msg } // Per-torrent PEX state type pexState struct { - ev []pexEvent // event feed, append-only - hold []pexEvent // delayed drops - rest time.Time // cooldown deadline on inbound - nc int // net number of alive conns - initCache pexMsgFactory // last generated initial message - initSeq int // number of events which went into initCache - initLock sync.RWMutex // serialise access to initCache and initSeq + sync.RWMutex + tail *pexEvent // event feed list + hold []pexEvent // delayed drops + // Torrent-wide cooldown deadline on inbound. This exists to prevent PEX from drowning out other + // peer address sources, until that is fixed. + rest time.Time + nc int // net number of alive conns + msg0 pexMsgFactory // initial message } // Reset wipes the state clean, releasing resources. Called from Torrent.Close(). func (s *pexState) Reset() { - s.ev = nil + s.Lock() + defer s.Unlock() + s.tail = nil s.hold = nil s.nc = 0 s.rest = time.Time{} - s.initLock.Lock() - s.initCache = pexMsgFactory{} - s.initSeq = 0 - s.initLock.Unlock() + s.msg0 = pexMsgFactory{} +} + +func (s *pexState) append(e *pexEvent) { + if s.tail != nil { + s.tail.next = e + } + s.tail = e + s.msg0.append(*e) } func (s *pexState) Add(c *PeerConn) { + e, err := c.pexEvent(pexAdd) + if err != nil { + return + } + s.Lock() + defer s.Unlock() s.nc++ if s.nc >= pexTargAdded { - s.ev = append(s.ev, s.hold...) + for _, e := range s.hold { + ne := e + s.append(&ne) + } s.hold = s.hold[:0] } - e := c.pexEvent(pexAdd) - s.ev = append(s.ev, e) c.pex.Listed = true + s.append(&e) } func (s *pexState) Drop(c *PeerConn) { @@ -215,44 +197,50 @@ func (s *pexState) Drop(c *PeerConn) { // skip connections which were not previously Added return } - e := c.pexEvent(pexDrop) + e, err := c.pexEvent(pexDrop) + if err != nil { + return + } + s.Lock() + defer s.Unlock() s.nc-- if s.nc < pexTargAdded && len(s.hold) < pexMaxHold { s.hold = append(s.hold, e) } else { - s.ev = append(s.ev, e) + s.append(&e) } } -// Generate a PEX message based on the event feed. Also returns an index to pass to the subsequent -// calls, producing incremental deltas. -func (s *pexState) Genmsg(start int) (pp.PexMsg, int) { - if start == 0 { - return s.genmsg0() +// Generate a PEX message based on the event feed. +// Also returns a pointer to pass to the subsequent calls +// to produce incremental deltas. +func (s *pexState) Genmsg(start *pexEvent) (pp.PexMsg, *pexEvent) { + s.RLock() + defer s.RUnlock() + if start == nil { + return *s.msg0.PexMsg(), s.tail } - - var factory pexMsgFactory - n := start - for _, e := range s.ev[start:] { - if start > 0 && factory.DeltaLen() >= pexMaxDelta { + var msg pexMsgFactory + last := start + for e := start.next; e != nil; e = e.next { + if msg.DeltaLen() >= pexMaxDelta { break } - factory.addEvent(e) - n++ + msg.append(*e) + last = e } - return factory.PexMsg(), n + return *msg.PexMsg(), last } -func (s *pexState) genmsg0() (pp.PexMsg, int) { - s.initLock.Lock() - for _, e := range s.ev[s.initSeq:] { - s.initCache.addEvent(e) - s.initSeq++ +// The same as Genmsg but just counts up the distinct events that haven't been sent. +func (s *pexState) numPending(start *pexEvent) (num int) { + s.RLock() + defer s.RUnlock() + if start == nil { + return s.msg0.PexMsg().Len() + } + for e := start.next; e != nil; e = e.next { + num++ } - s.initLock.Unlock() - s.initLock.RLock() - n := s.initSeq - msg := s.initCache.PexMsg() - s.initLock.RUnlock() - return msg, n + return }