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.

166 lines
4.0 KiB

  1. package consensus
  2. import (
  3. "encoding/binary"
  4. "testing"
  5. "time"
  6. "github.com/tendermint/tendermint/config/tendermint_test"
  7. "github.com/tendermint/tendermint/types"
  8. tmsp "github.com/tendermint/tmsp/types"
  9. . "github.com/tendermint/go-common"
  10. )
  11. func init() {
  12. config = tendermint_test.ResetConfig("consensus_mempool_test")
  13. }
  14. func TestTxConcurrentWithCommit(t *testing.T) {
  15. state, privVals := randGenesisState(1, false, 10)
  16. cs := newConsensusState(state, privVals[0], NewCounterApplication())
  17. height, round := cs.Height, cs.Round
  18. newBlockCh := subscribeToEvent(cs.evsw, "tester", types.EventStringNewBlock(), 1)
  19. appendTxsRange := func(start, end int) {
  20. // Append some txs.
  21. for i := start; i < end; i++ {
  22. txBytes := make([]byte, 8)
  23. binary.BigEndian.PutUint64(txBytes, uint64(i))
  24. err := cs.mempool.CheckTx(txBytes, nil)
  25. if err != nil {
  26. panic(Fmt("Error after CheckTx: %v", err))
  27. }
  28. // time.Sleep(time.Microsecond * time.Duration(rand.Int63n(3000)))
  29. }
  30. }
  31. NTxs := 10000
  32. go appendTxsRange(0, NTxs)
  33. startTestRound(cs, height, round)
  34. ticker := time.NewTicker(time.Second * 20)
  35. for nTxs := 0; nTxs < NTxs; {
  36. select {
  37. case b := <-newBlockCh:
  38. nTxs += b.(types.EventDataNewBlock).Block.Header.NumTxs
  39. case <-ticker.C:
  40. panic("Timed out waiting to commit blocks with transactions")
  41. }
  42. }
  43. }
  44. func TestRmBadTx(t *testing.T) {
  45. state, privVals := randGenesisState(1, false, 10)
  46. app := NewCounterApplication()
  47. cs := newConsensusState(state, privVals[0], app)
  48. // increment the counter by 1
  49. txBytes := make([]byte, 8)
  50. binary.BigEndian.PutUint64(txBytes, uint64(0))
  51. app.AppendTx(txBytes)
  52. app.Commit()
  53. ch := make(chan struct{})
  54. cbCh := make(chan struct{})
  55. go func() {
  56. // Try to send the tx through the mempool.
  57. // CheckTx should not err, but the app should return a bad tmsp code
  58. // and the tx should get removed from the pool
  59. err := cs.mempool.CheckTx(txBytes, func(r *tmsp.Response) {
  60. if r.GetCheckTx().Code != tmsp.CodeType_BadNonce {
  61. t.Fatalf("expected checktx to return bad nonce, got %v", r)
  62. }
  63. cbCh <- struct{}{}
  64. })
  65. if err != nil {
  66. t.Fatal("Error after CheckTx: %v", err)
  67. }
  68. // check for the tx
  69. for {
  70. time.Sleep(time.Second)
  71. select {
  72. case <-ch:
  73. default:
  74. txs := cs.mempool.Reap(1)
  75. if len(txs) == 0 {
  76. ch <- struct{}{}
  77. }
  78. }
  79. }
  80. }()
  81. // Wait until the tx returns
  82. ticker := time.After(time.Second * 5)
  83. select {
  84. case <-cbCh:
  85. // success
  86. case <-ticker:
  87. t.Fatalf("Timed out waiting for tx to return")
  88. }
  89. // Wait until the tx is removed
  90. ticker = time.After(time.Second * 5)
  91. select {
  92. case <-ch:
  93. // success
  94. case <-ticker:
  95. t.Fatalf("Timed out waiting for tx to be removed")
  96. }
  97. }
  98. // CounterApplication that maintains a mempool state and resets it upon commit
  99. type CounterApplication struct {
  100. txCount int
  101. mempoolTxCount int
  102. }
  103. func NewCounterApplication() *CounterApplication {
  104. return &CounterApplication{}
  105. }
  106. func (app *CounterApplication) Info() string {
  107. return Fmt("txs:%v", app.txCount)
  108. }
  109. func (app *CounterApplication) SetOption(key string, value string) (log string) {
  110. return ""
  111. }
  112. func (app *CounterApplication) AppendTx(tx []byte) tmsp.Result {
  113. return runTx(tx, &app.txCount)
  114. }
  115. func (app *CounterApplication) CheckTx(tx []byte) tmsp.Result {
  116. return runTx(tx, &app.mempoolTxCount)
  117. }
  118. func runTx(tx []byte, countPtr *int) tmsp.Result {
  119. count := *countPtr
  120. tx8 := make([]byte, 8)
  121. copy(tx8[len(tx8)-len(tx):], tx)
  122. txValue := binary.BigEndian.Uint64(tx8)
  123. if txValue != uint64(count) {
  124. return tmsp.ErrBadNonce.AppendLog(Fmt("Invalid nonce. Expected %v, got %v", count, txValue))
  125. }
  126. *countPtr += 1
  127. return tmsp.OK
  128. }
  129. func (app *CounterApplication) Commit() tmsp.Result {
  130. app.mempoolTxCount = app.txCount
  131. if app.txCount == 0 {
  132. return tmsp.OK
  133. } else {
  134. hash := make([]byte, 8)
  135. binary.BigEndian.PutUint64(hash, uint64(app.txCount))
  136. return tmsp.NewResultOK(hash, "")
  137. }
  138. }
  139. func (app *CounterApplication) Query(query []byte) tmsp.Result {
  140. return tmsp.NewResultOK(nil, Fmt("Query is not supported"))
  141. }