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.

869 lines
26 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
9 years ago
  1. package mempool
  2. import (
  3. "bytes"
  4. "context"
  5. "errors"
  6. "fmt"
  7. "reflect"
  8. "sync/atomic"
  9. "time"
  10. abci "github.com/tendermint/tendermint/abci/types"
  11. "github.com/tendermint/tendermint/config"
  12. "github.com/tendermint/tendermint/internal/libs/clist"
  13. tmsync "github.com/tendermint/tendermint/internal/libs/sync"
  14. "github.com/tendermint/tendermint/internal/proxy"
  15. "github.com/tendermint/tendermint/libs/log"
  16. tmmath "github.com/tendermint/tendermint/libs/math"
  17. "github.com/tendermint/tendermint/types"
  18. )
  19. var _ Mempool = (*TxMempool)(nil)
  20. // TxMempoolOption sets an optional parameter on the TxMempool.
  21. type TxMempoolOption func(*TxMempool)
  22. // TxMempool defines a prioritized mempool data structure used by the v1 mempool
  23. // reactor. It keeps a thread-safe priority queue of transactions that is used
  24. // when a block proposer constructs a block and a thread-safe linked-list that
  25. // is used to gossip transactions to peers in a FIFO manner.
  26. type TxMempool struct {
  27. logger log.Logger
  28. metrics *Metrics
  29. config *config.MempoolConfig
  30. proxyAppConn proxy.AppConnMempool
  31. // txsAvailable fires once for each height when the mempool is not empty
  32. txsAvailable chan struct{}
  33. notifiedTxsAvailable bool
  34. // height defines the last block height process during Update()
  35. height int64
  36. // sizeBytes defines the total size of the mempool (sum of all tx bytes)
  37. sizeBytes int64
  38. // cache defines a fixed-size cache of already seen transactions as this
  39. // reduces pressure on the proxyApp.
  40. cache TxCache
  41. // txStore defines the main storage of valid transactions. Indexes are built
  42. // on top of this store.
  43. txStore *TxStore
  44. // gossipIndex defines the gossiping index of valid transactions via a
  45. // thread-safe linked-list. We also use the gossip index as a cursor for
  46. // rechecking transactions already in the mempool.
  47. gossipIndex *clist.CList
  48. // recheckCursor and recheckEnd are used as cursors based on the gossip index
  49. // to recheck transactions that are already in the mempool. Iteration is not
  50. // thread-safe and transaction may be mutated in serial order.
  51. //
  52. // XXX/TODO: It might be somewhat of a codesmell to use the gossip index for
  53. // iterator and cursor management when rechecking transactions. If the gossip
  54. // index changes or is removed in a future refactor, this will have to be
  55. // refactored. Instead, we should consider just keeping a slice of a snapshot
  56. // of the mempool's current transactions during Update and an integer cursor
  57. // into that slice. This, however, requires additional O(n) space complexity.
  58. recheckCursor *clist.CElement // next expected response
  59. recheckEnd *clist.CElement // re-checking stops here
  60. // priorityIndex defines the priority index of valid transactions via a
  61. // thread-safe priority queue.
  62. priorityIndex *TxPriorityQueue
  63. // heightIndex defines a height-based, in ascending order, transaction index.
  64. // i.e. older transactions are first.
  65. heightIndex *WrappedTxList
  66. // timestampIndex defines a timestamp-based, in ascending order, transaction
  67. // index. i.e. older transactions are first.
  68. timestampIndex *WrappedTxList
  69. // A read/write lock is used to safe guard updates, insertions and deletions
  70. // from the mempool. A read-lock is implicitly acquired when executing CheckTx,
  71. // however, a caller must explicitly grab a write-lock via Lock when updating
  72. // the mempool via Update().
  73. mtx tmsync.RWMutex
  74. preCheck PreCheckFunc
  75. postCheck PostCheckFunc
  76. }
  77. func NewTxMempool(
  78. logger log.Logger,
  79. cfg *config.MempoolConfig,
  80. proxyAppConn proxy.AppConnMempool,
  81. height int64,
  82. options ...TxMempoolOption,
  83. ) *TxMempool {
  84. txmp := &TxMempool{
  85. logger: logger,
  86. config: cfg,
  87. proxyAppConn: proxyAppConn,
  88. height: height,
  89. cache: NopTxCache{},
  90. metrics: NopMetrics(),
  91. txStore: NewTxStore(),
  92. gossipIndex: clist.New(),
  93. priorityIndex: NewTxPriorityQueue(),
  94. heightIndex: NewWrappedTxList(func(wtx1, wtx2 *WrappedTx) bool {
  95. return wtx1.height >= wtx2.height
  96. }),
  97. timestampIndex: NewWrappedTxList(func(wtx1, wtx2 *WrappedTx) bool {
  98. return wtx1.timestamp.After(wtx2.timestamp) || wtx1.timestamp.Equal(wtx2.timestamp)
  99. }),
  100. }
  101. if cfg.CacheSize > 0 {
  102. txmp.cache = NewLRUTxCache(cfg.CacheSize)
  103. }
  104. proxyAppConn.SetResponseCallback(txmp.defaultTxCallback)
  105. for _, opt := range options {
  106. opt(txmp)
  107. }
  108. return txmp
  109. }
  110. // WithPreCheck sets a filter for the mempool to reject a transaction if f(tx)
  111. // returns an error. This is executed before CheckTx. It only applies to the
  112. // first created block. After that, Update() overwrites the existing value.
  113. func WithPreCheck(f PreCheckFunc) TxMempoolOption {
  114. return func(txmp *TxMempool) { txmp.preCheck = f }
  115. }
  116. // WithPostCheck sets a filter for the mempool to reject a transaction if
  117. // f(tx, resp) returns an error. This is executed after CheckTx. It only applies
  118. // to the first created block. After that, Update overwrites the existing value.
  119. func WithPostCheck(f PostCheckFunc) TxMempoolOption {
  120. return func(txmp *TxMempool) { txmp.postCheck = f }
  121. }
  122. // WithMetrics sets the mempool's metrics collector.
  123. func WithMetrics(metrics *Metrics) TxMempoolOption {
  124. return func(txmp *TxMempool) { txmp.metrics = metrics }
  125. }
  126. // Lock obtains a write-lock on the mempool. A caller must be sure to explicitly
  127. // release the lock when finished.
  128. func (txmp *TxMempool) Lock() {
  129. txmp.mtx.Lock()
  130. }
  131. // Unlock releases a write-lock on the mempool.
  132. func (txmp *TxMempool) Unlock() {
  133. txmp.mtx.Unlock()
  134. }
  135. // Size returns the number of valid transactions in the mempool. It is
  136. // thread-safe.
  137. func (txmp *TxMempool) Size() int {
  138. return txmp.txStore.Size()
  139. }
  140. // SizeBytes return the total sum in bytes of all the valid transactions in the
  141. // mempool. It is thread-safe.
  142. func (txmp *TxMempool) SizeBytes() int64 {
  143. return atomic.LoadInt64(&txmp.sizeBytes)
  144. }
  145. // FlushAppConn executes FlushSync on the mempool's proxyAppConn.
  146. //
  147. // NOTE: The caller must obtain a write-lock prior to execution.
  148. func (txmp *TxMempool) FlushAppConn() error {
  149. return txmp.proxyAppConn.FlushSync(context.Background())
  150. }
  151. // WaitForNextTx returns a blocking channel that will be closed when the next
  152. // valid transaction is available to gossip. It is thread-safe.
  153. func (txmp *TxMempool) WaitForNextTx() <-chan struct{} {
  154. return txmp.gossipIndex.WaitChan()
  155. }
  156. // NextGossipTx returns the next valid transaction to gossip. A caller must wait
  157. // for WaitForNextTx to signal a transaction is available to gossip first. It is
  158. // thread-safe.
  159. func (txmp *TxMempool) NextGossipTx() *clist.CElement {
  160. return txmp.gossipIndex.Front()
  161. }
  162. // EnableTxsAvailable enables the mempool to trigger events when transactions
  163. // are available on a block by block basis.
  164. func (txmp *TxMempool) EnableTxsAvailable() {
  165. txmp.mtx.Lock()
  166. defer txmp.mtx.Unlock()
  167. txmp.txsAvailable = make(chan struct{}, 1)
  168. }
  169. // TxsAvailable returns a channel which fires once for every height, and only
  170. // when transactions are available in the mempool. It is thread-safe.
  171. func (txmp *TxMempool) TxsAvailable() <-chan struct{} {
  172. return txmp.txsAvailable
  173. }
  174. // CheckTx executes the ABCI CheckTx method for a given transaction. It acquires
  175. // a read-lock attempts to execute the application's CheckTx ABCI method via
  176. // CheckTxAsync. We return an error if any of the following happen:
  177. //
  178. // - The CheckTxAsync execution fails.
  179. // - The transaction already exists in the cache and we've already received the
  180. // transaction from the peer. Otherwise, if it solely exists in the cache, we
  181. // return nil.
  182. // - The transaction size exceeds the maximum transaction size as defined by the
  183. // configuration provided to the mempool.
  184. // - The transaction fails Pre-Check (if it is defined).
  185. // - The proxyAppConn fails, e.g. the buffer is full.
  186. //
  187. // If the mempool is full, we still execute CheckTx and attempt to find a lower
  188. // priority transaction to evict. If such a transaction exists, we remove the
  189. // lower priority transaction and add the new one with higher priority.
  190. //
  191. // NOTE:
  192. // - The applications' CheckTx implementation may panic.
  193. // - The caller is not to explicitly require any locks for executing CheckTx.
  194. func (txmp *TxMempool) CheckTx(
  195. ctx context.Context,
  196. tx types.Tx,
  197. cb func(*abci.Response),
  198. txInfo TxInfo,
  199. ) error {
  200. if ctx == nil {
  201. ctx = context.TODO()
  202. }
  203. txmp.mtx.RLock()
  204. defer txmp.mtx.RUnlock()
  205. if txSize := len(tx); txSize > txmp.config.MaxTxBytes {
  206. return types.ErrTxTooLarge{
  207. Max: txmp.config.MaxTxBytes,
  208. Actual: txSize,
  209. }
  210. }
  211. if txmp.preCheck != nil {
  212. if err := txmp.preCheck(tx); err != nil {
  213. return types.ErrPreCheck{Reason: err}
  214. }
  215. }
  216. if err := txmp.proxyAppConn.Error(); err != nil {
  217. return err
  218. }
  219. txHash := tx.Key()
  220. // We add the transaction to the mempool's cache and if the transaction already
  221. // exists, i.e. false is returned, then we check if we've seen this transaction
  222. // from the same sender and error if we have. Otherwise, we return nil.
  223. if !txmp.cache.Push(tx) {
  224. wtx, ok := txmp.txStore.GetOrSetPeerByTxHash(txHash, txInfo.SenderID)
  225. if wtx != nil && ok {
  226. // We already have the transaction stored and the we've already seen this
  227. // transaction from txInfo.SenderID.
  228. return types.ErrTxInCache
  229. }
  230. txmp.logger.Debug("tx exists already in cache", "tx_hash", tx.Hash())
  231. return nil
  232. }
  233. reqRes, err := txmp.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{Tx: tx})
  234. if err != nil {
  235. txmp.cache.Remove(tx)
  236. return err
  237. }
  238. reqRes.SetCallback(func(res *abci.Response) {
  239. if txmp.recheckCursor != nil {
  240. panic("recheck cursor is non-nil in CheckTx callback")
  241. }
  242. wtx := &WrappedTx{
  243. tx: tx,
  244. hash: txHash,
  245. timestamp: time.Now().UTC(),
  246. height: txmp.height,
  247. }
  248. txmp.initTxCallback(wtx, res, txInfo)
  249. if cb != nil {
  250. cb(res)
  251. }
  252. })
  253. return nil
  254. }
  255. func (txmp *TxMempool) RemoveTxByKey(txKey types.TxKey) error {
  256. txmp.Lock()
  257. defer txmp.Unlock()
  258. // remove the committed transaction from the transaction store and indexes
  259. if wtx := txmp.txStore.GetTxByHash(txKey); wtx != nil {
  260. txmp.removeTx(wtx, false)
  261. return nil
  262. }
  263. return errors.New("transaction not found")
  264. }
  265. // Flush empties the mempool. It acquires a read-lock, fetches all the
  266. // transactions currently in the transaction store and removes each transaction
  267. // from the store and all indexes and finally resets the cache.
  268. //
  269. // NOTE:
  270. // - Flushing the mempool may leave the mempool in an inconsistent state.
  271. func (txmp *TxMempool) Flush() {
  272. txmp.mtx.RLock()
  273. defer txmp.mtx.RUnlock()
  274. txmp.heightIndex.Reset()
  275. txmp.timestampIndex.Reset()
  276. for _, wtx := range txmp.txStore.GetAllTxs() {
  277. txmp.removeTx(wtx, false)
  278. }
  279. atomic.SwapInt64(&txmp.sizeBytes, 0)
  280. txmp.cache.Reset()
  281. }
  282. // ReapMaxBytesMaxGas returns a list of transactions within the provided size
  283. // and gas constraints. Transaction are retrieved in priority order.
  284. //
  285. // NOTE:
  286. // - Transactions returned are not removed from the mempool transaction
  287. // store or indexes.
  288. func (txmp *TxMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs {
  289. txmp.mtx.RLock()
  290. defer txmp.mtx.RUnlock()
  291. var (
  292. totalGas int64
  293. totalSize int64
  294. )
  295. // wTxs contains a list of *WrappedTx retrieved from the priority queue that
  296. // need to be re-enqueued prior to returning.
  297. wTxs := make([]*WrappedTx, 0, txmp.priorityIndex.NumTxs())
  298. defer func() {
  299. for _, wtx := range wTxs {
  300. txmp.priorityIndex.PushTx(wtx)
  301. }
  302. }()
  303. txs := make([]types.Tx, 0, txmp.priorityIndex.NumTxs())
  304. for txmp.priorityIndex.NumTxs() > 0 {
  305. wtx := txmp.priorityIndex.PopTx()
  306. txs = append(txs, wtx.tx)
  307. wTxs = append(wTxs, wtx)
  308. size := types.ComputeProtoSizeForTxs([]types.Tx{wtx.tx})
  309. // Ensure we have capacity for the transaction with respect to the
  310. // transaction size.
  311. if maxBytes > -1 && totalSize+size > maxBytes {
  312. return txs[:len(txs)-1]
  313. }
  314. totalSize += size
  315. // ensure we have capacity for the transaction with respect to total gas
  316. gas := totalGas + wtx.gasWanted
  317. if maxGas > -1 && gas > maxGas {
  318. return txs[:len(txs)-1]
  319. }
  320. totalGas = gas
  321. }
  322. return txs
  323. }
  324. // ReapMaxTxs returns a list of transactions within the provided number of
  325. // transactions bound. Transaction are retrieved in priority order.
  326. //
  327. // NOTE:
  328. // - Transactions returned are not removed from the mempool transaction
  329. // store or indexes.
  330. func (txmp *TxMempool) ReapMaxTxs(max int) types.Txs {
  331. txmp.mtx.RLock()
  332. defer txmp.mtx.RUnlock()
  333. numTxs := txmp.priorityIndex.NumTxs()
  334. if max < 0 {
  335. max = numTxs
  336. }
  337. cap := tmmath.MinInt(numTxs, max)
  338. // wTxs contains a list of *WrappedTx retrieved from the priority queue that
  339. // need to be re-enqueued prior to returning.
  340. wTxs := make([]*WrappedTx, 0, cap)
  341. txs := make([]types.Tx, 0, cap)
  342. for txmp.priorityIndex.NumTxs() > 0 && len(txs) < max {
  343. wtx := txmp.priorityIndex.PopTx()
  344. txs = append(txs, wtx.tx)
  345. wTxs = append(wTxs, wtx)
  346. }
  347. for _, wtx := range wTxs {
  348. txmp.priorityIndex.PushTx(wtx)
  349. }
  350. return txs
  351. }
  352. // Update iterates over all the transactions provided by the block producer,
  353. // removes them from the cache (if applicable), and removes
  354. // the transactions from the main transaction store and associated indexes.
  355. // If there are transactions remaining in the mempool, we initiate a
  356. // re-CheckTx for them (if applicable), otherwise, we notify the caller more
  357. // transactions are available.
  358. //
  359. // NOTE:
  360. // - The caller must explicitly acquire a write-lock.
  361. func (txmp *TxMempool) Update(
  362. blockHeight int64,
  363. blockTxs types.Txs,
  364. deliverTxResponses []*abci.ResponseDeliverTx,
  365. newPreFn PreCheckFunc,
  366. newPostFn PostCheckFunc,
  367. ) error {
  368. txmp.height = blockHeight
  369. txmp.notifiedTxsAvailable = false
  370. if newPreFn != nil {
  371. txmp.preCheck = newPreFn
  372. }
  373. if newPostFn != nil {
  374. txmp.postCheck = newPostFn
  375. }
  376. for i, tx := range blockTxs {
  377. if deliverTxResponses[i].Code == abci.CodeTypeOK {
  378. // add the valid committed transaction to the cache (if missing)
  379. _ = txmp.cache.Push(tx)
  380. } else if !txmp.config.KeepInvalidTxsInCache {
  381. // allow invalid transactions to be re-submitted
  382. txmp.cache.Remove(tx)
  383. }
  384. // remove the committed transaction from the transaction store and indexes
  385. if wtx := txmp.txStore.GetTxByHash(tx.Key()); wtx != nil {
  386. txmp.removeTx(wtx, false)
  387. }
  388. }
  389. txmp.purgeExpiredTxs(blockHeight)
  390. // If there any uncommitted transactions left in the mempool, we either
  391. // initiate re-CheckTx per remaining transaction or notify that remaining
  392. // transactions are left.
  393. if txmp.Size() > 0 {
  394. if txmp.config.Recheck {
  395. txmp.logger.Debug(
  396. "executing re-CheckTx for all remaining transactions",
  397. "num_txs", txmp.Size(),
  398. "height", blockHeight,
  399. )
  400. txmp.updateReCheckTxs()
  401. } else {
  402. txmp.notifyTxsAvailable()
  403. }
  404. }
  405. txmp.metrics.Size.Set(float64(txmp.Size()))
  406. return nil
  407. }
  408. // initTxCallback is the callback invoked for a new unique transaction after CheckTx
  409. // has been executed by the ABCI application for the first time on that transaction.
  410. // CheckTx can be called again for the same transaction later when re-checking;
  411. // however, this callback will not be called.
  412. //
  413. // initTxCallback runs after the ABCI application executes CheckTx.
  414. // It runs the postCheck hook if one is defined on the mempool.
  415. // If the CheckTx response response code is not OK, or if the postCheck hook
  416. // reports an error, the transaction is rejected. Otherwise, we attempt to insert
  417. // the transaction into the mempool.
  418. //
  419. // When inserting a transaction, we first check if there is sufficient capacity.
  420. // If there is, the transaction is added to the txStore and all indexes.
  421. // Otherwise, if the mempool is full, we attempt to find a lower priority transaction
  422. // to evict in place of the new incoming transaction. If no such transaction exists,
  423. // the new incoming transaction is rejected.
  424. //
  425. // NOTE:
  426. // - An explicit lock is NOT required.
  427. func (txmp *TxMempool) initTxCallback(wtx *WrappedTx, res *abci.Response, txInfo TxInfo) {
  428. checkTxRes, ok := res.Value.(*abci.Response_CheckTx)
  429. if !ok {
  430. return
  431. }
  432. var err error
  433. if txmp.postCheck != nil {
  434. err = txmp.postCheck(wtx.tx, checkTxRes.CheckTx)
  435. }
  436. if err != nil || checkTxRes.CheckTx.Code != abci.CodeTypeOK {
  437. // ignore bad transactions
  438. txmp.logger.Info(
  439. "rejected bad transaction",
  440. "priority", wtx.priority,
  441. "tx", fmt.Sprintf("%X", wtx.tx.Hash()),
  442. "peer_id", txInfo.SenderNodeID,
  443. "code", checkTxRes.CheckTx.Code,
  444. "post_check_err", err,
  445. )
  446. txmp.metrics.FailedTxs.Add(1)
  447. if !txmp.config.KeepInvalidTxsInCache {
  448. txmp.cache.Remove(wtx.tx)
  449. }
  450. if err != nil {
  451. checkTxRes.CheckTx.MempoolError = err.Error()
  452. }
  453. return
  454. }
  455. sender := checkTxRes.CheckTx.Sender
  456. priority := checkTxRes.CheckTx.Priority
  457. if len(sender) > 0 {
  458. if wtx := txmp.txStore.GetTxBySender(sender); wtx != nil {
  459. txmp.logger.Error(
  460. "rejected incoming good transaction; tx already exists for sender",
  461. "tx", fmt.Sprintf("%X", wtx.tx.Hash()),
  462. "sender", sender,
  463. )
  464. txmp.metrics.RejectedTxs.Add(1)
  465. return
  466. }
  467. }
  468. if err := txmp.canAddTx(wtx); err != nil {
  469. evictTxs := txmp.priorityIndex.GetEvictableTxs(
  470. priority,
  471. int64(wtx.Size()),
  472. txmp.SizeBytes(),
  473. txmp.config.MaxTxsBytes,
  474. )
  475. if len(evictTxs) == 0 {
  476. // No room for the new incoming transaction so we just remove it from
  477. // the cache.
  478. txmp.cache.Remove(wtx.tx)
  479. txmp.logger.Error(
  480. "rejected incoming good transaction; mempool full",
  481. "tx", fmt.Sprintf("%X", wtx.tx.Hash()),
  482. "err", err.Error(),
  483. )
  484. txmp.metrics.RejectedTxs.Add(1)
  485. return
  486. }
  487. // evict an existing transaction(s)
  488. //
  489. // NOTE:
  490. // - The transaction, toEvict, can be removed while a concurrent
  491. // reCheckTx callback is being executed for the same transaction.
  492. for _, toEvict := range evictTxs {
  493. txmp.removeTx(toEvict, true)
  494. txmp.logger.Debug(
  495. "evicted existing good transaction; mempool full",
  496. "old_tx", fmt.Sprintf("%X", toEvict.tx.Hash()),
  497. "old_priority", toEvict.priority,
  498. "new_tx", fmt.Sprintf("%X", wtx.tx.Hash()),
  499. "new_priority", wtx.priority,
  500. )
  501. txmp.metrics.EvictedTxs.Add(1)
  502. }
  503. }
  504. wtx.gasWanted = checkTxRes.CheckTx.GasWanted
  505. wtx.priority = priority
  506. wtx.sender = sender
  507. wtx.peers = map[uint16]struct{}{
  508. txInfo.SenderID: {},
  509. }
  510. txmp.metrics.TxSizeBytes.Observe(float64(wtx.Size()))
  511. txmp.metrics.Size.Set(float64(txmp.Size()))
  512. txmp.insertTx(wtx)
  513. txmp.logger.Debug(
  514. "inserted good transaction",
  515. "priority", wtx.priority,
  516. "tx", fmt.Sprintf("%X", wtx.tx.Hash()),
  517. "height", txmp.height,
  518. "num_txs", txmp.Size(),
  519. )
  520. txmp.notifyTxsAvailable()
  521. }
  522. // defaultTxCallback is the CheckTx application callback used when a transaction
  523. // is being re-checked (if re-checking is enabled). The caller must hold a mempool
  524. // write-lock (via Lock()) and when executing Update(), if the mempool is non-empty
  525. // and Recheck is enabled, then all remaining transactions will be rechecked via
  526. // CheckTxAsync. The order transactions are rechecked must be the same as the
  527. // order in which this callback is called.
  528. func (txmp *TxMempool) defaultTxCallback(req *abci.Request, res *abci.Response) {
  529. if txmp.recheckCursor == nil {
  530. return
  531. }
  532. txmp.metrics.RecheckTimes.Add(1)
  533. checkTxRes, ok := res.Value.(*abci.Response_CheckTx)
  534. if !ok {
  535. txmp.logger.Error("received incorrect type in mempool callback",
  536. "expected", reflect.TypeOf(&abci.Response_CheckTx{}).Name(),
  537. "got", reflect.TypeOf(res.Value).Name(),
  538. )
  539. return
  540. }
  541. tx := req.GetCheckTx().Tx
  542. wtx := txmp.recheckCursor.Value.(*WrappedTx)
  543. // Search through the remaining list of tx to recheck for a transaction that matches
  544. // the one we received from the ABCI application.
  545. for {
  546. if bytes.Equal(tx, wtx.tx) {
  547. // We've found a tx in the recheck list that matches the tx that we
  548. // received from the ABCI application.
  549. // Break, and use this transaction for further checks.
  550. break
  551. }
  552. txmp.logger.Error(
  553. "re-CheckTx transaction mismatch",
  554. "got", wtx.tx.Hash(),
  555. "expected", types.Tx(tx).Key(),
  556. )
  557. if txmp.recheckCursor == txmp.recheckEnd {
  558. // we reached the end of the recheckTx list without finding a tx
  559. // matching the one we received from the ABCI application.
  560. // Return without processing any tx.
  561. txmp.recheckCursor = nil
  562. return
  563. }
  564. txmp.recheckCursor = txmp.recheckCursor.Next()
  565. wtx = txmp.recheckCursor.Value.(*WrappedTx)
  566. }
  567. // Only evaluate transactions that have not been removed. This can happen
  568. // if an existing transaction is evicted during CheckTx and while this
  569. // callback is being executed for the same evicted transaction.
  570. if !txmp.txStore.IsTxRemoved(wtx.hash) {
  571. var err error
  572. if txmp.postCheck != nil {
  573. err = txmp.postCheck(tx, checkTxRes.CheckTx)
  574. }
  575. if checkTxRes.CheckTx.Code == abci.CodeTypeOK && err == nil {
  576. wtx.priority = checkTxRes.CheckTx.Priority
  577. } else {
  578. txmp.logger.Debug(
  579. "existing transaction no longer valid; failed re-CheckTx callback",
  580. "priority", wtx.priority,
  581. "tx", fmt.Sprintf("%X", wtx.tx.Hash()),
  582. "err", err,
  583. "code", checkTxRes.CheckTx.Code,
  584. )
  585. if wtx.gossipEl != txmp.recheckCursor {
  586. panic("corrupted reCheckTx cursor")
  587. }
  588. txmp.removeTx(wtx, !txmp.config.KeepInvalidTxsInCache)
  589. }
  590. }
  591. // move reCheckTx cursor to next element
  592. if txmp.recheckCursor == txmp.recheckEnd {
  593. txmp.recheckCursor = nil
  594. } else {
  595. txmp.recheckCursor = txmp.recheckCursor.Next()
  596. }
  597. if txmp.recheckCursor == nil {
  598. txmp.logger.Debug("finished rechecking transactions")
  599. if txmp.Size() > 0 {
  600. txmp.notifyTxsAvailable()
  601. }
  602. }
  603. txmp.metrics.Size.Set(float64(txmp.Size()))
  604. }
  605. // updateReCheckTxs updates the recheck cursors using the gossipIndex. For
  606. // each transaction, it executes CheckTxAsync. The global callback defined on
  607. // the proxyAppConn will be executed for each transaction after CheckTx is
  608. // executed.
  609. //
  610. // NOTE:
  611. // - The caller must have a write-lock when executing updateReCheckTxs.
  612. func (txmp *TxMempool) updateReCheckTxs() {
  613. if txmp.Size() == 0 {
  614. panic("attempted to update re-CheckTx txs when mempool is empty")
  615. }
  616. txmp.recheckCursor = txmp.gossipIndex.Front()
  617. txmp.recheckEnd = txmp.gossipIndex.Back()
  618. ctx := context.Background()
  619. for e := txmp.gossipIndex.Front(); e != nil; e = e.Next() {
  620. wtx := e.Value.(*WrappedTx)
  621. // Only execute CheckTx if the transaction is not marked as removed which
  622. // could happen if the transaction was evicted.
  623. if !txmp.txStore.IsTxRemoved(wtx.hash) {
  624. _, err := txmp.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{
  625. Tx: wtx.tx,
  626. Type: abci.CheckTxType_Recheck,
  627. })
  628. if err != nil {
  629. // no need in retrying since the tx will be rechecked after the next block
  630. txmp.logger.Error("failed to execute CheckTx during rechecking", "err", err)
  631. }
  632. }
  633. }
  634. if _, err := txmp.proxyAppConn.FlushAsync(ctx); err != nil {
  635. txmp.logger.Error("failed to flush transactions during rechecking", "err", err)
  636. }
  637. }
  638. // canAddTx returns an error if we cannot insert the provided *WrappedTx into
  639. // the mempool due to mempool configured constraints. If it returns nil,
  640. // the transaction can be inserted into the mempool.
  641. func (txmp *TxMempool) canAddTx(wtx *WrappedTx) error {
  642. var (
  643. numTxs = txmp.Size()
  644. sizeBytes = txmp.SizeBytes()
  645. )
  646. if numTxs >= txmp.config.Size || int64(wtx.Size())+sizeBytes > txmp.config.MaxTxsBytes {
  647. return types.ErrMempoolIsFull{
  648. NumTxs: numTxs,
  649. MaxTxs: txmp.config.Size,
  650. TxsBytes: sizeBytes,
  651. MaxTxsBytes: txmp.config.MaxTxsBytes,
  652. }
  653. }
  654. return nil
  655. }
  656. func (txmp *TxMempool) insertTx(wtx *WrappedTx) {
  657. txmp.txStore.SetTx(wtx)
  658. txmp.priorityIndex.PushTx(wtx)
  659. txmp.heightIndex.Insert(wtx)
  660. txmp.timestampIndex.Insert(wtx)
  661. // Insert the transaction into the gossip index and mark the reference to the
  662. // linked-list element, which will be needed at a later point when the
  663. // transaction is removed.
  664. gossipEl := txmp.gossipIndex.PushBack(wtx)
  665. wtx.gossipEl = gossipEl
  666. atomic.AddInt64(&txmp.sizeBytes, int64(wtx.Size()))
  667. }
  668. func (txmp *TxMempool) removeTx(wtx *WrappedTx, removeFromCache bool) {
  669. if txmp.txStore.IsTxRemoved(wtx.hash) {
  670. return
  671. }
  672. txmp.txStore.RemoveTx(wtx)
  673. txmp.priorityIndex.RemoveTx(wtx)
  674. txmp.heightIndex.Remove(wtx)
  675. txmp.timestampIndex.Remove(wtx)
  676. // Remove the transaction from the gossip index and cleanup the linked-list
  677. // element so it can be garbage collected.
  678. txmp.gossipIndex.Remove(wtx.gossipEl)
  679. wtx.gossipEl.DetachPrev()
  680. atomic.AddInt64(&txmp.sizeBytes, int64(-wtx.Size()))
  681. if removeFromCache {
  682. txmp.cache.Remove(wtx.tx)
  683. }
  684. }
  685. // purgeExpiredTxs removes all transactions that have exceeded their respective
  686. // height- and/or time-based TTLs from their respective indexes. Every expired
  687. // transaction will be removed from the mempool, but preserved in the cache.
  688. //
  689. // NOTE: purgeExpiredTxs must only be called during TxMempool#Update in which
  690. // the caller has a write-lock on the mempool and so we can safely iterate over
  691. // the height and time based indexes.
  692. func (txmp *TxMempool) purgeExpiredTxs(blockHeight int64) {
  693. now := time.Now()
  694. expiredTxs := make(map[types.TxKey]*WrappedTx)
  695. if txmp.config.TTLNumBlocks > 0 {
  696. purgeIdx := -1
  697. for i, wtx := range txmp.heightIndex.txs {
  698. if (blockHeight - wtx.height) > txmp.config.TTLNumBlocks {
  699. expiredTxs[wtx.tx.Key()] = wtx
  700. purgeIdx = i
  701. } else {
  702. // since the index is sorted, we know no other txs can be be purged
  703. break
  704. }
  705. }
  706. if purgeIdx >= 0 {
  707. txmp.heightIndex.txs = txmp.heightIndex.txs[purgeIdx+1:]
  708. }
  709. }
  710. if txmp.config.TTLDuration > 0 {
  711. purgeIdx := -1
  712. for i, wtx := range txmp.timestampIndex.txs {
  713. if now.Sub(wtx.timestamp) > txmp.config.TTLDuration {
  714. expiredTxs[wtx.tx.Key()] = wtx
  715. purgeIdx = i
  716. } else {
  717. // since the index is sorted, we know no other txs can be be purged
  718. break
  719. }
  720. }
  721. if purgeIdx >= 0 {
  722. txmp.timestampIndex.txs = txmp.timestampIndex.txs[purgeIdx+1:]
  723. }
  724. }
  725. for _, wtx := range expiredTxs {
  726. txmp.removeTx(wtx, false)
  727. }
  728. }
  729. func (txmp *TxMempool) notifyTxsAvailable() {
  730. if txmp.Size() == 0 {
  731. panic("attempt to notify txs available but mempool is empty!")
  732. }
  733. if txmp.txsAvailable != nil && !txmp.notifiedTxsAvailable {
  734. // channel cap is 1, so this will send once
  735. txmp.notifiedTxsAvailable = true
  736. select {
  737. case txmp.txsAvailable <- struct{}{}:
  738. default:
  739. }
  740. }
  741. }