You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

219 lines
6.4 KiB

8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
8 years ago
  1. package consensus
  2. import (
  3. "encoding/binary"
  4. "testing"
  5. "time"
  6. abci "github.com/tendermint/abci/types"
  7. "github.com/tendermint/tendermint/types"
  8. . "github.com/tendermint/tmlibs/common"
  9. )
  10. func init() {
  11. config = ResetConfig("consensus_mempool_test")
  12. }
  13. func TestNoProgressUntilTxsAvailable(t *testing.T) {
  14. config := ResetConfig("consensus_mempool_txs_available_test")
  15. config.Consensus.CreateEmptyBlocks = false
  16. state, privVals := randGenesisState(1, false, 10)
  17. cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication())
  18. cs.mempool.EnableTxsAvailable()
  19. height, round := cs.Height, cs.Round
  20. newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1)
  21. startTestRound(cs, height, round)
  22. ensureNewStep(newBlockCh) // first block gets committed
  23. ensureNoNewStep(newBlockCh)
  24. deliverTxsRange(cs, 0, 2)
  25. ensureNewStep(newBlockCh) // commit txs
  26. ensureNewStep(newBlockCh) // commit updated app hash
  27. ensureNoNewStep(newBlockCh)
  28. }
  29. func TestProgressAfterCreateEmptyBlocksInterval(t *testing.T) {
  30. config := ResetConfig("consensus_mempool_txs_available_test")
  31. config.Consensus.CreateEmptyBlocksInterval = int(ensureTimeout.Seconds())
  32. state, privVals := randGenesisState(1, false, 10)
  33. cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication())
  34. cs.mempool.EnableTxsAvailable()
  35. height, round := cs.Height, cs.Round
  36. newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1)
  37. startTestRound(cs, height, round)
  38. ensureNewStep(newBlockCh) // first block gets committed
  39. ensureNoNewStep(newBlockCh) // then we dont make a block ...
  40. ensureNewStep(newBlockCh) // until the CreateEmptyBlocksInterval has passed
  41. }
  42. func TestProgressInHigherRound(t *testing.T) {
  43. config := ResetConfig("consensus_mempool_txs_available_test")
  44. config.Consensus.CreateEmptyBlocks = false
  45. state, privVals := randGenesisState(1, false, 10)
  46. cs := newConsensusStateWithConfig(config, state, privVals[0], NewCounterApplication())
  47. cs.mempool.EnableTxsAvailable()
  48. height, round := cs.Height, cs.Round
  49. newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1)
  50. newRoundCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewRound(), 1)
  51. timeoutCh := subscribeToEvent(cs.evsw, "tester", types.EventStringTimeoutPropose(), 1)
  52. cs.setProposal = func(proposal *types.Proposal) error {
  53. if cs.Height == 2 && cs.Round == 0 {
  54. // dont set the proposal in round 0 so we timeout and
  55. // go to next round
  56. cs.Logger.Info("Ignoring set proposal at height 2, round 0")
  57. return nil
  58. }
  59. return cs.defaultSetProposal(proposal)
  60. }
  61. startTestRound(cs, height, round)
  62. ensureNewStep(newRoundCh) // first round at first height
  63. ensureNewStep(newBlockCh) // first block gets committed
  64. ensureNewStep(newRoundCh) // first round at next height
  65. deliverTxsRange(cs, 0, 2) // we deliver txs, but dont set a proposal so we get the next round
  66. <-timeoutCh
  67. ensureNewStep(newRoundCh) // wait for the next round
  68. ensureNewStep(newBlockCh) // now we can commit the block
  69. }
  70. func deliverTxsRange(cs *ConsensusState, start, end int) {
  71. // Deliver some txs.
  72. for i := start; i < end; i++ {
  73. txBytes := make([]byte, 8)
  74. binary.BigEndian.PutUint64(txBytes, uint64(i))
  75. err := cs.mempool.CheckTx(txBytes, nil)
  76. if err != nil {
  77. panic(Fmt("Error after CheckTx: %v", err))
  78. }
  79. }
  80. }
  81. func TestTxConcurrentWithCommit(t *testing.T) {
  82. state, privVals := randGenesisState(1, false, 10)
  83. cs := newConsensusState(state, privVals[0], NewCounterApplication())
  84. height, round := cs.Height, cs.Round
  85. newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1)
  86. NTxs := 10000
  87. go deliverTxsRange(cs, 0, NTxs)
  88. startTestRound(cs, height, round)
  89. ticker := time.NewTicker(time.Second * 20)
  90. for nTxs := 0; nTxs < NTxs; {
  91. select {
  92. case b := <-newBlockCh:
  93. nTxs += b.(types.TMEventData).Unwrap().(types.EventDataNewBlock).Block.Header.NumTxs
  94. case <-ticker.C:
  95. panic("Timed out waiting to commit blocks with transactions")
  96. }
  97. }
  98. }
  99. func TestRmBadTx(t *testing.T) {
  100. state, privVals := randGenesisState(1, false, 10)
  101. app := NewCounterApplication()
  102. cs := newConsensusState(state, privVals[0], app)
  103. // increment the counter by 1
  104. txBytes := make([]byte, 8)
  105. binary.BigEndian.PutUint64(txBytes, uint64(0))
  106. app.DeliverTx(txBytes)
  107. app.Commit()
  108. ch := make(chan struct{})
  109. cbCh := make(chan struct{})
  110. go func() {
  111. // Try to send the tx through the mempool.
  112. // CheckTx should not err, but the app should return a bad abci code
  113. // and the tx should get removed from the pool
  114. err := cs.mempool.CheckTx(txBytes, func(r *abci.Response) {
  115. if r.GetCheckTx().Code != abci.CodeType_BadNonce {
  116. t.Fatalf("expected checktx to return bad nonce, got %v", r)
  117. }
  118. cbCh <- struct{}{}
  119. })
  120. if err != nil {
  121. t.Fatal("Error after CheckTx: %v", err)
  122. }
  123. // check for the tx
  124. for {
  125. time.Sleep(time.Second)
  126. txs := cs.mempool.Reap(1)
  127. if len(txs) == 0 {
  128. ch <- struct{}{}
  129. return
  130. }
  131. }
  132. }()
  133. // Wait until the tx returns
  134. ticker := time.After(time.Second * 5)
  135. select {
  136. case <-cbCh:
  137. // success
  138. case <-ticker:
  139. t.Fatalf("Timed out waiting for tx to return")
  140. }
  141. // Wait until the tx is removed
  142. ticker = time.After(time.Second * 5)
  143. select {
  144. case <-ch:
  145. // success
  146. case <-ticker:
  147. t.Fatalf("Timed out waiting for tx to be removed")
  148. }
  149. }
  150. // CounterApplication that maintains a mempool state and resets it upon commit
  151. type CounterApplication struct {
  152. abci.BaseApplication
  153. txCount int
  154. mempoolTxCount int
  155. }
  156. func NewCounterApplication() *CounterApplication {
  157. return &CounterApplication{}
  158. }
  159. func (app *CounterApplication) Info() abci.ResponseInfo {
  160. return abci.ResponseInfo{Data: Fmt("txs:%v", app.txCount)}
  161. }
  162. func (app *CounterApplication) DeliverTx(tx []byte) abci.Result {
  163. return runTx(tx, &app.txCount)
  164. }
  165. func (app *CounterApplication) CheckTx(tx []byte) abci.Result {
  166. return runTx(tx, &app.mempoolTxCount)
  167. }
  168. func runTx(tx []byte, countPtr *int) abci.Result {
  169. count := *countPtr
  170. tx8 := make([]byte, 8)
  171. copy(tx8[len(tx8)-len(tx):], tx)
  172. txValue := binary.BigEndian.Uint64(tx8)
  173. if txValue != uint64(count) {
  174. return abci.ErrBadNonce.AppendLog(Fmt("Invalid nonce. Expected %v, got %v", count, txValue))
  175. }
  176. *countPtr += 1
  177. return abci.OK
  178. }
  179. func (app *CounterApplication) Commit() abci.Result {
  180. app.mempoolTxCount = app.txCount
  181. if app.txCount == 0 {
  182. return abci.OK
  183. } else {
  184. hash := make([]byte, 8)
  185. binary.BigEndian.PutUint64(hash, uint64(app.txCount))
  186. return abci.NewResultOK(hash, "")
  187. }
  188. }