2019-09-25 16:54:31 +00:00
|
|
|
package network
|
|
|
|
|
|
|
|
import (
|
2021-10-31 09:23:07 +00:00
|
|
|
"sync"
|
|
|
|
|
2020-03-03 14:21:42 +00:00
|
|
|
"github.com/nspcc-dev/neo-go/pkg/core/block"
|
2021-07-30 13:57:42 +00:00
|
|
|
"go.uber.org/atomic"
|
2019-12-30 07:43:05 +00:00
|
|
|
"go.uber.org/zap"
|
2019-09-25 16:54:31 +00:00
|
|
|
)
|
|
|
|
|
2022-01-14 01:09:54 +00:00
|
|
|
// Blockqueuer is the interface for block queue.
|
|
|
|
type Blockqueuer interface {
|
|
|
|
AddBlock(block *block.Block) error
|
|
|
|
AddHeaders(...*block.Header) error
|
|
|
|
BlockHeight() uint32
|
|
|
|
}
|
|
|
|
|
2019-09-25 16:54:31 +00:00
|
|
|
type blockQueue struct {
|
2019-12-30 07:43:05 +00:00
|
|
|
log *zap.Logger
|
2022-01-17 21:04:41 +00:00
|
|
|
queueLock sync.RWMutex
|
2021-10-31 09:23:07 +00:00
|
|
|
queue []*block.Block
|
2022-01-17 21:04:41 +00:00
|
|
|
lastQ uint32
|
2019-09-25 16:54:31 +00:00
|
|
|
checkBlocks chan struct{}
|
2022-01-14 01:09:54 +00:00
|
|
|
chain Blockqueuer
|
2020-02-04 16:32:29 +00:00
|
|
|
relayF func(*block.Block)
|
2021-07-30 13:57:42 +00:00
|
|
|
discarded *atomic.Bool
|
2021-10-31 09:23:07 +00:00
|
|
|
len int
|
2019-09-25 16:54:31 +00:00
|
|
|
}
|
|
|
|
|
2020-09-02 12:00:53 +00:00
|
|
|
const (
|
|
|
|
// blockCacheSize is the amount of blocks above current height
|
|
|
|
// which are stored in queue.
|
|
|
|
blockCacheSize = 2000
|
|
|
|
)
|
|
|
|
|
2022-01-17 21:01:26 +00:00
|
|
|
func indexToPosition(i uint32) int {
|
|
|
|
return int(i) % blockCacheSize
|
|
|
|
}
|
|
|
|
|
2022-01-14 01:09:54 +00:00
|
|
|
func newBlockQueue(capacity int, bc Blockqueuer, log *zap.Logger, relayer func(*block.Block)) *blockQueue {
|
2019-12-30 07:43:05 +00:00
|
|
|
if log == nil {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2019-09-25 16:54:31 +00:00
|
|
|
return &blockQueue{
|
2019-12-30 07:43:05 +00:00
|
|
|
log: log,
|
2021-10-31 09:23:07 +00:00
|
|
|
queue: make([]*block.Block, blockCacheSize),
|
2019-09-25 16:54:31 +00:00
|
|
|
checkBlocks: make(chan struct{}, 1),
|
|
|
|
chain: bc,
|
2020-02-04 16:32:29 +00:00
|
|
|
relayF: relayer,
|
2021-07-30 13:57:42 +00:00
|
|
|
discarded: atomic.NewBool(false),
|
2019-09-25 16:54:31 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (bq *blockQueue) run() {
|
2021-10-31 09:23:07 +00:00
|
|
|
var lastHeight = bq.chain.BlockHeight()
|
2019-09-25 16:54:31 +00:00
|
|
|
for {
|
|
|
|
_, ok := <-bq.checkBlocks
|
|
|
|
if !ok {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
for {
|
2021-10-31 09:23:07 +00:00
|
|
|
h := bq.chain.BlockHeight()
|
2022-01-17 21:01:26 +00:00
|
|
|
pos := indexToPosition(h + 1)
|
2021-10-31 09:23:07 +00:00
|
|
|
bq.queueLock.Lock()
|
|
|
|
b := bq.queue[pos]
|
|
|
|
// The chain moved forward using blocks from other sources (consensus).
|
|
|
|
for i := lastHeight; i < h; i++ {
|
2022-01-17 21:01:26 +00:00
|
|
|
old := indexToPosition(i + 1)
|
2021-10-31 09:23:07 +00:00
|
|
|
if bq.queue[old] != nil && bq.queue[old].Index == i {
|
|
|
|
bq.len--
|
|
|
|
bq.queue[old] = nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
bq.queueLock.Unlock()
|
|
|
|
lastHeight = h
|
|
|
|
if b == nil {
|
2019-09-25 16:54:31 +00:00
|
|
|
break
|
|
|
|
}
|
2021-10-31 09:23:07 +00:00
|
|
|
|
|
|
|
err := bq.chain.AddBlock(b)
|
|
|
|
if err != nil {
|
|
|
|
// The block might already be added by consensus.
|
|
|
|
if bq.chain.BlockHeight() < b.Index {
|
|
|
|
bq.log.Warn("blockQueue: failed adding block into the blockchain",
|
|
|
|
zap.String("error", err.Error()),
|
|
|
|
zap.Uint32("blockHeight", bq.chain.BlockHeight()),
|
|
|
|
zap.Uint32("nextIndex", b.Index))
|
2019-09-25 16:54:31 +00:00
|
|
|
}
|
2021-10-31 09:23:07 +00:00
|
|
|
} else if bq.relayF != nil {
|
|
|
|
bq.relayF(b)
|
2019-09-25 16:54:31 +00:00
|
|
|
}
|
2021-10-31 09:23:07 +00:00
|
|
|
bq.queueLock.Lock()
|
|
|
|
bq.len--
|
|
|
|
l := bq.len
|
|
|
|
if bq.queue[pos] == b {
|
|
|
|
bq.queue[pos] = nil
|
|
|
|
}
|
|
|
|
bq.queueLock.Unlock()
|
|
|
|
updateBlockQueueLenMetric(l)
|
2019-09-25 16:54:31 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-01-14 12:32:07 +00:00
|
|
|
func (bq *blockQueue) putBlock(block *block.Block) error {
|
2020-09-02 12:00:53 +00:00
|
|
|
h := bq.chain.BlockHeight()
|
2021-10-31 09:23:07 +00:00
|
|
|
bq.queueLock.Lock()
|
2022-04-22 11:45:37 +00:00
|
|
|
defer bq.queueLock.Unlock()
|
|
|
|
if bq.discarded.Load() {
|
|
|
|
return nil
|
|
|
|
}
|
2020-09-02 12:00:53 +00:00
|
|
|
if block.Index <= h || h+blockCacheSize < block.Index {
|
2019-09-25 16:54:31 +00:00
|
|
|
// can easily happen when fetching the same blocks from
|
|
|
|
// different peers, thus not considered as error
|
|
|
|
return nil
|
|
|
|
}
|
2022-01-17 21:01:26 +00:00
|
|
|
pos := indexToPosition(block.Index)
|
2021-10-31 09:23:07 +00:00
|
|
|
// If we already have it, keep the old block, throw away new one.
|
|
|
|
if bq.queue[pos] == nil || bq.queue[pos].Index < block.Index {
|
|
|
|
bq.len++
|
|
|
|
bq.queue[pos] = block
|
2022-01-17 21:04:41 +00:00
|
|
|
for pos < blockCacheSize && bq.queue[pos] != nil && bq.lastQ+1 == bq.queue[pos].Index {
|
|
|
|
bq.lastQ = bq.queue[pos].Index
|
|
|
|
pos++
|
|
|
|
}
|
2021-10-31 09:23:07 +00:00
|
|
|
}
|
|
|
|
l := bq.len
|
2019-10-29 17:51:17 +00:00
|
|
|
// update metrics
|
2021-10-31 09:23:07 +00:00
|
|
|
updateBlockQueueLenMetric(l)
|
2019-09-25 16:54:31 +00:00
|
|
|
select {
|
|
|
|
case bq.checkBlocks <- struct{}{}:
|
|
|
|
// ok, signalled to goroutine processing queue
|
|
|
|
default:
|
|
|
|
// it's already busy processing blocks
|
|
|
|
}
|
2021-10-31 09:23:07 +00:00
|
|
|
return nil
|
2019-09-25 16:54:31 +00:00
|
|
|
}
|
|
|
|
|
2022-01-17 21:04:41 +00:00
|
|
|
func (bq *blockQueue) lastQueued() uint32 {
|
|
|
|
bq.queueLock.RLock()
|
|
|
|
defer bq.queueLock.RUnlock()
|
|
|
|
return bq.lastQ
|
|
|
|
}
|
|
|
|
|
2019-09-25 16:54:31 +00:00
|
|
|
func (bq *blockQueue) discard() {
|
2021-07-30 13:57:42 +00:00
|
|
|
if bq.discarded.CAS(false, true) {
|
2021-10-31 09:23:07 +00:00
|
|
|
bq.queueLock.Lock()
|
2022-04-22 11:45:37 +00:00
|
|
|
close(bq.checkBlocks)
|
2021-10-31 09:23:07 +00:00
|
|
|
// Technically we could bq.queue = nil, but this would cost
|
|
|
|
// another if in run().
|
|
|
|
for i := 0; i < len(bq.queue); i++ {
|
|
|
|
bq.queue[i] = nil
|
|
|
|
}
|
|
|
|
bq.len = 0
|
|
|
|
bq.queueLock.Unlock()
|
2021-07-30 13:57:42 +00:00
|
|
|
}
|
2019-09-25 16:54:31 +00:00
|
|
|
}
|