consensus: remove OnNewBlock(), use Blockchain subscription

Get new blocks directly from the Blockchain. It may lead to some duplications
(as we'll also receive our own blocks), but at the same time it's more
correct, because technically we can also get blocks via other means besides
network server like RPC (submitblock call). And it simplifies network server
at the same time.
This commit is contained in:
Roman Khimov 2020-05-07 22:45:06 +03:00
parent 1ac4f8528d
commit dd8bcfae47
2 changed files with 12 additions and 28 deletions

View file

@ -41,9 +41,6 @@ type Service interface {
OnTransaction(tx *transaction.Transaction) OnTransaction(tx *transaction.Transaction)
// GetPayload returns Payload with specified hash if it is present in the local cache. // GetPayload returns Payload with specified hash if it is present in the local cache.
GetPayload(h util.Uint256) *Payload GetPayload(h util.Uint256) *Payload
// OnNewBlock notifies consensus service that there is a new block in
// the chain (without explicitly passing it to the service).
OnNewBlock()
} }
type service struct { type service struct {
@ -61,7 +58,7 @@ type service struct {
transactions chan *transaction.Transaction transactions chan *transaction.Transaction
// blockEvents is used to pass a new block event to the consensus // blockEvents is used to pass a new block event to the consensus
// process. // process.
blockEvents chan struct{} blockEvents chan *coreb.Block
lastProposal []util.Uint256 lastProposal []util.Uint256
wallet *wallet.Wallet wallet *wallet.Wallet
} }
@ -106,7 +103,7 @@ func NewService(cfg Config) (Service, error) {
messages: make(chan Payload, 100), messages: make(chan Payload, 100),
transactions: make(chan *transaction.Transaction, 100), transactions: make(chan *transaction.Transaction, 100),
blockEvents: make(chan struct{}, 1), blockEvents: make(chan *coreb.Block, 1),
} }
if cfg.Wallet == nil { if cfg.Wallet == nil {
@ -163,7 +160,7 @@ var (
func (s *service) Start() { func (s *service) Start() {
s.dbft.Start() s.dbft.Start()
s.Chain.SubscribeForBlocks(s.blockEvents)
go s.eventLoop() go s.eventLoop()
} }
@ -203,11 +200,14 @@ func (s *service) eventLoop() {
s.dbft.OnReceive(&msg) s.dbft.OnReceive(&msg)
case tx := <-s.transactions: case tx := <-s.transactions:
s.dbft.OnTransaction(tx) s.dbft.OnTransaction(tx)
case <-s.blockEvents: case b := <-s.blockEvents:
s.log.Debug("new block in the chain", // We also receive our own blocks here, so check for index.
zap.Uint32("dbft index", s.dbft.BlockIndex), if b.Index >= s.dbft.BlockIndex {
zap.Uint32("chain index", s.Chain.BlockHeight())) s.log.Debug("new block in the chain",
s.dbft.InitializeConsensus(0) zap.Uint32("dbft index", s.dbft.BlockIndex),
zap.Uint32("chain index", s.Chain.BlockHeight()))
s.dbft.InitializeConsensus(0)
}
} }
} }
} }
@ -287,20 +287,6 @@ func (s *service) OnTransaction(tx *transaction.Transaction) {
} }
} }
// OnNewBlock notifies consensus process that there is a new block in the chain
// and dbft should probably be reinitialized.
func (s *service) OnNewBlock() {
if s.dbft != nil {
// If there is something in the queue already, the second
// consecutive event doesn't make much sense (reinitializing
// dbft twice doesn't improve it in any way).
select {
case s.blockEvents <- struct{}{}:
default:
}
}
}
// GetPayload returns payload stored in cache. // GetPayload returns payload stored in cache.
func (s *service) GetPayload(h util.Uint256) *Payload { func (s *service) GetPayload(h util.Uint256) *Payload {
p := s.cache.Get(h) p := s.cache.Get(h)

View file

@ -102,9 +102,7 @@ func NewServer(config ServerConfig, chain core.Blockchainer, log *zap.Logger) (*
transactions: make(chan *transaction.Transaction, 64), transactions: make(chan *transaction.Transaction, 64),
} }
s.bQueue = newBlockQueue(maxBlockBatch, chain, log, func(b *block.Block) { s.bQueue = newBlockQueue(maxBlockBatch, chain, log, func(b *block.Block) {
if s.consensusStarted.Load() { if !s.consensusStarted.Load() {
s.consensus.OnNewBlock()
} else {
s.tryStartConsensus() s.tryStartConsensus()
} }
s.relayBlock(b) s.relayBlock(b)