func (mem *CListMempool) removeTx(tx types.Tx, elem *clist.CElement, removeFromCache bool) {
	mem.txs.Remove(elem)
	elem.DetachPrev()
	mem.txsMap.Delete(tx.Key())
	atomic.AddInt64(&mem.txsBytes, int64(-len(tx)))
	if removeFromCache {
		mem.cache.Remove(tx)
	}
}
Called from:
- Update (lock held) if tx was committed
- resCbRecheck (lock not held) if tx was invalidated
package v0
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	abci "github.com/tendermint/tendermint/abci/types"
	"github.com/tendermint/tendermint/config"
	"github.com/tendermint/tendermint/internal/libs/clist"
	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
	"github.com/tendermint/tendermint/internal/mempool"
	"github.com/tendermint/tendermint/internal/proxy"
	"github.com/tendermint/tendermint/libs/log"
	tmmath "github.com/tendermint/tendermint/libs/math"
	"github.com/tendermint/tendermint/types"
)
type CListMempool struct {
	height   int64  
	txsBytes int64  
	notifiedTxsAvailable bool
	txsAvailable         chan struct{}  
	config *config.MempoolConfig
	updateMtx tmsync.RWMutex
	preCheck  mempool.PreCheckFunc
	postCheck mempool.PostCheckFunc
	txs          *clist.CList  
	proxyAppConn proxy.AppConnMempool
	recheckCursor *clist.CElement  
	recheckEnd    *clist.CElement  
	txsMap sync.Map
	cache mempool.TxCache
	logger  log.Logger
	metrics *mempool.Metrics
}
var _ mempool.Mempool = &CListMempool{}
type CListMempoolOption func(*CListMempool)
func NewCListMempool(
	cfg *config.MempoolConfig,
	proxyAppConn proxy.AppConnMempool,
	height int64,
	options ...CListMempoolOption,
) *CListMempool {
	mp := &CListMempool{
		config:        cfg,
		proxyAppConn:  proxyAppConn,
		txs:           clist.New(),
		height:        height,
		recheckCursor: nil,
		recheckEnd:    nil,
		logger:        log.NewNopLogger(),
		metrics:       mempool.NopMetrics(),
	}
	if cfg.CacheSize > 0 {
		mp.cache = mempool.NewLRUTxCache(cfg.CacheSize)
	} else {
		mp.cache = mempool.NopTxCache{}
	}
	proxyAppConn.SetResponseCallback(mp.globalCb)
	for _, option := range options {
		option(mp)
	}
	return mp
}
func (mem *CListMempool) EnableTxsAvailable() {
	mem.txsAvailable = make(chan struct{}, 1)
}
func (mem *CListMempool) SetLogger(l log.Logger) {
	mem.logger = l
}
func WithPreCheck(f mempool.PreCheckFunc) CListMempoolOption {
	return func(mem *CListMempool) { mem.preCheck = f }
}
func WithPostCheck(f mempool.PostCheckFunc) CListMempoolOption {
	return func(mem *CListMempool) { mem.postCheck = f }
}
func WithMetrics(metrics *mempool.Metrics) CListMempoolOption {
	return func(mem *CListMempool) { mem.metrics = metrics }
}
func (mem *CListMempool) Lock() {
	mem.updateMtx.Lock()
}
func (mem *CListMempool) Unlock() {
	mem.updateMtx.Unlock()
}
func (mem *CListMempool) Size() int {
	return mem.txs.Len()
}
func (mem *CListMempool) SizeBytes() int64 {
	return atomic.LoadInt64(&mem.txsBytes)
}
func (mem *CListMempool) FlushAppConn() error {
	return mem.proxyAppConn.FlushSync(context.Background())
}
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
	})
}
func (mem *CListMempool) TxsFront() *clist.CElement {
	return mem.txs.Front()
}
func (mem *CListMempool) TxsWaitChan() <-chan struct{} {
	return mem.txs.WaitChan()
}
func (mem *CListMempool) CheckTx(
	ctx context.Context,
	tx types.Tx,
	cb func(*abci.Response),
	txInfo mempool.TxInfo,
) error {
	mem.updateMtx.RLock()
	defer mem.updateMtx.RUnlock()
	txSize := len(tx)
	if err := mem.isFull(txSize); err != nil {
		return err
	}
	if txSize > mem.config.MaxTxBytes {
		return types.ErrTxTooLarge{
			Max:    mem.config.MaxTxBytes,
			Actual: txSize,
		}
	}
	if mem.preCheck != nil {
		if err := mem.preCheck(tx); err != nil {
			return types.ErrPreCheck{
				Reason: err,
			}
		}
	}
	if err := mem.proxyAppConn.Error(); err != nil {
		return err
	}
	if !mem.cache.Push(tx) {  
		if e, ok := mem.txsMap.Load(tx.Key()); ok {
			memTx := e.(*clist.CElement).Value.(*mempoolTx)
			_, loaded := memTx.senders.LoadOrStore(txInfo.SenderID, true)
			if loaded {
				return types.ErrTxInCache
			}
		}
		mem.logger.Debug("tx exists already in cache", "tx_hash", tx.Hash())
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	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
}
func (mem *CListMempool) globalCb(req *abci.Request, res *abci.Response) {
	if mem.recheckCursor == nil {
		return
	}
	mem.metrics.RecheckTimes.Add(1)
	mem.resCbRecheck(req, res)
	mem.metrics.Size.Set(float64(mem.Size()))
}
func (mem *CListMempool) reqResCb(
	tx []byte,
	peerID uint16,
	peerP2PID types.NodeID,
	externalCb func(*abci.Response),
) func(res *abci.Response) {
	return func(res *abci.Response) {
		if mem.recheckCursor != nil {
			panic("recheck cursor is not nil in reqResCb")
		}
		mem.resCbFirstTime(tx, peerID, peerP2PID, res)
		mem.metrics.Size.Set(float64(mem.Size()))
		if externalCb != nil {
			externalCb(res)
		}
	}
}
func (mem *CListMempool) addTx(memTx *mempoolTx) {
	e := mem.txs.PushBack(memTx)
	mem.txsMap.Store(memTx.tx.Key(), e)
	atomic.AddInt64(&mem.txsBytes, int64(len(memTx.tx)))
	mem.metrics.TxSizeBytes.Observe(float64(len(memTx.tx)))
}