// Copyright 2018 The dexon-consensus Authors
// This file is part of the dexon-consensus library.
//
// The dexon-consensus 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 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 library. If not, see
// <http://www.gnu.org/licenses/>.
package core
import (
"context"
"errors"
"math"
"sync"
"sync/atomic"
"time"
"github.com/dexon-foundation/dexon-consensus/common"
"github.com/dexon-foundation/dexon-consensus/core/types"
"github.com/dexon-foundation/dexon-consensus/core/utils"
)
// Errors returned from BA modules
var (
ErrPreviousRoundIsNotFinished = errors.New("previous round is not finished")
ErrRoundOutOfRange = errors.New("round out of range")
ErrInvalidBlock = errors.New("invalid block")
)
const maxResultCache = 100
// genValidLeader generate a validLeader function for agreement modules.
func genValidLeader(
mgr *agreementMgr) func(*types.Block) (bool, error) {
return func(block *types.Block) (bool, error) {
if block.Timestamp.After(time.Now()) {
return false, nil
}
if err := mgr.bcModule.sanityCheck(block); err != nil {
if err == ErrRetrySanityCheckLater {
return false, nil
}
return false, err
}
mgr.logger.Debug("Calling Application.VerifyBlock", "block", block)
switch mgr.app.VerifyBlock(block) {
case types.VerifyInvalidBlock:
return false, ErrInvalidBlock
case types.VerifyRetryLater:
return false, nil
default:
}
return true, nil
}
}
type agreementMgrConfig struct {
beginTime time.Time
roundInterval time.Duration
notarySetSize uint32
lambdaBA time.Duration
crs common.Hash
}
type baRoundSetting struct {
notarySet map[types.NodeID]struct{}
agr *agreement
recv *consensusBAReceiver
ticker Ticker
crs common.Hash
}
type agreementMgr struct {
// TODO(mission): unbound Consensus instance from this module.
con *Consensus
ID types.NodeID
app Application
gov Governance
network Network
logger common.Logger
cache *utils.NodeSetCache
signer *utils.Signer
bcModule *blockChain
ctx context.Context
lastEndTime time.Time
initRound uint64
configs []*agreementMgrConfig
baModule *agreement
processedBAResult map[types.Position]struct{}
voteFilter *utils.VoteFilter
waitGroup sync.WaitGroup
isRunning bool
lock sync.RWMutex
}
func newAgreementMgr(con *Consensus, initRound uint64,
initRoundBeginTime time.Time) *agreementMgr {
return &agreementMgr{
con: con,
ID: con.ID,
app: con.app,
gov: con.gov,
network: con.network,
logger: con.logger,
cache: con.nodeSetCache,
signer: con.signer,
bcModule: con.bcModule,
ctx: con.ctx,
initRound: initRound,
lastEndTime: initRoundBeginTime,
processedBAResult: make(map[types.Position]struct{}, maxResultCache),
}
}
func (mgr *agreementMgr) run() {
mgr.lock.Lock()
defer mgr.lock.Unlock()
if mgr.isRunning {
return
}
mgr.isRunning = true
mgr.waitGroup.Add(1)
go func() {
defer mgr.waitGroup.Done()
mgr.runBA(mgr.initRound)
}()
}
func (mgr *agreementMgr) getConfig(round uint64) *agreementMgrConfig {
mgr.lock.RLock()
defer mgr.lock.RUnlock()
if round < mgr.initRound {
panic(ErrRoundOutOfRange)
}
roundIndex := round - mgr.initRound
if roundIndex >= uint64(len(mgr.configs)) {
return nil
}
return mgr.configs[roundIndex]
}
func (mgr *agreementMgr) appendConfig(
round uint64, config *types.Config, crs common.Hash) (err error) {
mgr.lock.Lock()
defer mgr.lock.Unlock()
if round != uint64(len(mgr.configs))+mgr.initRound {
return ErrRoundNotIncreasing
}
newConfig := &agreementMgrConfig{
beginTime: mgr.lastEndTime,
roundInterval: config.RoundInterval,
notarySetSize: config.NotarySetSize,
lambdaBA: config.LambdaBA,
crs: crs,
}
mgr.configs = append(mgr.configs, newConfig)
mgr.lastEndTime = mgr.lastEndTime.Add(config.RoundInterval)
// Prepare modules.
if mgr.baModule != nil {
return nil
}
recv := &consensusBAReceiver{
consensus: mgr.con,
restartNotary: make(chan types.Position, 1),
roundValue: &atomic.Value{},
}
recv.roundValue.Store(uint64(0))
agrModule := newAgreement(
mgr.con.ID,
recv,
newLeaderSelector(genValidLeader(mgr), mgr.logger),
mgr.signer,
mgr.logger)
// Hacky way to initialize first notarySet.
nodes, err := mgr.cache.GetNodeSet(round)
if err != nil {
return err
}
agrModule.notarySet = nodes.GetSubSet(
int(config.NotarySetSize), types.NewNotarySetTarget(crs))
// Hacky way to make agreement module self contained.
recv.agreementModule = agrModule
mgr.baModule = agrModule
mgr.voteFilter = utils.NewVoteFilter()
return nil
}
func (mgr *agreementMgr) processVote(v *types.Vote) (err error) {
if v.Position.ChainID > 0 {
mgr.logger.Error("Process vote for unknown chain to BA",
"position", v.Position,
"initRound", mgr.initRound)
return utils.ErrInvalidChainID
}
if mgr.voteFilter.Filter(v) {
return nil
}
if err = mgr.baModule.processVote(v); err == nil {
mgr.baModule.updateFilter(mgr.voteFilter)
}
return
}
func (mgr *agreementMgr) processBlock(b *types.Block) error {
if b.Position.ChainID > 0 {
mgr.logger.Error("Process block for unknown chain to BA",
"position", b.Position,
"baRound", len(mgr.configs),
"initRound", mgr.initRound)
return utils.ErrInvalidChainID
}
return mgr.baModule.processBlock(b)
}
func (mgr *agreementMgr) touchAgreementResult(
result *types.AgreementResult) (first bool) {
// DO NOT LOCK THIS FUNCTION!!!!!!!! YOU WILL REGRET IT!!!!!
if _, exist := mgr.processedBAResult[result.Position]; !exist {
first = true
if len(mgr.processedBAResult) > maxResultCache {
for k := range mgr.processedBAResult {
// Randomly drop one element.
delete(mgr.processedBAResult, k)
break
}
}
mgr.processedBAResult[result.Position] = struct{}{}
}
return
}
func (mgr *agreementMgr) untouchAgreementResult(
result *types.AgreementResult) {
// DO NOT LOCK THIS FUNCTION!!!!!!!! YOU WILL REGRET IT!!!!!
delete(mgr.processedBAResult, result.Position)
}
func (mgr *agreementMgr) processAgreementResult(
result *types.AgreementResult) error {
if result.Position.ChainID > 0 {
mgr.logger.Error("Process unknown result for unknown chain to BA",
"position", result.Position,
"baRound", len(mgr.configs),
"initRound", mgr.initRound)
return utils.ErrInvalidChainID
}
aID := mgr.baModule.agreementID()
if isStop(aID) {
return nil
}
if result.Position == aID && !mgr.baModule.confirmed() {
mgr.logger.Info("Syncing BA", "position", result.Position)
for key := range result.Votes {
if err := mgr.baModule.processVote(&result.Votes[key]); err != nil {
return err
}
}
} else if result.Position.Newer(aID) {
mgr.logger.Info("Fast syncing BA", "position", result.Position)
nodes, err := mgr.cache.GetNodeSet(result.Position.Round)
if err != nil {
return err
}
mgr.logger.Debug("Calling Network.PullBlocks for fast syncing BA",
"hash", result.BlockHash)
mgr.network.PullBlocks(common.Hashes{result.BlockHash})
mgr.logger.Debug("Calling Governance.CRS", "round", result.Position.Round)
crs := utils.GetCRSWithPanic(mgr.gov, result.Position.Round, mgr.logger)
nIDs := nodes.GetSubSet(
int(utils.GetConfigWithPanic(
mgr.gov, result.Position.Round, mgr.logger).NotarySetSize),
types.NewNotarySetTarget(crs))
for key := range result.Votes {
if err := mgr.baModule.processVote(&result.Votes[key]); err != nil {
return err
}
}
leader, err := mgr.cache.GetLeaderNode(result.Position)
if err != nil {
return err
}
mgr.baModule.restart(nIDs, result.Position, leader, crs)
}
return nil
}
func (mgr *agreementMgr) stop() {
// Stop all running agreement modules.
func() {
mgr.lock.Lock()
defer mgr.lock.Unlock()
mgr.baModule.stop()
}()
// Block until all routines are done.
mgr.waitGroup.Wait()
}
func (mgr *agreementMgr) runBA(initRound uint64) {
// These are round based variables.
var (
currentRound uint64
nextRound = initRound
setting = baRoundSetting{
agr: mgr.baModule,
recv: mgr.baModule.data.recv.(*consensusBAReceiver),
}
roundBeginTime time.Time
roundEndTime time.Time
tickDuration time.Duration
)
// Check if this routine needs to awake in this round and prepare essential
// variables when yes.
checkRound := func() (isNotary bool) {
defer func() {
currentRound = nextRound
nextRound++
}()
// Wait until the configuartion for next round is ready.
var config *agreementMgrConfig
for {
if config = mgr.getConfig(nextRound); config != nil {
break
} else {
mgr.logger.Debug("round is not ready", "round", nextRound)
time.Sleep(1 * time.Second)
}
}
// Set next checkpoint.
roundBeginTime = config.beginTime
roundEndTime = config.beginTime.Add(config.roundInterval)
// Check if this node in notary set of this chain in this round.
notarySet, err := mgr.cache.GetNotarySet(nextRound, 0)
if err != nil {
panic(err)
}
setting.crs = config.crs
setting.notarySet = notarySet
_, isNotary = setting.notarySet[mgr.ID]
if isNotary {
mgr.logger.Info("selected as notary set",
"ID", mgr.ID,
"round", nextRound)
} else {
mgr.logger.Info("not selected as notary set",
"ID", mgr.ID,
"round", nextRound)
}
// Setup ticker
if tickDuration != config.lambdaBA {
if setting.ticker != nil {
setting.ticker.Stop()
}
setting.ticker = newTicker(mgr.gov, nextRound, TickerBA)
tickDuration = config.lambdaBA
}
return
}
Loop:
for {
select {
case <-mgr.ctx.Done():
break Loop
default:
}
now := time.Now().UTC()
setting.recv.isNotary = checkRound()
// Sleep until round begin. Here a biased round begin time would be
// used instead of the one in config. The reason it to disperse the load
// of fullnodes to verify confirmed blocks from each chain.
if now.Before(pickBiasedTime(roundBeginTime, 4*tickDuration)) {
select {
case <-mgr.ctx.Done():
break Loop
case <-time.After(roundBeginTime.Sub(now)):
}
// Clean the tick channel after awake: the tick would be queued in
// channel, thus the first few ticks would not tick on expected
// interval.
<-setting.ticker.Tick()
<-setting.ticker.Tick()
}
// Run BA for this round.
setting.recv.roundValue.Store(currentRound)
setting.recv.changeNotaryTime = roundEndTime
setting.recv.restartNotary <- types.Position{
Round: setting.recv.round(),
ChainID: math.MaxUint32,
}
mgr.voteFilter = utils.NewVoteFilter()
if err := mgr.baRoutineForOneRound(&setting); err != nil {
mgr.logger.Error("BA routine failed",
"error", err,
"nodeID", mgr.ID)
break Loop
}
}
}
func (mgr *agreementMgr) baRoutineForOneRound(
setting *baRoundSetting) (err error) {
agr := setting.agr
recv := setting.recv
oldPos := agr.agreementID()
restart := func(restartPos types.Position) (breakLoop bool, err error) {
if !isStop(restartPos) {
if restartPos.Round > oldPos.Round {
for {
select {
case <-mgr.ctx.Done():
break
default:
}
tipRound := mgr.bcModule.tipRound()
if tipRound > restartPos.Round {
// It's a vary rare that this go routine sleeps for entire round.
break
} else if tipRound != restartPos.Round {
mgr.logger.Debug("Waiting blockChain to change round...",
"pos", restartPos)
} else {
break
}
time.Sleep(100 * time.Millisecond)
}
// This round is finished.
breakLoop = true
return
}
if restartPos.Older(oldPos) {
// The restartNotary event is triggered by 'BlockConfirmed'
// of some older block.
return
}
}
var nextHeight uint64
var nextTime time.Time
for {
nextHeight, nextTime = mgr.bcModule.nextBlock()
if isStop(oldPos) && nextHeight == 0 {
break
}
if isStop(restartPos) && nextHeight == 0 {
break
}
if nextHeight > restartPos.Height {
break
}
mgr.logger.Debug("BlockChain not ready!!!",
"old", oldPos, "restart", restartPos, "next", nextHeight)
time.Sleep(100 * time.Millisecond)
}
nextPos := types.Position{
Round: recv.round(),
Height: nextHeight,
}
oldPos = nextPos
var leader types.NodeID
leader, err = mgr.cache.GetLeaderNode(nextPos)
if err != nil {
return
}
time.Sleep(nextTime.Sub(time.Now()))
setting.ticker.Restart()
agr.restart(setting.notarySet, nextPos, leader, setting.crs)
return
}
Loop:
for {
select {
case <-mgr.ctx.Done():
break Loop
default:
}
if agr.confirmed() {
// Block until receive restartPos
select {
case restartPos := <-recv.restartNotary:
breakLoop, err := restart(restartPos)
if err != nil {
return err
}
if breakLoop {
break Loop
}
case <-mgr.ctx.Done():
break Loop
}
}
select {
case restartPos := <-recv.restartNotary:
breakLoop, err := restart(restartPos)
if err != nil {
return err
}
if breakLoop {
break Loop
}
default:
}
if err = agr.nextState(); err != nil {
mgr.logger.Error("Failed to proceed to next state",
"nodeID", mgr.ID.String(),
"error", err)
break Loop
}
if agr.pullVotes() {
pos := agr.agreementID()
mgr.logger.Debug("Calling Network.PullVotes for syncing votes",
"position", pos)
mgr.network.PullVotes(pos)
}
for i := 0; i < agr.clocks(); i++ {
// Priority select for agreement.done().
select {
case <-agr.done():
continue Loop
default:
}
select {
case <-agr.done():
continue Loop
case <-setting.ticker.Tick():
}
}
}
return nil
}