feat: Cache Tx Decoder (#528)
* init * nit * nits * nit * more nits * go version bump * image bump * nit
This commit is contained in:
David Terpay
@@ -0,0 +1,121 @@
|
||||
package utils
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
sdk "github.com/cosmos/cosmos-sdk/types"
|
||||
)
|
||||
|
||||
// DefaultMaxSize is the default maximum size of the cache.
|
||||
var DefaultMaxSize uint64 = 500
|
||||
|
||||
// CacheTxDecoder wraps the sdk.TxDecoder and caches the decoded transactions with
|
||||
// an LRU'esque cache. Each transaction is cached using the transaction's hash
|
||||
// as the key. The cache is purged when the number of transactions in the cache
|
||||
// exceeds the maximum size. The oldest transactions are removed first.
|
||||
type CacheTxDecoder struct {
|
||||
mut sync.Mutex
|
||||
|
||||
decoder sdk.TxDecoder
|
||||
cache map[string]sdk.Tx
|
||||
window []string
|
||||
insertIndex int
|
||||
oldestIndex int
|
||||
maxSize uint64
|
||||
}
|
||||
|
||||
// NewDefaultCacheTxDecoder returns a new CacheTxDecoder.
|
||||
func NewDefaultCacheTxDecoder(
|
||||
decoder sdk.TxDecoder,
|
||||
) (*CacheTxDecoder, error) {
|
||||
if decoder == nil {
|
||||
return nil, fmt.Errorf("decoder cannot be nil")
|
||||
}
|
||||
|
||||
return &CacheTxDecoder{
|
||||
decoder: decoder,
|
||||
cache: make(map[string]sdk.Tx),
|
||||
window: make([]string, DefaultMaxSize),
|
||||
insertIndex: 0,
|
||||
oldestIndex: 0,
|
||||
maxSize: DefaultMaxSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// NewCacheTxDecoder returns a new CacheTxDecoder with the given cache interval.
|
||||
func NewCacheTxDecoder(
|
||||
decoder sdk.TxDecoder,
|
||||
maxSize uint64,
|
||||
) (*CacheTxDecoder, error) {
|
||||
if decoder == nil {
|
||||
return nil, fmt.Errorf("decoder cannot be nil")
|
||||
}
|
||||
|
||||
return &CacheTxDecoder{
|
||||
decoder: decoder,
|
||||
cache: make(map[string]sdk.Tx),
|
||||
window: make([]string, maxSize),
|
||||
insertIndex: 0,
|
||||
oldestIndex: 0,
|
||||
maxSize: maxSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Decode decodes the transaction bytes into a sdk.Tx. It caches the decoded
|
||||
// transaction using the transaction's hash as the key.
|
||||
func (ctd *CacheTxDecoder) TxDecoder() sdk.TxDecoder {
|
||||
return func(txBytes []byte) (sdk.Tx, error) {
|
||||
ctd.mut.Lock()
|
||||
defer ctd.mut.Unlock()
|
||||
|
||||
hash := TxHash(txBytes)
|
||||
if tx, ok := ctd.cache[hash]; ok {
|
||||
return tx, nil
|
||||
}
|
||||
|
||||
tx, err := ctd.decoder(txBytes)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Purge the cache if necessary
|
||||
if uint64(len(ctd.cache)) >= ctd.maxSize {
|
||||
// Purge the oldest transaction
|
||||
entry := ctd.window[ctd.oldestIndex]
|
||||
delete(ctd.cache, entry)
|
||||
|
||||
// Increment the oldest index
|
||||
ctd.oldestIndex++
|
||||
ctd.oldestIndex %= int(ctd.maxSize)
|
||||
}
|
||||
|
||||
// Update the cache and window
|
||||
ctd.cache[hash] = tx
|
||||
ctd.window[ctd.insertIndex] = hash
|
||||
|
||||
// Increment the insert index
|
||||
ctd.insertIndex++
|
||||
ctd.insertIndex %= int(ctd.maxSize)
|
||||
|
||||
return tx, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Len returns the number of transactions in the cache.
|
||||
func (ctd *CacheTxDecoder) Len() int {
|
||||
ctd.mut.Lock()
|
||||
defer ctd.mut.Unlock()
|
||||
|
||||
return len(ctd.cache)
|
||||
}
|
||||
|
||||
// Contains returns true if the cache contains the transaction with the given hash.
|
||||
func (ctd *CacheTxDecoder) Contains(txBytes []byte) bool {
|
||||
ctd.mut.Lock()
|
||||
defer ctd.mut.Unlock()
|
||||
|
||||
hash := TxHash(txBytes)
|
||||
_, ok := ctd.cache[hash]
|
||||
return ok
|
||||
}
|
||||
@@ -0,0 +1,218 @@
|
||||
package utils_test
|
||||
|
||||
import (
|
||||
"math/rand"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/skip-mev/block-sdk/v2/block/utils"
|
||||
"github.com/skip-mev/block-sdk/v2/testutils"
|
||||
)
|
||||
|
||||
var (
|
||||
numAccounts = 5
|
||||
numTxsPerAcct = 100
|
||||
cacheSize = 500
|
||||
)
|
||||
|
||||
func BenchmarkCacheDecoding(b *testing.B) {
|
||||
encodingCfg := testutils.CreateTestEncodingConfig()
|
||||
decoder := encodingCfg.TxConfig.TxDecoder()
|
||||
|
||||
random := rand.New(rand.NewSource(time.Now().Unix()))
|
||||
account := testutils.RandomAccounts(random, numAccounts)
|
||||
|
||||
txs := make([][]byte, numAccounts*numTxsPerAcct)
|
||||
for i := 0; i < numAccounts; i++ {
|
||||
for j := 0; j < numTxsPerAcct; j++ {
|
||||
txBytes, err := testutils.CreateRandomTxBz(
|
||||
encodingCfg.TxConfig,
|
||||
account[i],
|
||||
uint64(j),
|
||||
1,
|
||||
2,
|
||||
0,
|
||||
)
|
||||
require.NoError(b, err)
|
||||
|
||||
txs[i*numTxsPerAcct+j] = txBytes
|
||||
}
|
||||
}
|
||||
|
||||
cacheTxDecoder, err := utils.NewCacheTxDecoder(decoder, uint64(cacheSize))
|
||||
require.NoError(b, err)
|
||||
decoder = cacheTxDecoder.TxDecoder()
|
||||
b.ResetTimer()
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
for _, txBytes := range txs {
|
||||
_, err := decoder(txBytes)
|
||||
require.NoError(b, err)
|
||||
}
|
||||
|
||||
for _, txBytes := range txs {
|
||||
_, err := decoder(txBytes)
|
||||
require.NoError(b, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkStandardDecoding(b *testing.B) {
|
||||
encodingCfg := testutils.CreateTestEncodingConfig()
|
||||
decoder := encodingCfg.TxConfig.TxDecoder()
|
||||
|
||||
random := rand.New(rand.NewSource(time.Now().Unix()))
|
||||
account := testutils.RandomAccounts(random, numAccounts)
|
||||
|
||||
txs := make([][]byte, numAccounts*numTxsPerAcct)
|
||||
for i := 0; i < numAccounts; i++ {
|
||||
for j := 0; j < numTxsPerAcct; j++ {
|
||||
txBytes, err := testutils.CreateRandomTxBz(
|
||||
encodingCfg.TxConfig,
|
||||
account[i],
|
||||
uint64(j),
|
||||
1,
|
||||
2,
|
||||
0,
|
||||
)
|
||||
require.NoError(b, err)
|
||||
|
||||
txs[i*numTxsPerAcct+j] = txBytes
|
||||
}
|
||||
}
|
||||
|
||||
b.ResetTimer()
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
for _, txBytes := range txs {
|
||||
_, err := decoder(txBytes)
|
||||
require.NoError(b, err)
|
||||
}
|
||||
|
||||
for _, txBytes := range txs {
|
||||
_, err := decoder(txBytes)
|
||||
require.NoError(b, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewCacheTxDecoder(t *testing.T) {
|
||||
encodingCfg := testutils.CreateTestEncodingConfig()
|
||||
decoder := encodingCfg.TxConfig.TxDecoder()
|
||||
|
||||
_, err := utils.NewDefaultCacheTxDecoder(decoder)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = utils.NewCacheTxDecoder(decoder, 100)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = utils.NewCacheTxDecoder(nil, 100)
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
func TestDecode(t *testing.T) {
|
||||
encodingCfg := testutils.CreateTestEncodingConfig()
|
||||
decoder := encodingCfg.TxConfig.TxDecoder()
|
||||
|
||||
random := rand.New(rand.NewSource(time.Now().Unix()))
|
||||
account := testutils.RandomAccounts(random, 1)
|
||||
|
||||
t.Run("decode valid tx and check that it is cached", func(t *testing.T) {
|
||||
txBytes, err := testutils.CreateRandomTxBz(
|
||||
encodingCfg.TxConfig,
|
||||
account[0],
|
||||
0,
|
||||
1,
|
||||
1,
|
||||
0,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
cacheTxDecoder, err := utils.NewDefaultCacheTxDecoder(decoder)
|
||||
require.NoError(t, err)
|
||||
|
||||
decoder := cacheTxDecoder.TxDecoder()
|
||||
|
||||
tx, err := decoder(txBytes)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, tx)
|
||||
require.Equal(t, 1, cacheTxDecoder.Len())
|
||||
require.True(t, cacheTxDecoder.Contains(txBytes))
|
||||
|
||||
// decode the same tx again
|
||||
tx, err = decoder(txBytes)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, tx)
|
||||
require.Equal(t, 1, cacheTxDecoder.Len())
|
||||
require.True(t, cacheTxDecoder.Contains(txBytes))
|
||||
})
|
||||
|
||||
t.Run("decode invalid tx", func(t *testing.T) {
|
||||
cacheTxDecoder, err := utils.NewDefaultCacheTxDecoder(decoder)
|
||||
require.NoError(t, err)
|
||||
|
||||
decoder := cacheTxDecoder.TxDecoder()
|
||||
tx, err := decoder([]byte("invalid tx"))
|
||||
require.Error(t, err)
|
||||
require.Nil(t, tx)
|
||||
require.Equal(t, 0, cacheTxDecoder.Len())
|
||||
})
|
||||
|
||||
t.Run("decode multiple txs without hitting limit", func(t *testing.T) {
|
||||
cacheTxDecoder, err := utils.NewCacheTxDecoder(decoder, 100)
|
||||
require.NoError(t, err)
|
||||
|
||||
for i := 0; i < 100; i++ {
|
||||
txBytes, err := testutils.CreateRandomTxBz(
|
||||
encodingCfg.TxConfig,
|
||||
account[0],
|
||||
uint64(i),
|
||||
1,
|
||||
1,
|
||||
0,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
decoder := cacheTxDecoder.TxDecoder()
|
||||
tx, err := decoder(txBytes)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, tx)
|
||||
require.Equal(t, i+1, cacheTxDecoder.Len())
|
||||
require.True(t, cacheTxDecoder.Contains(txBytes))
|
||||
}
|
||||
require.Equal(t, 100, cacheTxDecoder.Len())
|
||||
})
|
||||
|
||||
t.Run("decode multiple txs hitting limit", func(t *testing.T) {
|
||||
maxSize := uint64(2)
|
||||
cacheTxDecoder, err := utils.NewCacheTxDecoder(decoder, maxSize)
|
||||
require.NoError(t, err)
|
||||
|
||||
for i := 0; i < int(maxSize*3); i++ {
|
||||
txBytes, err := testutils.CreateRandomTxBz(
|
||||
encodingCfg.TxConfig,
|
||||
account[0],
|
||||
uint64(i),
|
||||
1,
|
||||
1,
|
||||
0,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
decoder := cacheTxDecoder.TxDecoder()
|
||||
tx, err := decoder(txBytes)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, tx)
|
||||
require.True(t, cacheTxDecoder.Contains(txBytes))
|
||||
|
||||
if i < int(maxSize) {
|
||||
require.Equal(t, i+1, cacheTxDecoder.Len())
|
||||
} else {
|
||||
require.Equal(t, int(maxSize), cacheTxDecoder.Len())
|
||||
}
|
||||
}
|
||||
require.Equal(t, int(maxSize), cacheTxDecoder.Len())
|
||||
})
|
||||
}
|
||||
@@ -61,8 +61,12 @@ func GetTxHash(encoder sdk.TxEncoder, tx sdk.Tx) (string, error) {
|
||||
return "", fmt.Errorf("failed to encode transaction: %w", err)
|
||||
}
|
||||
|
||||
txHashStr := strings.ToUpper(hex.EncodeToString(comettypes.Tx(txBz).Hash()))
|
||||
return txHashStr, nil
|
||||
return TxHash(txBz), nil
|
||||
}
|
||||
|
||||
// TxHash returns the string hash representation of the given transactions.
|
||||
func TxHash(txBytes []byte) string {
|
||||
return strings.ToUpper(hex.EncodeToString(comettypes.Tx(txBytes).Hash()))
|
||||
}
|
||||
|
||||
// GetDecodedTxs returns the decoded transactions from the given bytes.
|
||||
|
||||
Reference in New Issue
Block a user