import (
"net"
+ "net/netip"
+ "sync"
+ "time"
- "github.com/anacrolix/dht/v2/krpc"
pp "github.com/anacrolix/torrent/peer_protocol"
)
// represents a single connection (t=pexAdd) or disconnection (t=pexDrop) event
type pexEvent struct {
t pexEventType
- addr net.Addr
+ addr netip.AddrPort
f pp.PexPeerFlags
+ next *pexEvent // event feed list
}
-// Combines the node addr, as required for pp.PexMsg.
-type pexMsgAdded struct {
- krpc.NodeAddr
- pp.PexPeerFlags
-}
-
-// Makes generating a PexMsg more efficient.
+// facilitates efficient de-duplication while generating PEX messages
type pexMsgFactory struct {
- added map[string]pexMsgAdded
- dropped map[string]krpc.NodeAddr
+ msg pp.PexMsg
+ added map[netip.AddrPort]struct{}
+ dropped map[netip.AddrPort]struct{}
}
func (me *pexMsgFactory) DeltaLen() int {
int64(len(me.dropped))))
}
-// Returns the key to use to identify a given addr in the factory. Panics if we can't support the
-// addr later in generating a PexMsg (since adding an unusable addr will cause DeltaLen to be out.)
-func (me *pexMsgFactory) addrKey(addr krpc.NodeAddr) string {
- if addr.IP.To4() != nil {
- addr.IP = addr.IP.To4()
- }
- keyBytes, err := addr.MarshalBinary()
- if err != nil {
- panic(err)
- }
- switch len(keyBytes) {
- case compactIpv4NodeAddrElemSize:
- case compactIpv6NodeAddrElemSize:
- default:
- panic(len(keyBytes))
- }
- return string(keyBytes)
+type addrKey = netip.AddrPort
+
+// Returns the key to use to identify a given addr in the factory.
+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
// won't hit the limit consuming this event).
-func (me *pexMsgFactory) Add(addr krpc.NodeAddr, flags pp.PexPeerFlags) bool {
- key := me.addrKey(addr)
- if _, ok := me.dropped[key]; ok {
- delete(me.dropped, key)
- return true
- }
- if me.DeltaLen() >= pexMaxDelta {
- return false
+func (me *pexMsgFactory) add(e pexEvent) {
+ key := me.addrKey(e.addr)
+ if _, ok := me.added[key]; ok {
+ return
}
if me.added == nil {
- me.added = make(map[string]pexMsgAdded, pexMaxDelta)
+ me.added = make(map[addrKey]struct{}, pexMaxDelta)
+ }
+ addr := krpcNodeAddrFromAddrPort(e.addr)
+ m := &me.msg
+ switch {
+ case addr.IP.To4() != nil:
+ if _, ok := me.dropped[key]; ok {
+ if i := m.Dropped.Index(addr); i >= 0 {
+ m.Dropped = append(m.Dropped[:i], m.Dropped[i+1:]...)
+ }
+ delete(me.dropped, key)
+ return
+ }
+ m.Added = append(m.Added, addr)
+ m.AddedFlags = append(m.AddedFlags, e.f)
+ case len(addr.IP) == net.IPv6len:
+ if _, ok := me.dropped[key]; ok {
+ if i := m.Dropped6.Index(addr); i >= 0 {
+ m.Dropped6 = append(m.Dropped6[:i], m.Dropped6[i+1:]...)
+ }
+ delete(me.dropped, key)
+ return
+ }
+ m.Added6 = append(m.Added6, addr)
+ m.Added6Flags = append(m.Added6Flags, e.f)
+ default:
+ panic(addr)
}
- me.added[key] = pexMsgAdded{addr, flags}
- return true
-
+ me.added[key] = struct{}{}
}
// 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(addr krpc.NodeAddr) bool {
- key := me.addrKey(addr)
- if _, ok := me.added[key]; ok {
- delete(me.added, key)
- return true
+func (me *pexMsgFactory) drop(e pexEvent) {
+ addr := krpcNodeAddrFromAddrPort(e.addr)
+ key := me.addrKey(e.addr)
+ if me.dropped == nil {
+ me.dropped = make(map[addrKey]struct{}, pexMaxDelta)
}
- if me.DeltaLen() >= pexMaxDelta {
- return false
+ if _, ok := me.dropped[key]; ok {
+ return
}
- if me.dropped == nil {
- me.dropped = make(map[string]krpc.NodeAddr, pexMaxDelta)
+ m := &me.msg
+ switch {
+ case addr.IP.To4() != nil:
+ if _, ok := me.added[key]; ok {
+ 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:]...)
+ }
+ delete(me.added, key)
+ return
+ }
+ m.Dropped = append(m.Dropped, addr)
+ case len(addr.IP) == net.IPv6len:
+ if _, ok := me.added[key]; ok {
+ 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:]...)
+ }
+ delete(me.added, key)
+ return
+ }
+ m.Dropped6 = append(m.Dropped6, addr)
}
- me.dropped[key] = addr
- return true
+ me.dropped[key] = struct{}{}
}
-// 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) addEvent(event pexEvent) bool {
- addr, ok := nodeAddr(event.addr)
- if !ok {
- return true
- }
+func (me *pexMsgFactory) append(event pexEvent) {
switch event.t {
case pexAdd:
- return me.Add(addr, event.f)
+ me.add(event)
case pexDrop:
- return me.Drop(addr)
+ me.drop(event)
default:
panic(event.t)
}
}
-var compactIpv4NodeAddrElemSize = krpc.CompactIPv4NodeAddrs{}.ElemSize()
-var compactIpv6NodeAddrElemSize = krpc.CompactIPv6NodeAddrs{}.ElemSize()
-
-func (me *pexMsgFactory) PexMsg() (ret pp.PexMsg) {
- for key, added := range me.added {
- switch len(key) {
- case compactIpv4NodeAddrElemSize:
- ret.Added = append(ret.Added, added.NodeAddr)
- ret.AddedFlags = append(ret.AddedFlags, added.PexPeerFlags)
- case compactIpv6NodeAddrElemSize:
- ret.Added6 = append(ret.Added6, added.NodeAddr)
- ret.Added6Flags = append(ret.Added6Flags, added.PexPeerFlags)
- default:
- panic(key)
- }
- }
- for key, addr := range me.dropped {
- switch len(key) {
- case compactIpv4NodeAddrElemSize:
- ret.Dropped = append(ret.Dropped, addr)
- case compactIpv6NodeAddrElemSize:
- ret.Dropped6 = append(ret.Dropped6, addr)
- default:
- panic(key)
- }
- }
- return
-}
-
-func mustNodeAddr(addr net.Addr) krpc.NodeAddr {
- ret, ok := nodeAddr(addr)
- if !ok {
- panic(addr)
- }
- return ret
-}
-
-// Convert an arbitrary torrent peer Addr into one that can be represented by the compact addr
-// format.
-func nodeAddr(addr net.Addr) (_ krpc.NodeAddr, ok bool) {
- ipport, ok := tryIpPortFromNetAddr(addr)
- if !ok {
- return
- }
- return krpc.NodeAddr{IP: shortestIP(ipport.IP), Port: ipport.Port}, true
-}
-
-// 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
+ sync.RWMutex
+ tail *pexEvent // event feed list
hold []pexEvent // delayed drops
- nc int // net number of alive conns
+ // 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.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) {
// 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) {
- var factory pexMsgFactory
- n := start
- for _, e := range s.ev[start:] {
- if !factory.addEvent(e) {
+// 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 msg pexMsgFactory
+ last := start
+ for e := start.next; e != nil; e = e.next {
+ if msg.DeltaLen() >= pexMaxDelta {
break
}
- n++
+ msg.append(*e)
+ last = e
}
- return factory.PexMsg(), n
+ return *msg.PexMsg(), last
+}
+
+// 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++
+ }
+ return
}