Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
216 changes: 165 additions & 51 deletions sei-tendermint/internal/mempool/tx.go
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,8 @@ type txStoreInner struct {
// Snapshot of the mempool state: transactions in inclusion order.
// Recomputed every time inInclusionOrder is called.
snapshot types.Txs
// snapshotStale is set when the store changed without recomputing the snapshot.
snapshotStale bool
// Cache of known txs, reducess pressure on app. It contains:
// * known unconditionally invalid txs
// * metadata allowing to do initial tx assessment before calling app.CheckTx
Expand Down Expand Up @@ -238,8 +240,17 @@ func (s *txStore) MarkInvalid(txHash types.TxHash) {
func (s *txStore) State() txStoreState { return s.state.Load() }

// Recent snapshot of the mempool.
// The snapshot is recomputed here lazily if it went stale since the last block.
func (s *txStore) RecentSnapshot() types.Txs {
for inner := range s.inner.RLock() {
if !inner.snapshotStale {
return inner.snapshot
}
}
for inner := range s.inner.Lock() {

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[suggestion] RecentSnapshot can now escalate from RLock to the exclusive Lock and run the O(m log m) inInclusionOrder sort inline. Since refresh/Reap set snapshotStale, the first /unconfirmed_txs request after each block pays that sort while holding the write lock, blocking Insert/CheckTx/Reap and the consensus Update for roughly the 9–20 ms the PR description measures at 10k–20k txs. Total work is not higher than before (consensus used to pay it unconditionally), but it is now triggerable at an arbitrary moment by an RPC caller rather than at a point consensus controls.

Two cheaper options: keep the snapshot a byproduct of the inInclusionOrder that Reap already runs under the lock, or compute the ordering into a local slice and publish it with a short critical section. Worth at least a comment recording the intended trade-off.

Separately, this makes RecentSnapshot unsafe to call from any context already holding inner.RLock() (sync.RWMutex is not reentrant). No current caller does, but the previous version was safe there.

if inner.snapshotStale {
inner.inInclusionOrder()
}
return inner.snapshot
}
panic("unreachable")
Expand Down Expand Up @@ -353,35 +364,103 @@ func (inner *txStoreInner) shouldReject(txHash types.TxHash) bool {
return false
}

// account returns the tracked state of the evm account, fetching it from the app on first use.
func (s *txStore) account(inner *txStoreInner, evm *evmTx) *evmAccount {
account, ok := inner.accounts[evm.address]
if !ok {
// TODO(gprusak): consider whether we should move these queries out of the mutex.
b := s.app.EvmBalance(evm.address, evm.seiAddress)
n := s.app.EvmNonce(evm.address)
account = &evmAccount{b, n, n}
inner.accounts[evm.address] = account
}
return account
}

// cacheMetadata records the evm metadata of wtx in the cache.
func (inner *txStoreInner) cacheMetadata(wtx *WrappedTx, evm *evmTx) {
inner.cache.Push(wtx.Hash(), utils.Some(cacheEvm{
priority: wtx.priority,
address: evm.address,
nonce: evm.nonce,
requiredBalance: evm.requiredBalance,
}))
}

// advanceReady marks the account's txs ready in nonce order, starting at nextNonce,
// until it hits a nonce gap or a tx the account cannot afford.
// The tx in skipAccepted is not reported as a newly accepted pending tx.
func (s *txStore) advanceReady(inner *txStoreInner, addr common.Address, account *evmAccount, state *txStoreState, skipAccepted utils.Option[*WrappedTx]) {
an := evmAddrNonce{Address: addr}
for {
an.Nonce = account.nextNonce
wtx, ok := inner.byNonce[an]
if !ok {
break
}
requiredBalance := wtx.evm.OrPanic("non-evm tx").requiredBalance
if account.balance.Cmp(&requiredBalance) < 0 {
break
}
account.nextNonce += 1
state.ready.Inc(wtx.Size())
if !wtx.readyEl.IsPresent() {
s.priorityReservoir.Add(wtx.priority)
wtx.readyEl = utils.Some(s.readyTxs.PushBack(wtx.Tx()))
if skip, ok := skipAccepted.Get(); !ok || wtx != skip {
recordPendingNonceAccepted()
}
}
}
}

// refreshReady recomputes which of the account's txs are ready, starting over from firstNonce.
func (s *txStore) refreshReady(inner *txStoreInner, addr common.Address, account *evmAccount, state *txStoreState) {
for nonce := account.firstNonce; nonce < account.nextNonce; nonce++ {
if wtx, ok := inner.byNonce[evmAddrNonce{addr, nonce}]; ok {
state.ready.Dec(wtx.Size())
}
}
account.nextNonce = account.firstNonce
s.advanceReady(inner, addr, account, state, utils.None[*WrappedTx]())
}

// remove drops wtx from every index and from the gossip list.
// Readiness of the remaining txs of the same account is left for the caller to refresh.
func (s *txStore) remove(inner *txStoreInner, wtx *WrappedTx, state *txStoreState) {
delete(inner.byHash, wtx.Hash())
if evm, ok := wtx.evm.Get(); ok {
delete(inner.byEvmHash, evm.hash)
delete(inner.byNonce, evmAddrNonce{evm.address, evm.nonce})
}
state.total.Dec(wtx.Size())
if inner.isReady(wtx) {
state.ready.Dec(wtx.Size())
}
Global.RemovedTxsAt().Add(1)
if el, ok := wtx.readyEl.Get(); ok {
s.readyTxs.Remove(el)
}
}

func (s *txStore) insert(inner *txStoreInner, wtx *WrappedTx, recordAdded bool) error {
if _, ok := inner.byHash[wtx.Hash()]; ok {
return errDuplicateTx
}
state := inner.state.Load()
if evm, ok := wtx.evm.Get(); ok {
// Fetch the evm account state.
account, ok := inner.accounts[evm.address]
if !ok {
// TODO(gprusak): consider whether we should move these queries out of the mutex.
b := s.app.EvmBalance(evm.address, evm.seiAddress)
n := s.app.EvmNonce(evm.address)
account = &evmAccount{b, n, n}
inner.accounts[evm.address] = account
}
account := s.account(inner, &evm)
// Reject transactions with old nonces.
if evm.nonce < account.firstNonce {
inner.cache.Push(wtx.Hash(), utils.None[cacheEvm]())
recordPendingNonceRejected()
return errOldNonce
}
insertedAtNextNonce := evm.nonce == account.nextNonce
newTx := wtx
inner.cache.Push(wtx.Hash(), utils.Some(cacheEvm{
priority: wtx.priority,
address: evm.address,
nonce: evm.nonce,
requiredBalance: evm.requiredBalance,
}))
skipAccepted := utils.None[*WrappedTx]()
if recordAdded && evm.nonce == account.nextNonce {
skipAccepted = utils.Some(wtx)
}
inner.cacheMetadata(wtx, &evm)
// We check the evm hash only AFTER caching the evm metadata.
if _, ok := inner.byEvmHash[evm.hash]; ok {
return errDuplicateTx
Expand Down Expand Up @@ -416,27 +495,7 @@ func (s *txStore) insert(inner *txStoreInner, wtx *WrappedTx, recordAdded bool)
state.total.Inc(wtx.Size())
inner.byEvmHash[evm.hash] = wtx
inner.byNonce[an] = wtx
// Update account ready txs.
for {
an.Nonce = account.nextNonce
wtx, ok := inner.byNonce[an]
if !ok {
break
}
requiredBalance := wtx.evm.OrPanic("non-evm tx").requiredBalance
if account.balance.Cmp(&requiredBalance) < 0 {
break
}
account.nextNonce += 1
state.ready.Inc(wtx.Size())
if !wtx.readyEl.IsPresent() {
s.priorityReservoir.Add(wtx.priority)
wtx.readyEl = utils.Some(s.readyTxs.PushBack(wtx.Tx()))
if !recordAdded || wtx != newTx || !insertedAtNextNonce {
recordPendingNonceAccepted()
}
}
}
s.advanceReady(inner, evm.address, account, &state, skipAccepted)
} else {
// Non-evm txs are automatically ready
state.total.Inc(wtx.Size())
Expand Down Expand Up @@ -503,6 +562,7 @@ func (inner *txStoreInner) inInclusionOrder() []*WrappedTx {
for i := range inner.snapshot {
inner.snapshot[i] = res[i].Tx()
}
inner.snapshotStale = false
return res
}

Expand Down Expand Up @@ -558,6 +618,46 @@ func (s *txStore) compact(inner *txStoreInner, clearAccounts bool) {
Global.CacheSizeAt().Set(int64(inner.cache.Size()))
}

// O(m), re-evaluates account nonces and balances against the app and drops txs
// which fell below their account nonce. Unlike compact, it keeps the indices
// in place and never evicts, so the caller must ensure the store is within softLimit.
func (s *txStore) refresh(inner *txStoreInner) {
state := txStoreState{}
inner.accounts = map[common.Address]*evmAccount{}
for txHash, wtx := range inner.byHash {
state.total.Inc(wtx.Size())
evm, ok := wtx.evm.Get()
if !ok {
// Non-evm txs are automatically ready
state.ready.Inc(wtx.Size())
continue
}
account := s.account(inner, &evm)
if evm.nonce < account.firstNonce {
inner.cache.Push(txHash, utils.None[cacheEvm]())
recordPendingNonceRejected()
state.total.Dec(wtx.Size())
delete(inner.byHash, txHash)
delete(inner.byEvmHash, evm.hash)
delete(inner.byNonce, evmAddrNonce{evm.address, evm.nonce})
Global.RemovedTxsAt().Add(1)
Global.EvictedTxsAt().Add(1)
if el, ok := wtx.readyEl.Get(); ok {
s.readyTxs.Remove(el)
}
continue
}
inner.cacheMetadata(wtx, &evm)
}
for addr, account := range inner.accounts {
account.nextNonce = account.firstNonce
s.advanceReady(inner, addr, account, &state, utils.None[*WrappedTx]())
}
inner.state.Store(state)
inner.snapshotStale = true
Global.CacheSizeAt().Set(int64(inner.cache.Size()))
}

type updateSpec struct {
Now time.Time
Height int64
Expand Down Expand Up @@ -596,6 +696,7 @@ func (s *txStore) Update(spec updateSpec) {
inner.failedTxs.Push(txHash, struct{}{})
}
}
state := inner.state.Load()
for txHash, wtx := range inner.byHash {
expired := isExpired(wtx)
if expired {
Expand All @@ -614,15 +715,18 @@ func (s *txStore) Update(spec updateSpec) {
if s.config.KeepInvalidTxsInCache && !executed {
inner.cache.Push(txHash, utils.None[cacheEvm]())
}
delete(inner.byHash, txHash)
Global.RemovedTxsAt().Add(1)
if el, ok := wtx.readyEl.Get(); ok {
s.readyTxs.Remove(el)
}
s.remove(inner, wtx, &state)
} else if newPriority, ok := spec.NewPriorities[wtx.Hash()]; ok {
wtx.priority = newPriority
}
}
inner.state.Store(state)
// Eviction needs the full inclusion order, so fall back to compact only when
// the store may still exceed softLimit after the removals.
if state.total.LessEqual(&inner.softLimit) {
s.refresh(inner)
continue
}
start := time.Now()
s.compact(inner, true)
otelMetrics.compactTotal.Add(context.Background(), 1, triggerUpdateAttr)
Expand Down Expand Up @@ -683,17 +787,27 @@ func (s *txStore) Reap(l ReapLimits, remove bool) (types.Txs, int64) {
}
}
if remove {
state := inner.state.Load()
affected := map[common.Address]*evmAccount{}
for _, wtx := range wtxs {
delete(inner.byHash, wtx.Hash())
Global.RemovedTxsAt().Add(1)
if el, ok := wtx.readyEl.Get(); ok {
s.readyTxs.Remove(el)
s.remove(inner, wtx, &state)
if evm, ok := wtx.evm.Get(); ok {
affected[evm.address] = inner.accounts[evm.address]
}
}
start := time.Now()
s.compact(inner, false)
otelMetrics.compactTotal.Add(context.Background(), 1, triggerReapAttr)
otelMetrics.compactDurationSeconds.Record(context.Background(), time.Since(start).Seconds())
// Account nonces are only re-evaluated by Update, so the successors of the
// reaped txs become pending until the next block.
for addr, account := range affected {
s.refreshReady(inner, addr, account, &state)
}
inner.state.Store(state)
inner.snapshotStale = true
if !state.total.LessEqual(&inner.softLimit) {
start := time.Now()
s.compact(inner, false)
otelMetrics.compactTotal.Add(context.Background(), 1, triggerReapAttr)
otelMetrics.compactDurationSeconds.Record(context.Background(), time.Since(start).Seconds())
}
}
}

Expand Down
71 changes: 71 additions & 0 deletions sei-tendermint/internal/mempool/tx_bench_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
package mempool

import (
"fmt"
"testing"
"time"

"github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy"
"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils"
"github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require"
)

func benchTxStore(b *testing.B, numAccounts, txsPerAccount int) *txStore {
rng := utils.TestRng()
app := newEVMNonceApp()
cfg := TestConfig()
cfg.Size = numAccounts * txsPerAccount
cfg.PendingSize = numAccounts * txsPerAccount
cfg.MaxTxsBytes = 1 << 40
cfg.MaxPendingTxsBytes = 1 << 40
store := NewTxStore(cfg, proxy.New(app))
for range numAccounts {
addr := genEvmAddress(rng)
app.setNonce(addr, 0)
app.setBalance(addr, 1<<30)
for n := range txsPerAccount {
wtx := makeEvmTxForTest(rng, addr, uint64(n), int64(rng.Intn(1000)), 1)
require.NoError(b, store.Insert(wtx))
}
}
return store
}

func BenchmarkTxStore_Update(b *testing.B) {
for _, tc := range []struct{ accounts, per int }{{10000, 1}, {5000, 4}, {50000, 1}} {
b.Run(fmt.Sprintf("accounts=%d,per=%d", tc.accounts, tc.per), func(b *testing.B) {
store := benchTxStore(b, tc.accounts, tc.per)
b.ResetTimer()
for range b.N {
store.Update(updateSpec{Now: time.Now(), Height: 1, Constraints: TxConstraints{MaxGas: -1}})
}
})
}
}

func BenchmarkTxStore_InInclusionOrder(b *testing.B) {
for _, tc := range []struct{ accounts, per int }{{10000, 1}, {5000, 4}, {50000, 1}} {
b.Run(fmt.Sprintf("accounts=%d,per=%d", tc.accounts, tc.per), func(b *testing.B) {
store := benchTxStore(b, tc.accounts, tc.per)
b.ResetTimer()
for range b.N {
for inner := range store.inner.Lock() {
inner.inInclusionOrder()
}
}
})
}
}

func BenchmarkTxStore_ReapRemove(b *testing.B) {
for _, tc := range []struct{ accounts, per int }{{10000, 1}, {5000, 4}} {
b.Run(fmt.Sprintf("accounts=%d,per=%d", tc.accounts, tc.per), func(b *testing.B) {
for range b.N {
b.StopTimer()
store := benchTxStore(b, tc.accounts, tc.per)
b.StartTimer()
store.Reap(ReapLimits{MaxTxs: utils.Some(uint64(100))}, true)
}
})
}
}
Loading