// Copyright 2018 The dexon-consensus Authors
// This file is part of the dexon-consensus-core library.
//
// The dexon-consensus-core library is free software: you can redistribute it
// and/or modify it under the terms of the GNU Lesser General Public License as
// published by the Free Software Foundation, either version 3 of the License,
// or (at your option) any later version.
//
// The dexon-consensus-core library is distributed in the hope that it will be
// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
// General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the dexon-consensus-core library. If not, see
// <http://www.gnu.org/licenses/>.
package syncer
import (
"context"
"time"
"github.com/dexon-foundation/dexon-consensus/common"
"github.com/dexon-foundation/dexon-consensus/core"
"github.com/dexon-foundation/dexon-consensus/core/types"
"github.com/dexon-foundation/dexon-consensus/core/utils"
)
// Struct agreement implements struct of BA (Byzantine Agreement) protocol
// needed in syncer, which only receives agreement results.
type agreement struct {
cache *utils.NodeSetCache
inputChan chan interface{}
outputChan chan<- *types.Block
pullChan chan<- common.Hash
blocks map[types.Position]map[common.Hash]*types.Block
agreementResults map[common.Hash]struct{}
latestCRSRound uint64
pendings map[uint64]map[common.Hash]*types.AgreementResult
logger common.Logger
confirmedBlocks map[common.Hash]struct{}
ctx context.Context
ctxCancel context.CancelFunc
}
// newAgreement creates a new agreement instance.
func newAgreement(ch chan<- *types.Block, pullChan chan<- common.Hash,
cache *utils.NodeSetCache, logger common.Logger) *agreement {
a := &agreement{
cache: cache,
inputChan: make(chan interface{}, 1000),
outputChan: ch,
pullChan: pullChan,
blocks: make(map[types.Position]map[common.Hash]*types.Block),
agreementResults: make(map[common.Hash]struct{}),
logger: logger,
pendings: make(
map[uint64]map[common.Hash]*types.AgreementResult),
confirmedBlocks: make(map[common.Hash]struct{}),
}
a.ctx, a.ctxCancel = context.WithCancel(context.Background())
return a
}
// run starts the agreement, this does not start a new routine, go a new
// routine explicitly in the caller.
func (a *agreement) run() {
defer a.ctxCancel()
for {
select {
case val, ok := <-a.inputChan:
if !ok {
// InputChan is closed by network when network ends.
return
}
switch v := val.(type) {
case *types.Block:
a.processBlock(v)
case *types.AgreementResult:
a.processAgreementResult(v)
case uint64:
a.processNewCRS(v)
}
}
}
}
func (a *agreement) processBlock(b *types.Block) {
if _, exist := a.confirmedBlocks[b.Hash]; exist {
return
}
if _, exist := a.agreementResults[b.Hash]; exist {
a.confirm(b)
} else {
if _, exist := a.blocks[b.Position]; !exist {
a.blocks[b.Position] = make(map[common.Hash]*types.Block)
}
a.blocks[b.Position][b.Hash] = b
}
}
func (a *agreement) processAgreementResult(r *types.AgreementResult) {
// Cache those results that CRS is not ready yet.
if _, exists := a.confirmedBlocks[r.BlockHash]; exists {
a.logger.Trace("agreement result already confirmed", "result", r)
return
}
if r.Position.Round > a.latestCRSRound {
pendingsForRound, exists := a.pendings[r.Position.Round]
if !exists {
pendingsForRound = make(map[common.Hash]*types.AgreementResult)
a.pendings[r.Position.Round] = pendingsForRound
}
pendingsForRound[r.BlockHash] = r
a.logger.Trace("agreement result cached", "result", r)
return
}
if err := core.VerifyAgreementResult(r, a.cache); err != nil {
a.logger.Error("agreement result verification failed",
"result", r,
"error", err)
return
}
if r.IsEmptyBlock {
b := &types.Block{
Position: r.Position,
}
// Empty blocks should be confirmed directly, they won't be sent over
// the wire.
a.confirm(b)
return
}
if bs, exist := a.blocks[r.Position]; exist {
if b, exist := bs[r.BlockHash]; exist {
a.confirm(b)
return
}
}
a.agreementResults[r.BlockHash] = struct{}{}
loop:
for {
select {
case a.pullChan <- r.BlockHash:
break loop
case <-a.ctx.Done():
a.logger.Error("pull request is not sent",
"position", &r.Position,
"hash", r.BlockHash.String()[:6])
return
case <-time.After(500 * time.Millisecond):
a.logger.Debug("pull request is unable to send",
"position", &r.Position,
"hash", r.BlockHash.String()[:6])
}
}
}
func (a *agreement) processNewCRS(round uint64) {
if round <= a.latestCRSRound {
return
}
prevRound := a.latestCRSRound + 1
a.latestCRSRound = round
// Verify all pending results.
for r := prevRound; r <= a.latestCRSRound; r++ {
pendingsForRound := a.pendings[r]
if pendingsForRound == nil {
continue
}
delete(a.pendings, r)
for _, res := range pendingsForRound {
if err := core.VerifyAgreementResult(res, a.cache); err != nil {
a.logger.Error("invalid agreement result",
"result", res,
"error", err)
continue
}
a.logger.Error("flush agreement result", "result", res)
a.processAgreementResult(res)
break
}
}
}
// confirm notifies consensus the confirmation of a block in BA.
func (a *agreement) confirm(b *types.Block) {
if _, exist := a.confirmedBlocks[b.Hash]; !exist {
delete(a.blocks, b.Position)
delete(a.agreementResults, b.Hash)
loop:
for {
select {
case a.outputChan <- b:
break loop
case <-a.ctx.Done():
a.logger.Error("confirmed block is not sent", "block", b)
return
case <-time.After(500 * time.Millisecond):
a.logger.Debug("agreement output channel is full", "block", b)
}
}
a.confirmedBlocks[b.Hash] = struct{}{}
}
}