peer.go

package v1

import (
	"fmt"
	"math"
	"time"

	flow "github.com/tendermint/tendermint/libs/flowrate"
	"github.com/tendermint/tendermint/libs/log"
	"github.com/tendermint/tendermint/p2p"
	"github.com/tendermint/tendermint/types"
)

//--------
// Peer

// BpPeerParams stores the peer parameters that are used when creating a peer.
type BpPeerParams struct {
	timeout     time.Duration
	minRecvRate int64
	sampleRate  time.Duration
	windowSize  time.Duration
}

// BpPeer is the datastructure associated with a fast sync peer.
type BpPeer struct {
	logger log.Logger
	ID     p2p.ID

	Height                  int64                  // the peer reported height
	NumPendingBlockRequests int                    // number of requests still waiting for block responses
	blocks                  map[int64]*types.Block // blocks received or expected to be received from this peer
	blockResponseTimer      *time.Timer
	recvMonitor             *flow.Monitor
	params                  *BpPeerParams // parameters for timer and monitor

	onErr func(err error, peerID p2p.ID) // function to call on error
}

// NewBpPeer creates a new peer.
func NewBpPeer(
	peerID p2p.ID, height int64, onErr func(err error, peerID p2p.ID), params *BpPeerParams) *BpPeer {

	if params == nil {
		params = BpPeerDefaultParams()
	}
	return &BpPeer{
		ID:     peerID,
		Height: height,
		blocks: make(map[int64]*types.Block, maxRequestsPerPeer),
		logger: log.NewNopLogger(),
		onErr:  onErr,
		params: params,
	}
}

// String returns a string representation of a peer.
func (peer *BpPeer) String() string {
	return fmt.Sprintf("peer: %v height: %v pending: %v", peer.ID, peer.Height, peer.NumPendingBlockRequests)
}

// SetLogger sets the logger of the peer.
func (peer *BpPeer) SetLogger(l log.Logger) {
	peer.logger = l
}

// Cleanup performs cleanup of the peer, removes blocks, requests, stops timer and monitor.
func (peer *BpPeer) Cleanup() {
	if peer.blockResponseTimer != nil {
		peer.blockResponseTimer.Stop()
	}
	if peer.NumPendingBlockRequests != 0 {
		peer.logger.Info("peer with pending requests is being cleaned", "peer", peer.ID)
	}
	if len(peer.blocks)-peer.NumPendingBlockRequests != 0 {
		peer.logger.Info("peer with pending blocks is being cleaned", "peer", peer.ID)
	}
	for h := range peer.blocks {
		delete(peer.blocks, h)
	}
	peer.NumPendingBlockRequests = 0
	peer.recvMonitor = nil
}

// BlockAtHeight returns the block at a given height if available and errMissingBlock otherwise.
func (peer *BpPeer) BlockAtHeight(height int64) (*types.Block, error) {
	block, ok := peer.blocks[height]
	if !ok {
		return nil, errMissingBlock
	}
	if block == nil {
		return nil, errMissingBlock
	}
	return peer.blocks[height], nil
}

// AddBlock adds a block at peer level. Block must be non-nil and recvSize a positive integer
// The peer must have a pending request for this block.
func (peer *BpPeer) AddBlock(block *types.Block, recvSize int) error {
	if block == nil || recvSize < 0 {
		panic("bad parameters")
	}
	existingBlock, ok := peer.blocks[block.Height]
	if !ok {
		peer.logger.Error("unsolicited block", "blockHeight", block.Height, "peer", peer.ID)
		return errMissingBlock
	}
	if existingBlock != nil {
		peer.logger.Error("already have a block for height", "height", block.Height)
		return errDuplicateBlock
	}
	if peer.NumPendingBlockRequests == 0 {
		panic("peer does not have pending requests")
	}
	peer.blocks[block.Height] = block
	peer.NumPendingBlockRequests--
	if peer.NumPendingBlockRequests == 0 {
		peer.stopMonitor()
		peer.stopBlockResponseTimer()
	} else {
		peer.recvMonitor.Update(recvSize)
		peer.resetBlockResponseTimer()
	}
	return nil
}

// RemoveBlock removes the block of given height
func (peer *BpPeer) RemoveBlock(height int64) {
	delete(peer.blocks, height)
}

// RequestSent records that a request was sent, and starts the peer timer and monitor if needed.
func (peer *BpPeer) RequestSent(height int64) {
	peer.blocks[height] = nil

	if peer.NumPendingBlockRequests == 0 {
		peer.startMonitor()
		peer.resetBlockResponseTimer()
	}
	peer.NumPendingBlockRequests++
}

// CheckRate verifies that the response rate of the peer is acceptable (higher than the minimum allowed).
func (peer *BpPeer) CheckRate() error {
	if peer.NumPendingBlockRequests == 0 {
		return nil
	}
	curRate := peer.recvMonitor.Status().CurRate
	// curRate can be 0 on start
	if curRate != 0 && curRate < peer.params.minRecvRate {
		err := errSlowPeer
		peer.logger.Error("SendTimeout", "peer", peer,
			"reason", err,
			"curRate", fmt.Sprintf("%d KB/s", curRate/1024),
			"minRate", fmt.Sprintf("%d KB/s", peer.params.minRecvRate/1024))
		return err
	}
	return nil
}

func (peer *BpPeer) onTimeout() {
	peer.onErr(errNoPeerResponse, peer.ID)
}

func (peer *BpPeer) stopMonitor() {
	peer.recvMonitor.Done()
	peer.recvMonitor = nil
}

func (peer *BpPeer) startMonitor() {
	peer.recvMonitor = flow.New(peer.params.sampleRate, peer.params.windowSize)
	initialValue := float64(peer.params.minRecvRate) * math.E
	peer.recvMonitor.SetREMA(initialValue)
}

func (peer *BpPeer) resetBlockResponseTimer() {
	if peer.blockResponseTimer == nil {
		peer.blockResponseTimer = time.AfterFunc(peer.params.timeout, peer.onTimeout)
	} else {
		peer.blockResponseTimer.Reset(peer.params.timeout)
	}
}

func (peer *BpPeer) stopBlockResponseTimer() bool {
	if peer.blockResponseTimer == nil {
		return false
	}
	return peer.blockResponseTimer.Stop()
}

// BpPeerDefaultParams returns the default peer parameters.
func BpPeerDefaultParams() *BpPeerParams {
	return &BpPeerParams{
		// Timeout for a peer to respond to a block request.
		timeout: 15 * time.Second,

		// Minimum recv rate to ensure we're receiving blocks from a peer fast
		// enough. If a peer is not sending data at at least that rate, we
		// consider them to have timedout and we disconnect.
		//
		// Assuming a DSL connection (not a good choice) 128 Kbps (upload) ~ 15 KB/s,
		// sending data across atlantic ~ 7.5 KB/s.
		minRecvRate: int64(7680),

		// Monitor parameters
		sampleRate: time.Second,
		windowSize: 40 * time.Second,
	}
}