fluencelabs/tendermint
1
0
Fork 0
mirror of https://github.com/fluencelabs/tendermint synced 2025-04-25 06:42:16 +00:00
Code Issues Projects Releases Wiki Activity

refactor mempool.Update

Browse Source 
- rename filterTxs to removeTxs
- move txsMap into removeTxs func
- rename goodTxs to txsLeft
This commit is contained in:
Anton Kaliaev 2018-11-19 16:16:33 +04:00
parent 7b883a5457
commit e9efbfe267
1 changed files with 22 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
mempool/mempool.go
View File

@ -534,12 +534,6 @@ func (mem *Mempool) Update(
preCheck PreCheckFunc, preCheck PreCheckFunc,
postCheck PostCheckFunc, postCheck PostCheckFunc,
) error { ) error {
// First, create a lookup map of txns in new txs.
txsMap := make(map[string]struct{}, len(txs))
for _, tx := range txs {
txsMap[string(tx)] = struct{}{}
}
// Set height // Set height
mem.height = height mem.height = height
mem.notifiedTxsAvailable = false mem.notifiedTxsAvailable = false
@ -551,12 +545,13 @@ func (mem *Mempool) Update(
mem.postCheck = postCheck mem.postCheck = postCheck
} }
// Remove transactions that are already in txs. // Remove committed transactions.
goodTxs := mem.filterTxs(txsMap) txsLeft := mem.removeTxs(txs)
// Recheck mempool txs if any txs were committed in the block // Recheck mempool txs if any txs were committed in the block
if mem.config.Recheck && len(goodTxs) > 0 { if mem.config.Recheck && len(txsLeft) > 0 {
mem.logger.Info("Recheck txs", "numtxs", len(goodTxs), "height", height) mem.logger.Info("Recheck txs", "numtxs", len(txsLeft), "height", height)
mem.recheckTxs(goodTxs) mem.recheckTxs(txsLeft)
// At this point, mem.txs are being rechecked. // At this point, mem.txs are being rechecked.
// mem.recheckCursor re-scans mem.txs and possibly removes some txs. // mem.recheckCursor re-scans mem.txs and possibly removes some txs.
// Before mem.Reap(), we should wait for mem.recheckCursor to be nil. // Before mem.Reap(), we should wait for mem.recheckCursor to be nil.
@ -568,12 +563,18 @@ func (mem *Mempool) Update(
return nil return nil
} }
func (mem *Mempool) filterTxs(blockTxsMap map[string]struct{}) []types.Tx { func (mem *Mempool) removeTxs(txs types.Txs) []types.Tx {
goodTxs := make([]types.Tx, 0, mem.txs.Len()) // Build a map for faster lookups.
txsMap := make(map[string]struct{}, len(txs))
for _, tx := range txs {
txsMap[string(tx)] = struct{}{}
}
txsLeft := make([]types.Tx, 0, mem.txs.Len())
for e := mem.txs.Front(); e != nil; e = e.Next() { for e := mem.txs.Front(); e != nil; e = e.Next() {
memTx := e.Value.(*mempoolTx) memTx := e.Value.(*mempoolTx)
// Remove the tx if it's alredy in a block. // Remove the tx if it's already in a block.
if _, ok := blockTxsMap[string(memTx.tx)]; ok { if _, ok := txsMap[string(memTx.tx)]; ok {
// remove from clist // remove from clist
mem.txs.Remove(e) mem.txs.Remove(e)
e.DetachPrev() e.DetachPrev()
@ -581,15 +582,14 @@ func (mem *Mempool) filterTxs(blockTxsMap map[string]struct{}) []types.Tx {
// NOTE: we don't remove committed txs from the cache. // NOTE: we don't remove committed txs from the cache.
continue continue
} }
// Good tx! txsLeft = append(txsLeft, memTx.tx)
goodTxs = append(goodTxs, memTx.tx)
} }
return goodTxs return txsLeft
} }
// NOTE: pass in goodTxs because mem.txs can mutate concurrently. // NOTE: pass in txs because mem.txs can mutate concurrently.
func (mem *Mempool) recheckTxs(goodTxs []types.Tx) { func (mem *Mempool) recheckTxs(txs []types.Tx) {
if len(goodTxs) == 0 { if len(txs) == 0 {
return return
} }
atomic.StoreInt32(&mem.rechecking, 1) atomic.StoreInt32(&mem.rechecking, 1)
@ -598,7 +598,7 @@ func (mem *Mempool) recheckTxs(goodTxs []types.Tx) {
// Push txs to proxyAppConn // Push txs to proxyAppConn
// NOTE: resCb() may be called concurrently. // NOTE: resCb() may be called concurrently.
for _, tx := range goodTxs { for _, tx := range txs {
mem.proxyAppConn.CheckTxAsync(tx) mem.proxyAppConn.CheckTxAsync(tx)
} }
mem.proxyAppConn.FlushAsync() mem.proxyAppConn.FlushAsync()