From 39493bcd9a26f1d19e1a407fbe23da1938bf9d87 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Tue, 11 Jul 2017 19:18:15 -0400 Subject: [PATCH 01/20] consensus: isProposer func --- consensus/state.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/consensus/state.go b/consensus/state.go index 415f6af3f..6259a620b 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -804,7 +804,7 @@ func (cs *ConsensusState) enterPropose(height int, round int) { return } - if !bytes.Equal(cs.Validators.GetProposer().Address, cs.privValidator.GetAddress()) { + if !cs.isProposer() { cs.Logger.Info("enterPropose: Not our turn to propose", "proposer", cs.Validators.GetProposer().Address, "privValidator", cs.privValidator) if cs.Validators.HasAddress(cs.privValidator.GetAddress()) { cs.Logger.Debug("This node is a validator") @@ -818,6 +818,10 @@ func (cs *ConsensusState) enterPropose(height int, round int) { } } +func (cs *ConsensusState) isProposer() bool { + return bytes.Equal(cs.Validators.GetProposer().Address, cs.privValidator.GetAddress()) +} + func (cs *ConsensusState) defaultDecideProposal(height, round int) { var block *types.Block var blockParts *types.PartSet From 4beac54bd9e2ae670aaf766887a779ea4f5ecb26 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Wed, 12 Jul 2017 01:02:16 -0400 Subject: [PATCH 02/20] no empty blocks --- consensus/state.go | 48 +++++++++++++++++++++++++++++++++++++++++----- mempool/mempool.go | 23 +++++++++++++++++++++- node/node.go | 1 + types/services.go | 3 +++ 4 files changed, 69 insertions(+), 6 deletions(-) diff --git a/consensus/state.go b/consensus/state.go index 6259a620b..6378297a0 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -10,13 +10,15 @@ import ( "time" fail "github.com/ebuchman/fail-test" + wire "github.com/tendermint/go-wire" + cmn "github.com/tendermint/tmlibs/common" + "github.com/tendermint/tmlibs/log" + cfg "github.com/tendermint/tendermint/config" "github.com/tendermint/tendermint/proxy" sm "github.com/tendermint/tendermint/state" "github.com/tendermint/tendermint/types" - cmn "github.com/tendermint/tmlibs/common" - "github.com/tendermint/tmlibs/log" ) //----------------------------------------------------------------------------- @@ -225,6 +227,9 @@ type ConsensusState struct { doPrevote func(height, round int) setProposal func(proposal *types.Proposal) error + // signifies that txs are available for proposal + txsAvailable chan RoundState + // closed when we finish shutting down done chan struct{} } @@ -239,6 +244,7 @@ func NewConsensusState(config *cfg.ConsensusConfig, state *sm.State, proxyAppCon peerMsgQueue: make(chan msgInfo, msgQueueSize), internalMsgQueue: make(chan msgInfo, msgQueueSize), timeoutTicker: NewTimeoutTicker(), + txsAvailable: make(chan RoundState), done: make(chan struct{}), } // set function defaults (may be overwritten before calling Start) @@ -619,6 +625,8 @@ func (cs *ConsensusState) receiveRoutine(maxSteps int) { var mi msgInfo select { + case rs_ := <-cs.txsAvailable: + cs.enterPropose(rs_.Height, rs_.Round) case mi = <-cs.peerMsgQueue: cs.wal.Save(mi) // handles proposals, block parts, votes @@ -770,11 +778,41 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { types.FireEventNewRound(cs.evsw, cs.RoundStateEvent()) - // Immediately go to enterPropose. - cs.enterPropose(height, round) + // Wait for txs to be available in the mempool + // before we enterPropose + go cs.waitForTxs(height, round) } -// Enter: from enterNewRound(height,round). +func (cs *ConsensusState) waitForTxs(height, round int) { + // if we're the proposer, start a heartbeat routine + // to tell other peers we're just waiting for txs (for debugging) + done := make(chan struct{}) + defer close(done) + if cs.isProposer() { + go cs.proposerHeartbeat(done) + } + + // wait for the mempool to have some txs + <-cs.mempool.TxsAvailable() + + // now we can enterPropose + cs.txsAvailable <- RoundState{Height: height, Round: round} +} + +func (cs *ConsensusState) proposerHeartbeat(done chan struct{}) { + for { + select { + case <-done: + return + default: + // TODO: broadcast heartbeat + + time.Sleep(time.Second) + } + } +} + +// Enter: from enter NewRound(height,round), once txs are in the mempool func (cs *ConsensusState) enterPropose(height int, round int) { if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPropose <= cs.Step) { cs.Logger.Debug(cmn.Fmt("enterPropose(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step)) diff --git a/mempool/mempool.go b/mempool/mempool.go index e804f9b35..e3c9e216c 100644 --- a/mempool/mempool.go +++ b/mempool/mempool.go @@ -72,6 +72,10 @@ type Mempool struct { // A log of mempool txs wal *auto.AutoFile + // fires once for each height, when the mempool is not empty + txsAvailable chan struct{} + notifiedTxsAvailable bool + logger log.Logger } @@ -88,6 +92,7 @@ func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool) *M recheckEnd: nil, logger: log.NewNopLogger(), cache: newTxCache(cacheSize), + txsAvailable: make(chan struct{}, 1), } mempool.initWAL() proxyAppConn.SetResponseCallback(mempool.resCb) @@ -215,6 +220,7 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) { tx: req.GetCheckTx().Tx, } mem.txs.PushBack(memTx) + mem.alertIfTxsAvailable() } else { // ignore bad transaction mem.logger.Info("Bad Transaction", "res", r) @@ -256,12 +262,25 @@ func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) { // Done! atomic.StoreInt32(&mem.rechecking, 0) mem.logger.Info("Done rechecking txs") + + mem.alertIfTxsAvailable() } default: // ignore other messages } } +func (mem *Mempool) alertIfTxsAvailable() { + if !mem.notifiedTxsAvailable && mem.Size() > 0 { + mem.notifiedTxsAvailable = true + mem.txsAvailable <- struct{}{} + } +} + +func (mem *Mempool) TxsAvailable() chan struct{} { + return mem.txsAvailable +} + // Reap returns a list of transactions currently in the mempool. // If maxTxs is -1, there is no cap on the number of returned transactions. func (mem *Mempool) Reap(maxTxs int) types.Txs { @@ -307,13 +326,15 @@ func (mem *Mempool) Update(height int, txs types.Txs) { // Set height mem.height = height + mem.notifiedTxsAvailable = false + // Remove transactions that are already in txs. goodTxs := mem.filterTxs(txsMap) // Recheck mempool txs if any txs were committed in the block // NOTE/XXX: in some apps a tx could be invalidated due to EndBlock, // so we really still do need to recheck, but this is for debugging if mem.config.Recheck && (mem.config.RecheckEmpty || len(txs) > 0) { - mem.logger.Info("Recheck txs", "numtxs", len(goodTxs)) + mem.logger.Info("Recheck txs", "numtxs", len(goodTxs), "height", height) mem.recheckTxs(goodTxs) // At this point, mem.txs are being rechecked. // mem.recheckCursor re-scans mem.txs and possibly removes some txs. diff --git a/node/node.go b/node/node.go index 672e384b9..513902000 100644 --- a/node/node.go +++ b/node/node.go @@ -139,6 +139,7 @@ func NewNode(config *cfg.Config, privValidator *types.PrivValidator, clientCreat mempoolLogger := logger.With("module", "mempool") mempool := mempl.NewMempool(config.Mempool, proxyApp.Mempool()) mempool.SetLogger(mempoolLogger) + mempool.Update(state.LastBlockHeight, nil) mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool) mempoolReactor.SetLogger(mempoolLogger) diff --git a/types/services.go b/types/services.go index ee20487e2..77c84a091 100644 --- a/types/services.go +++ b/types/services.go @@ -23,6 +23,8 @@ type Mempool interface { Reap(int) Txs Update(height int, txs Txs) Flush() + + TxsAvailable() chan struct{} } type MockMempool struct { @@ -35,6 +37,7 @@ func (m MockMempool) CheckTx(tx Tx, cb func(*abci.Response)) error { return nil func (m MockMempool) Reap(n int) Txs { return Txs{} } func (m MockMempool) Update(height int, txs Txs) {} func (m MockMempool) Flush() {} +func (m MockMempool) TxsAvailable() chan struct{} { return make(chan struct{}) } //------------------------------------------------------ // blockstore From 124032e3e93b614fa9fdfc61ed366a9c4bd28b75 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 13 Jul 2017 13:19:44 -0400 Subject: [PATCH 03/20] NoEmptyBlocks config option --- config/config.go | 9 ++++++--- consensus/state.go | 6 +++++- mempool/mempool.go | 22 +++++++++++++++++++++- node/node.go | 4 ++++ 4 files changed, 36 insertions(+), 5 deletions(-) diff --git a/config/config.go b/config/config.go index e552b33b4..a633b7fbf 100644 --- a/config/config.go +++ b/config/config.go @@ -301,8 +301,9 @@ type ConsensusConfig struct { SkipTimeoutCommit bool `mapstructure:"skip_timeout_commit"` // BlockSize - MaxBlockSizeTxs int `mapstructure:"max_block_size_txs"` - MaxBlockSizeBytes int `mapstructure:"max_block_size_bytes"` + MaxBlockSizeTxs int `mapstructure:"max_block_size_txs"` + MaxBlockSizeBytes int `mapstructure:"max_block_size_bytes"` + NoEmptyBlocks bool `mapstructure:"no_empty_blocks"` // TODO: This probably shouldn't be exposed but it makes it // easy to write tests for the wal/replay @@ -357,7 +358,8 @@ func DefaultConsensusConfig() *ConsensusConfig { TimeoutCommit: 1000, SkipTimeoutCommit: false, MaxBlockSizeTxs: 10000, - MaxBlockSizeBytes: 1, // TODO + MaxBlockSizeBytes: 1, // TODO + NoEmptyBlocks: true, BlockPartSize: types.DefaultBlockPartSize, // TODO: we shouldnt be importing types PeerGossipSleepDuration: 100, PeerQueryMaj23SleepDuration: 2000, @@ -375,6 +377,7 @@ func TestConsensusConfig() *ConsensusConfig { config.TimeoutPrecommitDelta = 1 config.TimeoutCommit = 10 config.SkipTimeoutCommit = true + config.NoEmptyBlocks = false return config } diff --git a/consensus/state.go b/consensus/state.go index 6378297a0..ff5b42c7b 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -780,7 +780,11 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { // Wait for txs to be available in the mempool // before we enterPropose - go cs.waitForTxs(height, round) + if cs.config.NoEmptyBlocks { + go cs.waitForTxs(height, round) + } else { + cs.enterPropose(height, round) + } } func (cs *ConsensusState) waitForTxs(height, round int) { diff --git a/mempool/mempool.go b/mempool/mempool.go index e3c9e216c..160ab7d02 100644 --- a/mempool/mempool.go +++ b/mempool/mempool.go @@ -92,13 +92,18 @@ func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool) *M recheckEnd: nil, logger: log.NewNopLogger(), cache: newTxCache(cacheSize), - txsAvailable: make(chan struct{}, 1), } mempool.initWAL() proxyAppConn.SetResponseCallback(mempool.resCb) return mempool } +// FireOnTxsAvailable initializes the TxsAvailable channel, +// ensuring it will trigger once every height when transactions are available. +func (mem *Mempool) FireOnTxsAvailable() { + mem.txsAvailable = make(chan struct{}, 1) +} + // SetLogger sets the Logger. func (mem *Mempool) SetLogger(l log.Logger) { mem.logger = l @@ -277,10 +282,25 @@ func (mem *Mempool) alertIfTxsAvailable() { } } +// TxsAvailable returns a channel which fires once for every height, +// and only when transactions are available in the mempool. +// XXX: Will panic if mem.FireOnTxsAvailable() has not been called. func (mem *Mempool) TxsAvailable() chan struct{} { + if mem.txsAvailable == nil { + panic("mem.txsAvailable is nil") + } return mem.txsAvailable } +func (mem *Mempool) alertIfTxsAvailable() { + if mem.txsAvailable != nil && + !mem.notifiedTxsAvailable && mem.Size() > 0 { + + mem.notifiedTxsAvailable = true + mem.txsAvailable <- struct{}{} + } +} + // Reap returns a list of transactions currently in the mempool. // If maxTxs is -1, there is no cap on the number of returned transactions. func (mem *Mempool) Reap(maxTxs int) types.Txs { diff --git a/node/node.go b/node/node.go index 513902000..c76ae5fa3 100644 --- a/node/node.go +++ b/node/node.go @@ -143,6 +143,10 @@ func NewNode(config *cfg.Config, privValidator *types.PrivValidator, clientCreat mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool) mempoolReactor.SetLogger(mempoolLogger) + if config.Consensus.NoEmptyBlocks { + mempool.FireOnTxsAvailable() + } + // Make ConsensusReactor consensusState := consensus.NewConsensusState(config.Consensus, state.Copy(), proxyApp.Consensus(), blockStore, mempool) consensusState.SetLogger(consensusLogger) From 678a9a2e428b68c4a561c0af7d2e9cd192af876c Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 13 Jul 2017 15:03:19 -0400 Subject: [PATCH 04/20] TxsAvailable tests --- consensus/common_test.go | 16 +++++-- consensus/mempool_test.go | 43 ++++++++++++------ mempool/mempool_test.go | 95 ++++++++++++++++++++++++++++++++++++--- types/services.go | 2 + 4 files changed, 132 insertions(+), 24 deletions(-) diff --git a/consensus/common_test.go b/consensus/common_test.go index 103294ab2..63f77b09c 100644 --- a/consensus/common_test.go +++ b/consensus/common_test.go @@ -294,12 +294,22 @@ func randConsensusState(nValidators int) (*ConsensusState, []*validatorStub) { //------------------------------------------------------------------------------- func ensureNoNewStep(stepCh chan interface{}) { - timeout := time.NewTicker(ensureTimeout * time.Second) + timer := time.NewTimer(ensureTimeout * time.Second) select { - case <-timeout.C: + case <-timer.C: break case <-stepCh: - panic("We should be stuck waiting for more votes, not moving to the next step") + panic("We should be stuck waiting, not moving to the next step") + } +} + +func ensureNewStep(stepCh chan interface{}) { + timer := time.NewTimer(ensureTimeout * time.Second) + select { + case <-timer.C: + panic("We shouldnt be stuck waiting") + case <-stepCh: + break } } diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index 327ad733c..9f332e15c 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -15,6 +15,34 @@ func init() { config = ResetConfig("consensus_mempool_test") } +func TestTxsAvailable(t *testing.T) { + config := ResetConfig("consensus_mempool_txs_available_test") + config.Consensus.NoEmptyBlocks = true + state, privVals := randGenesisState(1, false, 10) + cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication()) + cs.mempool.FireOnTxsAvailable() + height, round := cs.Height, cs.Round + newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1) + startTestRound(cs, height, round) + + // we shouldnt make progress until theres a tx + ensureNoNewStep(newBlockCh) + deliverTxsRange(cs, 0, 100) + ensureNewStep(newBlockCh) +} + +func deliverTxsRange(cs *ConsensusState, start, end int) { + // Deliver some txs. + for i := start; i < end; i++ { + txBytes := make([]byte, 8) + binary.BigEndian.PutUint64(txBytes, uint64(i)) + err := cs.mempool.CheckTx(txBytes, nil) + if err != nil { + panic(Fmt("Error after CheckTx: %v", err)) + } + } +} + func TestTxConcurrentWithCommit(t *testing.T) { state, privVals := randGenesisState(1, false, 10) @@ -22,21 +50,8 @@ func TestTxConcurrentWithCommit(t *testing.T) { height, round := cs.Height, cs.Round newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1) - deliverTxsRange := func(start, end int) { - // Deliver some txs. - for i := start; i < end; i++ { - txBytes := make([]byte, 8) - binary.BigEndian.PutUint64(txBytes, uint64(i)) - err := cs.mempool.CheckTx(txBytes, nil) - if err != nil { - panic(Fmt("Error after CheckTx: %v", err)) - } - // time.Sleep(time.Microsecond * time.Duration(rand.Int63n(3000))) - } - } - NTxs := 10000 - go deliverTxsRange(0, NTxs) + go deliverTxsRange(cs, 0, NTxs) startTestRound(cs, height, round) ticker := time.NewTicker(time.Second * 20) diff --git a/mempool/mempool_test.go b/mempool/mempool_test.go index 6451adb2d..6da9eff51 100644 --- a/mempool/mempool_test.go +++ b/mempool/mempool_test.go @@ -1,34 +1,115 @@ package mempool import ( + "crypto/rand" "encoding/binary" "testing" + "time" "github.com/tendermint/abci/example/counter" + "github.com/tendermint/abci/example/dummy" + "github.com/tendermint/tmlibs/log" + cfg "github.com/tendermint/tendermint/config" "github.com/tendermint/tendermint/proxy" "github.com/tendermint/tendermint/types" - "github.com/tendermint/tmlibs/log" ) -func TestSerialReap(t *testing.T) { +func newMempoolWithApp(t *testing.T, cc proxy.ClientCreator) *Mempool { config := cfg.ResetTestRoot("mempool_test") - app := counter.NewCounterApplication(true) - app.SetOption("serial", "on") - cc := proxy.NewLocalClientCreator(app) appConnMem, _ := cc.NewABCIClient() appConnMem.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "mempool")) if _, err := appConnMem.Start(); err != nil { t.Fatalf("Error starting ABCI client: %v", err.Error()) } + mempool := NewMempool(config.Mempool, appConnMem) + mempool.SetLogger(log.TestingLogger()) + return mempool +} + +func ensureNoFire(t *testing.T, ch chan struct{}, timeoutMS int) { + timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond) + select { + case <-ch: + t.Fatal("Expected not to fire") + case <-timer.C: + } +} + +func ensureFire(t *testing.T, ch chan struct{}, timeoutMS int) { + timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond) + select { + case <-ch: + case <-timer.C: + t.Fatal("Expected to fire") + } +} + +func sendTxs(t *testing.T, mempool *Mempool, count int) types.Txs { + txs := make(types.Txs, count) + for i := 0; i < count; i++ { + txBytes := make([]byte, 20) + txs[i] = txBytes + rand.Read(txBytes) + err := mempool.CheckTx(txBytes, nil) + if err != nil { + t.Fatal("Error after CheckTx: %v", err) + } + } + return txs +} + +func TestTxsAvailable(t *testing.T) { + app := dummy.NewDummyApplication() + cc := proxy.NewLocalClientCreator(app) + mempool := newMempoolWithApp(t, cc) + mempool.FireOnTxsAvailable() + + timeoutMS := 500 + + // with no txs, it shouldnt fire + ensureNoFire(t, mempool.TxsAvailable(), timeoutMS) + + // send a bunch of txs, it should only fire once + txs := sendTxs(t, mempool, 100) + ensureFire(t, mempool.TxsAvailable(), timeoutMS) + ensureNoFire(t, mempool.TxsAvailable(), timeoutMS) + + // call update with half the txs. + // it should fire once now for the new height + // since there are still txs left + committedTxs, txs := txs[:50], txs[50:] + mempool.Update(1, committedTxs) + ensureFire(t, mempool.TxsAvailable(), timeoutMS) + ensureNoFire(t, mempool.TxsAvailable(), timeoutMS) + + // send a bunch more txs. we already fired for this height so it shouldnt fire again + moreTxs := sendTxs(t, mempool, 50) + ensureNoFire(t, mempool.TxsAvailable(), timeoutMS) + + // now call update with all the txs. it should not fire as there are no txs left + committedTxs = append(txs, moreTxs...) + mempool.Update(2, committedTxs) + ensureNoFire(t, mempool.TxsAvailable(), timeoutMS) + + // send a bunch more txs, it should only fire once + sendTxs(t, mempool, 100) + ensureFire(t, mempool.TxsAvailable(), timeoutMS) + ensureNoFire(t, mempool.TxsAvailable(), timeoutMS) +} + +func TestSerialReap(t *testing.T) { + app := counter.NewCounterApplication(true) + app.SetOption("serial", "on") + cc := proxy.NewLocalClientCreator(app) + + mempool := newMempoolWithApp(t, cc) appConnCon, _ := cc.NewABCIClient() appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus")) if _, err := appConnCon.Start(); err != nil { t.Fatalf("Error starting ABCI client: %v", err.Error()) } - mempool := NewMempool(config.Mempool, appConnMem) - mempool.SetLogger(log.TestingLogger()) deliverTxsRange := func(start, end int) { // Deliver some txs. diff --git a/types/services.go b/types/services.go index 77c84a091..f68702ec6 100644 --- a/types/services.go +++ b/types/services.go @@ -25,6 +25,7 @@ type Mempool interface { Flush() TxsAvailable() chan struct{} + FireOnTxsAvailable() } type MockMempool struct { @@ -38,6 +39,7 @@ func (m MockMempool) Reap(n int) Txs { return Txs{ func (m MockMempool) Update(height int, txs Txs) {} func (m MockMempool) Flush() {} func (m MockMempool) TxsAvailable() chan struct{} { return make(chan struct{}) } +func (m MockMempool) FireOnTxsAvailable() {} //------------------------------------------------------ // blockstore From e9a23893004c6a738f62ce969d4aeb488f72fdb7 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 13 Jul 2017 15:08:00 -0400 Subject: [PATCH 05/20] cmd: --consensus.no_empty_blocks --- cmd/tendermint/commands/run_node.go | 3 +++ config/config.go | 3 +-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/cmd/tendermint/commands/run_node.go b/cmd/tendermint/commands/run_node.go index e99b8609a..fce96edcb 100644 --- a/cmd/tendermint/commands/run_node.go +++ b/cmd/tendermint/commands/run_node.go @@ -45,6 +45,9 @@ func AddNodeFlags(cmd *cobra.Command) { cmd.Flags().String("p2p.seeds", config.P2P.Seeds, "Comma delimited host:port seed nodes") cmd.Flags().Bool("p2p.skip_upnp", config.P2P.SkipUPNP, "Skip UPNP configuration") cmd.Flags().Bool("p2p.pex", config.P2P.PexReactor, "Enable Peer-Exchange (dev feature)") + + // consensus flags + cmd.Flags().Bool("consensus.no_empty_blocks", config.Consensus.NoEmptyBlocks, "Prevent empty blocks from being proposed by correct processes") } // Users wishing to: diff --git a/config/config.go b/config/config.go index a633b7fbf..323b36a25 100644 --- a/config/config.go +++ b/config/config.go @@ -359,7 +359,7 @@ func DefaultConsensusConfig() *ConsensusConfig { SkipTimeoutCommit: false, MaxBlockSizeTxs: 10000, MaxBlockSizeBytes: 1, // TODO - NoEmptyBlocks: true, + NoEmptyBlocks: false, BlockPartSize: types.DefaultBlockPartSize, // TODO: we shouldnt be importing types PeerGossipSleepDuration: 100, PeerQueryMaj23SleepDuration: 2000, @@ -377,7 +377,6 @@ func TestConsensusConfig() *ConsensusConfig { config.TimeoutPrecommitDelta = 1 config.TimeoutCommit = 10 config.SkipTimeoutCommit = true - config.NoEmptyBlocks = false return config } From d39605387238447938bbedbc8b141b3819ec3026 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 13 Jul 2017 15:09:26 -0400 Subject: [PATCH 06/20] changelog --- CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b460dc39e..4c233a264 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Changelog +## 0.10.3 (TBD) + +FEATURES: +- New `--consensus.no_empty_blocks` flag prevents the creation of blocks until there are transactions available + ## 0.10.2 (July 10, 2017) FEATURES: From fc3fe9292f7b4e3464b90853e1122de5d40f1e5f Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Mon, 17 Jul 2017 13:49:35 -0400 Subject: [PATCH 07/20] fixes from review --- cmd/tendermint/commands/run_node.go | 2 +- consensus/state.go | 7 ++++--- mempool/mempool.go | 16 ++++++---------- 3 files changed, 11 insertions(+), 14 deletions(-) diff --git a/cmd/tendermint/commands/run_node.go b/cmd/tendermint/commands/run_node.go index fce96edcb..b9df28ab8 100644 --- a/cmd/tendermint/commands/run_node.go +++ b/cmd/tendermint/commands/run_node.go @@ -47,7 +47,7 @@ func AddNodeFlags(cmd *cobra.Command) { cmd.Flags().Bool("p2p.pex", config.P2P.PexReactor, "Enable Peer-Exchange (dev feature)") // consensus flags - cmd.Flags().Bool("consensus.no_empty_blocks", config.Consensus.NoEmptyBlocks, "Prevent empty blocks from being proposed by correct processes") + cmd.Flags().Bool("consensus.no_empty_blocks", config.Consensus.NoEmptyBlocks, "Only produce blocks when there are txs and when the AppHash changes") } // Users wishing to: diff --git a/consensus/state.go b/consensus/state.go index ff5b42c7b..878b235a3 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -790,9 +790,9 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { func (cs *ConsensusState) waitForTxs(height, round int) { // if we're the proposer, start a heartbeat routine // to tell other peers we're just waiting for txs (for debugging) - done := make(chan struct{}) - defer close(done) if cs.isProposer() { + done := make(chan struct{}) + defer close(done) go cs.proposerHeartbeat(done) } @@ -816,7 +816,8 @@ func (cs *ConsensusState) proposerHeartbeat(done chan struct{}) { } } -// Enter: from enter NewRound(height,round), once txs are in the mempool +// Enter (!NoEmptyBlocks): from enterNewRound(height,round) +// Enter (NoEmptyBlocks) : after enterNewRound(height,round), once txs are in the mempool func (cs *ConsensusState) enterPropose(height int, round int) { if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPropose <= cs.Step) { cs.Logger.Debug(cmn.Fmt("enterPropose(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step)) diff --git a/mempool/mempool.go b/mempool/mempool.go index 160ab7d02..fadeb5dac 100644 --- a/mempool/mempool.go +++ b/mempool/mempool.go @@ -100,6 +100,7 @@ func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool) *M // FireOnTxsAvailable initializes the TxsAvailable channel, // ensuring it will trigger once every height when transactions are available. +// NOTE: not thread safe - should only be called once, on startup func (mem *Mempool) FireOnTxsAvailable() { mem.txsAvailable = make(chan struct{}, 1) } @@ -225,7 +226,7 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) { tx: req.GetCheckTx().Tx, } mem.txs.PushBack(memTx) - mem.alertIfTxsAvailable() + mem.notifyIfTxsAvailable() } else { // ignore bad transaction mem.logger.Info("Bad Transaction", "res", r) @@ -268,20 +269,13 @@ func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) { atomic.StoreInt32(&mem.rechecking, 0) mem.logger.Info("Done rechecking txs") - mem.alertIfTxsAvailable() + mem.notifyIfTxsAvailable() } default: // ignore other messages } } -func (mem *Mempool) alertIfTxsAvailable() { - if !mem.notifiedTxsAvailable && mem.Size() > 0 { - mem.notifiedTxsAvailable = true - mem.txsAvailable <- struct{}{} - } -} - // TxsAvailable returns a channel which fires once for every height, // and only when transactions are available in the mempool. // XXX: Will panic if mem.FireOnTxsAvailable() has not been called. @@ -292,7 +286,7 @@ func (mem *Mempool) TxsAvailable() chan struct{} { return mem.txsAvailable } -func (mem *Mempool) alertIfTxsAvailable() { +func (mem *Mempool) notifyIfTxsAvailable() { if mem.txsAvailable != nil && !mem.notifiedTxsAvailable && mem.Size() > 0 { @@ -345,6 +339,8 @@ func (mem *Mempool) Update(height int, txs types.Txs) { } // Set height + // NOTE: the height is not set until Update is first called + // (so it will be wrong after a restart until the next block) mem.height = height mem.notifiedTxsAvailable = false From ecdda69fabd6ac5b5c82d40f6d2de76af7cb0d2e Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 20 Jul 2017 14:43:16 -0400 Subject: [PATCH 08/20] commit empty blocks when needed to prove app hash --- consensus/mempool_test.go | 10 ++++++---- consensus/state.go | 19 +++++++++++++++++-- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index 9f332e15c..18daf67a6 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -15,7 +15,7 @@ func init() { config = ResetConfig("consensus_mempool_test") } -func TestTxsAvailable(t *testing.T) { +func TestNoProgressUntilTxsAvailable(t *testing.T) { config := ResetConfig("consensus_mempool_txs_available_test") config.Consensus.NoEmptyBlocks = true state, privVals := randGenesisState(1, false, 10) @@ -25,10 +25,12 @@ func TestTxsAvailable(t *testing.T) { newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1) startTestRound(cs, height, round) - // we shouldnt make progress until theres a tx + ensureNewStep(newBlockCh) // first block gets committed + ensureNoNewStep(newBlockCh) + deliverTxsRange(cs, 0, 2) + ensureNewStep(newBlockCh) // commit txs + ensureNewStep(newBlockCh) // commit updated app hash ensureNoNewStep(newBlockCh) - deliverTxsRange(cs, 0, 100) - ensureNewStep(newBlockCh) } func deliverTxsRange(cs *ConsensusState, start, end int) { diff --git a/consensus/state.go b/consensus/state.go index 878b235a3..b18b40db3 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -779,14 +779,29 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { types.FireEventNewRound(cs.evsw, cs.RoundStateEvent()) // Wait for txs to be available in the mempool - // before we enterPropose - if cs.config.NoEmptyBlocks { + // before we enterPropose. If the last block changed the app hash, + // we may need an empty "proof" block, and enterPropose immediately. + if cs.config.NoEmptyBlocks && !cs.needProofBlock(height) { go cs.waitForTxs(height, round) } else { cs.enterPropose(height, round) } } +// needProofBlock returns true on the first height (so the genesis app hash is signed right away) +// and where the last block (height-1) caused the app hash to change +func (cs *ConsensusState) needProofBlock(height int) bool { + if height == 1 { + return true + } + + lastBlockMeta := cs.blockStore.LoadBlockMeta(height - 1) + if !bytes.Equal(cs.state.AppHash, lastBlockMeta.Header.AppHash) { + return true + } + return false +} + func (cs *ConsensusState) waitForTxs(height, round int) { // if we're the proposer, start a heartbeat routine // to tell other peers we're just waiting for txs (for debugging) From cf3abe5096682fafeab49e257fcaadc106d63abb Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Tue, 25 Jul 2017 10:52:14 -0400 Subject: [PATCH 09/20] consensus: remove rs from handleMsg --- consensus/replay.go | 2 +- consensus/state.go | 18 ++++++++++++------ 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/consensus/replay.go b/consensus/replay.go index c4ed684c7..0c7c27e90 100644 --- a/consensus/replay.go +++ b/consensus/replay.go @@ -82,7 +82,7 @@ func (cs *ConsensusState) readReplayMessage(msgBytes []byte, newStepCh chan inte "blockID", v.BlockID, "peer", peerKey) } - cs.handleMsg(m, cs.RoundState) + cs.handleMsg(m) case timeoutInfo: cs.Logger.Info("Replay: Timeout", "height", m.Height, "round", m.Round, "step", m.Step, "dur", m.Duration) cs.handleTimeout(m, cs.RoundState) diff --git a/consensus/state.go b/consensus/state.go index b18b40db3..d1dec7f59 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -611,7 +611,8 @@ func (cs *ConsensusState) newStep() { // receiveRoutine handles messages which may cause state transitions. // it's argument (n) is the number of messages to process before exiting - use 0 to run forever // It keeps the RoundState and is the only thing that updates it. -// Updates (state transitions) happen on timeouts, complete proposals, and 2/3 majorities +// Updates (state transitions) happen on timeouts, complete proposals, and 2/3 majorities. +// ConsensusState must be locked before any internal state is updated. func (cs *ConsensusState) receiveRoutine(maxSteps int) { for { if maxSteps > 0 { @@ -625,17 +626,19 @@ func (cs *ConsensusState) receiveRoutine(maxSteps int) { var mi msgInfo select { - case rs_ := <-cs.txsAvailable: - cs.enterPropose(rs_.Height, rs_.Round) + case <-cs.txsAvailable: + // use nil for this special internal message signalling txs are available. + // no need to write this to the wal + // cs.handleMsg(msgInfo{nil, ""}, rs_) case mi = <-cs.peerMsgQueue: cs.wal.Save(mi) // handles proposals, block parts, votes // may generate internal events (votes, complete proposals, 2/3 majorities) - cs.handleMsg(mi, rs) + cs.handleMsg(mi) case mi = <-cs.internalMsgQueue: cs.wal.Save(mi) // handles proposals, block parts, votes - cs.handleMsg(mi, rs) + cs.handleMsg(mi) case ti := <-cs.timeoutTicker.Chan(): // tockChan: cs.wal.Save(ti) // if the timeout is relevant to the rs @@ -659,13 +662,16 @@ func (cs *ConsensusState) receiveRoutine(maxSteps int) { } // state transitions on complete-proposal, 2/3-any, 2/3-one -func (cs *ConsensusState) handleMsg(mi msgInfo, rs RoundState) { +func (cs *ConsensusState) handleMsg(mi msgInfo) { cs.mtx.Lock() defer cs.mtx.Unlock() var err error msg, peerKey := mi.Msg, mi.PeerKey switch msg := msg.(type) { + case nil: + // transactions are available, so enterPropose + // cs.enterPropose(rs.Height, rs.Round) case *ProposalMessage: // will not cause transition. // once proposal is set, we can receive block parts From 3444bee47fe423c8832df092a336d2a1d6a80726 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Tue, 25 Jul 2017 13:57:11 -0400 Subject: [PATCH 10/20] fixes from review; use mempool.TxsAvailable() directly --- consensus/common_test.go | 5 +++- consensus/mempool_test.go | 2 +- consensus/state.go | 47 ++++++++++++-------------------------- mempool/mempool.go | 48 ++++++++++++++++++--------------------- mempool/mempool_test.go | 8 +++---- node/node.go | 5 ++-- types/services.go | 8 +++---- 7 files changed, 51 insertions(+), 72 deletions(-) diff --git a/consensus/common_test.go b/consensus/common_test.go index 63f77b09c..6b96ec311 100644 --- a/consensus/common_test.go +++ b/consensus/common_test.go @@ -240,8 +240,11 @@ func newConsensusStateWithConfig(thisConfig *cfg.Config, state *sm.State, pv *ty proxyAppConnCon := abcicli.NewLocalClient(mtx, app) // Make Mempool - mempool := mempl.NewMempool(thisConfig.Mempool, proxyAppConnMem) + mempool := mempl.NewMempool(thisConfig.Mempool, proxyAppConnMem, 0) mempool.SetLogger(log.TestingLogger().With("module", "mempool")) + if thisConfig.Consensus.NoEmptyBlocks { + mempool.EnableTxsAvailable() + } // Make ConsensusReactor cs := NewConsensusState(thisConfig.Consensus, state, proxyAppConnCon, blockStore, mempool) diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index 18daf67a6..843a85aa6 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -20,7 +20,7 @@ func TestNoProgressUntilTxsAvailable(t *testing.T) { config.Consensus.NoEmptyBlocks = true state, privVals := randGenesisState(1, false, 10) cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication()) - cs.mempool.FireOnTxsAvailable() + cs.mempool.EnableTxsAvailable() height, round := cs.Height, cs.Round newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1) startTestRound(cs, height, round) diff --git a/consensus/state.go b/consensus/state.go index d1dec7f59..4695a6304 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -227,9 +227,6 @@ type ConsensusState struct { doPrevote func(height, round int) setProposal func(proposal *types.Proposal) error - // signifies that txs are available for proposal - txsAvailable chan RoundState - // closed when we finish shutting down done chan struct{} } @@ -244,7 +241,6 @@ func NewConsensusState(config *cfg.ConsensusConfig, state *sm.State, proxyAppCon peerMsgQueue: make(chan msgInfo, msgQueueSize), internalMsgQueue: make(chan msgInfo, msgQueueSize), timeoutTicker: NewTimeoutTicker(), - txsAvailable: make(chan RoundState), done: make(chan struct{}), } // set function defaults (may be overwritten before calling Start) @@ -626,10 +622,8 @@ func (cs *ConsensusState) receiveRoutine(maxSteps int) { var mi msgInfo select { - case <-cs.txsAvailable: - // use nil for this special internal message signalling txs are available. - // no need to write this to the wal - // cs.handleMsg(msgInfo{nil, ""}, rs_) + case height := <-cs.mempool.TxsAvailable(): + cs.handleTxsAvailable(height) case mi = <-cs.peerMsgQueue: cs.wal.Save(mi) // handles proposals, block parts, votes @@ -669,9 +663,6 @@ func (cs *ConsensusState) handleMsg(mi msgInfo) { var err error msg, peerKey := mi.Msg, mi.PeerKey switch msg := msg.(type) { - case nil: - // transactions are available, so enterPropose - // cs.enterPropose(rs.Height, rs.Round) case *ProposalMessage: // will not cause transition. // once proposal is set, we can receive block parts @@ -737,6 +728,13 @@ func (cs *ConsensusState) handleTimeout(ti timeoutInfo, rs RoundState) { } +func (cs *ConsensusState) handleTxsAvailable(height int) { + cs.mtx.Lock() + defer cs.mtx.Unlock() + // we only need to do this for round 0 + cs.enterPropose(height, 0) +} + //----------------------------------------------------------------------------- // State functions // Used internally by handleTimeout and handleMsg to make state transitions @@ -785,10 +783,11 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { types.FireEventNewRound(cs.evsw, cs.RoundStateEvent()) // Wait for txs to be available in the mempool - // before we enterPropose. If the last block changed the app hash, + // before we enterPropose in round 0. If the last block changed the app hash, // we may need an empty "proof" block, and enterPropose immediately. - if cs.config.NoEmptyBlocks && !cs.needProofBlock(height) { - go cs.waitForTxs(height, round) + waitForTxs := cs.config.NoEmptyBlocks && round == 0 && !cs.needProofBlock(height) + if waitForTxs { + go cs.proposalHeartbeat() } else { cs.enterPropose(height, round) } @@ -808,27 +807,9 @@ func (cs *ConsensusState) needProofBlock(height int) bool { return false } -func (cs *ConsensusState) waitForTxs(height, round int) { - // if we're the proposer, start a heartbeat routine - // to tell other peers we're just waiting for txs (for debugging) - if cs.isProposer() { - done := make(chan struct{}) - defer close(done) - go cs.proposerHeartbeat(done) - } - - // wait for the mempool to have some txs - <-cs.mempool.TxsAvailable() - - // now we can enterPropose - cs.txsAvailable <- RoundState{Height: height, Round: round} -} - -func (cs *ConsensusState) proposerHeartbeat(done chan struct{}) { +func (cs *ConsensusState) proposalHeartbeat() { for { select { - case <-done: - return default: // TODO: broadcast heartbeat diff --git a/mempool/mempool.go b/mempool/mempool.go index fadeb5dac..9f9824f9e 100644 --- a/mempool/mempool.go +++ b/mempool/mempool.go @@ -56,14 +56,16 @@ const cacheSize = 100000 type Mempool struct { config *cfg.MempoolConfig - proxyMtx sync.Mutex - proxyAppConn proxy.AppConnMempool - txs *clist.CList // concurrent linked-list of good txs - counter int64 // simple incrementing counter - height int // the last block Update()'d to - rechecking int32 // for re-checking filtered txs on Update() - recheckCursor *clist.CElement // next expected response - recheckEnd *clist.CElement // re-checking stops here + proxyMtx sync.Mutex + proxyAppConn proxy.AppConnMempool + txs *clist.CList // concurrent linked-list of good txs + counter int64 // simple incrementing counter + height int // the last block Update()'d to + rechecking int32 // for re-checking filtered txs on Update() + recheckCursor *clist.CElement // next expected response + recheckEnd *clist.CElement // re-checking stops here + notifiedTxsAvailable bool // true if fired on txsAvailable for this height + txsAvailable chan int // fires the next height once for each height, when the mempool is not empty // Keep a cache of already-seen txs. // This reduces the pressure on the proxyApp. @@ -72,21 +74,17 @@ type Mempool struct { // A log of mempool txs wal *auto.AutoFile - // fires once for each height, when the mempool is not empty - txsAvailable chan struct{} - notifiedTxsAvailable bool - logger log.Logger } // NewMempool returns a new Mempool with the given configuration and connection to an application. -func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool) *Mempool { +func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool, height int) *Mempool { mempool := &Mempool{ config: config, proxyAppConn: proxyAppConn, txs: clist.New(), counter: 0, - height: 0, + height: height, rechecking: 0, recheckCursor: nil, recheckEnd: nil, @@ -98,11 +96,11 @@ func NewMempool(config *cfg.MempoolConfig, proxyAppConn proxy.AppConnMempool) *M return mempool } -// FireOnTxsAvailable initializes the TxsAvailable channel, +// EnableTxsAvailable initializes the TxsAvailable channel, // ensuring it will trigger once every height when transactions are available. // NOTE: not thread safe - should only be called once, on startup -func (mem *Mempool) FireOnTxsAvailable() { - mem.txsAvailable = make(chan struct{}, 1) +func (mem *Mempool) EnableTxsAvailable() { + mem.txsAvailable = make(chan int, 1) } // SetLogger sets the Logger. @@ -278,20 +276,20 @@ func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) { // TxsAvailable returns a channel which fires once for every height, // and only when transactions are available in the mempool. -// XXX: Will panic if mem.FireOnTxsAvailable() has not been called. -func (mem *Mempool) TxsAvailable() chan struct{} { - if mem.txsAvailable == nil { - panic("mem.txsAvailable is nil") - } +// NOTE: the returned channel may be nil if EnableTxsAvailable was not called. +func (mem *Mempool) TxsAvailable() chan int { return mem.txsAvailable } func (mem *Mempool) notifyIfTxsAvailable() { + if mem.Size() == 0 { + panic("notified txs available but mempool is empty!") + } if mem.txsAvailable != nil && - !mem.notifiedTxsAvailable && mem.Size() > 0 { + !mem.notifiedTxsAvailable { mem.notifiedTxsAvailable = true - mem.txsAvailable <- struct{}{} + mem.txsAvailable <- mem.height + 1 } } @@ -339,8 +337,6 @@ func (mem *Mempool) Update(height int, txs types.Txs) { } // Set height - // NOTE: the height is not set until Update is first called - // (so it will be wrong after a restart until the next block) mem.height = height mem.notifiedTxsAvailable = false diff --git a/mempool/mempool_test.go b/mempool/mempool_test.go index 6da9eff51..2e7a575cf 100644 --- a/mempool/mempool_test.go +++ b/mempool/mempool_test.go @@ -23,12 +23,12 @@ func newMempoolWithApp(t *testing.T, cc proxy.ClientCreator) *Mempool { if _, err := appConnMem.Start(); err != nil { t.Fatalf("Error starting ABCI client: %v", err.Error()) } - mempool := NewMempool(config.Mempool, appConnMem) + mempool := NewMempool(config.Mempool, appConnMem, 0) mempool.SetLogger(log.TestingLogger()) return mempool } -func ensureNoFire(t *testing.T, ch chan struct{}, timeoutMS int) { +func ensureNoFire(t *testing.T, ch chan int, timeoutMS int) { timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond) select { case <-ch: @@ -37,7 +37,7 @@ func ensureNoFire(t *testing.T, ch chan struct{}, timeoutMS int) { } } -func ensureFire(t *testing.T, ch chan struct{}, timeoutMS int) { +func ensureFire(t *testing.T, ch chan int, timeoutMS int) { timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond) select { case <-ch: @@ -64,7 +64,7 @@ func TestTxsAvailable(t *testing.T) { app := dummy.NewDummyApplication() cc := proxy.NewLocalClientCreator(app) mempool := newMempoolWithApp(t, cc) - mempool.FireOnTxsAvailable() + mempool.EnableTxsAvailable() timeoutMS := 500 diff --git a/node/node.go b/node/node.go index c76ae5fa3..0e8cc8fae 100644 --- a/node/node.go +++ b/node/node.go @@ -137,14 +137,13 @@ func NewNode(config *cfg.Config, privValidator *types.PrivValidator, clientCreat // Make MempoolReactor mempoolLogger := logger.With("module", "mempool") - mempool := mempl.NewMempool(config.Mempool, proxyApp.Mempool()) + mempool := mempl.NewMempool(config.Mempool, proxyApp.Mempool(), state.LastBlockHeight) mempool.SetLogger(mempoolLogger) - mempool.Update(state.LastBlockHeight, nil) mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool) mempoolReactor.SetLogger(mempoolLogger) if config.Consensus.NoEmptyBlocks { - mempool.FireOnTxsAvailable() + mempool.EnableTxsAvailable() } // Make ConsensusReactor diff --git a/types/services.go b/types/services.go index f68702ec6..0008b68e7 100644 --- a/types/services.go +++ b/types/services.go @@ -24,8 +24,8 @@ type Mempool interface { Update(height int, txs Txs) Flush() - TxsAvailable() chan struct{} - FireOnTxsAvailable() + TxsAvailable() chan int + EnableTxsAvailable() } type MockMempool struct { @@ -38,8 +38,8 @@ func (m MockMempool) CheckTx(tx Tx, cb func(*abci.Response)) error { return nil func (m MockMempool) Reap(n int) Txs { return Txs{} } func (m MockMempool) Update(height int, txs Txs) {} func (m MockMempool) Flush() {} -func (m MockMempool) TxsAvailable() chan struct{} { return make(chan struct{}) } -func (m MockMempool) FireOnTxsAvailable() {} +func (m MockMempool) TxsAvailable() chan int { return make(chan int) } +func (m MockMempool) EnableTxsAvailable() {} //------------------------------------------------------ // blockstore From b96d28a42bc67e4107a122e57c95c6144bfcf408 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Fri, 28 Jul 2017 23:39:10 -0400 Subject: [PATCH 11/20] test progress in higher round --- consensus/mempool_test.go | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index 843a85aa6..fdb4349ad 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -31,6 +31,37 @@ func TestNoProgressUntilTxsAvailable(t *testing.T) { ensureNewStep(newBlockCh) // commit txs ensureNewStep(newBlockCh) // commit updated app hash ensureNoNewStep(newBlockCh) + +} + +func TestProgressInHigherRound(t *testing.T) { + config := ResetConfig("consensus_mempool_txs_available_test") + config.Consensus.NoEmptyBlocks = true + state, privVals := randGenesisState(1, false, 10) + cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication()) + cs.mempool.EnableTxsAvailable() + height, round := cs.Height, cs.Round + newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1) + newRoundCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewRound(), 1) + timeoutCh := subscribeToEvent(cs.evsw, "tester", types.EventStringTimeoutPropose(), 1) + cs.setProposal = func(proposal *types.Proposal) error { + if cs.Height == 2 && cs.Round == 0 { + // dont set the proposal in round 0 so we timeout and + // go to next round + cs.Logger.Info("Ignoring set proposal at height 2, round 0") + return nil + } + return cs.defaultSetProposal(proposal) + } + startTestRound(cs, height, round) + + ensureNewStep(newRoundCh) // first round at first height + ensureNewStep(newBlockCh) // first block gets committed + ensureNewStep(newRoundCh) // first round at next height + deliverTxsRange(cs, 0, 2) // we deliver txs, but dont set a proposal so we get the next round + <-timeoutCh + ensureNewStep(newRoundCh) // wait for the next round + ensureNewStep(newBlockCh) // now we can commit the block } func deliverTxsRange(cs *ConsensusState, start, end int) { From 530626dab7fd2986d1949ec41a568e18aeff4367 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 20 Jul 2017 15:09:44 -0400 Subject: [PATCH 12/20] broadcast proposer heartbeat msg --- consensus/reactor.go | 30 ++++++++++++++++++++++++++++++ consensus/state.go | 10 ++++++++-- types/events.go | 22 +++++++++++++++++++++- 3 files changed, 59 insertions(+), 3 deletions(-) diff --git a/consensus/reactor.go b/consensus/reactor.go index ce25442fe..cd0ebbd3f 100644 --- a/consensus/reactor.go +++ b/consensus/reactor.go @@ -311,6 +311,21 @@ func (conR *ConsensusReactor) registerEventCallbacks() { edv := data.Unwrap().(types.EventDataVote) conR.broadcastHasVoteMessage(edv.Vote) }) + + types.AddListenerForEvent(conR.evsw, "conR", types.EventStringProposerHeartbeat(), func(data types.TMEventData) { + heartbeat := data.Unwrap().(types.EventDataProposerHeartbeat) + conR.broadcastProposerHeartbeatMessage(heartbeat) + }) +} + +func (conR *ConsensusReactor) broadcastProposerHeartbeatMessage(heartbeat types.EventDataProposerHeartbeat) { + msg := &ProposerHeartbeatMessage{ + Height: heartbeat.Height, + Round: heartbeat.Round, + Proposer: heartbeat.Proposer, + Sequence: heartbeat.Sequence, + } + conR.Switch.Broadcast(StateChannel, struct{ ConsensusMessage }{msg}) } func (conR *ConsensusReactor) broadcastNewRoundStep(rs *RoundState) { @@ -1305,3 +1320,18 @@ type VoteSetBitsMessage struct { func (m *VoteSetBitsMessage) String() string { return fmt.Sprintf("[VSB %v/%02d/%v %v %v]", m.Height, m.Round, m.Type, m.BlockID, m.Votes) } + +//------------------------------------- + +// ProposerHeartbeatMessage is sent to signal that the proposer is alive and waiting for transactions +type ProposerHeartbeatMessage struct { + Height int + Round int + Proposer []byte + Sequence int +} + +// String returns a string representation. +func (m *ProposerHeartbeatMessage) String() string { + return fmt.Sprintf("[HEARTBEAT %v/%02d %X %d]", m.Height, m.Round, m.Proposer, m.Sequence) +} diff --git a/consensus/state.go b/consensus/state.go index 4695a6304..222e2eb10 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -808,11 +808,17 @@ func (cs *ConsensusState) needProofBlock(height int) bool { } func (cs *ConsensusState) proposalHeartbeat() { + counter := 0 + addr := cs.privValidator.GetAddress() for { select { default: - // TODO: broadcast heartbeat - + if cs.evsw != nil { + rs := cs.RoundStateEvent() + heartbeat := types.EventDataProposerHeartbeat{rs, addr, counter} + types.FireEventProposerHeartbeat(cs.evsw, heartbeat) + counter += 1 + } time.Sleep(time.Second) } } diff --git a/types/events.go b/types/events.go index 8c29c4445..625e0aa9d 100644 --- a/types/events.go +++ b/types/events.go @@ -31,6 +31,8 @@ func EventStringRelock() string { return "Relock" } func EventStringTimeoutWait() string { return "TimeoutWait" } func EventStringVote() string { return "Vote" } +func EventStringProposerHeartbeat() string { return "ProposerHeartbeat" } + //---------------------------------------- var ( @@ -39,6 +41,8 @@ var ( EventDataNameTx = "tx" EventDataNameRoundState = "round_state" EventDataNameVote = "vote" + + EventDataNameProposerHeartbeat = "proposer_heartbeat" ) //---------------------------------------- @@ -84,6 +88,8 @@ const ( EventDataTypeRoundState = byte(0x11) EventDataTypeVote = byte(0x12) + + EventDataTypeProposerHeartbeat = byte(0x20) ) var tmEventDataMapper = data.NewMapper(TMEventData{}). @@ -91,7 +97,8 @@ var tmEventDataMapper = data.NewMapper(TMEventData{}). RegisterImplementation(EventDataNewBlockHeader{}, EventDataNameNewBlockHeader, EventDataTypeNewBlockHeader). RegisterImplementation(EventDataTx{}, EventDataNameTx, EventDataTypeTx). RegisterImplementation(EventDataRoundState{}, EventDataNameRoundState, EventDataTypeRoundState). - RegisterImplementation(EventDataVote{}, EventDataNameVote, EventDataTypeVote) + RegisterImplementation(EventDataVote{}, EventDataNameVote, EventDataTypeVote). + RegisterImplementation(EventDataProposerHeartbeat{}, EventDataNameProposerHeartbeat, EventDataTypeProposerHeartbeat) // Most event messages are basic types (a block, a transaction) // but some (an input to a call tx or a receive) are more exotic @@ -115,6 +122,13 @@ type EventDataTx struct { Error string `json:"error"` // this is redundant information for now } +type EventDataProposerHeartbeat struct { + EventDataRoundState + + Proposer []byte `json:"proposer"` + Sequence int `json:"sequence"` +} + // NOTE: This goes into the replay WAL type EventDataRoundState struct { Height int `json:"height"` @@ -135,6 +149,8 @@ func (_ EventDataTx) AssertIsTMEventData() {} func (_ EventDataRoundState) AssertIsTMEventData() {} func (_ EventDataVote) AssertIsTMEventData() {} +func (_ EventDataProposerHeartbeat) AssertIsTMEventData() {} + //---------------------------------------- // Wrappers for type safety @@ -232,3 +248,7 @@ func FireEventRelock(fireable events.Fireable, rs EventDataRoundState) { func FireEventLock(fireable events.Fireable, rs EventDataRoundState) { fireEvent(fireable, EventStringLock(), TMEventData{rs}) } + +func FireEventProposerHeartbeat(fireable events.Fireable, rs EventDataProposerHeartbeat) { + fireEvent(fireable, EventStringProposerHeartbeat(), TMEventData{rs}) +} From 10f81013142d653a14cbc7c36dfa7b2f024c6c2c Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Tue, 25 Jul 2017 11:55:34 -0400 Subject: [PATCH 13/20] fix race --- consensus/state.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/consensus/state.go b/consensus/state.go index 222e2eb10..5d1e106da 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -814,7 +814,7 @@ func (cs *ConsensusState) proposalHeartbeat() { select { default: if cs.evsw != nil { - rs := cs.RoundStateEvent() + rs := cs.GetRoundState().RoundStateEvent() heartbeat := types.EventDataProposerHeartbeat{rs, addr, counter} types.FireEventProposerHeartbeat(cs.evsw, heartbeat) counter += 1 From ab753abfa0ab519f3fc8ca68765a6a632ebc261b Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Sat, 29 Jul 2017 14:15:10 -0400 Subject: [PATCH 14/20] Proposer->Proposal; sign heartbeats --- consensus/byzantine_test.go | 9 +++++++++ consensus/reactor.go | 28 ++++++++++------------------ consensus/state.go | 18 ++++++++++++++++-- types/canonical_json.go | 23 +++++++++++++++++++++++ types/events.go | 21 +++++++++------------ types/priv_validator.go | 7 +++++++ 6 files changed, 74 insertions(+), 32 deletions(-) diff --git a/consensus/byzantine_test.go b/consensus/byzantine_test.go index 3a68a2f5d..94a03c7aa 100644 --- a/consensus/byzantine_test.go +++ b/consensus/byzantine_test.go @@ -293,6 +293,15 @@ func (privVal *ByzantinePrivValidator) SignProposal(chainID string, proposal *ty return nil } +func (privVal *ByzantinePrivValidator) SignHeartbeat(chainID string, heartbeat *types.Heartbeat) error { + privVal.mtx.Lock() + defer privVal.mtx.Unlock() + + // Sign + heartbeat.Signature = privVal.Sign(types.SignBytes(chainID, heartbeat)) + return nil +} + func (privVal *ByzantinePrivValidator) String() string { return Fmt("PrivValidator{%X}", privVal.Address) } diff --git a/consensus/reactor.go b/consensus/reactor.go index cd0ebbd3f..6c8a75441 100644 --- a/consensus/reactor.go +++ b/consensus/reactor.go @@ -312,19 +312,14 @@ func (conR *ConsensusReactor) registerEventCallbacks() { conR.broadcastHasVoteMessage(edv.Vote) }) - types.AddListenerForEvent(conR.evsw, "conR", types.EventStringProposerHeartbeat(), func(data types.TMEventData) { - heartbeat := data.Unwrap().(types.EventDataProposerHeartbeat) - conR.broadcastProposerHeartbeatMessage(heartbeat) + types.AddListenerForEvent(conR.evsw, "conR", types.EventStringProposalHeartbeat(), func(data types.TMEventData) { + heartbeat := data.Unwrap().(types.EventDataProposalHeartbeat) + conR.broadcastProposalHeartbeatMessage(heartbeat) }) } -func (conR *ConsensusReactor) broadcastProposerHeartbeatMessage(heartbeat types.EventDataProposerHeartbeat) { - msg := &ProposerHeartbeatMessage{ - Height: heartbeat.Height, - Round: heartbeat.Round, - Proposer: heartbeat.Proposer, - Sequence: heartbeat.Sequence, - } +func (conR *ConsensusReactor) broadcastProposalHeartbeatMessage(heartbeat types.EventDataProposalHeartbeat) { + msg := &ProposalHeartbeatMessage{heartbeat.Heartbeat} conR.Switch.Broadcast(StateChannel, struct{ ConsensusMessage }{msg}) } @@ -1323,15 +1318,12 @@ func (m *VoteSetBitsMessage) String() string { //------------------------------------- -// ProposerHeartbeatMessage is sent to signal that the proposer is alive and waiting for transactions -type ProposerHeartbeatMessage struct { - Height int - Round int - Proposer []byte - Sequence int +// ProposalHeartbeatMessage is sent to signal that the proposer is alive and waiting for transactions +type ProposalHeartbeatMessage struct { + Heartbeat *types.Heartbeat } // String returns a string representation. -func (m *ProposerHeartbeatMessage) String() string { - return fmt.Sprintf("[HEARTBEAT %v/%02d %X %d]", m.Height, m.Round, m.Proposer, m.Sequence) +func (m *ProposalHeartbeatMessage) String() string { + return fmt.Sprintf("[HEARTBEAT %v]", m.Heartbeat) } diff --git a/consensus/state.go b/consensus/state.go index 5d1e106da..2d0a9a21d 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -181,6 +181,7 @@ type PrivValidator interface { GetAddress() []byte SignVote(chainID string, vote *types.Vote) error SignProposal(chainID string, proposal *types.Proposal) error + SignHeartbeat(chainID string, heartbeat *types.Heartbeat) error } // ConsensusState handles execution of the consensus algorithm. @@ -810,13 +811,26 @@ func (cs *ConsensusState) needProofBlock(height int) bool { func (cs *ConsensusState) proposalHeartbeat() { counter := 0 addr := cs.privValidator.GetAddress() + valIndex, v := cs.Validators.GetByAddress(addr) + if v == nil { + // not a validator + valIndex = -1 + } for { select { default: if cs.evsw != nil { rs := cs.GetRoundState().RoundStateEvent() - heartbeat := types.EventDataProposerHeartbeat{rs, addr, counter} - types.FireEventProposerHeartbeat(cs.evsw, heartbeat) + heartbeat := &types.Heartbeat{ + Height: rs.Height, + Round: rs.Round, + Sequence: counter, + ValidatorAddress: addr, + ValidatorIndex: valIndex, + } + cs.privValidator.SignHeartbeat(cs.state.ChainID, heartbeat) + heartbeatEvent := types.EventDataProposalHeartbeat{heartbeat} + types.FireEventProposalHeartbeat(cs.evsw, heartbeatEvent) counter += 1 } time.Sleep(time.Second) diff --git a/types/canonical_json.go b/types/canonical_json.go index 2e8583a4a..5f1a0acaa 100644 --- a/types/canonical_json.go +++ b/types/canonical_json.go @@ -31,6 +31,14 @@ type CanonicalJSONVote struct { Type byte `json:"type"` } +type CanonicalJSONHeartbeat struct { + Height int `json:"height"` + Round int `json:"round"` + Sequence int `json:"sequence"` + ValidatorAddress data.Bytes `json:"validator_address"` + ValidatorIndex int `json:"validator_index"` +} + //------------------------------------ // Messages including a "chain id" can only be applied to one chain, hence "Once" @@ -44,6 +52,11 @@ type CanonicalJSONOnceVote struct { Vote CanonicalJSONVote `json:"vote"` } +type CanonicalJSONOnceHeartbeat struct { + ChainID string `json:"chain_id"` + Heartbeat CanonicalJSONHeartbeat `json:"heartbeat"` +} + //----------------------------------- // Canonicalize the structs @@ -79,3 +92,13 @@ func CanonicalVote(vote *Vote) CanonicalJSONVote { vote.Type, } } + +func CanonicalHeartbeat(heartbeat *Heartbeat) CanonicalJSONHeartbeat { + return CanonicalJSONHeartbeat{ + heartbeat.Height, + heartbeat.Round, + heartbeat.Sequence, + heartbeat.ValidatorAddress, + heartbeat.ValidatorIndex, + } +} diff --git a/types/events.go b/types/events.go index 625e0aa9d..79e17fe0f 100644 --- a/types/events.go +++ b/types/events.go @@ -31,7 +31,7 @@ func EventStringRelock() string { return "Relock" } func EventStringTimeoutWait() string { return "TimeoutWait" } func EventStringVote() string { return "Vote" } -func EventStringProposerHeartbeat() string { return "ProposerHeartbeat" } +func EventStringProposalHeartbeat() string { return "ProposalHeartbeat" } //---------------------------------------- @@ -42,7 +42,7 @@ var ( EventDataNameRoundState = "round_state" EventDataNameVote = "vote" - EventDataNameProposerHeartbeat = "proposer_heartbeat" + EventDataNameProposalHeartbeat = "proposer_heartbeat" ) //---------------------------------------- @@ -89,7 +89,7 @@ const ( EventDataTypeRoundState = byte(0x11) EventDataTypeVote = byte(0x12) - EventDataTypeProposerHeartbeat = byte(0x20) + EventDataTypeProposalHeartbeat = byte(0x20) ) var tmEventDataMapper = data.NewMapper(TMEventData{}). @@ -98,7 +98,7 @@ var tmEventDataMapper = data.NewMapper(TMEventData{}). RegisterImplementation(EventDataTx{}, EventDataNameTx, EventDataTypeTx). RegisterImplementation(EventDataRoundState{}, EventDataNameRoundState, EventDataTypeRoundState). RegisterImplementation(EventDataVote{}, EventDataNameVote, EventDataTypeVote). - RegisterImplementation(EventDataProposerHeartbeat{}, EventDataNameProposerHeartbeat, EventDataTypeProposerHeartbeat) + RegisterImplementation(EventDataProposalHeartbeat{}, EventDataNameProposalHeartbeat, EventDataTypeProposalHeartbeat) // Most event messages are basic types (a block, a transaction) // but some (an input to a call tx or a receive) are more exotic @@ -122,11 +122,8 @@ type EventDataTx struct { Error string `json:"error"` // this is redundant information for now } -type EventDataProposerHeartbeat struct { - EventDataRoundState - - Proposer []byte `json:"proposer"` - Sequence int `json:"sequence"` +type EventDataProposalHeartbeat struct { + Heartbeat *Heartbeat } // NOTE: This goes into the replay WAL @@ -149,7 +146,7 @@ func (_ EventDataTx) AssertIsTMEventData() {} func (_ EventDataRoundState) AssertIsTMEventData() {} func (_ EventDataVote) AssertIsTMEventData() {} -func (_ EventDataProposerHeartbeat) AssertIsTMEventData() {} +func (_ EventDataProposalHeartbeat) AssertIsTMEventData() {} //---------------------------------------- // Wrappers for type safety @@ -249,6 +246,6 @@ func FireEventLock(fireable events.Fireable, rs EventDataRoundState) { fireEvent(fireable, EventStringLock(), TMEventData{rs}) } -func FireEventProposerHeartbeat(fireable events.Fireable, rs EventDataProposerHeartbeat) { - fireEvent(fireable, EventStringProposerHeartbeat(), TMEventData{rs}) +func FireEventProposalHeartbeat(fireable events.Fireable, rs EventDataProposalHeartbeat) { + fireEvent(fireable, EventStringProposalHeartbeat(), TMEventData{rs}) } diff --git a/types/priv_validator.go b/types/priv_validator.go index 8c9a011ca..690824934 100644 --- a/types/priv_validator.go +++ b/types/priv_validator.go @@ -252,6 +252,13 @@ func (privVal *PrivValidator) signBytesHRS(height, round int, step int8, signByt } +func (privVal *PrivValidator) SignHeartbeat(chainID string, heartbeat *Heartbeat) error { + privVal.mtx.Lock() + defer privVal.mtx.Unlock() + heartbeat.Signature = privVal.Sign(SignBytes(chainID, heartbeat)) + return nil +} + func (privVal *PrivValidator) String() string { return fmt.Sprintf("PrivValidator{%v LH:%v, LR:%v, LS:%v}", privVal.Address, privVal.LastHeight, privVal.LastRound, privVal.LastStep) } From b8ac67e240148102dc4e1d4064ddb1fd7e6e2ab0 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 3 Aug 2017 13:25:26 -0400 Subject: [PATCH 15/20] some fixes --- consensus/reactor.go | 2 +- consensus/state.go | 37 ++++++++++++++++++------------------- 2 files changed, 19 insertions(+), 20 deletions(-) diff --git a/consensus/reactor.go b/consensus/reactor.go index 6c8a75441..3e12a3148 100644 --- a/consensus/reactor.go +++ b/consensus/reactor.go @@ -1318,7 +1318,7 @@ func (m *VoteSetBitsMessage) String() string { //------------------------------------- -// ProposalHeartbeatMessage is sent to signal that the proposer is alive and waiting for transactions +// ProposalHeartbeatMessage is sent to signal that a node is alive and waiting for transactions for a proposal. type ProposalHeartbeatMessage struct { Heartbeat *types.Heartbeat } diff --git a/consensus/state.go b/consensus/state.go index 2d0a9a21d..e6fd665a5 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -788,7 +788,7 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { // we may need an empty "proof" block, and enterPropose immediately. waitForTxs := cs.config.NoEmptyBlocks && round == 0 && !cs.needProofBlock(height) if waitForTxs { - go cs.proposalHeartbeat() + go cs.proposalHeartbeat(height, round) } else { cs.enterPropose(height, round) } @@ -808,7 +808,7 @@ func (cs *ConsensusState) needProofBlock(height int) bool { return false } -func (cs *ConsensusState) proposalHeartbeat() { +func (cs *ConsensusState) proposalHeartbeat(height, round int) { counter := 0 addr := cs.privValidator.GetAddress() valIndex, v := cs.Validators.GetByAddress(addr) @@ -817,24 +817,23 @@ func (cs *ConsensusState) proposalHeartbeat() { valIndex = -1 } for { - select { - default: - if cs.evsw != nil { - rs := cs.GetRoundState().RoundStateEvent() - heartbeat := &types.Heartbeat{ - Height: rs.Height, - Round: rs.Round, - Sequence: counter, - ValidatorAddress: addr, - ValidatorIndex: valIndex, - } - cs.privValidator.SignHeartbeat(cs.state.ChainID, heartbeat) - heartbeatEvent := types.EventDataProposalHeartbeat{heartbeat} - types.FireEventProposalHeartbeat(cs.evsw, heartbeatEvent) - counter += 1 - } - time.Sleep(time.Second) + rs := cs.GetRoundState() + // if we've already moved on, no need to send more heartbeats + if rs.Step > RoundStepNewRound || rs.Round > round || rs.Height > height { + return } + heartbeat := &types.Heartbeat{ + Height: rs.Height, + Round: rs.Round, + Sequence: counter, + ValidatorAddress: addr, + ValidatorIndex: valIndex, + } + cs.privValidator.SignHeartbeat(cs.state.ChainID, heartbeat) + heartbeatEvent := types.EventDataProposalHeartbeat{heartbeat} + types.FireEventProposalHeartbeat(cs.evsw, heartbeatEvent) + counter += 1 + time.Sleep(time.Second) } } From d0965cca05981e10609ddf19865e4f8620acf5b6 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Thu, 3 Aug 2017 13:58:17 -0400 Subject: [PATCH 16/20] forgot heartbeat file --- types/heartbeat.go | 42 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) create mode 100644 types/heartbeat.go diff --git a/types/heartbeat.go b/types/heartbeat.go new file mode 100644 index 000000000..378f6202b --- /dev/null +++ b/types/heartbeat.go @@ -0,0 +1,42 @@ +package types + +import ( + "fmt" + "io" + + "github.com/tendermint/go-crypto" + "github.com/tendermint/go-wire" + "github.com/tendermint/go-wire/data" + cmn "github.com/tendermint/tmlibs/common" +) + +type Heartbeat struct { + ValidatorAddress data.Bytes `json:"validator_address"` + ValidatorIndex int `json:"validator_index"` + Height int `json:"height"` + Round int `json:"round"` + Sequence int `json:"sequence"` + Signature crypto.Signature `json:"signature"` +} + +func (heartbeat *Heartbeat) WriteSignBytes(chainID string, w io.Writer, n *int, err *error) { + wire.WriteJSON(CanonicalJSONOnceHeartbeat{ + chainID, + CanonicalHeartbeat(heartbeat), + }, w, n, err) +} + +func (heartbeat *Heartbeat) Copy() *Heartbeat { + heartbeatCopy := *heartbeat + return &heartbeatCopy +} + +func (heartbeat *Heartbeat) String() string { + if heartbeat == nil { + return "nil-heartbeat" + } + + return fmt.Sprintf("Heartbeat{%v:%X %v/%02d (%v) %v}", + heartbeat.ValidatorIndex, cmn.Fingerprint(heartbeat.ValidatorAddress), + heartbeat.Height, heartbeat.Round, heartbeat.Sequence, heartbeat.Signature) +} From fb47ca6d35e43e9254c7d94e9694472001320dbb Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Fri, 4 Aug 2017 21:36:11 -0400 Subject: [PATCH 17/20] fixes from review --- consensus/state.go | 6 +++++- mempool/mempool.go | 6 +++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/consensus/state.go b/consensus/state.go index e6fd665a5..e3a153651 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -24,6 +24,10 @@ import ( //----------------------------------------------------------------------------- // Config +const ( + proposalHeartbeatIntervalSeconds = 2 +) + //----------------------------------------------------------------------------- // Errors @@ -833,7 +837,7 @@ func (cs *ConsensusState) proposalHeartbeat(height, round int) { heartbeatEvent := types.EventDataProposalHeartbeat{heartbeat} types.FireEventProposalHeartbeat(cs.evsw, heartbeatEvent) counter += 1 - time.Sleep(time.Second) + time.Sleep(proposalHeartbeatIntervalSeconds * time.Second) } } diff --git a/mempool/mempool.go b/mempool/mempool.go index 9f9824f9e..6113910b7 100644 --- a/mempool/mempool.go +++ b/mempool/mempool.go @@ -224,7 +224,7 @@ func (mem *Mempool) resCbNormal(req *abci.Request, res *abci.Response) { tx: req.GetCheckTx().Tx, } mem.txs.PushBack(memTx) - mem.notifyIfTxsAvailable() + mem.notifyTxsAvailable() } else { // ignore bad transaction mem.logger.Info("Bad Transaction", "res", r) @@ -267,7 +267,7 @@ func (mem *Mempool) resCbRecheck(req *abci.Request, res *abci.Response) { atomic.StoreInt32(&mem.rechecking, 0) mem.logger.Info("Done rechecking txs") - mem.notifyIfTxsAvailable() + mem.notifyTxsAvailable() } default: // ignore other messages @@ -281,7 +281,7 @@ func (mem *Mempool) TxsAvailable() chan int { return mem.txsAvailable } -func (mem *Mempool) notifyIfTxsAvailable() { +func (mem *Mempool) notifyTxsAvailable() { if mem.Size() == 0 { panic("notified txs available but mempool is empty!") } From 37f139047334dabffdc7cbffe376f4601a157ad7 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Fri, 4 Aug 2017 21:46:17 -0400 Subject: [PATCH 18/20] CreateEmptyBlocks and CreateEmptyBlocksInterval --- CHANGELOG.md | 3 ++- cmd/tendermint/commands/run_node.go | 2 +- config/config.go | 22 ++++++++++++++++++---- consensus/common_test.go | 2 +- consensus/mempool_test.go | 4 ++-- consensus/state.go | 12 +++++++++--- node/node.go | 2 +- 7 files changed, 34 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c233a264..0be66e3d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,8 @@ ## 0.10.3 (TBD) FEATURES: -- New `--consensus.no_empty_blocks` flag prevents the creation of blocks until there are transactions available +- New `--consensus.create_empty_blocks` flag; when set to false, only creates blocks when there are txs or when the AppHash changes +- New `consensus.create_empty_blocks_interval` config option; when greater than 0, will create an empty block after waiting that many seconds ## 0.10.2 (July 10, 2017) diff --git a/cmd/tendermint/commands/run_node.go b/cmd/tendermint/commands/run_node.go index b9df28ab8..e92911da2 100644 --- a/cmd/tendermint/commands/run_node.go +++ b/cmd/tendermint/commands/run_node.go @@ -47,7 +47,7 @@ func AddNodeFlags(cmd *cobra.Command) { cmd.Flags().Bool("p2p.pex", config.P2P.PexReactor, "Enable Peer-Exchange (dev feature)") // consensus flags - cmd.Flags().Bool("consensus.no_empty_blocks", config.Consensus.NoEmptyBlocks, "Only produce blocks when there are txs and when the AppHash changes") + cmd.Flags().Bool("consensus.create_empty_blocks", config.Consensus.CreateEmptyBlocks, "Set this to false to only produce blocks when there are txs or when the AppHash changes") } // Users wishing to: diff --git a/config/config.go b/config/config.go index 323b36a25..a16fa7103 100644 --- a/config/config.go +++ b/config/config.go @@ -301,9 +301,12 @@ type ConsensusConfig struct { SkipTimeoutCommit bool `mapstructure:"skip_timeout_commit"` // BlockSize - MaxBlockSizeTxs int `mapstructure:"max_block_size_txs"` - MaxBlockSizeBytes int `mapstructure:"max_block_size_bytes"` - NoEmptyBlocks bool `mapstructure:"no_empty_blocks"` + MaxBlockSizeTxs int `mapstructure:"max_block_size_txs"` + MaxBlockSizeBytes int `mapstructure:"max_block_size_bytes"` + + // EmptyBlocks mode and possible interval between empty blocks in seconds + CreateEmptyBlocks bool `mapstructure:"create_empty_blocks"` + CreateEmptyBlocksInterval int `mapstructure:"create_empty_blocks_interval"` // TODO: This probably shouldn't be exposed but it makes it // easy to write tests for the wal/replay @@ -314,6 +317,16 @@ type ConsensusConfig struct { PeerQueryMaj23SleepDuration int `mapstructure:"peer_query_maj23_sleep_duration"` } +// WaitForTxs returns true if the consensus should wait for transactions before entering the propose step +func (cfg *ConsensusConfig) WaitForTxs() bool { + return !cfg.CreateEmptyBlocks || cfg.CreateEmptyBlocksInterval > 0 +} + +// EmptyBlocks returns the amount of time to wait before proposing an empty block or starting the propose timer if there are no txs available +func (cfg *ConsensusConfig) EmptyBlocks() time.Duration { + return time.Duration(cfg.CreateEmptyBlocksInterval) * time.Second +} + // Propose returns the amount of time to wait for a proposal func (cfg *ConsensusConfig) Propose(round int) time.Duration { return time.Duration(cfg.TimeoutPropose+cfg.TimeoutProposeDelta*round) * time.Millisecond @@ -359,7 +372,8 @@ func DefaultConsensusConfig() *ConsensusConfig { SkipTimeoutCommit: false, MaxBlockSizeTxs: 10000, MaxBlockSizeBytes: 1, // TODO - NoEmptyBlocks: false, + CreateEmptyBlocks: true, + CreateEmptyBlocksInterval: 0, BlockPartSize: types.DefaultBlockPartSize, // TODO: we shouldnt be importing types PeerGossipSleepDuration: 100, PeerQueryMaj23SleepDuration: 2000, diff --git a/consensus/common_test.go b/consensus/common_test.go index 6b96ec311..8197d6d65 100644 --- a/consensus/common_test.go +++ b/consensus/common_test.go @@ -242,7 +242,7 @@ func newConsensusStateWithConfig(thisConfig *cfg.Config, state *sm.State, pv *ty // Make Mempool mempool := mempl.NewMempool(thisConfig.Mempool, proxyAppConnMem, 0) mempool.SetLogger(log.TestingLogger().With("module", "mempool")) - if thisConfig.Consensus.NoEmptyBlocks { + if thisConfig.Consensus.WaitForTxs() { mempool.EnableTxsAvailable() } diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index fdb4349ad..e4069b876 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -17,7 +17,7 @@ func init() { func TestNoProgressUntilTxsAvailable(t *testing.T) { config := ResetConfig("consensus_mempool_txs_available_test") - config.Consensus.NoEmptyBlocks = true + config.Consensus.CreateEmptyBlocks = false state, privVals := randGenesisState(1, false, 10) cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication()) cs.mempool.EnableTxsAvailable() @@ -36,7 +36,7 @@ func TestNoProgressUntilTxsAvailable(t *testing.T) { func TestProgressInHigherRound(t *testing.T) { config := ResetConfig("consensus_mempool_txs_available_test") - config.Consensus.NoEmptyBlocks = true + config.Consensus.CreateEmptyBlocks = false state, privVals := randGenesisState(1, false, 10) cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication()) cs.mempool.EnableTxsAvailable() diff --git a/consensus/state.go b/consensus/state.go index e3a153651..0a763f914 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -718,6 +718,8 @@ func (cs *ConsensusState) handleTimeout(ti timeoutInfo, rs RoundState) { // NewRound event fired from enterNewRound. // XXX: should we fire timeout here (for timeout commit)? cs.enterNewRound(ti.Height, 0) + case RoundStepNewRound: + cs.enterPropose(ti.Height, 0) case RoundStepPropose: types.FireEventTimeoutPropose(cs.evsw, cs.RoundStateEvent()) cs.enterPrevote(ti.Height, ti.Round) @@ -790,8 +792,11 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { // Wait for txs to be available in the mempool // before we enterPropose in round 0. If the last block changed the app hash, // we may need an empty "proof" block, and enterPropose immediately. - waitForTxs := cs.config.NoEmptyBlocks && round == 0 && !cs.needProofBlock(height) + waitForTxs := cs.config.WaitForTxs() && round == 0 && !cs.needProofBlock(height) if waitForTxs { + if cs.config.CreateEmptyBlocksInterval > 0 { + cs.scheduleTimeout(cs.config.EmptyBlocks(), height, round, RoundStepNewRound) + } go cs.proposalHeartbeat(height, round) } else { cs.enterPropose(height, round) @@ -841,8 +846,9 @@ func (cs *ConsensusState) proposalHeartbeat(height, round int) { } } -// Enter (!NoEmptyBlocks): from enterNewRound(height,round) -// Enter (NoEmptyBlocks) : after enterNewRound(height,round), once txs are in the mempool +// Enter (CreateEmptyBlocks): from enterNewRound(height,round) +// Enter (CreateEmptyBlocks, CreateEmptyBlocksInterval > 0 ): after enterNewRound(height,round), after timeout of CreateEmptyBlocksInterval +// Enter (!CreateEmptyBlocks) : after enterNewRound(height,round), once txs are in the mempool func (cs *ConsensusState) enterPropose(height int, round int) { if cs.Height != height || round < cs.Round || (cs.Round == round && RoundStepPropose <= cs.Step) { cs.Logger.Debug(cmn.Fmt("enterPropose(%v/%v): Invalid args. Current step: %v/%v/%v", height, round, cs.Height, cs.Round, cs.Step)) diff --git a/node/node.go b/node/node.go index 0e8cc8fae..e55731f4f 100644 --- a/node/node.go +++ b/node/node.go @@ -142,7 +142,7 @@ func NewNode(config *cfg.Config, privValidator *types.PrivValidator, clientCreat mempoolReactor := mempl.NewMempoolReactor(config.Mempool, mempool) mempoolReactor.SetLogger(mempoolLogger) - if config.Consensus.NoEmptyBlocks { + if config.Consensus.WaitForTxs() { mempool.EnableTxsAvailable() } From d2f3e9faf41409e3e460b3a9759116eb471d4fae Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Tue, 8 Aug 2017 16:35:25 -0400 Subject: [PATCH 19/20] test CreateEmptyBlocksInterval --- consensus/common_test.go | 6 +++--- consensus/mempool_test.go | 15 +++++++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/consensus/common_test.go b/consensus/common_test.go index 8197d6d65..bf805e095 100644 --- a/consensus/common_test.go +++ b/consensus/common_test.go @@ -31,7 +31,7 @@ import ( // genesis, chain_id, priv_val var config *cfg.Config // NOTE: must be reset for each _test.go file -var ensureTimeout = time.Duration(2) +var ensureTimeout = time.Second * 2 func ensureDir(dir string, mode os.FileMode) { if err := EnsureDir(dir, mode); err != nil { @@ -297,7 +297,7 @@ func randConsensusState(nValidators int) (*ConsensusState, []*validatorStub) { //------------------------------------------------------------------------------- func ensureNoNewStep(stepCh chan interface{}) { - timer := time.NewTimer(ensureTimeout * time.Second) + timer := time.NewTimer(ensureTimeout) select { case <-timer.C: break @@ -307,7 +307,7 @@ func ensureNoNewStep(stepCh chan interface{}) { } func ensureNewStep(stepCh chan interface{}) { - timer := time.NewTimer(ensureTimeout * time.Second) + timer := time.NewTimer(ensureTimeout) select { case <-timer.C: panic("We shouldnt be stuck waiting") diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index e4069b876..0f726b39e 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -34,6 +34,21 @@ func TestNoProgressUntilTxsAvailable(t *testing.T) { } +func TestProgressAfterCreateEmptyBlocksInterval(t *testing.T) { + config := ResetConfig("consensus_mempool_txs_available_test") + config.Consensus.CreateEmptyBlocksInterval = int(ensureTimeout.Seconds()) + state, privVals := randGenesisState(1, false, 10) + cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication()) + cs.mempool.EnableTxsAvailable() + height, round := cs.Height, cs.Round + newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1) + startTestRound(cs, height, round) + + ensureNewStep(newBlockCh) // first block gets committed + ensureNoNewStep(newBlockCh) // then we dont make a block ... + ensureNewStep(newBlockCh) // until the CreateEmptyBlocksInterval has passed +} + func TestProgressInHigherRound(t *testing.T) { config := ResetConfig("consensus_mempool_txs_available_test") config.Consensus.CreateEmptyBlocks = false From 0bf66deb3cd04a7de25375946efefe0e228acb60 Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Tue, 8 Aug 2017 17:09:04 -0400 Subject: [PATCH 20/20] fixes from review --- config/config.go | 2 +- consensus/state.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/config/config.go b/config/config.go index a16fa7103..cc1c7d719 100644 --- a/config/config.go +++ b/config/config.go @@ -323,7 +323,7 @@ func (cfg *ConsensusConfig) WaitForTxs() bool { } // EmptyBlocks returns the amount of time to wait before proposing an empty block or starting the propose timer if there are no txs available -func (cfg *ConsensusConfig) EmptyBlocks() time.Duration { +func (cfg *ConsensusConfig) EmptyBlocksInterval() time.Duration { return time.Duration(cfg.CreateEmptyBlocksInterval) * time.Second } diff --git a/consensus/state.go b/consensus/state.go index 0a763f914..48c91d273 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -795,7 +795,7 @@ func (cs *ConsensusState) enterNewRound(height int, round int) { waitForTxs := cs.config.WaitForTxs() && round == 0 && !cs.needProofBlock(height) if waitForTxs { if cs.config.CreateEmptyBlocksInterval > 0 { - cs.scheduleTimeout(cs.config.EmptyBlocks(), height, round, RoundStepNewRound) + cs.scheduleTimeout(cs.config.EmptyBlocksInterval(), height, round, RoundStepNewRound) } go cs.proposalHeartbeat(height, round) } else {