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.

157 lines
3.9 KiB

  1. package mempool
  2. import (
  3. "fmt"
  4. "sync"
  5. "testing"
  6. "time"
  7. "github.com/fortytw2/leaktest"
  8. "github.com/pkg/errors"
  9. "github.com/stretchr/testify/assert"
  10. "github.com/go-kit/kit/log/term"
  11. "github.com/tendermint/abci/example/kvstore"
  12. "github.com/tendermint/tmlibs/log"
  13. cfg "github.com/tendermint/tendermint/config"
  14. "github.com/tendermint/tendermint/p2p"
  15. "github.com/tendermint/tendermint/proxy"
  16. "github.com/tendermint/tendermint/types"
  17. )
  18. // mempoolLogger is a TestingLogger which uses a different
  19. // color for each validator ("validator" key must exist).
  20. func mempoolLogger() log.Logger {
  21. return log.TestingLoggerWithColorFn(func(keyvals ...interface{}) term.FgBgColor {
  22. for i := 0; i < len(keyvals)-1; i += 2 {
  23. if keyvals[i] == "validator" {
  24. return term.FgBgColor{Fg: term.Color(uint8(keyvals[i+1].(int) + 1))}
  25. }
  26. }
  27. return term.FgBgColor{}
  28. })
  29. }
  30. // connect N mempool reactors through N switches
  31. func makeAndConnectMempoolReactors(config *cfg.Config, N int) []*MempoolReactor {
  32. reactors := make([]*MempoolReactor, N)
  33. logger := mempoolLogger()
  34. for i := 0; i < N; i++ {
  35. app := kvstore.NewKVStoreApplication()
  36. cc := proxy.NewLocalClientCreator(app)
  37. mempool := newMempoolWithApp(cc)
  38. reactors[i] = NewMempoolReactor(config.Mempool, mempool) // so we dont start the consensus states
  39. reactors[i].SetLogger(logger.With("validator", i))
  40. }
  41. p2p.MakeConnectedSwitches(config.P2P, N, func(i int, s *p2p.Switch) *p2p.Switch {
  42. s.AddReactor("MEMPOOL", reactors[i])
  43. return s
  44. }, p2p.Connect2Switches)
  45. return reactors
  46. }
  47. // wait for all txs on all reactors
  48. func waitForTxs(t *testing.T, txs types.Txs, reactors []*MempoolReactor) {
  49. // wait for the txs in all mempools
  50. wg := new(sync.WaitGroup)
  51. for i := 0; i < len(reactors); i++ {
  52. wg.Add(1)
  53. go _waitForTxs(t, wg, txs, i, reactors)
  54. }
  55. done := make(chan struct{})
  56. go func() {
  57. wg.Wait()
  58. close(done)
  59. }()
  60. timer := time.After(TIMEOUT)
  61. select {
  62. case <-timer:
  63. t.Fatal("Timed out waiting for txs")
  64. case <-done:
  65. }
  66. }
  67. // wait for all txs on a single mempool
  68. func _waitForTxs(t *testing.T, wg *sync.WaitGroup, txs types.Txs, reactorIdx int, reactors []*MempoolReactor) {
  69. mempool := reactors[reactorIdx].Mempool
  70. for mempool.Size() != len(txs) {
  71. time.Sleep(time.Millisecond * 100)
  72. }
  73. reapedTxs := mempool.Reap(len(txs))
  74. for i, tx := range txs {
  75. 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]))
  76. }
  77. wg.Done()
  78. }
  79. const (
  80. NUM_TXS = 1000
  81. TIMEOUT = 120 * time.Second // ridiculously high because CircleCI is slow
  82. )
  83. func TestReactorBroadcastTxMessage(t *testing.T) {
  84. config := cfg.TestConfig()
  85. const N = 4
  86. reactors := makeAndConnectMempoolReactors(config, N)
  87. defer func() {
  88. for _, r := range reactors {
  89. r.Stop()
  90. }
  91. }()
  92. // send a bunch of txs to the first reactor's mempool
  93. // and wait for them all to be received in the others
  94. txs := checkTxs(t, reactors[0].Mempool, NUM_TXS)
  95. waitForTxs(t, txs, reactors)
  96. }
  97. func TestBroadcastTxForPeerStopsWhenPeerStops(t *testing.T) {
  98. if testing.Short() {
  99. t.Skip("skipping test in short mode.")
  100. }
  101. config := cfg.TestConfig()
  102. const N = 2
  103. reactors := makeAndConnectMempoolReactors(config, N)
  104. defer func() {
  105. for _, r := range reactors {
  106. r.Stop()
  107. }
  108. }()
  109. // stop peer
  110. sw := reactors[1].Switch
  111. sw.StopPeerForError(sw.Peers().List()[0], errors.New("some reason"))
  112. // check that we are not leaking any go-routines
  113. // i.e. broadcastTxRoutine finishes when peer is stopped
  114. leaktest.CheckTimeout(t, 10*time.Second)()
  115. }
  116. func TestBroadcastTxForPeerStopsWhenReactorStops(t *testing.T) {
  117. if testing.Short() {
  118. t.Skip("skipping test in short mode.")
  119. }
  120. config := cfg.TestConfig()
  121. const N = 2
  122. reactors := makeAndConnectMempoolReactors(config, N)
  123. // stop reactors
  124. for _, r := range reactors {
  125. r.Stop()
  126. }
  127. // check that we are not leaking any go-routines
  128. // i.e. broadcastTxRoutine finishes when reactor is stopped
  129. leaktest.CheckTimeout(t, 10*time.Second)()
  130. }