Mempool optimization: on transaction addition, call "GetTransferredValue" and retain its value

dataRetriever/factory/dataPoolFactory.go

@@ -64,9 +64,10 @@ func NewDataPoolFromConfig(args ArgsDataPool) (dataRetriever.PoolsHolder, error)
 
 	txPool, err := txpool.NewShardedTxPool(txpool.ArgShardedTxPool{
 		Config:         factory.GetCacherFromConfig(mainConfig.TxDataPool),
+		TxGasHandler:   args.EconomicsData,
+		Marshalizer:    args.Marshalizer,
 		NumberOfShards: args.ShardCoordinator.NumberOfShards(),
 		SelfShardID:    args.ShardCoordinator.SelfId(),
-		TxGasHandler:   args.EconomicsData,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("%w while creating the cache for the transactions", err)

dataRetriever/txpool/argShardedTxPool.go

@@ -4,15 +4,16 @@ import (
 	"fmt"
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
+	"github.com/multiversx/mx-chain-core-go/marshal"
 	"github.com/multiversx/mx-chain-go/dataRetriever"
 	"github.com/multiversx/mx-chain-go/storage/storageunit"
-	"github.com/multiversx/mx-chain-go/storage/txcache"
 )
 
 // ArgShardedTxPool is the argument for ShardedTxPool's constructor
 type ArgShardedTxPool struct {
 	Config         storageunit.CacheConfig
-	TxGasHandler   txcache.TxGasHandler
+	TxGasHandler   txGasHandler
+	Marshalizer    marshal.Marshalizer
 	NumberOfShards uint32
 	SelfShardID    uint32
 }
@@ -39,6 +40,9 @@ func (args *ArgShardedTxPool) verify() error {
 	if check.IfNil(args.TxGasHandler) {
 		return fmt.Errorf("%w: TxGasHandler is not valid", dataRetriever.ErrNilTxGasHandler)
 	}
+	if check.IfNil(args.Marshalizer) {
+		return fmt.Errorf("%w: Marshalizer is not valid", dataRetriever.ErrNilMarshalizer)
+	}
 	if args.NumberOfShards == 0 {
 		return fmt.Errorf("%w: NumberOfShards is not valid", dataRetriever.ErrCacheConfigInvalidSharding)
 	}

dataRetriever/txpool/interface.go

@@ -1,6 +1,9 @@
 package txpool
 
 import (
+	"math/big"
+
+	"github.com/multiversx/mx-chain-core-go/data"
 	"github.com/multiversx/mx-chain-go/storage"
 	"github.com/multiversx/mx-chain-go/storage/txcache"
 )
@@ -17,3 +20,8 @@ type txCache interface {
 	Diagnose(deep bool)
 	GetTransactionsPoolForSender(sender string) []*txcache.WrappedTransaction
 }
+
+type txGasHandler interface {
+	ComputeTxFee(tx data.TransactionWithFeeHandler) *big.Int
+	IsInterfaceNil() bool
+}

dataRetriever/txpool/memorytests/memory_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/multiversx/mx-chain-core-go/core"
 	"github.com/multiversx/mx-chain-core-go/data/transaction"
+	"github.com/multiversx/mx-chain-core-go/marshal"
 	"github.com/multiversx/mx-chain-go/dataRetriever"
 	"github.com/multiversx/mx-chain-go/dataRetriever/txpool"
 	"github.com/multiversx/mx-chain-go/storage/storageunit"
@@ -112,6 +113,7 @@ func newPool() dataRetriever.ShardedDataCacherNotifier {
 	args := txpool.ArgShardedTxPool{
 		Config:         config,
 		TxGasHandler:   txcachemocks.NewTxGasHandlerMock(),
+		Marshalizer:    &marshal.GogoProtoMarshalizer{},
 		NumberOfShards: 2,
 		SelfShardID:    0,
 	}

dataRetriever/txpool/mempoolHost.go

@@ -0,0 +1,112 @@
+package txpool
+
+import (
+	"bytes"
+	"math/big"
+
+	"github.com/multiversx/mx-chain-core-go/core"
+	"github.com/multiversx/mx-chain-core-go/core/check"
+	"github.com/multiversx/mx-chain-core-go/data"
+	"github.com/multiversx/mx-chain-core-go/marshal"
+	"github.com/multiversx/mx-chain-go/dataRetriever"
+	"github.com/multiversx/mx-chain-go/process"
+	vmcommon "github.com/multiversx/mx-chain-vm-common-go"
+	"github.com/multiversx/mx-chain-vm-common-go/parsers"
+)
+
+type argsMempoolHost struct {
+	txGasHandler txGasHandler
+	marshalizer  marshal.Marshalizer
+}
+
+type mempoolHost struct {
+	txGasHandler        txGasHandler
+	callArgumentsParser process.CallArgumentsParser
+	esdtTransferParser  vmcommon.ESDTTransferParser
+}
+
+func newMempoolHost(args argsMempoolHost) (*mempoolHost, error) {
+	if check.IfNil(args.txGasHandler) {
+		return nil, dataRetriever.ErrNilTxGasHandler
+	}
+	if check.IfNil(args.marshalizer) {
+		return nil, dataRetriever.ErrNilMarshalizer
+	}
+
+	argsParser := parsers.NewCallArgsParser()
+
+	esdtTransferParser, err := parsers.NewESDTTransferParser(args.marshalizer)
+	if err != nil {
+		return nil, err
+	}
+
+	return &mempoolHost{
+		txGasHandler:        args.txGasHandler,
+		callArgumentsParser: argsParser,
+		esdtTransferParser:  esdtTransferParser,
+	}, nil
+}
+
+// ComputeTxFee computes the fee for a transaction.
+func (host *mempoolHost) ComputeTxFee(tx data.TransactionWithFeeHandler) *big.Int {
+	return host.txGasHandler.ComputeTxFee(tx)
+}
+
+// GetTransferredValue returns the value transferred by a transaction.
+func (host *mempoolHost) GetTransferredValue(tx data.TransactionHandler) *big.Int {
+	value := tx.GetValue()
+	hasValue := value != nil && value.Sign() != 0
+	if hasValue {
+		// Early exit (optimization): a transaction can either bear a regular value or be a "MultiESDTNFTTransfer".
+		return value
+	}
+
+	data := tx.GetData()
+	hasData := len(data) > 0
+	if !hasData {
+		// Early exit (optimization): no "MultiESDTNFTTransfer" to parse.
+		return tx.GetValue()
+	}
+
+	maybeMultiTransfer := bytes.HasPrefix(data, []byte(core.BuiltInFunctionMultiESDTNFTTransfer))
+	if !maybeMultiTransfer {
+		// Early exit (optimization).
+		return big.NewInt(0)
+	}
+
+	function, args, err := host.callArgumentsParser.ParseData(string(data))
+	if err != nil {
+		return big.NewInt(0)
+	}
+
+	if function != core.BuiltInFunctionMultiESDTNFTTransfer {
+		// Early exit (optimization).
+		return big.NewInt(0)
+	}
+
+	esdtTransfers, err := host.esdtTransferParser.ParseESDTTransfers(tx.GetSndAddr(), tx.GetRcvAddr(), function, args)
+	if err != nil {
+		return big.NewInt(0)
+	}
+
+	accumulatedNativeValue := big.NewInt(0)
+
+	for _, transfer := range esdtTransfers.ESDTTransfers {
+		if transfer.ESDTTokenNonce != 0 {
+			continue
+		}
+		if string(transfer.ESDTTokenName) != vmcommon.EGLDIdentifier {
+			// We only care about native transfers.
+			continue
+		}
+
+		_ = accumulatedNativeValue.Add(accumulatedNativeValue, transfer.ESDTValue)
+	}
+
+	return accumulatedNativeValue
+}
+
+// IsInterfaceNil returns true if there is no value under the interface
+func (host *mempoolHost) IsInterfaceNil() bool {
+	return host == nil
+}

dataRetriever/txpool/mempoolHost_test.go

@@ -0,0 +1,182 @@
+package txpool
+
+import (
+	"encoding/hex"
+	"fmt"
+	"math/big"
+	"testing"
+
+	"github.com/multiversx/mx-chain-core-go/core"
+	"github.com/multiversx/mx-chain-core-go/data/transaction"
+	"github.com/multiversx/mx-chain-core-go/marshal"
+	"github.com/multiversx/mx-chain-go/dataRetriever"
+	"github.com/multiversx/mx-chain-go/testscommon"
+	"github.com/multiversx/mx-chain-go/testscommon/txcachemocks"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewMempoolHost(t *testing.T) {
+	t.Parallel()
+
+	host, err := newMempoolHost(argsMempoolHost{
+		txGasHandler: nil,
+		marshalizer:  &marshal.GogoProtoMarshalizer{},
+	})
+	require.Nil(t, host)
+	require.ErrorIs(t, err, dataRetriever.ErrNilTxGasHandler)
+
+	host, err = newMempoolHost(argsMempoolHost{
+		txGasHandler: txcachemocks.NewTxGasHandlerMock(),
+		marshalizer:  nil,
+	})
+	require.Nil(t, host)
+	require.ErrorIs(t, err, dataRetriever.ErrNilMarshalizer)
+
+	host, err = newMempoolHost(argsMempoolHost{
+		txGasHandler: txcachemocks.NewTxGasHandlerMock(),
+		marshalizer:  &marshal.GogoProtoMarshalizer{},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, host)
+}
+
+func TestMempoolHost_GetTransferredValue(t *testing.T) {
+	t.Parallel()
+
+	host, err := newMempoolHost(argsMempoolHost{
+		txGasHandler: txcachemocks.NewTxGasHandlerMock(),
+		marshalizer:  &marshal.GogoProtoMarshalizer{},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, host)
+
+	t.Run("with value", func(t *testing.T) {
+		value := host.GetTransferredValue(&transaction.Transaction{
+			Value: big.NewInt(1000000000000000000),
+		})
+		require.Equal(t, big.NewInt(1000000000000000000), value)
+	})
+
+	t.Run("with value and data", func(t *testing.T) {
+		value := host.GetTransferredValue(&transaction.Transaction{
+			Value: big.NewInt(1000000000000000000),
+			Data:  []byte("data"),
+		})
+		require.Equal(t, big.NewInt(1000000000000000000), value)
+	})
+
+	t.Run("native transfer within MultiESDTNFTTransfer", func(t *testing.T) {
+		value := host.GetTransferredValue(&transaction.Transaction{
+			SndAddr: testscommon.TestPubKeyAlice,
+			RcvAddr: testscommon.TestPubKeyAlice,
+			Data:    []byte("MultiESDTNFTTransfer@8049d639e5a6980d1cd2392abcce41029cda74a1563523a202f09641cc2618f8@03@4e46542d313233343536@0a@01@544553542d393837363534@01@01@45474c442d303030303030@@0de0b6b3a7640000"),
+		})
+		require.Equal(t, big.NewInt(1000000000000000000), value)
+	})
+
+	t.Run("native transfer within MultiESDTNFTTransfer; transfer & execute", func(t *testing.T) {
+		value := host.GetTransferredValue(&transaction.Transaction{
+			SndAddr: testscommon.TestPubKeyAlice,
+			RcvAddr: testscommon.TestPubKeyAlice,
+			Data:    []byte("MultiESDTNFTTransfer@00000000000000000500b9353fe8407f87310c87e12fa1ac807f0485da39d152@03@4e46542d313233343536@01@01@4e46542d313233343536@2a@01@45474c442d303030303030@@0de0b6b3a7640000@64756d6d79@07"),
+		})
+		require.Equal(t, big.NewInt(1000000000000000000), value)
+	})
+}
+
+func TestBenchmarkMempoolHost_GetTransferredValue(t *testing.T) {
+	host, err := newMempoolHost(argsMempoolHost{
+		txGasHandler: txcachemocks.NewTxGasHandlerMock(),
+		marshalizer:  &marshal.GogoProtoMarshalizer{},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, host)
+
+	sw := core.NewStopWatch()
+
+	valueMultiplier := int64(1_000_000_000_000)
+
+	t.Run("numTransactions = 5_000", func(t *testing.T) {
+		numTransactions := 5_000
+		transactions := createMultiESDTNFTTransfersWithNativeTransfer(numTransactions, valueMultiplier)
+
+		sw.Start(t.Name())
+
+		for i := 0; i < numTransactions; i++ {
+			tx := transactions[i]
+			value := host.GetTransferredValue(tx)
+			require.Equal(t, big.NewInt(int64(i)*valueMultiplier), value)
+		}
+
+		sw.Stop(t.Name())
+	})
+
+	t.Run("numTransactions = 10_000", func(t *testing.T) {
+		numTransactions := 10_000
+		transactions := createMultiESDTNFTTransfersWithNativeTransfer(numTransactions, valueMultiplier)
+
+		sw.Start(t.Name())
+
+		for i := 0; i < numTransactions; i++ {
+			tx := transactions[i]
+			value := host.GetTransferredValue(tx)
+			require.Equal(t, big.NewInt(int64(i)*valueMultiplier), value)
+		}
+
+		sw.Stop(t.Name())
+	})
+
+	t.Run("numTransactions = 20_000", func(t *testing.T) {
+		numTransactions := 20_000
+		transactions := createMultiESDTNFTTransfersWithNativeTransfer(numTransactions, valueMultiplier)
+
+		sw.Start(t.Name())
+
+		for i := 0; i < numTransactions; i++ {
+			tx := transactions[i]
+			value := host.GetTransferredValue(tx)
+			require.Equal(t, big.NewInt(int64(i)*valueMultiplier), value)
+		}
+
+		sw.Stop(t.Name())
+	})
+
+	for name, measurement := range sw.GetMeasurementsMap() {
+		fmt.Printf("%fs (%s)\n", measurement, name)
+	}
+
+	// (1)
+	// Vendor ID:                GenuineIntel
+	//   Model name:             11th Gen Intel(R) Core(TM) i7-1165G7 @ 2.80GHz
+	//     CPU family:           6
+	//     Model:                140
+	//     Thread(s) per core:   2
+	//     Core(s) per socket:   4
+	//
+	// NOTE: 20% is also due to the require() / assert() calls.
+	// 0.012993s (TestBenchmarkMempoolHost_GetTransferredValue/numTransactions_=_5_000)
+	// 0.024580s (TestBenchmarkMempoolHost_GetTransferredValue/numTransactions_=_10_000)
+	// 0.048808s (TestBenchmarkMempoolHost_GetTransferredValue/numTransactions_=_20_000)
+}
+
+func createMultiESDTNFTTransfersWithNativeTransfer(numTransactions int, valueMultiplier int64) []*transaction.Transaction {
+	transactions := make([]*transaction.Transaction, 0, numTransactions)
+
+	for i := 0; i < numTransactions; i++ {
+		nativeValue := big.NewInt(int64(i) * valueMultiplier)
+		data := fmt.Sprintf(
+			"MultiESDTNFTTransfer@8049d639e5a6980d1cd2392abcce41029cda74a1563523a202f09641cc2618f8@03@4e46542d313233343536@0a@01@544553542d393837363534@01@01@45474c442d303030303030@@%s",
+			hex.EncodeToString(nativeValue.Bytes()),
+		)
+
+		tx := &transaction.Transaction{
+			SndAddr: testscommon.TestPubKeyAlice,
+			RcvAddr: testscommon.TestPubKeyAlice,
+			Data:    []byte(data),
+		}
+
+		transactions = append(transactions, tx)
+	}
+
+	return transactions
+}