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.

98 lines
2.5 KiB

  1. /*
  2. Mempool receives new transactions and applies them to the latest committed state.
  3. If the transaction is acceptable, then it broadcasts the tx to peers.
  4. When this node happens to be the next proposer, it simply uses the recently
  5. modified state (and the associated transactions) to construct a proposal.
  6. */
  7. package mempool
  8. import (
  9. "sync"
  10. "github.com/tendermint/tendermint/binary"
  11. blk "github.com/tendermint/tendermint/block"
  12. sm "github.com/tendermint/tendermint/state"
  13. )
  14. type Mempool struct {
  15. mtx sync.Mutex
  16. state *sm.State
  17. txs []blk.Tx
  18. }
  19. func NewMempool(state *sm.State) *Mempool {
  20. return &Mempool{
  21. state: state,
  22. }
  23. }
  24. // Apply tx to the state and remember it.
  25. func (mem *Mempool) AddTx(tx blk.Tx) (err error) {
  26. mem.mtx.Lock()
  27. defer mem.mtx.Unlock()
  28. err = mem.state.ExecTx(tx)
  29. if err != nil {
  30. log.Debug("AddTx() error", "tx", tx, "error", err)
  31. return err
  32. } else {
  33. log.Debug("AddTx() success", "tx", tx)
  34. mem.txs = append(mem.txs, tx)
  35. return nil
  36. }
  37. }
  38. func (mem *Mempool) GetProposalTxs() []blk.Tx {
  39. mem.mtx.Lock()
  40. defer mem.mtx.Unlock()
  41. log.Debug("GetProposalTxs:", "txs", mem.txs)
  42. return mem.txs
  43. }
  44. // "block" is the new block being committed.
  45. // "state" is the result of state.AppendBlock("block").
  46. // Txs that are present in "block" are discarded from mempool.
  47. // Txs that have become invalid in the new "state" are also discarded.
  48. func (mem *Mempool) ResetForBlockAndState(block *blk.Block, state *sm.State) {
  49. mem.mtx.Lock()
  50. defer mem.mtx.Unlock()
  51. mem.state = state.Copy()
  52. // First, create a lookup map of txns in new block.
  53. blockTxsMap := make(map[string]struct{})
  54. for _, tx := range block.Data.Txs {
  55. txHash := binary.BinarySha256(tx)
  56. blockTxsMap[string(txHash)] = struct{}{}
  57. }
  58. // Next, filter all txs from mem.txs that are in blockTxsMap
  59. txs := []blk.Tx{}
  60. for _, tx := range mem.txs {
  61. txHash := binary.BinarySha256(tx)
  62. if _, ok := blockTxsMap[string(txHash)]; ok {
  63. log.Debug("Filter out, already committed", "tx", tx, "txHash", txHash)
  64. continue
  65. } else {
  66. log.Debug("Filter in, still new", "tx", tx, "txHash", txHash)
  67. txs = append(txs, tx)
  68. }
  69. }
  70. // Next, filter all txs that aren't valid given new state.
  71. validTxs := []blk.Tx{}
  72. for _, tx := range txs {
  73. err := mem.state.ExecTx(tx)
  74. if err == nil {
  75. log.Debug("Filter in, valid", "tx", tx)
  76. validTxs = append(validTxs, tx)
  77. } else {
  78. // tx is no longer valid.
  79. log.Debug("Filter out, no longer valid", "tx", tx, "error", err)
  80. }
  81. }
  82. // We're done!
  83. log.Debug("New txs", "txs", validTxs, "oldTxs", mem.txs)
  84. mem.txs = validTxs
  85. }