func (pool *BlockPool) pickIncrAvailablePeer(minHeight def.INT) *bpPeer {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	for _, peer := range pool.peers {
		if peer.didTimeout {
			pool.removePeer(peer.id)
			continue
		} else {
		}
		if peer.numPending >= maxPendingRequestsPerPeer {
			continue
		}
		if peer.height < minHeight {
			continue
		}
		peer.incrPending()
		return peer
	}
	return nil
}
Pick an available peer with at least the given minHeight.
If no peers are available, returns nil.
package blockchain
import (
	"math"
	"sync"
	"time"
	"go.uber.org/zap"
	pbtypes "github.com/annchain/annchain/angine/protos/types"
	agtypes "github.com/annchain/annchain/angine/types"
	. "github.com/annchain/annchain/module/lib/go-common"
	flow "github.com/annchain/annchain/module/lib/go-flowrate/flowrate"
	"github.com/annchain/annchain/module/xlib/def"
)
const (
	requestIntervalMS         = 250
	maxTotalRequesters        = 300
	maxPendingRequests        = maxTotalRequesters
	maxPendingRequestsPerPeer = 75
	minRecvRate               = 10240  
)
var peerTimeoutSeconds = time.Duration(15)  
type BlockPool struct {
	BaseService
	startTime time.Time
	mtx sync.Mutex
	requesters map[def.INT]*bpRequester
	height     def.INT  
	numPending int32    
	peers map[string]*bpPeer
	requestsCh chan<- BlockRequest
	timeoutsCh chan<- string
	logger *zap.Logger
}
func NewBlockPool(logger *zap.Logger, start def.INT, requestsCh chan<- BlockRequest, timeoutsCh chan<- string) *BlockPool {
	bp := &BlockPool{
		peers: make(map[string]*bpPeer),
		requesters: make(map[def.INT]*bpRequester),
		height:     start,
		numPending: 0,
		requestsCh: requestsCh,
		timeoutsCh: timeoutsCh,
		logger: logger,
	}
	bp.BaseService = *NewBaseService(logger, "BlockPool", bp)
	return bp
}
func (pool *BlockPool) OnStart() error {
	pool.BaseService.OnStart()
	go pool.makeRequestersRoutine()
	pool.startTime = time.Now()
	return nil
}
func (pool *BlockPool) OnStop() {
	pool.BaseService.OnStop()
}
func (pool *BlockPool) makeRequestersRoutine() {
	for {
		if !pool.IsRunning() {
			break
		}
		_, numPending, lenRequesters := pool.GetStatus()
		if numPending >= maxPendingRequests {
			time.Sleep(requestIntervalMS * time.Millisecond)
			pool.removeTimedoutPeers()
		} else if lenRequesters >= maxTotalRequesters {
			time.Sleep(requestIntervalMS * time.Millisecond)
			pool.removeTimedoutPeers()
		} else {
			pool.makeNextRequester()
		}
	}
}
func (pool *BlockPool) removeTimedoutPeers() {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	for _, peer := range pool.peers {
		if !peer.didTimeout && peer.numPending > 0 {
			curRate := peer.recvMonitor.Status().CurRate
			if curRate != 0 && curRate < minRecvRate {
				pool.sendTimeout(peer.id)
				pool.logger.Warn("SendTimeout", zap.String("peer", peer.id), zap.String("reason", "curRate too low"))
				peer.didTimeout = true
			}
		}
		if peer.didTimeout {
			pool.removePeer(peer.id)
		}
	}
}
func (pool *BlockPool) GetStatus() (height def.INT, numPending int32, lenRequesters int) {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	return pool.height, pool.numPending, len(pool.requesters)
}
func (pool *BlockPool) IsCaughtUp() bool {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	height := pool.height
	if len(pool.peers) == 0 {
		pool.logger.Debug("Blockpool has no peers")
		return false
	}
	var maxPeerHeight def.INT = 0
	for _, peer := range pool.peers {
		maxPeerHeight = MaxInt64(maxPeerHeight, peer.height)
	}
	isCaughtUp := (height > 0 || time.Now().Sub(pool.startTime) > 5*time.Second) && (maxPeerHeight == 0 || height >= maxPeerHeight)
	pool.logger.Info(Fmt("IsCaughtUp: %v", isCaughtUp), zap.Int64("height", height), zap.Int64("maxPeerHeight", maxPeerHeight))
	return isCaughtUp
}
func (pool *BlockPool) PeekTwoBlocks() (first *agtypes.BlockCache, second *agtypes.BlockCache) {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	if r := pool.requesters[pool.height]; r != nil {
		first = r.getBlock()
	}
	if r := pool.requesters[pool.height+1]; r != nil {
		second = r.getBlock()
	}
	return
}
func (pool *BlockPool) PopRequest() {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	if r := pool.requesters[pool.height]; r != nil {
		r.Stop()
		delete(pool.requesters, pool.height)
		pool.height++
	} else {
		PanicSanity(Fmt("Expected requester to pop, got nothing at height %v", pool.height))
	}
}
func (pool *BlockPool) RedoRequest(height def.INT) {
	pool.mtx.Lock()
	request := pool.requesters[height]
	pool.mtx.Unlock()
	if request.block == nil {
		PanicSanity("Expected block to be non-nil")
	}
	pool.RemovePeer(request.peerID)
}
func (pool *BlockPool) AddBlock(peerID string, block *pbtypes.Block, blockSize int) {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	requester := pool.requesters[block.Header.Height]
	if requester == nil {
		return
	}
	if requester.setBlock(block, peerID) {
		pool.numPending--
		peer := pool.peers[peerID]
		peer.decrPending(blockSize)
	} else {
	}
}
func (pool *BlockPool) SetPeerHeight(peerID string, height def.INT) {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	peer := pool.peers[peerID]
	if peer != nil {
		peer.height = height
	} else {
		peer = newBPPeer(pool, peerID, height)
		pool.peers[peerID] = peer
	}
}
func (pool *BlockPool) RemovePeer(peerID string) {
	pool.mtx.Lock()
	defer pool.mtx.Unlock()
	pool.removePeer(peerID)
}
func (pool *BlockPool) removePeer(peerID string) {
	for _, requester := range pool.requesters {
		if requester.getPeerID() == peerID {
			pool.numPending++
			go requester.redo()  
		}
	}
	delete(pool.peers, peerID)
}