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.

108 lines
2.8 KiB

  1. package mempool
  2. import (
  3. "fmt"
  4. "sync"
  5. "testing"
  6. "time"
  7. "github.com/stretchr/testify/assert"
  8. "github.com/go-kit/kit/log/term"
  9. "github.com/tendermint/abci/example/dummy"
  10. "github.com/tendermint/tmlibs/log"
  11. cfg "github.com/tendermint/tendermint/config"
  12. "github.com/tendermint/tendermint/p2p"
  13. "github.com/tendermint/tendermint/proxy"
  14. "github.com/tendermint/tendermint/types"
  15. )
  16. // mempoolLogger is a TestingLogger which uses a different
  17. // color for each validator ("validator" key must exist).
  18. func mempoolLogger() log.Logger {
  19. return log.TestingLoggerWithColorFn(func(keyvals ...interface{}) term.FgBgColor {
  20. for i := 0; i < len(keyvals)-1; i += 2 {
  21. if keyvals[i] == "validator" {
  22. return term.FgBgColor{Fg: term.Color(uint8(keyvals[i+1].(int) + 1))}
  23. }
  24. }
  25. return term.FgBgColor{}
  26. })
  27. }
  28. // connect N mempool reactors through N switches
  29. func makeAndConnectMempoolReactors(config *cfg.Config, N int) []*MempoolReactor {
  30. reactors := make([]*MempoolReactor, N)
  31. logger := mempoolLogger()
  32. for i := 0; i < N; i++ {
  33. app := dummy.NewDummyApplication()
  34. cc := proxy.NewLocalClientCreator(app)
  35. mempool := newMempoolWithApp(cc)
  36. reactors[i] = NewMempoolReactor(config.Mempool, mempool) // so we dont start the consensus states
  37. reactors[i].SetLogger(logger.With("validator", i))
  38. }
  39. p2p.MakeConnectedSwitches(config.P2P, N, func(i int, s *p2p.Switch) *p2p.Switch {
  40. s.AddReactor("MEMPOOL", reactors[i])
  41. return s
  42. }, p2p.Connect2Switches)
  43. return reactors
  44. }
  45. // wait for all txs on all reactors
  46. func waitForTxs(t *testing.T, txs types.Txs, reactors []*MempoolReactor) {
  47. // wait for the txs in all mempools
  48. wg := new(sync.WaitGroup)
  49. for i := 0; i < len(reactors); i++ {
  50. wg.Add(1)
  51. go _waitForTxs(t, wg, txs, i, reactors)
  52. }
  53. done := make(chan struct{})
  54. go func() {
  55. wg.Wait()
  56. close(done)
  57. }()
  58. timer := time.After(TIMEOUT)
  59. select {
  60. case <-timer:
  61. t.Fatal("Timed out waiting for txs")
  62. case <-done:
  63. }
  64. }
  65. // wait for all txs on a single mempool
  66. func _waitForTxs(t *testing.T, wg *sync.WaitGroup, txs types.Txs, reactorIdx int, reactors []*MempoolReactor) {
  67. mempool := reactors[reactorIdx].Mempool
  68. for mempool.Size() != len(txs) {
  69. time.Sleep(time.Millisecond * 100)
  70. }
  71. reapedTxs := mempool.Reap(len(txs))
  72. for i, tx := range txs {
  73. assert.Equal(t, tx, reapedTxs[i], fmt.Sprintf("txs at index %d on reactor %d don't match: %v vs %v", i, reactorIdx, tx, reapedTxs[i]))
  74. }
  75. wg.Done()
  76. }
  77. var (
  78. NUM_TXS = 1000
  79. TIMEOUT = 120 * time.Second // ridiculously high because CircleCI is slow
  80. )
  81. func TestReactorBroadcastTxMessage(t *testing.T) {
  82. config := cfg.TestConfig()
  83. N := 4
  84. reactors := makeAndConnectMempoolReactors(config, N)
  85. // send a bunch of txs to the first reactor's mempool
  86. // and wait for them all to be received in the others
  87. txs := checkTxs(t, reactors[0].Mempool, NUM_TXS)
  88. waitForTxs(t, txs, reactors)
  89. }