pex.go

package torrent

import (
	"net"
	"net/netip"
	"sync"
	"time"

	pp "github.com/anacrolix/torrent/peer_protocol"
)

type pexEventType int

const (
	pexAdd pexEventType = iota
	pexDrop
)

// internal, based on BEP11
const (
	pexTargAdded = 25 // put drops on hold when the number of alive connections is lower than this
	pexMaxHold   = 25 // length of the drop hold-back buffer
	pexMaxDelta  = 50 // upper bound on added+added6 and dropped+dropped6 in a single PEX message
)

// represents a single connection (t=pexAdd) or disconnection (t=pexDrop) event
type pexEvent struct {
	t    pexEventType
	addr netip.AddrPort
	f    pp.PexPeerFlags
	next *pexEvent // event feed list
}

// facilitates efficient de-duplication while generating PEX messages
type pexMsgFactory struct {
	msg     pp.PexMsg
	added   map[netip.AddrPort]struct{}
	dropped map[netip.AddrPort]struct{}
}

func (me *pexMsgFactory) DeltaLen() int {
	return int(max(
		int64(len(me.added)),
		int64(len(me.dropped))))
}

type addrKey = netip.AddrPort

// Returns the key to use to identify a given addr in the factory.
func (me *pexMsgFactory) addrKey(addr netip.AddrPort) addrKey {
	return addr
}

// Returns whether the entry was added (we can check if we're cancelling out another entry and so
// won't hit the limit consuming this event).
func (me *pexMsgFactory) add(e pexEvent) {
	key := me.addrKey(e.addr)
	if _, ok := me.added[key]; ok {
		return
	}
	if me.added == nil {
		me.added = make(map[addrKey]struct{}, pexMaxDelta)
	}
	addr := krpcNodeAddrFromAddrPort(e.addr)
	m := &me.msg
	switch {
	case addr.IP.To4() != nil:
		if _, ok := me.dropped[key]; ok {
			if i := m.Dropped.Index(addr); i >= 0 {
				m.Dropped = append(m.Dropped[:i], m.Dropped[i+1:]...)
			}
			delete(me.dropped, key)
			return
		}
		m.Added = append(m.Added, addr)
		m.AddedFlags = append(m.AddedFlags, e.f)
	case len(addr.IP) == net.IPv6len:
		if _, ok := me.dropped[key]; ok {
			if i := m.Dropped6.Index(addr); i >= 0 {
				m.Dropped6 = append(m.Dropped6[:i], m.Dropped6[i+1:]...)
			}
			delete(me.dropped, key)
			return
		}
		m.Added6 = append(m.Added6, addr)
		m.Added6Flags = append(m.Added6Flags, e.f)
	default:
		panic(addr)
	}
	me.added[key] = struct{}{}
}

// Returns whether the entry was added (we can check if we're cancelling out another entry and so
// won't hit the limit consuming this event).
func (me *pexMsgFactory) drop(e pexEvent) {
	addr := krpcNodeAddrFromAddrPort(e.addr)
	key := me.addrKey(e.addr)
	if me.dropped == nil {
		me.dropped = make(map[addrKey]struct{}, pexMaxDelta)
	}
	if _, ok := me.dropped[key]; ok {
		return
	}
	m := &me.msg
	switch {
	case addr.IP.To4() != nil:
		if _, ok := me.added[key]; ok {
			if i := m.Added.Index(addr); i >= 0 {
				m.Added = append(m.Added[:i], m.Added[i+1:]...)
				m.AddedFlags = append(m.AddedFlags[:i], m.AddedFlags[i+1:]...)
			}
			delete(me.added, key)
			return
		}
		m.Dropped = append(m.Dropped, addr)
	case len(addr.IP) == net.IPv6len:
		if _, ok := me.added[key]; ok {
			if i := m.Added6.Index(addr); i >= 0 {
				m.Added6 = append(m.Added6[:i], m.Added6[i+1:]...)
				m.Added6Flags = append(m.Added6Flags[:i], m.Added6Flags[i+1:]...)
			}
			delete(me.added, key)
			return
		}
		m.Dropped6 = append(m.Dropped6, addr)
	}
	me.dropped[key] = struct{}{}
}

func (me *pexMsgFactory) append(event pexEvent) {
	switch event.t {
	case pexAdd:
		me.add(event)
	case pexDrop:
		me.drop(event)
	default:
		panic(event.t)
	}
}

func (me *pexMsgFactory) PexMsg() *pp.PexMsg {
	return &me.msg
}

// Per-torrent PEX state
type pexState struct {
	sync.RWMutex
	tail *pexEvent  // event feed list
	hold []pexEvent // delayed drops
	// Torrent-wide cooldown deadline on inbound. This exists to prevent PEX from drowning out other
	// peer address sources, until that is fixed.
	rest time.Time
	nc   int           // net number of alive conns
	msg0 pexMsgFactory // initial message
}

// Reset wipes the state clean, releasing resources. Called from Torrent.Close().
func (s *pexState) Reset() {
	s.Lock()
	defer s.Unlock()
	s.tail = nil
	s.hold = nil
	s.nc = 0
	s.rest = time.Time{}
	s.msg0 = pexMsgFactory{}
}

func (s *pexState) append(e *pexEvent) {
	if s.tail != nil {
		s.tail.next = e
	}
	s.tail = e
	s.msg0.append(*e)
}

func (s *pexState) Add(c *PeerConn) {
	e, err := c.pexEvent(pexAdd)
	if err != nil {
		return
	}
	s.Lock()
	defer s.Unlock()
	s.nc++
	if s.nc >= pexTargAdded {
		for _, e := range s.hold {
			ne := e
			s.append(&ne)
		}
		s.hold = s.hold[:0]
	}
	c.pex.Listed = true
	s.append(&e)
}

func (s *pexState) Drop(c *PeerConn) {
	if !c.pex.Listed {
		// skip connections which were not previously Added
		return
	}
	e, err := c.pexEvent(pexDrop)
	if err != nil {
		return
	}
	s.Lock()
	defer s.Unlock()
	s.nc--
	if s.nc < pexTargAdded && len(s.hold) < pexMaxHold {
		s.hold = append(s.hold, e)
	} else {
		s.append(&e)
	}
}

// Generate a PEX message based on the event feed.
// Also returns a pointer to pass to the subsequent calls
// to produce incremental deltas.
func (s *pexState) Genmsg(start *pexEvent) (pp.PexMsg, *pexEvent) {
	s.RLock()
	defer s.RUnlock()
	if start == nil {
		return *s.msg0.PexMsg(), s.tail
	}
	var msg pexMsgFactory
	last := start
	for e := start.next; e != nil; e = e.next {
		if msg.DeltaLen() >= pexMaxDelta {
			break
		}
		msg.append(*e)
		last = e
	}
	return *msg.PexMsg(), last
}

// The same as Genmsg but just counts up the distinct events that haven't been sent.
func (s *pexState) numPending(start *pexEvent) (num int) {
	s.RLock()
	defer s.RUnlock()
	if start == nil {
		return s.msg0.PexMsg().Len()
	}
	for e := start.next; e != nil; e = e.next {
		num++
	}
	return
}