import (
"net"
"sync"
- "time"
"github.com/anacrolix/dht/v2/krpc"
+
pp "github.com/anacrolix/torrent/peer_protocol"
)
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 := 0; i < len(v); i += 1 {
- if addrEqual(&v[i], a) {
- return i
- }
- }
- return -1
-}
-
// 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(e pexEvent) {
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)
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)
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:]...)
}
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:]...)
}
}
}
-func (me *pexMsgFactory) PexMsg() pp.PexMsg {
- return me.msg
+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
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
+ return krpc.NodeAddr{IP: ipport.IP, Port: ipport.Port}, ok
}
// Per-torrent PEX state
sync.RWMutex
tail *pexEvent // event feed list
hold []pexEvent // delayed drops
- rest time.Time // cooldown deadline on inbound
nc int // net number of alive conns
msg0 pexMsgFactory // initial message
}
s.tail = nil
s.hold = nil
s.nc = 0
- s.rest = time.Time{}
s.msg0 = pexMsgFactory{}
}
s.RLock()
defer s.RUnlock()
if start == nil {
- return s.msg0.PexMsg(), s.tail
+ return *s.msg0.PexMsg(), s.tail
}
var msg pexMsgFactory
last := start
msg.append(*e)
last = e
}
- return msg.PexMsg(), last
+ 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
}