8 "github.com/anacrolix/dht/v2/krpc"
9 pp "github.com/anacrolix/torrent/peer_protocol"
15 pexAdd pexEventType = iota
19 // internal, based on BEP11
21 pexTargAdded = 25 // put drops on hold when the number of alive connections is lower than this
22 pexMaxHold = 25 // length of the drop hold-back buffer
23 pexMaxDelta = 50 // upper bound on added+added6 and dropped+dropped6 in a single PEX message
26 // represents a single connection (t=pexAdd) or disconnection (t=pexDrop) event
27 type pexEvent struct {
33 // facilitates efficient de-duplication while generating PEX messages
34 type pexMsgFactory struct {
36 added map[addrKey]struct{}
37 dropped map[addrKey]struct{}
40 func (me *pexMsgFactory) DeltaLen() int {
43 int64(len(me.dropped))))
48 // Returns the key to use to identify a given addr in the factory.
49 func (me *pexMsgFactory) addrKey(addr PeerRemoteAddr) addrKey {
50 return addrKey(addr.String())
53 func addrEqual(a, b *krpc.NodeAddr) bool {
54 return a.IP.Equal(b.IP) && a.Port == b.Port
57 func addrIndex(v []krpc.NodeAddr, a *krpc.NodeAddr) int {
58 for i := 0; i < len(v); i += 1 {
59 if addrEqual(&v[i], a) {
66 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
67 // won't hit the limit consuming this event).
68 func (me *pexMsgFactory) add(e pexEvent) {
69 key := me.addrKey(e.addr)
70 if _, ok := me.added[key]; ok {
74 me.added = make(map[addrKey]struct{}, pexMaxDelta)
76 addr, ok := nodeAddr(e.addr)
82 case addr.IP.To4() != nil:
83 if _, ok := me.dropped[key]; ok {
84 if i := addrIndex(m.Dropped.NodeAddrs(), &addr); i >= 0 {
85 m.Dropped = append(m.Dropped[:i], m.Dropped[i+1:]...)
87 delete(me.dropped, key)
90 m.Added = append(m.Added, addr)
91 m.AddedFlags = append(m.AddedFlags, e.f)
92 case len(addr.IP) == net.IPv6len:
93 if _, ok := me.dropped[key]; ok {
94 if i := addrIndex(m.Dropped6.NodeAddrs(), &addr); i >= 0 {
95 m.Dropped6 = append(m.Dropped6[:i], m.Dropped6[i+1:]...)
97 delete(me.dropped, key)
100 m.Added6 = append(m.Added6, addr)
101 m.Added6Flags = append(m.Added6Flags, e.f)
105 me.added[key] = struct{}{}
108 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
109 // won't hit the limit consuming this event).
110 func (me *pexMsgFactory) drop(e pexEvent) {
111 addr, ok := nodeAddr(e.addr)
115 key := me.addrKey(e.addr)
116 if me.dropped == nil {
117 me.dropped = make(map[addrKey]struct{}, pexMaxDelta)
119 if _, ok := me.dropped[key]; ok {
124 case addr.IP.To4() != nil:
125 if _, ok := me.added[key]; ok {
126 if i := addrIndex(m.Added.NodeAddrs(), &addr); i >= 0 {
127 m.Added = append(m.Added[:i], m.Added[i+1:]...)
128 m.AddedFlags = append(m.AddedFlags[:i], m.AddedFlags[i+1:]...)
130 delete(me.added, key)
133 m.Dropped = append(m.Dropped, addr)
134 case len(addr.IP) == net.IPv6len:
135 if _, ok := me.added[key]; ok {
136 if i := addrIndex(m.Added6.NodeAddrs(), &addr); i >= 0 {
137 m.Added6 = append(m.Added6[:i], m.Added6[i+1:]...)
138 m.Added6Flags = append(m.Added6Flags[:i], m.Added6Flags[i+1:]...)
140 delete(me.added, key)
143 m.Dropped6 = append(m.Dropped6, addr)
145 me.dropped[key] = struct{}{}
148 func (me *pexMsgFactory) addEvent(event pexEvent) {
159 func (me *pexMsgFactory) PexMsg() pp.PexMsg {
163 // Convert an arbitrary torrent peer Addr into one that can be represented by the compact addr
165 func nodeAddr(addr PeerRemoteAddr) (krpc.NodeAddr, bool) {
166 ipport, _ := tryIpPortFromNetAddr(addr)
167 ok := ipport.IP != nil
168 return krpc.NodeAddr{IP: shortestIP(ipport.IP), Port: ipport.Port}, ok
171 // mainly for the krpc marshallers
172 func shortestIP(ip net.IP) net.IP {
173 if ip4 := ip.To4(); ip4 != nil {
179 // Per-torrent PEX state
180 type pexState struct {
181 ev []pexEvent // event feed, append-only
182 hold []pexEvent // delayed drops
183 rest time.Time // cooldown deadline on inbound
184 nc int // net number of alive conns
185 initCache pexMsgFactory // last generated initial message
186 initSeq int // number of events which went into initCache
187 initLock sync.RWMutex // serialise access to initCache and initSeq
190 // Reset wipes the state clean, releasing resources. Called from Torrent.Close().
191 func (s *pexState) Reset() {
197 s.initCache = pexMsgFactory{}
202 func (s *pexState) Add(c *PeerConn) {
204 if s.nc >= pexTargAdded {
205 s.ev = append(s.ev, s.hold...)
208 e := c.pexEvent(pexAdd)
209 s.ev = append(s.ev, e)
213 func (s *pexState) Drop(c *PeerConn) {
215 // skip connections which were not previously Added
218 e := c.pexEvent(pexDrop)
220 if s.nc < pexTargAdded && len(s.hold) < pexMaxHold {
221 s.hold = append(s.hold, e)
223 s.ev = append(s.ev, e)
227 // Generate a PEX message based on the event feed. Also returns an index to pass to the subsequent
228 // calls, producing incremental deltas.
229 func (s *pexState) Genmsg(start int) (pp.PexMsg, int) {
234 var factory pexMsgFactory
236 for _, e := range s.ev[start:] {
237 if start > 0 && factory.DeltaLen() >= pexMaxDelta {
243 return factory.PexMsg(), n
246 func (s *pexState) genmsg0() (pp.PexMsg, int) {
248 for _, e := range s.ev[s.initSeq:] {
249 s.initCache.addEvent(e)
255 msg := s.initCache.PexMsg()