func (tp *ethTxPool) demoteUnexecutables() {
	for addr, accountTxs := range tp.pending {
		nonce := tp.safeGetNonce(addr)
		for _, tx := range accountTxs.Forward(nonce) {
			hash := tx.Hash()
			delete(tp.all, hash)
		}
		if accountTxs.Len() == 0 {
			delete(tp.pending, addr)
		} else if accountTxs.Len() > 0 && accountTxs.Get(nonce) == nil {
			for _, tx := range accountTxs.items {
				log.Warn("Demoting invalidated transaction", zap.String("hash", tx.Hash().Hex()))
				if err := tp.addWaiting(tx, addr); err != nil {
					delete(tp.all, tx.Hash())
				}
			}
			delete(tp.pending, addr)
		}
	}
}
demoteUnexecutables removes invalid and processed transactions from the pools
executable/pending queue and any subsequent transactions that become unexecutable
are moved back into the waiting queue.
package evm
import (
	"bytes"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/spf13/viper"
	"go.uber.org/zap"
	"github.com/dappledger/AnnChain/eth/common"
	etypes "github.com/dappledger/AnnChain/eth/core/types"
	"github.com/dappledger/AnnChain/eth/rlp"
	"github.com/dappledger/AnnChain/gemmill/modules/go-clist"
	"github.com/dappledger/AnnChain/gemmill/modules/go-log"
	"github.com/dappledger/AnnChain/gemmill/types"
)
const (
	txEvictInterval = 1 * time.Minute
	waitingLifeTime = 10 * time.Minute
)
var (
	errTxExist                  = errors.New("tx already exist in cache")
	errTxPoolWaitingQueueIsFull = errors.New("evm tx pool waiting queue is full")
)
type ethTxPool struct {
	pending         map[common.Address]*txSortedMap  
	waiting         map[common.Address]*txSortedMap  
	waitingBeats    map[common.Address]time.Time     
	broadcastQueue  *clist.CList                     
	all             map[common.Hash]types.Tx         
	extTxs          *clist.CList                     
	mtx             sync.Mutex
	app             *EVMApp
	waitingLifeTime time.Duration  
	waitingLimit    int            
	pendingLimit    int            
	height          int64
	filter          []types.IFilter
}
func NewEthTxPool(app *EVMApp, conf *viper.Viper) *ethTxPool {
	return &ethTxPool{
		all:             make(map[common.Hash]types.Tx),
		waiting:         make(map[common.Address]*txSortedMap),
		waitingBeats:    make(map[common.Address]time.Time),
		pending:         make(map[common.Address]*txSortedMap),
		extTxs:          clist.New(),
		broadcastQueue:  clist.New(),
		waitingLimit:    conf.GetInt("block_size") * 10,
		pendingLimit:    conf.GetInt("block_size") * 10,
		waitingLifeTime: waitingLifeTime,
		app:             app,
	}
}
func (tp *ethTxPool) Start(height int64) {
	tp.setHeight(height)
	go tp.loop()
}
func (tp *ethTxPool) loop() {
	evict := time.NewTicker(txEvictInterval)
	defer evict.Stop()
	for {
		select {
		case <-evict.C:
			tp.Lock()
			for addr := range tp.waitingBeats {
				if time.Since(tp.waitingBeats[addr]) > tp.waitingLifeTime {
					if tp.waiting[addr].Get(tp.safeGetNonce(addr)) != nil {
						continue
					}
					for _, tx := range tp.waiting[addr].Flatten() {
						delete(tp.all, tx.Hash())
					}
					delete(tp.waitingBeats, addr)
					delete(tp.waiting, addr)
				}
			}
			tp.Unlock()
		}
	}
}
func (tp *ethTxPool) Lock() {
	tp.mtx.Lock()
}
func (tp *ethTxPool) Unlock() {
	tp.mtx.Unlock()
}
func (tp *ethTxPool) setHeight(height int64) {
	atomic.StoreInt64(&tp.height, height)
}
func (tp *ethTxPool) RegisterFilter(filter types.IFilter) {
	tp.filter = append(tp.filter, filter)
}
func (tp *ethTxPool) Reap(maxTxs int) []types.Tx {
	tp.Lock()
	defer tp.Unlock()
	if maxTxs == 0 {
		return []types.Tx{}
	} else if maxTxs < 0 {
		maxTxs = tp.pendingLimit
	}
	allTxs := make([]types.Tx, 0, maxTxs)
	if extTxs := tp.reapAdminOP(maxTxs); len(extTxs) == maxTxs {
		return extTxs
	} else {
		allTxs = append(allTxs, extTxs...)
	}
OUTLOOP:  
	for _, accountTxs := range tp.pending {
		for _, tx := range accountTxs.Flatten() {
			txBytes, exist := tp.all[tx.Hash()]
			if !exist {
				txBytes, _ = rlp.EncodeToBytes(tx)
			}
			allTxs = append(allTxs, txBytes)
			if len(allTxs) == maxTxs {
				break OUTLOOP
			}
		}
	}
	log.Debug("reap return txs", zap.Int("count", len(allTxs)))
	return allTxs
}
func (tp *ethTxPool) ReceiveTx(rawTx types.Tx) error {
	if types.IsAdminOP(rawTx) {
		return tp.handleAdminOP(rawTx)
	}
	tx := &etypes.Transaction{}
	if err := rlp.DecodeBytes(rawTx, tx); err != nil {
		return err
	}
	if err := tp.CheckAndAdd(tx, rawTx); err != nil {
		return err
	}
	tp.Lock()
	tp.broadcastNewTx(rawTx)
	tp.Unlock()
	return nil
}
func (tp *ethTxPool) GetPendingMaxNonce(bytAddress []byte) (uint64, error) {
	tp.Lock()
	defer tp.Unlock()
	address := common.BytesToAddress(bytAddress)
	txWSMap, ok := tp.waiting[address]
	if ok && txWSMap != nil {
		pendingMaxNonce := tp.unSafeGetPendingMaxNonce(address)
		if pendingMaxNonce != txWSMap.MinNonce() {
			return pendingMaxNonce, nil
		}
		return txWSMap.MaxNonce() + 1, nil
	}
	return tp.unSafeGetPendingMaxNonce(address), nil
}
func (tp *ethTxPool) unSafeGetPendingMaxNonce(address common.Address) uint64 {
	txSMap, ok := tp.pending[address]
	if ok && txSMap != nil {
		return txSMap.MaxNonce() + 1
	}
	return tp.safeGetNonce(address)
}
func (tp *ethTxPool) handleAdminOP(tx types.Tx) error {
	tp.Lock()
	defer tp.Unlock()
	for e := tp.extTxs.Front(); e != nil; e = e.Next() {
		extTx := e.Value.(types.Tx)
		if bytes.Equal(tx, extTx) {
			return errTxExist
		}
	}
	if tp.extTxs.Len() >= tp.pendingLimit {
		front := tp.extTxs.Front()
		tp.extTxs.Remove(front)
		front.DetachPrev()
	}
	tp.extTxs.PushBack(tx)
	tp.broadcastNewTx(tx)
	return nil
}
func (tp *ethTxPool) reapAdminOP(maxTxs int) []types.Tx {
	if tp.extTxs.Len() == 0 {
		return []types.Tx{}
	}
	txs := make([]types.Tx, 0, maxTxs)
	for e := tp.extTxs.Front(); e != nil && len(txs) < maxTxs; e = e.Next() {
		extTx := e.Value.(types.Tx)
		txs = append(txs, extTx)
	}
	return txs
}
func (tp *ethTxPool) refreshAdminOP(txsMap map[string]struct{}) {
	if tp.extTxs.Len() == 0 {
		return
	}
	txsLen := tp.extTxs.Len()
	index := 0
	for e := tp.extTxs.Front(); e != nil && index < txsLen; e = e.Next() {
		index++
		extTx := e.Value.(types.Tx)
		if _, ok := txsMap[string(extTx)]; ok {
			tp.extTxs.Remove(e)
			e.DetachPrev()
		}
	}
}
func (tp *ethTxPool) safeGetNonce(addr common.Address) uint64 {
	tp.app.stateMtx.Lock()
	nonce := tp.app.state.GetNonce(addr)
	tp.app.stateMtx.Unlock()
	return nonce
}
func (tp *ethTxPool) CheckAndAdd(tx *etypes.Transaction, rawTx types.Tx) error {
	tp.Lock()
	defer tp.Unlock()
	if _, exist := tp.all[tx.Hash()]; exist {
		return errTxExist
	}
	from, _ := etypes.Sender(tp.app.Signer, tx)
	currentNonce := tp.safeGetNonce(from)
	if currentNonce > tx.Nonce() {
		return fmt.Errorf("nonce(%d) different with getNonce(%d)", tx.Nonce(), currentNonce)
	}
	if err := tp.addWaiting(tx, from); err != nil {
		return err
	}
	tp.all[tx.Hash()] = rawTx
	if currentNonce == tx.Nonce() {
		tp.promoteExecutables([]common.Address{from})
	}
	return nil
}
func (tp *ethTxPool) Update(height int64, txs []types.Tx) {
	log.Debug("update tx pool txs", zap.Int64("height", height))
	tp.Lock()
	defer tp.Unlock()
	tp.setHeight(height)
	if len(txs) == 0 {
		return
	}
	txsMap := make(map[string]struct{})
	for _, tx := range txs {
		txsMap[string(tx)] = struct{}{}
	}
	tp.refreshBroadcastList(txsMap)
	tp.refreshAdminOP(txsMap)
	return
}
func (tp *ethTxPool) updateToState() {
	log.Debug("update txpool to evm app state")
	tp.Lock()
	defer tp.Unlock()
	tp.demoteUnexecutables()
	tp.promoteExecutables(nil)
}
func (tp *ethTxPool) Size() int {
	tp.Lock()
	defer tp.Unlock()
	return tp.extTxs.Len() + len(tp.all)
}
func (tp *ethTxPool) TxsFrontWait() *clist.CElement {
	return tp.broadcastQueue.FrontWait()
}
func (tp *ethTxPool) Flush() {
	tp.Lock()
	tp.waiting = make(map[common.Address]*txSortedMap)
	tp.pending = make(map[common.Address]*txSortedMap)
	tp.waitingBeats = make(map[common.Address]time.Time)
	tp.all = make(map[common.Hash]types.Tx)
	tp.broadcastQueue = clist.New()
	tp.extTxs = clist.New()
	tp.Unlock()
}
func (tp *ethTxPool) promoteExecutables(addrs []common.Address) {
	pendingTxCount := 0
	for _, accountTxs := range tp.pending {
		pendingTxCount += accountTxs.Len()
	}
	if pendingTxCount >= tp.pendingLimit {
		return
	}
	if addrs == nil {
		addrs = make([]common.Address, 0, len(tp.waiting))
		for addr := range tp.waiting {
			addrs = append(addrs, addr)
		}
	}
	for _, addr := range addrs {
		if pendingTxCount >= tp.pendingLimit {
			return
		}
		nonce := tp.safeGetNonce(addr)
		waiting := tp.waiting[addr]
		oldTxs := waiting.Forward(nonce)
		for _, otx := range oldTxs {
			delete(tp.all, otx.Hash())
		}
		txs := waiting.ReadyN(nonce, tp.pendingLimit-pendingTxCount)
		if waiting.Len() == 0 {
			delete(tp.waiting, addr)
			delete(tp.waitingBeats, addr)
		}
		if len(txs) == 0 {
			continue
		}
		if tp.pending[addr] == nil {
			tp.pending[addr] = newTxSortedMap()
		}
		for _, tx := range txs {
			if err := tp.pending[addr].Add(tx); err == nil {
				pendingTxCount++
			}
		}
	}
}
func (tp *ethTxPool) addWaiting(tx *etypes.Transaction, address common.Address) error {
	waitingTxCount := 0
	for _, txs := range tp.waiting {
		waitingTxCount += txs.Len()
	}
	if waitingTxCount >= tp.waitingLimit {
		if tp.waiting[address] == nil || !tp.waiting[address].TryReplace(tx) {
			return errTxPoolWaitingQueueIsFull
		}
	} else {
		if tp.waiting[address] == nil {
			tp.waiting[address] = newTxSortedMap()
		}
		err := tp.waiting[address].Add(tx)
		if err != nil {
			return err
		}
	}
	tp.waitingBeats[address] = time.Now()
	return nil
}