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 {
34 added map[addrKey]pexEvent
35 dropped map[addrKey]pexEvent
38 func (me *pexMsgFactory) DeltaLen() int {
41 int64(len(me.dropped))))
46 // Returns the key to use to identify a given addr in the factory.
47 func (me *pexMsgFactory) addrKey(addr net.Addr) addrKey {
48 return addrKey(addr.String())
51 // Returns whether the entry was added (we can check if we're cancelling out another entry and so
52 // won't hit the limit consuming this event).
53 func (me *pexMsgFactory) add(e pexEvent) {
54 key := me.addrKey(e.addr)
55 if _, ok := me.dropped[key]; ok {
56 delete(me.dropped, key)
60 me.added = make(map[addrKey]pexEvent, pexMaxDelta)
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) drop(e pexEvent) {
68 key := me.addrKey(e.addr)
69 if _, ok := me.added[key]; ok {
73 if me.dropped == nil {
74 me.dropped = make(map[addrKey]pexEvent, pexMaxDelta)
79 func (me *pexMsgFactory) addEvent(event pexEvent) {
90 func (me *pexMsgFactory) PexMsg() (ret pp.PexMsg) {
91 for key, added := range me.added {
92 addr, ok := nodeAddr(added.addr)
98 ret.Added = append(ret.Added, addr)
99 ret.AddedFlags = append(ret.AddedFlags, added.f)
101 ret.Added6 = append(ret.Added6, addr)
102 ret.Added6Flags = append(ret.Added6Flags, added.f)
107 for key, dropped := range me.dropped {
108 addr, ok := nodeAddr(dropped.addr)
112 switch len(addr.IP) {
114 ret.Dropped = append(ret.Dropped, addr)
116 ret.Dropped6 = append(ret.Dropped6, addr)
124 // Convert an arbitrary torrent peer Addr into one that can be represented by the compact addr
126 func nodeAddr(addr net.Addr) (_ krpc.NodeAddr, ok bool) {
127 ipport, ok := tryIpPortFromNetAddr(addr)
131 return krpc.NodeAddr{IP: shortestIP(ipport.IP), Port: ipport.Port}, true
134 // mainly for the krpc marshallers
135 func shortestIP(ip net.IP) net.IP {
136 if ip4 := ip.To4(); ip4 != nil {
142 // Per-torrent PEX state
143 type pexState struct {
144 ev []pexEvent // event feed, append-only
145 hold []pexEvent // delayed drops
146 nc int // net number of alive conns
147 initCache pexMsgFactory // last generated initial message
148 initSeq int // number of events which went into initCache
149 initLock sync.RWMutex // serialise access to initCache and initSeq
152 // Reset wipes the state clean, releasing resources. Called from Torrent.Close().
153 func (s *pexState) Reset() {
158 s.initCache = pexMsgFactory{}
163 func (s *pexState) Add(c *PeerConn) {
165 if s.nc >= pexTargAdded {
166 s.ev = append(s.ev, s.hold...)
169 e := c.pexEvent(pexAdd)
170 s.ev = append(s.ev, e)
174 func (s *pexState) Drop(c *PeerConn) {
176 // skip connections which were not previously Added
179 e := c.pexEvent(pexDrop)
181 if s.nc < pexTargAdded && len(s.hold) < pexMaxHold {
182 s.hold = append(s.hold, e)
184 s.ev = append(s.ev, e)
188 // Generate a PEX message based on the event feed. Also returns an index to pass to the subsequent
189 // calls, producing incremental deltas.
190 func (s *pexState) Genmsg(start int) (pp.PexMsg, int) {
195 var factory pexMsgFactory
197 for _, e := range s.ev[start:] {
198 if start > 0 && factory.DeltaLen() >= pexMaxDelta {
204 return factory.PexMsg(), n
207 func (s *pexState) genmsg0() (pp.PexMsg, int) {
209 for _, e := range s.ev[s.initSeq:] {
210 s.initCache.addEvent(e)
216 msg := s.initCache.PexMsg()