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.

286 lines
8.0 KiB

8 years ago
8 years ago
8 years ago
  1. package mempool
  2. import (
  3. "crypto/md5"
  4. "crypto/rand"
  5. "encoding/binary"
  6. "fmt"
  7. "io/ioutil"
  8. "os"
  9. "path/filepath"
  10. "testing"
  11. "time"
  12. "github.com/tendermint/abci/example/counter"
  13. "github.com/tendermint/abci/example/kvstore"
  14. abci "github.com/tendermint/abci/types"
  15. cmn "github.com/tendermint/tmlibs/common"
  16. "github.com/tendermint/tmlibs/log"
  17. cfg "github.com/tendermint/tendermint/config"
  18. "github.com/tendermint/tendermint/proxy"
  19. "github.com/tendermint/tendermint/types"
  20. "github.com/stretchr/testify/require"
  21. )
  22. func newMempoolWithApp(cc proxy.ClientCreator) *Mempool {
  23. config := cfg.ResetTestRoot("mempool_test")
  24. appConnMem, _ := cc.NewABCIClient()
  25. appConnMem.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "mempool"))
  26. err := appConnMem.Start()
  27. if err != nil {
  28. panic(err)
  29. }
  30. mempool := NewMempool(config.Mempool, appConnMem, 0)
  31. mempool.SetLogger(log.TestingLogger())
  32. return mempool
  33. }
  34. func ensureNoFire(t *testing.T, ch <-chan int64, timeoutMS int) {
  35. timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond)
  36. select {
  37. case <-ch:
  38. t.Fatal("Expected not to fire")
  39. case <-timer.C:
  40. }
  41. }
  42. func ensureFire(t *testing.T, ch <-chan int64, timeoutMS int) {
  43. timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond)
  44. select {
  45. case <-ch:
  46. case <-timer.C:
  47. t.Fatal("Expected to fire")
  48. }
  49. }
  50. func checkTxs(t *testing.T, mempool *Mempool, count int) types.Txs {
  51. txs := make(types.Txs, count)
  52. for i := 0; i < count; i++ {
  53. txBytes := make([]byte, 20)
  54. txs[i] = txBytes
  55. _, err := rand.Read(txBytes)
  56. if err != nil {
  57. t.Error(err)
  58. }
  59. if err := mempool.CheckTx(txBytes, nil); err != nil {
  60. t.Fatalf("Error after CheckTx: %v", err)
  61. }
  62. }
  63. return txs
  64. }
  65. func TestTxsAvailable(t *testing.T) {
  66. app := kvstore.NewKVStoreApplication()
  67. cc := proxy.NewLocalClientCreator(app)
  68. mempool := newMempoolWithApp(cc)
  69. mempool.EnableTxsAvailable()
  70. timeoutMS := 500
  71. // with no txs, it shouldnt fire
  72. ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
  73. // send a bunch of txs, it should only fire once
  74. txs := checkTxs(t, mempool, 100)
  75. ensureFire(t, mempool.TxsAvailable(), timeoutMS)
  76. ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
  77. // call update with half the txs.
  78. // it should fire once now for the new height
  79. // since there are still txs left
  80. committedTxs, txs := txs[:50], txs[50:]
  81. if err := mempool.Update(1, committedTxs); err != nil {
  82. t.Error(err)
  83. }
  84. ensureFire(t, mempool.TxsAvailable(), timeoutMS)
  85. ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
  86. // send a bunch more txs. we already fired for this height so it shouldnt fire again
  87. moreTxs := checkTxs(t, mempool, 50)
  88. ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
  89. // now call update with all the txs. it should not fire as there are no txs left
  90. committedTxs = append(txs, moreTxs...)
  91. if err := mempool.Update(2, committedTxs); err != nil {
  92. t.Error(err)
  93. }
  94. ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
  95. // send a bunch more txs, it should only fire once
  96. checkTxs(t, mempool, 100)
  97. ensureFire(t, mempool.TxsAvailable(), timeoutMS)
  98. ensureNoFire(t, mempool.TxsAvailable(), timeoutMS)
  99. }
  100. func TestSerialReap(t *testing.T) {
  101. app := counter.NewCounterApplication(true)
  102. app.SetOption(abci.RequestSetOption{"serial", "on"})
  103. cc := proxy.NewLocalClientCreator(app)
  104. mempool := newMempoolWithApp(cc)
  105. appConnCon, _ := cc.NewABCIClient()
  106. appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
  107. err := appConnCon.Start()
  108. require.Nil(t, err)
  109. cacheMap := make(map[string]struct{})
  110. deliverTxsRange := func(start, end int) {
  111. // Deliver some txs.
  112. for i := start; i < end; i++ {
  113. // This will succeed
  114. txBytes := make([]byte, 8)
  115. binary.BigEndian.PutUint64(txBytes, uint64(i))
  116. err := mempool.CheckTx(txBytes, nil)
  117. _, cached := cacheMap[string(txBytes)]
  118. if cached {
  119. require.NotNil(t, err, "expected error for cached tx")
  120. } else {
  121. require.Nil(t, err, "expected no err for uncached tx")
  122. }
  123. cacheMap[string(txBytes)] = struct{}{}
  124. // Duplicates are cached and should return error
  125. err = mempool.CheckTx(txBytes, nil)
  126. require.NotNil(t, err, "Expected error after CheckTx on duplicated tx")
  127. }
  128. }
  129. reapCheck := func(exp int) {
  130. txs := mempool.Reap(-1)
  131. require.Equal(t, len(txs), exp, cmn.Fmt("Expected to reap %v txs but got %v", exp, len(txs)))
  132. }
  133. updateRange := func(start, end int) {
  134. txs := make([]types.Tx, 0)
  135. for i := start; i < end; i++ {
  136. txBytes := make([]byte, 8)
  137. binary.BigEndian.PutUint64(txBytes, uint64(i))
  138. txs = append(txs, txBytes)
  139. }
  140. if err := mempool.Update(0, txs); err != nil {
  141. t.Error(err)
  142. }
  143. }
  144. commitRange := func(start, end int) {
  145. // Deliver some txs.
  146. for i := start; i < end; i++ {
  147. txBytes := make([]byte, 8)
  148. binary.BigEndian.PutUint64(txBytes, uint64(i))
  149. res, err := appConnCon.DeliverTxSync(txBytes)
  150. if err != nil {
  151. t.Errorf("Client error committing tx: %v", err)
  152. }
  153. if res.IsErr() {
  154. t.Errorf("Error committing tx. Code:%v result:%X log:%v",
  155. res.Code, res.Data, res.Log)
  156. }
  157. }
  158. res, err := appConnCon.CommitSync()
  159. if err != nil {
  160. t.Errorf("Client error committing: %v", err)
  161. }
  162. if len(res.Data) != 8 {
  163. t.Errorf("Error committing. Hash:%X", res.Data)
  164. }
  165. }
  166. //----------------------------------------
  167. // Deliver some txs.
  168. deliverTxsRange(0, 100)
  169. // Reap the txs.
  170. reapCheck(100)
  171. // Reap again. We should get the same amount
  172. reapCheck(100)
  173. // Deliver 0 to 999, we should reap 900 new txs
  174. // because 100 were already counted.
  175. deliverTxsRange(0, 1000)
  176. // Reap the txs.
  177. reapCheck(1000)
  178. // Reap again. We should get the same amount
  179. reapCheck(1000)
  180. // Commit from the conensus AppConn
  181. commitRange(0, 500)
  182. updateRange(0, 500)
  183. // We should have 500 left.
  184. reapCheck(500)
  185. // Deliver 100 invalid txs and 100 valid txs
  186. deliverTxsRange(900, 1100)
  187. // We should have 600 now.
  188. reapCheck(600)
  189. }
  190. func TestMempoolCloseWAL(t *testing.T) {
  191. // 1. Create the temporary directory for mempool and WAL testing.
  192. rootDir, err := ioutil.TempDir("", "mempool-test")
  193. require.Nil(t, err, "expecting successful tmpdir creation")
  194. defer os.RemoveAll(rootDir)
  195. // 2. Ensure that it doesn't contain any elements -- Sanity check
  196. m1, err := filepath.Glob(filepath.Join(rootDir, "*"))
  197. require.Nil(t, err, "successful globbing expected")
  198. require.Equal(t, 0, len(m1), "no matches yet")
  199. // 3. Create the mempool
  200. wcfg := cfg.DefaultMempoolConfig()
  201. wcfg.RootDir = rootDir
  202. app := kvstore.NewKVStoreApplication()
  203. cc := proxy.NewLocalClientCreator(app)
  204. appConnMem, _ := cc.NewABCIClient()
  205. mempool := NewMempool(wcfg, appConnMem, 10)
  206. mempool.InitWAL()
  207. // 4. Ensure that the directory contains the WAL file
  208. m2, err := filepath.Glob(filepath.Join(rootDir, "*"))
  209. require.Nil(t, err, "successful globbing expected")
  210. require.Equal(t, 1, len(m2), "expecting the wal match in")
  211. // 5. Write some contents to the WAL
  212. mempool.CheckTx(types.Tx([]byte("foo")), nil)
  213. walFilepath := mempool.wal.Path
  214. sum1 := checksumFile(walFilepath, t)
  215. // 6. Sanity check to ensure that the written TX matches the expectation.
  216. require.Equal(t, sum1, checksumIt([]byte("foo\n")), "foo with a newline should be written")
  217. // 7. Invoke CloseWAL() and ensure it discards the
  218. // WAL thus any other write won't go through.
  219. require.True(t, mempool.CloseWAL(), "CloseWAL should CloseWAL")
  220. mempool.CheckTx(types.Tx([]byte("bar")), nil)
  221. sum2 := checksumFile(walFilepath, t)
  222. require.Equal(t, sum1, sum2, "expected no change to the WAL after invoking CloseWAL() since it was discarded")
  223. // 8. Second CloseWAL should do nothing
  224. require.False(t, mempool.CloseWAL(), "CloseWAL should CloseWAL")
  225. // 9. Sanity check to ensure that the WAL file still exists
  226. m3, err := filepath.Glob(filepath.Join(rootDir, "*"))
  227. require.Nil(t, err, "successful globbing expected")
  228. require.Equal(t, 1, len(m3), "expecting the wal match in")
  229. }
  230. func checksumIt(data []byte) string {
  231. h := md5.New()
  232. h.Write(data)
  233. return fmt.Sprintf("%x", h.Sum(nil))
  234. }
  235. func checksumFile(p string, t *testing.T) string {
  236. data, err := ioutil.ReadFile(p)
  237. require.Nil(t, err, "expecting successful read of %q", p)
  238. return checksumIt(data)
  239. }