diff options
-rw-r--r-- | core/consensus.go | 94 | ||||
-rw-r--r-- | core/consensus_test.go | 23 | ||||
-rw-r--r-- | integration_test/network.go | 42 | ||||
-rw-r--r-- | integration_test/validator.go | 2 | ||||
-rw-r--r-- | simulation/validator.go | 5 |
5 files changed, 163 insertions, 3 deletions
diff --git a/core/consensus.go b/core/consensus.go index 9c109e4..d6b5efd 100644 --- a/core/consensus.go +++ b/core/consensus.go @@ -19,6 +19,7 @@ package core import ( "fmt" + "sort" "sync" "time" @@ -56,6 +57,7 @@ var ( // Consensus implements DEXON Consensus algorithm. type Consensus struct { + ID types.ValidatorID app Application gov Governance rbModule *reliableBroadcast @@ -63,9 +65,12 @@ type Consensus struct { ctModule *consensusTimestamp ccModule *compactionChain db blockdb.BlockDatabase + network Network + tick *time.Ticker prvKey crypto.PrivateKey sigToPub SigToPubFn lock sync.RWMutex + stopChan chan struct{} } // NewConsensus construct an Consensus instance. @@ -73,6 +78,8 @@ func NewConsensus( app Application, gov Governance, db blockdb.BlockDatabase, + network Network, + tick *time.Ticker, prv crypto.PrivateKey, sigToPub SigToPubFn) *Consensus { validatorSet := gov.GetValidatorSet() @@ -95,6 +102,7 @@ func NewConsensus( validators) return &Consensus{ + ID: types.NewValidatorID(prv.PublicKey()), rbModule: rb, toModule: to, ctModule: newConsensusTimestamp(), @@ -102,8 +110,94 @@ func NewConsensus( app: newNonBlockingApplication(app), gov: gov, db: db, + network: network, + tick: tick, prvKey: prv, sigToPub: sigToPub, + stopChan: make(chan struct{}), + } +} + +// Run starts running Consensus core. +func (con *Consensus) Run() { + go con.processMsg(con.network.ReceiveChan()) + + chainID := uint64(0) + hashes := make(common.Hashes, 0, len(con.gov.GetValidatorSet())) + for vID := range con.gov.GetValidatorSet() { + hashes = append(hashes, vID.Hash) + } + sort.Sort(hashes) + for i, hash := range hashes { + if hash == con.ID.Hash { + chainID = uint64(i) + break + } + } + + genesisBlock := &types.Block{ + ProposerID: con.ID, + ChainID: chainID, + } + if err := con.PrepareGenesisBlock(genesisBlock, time.Now().UTC()); err != nil { + fmt.Println(err) + } + if err := con.ProcessBlock(genesisBlock); err != nil { + fmt.Println(err) + } + con.network.BroadcastBlock(genesisBlock) + +ProposingBlockLoop: + for { + select { + case <-con.tick.C: + case <-con.stopChan: + break ProposingBlockLoop + } + block := &types.Block{ + ProposerID: con.ID, + ChainID: chainID, + } + if err := con.PrepareBlock(block, time.Now().UTC()); err != nil { + fmt.Println(err) + } + if err := con.ProcessBlock(block); err != nil { + fmt.Println(err) + } + con.network.BroadcastBlock(block) + } +} + +// Stop the Consensus core. +func (con *Consensus) Stop() { + con.stopChan <- struct{}{} + con.stopChan <- struct{}{} +} + +func (con *Consensus) processMsg(msgChan <-chan interface{}) { + for { + var msg interface{} + select { + case msg = <-msgChan: + case <-con.stopChan: + return + } + + switch val := msg.(type) { + case *types.Block: + if err := con.ProcessBlock(val); err != nil { + fmt.Println(err) + } + types.RecycleBlock(val) + case *types.NotaryAck: + if err := con.ProcessNotaryAck(val); err != nil { + fmt.Println(err) + } + case *types.Vote: + if err := con.ProcessVote(val); err != nil { + fmt.Println(err) + } + } } } diff --git a/core/consensus_test.go b/core/consensus_test.go index 46df8eb..70ab5aa 100644 --- a/core/consensus_test.go +++ b/core/consensus_test.go @@ -30,6 +30,26 @@ import ( "github.com/stretchr/testify/suite" ) +// network implements core.Network. +type network struct { +} + +// BroadcastVote broadcasts vote to all nodes in DEXON network. +func (n *network) BroadcastVote(vote *types.Vote) {} + +// BroadcastBlock broadcasts block to all nodes in DEXON network. +func (n *network) BroadcastBlock(block *types.Block) { +} + +// BroadcastNotaryAck broadcasts notaryAck to all nodes in DEXON network. +func (n *network) BroadcastNotaryAck(notaryAck *types.NotaryAck) { +} + +// ReceiveChan returns a channel to receive messages from DEXON network. +func (n *network) ReceiveChan() <-chan interface{} { + return make(chan interface{}) +} + type ConsensusTestSuite struct { suite.Suite } @@ -56,7 +76,8 @@ func (s *ConsensusTestSuite) prepareConsensus( s.Require().Nil(err) prv, exist := gov.PrivateKeys[vID] s.Require().True(exist) - con := NewConsensus(app, gov, db, prv, eth.SigToPub) + con := NewConsensus(app, gov, db, + &network{}, time.NewTicker(1), prv, eth.SigToPub) return &con.app, con } diff --git a/integration_test/network.go b/integration_test/network.go new file mode 100644 index 0000000..446d01a --- /dev/null +++ b/integration_test/network.go @@ -0,0 +1,42 @@ +// Copyright 2018 The dexon-consensus-core 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 integration + +import ( + "github.com/dexon-foundation/dexon-consensus-core/core/types" +) + +// Network implements core.Network. +type Network struct { +} + +// BroadcastVote broadcasts vote to all nodes in DEXON network. +func (n *Network) BroadcastVote(vote *types.Vote) {} + +// BroadcastBlock broadcasts block to all nodes in DEXON network. +func (n *Network) BroadcastBlock(block *types.Block) { +} + +// BroadcastNotaryAck broadcasts notaryAck to all nodes in DEXON network. +func (n *Network) BroadcastNotaryAck(notaryAck *types.NotaryAck) { +} + +// ReceiveChan returns a channel to receive messages from DEXON network. +func (n *Network) ReceiveChan() <-chan interface{} { + return make(chan interface{}) +} diff --git a/integration_test/validator.go b/integration_test/validator.go index 60c6b1f..53f9456 100644 --- a/integration_test/validator.go +++ b/integration_test/validator.go @@ -102,7 +102,7 @@ func NewValidator( networkLatency: networkLatency, proposingLatency: proposingLatency, cons: core.NewConsensus( - app, gov, db, privateKey, eth.SigToPub), + app, gov, db, &Network{}, time.NewTicker(1), privateKey, eth.SigToPub), } } diff --git a/simulation/validator.go b/simulation/validator.go index 8a672c5..a54c848 100644 --- a/simulation/validator.go +++ b/simulation/validator.go @@ -100,7 +100,10 @@ func (v *Validator) Run() { } } v.consensus = core.NewConsensus( - v.app, v.gov, v.db, v.prvKey, v.sigToPub) + v.app, v.gov, v.db, v.network, + time.NewTicker( + time.Duration(v.config.ProposeIntervalMean)*time.Millisecond), + v.prvKey, v.sigToPub) genesisBlock := &types.Block{ ProposerID: v.ID, |