func (r *Reactor) calculateNextRequestTime() {
	if ratio := r.peerManager.PeerRatio(); ratio >= 0.95 {
		r.Logger.Debug("peer manager near full ratio, sleeping...",
			"sleep_period", fullCapacityInterval, "ratio", ratio)
		r.nextRequestTime = time.Now().Add(fullCapacityInterval)
		return
	}
	peers := tmmath.MinInt(len(r.availablePeers), 50)
	baseTime := minReceiveRequestInterval
	if peers > 0 {
		baseTime = minReceiveRequestInterval * 2 / time.Duration(peers)
	}
	if r.totalPeers > 0 || r.discoveryRatio == 0 {
		ratio := float32(r.totalPeers+1) / float32(r.newPeers+1)
		r.discoveryRatio = ratio * ratio
		r.newPeers = 0
		r.totalPeers = 0
	}
	r.nextRequestTime = time.Now().Add(baseTime * time.Duration(r.discoveryRatio))
}
calculateNextRequestTime implements something of a proportional controller
to estimate how often the reactor should be requesting new peer addresses.
The dependent variable in this calculation is the ratio of new peers to
all peers that the reactor receives. The interval is thus calculated as the
inverse squared. In the beginning, all peers should be new peers.
We  expect this ratio to be near 1 and thus the interval to be as short
as possible. As the node becomes more familiar with the network the ratio of
new nodes will plummet to a very small number, meaning the interval expands
to its upper bound.
CONTRACT: Must use a write lock as nextRequestTime is updated
package pex
import (
	"context"
	"fmt"
	"runtime/debug"
	"sync"
	"time"
	"github.com/tendermint/tendermint/internal/p2p"
	"github.com/tendermint/tendermint/internal/p2p/conn"
	"github.com/tendermint/tendermint/libs/log"
	tmmath "github.com/tendermint/tendermint/libs/math"
	"github.com/tendermint/tendermint/libs/service"
	protop2p "github.com/tendermint/tendermint/proto/tendermint/p2p"
	"github.com/tendermint/tendermint/types"
)
var (
	_ service.Service = (*Reactor)(nil)
	_ p2p.Wrapper     = (*protop2p.PexMessage)(nil)
)
const (
	PexChannel = 0x00
	maxAddressSize = 256
	maxGetSelection = 250
	maxMsgSize = maxAddressSize * maxGetSelection
	minReceiveRequestInterval = 100 * time.Millisecond
	maxAddresses uint16 = 100
	resolveTimeout = 3 * time.Second
	noAvailablePeersWaitPeriod = 1 * time.Second
	fullCapacityInterval = 10 * time.Minute
)
func ChannelDescriptor() *conn.ChannelDescriptor {
	return &conn.ChannelDescriptor{
		ID:                  PexChannel,
		MessageType:         new(protop2p.PexMessage),
		Priority:            1,
		SendQueueCapacity:   10,
		RecvMessageCapacity: maxMsgSize,
		RecvBufferCapacity:  128,
	}
}
type Reactor struct {
	service.BaseService
	peerManager *p2p.PeerManager
	pexCh       *p2p.Channel
	peerUpdates *p2p.PeerUpdates
	closeCh     chan struct{}
	availablePeers map[types.NodeID]struct{}
	mtx sync.RWMutex
	requestsSent map[types.NodeID]struct{}
	lastReceivedRequests map[types.NodeID]time.Time
	nextRequestTime time.Time
	newPeers   uint32
	totalPeers uint32
	discoveryRatio float32
}
func NewReactor(
	logger log.Logger,
	peerManager *p2p.PeerManager,
	pexCh *p2p.Channel,
	peerUpdates *p2p.PeerUpdates,
) *Reactor {
	r := &Reactor{
		peerManager:          peerManager,
		pexCh:                pexCh,
		peerUpdates:          peerUpdates,
		closeCh:              make(chan struct{}),
		availablePeers:       make(map[types.NodeID]struct{}),
		requestsSent:         make(map[types.NodeID]struct{}),
		lastReceivedRequests: make(map[types.NodeID]time.Time),
	}
	r.BaseService = *service.NewBaseService(logger, "PEX", r)
	return r
}
func (r *Reactor) OnStart() error {
	go r.processPexCh()
	go r.processPeerUpdates()
	return nil
}
func (r *Reactor) OnStop() {
	close(r.closeCh)
	<-r.pexCh.Done()
	<-r.peerUpdates.Done()
}
func (r *Reactor) processPexCh() {
	defer r.pexCh.Close()
	for {
		select {
		case <-r.closeCh:
			r.Logger.Debug("stopped listening on PEX channel; closing...")
			return
		case <-r.waitUntilNextRequest():
			r.sendRequestForPeers()
		case envelope := <-r.pexCh.In:
			if err := r.handleMessage(r.pexCh.ID, envelope); err != nil {
				r.Logger.Error("failed to process message", "ch_id", r.pexCh.ID, "envelope", envelope, "err", err)
				r.pexCh.Error <- p2p.PeerError{
					NodeID: envelope.From,
					Err:    err,
				}
			}
		}
	}
}
func (r *Reactor) processPeerUpdates() {
	defer r.peerUpdates.Close()
	for {
		select {
		case peerUpdate := <-r.peerUpdates.Updates():
			r.processPeerUpdate(peerUpdate)
		case <-r.closeCh:
			r.Logger.Debug("stopped listening on peer updates channel; closing...")
			return
		}
	}
}
func (r *Reactor) handlePexMessage(envelope p2p.Envelope) error {
	logger := r.Logger.With("peer", envelope.From)
	switch msg := envelope.Message.(type) {
	case *protop2p.PexRequest:
		if err := r.markPeerRequest(envelope.From); err != nil {
			return err
		}
		pexAddresses := r.resolve(r.peerManager.Advertise(envelope.From, maxAddresses))
		r.pexCh.Out <- p2p.Envelope{
			To:      envelope.From,
			Message: &protop2p.PexResponse{Addresses: pexAddresses},
		}
	case *protop2p.PexResponse:
		if err := r.markPeerResponse(envelope.From); err != nil {
			return err
		}
		if len(msg.Addresses) > int(maxAddresses) {
			return fmt.Errorf("peer sent too many addresses (max: %d, got: %d)",
				maxAddresses,
				len(msg.Addresses),
			)
		}
		for _, pexAddress := range msg.Addresses {
			peerAddress, err := p2p.ParseNodeAddress(
				fmt.Sprintf("%s@%s:%d", pexAddress.ID, pexAddress.IP, pexAddress.Port))
			if err != nil {
				continue
			}
			added, err := r.peerManager.Add(peerAddress)
			if err != nil {
				logger.Error("failed to add PEX address", "address", peerAddress, "err", err)
			}
			if added {
				r.newPeers++
				logger.Debug("added PEX address", "address", peerAddress)
			}
			r.totalPeers++
		}
	case *protop2p.PexRequestV2:
		if err := r.markPeerRequest(envelope.From); err != nil {
			return err
		}
		nodeAddresses := r.peerManager.Advertise(envelope.From, maxAddresses)
		pexAddressesV2 := make([]protop2p.PexAddressV2, len(nodeAddresses))
		for idx, addr := range nodeAddresses {
			pexAddressesV2[idx] = protop2p.PexAddressV2{
				URL: addr.String(),
			}
		}
		r.pexCh.Out <- p2p.Envelope{
			To:      envelope.From,
			Message: &protop2p.PexResponseV2{Addresses: pexAddressesV2},
		}
	case *protop2p.PexResponseV2:
		if err := r.markPeerResponse(envelope.From); err != nil {
			return err
		}
		if len(msg.Addresses) > int(maxAddresses) {
			return fmt.Errorf("peer sent too many addresses (max: %d, got: %d)",
				maxAddresses,
				len(msg.Addresses),
			)
		}
		for _, pexAddress := range msg.Addresses {
			peerAddress, err := p2p.ParseNodeAddress(pexAddress.URL)
			if err != nil {
				continue
			}
			added, err := r.peerManager.Add(peerAddress)
			if err != nil {
				logger.Error("failed to add V2 PEX address", "address", peerAddress, "err", err)
			}
			if added {
				r.newPeers++
				logger.Debug("added V2 PEX address", "address", peerAddress)
			}
			r.totalPeers++
		}
	default:
		return fmt.Errorf("received unknown message: %T", msg)
	}
	return nil
}
func (r *Reactor) resolve(addresses []p2p.NodeAddress) []protop2p.PexAddress {
	limit := len(addresses)
	pexAddresses := make([]protop2p.PexAddress, 0, limit)
	for _, address := range addresses {
		ctx, cancel := context.WithTimeout(context.Background(), resolveTimeout)
		endpoints, err := address.Resolve(ctx)
		r.Logger.Debug("resolved node address", "endpoints", endpoints)
		cancel()
		if err != nil {
			r.Logger.Debug("failed to resolve address", "address", address, "err", err)
			continue
		}
		for _, endpoint := range endpoints {
			r.Logger.Debug("checking endpint", "IP", endpoint.IP, "Port", endpoint.Port)
			if len(pexAddresses) >= limit {
				return pexAddresses
			} else if endpoint.IP != nil {
				r.Logger.Debug("appending pex address")
				pexAddresses = append(pexAddresses, protop2p.PexAddress{
					ID:   string(address.NodeID),
					IP:   endpoint.IP.String(),
					Port: uint32(endpoint.Port),
				})
			}
		}
	}
	return pexAddresses
}
func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err error) {
	defer func() {
		if e := recover(); e != nil {
			err = fmt.Errorf("panic in processing message: %v", e)
			r.Logger.Error(
				"recovering from processing message panic",
				"err", err,
				"stack", string(debug.Stack()),
			)
		}
	}()
	r.Logger.Debug("received PEX message", "peer", envelope.From)
	switch chID {
	case p2p.ChannelID(PexChannel):
		err = r.handlePexMessage(envelope)
	default:
		err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
	}
	return err
}
func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
	r.Logger.Debug("received PEX peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
	r.mtx.Lock()
	defer r.mtx.Unlock()
	switch peerUpdate.Status {
	case p2p.PeerStatusUp:
		r.availablePeers[peerUpdate.NodeID] = struct{}{}
	case p2p.PeerStatusDown:
		delete(r.availablePeers, peerUpdate.NodeID)
		delete(r.requestsSent, peerUpdate.NodeID)
		delete(r.lastReceivedRequests, peerUpdate.NodeID)
	default:
	}
}
func (r *Reactor) waitUntilNextRequest() <-chan time.Time {
	return time.After(time.Until(r.nextRequestTime))
}
func (r *Reactor) sendRequestForPeers() {
	r.mtx.Lock()
	defer r.mtx.Unlock()
	if len(r.availablePeers) == 0 {
		r.Logger.Debug("no available peers to send request to, waiting...")
		r.nextRequestTime = time.Now().Add(noAvailablePeersWaitPeriod)
		return
	}
	var peerID types.NodeID
	for peerID = range r.availablePeers {
		break
	}
	if r.isLegacyPeer(peerID) {
		r.pexCh.Out <- p2p.Envelope{
			To:      peerID,
			Message: &protop2p.PexRequest{},
		}
	} else {
		r.pexCh.Out <- p2p.Envelope{
			To:      peerID,
			Message: &protop2p.PexRequestV2{},
		}
	}
	delete(r.availablePeers, peerID)
	r.requestsSent[peerID] = struct{}{}
	r.calculateNextRequestTime()
	r.Logger.Debug("peer request sent", "next_request_time", r.nextRequestTime)
}