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.

134 lines
4.9 KiB

8 years ago
mempool no gossip back (#2778) Closes #1798 This is done by making every mempool tx maintain a list of peers who its received the tx from. Instead of using the 20byte peer ID, it instead uses a local map from peerID to uint16 counter, so every peer adds 2 bytes. (Word aligned to probably make it 8 bytes) This also required resetting the callback function on every CheckTx. This likely has performance ramifications for instruction caching. The actual setting operation isn't costly with the removal of defers in this PR. * Make the mempool not gossip txs back to peers its received it from * Fix adversarial memleak * Don't break interface * Update changelog * Forgot to add a mtx * forgot a mutex * Update mempool/reactor.go Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * Update mempool/mempool.go Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * Use unknown peer ID Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * fix compilation * use next wait chan logic when skipping * Minor fixes * Add TxInfo * Add reverse map * Make activeID's auto-reserve 0 * 0 -> UnknownPeerID Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * Switch to making the normal case set a callback on the reqres object The recheck case is still done via the global callback, and stats are also set via global callback * fix merge conflict * Addres comments * Add cache tests * add cache tests * minor fixes * update metrics in reqResCb and reformat code * goimport -w mempool/reactor.go * mempool: update memTx senders I had to introduce txsMap for quick mempoolTx lookups. * change senders type from []uint16 to sync.Map Fixes DATA RACE: ``` Read at 0x00c0013fcd3a by goroutine 183: github.com/tendermint/tendermint/mempool.(*MempoolReactor).broadcastTxRoutine() /go/src/github.com/tendermint/tendermint/mempool/reactor.go:195 +0x3c7 Previous write at 0x00c0013fcd3a by D[2019-02-27|10:10:49.058] Read PacketMsg switch=3 peer=35bc1e3558c182927b31987eeff3feb3d58a0fc5@127.0.0.1 :46552 conn=MConn{pipe} packet="PacketMsg{30:2B06579D0A143EB78F3D3299DE8213A51D4E11FB05ACE4D6A14F T:1}" goroutine 190: github.com/tendermint/tendermint/mempool.(*Mempool).CheckTxWithInfo() /go/src/github.com/tendermint/tendermint/mempool/mempool.go:387 +0xdc1 github.com/tendermint/tendermint/mempool.(*MempoolReactor).Receive() /go/src/github.com/tendermint/tendermint/mempool/reactor.go:134 +0xb04 github.com/tendermint/tendermint/p2p.createMConnection.func1() /go/src/github.com/tendermint/tendermint/p2p/peer.go:374 +0x25b github.com/tendermint/tendermint/p2p/conn.(*MConnection).recvRoutine() /go/src/github.com/tendermint/tendermint/p2p/conn/connection.go:599 +0xcce Goroutine 183 (running) created at: D[2019-02-27|10:10:49.058] Send switch=2 peer=1efafad5443abeea4b7a8155218e4369525d987e@127.0.0.1:46193 channel=48 conn=MConn{pipe} m sgBytes=2B06579D0A146194480ADAE00C2836ED7125FEE65C1D9DD51049 github.com/tendermint/tendermint/mempool.(*MempoolReactor).AddPeer() /go/src/github.com/tendermint/tendermint/mempool/reactor.go:105 +0x1b1 github.com/tendermint/tendermint/p2p.(*Switch).startInitPeer() /go/src/github.com/tendermint/tendermint/p2p/switch.go:683 +0x13b github.com/tendermint/tendermint/p2p.(*Switch).addPeer() /go/src/github.com/tendermint/tendermint/p2p/switch.go:650 +0x585 github.com/tendermint/tendermint/p2p.(*Switch).addPeerWithConnection() /go/src/github.com/tendermint/tendermint/p2p/test_util.go:145 +0x939 github.com/tendermint/tendermint/p2p.Connect2Switches.func2() /go/src/github.com/tendermint/tendermint/p2p/test_util.go:109 +0x50 I[2019-02-27|10:10:49.058] Added good transaction validator=0 tx=43B4D1F0F03460BD262835C4AA560DB860CFBBE85BD02386D83DAC38C67B3AD7 res="&{CheckTx:gas_w anted:1 }" height=0 total=375 Goroutine 190 (running) created at: github.com/tendermint/tendermint/p2p/conn.(*MConnection).OnStart() /go/src/github.com/tendermint/tendermint/p2p/conn/connection.go:210 +0x313 github.com/tendermint/tendermint/libs/common.(*BaseService).Start() /go/src/github.com/tendermint/tendermint/libs/common/service.go:139 +0x4df github.com/tendermint/tendermint/p2p.(*peer).OnStart() /go/src/github.com/tendermint/tendermint/p2p/peer.go:179 +0x56 github.com/tendermint/tendermint/libs/common.(*BaseService).Start() /go/src/github.com/tendermint/tendermint/libs/common/service.go:139 +0x4df github.com/tendermint/tendermint/p2p.(*peer).Start() <autogenerated>:1 +0x43 github.com/tendermint/tendermint/p2p.(*Switch).startInitPeer() ``` * explain the choice of a map DS for senders * extract ids pool/mapper to a separate struct * fix literal copies lock value from senders: sync.Map contains sync.Mutex * use sync.Map#LoadOrStore instead of Load * fixes after Ismail's review * rename resCbNormal to resCbFirstTime
6 years ago
mempool no gossip back (#2778) Closes #1798 This is done by making every mempool tx maintain a list of peers who its received the tx from. Instead of using the 20byte peer ID, it instead uses a local map from peerID to uint16 counter, so every peer adds 2 bytes. (Word aligned to probably make it 8 bytes) This also required resetting the callback function on every CheckTx. This likely has performance ramifications for instruction caching. The actual setting operation isn't costly with the removal of defers in this PR. * Make the mempool not gossip txs back to peers its received it from * Fix adversarial memleak * Don't break interface * Update changelog * Forgot to add a mtx * forgot a mutex * Update mempool/reactor.go Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * Update mempool/mempool.go Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * Use unknown peer ID Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * fix compilation * use next wait chan logic when skipping * Minor fixes * Add TxInfo * Add reverse map * Make activeID's auto-reserve 0 * 0 -> UnknownPeerID Co-Authored-By: ValarDragon <ValarDragon@users.noreply.github.com> * Switch to making the normal case set a callback on the reqres object The recheck case is still done via the global callback, and stats are also set via global callback * fix merge conflict * Addres comments * Add cache tests * add cache tests * minor fixes * update metrics in reqResCb and reformat code * goimport -w mempool/reactor.go * mempool: update memTx senders I had to introduce txsMap for quick mempoolTx lookups. * change senders type from []uint16 to sync.Map Fixes DATA RACE: ``` Read at 0x00c0013fcd3a by goroutine 183: github.com/tendermint/tendermint/mempool.(*MempoolReactor).broadcastTxRoutine() /go/src/github.com/tendermint/tendermint/mempool/reactor.go:195 +0x3c7 Previous write at 0x00c0013fcd3a by D[2019-02-27|10:10:49.058] Read PacketMsg switch=3 peer=35bc1e3558c182927b31987eeff3feb3d58a0fc5@127.0.0.1 :46552 conn=MConn{pipe} packet="PacketMsg{30:2B06579D0A143EB78F3D3299DE8213A51D4E11FB05ACE4D6A14F T:1}" goroutine 190: github.com/tendermint/tendermint/mempool.(*Mempool).CheckTxWithInfo() /go/src/github.com/tendermint/tendermint/mempool/mempool.go:387 +0xdc1 github.com/tendermint/tendermint/mempool.(*MempoolReactor).Receive() /go/src/github.com/tendermint/tendermint/mempool/reactor.go:134 +0xb04 github.com/tendermint/tendermint/p2p.createMConnection.func1() /go/src/github.com/tendermint/tendermint/p2p/peer.go:374 +0x25b github.com/tendermint/tendermint/p2p/conn.(*MConnection).recvRoutine() /go/src/github.com/tendermint/tendermint/p2p/conn/connection.go:599 +0xcce Goroutine 183 (running) created at: D[2019-02-27|10:10:49.058] Send switch=2 peer=1efafad5443abeea4b7a8155218e4369525d987e@127.0.0.1:46193 channel=48 conn=MConn{pipe} m sgBytes=2B06579D0A146194480ADAE00C2836ED7125FEE65C1D9DD51049 github.com/tendermint/tendermint/mempool.(*MempoolReactor).AddPeer() /go/src/github.com/tendermint/tendermint/mempool/reactor.go:105 +0x1b1 github.com/tendermint/tendermint/p2p.(*Switch).startInitPeer() /go/src/github.com/tendermint/tendermint/p2p/switch.go:683 +0x13b github.com/tendermint/tendermint/p2p.(*Switch).addPeer() /go/src/github.com/tendermint/tendermint/p2p/switch.go:650 +0x585 github.com/tendermint/tendermint/p2p.(*Switch).addPeerWithConnection() /go/src/github.com/tendermint/tendermint/p2p/test_util.go:145 +0x939 github.com/tendermint/tendermint/p2p.Connect2Switches.func2() /go/src/github.com/tendermint/tendermint/p2p/test_util.go:109 +0x50 I[2019-02-27|10:10:49.058] Added good transaction validator=0 tx=43B4D1F0F03460BD262835C4AA560DB860CFBBE85BD02386D83DAC38C67B3AD7 res="&{CheckTx:gas_w anted:1 }" height=0 total=375 Goroutine 190 (running) created at: github.com/tendermint/tendermint/p2p/conn.(*MConnection).OnStart() /go/src/github.com/tendermint/tendermint/p2p/conn/connection.go:210 +0x313 github.com/tendermint/tendermint/libs/common.(*BaseService).Start() /go/src/github.com/tendermint/tendermint/libs/common/service.go:139 +0x4df github.com/tendermint/tendermint/p2p.(*peer).OnStart() /go/src/github.com/tendermint/tendermint/p2p/peer.go:179 +0x56 github.com/tendermint/tendermint/libs/common.(*BaseService).Start() /go/src/github.com/tendermint/tendermint/libs/common/service.go:139 +0x4df github.com/tendermint/tendermint/p2p.(*peer).Start() <autogenerated>:1 +0x43 github.com/tendermint/tendermint/p2p.(*Switch).startInitPeer() ``` * explain the choice of a map DS for senders * extract ids pool/mapper to a separate struct * fix literal copies lock value from senders: sync.Map contains sync.Mutex * use sync.Map#LoadOrStore instead of Load * fixes after Ismail's review * rename resCbNormal to resCbFirstTime
6 years ago
  1. package mempool
  2. import (
  3. "fmt"
  4. abci "github.com/tendermint/tendermint/abci/types"
  5. "github.com/tendermint/tendermint/libs/clist"
  6. "github.com/tendermint/tendermint/types"
  7. )
  8. // Mempool defines the mempool interface.
  9. //
  10. // Updates to the mempool need to be synchronized with committing a block so
  11. // apps can reset their transient state on Commit.
  12. type Mempool interface {
  13. // CheckTx executes a new transaction against the application to determine
  14. // its validity and whether it should be added to the mempool.
  15. CheckTx(tx types.Tx, callback func(*abci.Response)) error
  16. // CheckTxWithInfo performs the same operation as CheckTx, but with extra
  17. // meta data about the tx.
  18. // Currently this metadata is the peer who sent it, used to prevent the tx
  19. // from being gossiped back to them.
  20. CheckTxWithInfo(tx types.Tx, callback func(*abci.Response), txInfo TxInfo) error
  21. // ReapMaxBytesMaxGas reaps transactions from the mempool up to maxBytes
  22. // bytes total with the condition that the total gasWanted must be less than
  23. // maxGas.
  24. // If both maxes are negative, there is no cap on the size of all returned
  25. // transactions (~ all available transactions).
  26. ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs
  27. // ReapMaxTxs reaps up to max transactions from the mempool.
  28. // If max is negative, there is no cap on the size of all returned
  29. // transactions (~ all available transactions).
  30. ReapMaxTxs(max int) types.Txs
  31. // UNSAFE
  32. TxsWaitChan() <-chan struct{}
  33. TxsFront() *clist.CElement
  34. // Lock locks the mempool. The consensus must be able to hold lock to safely update.
  35. Lock()
  36. // Unlock unlocks the mempool.
  37. Unlock()
  38. // Update informs the mempool that the given txs were committed and can be discarded.
  39. // NOTE: this should be called *after* block is committed by consensus.
  40. // NOTE: unsafe; Lock/Unlock must be managed by caller
  41. Update(blockHeight int64, blockTxs types.Txs, newPreFn PreCheckFunc, newPostFn PostCheckFunc) error
  42. // FlushAppConn flushes the mempool connection to ensure async reqResCb calls are
  43. // done. E.g. from CheckTx.
  44. FlushAppConn() error
  45. // Flush removes all transactions from the mempool and cache
  46. Flush()
  47. // TxsAvailable returns a channel which fires once for every height,
  48. // and only when transactions are available in the mempool.
  49. // NOTE: the returned channel may be nil if EnableTxsAvailable was not called.
  50. TxsAvailable() <-chan struct{}
  51. // EnableTxsAvailable initializes the TxsAvailable channel, ensuring it will
  52. // trigger once every height when transactions are available.
  53. EnableTxsAvailable()
  54. // Size returns the number of transactions in the mempool.
  55. Size() int
  56. // TxsBytes returns the total size of all txs in the mempool.
  57. TxsBytes() int64
  58. }
  59. //--------------------------------------------------------------------------------
  60. // PreCheckFunc is an optional filter executed before CheckTx and rejects
  61. // transaction if false is returned. An example would be to ensure that a
  62. // transaction doesn't exceeded the block size.
  63. type PreCheckFunc func(types.Tx) error
  64. // PostCheckFunc is an optional filter executed after CheckTx and rejects
  65. // transaction if false is returned. An example would be to ensure a
  66. // transaction doesn't require more gas than available for the block.
  67. type PostCheckFunc func(types.Tx, *abci.ResponseCheckTx) error
  68. // TxInfo are parameters that get passed when attempting to add a tx to the
  69. // mempool.
  70. type TxInfo struct {
  71. // We don't use p2p.ID here because it's too big. The gain is to store max 2
  72. // bytes with each tx to identify the sender rather than 20 bytes.
  73. SenderID uint16
  74. }
  75. //--------------------------------------------------------------------------------
  76. // PreCheckAminoMaxBytes checks that the size of the transaction plus the amino
  77. // overhead is smaller or equal to the expected maxBytes.
  78. func PreCheckAminoMaxBytes(maxBytes int64) PreCheckFunc {
  79. return func(tx types.Tx) error {
  80. // We have to account for the amino overhead in the tx size as well
  81. // NOTE: fieldNum = 1 as types.Block.Data contains Txs []Tx as first field.
  82. // If this field order ever changes this needs to updated here accordingly.
  83. // NOTE: if some []Tx are encoded without a parenting struct, the
  84. // fieldNum is also equal to 1.
  85. aminoOverhead := types.ComputeAminoOverhead(tx, 1)
  86. txSize := int64(len(tx)) + aminoOverhead
  87. if txSize > maxBytes {
  88. return fmt.Errorf("Tx size (including amino overhead) is too big: %d, max: %d",
  89. txSize, maxBytes)
  90. }
  91. return nil
  92. }
  93. }
  94. // PostCheckMaxGas checks that the wanted gas is smaller or equal to the passed
  95. // maxGas. Returns nil if maxGas is -1.
  96. func PostCheckMaxGas(maxGas int64) PostCheckFunc {
  97. return func(tx types.Tx, res *abci.ResponseCheckTx) error {
  98. if maxGas == -1 {
  99. return nil
  100. }
  101. if res.GasWanted < 0 {
  102. return fmt.Errorf("gas wanted %d is negative",
  103. res.GasWanted)
  104. }
  105. if res.GasWanted > maxGas {
  106. return fmt.Errorf("gas wanted %d is greater than max gas %d",
  107. res.GasWanted, maxGas)
  108. }
  109. return nil
  110. }
  111. }