8 "github.com/anacrolix/dht/v2/krpc"
10 pp "github.com/anacrolix/torrent/peer_protocol"
16 pexAdd pexEventType = iota
20 // internal, based on BEP11
22 pexTargAdded = 25 // put drops on hold when the number of alive connections is lower than this
23 pexMaxHold = 25 // length of the drop hold-back buffer
24 pexMaxDelta = 50 // upper bound on added+added6 and dropped+dropped6 in a single PEX message
27 // represents a single connection (t=pexAdd) or disconnection (t=pexDrop) event
28 type pexEvent struct {
32 next *pexEvent // event feed list
35 // facilitates efficient de-duplication while generating PEX messages
36 type pexMsgFactory struct {
38 added map[addrKey]struct{}
39 dropped map[addrKey]struct{}
42 func (me *pexMsgFactory) DeltaLen() int {
45 int64(len(me.dropped))))
50 // Returns the key to use to identify a given addr in the factory.
51 func (me *pexMsgFactory) addrKey(addr PeerRemoteAddr) addrKey {
52 return addrKey(addr.String())
55 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
56 // won't hit the limit consuming this event).
57 func (me *pexMsgFactory) add(e pexEvent) {
58 key := me.addrKey(e.addr)
59 if _, ok := me.added[key]; ok {
63 me.added = make(map[addrKey]struct{}, pexMaxDelta)
65 addr, ok := nodeAddr(e.addr)
71 case addr.IP.To4() != nil:
72 if _, ok := me.dropped[key]; ok {
73 if i := m.Dropped.Index(addr); i >= 0 {
74 m.Dropped = append(m.Dropped[:i], m.Dropped[i+1:]...)
76 delete(me.dropped, key)
79 m.Added = append(m.Added, addr)
80 m.AddedFlags = append(m.AddedFlags, e.f)
81 case len(addr.IP) == net.IPv6len:
82 if _, ok := me.dropped[key]; ok {
83 if i := m.Dropped6.Index(addr); i >= 0 {
84 m.Dropped6 = append(m.Dropped6[:i], m.Dropped6[i+1:]...)
86 delete(me.dropped, key)
89 m.Added6 = append(m.Added6, addr)
90 m.Added6Flags = append(m.Added6Flags, e.f)
94 me.added[key] = struct{}{}
97 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
98 // won't hit the limit consuming this event).
99 func (me *pexMsgFactory) drop(e pexEvent) {
100 addr, ok := nodeAddr(e.addr)
104 key := me.addrKey(e.addr)
105 if me.dropped == nil {
106 me.dropped = make(map[addrKey]struct{}, pexMaxDelta)
108 if _, ok := me.dropped[key]; ok {
113 case addr.IP.To4() != nil:
114 if _, ok := me.added[key]; ok {
115 if i := m.Added.Index(addr); i >= 0 {
116 m.Added = append(m.Added[:i], m.Added[i+1:]...)
117 m.AddedFlags = append(m.AddedFlags[:i], m.AddedFlags[i+1:]...)
119 delete(me.added, key)
122 m.Dropped = append(m.Dropped, addr)
123 case len(addr.IP) == net.IPv6len:
124 if _, ok := me.added[key]; ok {
125 if i := m.Added6.Index(addr); i >= 0 {
126 m.Added6 = append(m.Added6[:i], m.Added6[i+1:]...)
127 m.Added6Flags = append(m.Added6Flags[:i], m.Added6Flags[i+1:]...)
129 delete(me.added, key)
132 m.Dropped6 = append(m.Dropped6, addr)
134 me.dropped[key] = struct{}{}
137 func (me *pexMsgFactory) append(event pexEvent) {
148 func (me *pexMsgFactory) PexMsg() *pp.PexMsg {
152 // Convert an arbitrary torrent peer Addr into one that can be represented by the compact addr
154 func nodeAddr(addr PeerRemoteAddr) (krpc.NodeAddr, bool) {
155 ipport, _ := tryIpPortFromNetAddr(addr)
156 ok := ipport.IP != nil
157 return krpc.NodeAddr{IP: ipport.IP, Port: ipport.Port}, ok
160 // Per-torrent PEX state
161 type pexState struct {
163 tail *pexEvent // event feed list
164 hold []pexEvent // delayed drops
165 rest time.Time // cooldown deadline on inbound
166 nc int // net number of alive conns
167 msg0 pexMsgFactory // initial message
170 // Reset wipes the state clean, releasing resources. Called from Torrent.Close().
171 func (s *pexState) Reset() {
178 s.msg0 = pexMsgFactory{}
181 func (s *pexState) append(e *pexEvent) {
189 func (s *pexState) Add(c *PeerConn) {
193 if s.nc >= pexTargAdded {
194 for _, e := range s.hold {
200 e := c.pexEvent(pexAdd)
205 func (s *pexState) Drop(c *PeerConn) {
207 // skip connections which were not previously Added
212 e := c.pexEvent(pexDrop)
214 if s.nc < pexTargAdded && len(s.hold) < pexMaxHold {
215 s.hold = append(s.hold, e)
221 // Generate a PEX message based on the event feed.
222 // Also returns a pointer to pass to the subsequent calls
223 // to produce incremental deltas.
224 func (s *pexState) Genmsg(start *pexEvent) (pp.PexMsg, *pexEvent) {
228 return *s.msg0.PexMsg(), s.tail
230 var msg pexMsgFactory
232 for e := start.next; e != nil; e = e.next {
233 if msg.DeltaLen() >= pexMaxDelta {
239 return *msg.PexMsg(), last
242 // The same as Genmsg but just counts up the distinct events that haven't been sent.
243 func (s *pexState) numPending(start *pexEvent) (num int) {
247 return s.msg0.PexMsg().Len()
249 for e := start.next; e != nil; e = e.next {