<commit_before>package block

import (
	"math/rand"
	"sync/atomic"
	"time"

	. "github.com/tendermint/tendermint/common"
)

const (
	maxOutstandingRequestsPerPeer = 10
	eventsChannelCapacity         = 100
	requestTimeoutSeconds         = 10
	maxTries                      = 3
	requestIntervalMS             = 500
	requestBatchSize              = 50
	maxPendingRequests            = 50
	maxTotalRequests              = 100
	maxPeersPerRequest            = 1
)

type BlockRequest struct {
	Height uint
	PeerId string
}

type BlockPool struct {
	peers      map[string]*bpPeer
	blockInfos map[uint]*bpBlockInfo
	height     uint  // the lowest key in blockInfos.
	started    int32 // atomic
	stopped    int32 // atomic
	numPending int32
	numTotal   int32
	eventsCh   chan interface{}    // internal events.
	requestsCh chan<- BlockRequest // output of new requests to make.
	timeoutsCh chan<- string       // output of peers that timed out.
	blocksCh   chan<- *Block       // output of ordered blocks.
	repeater   *RepeatTimer        // for requesting more bocks.
	quit       chan struct{}
}

func NewBlockPool(start uint, timeoutsCh chan<- string, requestsCh chan<- BlockRequest, blocksCh chan<- *Block) *BlockPool {
	return &BlockPool{
		peers:      make(map[string]*bpPeer),
		blockInfos: make(map[uint]*bpBlockInfo),
		height:     start,
		started:    0,
		stopped:    0,
		numPending: 0,
		numTotal:   0,
		quit:       make(chan struct{}),

		eventsCh:   make(chan interface{}, eventsChannelCapacity),
		requestsCh: requestsCh,
		timeoutsCh: timeoutsCh,
		blocksCh:   blocksCh,
		repeater:   NewRepeatTimer("", requestIntervalMS*time.Millisecond),
	}
}

func (bp *BlockPool) Start() {
	if atomic.CompareAndSwapInt32(&bp.started, 0, 1) {
		log.Info("Starting BlockPool")
		go bp.run()
	}
}

func (bp *BlockPool) Stop() {
	if atomic.CompareAndSwapInt32(&bp.stopped, 0, 1) {
		log.Info("Stopping BlockPool")
		close(bp.quit)
		close(bp.eventsCh)
		close(bp.requestsCh)
		close(bp.timeoutsCh)
		close(bp.blocksCh)
		bp.repeater.Stop()
	}
}

// AddBlock should be called when a block is received.
func (bp *BlockPool) AddBlock(block *Block, peerId string) {
	bp.eventsCh <- bpBlockResponse{block, peerId}
}

func (bp *BlockPool) SetPeerStatus(peerId string, height uint) {
	bp.eventsCh <- bpPeerStatus{peerId, height}
}

// Runs in a goroutine and processes messages.
func (bp *BlockPool) run() {
FOR_LOOP:
	for {
		select {
		case msg := <-bp.eventsCh:
			bp.handleEvent(msg)
		case <-bp.repeater.Ch:
			bp.makeMoreBlockInfos()
			bp.requestBlocksFromRandomPeers(10)
		case <-bp.quit:
			break FOR_LOOP
		}
	}
}

func (bp *BlockPool) handleEvent(event_ interface{}) {
	switch event := event_.(type) {
	case bpBlockResponse:
		peer := bp.peers[event.peerId]
		blockInfo := bp.blockInfos[event.block.Height]
		if blockInfo == nil {
			// block was unwanted.
			if peer != nil {
				peer.bad++
			}
		} else {
			// block was wanted.
			if peer != nil {
				peer.good++
			}
			delete(peer.requests, event.block.Height)
			if blockInfo.block == nil {
				// peer is the first to give it to us.
				blockInfo.block = event.block
				blockInfo.blockBy = peer.id
				bp.numPending--
				if event.block.Height == bp.height {
					go bp.pushBlocksFromStart()
				}
			}
		}
	case bpPeerStatus: // updated or new status from peer
		// request blocks if possible.
		peer := bp.peers[event.peerId]
		if peer == nil {
			peer = bpNewPeer(event.peerId, event.height)
			bp.peers[peer.id] = peer
		}
		bp.requestBlocksFromPeer(peer)
	case bpRequestTimeout: // unconditional timeout for each peer's request.
		peer := bp.peers[event.peerId]
		if peer == nil {
			// cleanup was already handled.
			return
		}
		height := event.height
		request := peer.requests[height]
		if request == nil || request.block != nil {
			// the request was fulfilled by some peer or this peer.
			return
		}

		// A request for peer timed out.
		peer.bad++
		if request.tries < maxTries {
			log.Warn("Timeout: Trying again.", "tries", request.tries, "peerId", peer.id)
			// try again.
			select {
			case bp.requestsCh <- BlockRequest{height, peer.id}:
				request.startAndTimeoutTo(bp.eventsCh) // also bumps request.tries
			default:
				// The request cannot be made because requestCh is full.
				// Just delete the request.
				delete(peer.requests, height)
			}
		} else {
			log.Warn("Timeout: Deleting request")
			// delete the request.
			delete(peer.requests, height)
			blockInfo := bp.blockInfos[height]
			if blockInfo != nil {
				delete(blockInfo.requests, peer.id)
			}
			select {
			case bp.timeoutsCh <- peer.id:
			default:
			}

		}
	}
}

func (bp *BlockPool) requestBlocksFromRandomPeers(maxPeers int) {
	chosen := bp.pickAvailablePeers(maxPeers)
	log.Debug("requestBlocksFromRandomPeers", "chosen", len(chosen))
	for _, peer := range chosen {
		bp.requestBlocksFromPeer(peer)
	}
}
<commit_msg>added NOTE on inefficiency.<commit_after>package block

import (
	"math/rand"
	"sync/atomic"
	"time"

	. "github.com/tendermint/tendermint/common"
)

const (
	maxOutstandingRequestsPerPeer = 10
	eventsChannelCapacity         = 100
	requestTimeoutSeconds         = 10
	maxTries                      = 3
	requestIntervalMS             = 500
	requestBatchSize              = 50
	maxPendingRequests            = 50
	maxTotalRequests              = 100
	maxPeersPerRequest            = 1
)

type BlockRequest struct {
	Height uint
	PeerId string
}

type BlockPool struct {
	peers      map[string]*bpPeer
	blockInfos map[uint]*bpBlockInfo
	height     uint  // the lowest key in blockInfos.
	started    int32 // atomic
	stopped    int32 // atomic
	numPending int32
	numTotal   int32
	eventsCh   chan interface{}    // internal events.
	requestsCh chan<- BlockRequest // output of new requests to make.
	timeoutsCh chan<- string       // output of peers that timed out.
	blocksCh   chan<- *Block       // output of ordered blocks.
	repeater   *RepeatTimer        // for requesting more bocks.
	quit       chan struct{}
}

func NewBlockPool(start uint, timeoutsCh chan<- string, requestsCh chan<- BlockRequest, blocksCh chan<- *Block) *BlockPool {
	return &BlockPool{
		peers:      make(map[string]*bpPeer),
		blockInfos: make(map[uint]*bpBlockInfo),
		height:     start,
		started:    0,
		stopped:    0,
		numPending: 0,
		numTotal:   0,
		quit:       make(chan struct{}),

		eventsCh:   make(chan interface{}, eventsChannelCapacity),
		requestsCh: requestsCh,
		timeoutsCh: timeoutsCh,
		blocksCh:   blocksCh,
		repeater:   NewRepeatTimer("", requestIntervalMS*time.Millisecond),
	}
}

func (bp *BlockPool) Start() {
	if atomic.CompareAndSwapInt32(&bp.started, 0, 1) {
		log.Info("Starting BlockPool")
		go bp.run()
	}
}

func (bp *BlockPool) Stop() {
	if atomic.CompareAndSwapInt32(&bp.stopped, 0, 1) {
		log.Info("Stopping BlockPool")
		close(bp.quit)
		close(bp.eventsCh)
		close(bp.requestsCh)
		close(bp.timeoutsCh)
		close(bp.blocksCh)
		bp.repeater.Stop()
	}
}

// AddBlock should be called when a block is received.
func (bp *BlockPool) AddBlock(block *Block, peerId string) {
	bp.eventsCh <- bpBlockResponse{block, peerId}
}

func (bp *BlockPool) SetPeerStatus(peerId string, height uint) {
	bp.eventsCh <- bpPeerStatus{peerId, height}
}

// Runs in a goroutine and processes messages.
func (bp *BlockPool) run() {
FOR_LOOP:
	for {
		select {
		case msg := <-bp.eventsCh:
			bp.handleEvent(msg)
		case <-bp.repeater.Ch:
			bp.makeMoreBlockInfos()
			bp.requestBlocksFromRandomPeers(10)
		case <-bp.quit:
			break FOR_LOOP
		}
	}
}

func (bp *BlockPool) handleEvent(event_ interface{}) {
	switch event := event_.(type) {
	case bpBlockResponse:
		peer := bp.peers[event.peerId]
		blockInfo := bp.blockInfos[event.block.Height]
		if blockInfo == nil {
			// block was unwanted.
			if peer != nil {
				peer.bad++
			}
		} else {
			// block was wanted.
			if peer != nil {
				peer.good++
			}
			delete(peer.requests, event.block.Height)
			if blockInfo.block == nil {
				// peer is the first to give it to us.
				blockInfo.block = event.block
				blockInfo.blockBy = peer.id
				bp.numPending--
				if event.block.Height == bp.height {
					go bp.pushBlocksFromStart()
				}
			}
		}
	case bpPeerStatus: // updated or new status from peer
		// request blocks if possible.
		peer := bp.peers[event.peerId]
		if peer == nil {
			peer = bpNewPeer(event.peerId, event.height)
			bp.peers[peer.id] = peer
		}
		bp.requestBlocksFromPeer(peer)
	case bpRequestTimeout: // unconditional timeout for each peer's request.
		peer := bp.peers[event.peerId]
		if peer == nil {
			// cleanup was already handled.
			return
		}
		height := event.height
		request := peer.requests[height]
		if request == nil || request.block != nil {
			// the request was fulfilled by some peer or this peer.
			return
		}

		// A request for peer timed out.
		peer.bad++
		if request.tries < maxTries {
			log.Warn("Timeout: Trying again.", "tries", request.tries, "peerId", peer.id)
			// try again.
			select {
			case bp.requestsCh <- BlockRequest{height, peer.id}:
				request.startAndTimeoutTo(bp.eventsCh) // also bumps request.tries
			default:
				// The request cannot be made because requestCh is full.
				// Just delete the request.
				delete(peer.requests, height)
			}
		} else {
			log.Warn("Timeout: Deleting request")
			// delete the request.
			delete(peer.requests, height)
			blockInfo := bp.blockInfos[height]
			if blockInfo != nil {
				delete(blockInfo.requests, peer.id)
			}
			select {
			case bp.timeoutsCh <- peer.id:
			default:
			}

		}
	}
}

// NOTE: This function is sufficient, but we should find pending blocks
// and sample the peers in one go rather than the current O(n^2) impl.
func (bp *BlockPool) requestBlocksFromRandomPeers(maxPeers int) {
	chosen := bp.pickAvailablePeers(maxPeers)
	log.Debug("requestBlocksFromRandomPeers", "chosen", len(chosen))
	for _, peer := range chosen {
		bp.requestBlocksFromPeer(peer)
	}
}

{'fraction_non_alphanumeric': 0.11465160834929919, 'fraction_numerical': 0.008369466572552184, 'mean_word_length': 2.9927536231884058, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1400063', 'n_tokens_mistral': 3792, 'n_tokens_neox': 3298, 'n_words': 871}