mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-27 18:34:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

mempool: v1 implementation (#6466)

Browse Source
This commit is contained in:
Aleksandr Bezobchuk
2021-06-01 11:17:45 -04:00
committed by GitHub
parent 4e06dfef8c
commit 1e4bc04cd6
42 changed files with 3323 additions and 736 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mempool/v0/bench_test.go
+100
View File
@@ -0,0 +1,100 @@
package v0
import (
"encoding/binary"
"sync/atomic"
"testing"
"github.com/tendermint/tendermint/abci/example/kvstore"
"github.com/tendermint/tendermint/mempool"
"github.com/tendermint/tendermint/proxy"
)
func BenchmarkReap(b *testing.B) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
mp.config.Size = 100000
size := 10000
for i := 0; i < size; i++ {
tx := make([]byte, 8)
binary.BigEndian.PutUint64(tx, uint64(i))
if err := mp.CheckTx(tx, nil, mempool.TxInfo{}); err != nil {
b.Fatal(err)
}
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
mp.ReapMaxBytesMaxGas(100000000, 10000000)
}
}
func BenchmarkCheckTx(b *testing.B) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
mp.config.Size = 1000000
b.ResetTimer()
for i := 0; i < b.N; i++ {
b.StopTimer()
tx := make([]byte, 8)
binary.BigEndian.PutUint64(tx, uint64(i))
b.StartTimer()
if err := mp.CheckTx(tx, nil, mempool.TxInfo{}); err != nil {
b.Fatal(err)
}
}
}
func BenchmarkParallelCheckTx(b *testing.B) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
mp.config.Size = 100000000
var txcnt uint64
next := func() uint64 {
return atomic.AddUint64(&txcnt, 1) - 1
}
b.ResetTimer()
b.RunParallel(func(pb *testing.PB) {
for pb.Next() {
tx := make([]byte, 8)
binary.BigEndian.PutUint64(tx, next())
if err := mp.CheckTx(tx, nil, mempool.TxInfo{}); err != nil {
b.Fatal(err)
}
}
})
}
func BenchmarkCheckDuplicateTx(b *testing.B) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
mp.config.Size = 1000000
for i := 0; i < b.N; i++ {
tx := make([]byte, 8)
binary.BigEndian.PutUint64(tx, uint64(i))
if err := mp.CheckTx(tx, nil, mempool.TxInfo{}); err != nil {
b.Fatal(err)
}
if err := mp.CheckTx(tx, nil, mempool.TxInfo{}); err == nil {
b.Fatal("tx should be duplicate")
}
}
}
mempool/v0/cache_test.go
+81
View File
@@ -0,0 +1,81 @@
package v0
import (
"crypto/sha256"
"testing"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/abci/example/kvstore"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/mempool"
"github.com/tendermint/tendermint/proxy"
"github.com/tendermint/tendermint/types"
)
func TestCacheAfterUpdate(t *testing.T) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
// reAddIndices & txsInCache can have elements > numTxsToCreate
// also assumes max index is 255 for convenience
// txs in cache also checks order of elements
tests := []struct {
numTxsToCreate int
updateIndices []int
reAddIndices []int
txsInCache []int
}{
{1, []int{}, []int{1}, []int{1, 0}}, // adding new txs works
{2, []int{1}, []int{}, []int{1, 0}}, // update doesn't remove tx from cache
{2, []int{2}, []int{}, []int{2, 1, 0}}, // update adds new tx to cache
{2, []int{1}, []int{1}, []int{1, 0}}, // re-adding after update doesn't make dupe
}
for tcIndex, tc := range tests {
for i := 0; i < tc.numTxsToCreate; i++ {
tx := types.Tx{byte(i)}
err := mp.CheckTx(tx, nil, mempool.TxInfo{})
require.NoError(t, err)
}
updateTxs := []types.Tx{}
for _, v := range tc.updateIndices {
tx := types.Tx{byte(v)}
updateTxs = append(updateTxs, tx)
}
err := mp.Update(int64(tcIndex), updateTxs, abciResponses(len(updateTxs), abci.CodeTypeOK), nil, nil)
require.NoError(t, err)
for _, v := range tc.reAddIndices {
tx := types.Tx{byte(v)}
_ = mp.CheckTx(tx, nil, mempool.TxInfo{})
}
cache := mp.cache.(*mempool.LRUTxCache)
node := cache.GetList().Front()
counter := 0
for node != nil {
require.NotEqual(t, len(tc.txsInCache), counter,
"cache larger than expected on testcase %d", tcIndex)
nodeVal := node.Value.([sha256.Size]byte)
expectedBz := sha256.Sum256([]byte{byte(tc.txsInCache[len(tc.txsInCache)-counter-1])})
// Reference for reading the errors:
// >>> sha256('\x00').hexdigest()
// '6e340b9cffb37a989ca544e6bb780a2c78901d3fb33738768511a30617afa01d'
// >>> sha256('\x01').hexdigest()
// '4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a'
// >>> sha256('\x02').hexdigest()
// 'dbc1b4c900ffe48d575b5da5c638040125f65db0fe3e24494b76ea986457d986'
require.Equal(t, expectedBz, nodeVal, "Equality failed on index %d, tc %d", counter, tcIndex)
counter++
node = node.Next()
}
require.Equal(t, len(tc.txsInCache), counter,
"cache smaller than expected on testcase %d", tcIndex)
mp.Flush()
}
}
mempool/v0/clist_mempool.go
+668
View File
@@ -0,0 +1,668 @@
package v0
import (
"bytes"
"context"
"fmt"
"sync"
"sync/atomic"
abci "github.com/tendermint/tendermint/abci/types"
cfg "github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/internal/libs/clist"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/libs/log"
tmmath "github.com/tendermint/tendermint/libs/math"
"github.com/tendermint/tendermint/mempool"
"github.com/tendermint/tendermint/p2p"
"github.com/tendermint/tendermint/proxy"
"github.com/tendermint/tendermint/types"
)
// CListMempool is an ordered in-memory pool for transactions before they are
// proposed in a consensus round. Transaction validity is checked using the
// CheckTx abci message before the transaction is added to the pool. The
// mempool uses a concurrent list structure for storing transactions that can
// be efficiently accessed by multiple concurrent readers.
type CListMempool struct {
// Atomic integers
height int64 // the last block Update()'d to
txsBytes int64 // total size of mempool, in bytes
// notify listeners (ie. consensus) when txs are available
notifiedTxsAvailable bool
txsAvailable chan struct{} // fires once for each height, when the mempool is not empty
config *cfg.MempoolConfig
// Exclusive mutex for Update method to prevent concurrent execution of
// CheckTx or ReapMaxBytesMaxGas(ReapMaxTxs) methods.
updateMtx tmsync.RWMutex
preCheck mempool.PreCheckFunc
postCheck mempool.PostCheckFunc
txs *clist.CList // concurrent linked-list of good txs
proxyAppConn proxy.AppConnMempool
// Track whether we're rechecking txs.
// These are not protected by a mutex and are expected to be mutated in
// serial (ie. by abci responses which are called in serial).
recheckCursor *clist.CElement // next expected response
recheckEnd *clist.CElement // re-checking stops here
// Map for quick access to txs to record sender in CheckTx.
// txsMap: txKey -> CElement
txsMap sync.Map
// Keep a cache of already-seen txs.
// This reduces the pressure on the proxyApp.
cache mempool.TxCache
logger log.Logger
metrics *mempool.Metrics
}
var _ mempool.Mempool = &CListMempool{}
// CListMempoolOption sets an optional parameter on the mempool.
type CListMempoolOption func(*CListMempool)
// NewCListMempool returns a new mempool with the given configuration and
// connection to an application.
func NewCListMempool(
config *cfg.MempoolConfig,
proxyAppConn proxy.AppConnMempool,
height int64,
options ...CListMempoolOption,
) *CListMempool {
mp := &CListMempool{
config: config,
proxyAppConn: proxyAppConn,
txs: clist.New(),
height: height,
recheckCursor: nil,
recheckEnd: nil,
logger: log.NewNopLogger(),
metrics: mempool.NopMetrics(),
}
if config.CacheSize > 0 {
mp.cache = mempool.NewLRUTxCache(config.CacheSize)
} else {
mp.cache = mempool.NopTxCache{}
}
proxyAppConn.SetResponseCallback(mp.globalCb)
for _, option := range options {
option(mp)
}
return mp
}
// NOTE: not thread safe - should only be called once, on startup
func (mem *CListMempool) EnableTxsAvailable() {
mem.txsAvailable = make(chan struct{}, 1)
}
// SetLogger sets the Logger.
func (mem *CListMempool) SetLogger(l log.Logger) {
mem.logger = l
}
// WithPreCheck sets a filter for the mempool to reject a tx if f(tx) returns
// false. This is ran before CheckTx. Only applies to the first created block.
// After that, Update overwrites the existing value.
func WithPreCheck(f mempool.PreCheckFunc) CListMempoolOption {
return func(mem *CListMempool) { mem.preCheck = f }
}
// WithPostCheck sets a filter for the mempool to reject a tx if f(tx) returns
// false. This is ran after CheckTx. Only applies to the first created block.
// After that, Update overwrites the existing value.
func WithPostCheck(f mempool.PostCheckFunc) CListMempoolOption {
return func(mem *CListMempool) { mem.postCheck = f }
}
// WithMetrics sets the metrics.
func WithMetrics(metrics *mempool.Metrics) CListMempoolOption {
return func(mem *CListMempool) { mem.metrics = metrics }
}
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) Lock() {
mem.updateMtx.Lock()
}
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) Unlock() {
mem.updateMtx.Unlock()
}
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) Size() int {
return mem.txs.Len()
}
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) SizeBytes() int64 {
return atomic.LoadInt64(&mem.txsBytes)
}
// Lock() must be help by the caller during execution.
func (mem *CListMempool) FlushAppConn() error {
return mem.proxyAppConn.FlushSync(context.Background())
}
// XXX: Unsafe! Calling Flush may leave mempool in inconsistent state.
func (mem *CListMempool) Flush() {
mem.updateMtx.RLock()
defer mem.updateMtx.RUnlock()
_ = atomic.SwapInt64(&mem.txsBytes, 0)
mem.cache.Reset()
for e := mem.txs.Front(); e != nil; e = e.Next() {
mem.txs.Remove(e)
e.DetachPrev()
}
mem.txsMap.Range(func(key, _ interface{}) bool {
mem.txsMap.Delete(key)
return true
})
}
// TxsFront returns the first transaction in the ordered list for peer
// goroutines to call .NextWait() on.
// FIXME: leaking implementation details!
//
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) TxsFront() *clist.CElement {
return mem.txs.Front()
}
// TxsWaitChan returns a channel to wait on transactions. It will be closed
// once the mempool is not empty (ie. the internal `mem.txs` has at least one
// element)
//
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) TxsWaitChan() <-chan struct{} {
return mem.txs.WaitChan()
}
// It blocks if we're waiting on Update() or Reap().
// cb: A callback from the CheckTx command.
// It gets called from another goroutine.
// CONTRACT: Either cb will get called, or err returned.
//
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) CheckTx(tx types.Tx, cb func(*abci.Response), txInfo mempool.TxInfo) error {
mem.updateMtx.RLock()
// use defer to unlock mutex because application (*local client*) might panic
defer mem.updateMtx.RUnlock()
txSize := len(tx)
if err := mem.isFull(txSize); err != nil {
return err
}
if txSize > mem.config.MaxTxBytes {
return mempool.ErrTxTooLarge{
Max: mem.config.MaxTxBytes,
Actual: txSize,
}
}
if mem.preCheck != nil {
if err := mem.preCheck(tx); err != nil {
return mempool.ErrPreCheck{
Reason: err,
}
}
}
// NOTE: proxyAppConn may error if tx buffer is full
if err := mem.proxyAppConn.Error(); err != nil {
return err
}
if !mem.cache.Push(tx) { // if the transaction already exists in the cache
// Record a new sender for a tx we've already seen.
// Note it's possible a tx is still in the cache but no longer in the mempool
// (eg. after committing a block, txs are removed from mempool but not cache),
// so we only record the sender for txs still in the mempool.
if e, ok := mem.txsMap.Load(mempool.TxKey(tx)); ok {
memTx := e.(*clist.CElement).Value.(*mempoolTx)
_, loaded := memTx.senders.LoadOrStore(txInfo.SenderID, true)
// TODO: consider punishing peer for dups,
// its non-trivial since invalid txs can become valid,
// but they can spam the same tx with little cost to them atm.
if loaded {
return mempool.ErrTxInCache
}
}
mem.logger.Debug("tx exists already in cache", "tx_hash", tx.Hash())
return nil
}
ctx := context.Background()
if txInfo.Context != nil {
ctx = txInfo.Context
}
reqRes, err := mem.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{Tx: tx})
if err != nil {
mem.cache.Remove(tx)
return err
}
reqRes.SetCallback(mem.reqResCb(tx, txInfo.SenderID, txInfo.SenderNodeID, cb))
return nil
}
// Global callback that will be called after every ABCI response.
// Having a single global callback avoids needing to set a callback for each request.
// However, processing the checkTx response requires the peerID (so we can track which txs we heard from who),
// and peerID is not included in the ABCI request, so we have to set request-specific callbacks that
// include this information. If we're not in the midst of a recheck, this function will just return,
// so the request specific callback can do the work.
//
// When rechecking, we don't need the peerID, so the recheck callback happens
// here.
func (mem *CListMempool) globalCb(req *abci.Request, res *abci.Response) {
if mem.recheckCursor == nil {
return
}
mem.metrics.RecheckTimes.Add(1)
mem.resCbRecheck(req, res)
// update metrics
mem.metrics.Size.Set(float64(mem.Size()))
}
// Request specific callback that should be set on individual reqRes objects
// to incorporate local information when processing the response.
// This allows us to track the peer that sent us this tx, so we can avoid sending it back to them.
// NOTE: alternatively, we could include this information in the ABCI request itself.
//
// External callers of CheckTx, like the RPC, can also pass an externalCb through here that is called
// when all other response processing is complete.
//
// Used in CheckTx to record PeerID who sent us the tx.
func (mem *CListMempool) reqResCb(
tx []byte,
peerID uint16,
peerP2PID p2p.NodeID,
externalCb func(*abci.Response),
) func(res *abci.Response) {
return func(res *abci.Response) {
if mem.recheckCursor != nil {
// this should never happen
panic("recheck cursor is not nil in reqResCb")
}
mem.resCbFirstTime(tx, peerID, peerP2PID, res)
// update metrics
mem.metrics.Size.Set(float64(mem.Size()))
// passed in by the caller of CheckTx, eg. the RPC
if externalCb != nil {
externalCb(res)
}
}
}
// Called from:
// - resCbFirstTime (lock not held) if tx is valid
func (mem *CListMempool) addTx(memTx *mempoolTx) {
e := mem.txs.PushBack(memTx)
mem.txsMap.Store(mempool.TxKey(memTx.tx), e)
atomic.AddInt64(&mem.txsBytes, int64(len(memTx.tx)))
mem.metrics.TxSizeBytes.Observe(float64(len(memTx.tx)))
}
// Called from:
// - Update (lock held) if tx was committed
// - resCbRecheck (lock not held) if tx was invalidated
func (mem *CListMempool) removeTx(tx types.Tx, elem *clist.CElement, removeFromCache bool) {
mem.txs.Remove(elem)
elem.DetachPrev()
mem.txsMap.Delete(mempool.TxKey(tx))
atomic.AddInt64(&mem.txsBytes, int64(-len(tx)))
if removeFromCache {
mem.cache.Remove(tx)
}
}
// RemoveTxByKey removes a transaction from the mempool by its TxKey index.
func (mem *CListMempool) RemoveTxByKey(txKey [mempool.TxKeySize]byte, removeFromCache bool) {
if e, ok := mem.txsMap.Load(txKey); ok {
memTx := e.(*clist.CElement).Value.(*mempoolTx)
if memTx != nil {
mem.removeTx(memTx.tx, e.(*clist.CElement), removeFromCache)
}
}
}
func (mem *CListMempool) isFull(txSize int) error {
var (
memSize = mem.Size()
txsBytes = mem.SizeBytes()
)
if memSize >= mem.config.Size || int64(txSize)+txsBytes > mem.config.MaxTxsBytes {
return mempool.ErrMempoolIsFull{
NumTxs: memSize,
MaxTxs: mem.config.Size,
TxsBytes: txsBytes,
MaxTxsBytes: mem.config.MaxTxsBytes,
}
}
return nil
}
// callback, which is called after the app checked the tx for the first time.
//
// The case where the app checks the tx for the second and subsequent times is
// handled by the resCbRecheck callback.
func (mem *CListMempool) resCbFirstTime(
tx []byte,
peerID uint16,
peerP2PID p2p.NodeID,
res *abci.Response,
) {
switch r := res.Value.(type) {
case *abci.Response_CheckTx:
var postCheckErr error
if mem.postCheck != nil {
postCheckErr = mem.postCheck(tx, r.CheckTx)
}
if (r.CheckTx.Code == abci.CodeTypeOK) && postCheckErr == nil {
// Check mempool isn't full again to reduce the chance of exceeding the
// limits.
if err := mem.isFull(len(tx)); err != nil {
// remove from cache (mempool might have a space later)
mem.cache.Remove(tx)
mem.logger.Error(err.Error())
return
}
memTx := &mempoolTx{
height: mem.height,
gasWanted: r.CheckTx.GasWanted,
tx: tx,
}
memTx.senders.Store(peerID, true)
mem.addTx(memTx)
mem.logger.Debug(
"added good transaction",
"tx", mempool.TxHashFromBytes(tx),
"res", r,
"height", memTx.height,
"total", mem.Size(),
)
mem.notifyTxsAvailable()
} else {
// ignore bad transaction
mem.logger.Debug(
"rejected bad transaction",
"tx", mempool.TxHashFromBytes(tx),
"peerID", peerP2PID,
"res", r,
"err", postCheckErr,
)
mem.metrics.FailedTxs.Add(1)
if !mem.config.KeepInvalidTxsInCache {
// remove from cache (it might be good later)
mem.cache.Remove(tx)
}
}
default:
// ignore other messages
}
}
// callback, which is called after the app rechecked the tx.
//
// The case where the app checks the tx for the first time is handled by the
// resCbFirstTime callback.
func (mem *CListMempool) resCbRecheck(req *abci.Request, res *abci.Response) {
switch r := res.Value.(type) {
case *abci.Response_CheckTx:
tx := req.GetCheckTx().Tx
memTx := mem.recheckCursor.Value.(*mempoolTx)
if !bytes.Equal(tx, memTx.tx) {
panic(fmt.Sprintf(
"Unexpected tx response from proxy during recheck\nExpected %X, got %X",
memTx.tx,
tx))
}
var postCheckErr error
if mem.postCheck != nil {
postCheckErr = mem.postCheck(tx, r.CheckTx)
}
if (r.CheckTx.Code == abci.CodeTypeOK) && postCheckErr == nil {
// Good, nothing to do.
} else {
// Tx became invalidated due to newly committed block.
mem.logger.Debug("tx is no longer valid", "tx", mempool.TxHashFromBytes(tx), "res", r, "err", postCheckErr)
// NOTE: we remove tx from the cache because it might be good later
mem.removeTx(tx, mem.recheckCursor, !mem.config.KeepInvalidTxsInCache)
}
if mem.recheckCursor == mem.recheckEnd {
mem.recheckCursor = nil
} else {
mem.recheckCursor = mem.recheckCursor.Next()
}
if mem.recheckCursor == nil {
// Done!
mem.logger.Debug("done rechecking txs")
// incase the recheck removed all txs
if mem.Size() > 0 {
mem.notifyTxsAvailable()
}
}
default:
// ignore other messages
}
}
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) TxsAvailable() <-chan struct{} {
return mem.txsAvailable
}
func (mem *CListMempool) notifyTxsAvailable() {
if mem.Size() == 0 {
panic("notified txs available but mempool is empty!")
}
if mem.txsAvailable != nil && !mem.notifiedTxsAvailable {
// channel cap is 1, so this will send once
mem.notifiedTxsAvailable = true
select {
case mem.txsAvailable <- struct{}{}:
default:
}
}
}
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) ReapMaxBytesMaxGas(maxBytes, maxGas int64) types.Txs {
mem.updateMtx.RLock()
defer mem.updateMtx.RUnlock()
var (
totalGas int64
runningSize int64
)
// TODO: we will get a performance boost if we have a good estimate of avg
// size per tx, and set the initial capacity based off of that.
// txs := make([]types.Tx, 0, tmmath.MinInt(mem.txs.Len(), max/mem.avgTxSize))
txs := make([]types.Tx, 0, mem.txs.Len())
for e := mem.txs.Front(); e != nil; e = e.Next() {
memTx := e.Value.(*mempoolTx)
txs = append(txs, memTx.tx)
dataSize := types.ComputeProtoSizeForTxs([]types.Tx{memTx.tx})
// Check total size requirement
if maxBytes > -1 && runningSize+dataSize > maxBytes {
return txs[:len(txs)-1]
}
runningSize += dataSize
// Check total gas requirement.
// If maxGas is negative, skip this check.
// Since newTotalGas < masGas, which
// must be non-negative, it follows that this won't overflow.
newTotalGas := totalGas + memTx.gasWanted
if maxGas > -1 && newTotalGas > maxGas {
return txs[:len(txs)-1]
}
totalGas = newTotalGas
}
return txs
}
// Safe for concurrent use by multiple goroutines.
func (mem *CListMempool) ReapMaxTxs(max int) types.Txs {
mem.updateMtx.RLock()
defer mem.updateMtx.RUnlock()
if max < 0 {
max = mem.txs.Len()
}
txs := make([]types.Tx, 0, tmmath.MinInt(mem.txs.Len(), max))
for e := mem.txs.Front(); e != nil && len(txs) <= max; e = e.Next() {
memTx := e.Value.(*mempoolTx)
txs = append(txs, memTx.tx)
}
return txs
}
// Lock() must be help by the caller during execution.
func (mem *CListMempool) Update(
height int64,
txs types.Txs,
deliverTxResponses []*abci.ResponseDeliverTx,
preCheck mempool.PreCheckFunc,
postCheck mempool.PostCheckFunc,
) error {
// Set height
mem.height = height
mem.notifiedTxsAvailable = false
if preCheck != nil {
mem.preCheck = preCheck
}
if postCheck != nil {
mem.postCheck = postCheck
}
for i, tx := range txs {
if deliverTxResponses[i].Code == abci.CodeTypeOK {
// Add valid committed tx to the cache (if missing).
_ = mem.cache.Push(tx)
} else if !mem.config.KeepInvalidTxsInCache {
// Allow invalid transactions to be resubmitted.
mem.cache.Remove(tx)
}
// Remove committed tx from the mempool.
//
// Note an evil proposer can drop valid txs!
// Mempool before:
// 100 -> 101 -> 102
// Block, proposed by an evil proposer:
// 101 -> 102
// Mempool after:
// 100
// https://github.com/tendermint/tendermint/issues/3322.
if e, ok := mem.txsMap.Load(mempool.TxKey(tx)); ok {
mem.removeTx(tx, e.(*clist.CElement), false)
}
}
// Either recheck non-committed txs to see if they became invalid
// or just notify there're some txs left.
if mem.Size() > 0 {
if mem.config.Recheck {
mem.logger.Debug("recheck txs", "numtxs", mem.Size(), "height", height)
mem.recheckTxs()
// At this point, mem.txs are being rechecked.
// mem.recheckCursor re-scans mem.txs and possibly removes some txs.
// Before mem.Reap(), we should wait for mem.recheckCursor to be nil.
} else {
mem.notifyTxsAvailable()
}
}
// Update metrics
mem.metrics.Size.Set(float64(mem.Size()))
return nil
}
func (mem *CListMempool) recheckTxs() {
if mem.Size() == 0 {
panic("recheckTxs is called, but the mempool is empty")
}
mem.recheckCursor = mem.txs.Front()
mem.recheckEnd = mem.txs.Back()
ctx := context.Background()
// Push txs to proxyAppConn
// NOTE: globalCb may be called concurrently.
for e := mem.txs.Front(); e != nil; e = e.Next() {
memTx := e.Value.(*mempoolTx)
_, err := mem.proxyAppConn.CheckTxAsync(ctx, abci.RequestCheckTx{
Tx: memTx.tx,
Type: abci.CheckTxType_Recheck,
})
if err != nil {
// No need in retrying since memTx will be rechecked after next block.
mem.logger.Error("Can't check tx", "err", err)
}
}
_, err := mem.proxyAppConn.FlushAsync(ctx)
if err != nil {
mem.logger.Error("Can't flush txs", "err", err)
}
}
//--------------------------------------------------------------------------------
// mempoolTx is a transaction that successfully ran
type mempoolTx struct {
height int64 // height that this tx had been validated in
gasWanted int64 // amount of gas this tx states it will require
tx types.Tx //
// ids of peers who've sent us this tx (as a map for quick lookups).
// senders: PeerID -> bool
senders sync.Map
}
// Height returns the height for this transaction
func (memTx *mempoolTx) Height() int64 {
return atomic.LoadInt64(&memTx.height)
}
mempool/v0/clist_mempool_test.go
+615
View File
@@ -0,0 +1,615 @@
package v0
import (
"context"
"crypto/rand"
"encoding/binary"
"fmt"
mrand "math/rand"
"os"
"testing"
"time"
"github.com/gogo/protobuf/proto"
gogotypes "github.com/gogo/protobuf/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/abci/example/counter"
"github.com/tendermint/tendermint/abci/example/kvstore"
abciserver "github.com/tendermint/tendermint/abci/server"
abci "github.com/tendermint/tendermint/abci/types"
cfg "github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/libs/log"
tmrand "github.com/tendermint/tendermint/libs/rand"
"github.com/tendermint/tendermint/libs/service"
"github.com/tendermint/tendermint/mempool"
"github.com/tendermint/tendermint/proxy"
"github.com/tendermint/tendermint/types"
)
// A cleanupFunc cleans up any config / test files created for a particular
// test.
type cleanupFunc func()
func newMempoolWithApp(cc proxy.ClientCreator) (*CListMempool, cleanupFunc) {
return newMempoolWithAppAndConfig(cc, cfg.ResetTestRoot("mempool_test"))
}
func newMempoolWithAppAndConfig(cc proxy.ClientCreator, config *cfg.Config) (*CListMempool, cleanupFunc) {
appConnMem, _ := cc.NewABCIClient()
appConnMem.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "mempool"))
err := appConnMem.Start()
if err != nil {
panic(err)
}
mp := NewCListMempool(config.Mempool, appConnMem, 0)
mp.SetLogger(log.TestingLogger())
return mp, func() { os.RemoveAll(config.RootDir) }
}
func ensureNoFire(t *testing.T, ch <-chan struct{}, timeoutMS int) {
timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond)
select {
case <-ch:
t.Fatal("Expected not to fire")
case <-timer.C:
}
}
func ensureFire(t *testing.T, ch <-chan struct{}, timeoutMS int) {
timer := time.NewTimer(time.Duration(timeoutMS) * time.Millisecond)
select {
case <-ch:
case <-timer.C:
t.Fatal("Expected to fire")
}
}
func checkTxs(t *testing.T, mp mempool.Mempool, count int, peerID uint16) types.Txs {
txs := make(types.Txs, count)
txInfo := mempool.TxInfo{SenderID: peerID}
for i := 0; i < count; i++ {
txBytes := make([]byte, 20)
txs[i] = txBytes
_, err := rand.Read(txBytes)
if err != nil {
t.Error(err)
}
if err := mp.CheckTx(txBytes, nil, txInfo); err != nil {
// Skip invalid txs.
// TestMempoolFilters will fail otherwise. It asserts a number of txs
// returned.
if mempool.IsPreCheckError(err) {
continue
}
t.Fatalf("CheckTx failed: %v while checking #%d tx", err, i)
}
}
return txs
}
func TestReapMaxBytesMaxGas(t *testing.T) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
// Ensure gas calculation behaves as expected
checkTxs(t, mp, 1, mempool.UnknownPeerID)
tx0 := mp.TxsFront().Value.(*mempoolTx)
// assert that kv store has gas wanted = 1.
require.Equal(t, app.CheckTx(abci.RequestCheckTx{Tx: tx0.tx}).GasWanted, int64(1), "KVStore had a gas value neq to 1")
require.Equal(t, tx0.gasWanted, int64(1), "transactions gas was set incorrectly")
// ensure each tx is 20 bytes long
require.Equal(t, len(tx0.tx), 20, "Tx is longer than 20 bytes")
mp.Flush()
// each table driven test creates numTxsToCreate txs with checkTx, and at the end clears all remaining txs.
// each tx has 20 bytes
tests := []struct {
numTxsToCreate int
maxBytes int64
maxGas int64
expectedNumTxs int
}{
{20, -1, -1, 20},
{20, -1, 0, 0},
{20, -1, 10, 10},
{20, -1, 30, 20},
{20, 0, -1, 0},
{20, 0, 10, 0},
{20, 10, 10, 0},
{20, 24, 10, 1},
{20, 240, 5, 5},
{20, 240, -1, 10},
{20, 240, 10, 10},
{20, 240, 15, 10},
{20, 20000, -1, 20},
{20, 20000, 5, 5},
{20, 20000, 30, 20},
}
for tcIndex, tt := range tests {
checkTxs(t, mp, tt.numTxsToCreate, mempool.UnknownPeerID)
got := mp.ReapMaxBytesMaxGas(tt.maxBytes, tt.maxGas)
assert.Equal(t, tt.expectedNumTxs, len(got), "Got %d txs, expected %d, tc #%d",
len(got), tt.expectedNumTxs, tcIndex)
mp.Flush()
}
}
func TestMempoolFilters(t *testing.T) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
emptyTxArr := []types.Tx{[]byte{}}
nopPreFilter := func(tx types.Tx) error { return nil }
nopPostFilter := func(tx types.Tx, res *abci.ResponseCheckTx) error { return nil }
// each table driven test creates numTxsToCreate txs with checkTx, and at the end clears all remaining txs.
// each tx has 20 bytes
tests := []struct {
numTxsToCreate int
preFilter mempool.PreCheckFunc
postFilter mempool.PostCheckFunc
expectedNumTxs int
}{
{10, nopPreFilter, nopPostFilter, 10},
{10, mempool.PreCheckMaxBytes(10), nopPostFilter, 0},
{10, mempool.PreCheckMaxBytes(22), nopPostFilter, 10},
{10, nopPreFilter, mempool.PostCheckMaxGas(-1), 10},
{10, nopPreFilter, mempool.PostCheckMaxGas(0), 0},
{10, nopPreFilter, mempool.PostCheckMaxGas(1), 10},
{10, nopPreFilter, mempool.PostCheckMaxGas(3000), 10},
{10, mempool.PreCheckMaxBytes(10), mempool.PostCheckMaxGas(20), 0},
{10, mempool.PreCheckMaxBytes(30), mempool.PostCheckMaxGas(20), 10},
{10, mempool.PreCheckMaxBytes(22), mempool.PostCheckMaxGas(1), 10},
{10, mempool.PreCheckMaxBytes(22), mempool.PostCheckMaxGas(0), 0},
}
for tcIndex, tt := range tests {
err := mp.Update(1, emptyTxArr, abciResponses(len(emptyTxArr), abci.CodeTypeOK), tt.preFilter, tt.postFilter)
require.NoError(t, err)
checkTxs(t, mp, tt.numTxsToCreate, mempool.UnknownPeerID)
require.Equal(t, tt.expectedNumTxs, mp.Size(), "mempool had the incorrect size, on test case %d", tcIndex)
mp.Flush()
}
}
func TestMempoolUpdate(t *testing.T) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
// 1. Adds valid txs to the cache
{
err := mp.Update(1, []types.Tx{[]byte{0x01}}, abciResponses(1, abci.CodeTypeOK), nil, nil)
require.NoError(t, err)
err = mp.CheckTx([]byte{0x01}, nil, mempool.TxInfo{})
require.NoError(t, err)
}
// 2. Removes valid txs from the mempool
{
err := mp.CheckTx([]byte{0x02}, nil, mempool.TxInfo{})
require.NoError(t, err)
err = mp.Update(1, []types.Tx{[]byte{0x02}}, abciResponses(1, abci.CodeTypeOK), nil, nil)
require.NoError(t, err)
assert.Zero(t, mp.Size())
}
// 3. Removes invalid transactions from the cache and the mempool (if present)
{
err := mp.CheckTx([]byte{0x03}, nil, mempool.TxInfo{})
require.NoError(t, err)
err = mp.Update(1, []types.Tx{[]byte{0x03}}, abciResponses(1, 1), nil, nil)
require.NoError(t, err)
assert.Zero(t, mp.Size())
err = mp.CheckTx([]byte{0x03}, nil, mempool.TxInfo{})
require.NoError(t, err)
}
}
func TestMempool_KeepInvalidTxsInCache(t *testing.T) {
app := counter.NewApplication(true)
cc := proxy.NewLocalClientCreator(app)
wcfg := cfg.DefaultConfig()
wcfg.Mempool.KeepInvalidTxsInCache = true
mp, cleanup := newMempoolWithAppAndConfig(cc, wcfg)
defer cleanup()
// 1. An invalid transaction must remain in the cache after Update
{
a := make([]byte, 8)
binary.BigEndian.PutUint64(a, 0)
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, 1)
err := mp.CheckTx(b, nil, mempool.TxInfo{})
require.NoError(t, err)
// simulate new block
_ = app.DeliverTx(abci.RequestDeliverTx{Tx: a})
_ = app.DeliverTx(abci.RequestDeliverTx{Tx: b})
err = mp.Update(1, []types.Tx{a, b},
[]*abci.ResponseDeliverTx{{Code: abci.CodeTypeOK}, {Code: 2}}, nil, nil)
require.NoError(t, err)
// a must be added to the cache
err = mp.CheckTx(a, nil, mempool.TxInfo{})
require.NoError(t, err)
// b must remain in the cache
err = mp.CheckTx(b, nil, mempool.TxInfo{})
require.NoError(t, err)
}
// 2. An invalid transaction must remain in the cache
{
a := make([]byte, 8)
binary.BigEndian.PutUint64(a, 0)
// remove a from the cache to test (2)
mp.cache.Remove(a)
err := mp.CheckTx(a, nil, mempool.TxInfo{})
require.NoError(t, err)
}
}
func TestTxsAvailable(t *testing.T) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
mp.EnableTxsAvailable()
timeoutMS := 500
// with no txs, it shouldnt fire
ensureNoFire(t, mp.TxsAvailable(), timeoutMS)
// send a bunch of txs, it should only fire once
txs := checkTxs(t, mp, 100, mempool.UnknownPeerID)
ensureFire(t, mp.TxsAvailable(), timeoutMS)
ensureNoFire(t, mp.TxsAvailable(), timeoutMS)
// call update with half the txs.
// it should fire once now for the new height
// since there are still txs left
committedTxs, txs := txs[:50], txs[50:]
if err := mp.Update(1, committedTxs, abciResponses(len(committedTxs), abci.CodeTypeOK), nil, nil); err != nil {
t.Error(err)
}
ensureFire(t, mp.TxsAvailable(), timeoutMS)
ensureNoFire(t, mp.TxsAvailable(), timeoutMS)
// send a bunch more txs. we already fired for this height so it shouldnt fire again
moreTxs := checkTxs(t, mp, 50, mempool.UnknownPeerID)
ensureNoFire(t, mp.TxsAvailable(), timeoutMS)
// now call update with all the txs. it should not fire as there are no txs left
committedTxs = append(txs, moreTxs...) //nolint: gocritic
if err := mp.Update(2, committedTxs, abciResponses(len(committedTxs), abci.CodeTypeOK), nil, nil); err != nil {
t.Error(err)
}
ensureNoFire(t, mp.TxsAvailable(), timeoutMS)
// send a bunch more txs, it should only fire once
checkTxs(t, mp, 100, mempool.UnknownPeerID)
ensureFire(t, mp.TxsAvailable(), timeoutMS)
ensureNoFire(t, mp.TxsAvailable(), timeoutMS)
}
func TestSerialReap(t *testing.T) {
app := counter.NewApplication(true)
cc := proxy.NewLocalClientCreator(app)
mp, cleanup := newMempoolWithApp(cc)
defer cleanup()
appConnCon, _ := cc.NewABCIClient()
appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
err := appConnCon.Start()
require.Nil(t, err)
cacheMap := make(map[string]struct{})
deliverTxsRange := func(start, end int) {
// Deliver some txs.
for i := start; i < end; i++ {
// This will succeed
txBytes := make([]byte, 8)
binary.BigEndian.PutUint64(txBytes, uint64(i))
err := mp.CheckTx(txBytes, nil, mempool.TxInfo{})
_, cached := cacheMap[string(txBytes)]
if cached {
require.NotNil(t, err, "expected error for cached tx")
} else {
require.Nil(t, err, "expected no err for uncached tx")
}
cacheMap[string(txBytes)] = struct{}{}
// Duplicates are cached and should return error
err = mp.CheckTx(txBytes, nil, mempool.TxInfo{})
require.NotNil(t, err, "Expected error after CheckTx on duplicated tx")
}
}
reapCheck := func(exp int) {
txs := mp.ReapMaxBytesMaxGas(-1, -1)
require.Equal(t, len(txs), exp, fmt.Sprintf("Expected to reap %v txs but got %v", exp, len(txs)))
}
updateRange := func(start, end int) {
txs := make([]types.Tx, 0)
for i := start; i < end; i++ {
txBytes := make([]byte, 8)
binary.BigEndian.PutUint64(txBytes, uint64(i))
txs = append(txs, txBytes)
}
if err := mp.Update(0, txs, abciResponses(len(txs), abci.CodeTypeOK), nil, nil); err != nil {
t.Error(err)
}
}
commitRange := func(start, end int) {
ctx := context.Background()
// Deliver some txs.
for i := start; i < end; i++ {
txBytes := make([]byte, 8)
binary.BigEndian.PutUint64(txBytes, uint64(i))
res, err := appConnCon.DeliverTxSync(ctx, abci.RequestDeliverTx{Tx: txBytes})
if err != nil {
t.Errorf("client error committing tx: %v", err)
}
if res.IsErr() {
t.Errorf("error committing tx. Code:%v result:%X log:%v",
res.Code, res.Data, res.Log)
}
}
res, err := appConnCon.CommitSync(ctx)
if err != nil {
t.Errorf("client error committing: %v", err)
}
if len(res.Data) != 8 {
t.Errorf("error committing. Hash:%X", res.Data)
}
}
//----------------------------------------
// Deliver some txs.
deliverTxsRange(0, 100)
// Reap the txs.
reapCheck(100)
// Reap again. We should get the same amount
reapCheck(100)
// Deliver 0 to 999, we should reap 900 new txs
// because 100 were already counted.
deliverTxsRange(0, 1000)
// Reap the txs.
reapCheck(1000)
// Reap again. We should get the same amount
reapCheck(1000)
// Commit from the conensus AppConn
commitRange(0, 500)
updateRange(0, 500)
// We should have 500 left.
reapCheck(500)
// Deliver 100 invalid txs and 100 valid txs
deliverTxsRange(900, 1100)
// We should have 600 now.
reapCheck(600)
}
func TestMempool_CheckTxChecksTxSize(t *testing.T) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
mempl, cleanup := newMempoolWithApp(cc)
defer cleanup()
maxTxSize := mempl.config.MaxTxBytes
testCases := []struct {
len int
err bool
}{
// check small txs. no error
0: {10, false},
1: {1000, false},
2: {1000000, false},
// check around maxTxSize
3: {maxTxSize - 1, false},
4: {maxTxSize, false},
5: {maxTxSize + 1, true},
}
for i, testCase := range testCases {
caseString := fmt.Sprintf("case %d, len %d", i, testCase.len)
tx := tmrand.Bytes(testCase.len)
err := mempl.CheckTx(tx, nil, mempool.TxInfo{})
bv := gogotypes.BytesValue{Value: tx}
bz, err2 := bv.Marshal()
require.NoError(t, err2)
require.Equal(t, len(bz), proto.Size(&bv), caseString)
if !testCase.err {
require.NoError(t, err, caseString)
} else {
require.Equal(t, err, mempool.ErrTxTooLarge{
Max: maxTxSize,
Actual: testCase.len,
}, caseString)
}
}
}
func TestMempoolTxsBytes(t *testing.T) {
app := kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(app)
config := cfg.ResetTestRoot("mempool_test")
config.Mempool.MaxTxsBytes = 10
mp, cleanup := newMempoolWithAppAndConfig(cc, config)
defer cleanup()
// 1. zero by default
assert.EqualValues(t, 0, mp.SizeBytes())
// 2. len(tx) after CheckTx
err := mp.CheckTx([]byte{0x01}, nil, mempool.TxInfo{})
require.NoError(t, err)
assert.EqualValues(t, 1, mp.SizeBytes())
// 3. zero again after tx is removed by Update
err = mp.Update(1, []types.Tx{[]byte{0x01}}, abciResponses(1, abci.CodeTypeOK), nil, nil)
require.NoError(t, err)
assert.EqualValues(t, 0, mp.SizeBytes())
// 4. zero after Flush
err = mp.CheckTx([]byte{0x02, 0x03}, nil, mempool.TxInfo{})
require.NoError(t, err)
assert.EqualValues(t, 2, mp.SizeBytes())
mp.Flush()
assert.EqualValues(t, 0, mp.SizeBytes())
// 5. ErrMempoolIsFull is returned when/if MaxTxsBytes limit is reached.
err = mp.CheckTx([]byte{0x04, 0x04, 0x04, 0x04, 0x04, 0x04, 0x04, 0x04, 0x04, 0x04}, nil, mempool.TxInfo{})
require.NoError(t, err)
err = mp.CheckTx([]byte{0x05}, nil, mempool.TxInfo{})
if assert.Error(t, err) {
assert.IsType(t, mempool.ErrMempoolIsFull{}, err)
}
// 6. zero after tx is rechecked and removed due to not being valid anymore
app2 := counter.NewApplication(true)
cc = proxy.NewLocalClientCreator(app2)
mp, cleanup = newMempoolWithApp(cc)
defer cleanup()
txBytes := make([]byte, 8)
binary.BigEndian.PutUint64(txBytes, uint64(0))
err = mp.CheckTx(txBytes, nil, mempool.TxInfo{})
require.NoError(t, err)
assert.EqualValues(t, 8, mp.SizeBytes())
appConnCon, _ := cc.NewABCIClient()
appConnCon.SetLogger(log.TestingLogger().With("module", "abci-client", "connection", "consensus"))
err = appConnCon.Start()
require.Nil(t, err)
t.Cleanup(func() {
if err := appConnCon.Stop(); err != nil {
t.Error(err)
}
})
ctx := context.Background()
res, err := appConnCon.DeliverTxSync(ctx, abci.RequestDeliverTx{Tx: txBytes})
require.NoError(t, err)
require.EqualValues(t, 0, res.Code)
res2, err := appConnCon.CommitSync(ctx)
require.NoError(t, err)
require.NotEmpty(t, res2.Data)
// Pretend like we committed nothing so txBytes gets rechecked and removed.
err = mp.Update(1, []types.Tx{}, abciResponses(0, abci.CodeTypeOK), nil, nil)
require.NoError(t, err)
assert.EqualValues(t, 0, mp.SizeBytes())
// 7. Test RemoveTxByKey function
err = mp.CheckTx([]byte{0x06}, nil, mempool.TxInfo{})
require.NoError(t, err)
assert.EqualValues(t, 1, mp.SizeBytes())
mp.RemoveTxByKey(mempool.TxKey([]byte{0x07}), true)
assert.EqualValues(t, 1, mp.SizeBytes())
mp.RemoveTxByKey(mempool.TxKey([]byte{0x06}), true)
assert.EqualValues(t, 0, mp.SizeBytes())
}
// This will non-deterministically catch some concurrency failures like
// https://github.com/tendermint/tendermint/issues/3509
// TODO: all of the tests should probably also run using the remote proxy app
// since otherwise we're not actually testing the concurrency of the mempool here!
func TestMempoolRemoteAppConcurrency(t *testing.T) {
sockPath := fmt.Sprintf("unix:///tmp/echo_%v.sock", tmrand.Str(6))
app := kvstore.NewApplication()
cc, server := newRemoteApp(t, sockPath, app)
t.Cleanup(func() {
if err := server.Stop(); err != nil {
t.Error(err)
}
})
config := cfg.ResetTestRoot("mempool_test")
mp, cleanup := newMempoolWithAppAndConfig(cc, config)
defer cleanup()
// generate small number of txs
nTxs := 10
txLen := 200
txs := make([]types.Tx, nTxs)
for i := 0; i < nTxs; i++ {
txs[i] = tmrand.Bytes(txLen)
}
// simulate a group of peers sending them over and over
N := config.Mempool.Size
maxPeers := 5
for i := 0; i < N; i++ {
peerID := mrand.Intn(maxPeers)
txNum := mrand.Intn(nTxs)
tx := txs[txNum]
// this will err with ErrTxInCache many times ...
mp.CheckTx(tx, nil, mempool.TxInfo{SenderID: uint16(peerID)}) //nolint: errcheck // will error
}
err := mp.FlushAppConn()
require.NoError(t, err)
}
// caller must close server
func newRemoteApp(
t *testing.T,
addr string,
app abci.Application,
) (
clientCreator proxy.ClientCreator,
server service.Service,
) {
clientCreator = proxy.NewRemoteClientCreator(addr, "socket", true)
// Start server
server = abciserver.NewSocketServer(addr, app)
server.SetLogger(log.TestingLogger().With("module", "abci-server"))
if err := server.Start(); err != nil {
t.Fatalf("Error starting socket server: %v", err.Error())
}
return clientCreator, server
}
func abciResponses(n int, code uint32) []*abci.ResponseDeliverTx {
responses := make([]*abci.ResponseDeliverTx, 0, n)
for i := 0; i < n; i++ {
responses = append(responses, &abci.ResponseDeliverTx{Code: code})
}
return responses
}
mempool/v0/doc.go
+23
View File
@@ -0,0 +1,23 @@
// The mempool pushes new txs onto the proxyAppConn.
// It gets a stream of (req, res) tuples from the proxy.
// The mempool stores good txs in a concurrent linked-list.
// Multiple concurrent go-routines can traverse this linked-list
// safely by calling .NextWait() on each element.
// So we have several go-routines:
// 1. Consensus calling Update() and ReapMaxBytesMaxGas() synchronously
// 2. Many mempool reactor's peer routines calling CheckTx()
// 3. Many mempool reactor's peer routines traversing the txs linked list
// To manage these goroutines, there are three methods of locking.
// 1. Mutations to the linked-list is protected by an internal mtx (CList is goroutine-safe)
// 2. Mutations to the linked-list elements are atomic
// 3. CheckTx() and/or ReapMaxBytesMaxGas() calls can be paused upon Update(), protected by .updateMtx
// Garbage collection of old elements from mempool.txs is handlde via the
// DetachPrev() call, which makes old elements not reachable by peer
// broadcastTxRoutine().
// TODO: Better handle abci client errors. (make it automatically handle connection errors)
package v0
mempool/v0/reactor.go
+391
View File
@@ -0,0 +1,391 @@
package v0
import (
"errors"
"fmt"
"sync"
"time"
cfg "github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/internal/libs/clist"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/libs/service"
"github.com/tendermint/tendermint/mempool"
"github.com/tendermint/tendermint/p2p"
protomem "github.com/tendermint/tendermint/proto/tendermint/mempool"
"github.com/tendermint/tendermint/types"
)
var (
_ service.Service = (*Reactor)(nil)
_ p2p.Wrapper = (*protomem.Message)(nil)
)
// PeerManager defines the interface contract required for getting necessary
// peer information. This should eventually be replaced with a message-oriented
// approach utilizing the p2p stack.
type PeerManager interface {
GetHeight(p2p.NodeID) int64
}
// Reactor implements a service that contains mempool of txs that are broadcasted
// amongst peers. It maintains a map from peer ID to counter, to prevent gossiping
// txs to the peers you received it from.
type Reactor struct {
service.BaseService
config *cfg.MempoolConfig
mempool *CListMempool
ids *mempool.MempoolIDs
// XXX: Currently, this is the only way to get information about a peer. Ideally,
// we rely on message-oriented communication to get necessary peer data.
// ref: https://github.com/tendermint/tendermint/issues/5670
peerMgr PeerManager
mempoolCh *p2p.Channel
peerUpdates *p2p.PeerUpdates
closeCh chan struct{}
// peerWG is used to coordinate graceful termination of all peer broadcasting
// goroutines.
peerWG sync.WaitGroup
mtx tmsync.Mutex
peerRoutines map[p2p.NodeID]*tmsync.Closer
}
// NewReactor returns a reference to a new reactor.
func NewReactor(
logger log.Logger,
config *cfg.MempoolConfig,
peerMgr PeerManager,
mp *CListMempool,
mempoolCh *p2p.Channel,
peerUpdates *p2p.PeerUpdates,
) *Reactor {
r := &Reactor{
config: config,
peerMgr: peerMgr,
mempool: mp,
ids: mempool.NewMempoolIDs(),
mempoolCh: mempoolCh,
peerUpdates: peerUpdates,
closeCh: make(chan struct{}),
peerRoutines: make(map[p2p.NodeID]*tmsync.Closer),
}
r.BaseService = *service.NewBaseService(logger, "Mempool", r)
return r
}
// GetChannelShims returns a map of ChannelDescriptorShim objects, where each
// object wraps a reference to a legacy p2p ChannelDescriptor and the corresponding
// p2p proto.Message the new p2p Channel is responsible for handling.
//
//
// TODO: Remove once p2p refactor is complete.
// ref: https://github.com/tendermint/tendermint/issues/5670
func GetChannelShims(config *cfg.MempoolConfig) map[p2p.ChannelID]*p2p.ChannelDescriptorShim {
largestTx := make([]byte, config.MaxTxBytes)
batchMsg := protomem.Message{
Sum: &protomem.Message_Txs{
Txs: &protomem.Txs{Txs: [][]byte{largestTx}},
},
}
return map[p2p.ChannelID]*p2p.ChannelDescriptorShim{
mempool.MempoolChannel: {
MsgType: new(protomem.Message),
Descriptor: &p2p.ChannelDescriptor{
ID: byte(mempool.MempoolChannel),
Priority: 5,
RecvMessageCapacity: batchMsg.Size(),
MaxSendBytes: 5000,
},
},
}
}
// OnStart starts separate go routines for each p2p Channel and listens for
// envelopes on each. In addition, it also listens for peer updates and handles
// messages on that p2p channel accordingly. The caller must be sure to execute
// OnStop to ensure the outbound p2p Channels are closed.
func (r *Reactor) OnStart() error {
if !r.config.Broadcast {
r.Logger.Info("tx broadcasting is disabled")
}
go r.processMempoolCh()
go r.processPeerUpdates()
return nil
}
// OnStop stops the reactor by signaling to all spawned goroutines to exit and
// blocking until they all exit.
func (r *Reactor) OnStop() {
r.mtx.Lock()
for _, c := range r.peerRoutines {
c.Close()
}
r.mtx.Unlock()
// wait for all spawned peer tx broadcasting goroutines to gracefully exit
r.peerWG.Wait()
// Close closeCh to signal to all spawned goroutines to gracefully exit. All
// p2p Channels should execute Close().
close(r.closeCh)
// Wait for all p2p Channels to be closed before returning. This ensures we
// can easily reason about synchronization of all p2p Channels and ensure no
// panics will occur.
<-r.mempoolCh.Done()
<-r.peerUpdates.Done()
}
// handleMempoolMessage handles envelopes sent from peers on the MempoolChannel.
// For every tx in the message, we execute CheckTx. It returns an error if an
// empty set of txs are sent in an envelope or if we receive an unexpected
// message type.
func (r *Reactor) handleMempoolMessage(envelope p2p.Envelope) error {
logger := r.Logger.With("peer", envelope.From)
switch msg := envelope.Message.(type) {
case *protomem.Txs:
protoTxs := msg.GetTxs()
if len(protoTxs) == 0 {
return errors.New("empty txs received from peer")
}
txInfo := mempool.TxInfo{SenderID: r.ids.GetForPeer(envelope.From)}
if len(envelope.From) != 0 {
txInfo.SenderNodeID = envelope.From
}
for _, tx := range protoTxs {
if err := r.mempool.CheckTx(types.Tx(tx), nil, txInfo); err != nil {
logger.Error("checktx failed for tx", "tx", fmt.Sprintf("%X", mempool.TxHashFromBytes(tx)), "err", err)
}
}
default:
return fmt.Errorf("received unknown message: %T", msg)
}
return nil
}
// handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
// It will handle errors and any possible panics gracefully. A caller can handle
// any error returned by sending a PeerError on the respective channel.
func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err error) {
defer func() {
if e := recover(); e != nil {
err = fmt.Errorf("panic in processing message: %v", e)
}
}()
r.Logger.Debug("received message", "peer", envelope.From)
switch chID {
case mempool.MempoolChannel:
err = r.handleMempoolMessage(envelope)
default:
err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
}
return err
}
// processMempoolCh implements a blocking event loop where we listen for p2p
// Envelope messages from the mempoolCh.
func (r *Reactor) processMempoolCh() {
defer r.mempoolCh.Close()
for {
select {
case envelope := <-r.mempoolCh.In:
if err := r.handleMessage(r.mempoolCh.ID, envelope); err != nil {
r.Logger.Error("failed to process message", "ch_id", r.mempoolCh.ID, "envelope", envelope, "err", err)
r.mempoolCh.Error <- p2p.PeerError{
NodeID: envelope.From,
Err: err,
}
}
case <-r.closeCh:
r.Logger.Debug("stopped listening on mempool channel; closing...")
return
}
}
}
// processPeerUpdate processes a PeerUpdate. For added peers, PeerStatusUp, we
// check if the reactor is running and if we've already started a tx broadcasting
// goroutine or not. If not, we start one for the newly added peer. For down or
// removed peers, we remove the peer from the mempool peer ID set and signal to
// stop the tx broadcasting goroutine.
func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
r.Logger.Debug("received peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
r.mtx.Lock()
defer r.mtx.Unlock()
switch peerUpdate.Status {
case p2p.PeerStatusUp:
// Do not allow starting new tx broadcast loops after reactor shutdown
// has been initiated. This can happen after we've manually closed all
// peer broadcast loops and closed r.closeCh, but the router still sends
// in-flight peer updates.
if !r.IsRunning() {
return
}
if r.config.Broadcast {
// Check if we've already started a goroutine for this peer, if not we create
// a new done channel so we can explicitly close the goroutine if the peer
// is later removed, we increment the waitgroup so the reactor can stop
// safely, and finally start the goroutine to broadcast txs to that peer.
_, ok := r.peerRoutines[peerUpdate.NodeID]
if !ok {
closer := tmsync.NewCloser()
r.peerRoutines[peerUpdate.NodeID] = closer
r.peerWG.Add(1)
r.ids.ReserveForPeer(peerUpdate.NodeID)
// start a broadcast routine ensuring all txs are forwarded to the peer
go r.broadcastTxRoutine(peerUpdate.NodeID, closer)
}
}
case p2p.PeerStatusDown:
r.ids.Reclaim(peerUpdate.NodeID)
// Check if we've started a tx broadcasting goroutine for this peer.
// If we have, we signal to terminate the goroutine via the channel's closure.
// This will internally decrement the peer waitgroup and remove the peer
// from the map of peer tx broadcasting goroutines.
closer, ok := r.peerRoutines[peerUpdate.NodeID]
if ok {
closer.Close()
}
}
}
// processPeerUpdates initiates a blocking process where we listen for and handle
// PeerUpdate messages. When the reactor is stopped, we will catch the signal and
// close the p2p PeerUpdatesCh gracefully.
func (r *Reactor) processPeerUpdates() {
defer r.peerUpdates.Close()
for {
select {
case peerUpdate := <-r.peerUpdates.Updates():
r.processPeerUpdate(peerUpdate)
case <-r.closeCh:
r.Logger.Debug("stopped listening on peer updates channel; closing...")
return
}
}
}
func (r *Reactor) broadcastTxRoutine(peerID p2p.NodeID, closer *tmsync.Closer) {
peerMempoolID := r.ids.GetForPeer(peerID)
var next *clist.CElement
// remove the peer ID from the map of routines and mark the waitgroup as done
defer func() {
r.mtx.Lock()
delete(r.peerRoutines, peerID)
r.mtx.Unlock()
r.peerWG.Done()
if e := recover(); e != nil {
r.Logger.Error("recovering from broadcasting mempool loop", "err", e)
}
}()
for {
if !r.IsRunning() {
return
}
// This happens because the CElement we were looking at got garbage
// collected (removed). That is, .NextWait() returned nil. Go ahead and
// start from the beginning.
if next == nil {
select {
case <-r.mempool.TxsWaitChan(): // wait until a tx is available
if next = r.mempool.TxsFront(); next == nil {
continue
}
case <-closer.Done():
// The peer is marked for removal via a PeerUpdate as the doneCh was
// explicitly closed to signal we should exit.
return
case <-r.closeCh:
// The reactor has signaled that we are stopped and thus we should
// implicitly exit this peer's goroutine.
return
}
}
memTx := next.Value.(*mempoolTx)
if r.peerMgr != nil {
height := r.peerMgr.GetHeight(peerID)
if height > 0 && height < memTx.Height()-1 {
// allow for a lag of one block
time.Sleep(mempool.PeerCatchupSleepIntervalMS * time.Millisecond)
continue
}
}
// NOTE: Transaction batching was disabled due to:
// https://github.com/tendermint/tendermint/issues/5796
if _, ok := memTx.senders.Load(peerMempoolID); !ok {
// Send the mempool tx to the corresponding peer. Note, the peer may be
// behind and thus would not be able to process the mempool tx correctly.
r.mempoolCh.Out <- p2p.Envelope{
To: peerID,
Message: &protomem.Txs{
Txs: [][]byte{memTx.tx},
},
}
r.Logger.Debug(
"gossiped tx to peer",
"tx", fmt.Sprintf("%X", mempool.TxHashFromBytes(memTx.tx)),
"peer", peerID,
)
}
select {
case <-next.NextWaitChan():
// see the start of the for loop for nil check
next = next.Next()
case <-closer.Done():
// The peer is marked for removal via a PeerUpdate as the doneCh was
// explicitly closed to signal we should exit.
return
case <-r.closeCh:
// The reactor has signaled that we are stopped and thus we should
// implicitly exit this peer's goroutine.
return
}
}
}
mempool/v0/reactor_test.go
+383
View File
@@ -0,0 +1,383 @@
package v0
import (
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/abci/example/kvstore"
abci "github.com/tendermint/tendermint/abci/types"
cfg "github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/libs/log"
tmrand "github.com/tendermint/tendermint/libs/rand"
"github.com/tendermint/tendermint/mempool"
"github.com/tendermint/tendermint/p2p"
"github.com/tendermint/tendermint/p2p/p2ptest"
protomem "github.com/tendermint/tendermint/proto/tendermint/mempool"
"github.com/tendermint/tendermint/proxy"
"github.com/tendermint/tendermint/types"
)
type reactorTestSuite struct {
network *p2ptest.Network
logger log.Logger
reactors map[p2p.NodeID]*Reactor
mempoolChnnels map[p2p.NodeID]*p2p.Channel
mempools map[p2p.NodeID]*CListMempool
kvstores map[p2p.NodeID]*kvstore.Application
peerChans map[p2p.NodeID]chan p2p.PeerUpdate
peerUpdates map[p2p.NodeID]*p2p.PeerUpdates
nodes []p2p.NodeID
}
func setup(t *testing.T, cfg *cfg.MempoolConfig, numNodes int, chBuf uint) *reactorTestSuite {
t.Helper()
rts := &reactorTestSuite{
logger: log.TestingLogger().With("testCase", t.Name()),
network: p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numNodes}),
reactors: make(map[p2p.NodeID]*Reactor, numNodes),
mempoolChnnels: make(map[p2p.NodeID]*p2p.Channel, numNodes),
mempools: make(map[p2p.NodeID]*CListMempool, numNodes),
kvstores: make(map[p2p.NodeID]*kvstore.Application, numNodes),
peerChans: make(map[p2p.NodeID]chan p2p.PeerUpdate, numNodes),
peerUpdates: make(map[p2p.NodeID]*p2p.PeerUpdates, numNodes),
}
chDesc := p2p.ChannelDescriptor{ID: byte(mempool.MempoolChannel)}
rts.mempoolChnnels = rts.network.MakeChannelsNoCleanup(t, chDesc, new(protomem.Message), int(chBuf))
for nodeID := range rts.network.Nodes {
rts.kvstores[nodeID] = kvstore.NewApplication()
cc := proxy.NewLocalClientCreator(rts.kvstores[nodeID])
mempool, memCleanup := newMempoolWithApp(cc)
t.Cleanup(memCleanup)
mempool.SetLogger(rts.logger)
rts.mempools[nodeID] = mempool
rts.peerChans[nodeID] = make(chan p2p.PeerUpdate)
rts.peerUpdates[nodeID] = p2p.NewPeerUpdates(rts.peerChans[nodeID], 1)
rts.network.Nodes[nodeID].PeerManager.Register(rts.peerUpdates[nodeID])
rts.reactors[nodeID] = NewReactor(
rts.logger.With("nodeID", nodeID),
cfg,
rts.network.Nodes[nodeID].PeerManager,
mempool,
rts.mempoolChnnels[nodeID],
rts.peerUpdates[nodeID],
)
rts.nodes = append(rts.nodes, nodeID)
require.NoError(t, rts.reactors[nodeID].Start())
require.True(t, rts.reactors[nodeID].IsRunning())
}
require.Len(t, rts.reactors, numNodes)
t.Cleanup(func() {
for nodeID := range rts.reactors {
if rts.reactors[nodeID].IsRunning() {
require.NoError(t, rts.reactors[nodeID].Stop())
require.False(t, rts.reactors[nodeID].IsRunning())
}
}
})
return rts
}
func (rts *reactorTestSuite) start(t *testing.T) {
t.Helper()
rts.network.Start(t)
require.Len(t,
rts.network.RandomNode().PeerManager.Peers(),
len(rts.nodes)-1,
"network does not have expected number of nodes")
}
func (rts *reactorTestSuite) assertMempoolChannelsDrained(t *testing.T) {
t.Helper()
for id, r := range rts.reactors {
require.NoError(t, r.Stop(), "stopping reactor %s", id)
r.Wait()
require.False(t, r.IsRunning(), "reactor %s did not stop", id)
}
for _, mch := range rts.mempoolChnnels {
require.Empty(t, mch.Out, "checking channel %q (len=%d)", mch.ID, len(mch.Out))
}
}
func (rts *reactorTestSuite) waitForTxns(t *testing.T, txs types.Txs, ids ...p2p.NodeID) {
t.Helper()
fn := func(pool *CListMempool) {
for pool.Size() < len(txs) {
time.Sleep(50 * time.Millisecond)
}
reapedTxs := pool.ReapMaxTxs(len(txs))
require.Equal(t, len(txs), len(reapedTxs))
for i, tx := range txs {
require.Equalf(t,
tx,
reapedTxs[i],
"txs at index %d in reactor mempool mismatch; got: %v, expected: %v", i, tx, reapedTxs[i],
)
}
}
if len(ids) == 1 {
fn(rts.reactors[ids[0]].mempool)
return
}
wg := &sync.WaitGroup{}
for id := range rts.mempools {
if len(ids) > 0 && !p2ptest.NodeInSlice(id, ids) {
continue
}
wg.Add(1)
func(nid p2p.NodeID) { defer wg.Done(); fn(rts.reactors[nid].mempool) }(id)
}
wg.Wait()
}
func TestReactorBroadcastTxs(t *testing.T) {
numTxs := 1000
numNodes := 10
config := cfg.TestConfig()
rts := setup(t, config.Mempool, numNodes, 0)
primary := rts.nodes[0]
secondaries := rts.nodes[1:]
txs := checkTxs(t, rts.reactors[primary].mempool, numTxs, mempool.UnknownPeerID)
// run the router
rts.start(t)
// Wait till all secondary suites (reactor) received all mempool txs from the
// primary suite (node).
rts.waitForTxns(t, txs, secondaries...)
for _, pool := range rts.mempools {
require.Equal(t, len(txs), pool.Size())
}
rts.assertMempoolChannelsDrained(t)
}
// regression test for https://github.com/tendermint/tendermint/issues/5408
func TestReactorConcurrency(t *testing.T) {
numTxs := 5
numNodes := 2
config := cfg.TestConfig()
rts := setup(t, config.Mempool, numNodes, 0)
primary := rts.nodes[0]
secondary := rts.nodes[1]
rts.start(t)
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
wg.Add(2)
// 1. submit a bunch of txs
// 2. update the whole mempool
txs := checkTxs(t, rts.reactors[primary].mempool, numTxs, mempool.UnknownPeerID)
go func() {
defer wg.Done()
mempool := rts.mempools[primary]
mempool.Lock()
defer mempool.Unlock()
deliverTxResponses := make([]*abci.ResponseDeliverTx, len(txs))
for i := range txs {
deliverTxResponses[i] = &abci.ResponseDeliverTx{Code: 0}
}
require.NoError(t, mempool.Update(1, txs, deliverTxResponses, nil, nil))
}()
// 1. submit a bunch of txs
// 2. update none
_ = checkTxs(t, rts.reactors[secondary].mempool, numTxs, mempool.UnknownPeerID)
go func() {
defer wg.Done()
mempool := rts.mempools[secondary]
mempool.Lock()
defer mempool.Unlock()
err := mempool.Update(1, []types.Tx{}, make([]*abci.ResponseDeliverTx, 0), nil, nil)
require.NoError(t, err)
}()
// flush the mempool
rts.mempools[secondary].Flush()
}
wg.Wait()
}
func TestReactorNoBroadcastToSender(t *testing.T) {
numTxs := 1000
numNodes := 2
config := cfg.TestConfig()
rts := setup(t, config.Mempool, numNodes, uint(numTxs))
primary := rts.nodes[0]
secondary := rts.nodes[1]
peerID := uint16(1)
_ = checkTxs(t, rts.mempools[primary], numTxs, peerID)
rts.start(t)
time.Sleep(100 * time.Millisecond)
require.Eventually(t, func() bool {
return rts.mempools[secondary].Size() == 0
}, time.Minute, 100*time.Millisecond)
rts.assertMempoolChannelsDrained(t)
}
func TestReactor_MaxTxBytes(t *testing.T) {
numNodes := 2
config := cfg.TestConfig()
rts := setup(t, config.Mempool, numNodes, 0)
primary := rts.nodes[0]
secondary := rts.nodes[1]
// Broadcast a tx, which has the max size and ensure it's received by the
// second reactor.
tx1 := tmrand.Bytes(config.Mempool.MaxTxBytes)
err := rts.reactors[primary].mempool.CheckTx(tx1, nil, mempool.TxInfo{SenderID: mempool.UnknownPeerID})
require.NoError(t, err)
rts.start(t)
// Wait till all secondary suites (reactor) received all mempool txs from the
// primary suite (node).
rts.waitForTxns(t, []types.Tx{tx1}, secondary)
rts.reactors[primary].mempool.Flush()
rts.reactors[secondary].mempool.Flush()
// broadcast a tx, which is beyond the max size and ensure it's not sent
tx2 := tmrand.Bytes(config.Mempool.MaxTxBytes + 1)
err = rts.mempools[primary].CheckTx(tx2, nil, mempool.TxInfo{SenderID: mempool.UnknownPeerID})
require.Error(t, err)
rts.assertMempoolChannelsDrained(t)
}
func TestDontExhaustMaxActiveIDs(t *testing.T) {
config := cfg.TestConfig()
// we're creating a single node network, but not starting the
// network.
rts := setup(t, config.Mempool, 1, mempool.MaxActiveIDs+1)
nodeID := rts.nodes[0]
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
// ensure the reactor does not panic (i.e. exhaust active IDs)
for i := 0; i < mempool.MaxActiveIDs+1; i++ {
rts.peerChans[nodeID] <- p2p.PeerUpdate{
Status: p2p.PeerStatusUp,
NodeID: peerID,
}
rts.mempoolChnnels[nodeID].Out <- p2p.Envelope{
To: peerID,
Message: &protomem.Txs{
Txs: [][]byte{},
},
}
}
require.Eventually(
t,
func() bool {
for _, mch := range rts.mempoolChnnels {
if len(mch.Out) > 0 {
return false
}
}
return true
},
time.Minute,
10*time.Millisecond,
)
rts.assertMempoolChannelsDrained(t)
}
func TestMempoolIDsPanicsIfNodeRequestsOvermaxActiveIDs(t *testing.T) {
if testing.Short() {
t.Skip("skipping test in short mode")
}
// 0 is already reserved for UnknownPeerID
ids := mempool.NewMempoolIDs()
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
for i := 0; i < mempool.MaxActiveIDs-1; i++ {
ids.ReserveForPeer(peerID)
}
require.Panics(t, func() {
ids.ReserveForPeer(peerID)
})
}
func TestBroadcastTxForPeerStopsWhenPeerStops(t *testing.T) {
if testing.Short() {
t.Skip("skipping test in short mode")
}
config := cfg.TestConfig()
rts := setup(t, config.Mempool, 2, 0)
primary := rts.nodes[0]
secondary := rts.nodes[1]
rts.start(t)
// disconnect peer
rts.peerChans[primary] <- p2p.PeerUpdate{
Status: p2p.PeerStatusDown,
NodeID: secondary,
}
}