7 "github.com/anacrolix/dht/v2/krpc"
8 pp "github.com/anacrolix/torrent/peer_protocol"
14 pexAdd pexEventType = iota
18 // internal, based on BEP11
20 pexTargAdded = 25 // put drops on hold when the number of alive connections is lower than this
21 pexMaxHold = 25 // length of the drop hold-back buffer
22 pexMaxDelta = 50 // upper bound on added+added6 and dropped+dropped6 in a single PEX message
25 // represents a single connection (t=pexAdd) or disconnection (t=pexDrop) event
26 type pexEvent struct {
32 // facilitates efficient de-duplication while generating PEX messages
33 type pexMsgFactory struct {
35 added map[addrKey]struct{}
36 dropped map[addrKey]struct{}
39 func (me *pexMsgFactory) DeltaLen() int {
42 int64(len(me.dropped))))
47 // Returns the key to use to identify a given addr in the factory.
48 func (me *pexMsgFactory) addrKey(addr PeerRemoteAddr) addrKey {
49 return addrKey(addr.String())
52 func addrEqual(a, b *krpc.NodeAddr) bool {
53 return a.IP.Equal(b.IP) && a.Port == b.Port
56 func addrIndex(v []krpc.NodeAddr, a *krpc.NodeAddr) int {
58 if addrEqual(&v[i], a) {
65 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
66 // won't hit the limit consuming this event).
67 func (me *pexMsgFactory) add(e pexEvent) {
68 key := me.addrKey(e.addr)
69 if _, ok := me.added[key]; ok {
73 me.added = make(map[addrKey]struct{}, pexMaxDelta)
75 addr, ok := nodeAddr(e.addr)
81 case addr.IP.To4() != nil:
82 if _, ok := me.dropped[key]; ok {
83 if i := addrIndex(m.Dropped.NodeAddrs(), &addr); i >= 0 {
84 m.Dropped = append(m.Dropped[:i], m.Dropped[i+1:]...)
86 delete(me.dropped, key)
89 m.Added = append(m.Added, addr)
90 m.AddedFlags = append(m.AddedFlags, e.f)
91 case len(addr.IP) == net.IPv6len:
92 if _, ok := me.dropped[key]; ok {
93 if i := addrIndex(m.Dropped6.NodeAddrs(), &addr); i >= 0 {
94 m.Dropped6 = append(m.Dropped6[:i], m.Dropped6[i+1:]...)
96 delete(me.dropped, key)
99 m.Added6 = append(m.Added6, addr)
100 m.Added6Flags = append(m.Added6Flags, e.f)
104 me.added[key] = struct{}{}
107 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
108 // won't hit the limit consuming this event).
109 func (me *pexMsgFactory) drop(e pexEvent) {
110 addr, ok := nodeAddr(e.addr)
114 key := me.addrKey(e.addr)
115 if me.dropped == nil {
116 me.dropped = make(map[addrKey]struct{}, pexMaxDelta)
118 if _, ok := me.dropped[key]; ok {
123 case addr.IP.To4() != nil:
124 if _, ok := me.added[key]; ok {
125 if i := addrIndex(m.Added.NodeAddrs(), &addr); i >= 0 {
126 m.Added = append(m.Added[:i], m.Added[i+1:]...)
127 m.AddedFlags = append(m.AddedFlags[:i], m.AddedFlags[i+1:]...)
129 delete(me.added, key)
132 m.Dropped = append(m.Dropped, addr)
133 case len(addr.IP) == net.IPv6len:
134 if _, ok := me.added[key]; ok {
135 if i := addrIndex(m.Added6.NodeAddrs(), &addr); i >= 0 {
136 m.Added6 = append(m.Added6[:i], m.Added6[i+1:]...)
137 m.Added6Flags = append(m.Added6Flags[:i], m.Added6Flags[i+1:]...)
139 delete(me.added, key)
142 m.Dropped6 = append(m.Dropped6, addr)
144 me.dropped[key] = struct{}{}
147 func (me *pexMsgFactory) addEvent(event pexEvent) {
158 func (me *pexMsgFactory) PexMsg() pp.PexMsg {
162 // Convert an arbitrary torrent peer Addr into one that can be represented by the compact addr
164 func nodeAddr(addr PeerRemoteAddr) (krpc.NodeAddr, bool) {
165 ipport, _ := tryIpPortFromNetAddr(addr)
166 ok := ipport.IP != nil
167 return krpc.NodeAddr{IP: shortestIP(ipport.IP), Port: ipport.Port}, ok
170 // mainly for the krpc marshallers
171 func shortestIP(ip net.IP) net.IP {
172 if ip4 := ip.To4(); ip4 != nil {
178 // Per-torrent PEX state
179 type pexState struct {
180 ev []pexEvent // event feed, append-only
181 hold []pexEvent // delayed drops
182 nc int // net number of alive conns
183 initCache pexMsgFactory // last generated initial message
184 initSeq int // number of events which went into initCache
185 initLock sync.RWMutex // serialise access to initCache and initSeq
188 // Reset wipes the state clean, releasing resources. Called from Torrent.Close().
189 func (s *pexState) Reset() {
194 s.initCache = pexMsgFactory{}
199 func (s *pexState) Add(c *PeerConn) {
201 if s.nc >= pexTargAdded {
202 s.ev = append(s.ev, s.hold...)
205 e := c.pexEvent(pexAdd)
206 s.ev = append(s.ev, e)
210 func (s *pexState) Drop(c *PeerConn) {
212 // skip connections which were not previously Added
215 e := c.pexEvent(pexDrop)
217 if s.nc < pexTargAdded && len(s.hold) < pexMaxHold {
218 s.hold = append(s.hold, e)
220 s.ev = append(s.ev, e)
224 // Generate a PEX message based on the event feed. Also returns an index to pass to the subsequent
225 // calls, producing incremental deltas.
226 func (s *pexState) Genmsg(start int) (pp.PexMsg, int) {
231 var factory pexMsgFactory
233 for _, e := range s.ev[start:] {
234 if start > 0 && factory.DeltaLen() >= pexMaxDelta {
240 return factory.PexMsg(), n
243 func (s *pexState) genmsg0() (pp.PexMsg, int) {
245 for _, e := range s.ev[s.initSeq:] {
246 s.initCache.addEvent(e)
252 msg := s.initCache.PexMsg()