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 {
31 next *pexEvent // event feed list
34 // facilitates efficient de-duplication while generating PEX messages
35 type pexMsgFactory struct {
37 added map[netip.AddrPort]struct{}
38 dropped map[netip.AddrPort]struct{}
41 func (me *pexMsgFactory) DeltaLen() int {
44 int64(len(me.dropped))))
47 type addrKey = netip.AddrPort
49 // Returns the key to use to identify a given addr in the factory.
50 func (me *pexMsgFactory) addrKey(addr netip.AddrPort) addrKey {
54 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
55 // won't hit the limit consuming this event).
56 func (me *pexMsgFactory) add(e pexEvent) {
57 key := me.addrKey(e.addr)
58 if _, ok := me.added[key]; ok {
62 me.added = make(map[addrKey]struct{}, pexMaxDelta)
64 addr := krpcNodeAddrFromAddrPort(e.addr)
67 case addr.IP.To4() != nil:
68 if _, ok := me.dropped[key]; ok {
69 if i := m.Dropped.Index(addr); i >= 0 {
70 m.Dropped = append(m.Dropped[:i], m.Dropped[i+1:]...)
72 delete(me.dropped, key)
75 m.Added = append(m.Added, addr)
76 m.AddedFlags = append(m.AddedFlags, e.f)
77 case len(addr.IP) == net.IPv6len:
78 if _, ok := me.dropped[key]; ok {
79 if i := m.Dropped6.Index(addr); i >= 0 {
80 m.Dropped6 = append(m.Dropped6[:i], m.Dropped6[i+1:]...)
82 delete(me.dropped, key)
85 m.Added6 = append(m.Added6, addr)
86 m.Added6Flags = append(m.Added6Flags, e.f)
90 me.added[key] = struct{}{}
93 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
94 // won't hit the limit consuming this event).
95 func (me *pexMsgFactory) drop(e pexEvent) {
96 addr := krpcNodeAddrFromAddrPort(e.addr)
97 key := me.addrKey(e.addr)
98 if me.dropped == nil {
99 me.dropped = make(map[addrKey]struct{}, pexMaxDelta)
101 if _, ok := me.dropped[key]; ok {
106 case addr.IP.To4() != nil:
107 if _, ok := me.added[key]; ok {
108 if i := m.Added.Index(addr); i >= 0 {
109 m.Added = append(m.Added[:i], m.Added[i+1:]...)
110 m.AddedFlags = append(m.AddedFlags[:i], m.AddedFlags[i+1:]...)
112 delete(me.added, key)
115 m.Dropped = append(m.Dropped, addr)
116 case len(addr.IP) == net.IPv6len:
117 if _, ok := me.added[key]; ok {
118 if i := m.Added6.Index(addr); i >= 0 {
119 m.Added6 = append(m.Added6[:i], m.Added6[i+1:]...)
120 m.Added6Flags = append(m.Added6Flags[:i], m.Added6Flags[i+1:]...)
122 delete(me.added, key)
125 m.Dropped6 = append(m.Dropped6, addr)
127 me.dropped[key] = struct{}{}
130 func (me *pexMsgFactory) append(event pexEvent) {
141 func (me *pexMsgFactory) PexMsg() *pp.PexMsg {
145 // Per-torrent PEX state
146 type pexState struct {
148 tail *pexEvent // event feed list
149 hold []pexEvent // delayed drops
150 // Torrent-wide cooldown deadline on inbound. This exists to prevent PEX from drowning out other
151 // peer address sources, until that is fixed.
153 nc int // net number of alive conns
154 msg0 pexMsgFactory // initial message
157 // Reset wipes the state clean, releasing resources. Called from Torrent.Close().
158 func (s *pexState) Reset() {
165 s.msg0 = pexMsgFactory{}
168 func (s *pexState) append(e *pexEvent) {
176 func (s *pexState) Add(c *PeerConn) {
177 e, err := c.pexEvent(pexAdd)
184 if s.nc >= pexTargAdded {
185 for _, e := range s.hold {
195 func (s *pexState) Drop(c *PeerConn) {
197 // skip connections which were not previously Added
200 e, err := c.pexEvent(pexDrop)
207 if s.nc < pexTargAdded && len(s.hold) < pexMaxHold {
208 s.hold = append(s.hold, e)
214 // Generate a PEX message based on the event feed.
215 // Also returns a pointer to pass to the subsequent calls
216 // to produce incremental deltas.
217 func (s *pexState) Genmsg(start *pexEvent) (pp.PexMsg, *pexEvent) {
221 return *s.msg0.PexMsg(), s.tail
223 var msg pexMsgFactory
225 for e := start.next; e != nil; e = e.next {
226 if msg.DeltaLen() >= pexMaxDelta {
232 return *msg.PexMsg(), last
235 // The same as Genmsg but just counts up the distinct events that haven't been sent.
236 func (s *pexState) numPending(start *pexEvent) (num int) {
240 return s.msg0.PexMsg().Len()
242 for e := start.next; e != nil; e = e.next {