core: types: less allocations when hashing and tx handling (#21265)

* core, crypto: various allocation savings regarding tx handling

* core: reduce allocs for gas price comparison

This change reduces the allocations needed for comparing different transactions to each other.
A call to `tx.GasPrice()` copies the gas price as it has to be safe against modifications and
also needs to be threadsafe. For comparing and ordering different transactions we don't need
these guarantees

* core: added tx.GasPriceIntCmp for comparison without allocation

adds a method to remove unneeded allocation in comparison to tx.gasPrice

* core/types: pool legacykeccak256 objects in rlpHash

rlpHash is by far the most used function in core that allocates a legacyKeccak256 object on each call.
Since it is so widely used it makes sense to add pooling here so we relieve the GC.
On my machine these changes result in > 100 MILLION less allocations and > 30 GB less allocated memory.

* reverted some changes

* reverted some changes

* trie: use crypto.KeccakState instead of replicating code

Co-authored-by: Martin Holst Swende <martin@swende.se>
This commit is contained in:
Marius van der Wijden 2020-06-30 11:59:06 +02:00 committed by GitHub
parent e376d2fb31
commit ddeea1e0c6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 72 additions and 32 deletions

View File

@ -256,7 +256,7 @@ func (l *txList) Add(tx *types.Transaction, priceBump uint64) (bool, *types.Tran
// Have to ensure that the new gas price is higher than the old gas // Have to ensure that the new gas price is higher than the old gas
// price as well as checking the percentage threshold to ensure that // price as well as checking the percentage threshold to ensure that
// this is accurate for low (Wei-level) gas price replacements // this is accurate for low (Wei-level) gas price replacements
if old.GasPrice().Cmp(tx.GasPrice()) >= 0 || threshold.Cmp(tx.GasPrice()) > 0 { if old.GasPriceCmp(tx) >= 0 || tx.GasPriceIntCmp(threshold) < 0 {
return false, nil return false, nil
} }
} }
@ -372,7 +372,7 @@ func (h priceHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h priceHeap) Less(i, j int) bool { func (h priceHeap) Less(i, j int) bool {
// Sort primarily by price, returning the cheaper one // Sort primarily by price, returning the cheaper one
switch h[i].GasPrice().Cmp(h[j].GasPrice()) { switch h[i].GasPriceCmp(h[j]) {
case -1: case -1:
return true return true
case 1: case 1:
@ -449,7 +449,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact
continue continue
} }
// Stop the discards if we've reached the threshold // Stop the discards if we've reached the threshold
if tx.GasPrice().Cmp(threshold) >= 0 { if tx.GasPriceIntCmp(threshold) >= 0 {
save = append(save, tx) save = append(save, tx)
break break
} }
@ -489,7 +489,7 @@ func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) boo
return false return false
} }
cheapest := []*types.Transaction(*l.items)[0] cheapest := []*types.Transaction(*l.items)[0]
return cheapest.GasPrice().Cmp(tx.GasPrice()) >= 0 return cheapest.GasPriceCmp(tx) >= 0
} }
// Discard finds a number of most underpriced transactions, removes them from the // Discard finds a number of most underpriced transactions, removes them from the

View File

@ -17,6 +17,7 @@
package core package core
import ( import (
"math/big"
"math/rand" "math/rand"
"testing" "testing"
@ -49,3 +50,21 @@ func TestStrictTxListAdd(t *testing.T) {
} }
} }
} }
func BenchmarkTxListAdd(t *testing.B) {
// Generate a list of transactions to insert
key, _ := crypto.GenerateKey()
txs := make(types.Transactions, 100000)
for i := 0; i < len(txs); i++ {
txs[i] = transaction(uint64(i), 0, key)
}
// Insert the transactions in a random order
list := newTxList(true)
priceLimit := big.NewInt(int64(DefaultTxPoolConfig.PriceLimit))
t.ResetTimer()
for _, v := range rand.Perm(len(txs)) {
list.Add(txs[v], DefaultTxPoolConfig.PriceBump)
list.Filter(priceLimit, DefaultTxPoolConfig.PriceBump)
}
}

View File

@ -534,7 +534,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
} }
// Drop non-local transactions under our own minimal accepted gas price // Drop non-local transactions under our own minimal accepted gas price
local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network
if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 { if !local && tx.GasPriceIntCmp(pool.gasPrice) < 0 {
return ErrUnderpriced return ErrUnderpriced
} }
// Ensure the transaction adheres to nonce ordering // Ensure the transaction adheres to nonce ordering
@ -1187,15 +1187,15 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
for _, tx := range forwards { for _, tx := range forwards {
hash := tx.Hash() hash := tx.Hash()
pool.all.Remove(hash) pool.all.Remove(hash)
log.Trace("Removed old queued transaction", "hash", hash)
} }
log.Trace("Removed old queued transactions", "count", len(forwards))
// Drop all transactions that are too costly (low balance or out of gas) // Drop all transactions that are too costly (low balance or out of gas)
drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas) drops, _ := list.Filter(pool.currentState.GetBalance(addr), pool.currentMaxGas)
for _, tx := range drops { for _, tx := range drops {
hash := tx.Hash() hash := tx.Hash()
pool.all.Remove(hash) pool.all.Remove(hash)
log.Trace("Removed unpayable queued transaction", "hash", hash)
} }
log.Trace("Removed unpayable queued transactions", "count", len(drops))
queuedNofundsMeter.Mark(int64(len(drops))) queuedNofundsMeter.Mark(int64(len(drops)))
// Gather all executable transactions and promote them // Gather all executable transactions and promote them
@ -1203,10 +1203,10 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) []*types.Trans
for _, tx := range readies { for _, tx := range readies {
hash := tx.Hash() hash := tx.Hash()
if pool.promoteTx(addr, hash, tx) { if pool.promoteTx(addr, hash, tx) {
log.Trace("Promoting queued transaction", "hash", hash)
promoted = append(promoted, tx) promoted = append(promoted, tx)
} }
} }
log.Trace("Promoted queued transactions", "count", len(promoted))
queuedGauge.Dec(int64(len(readies))) queuedGauge.Dec(int64(len(readies)))
// Drop all transactions over the allowed limit // Drop all transactions over the allowed limit

View File

@ -23,11 +23,13 @@ import (
"io" "io"
"math/big" "math/big"
"reflect" "reflect"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rlp"
"golang.org/x/crypto/sha3" "golang.org/x/crypto/sha3"
) )
@ -129,10 +131,19 @@ func (h *Header) SanityCheck() error {
return nil return nil
} }
// hasherPool holds LegacyKeccak hashers.
var hasherPool = sync.Pool{
New: func() interface{} {
return sha3.NewLegacyKeccak256()
},
}
func rlpHash(x interface{}) (h common.Hash) { func rlpHash(x interface{}) (h common.Hash) {
hw := sha3.NewLegacyKeccak256() sha := hasherPool.Get().(crypto.KeccakState)
rlp.Encode(hw, x) defer hasherPool.Put(sha)
hw.Sum(h[:0]) sha.Reset()
rlp.Encode(sha, x)
sha.Read(h[:])
return h return h
} }

View File

@ -175,9 +175,15 @@ func (tx *Transaction) UnmarshalJSON(input []byte) error {
func (tx *Transaction) Data() []byte { return common.CopyBytes(tx.data.Payload) } func (tx *Transaction) Data() []byte { return common.CopyBytes(tx.data.Payload) }
func (tx *Transaction) Gas() uint64 { return tx.data.GasLimit } func (tx *Transaction) Gas() uint64 { return tx.data.GasLimit }
func (tx *Transaction) GasPrice() *big.Int { return new(big.Int).Set(tx.data.Price) } func (tx *Transaction) GasPrice() *big.Int { return new(big.Int).Set(tx.data.Price) }
func (tx *Transaction) Value() *big.Int { return new(big.Int).Set(tx.data.Amount) } func (tx *Transaction) GasPriceCmp(other *Transaction) int {
func (tx *Transaction) Nonce() uint64 { return tx.data.AccountNonce } return tx.data.Price.Cmp(other.data.Price)
func (tx *Transaction) CheckNonce() bool { return true } }
func (tx *Transaction) GasPriceIntCmp(other *big.Int) int {
return tx.data.Price.Cmp(other)
}
func (tx *Transaction) Value() *big.Int { return new(big.Int).Set(tx.data.Amount) }
func (tx *Transaction) Nonce() uint64 { return tx.data.AccountNonce }
func (tx *Transaction) CheckNonce() bool { return true }
// To returns the recipient address of the transaction. // To returns the recipient address of the transaction.
// It returns nil if the transaction is a contract creation. // It returns nil if the transaction is a contract creation.

View File

@ -24,6 +24,7 @@ import (
"encoding/hex" "encoding/hex"
"errors" "errors"
"fmt" "fmt"
"hash"
"io" "io"
"io/ioutil" "io/ioutil"
"math/big" "math/big"
@ -51,23 +52,33 @@ var (
var errInvalidPubkey = errors.New("invalid secp256k1 public key") var errInvalidPubkey = errors.New("invalid secp256k1 public key")
// KeccakState wraps sha3.state. In addition to the usual hash methods, it also supports
// Read to get a variable amount of data from the hash state. Read is faster than Sum
// because it doesn't copy the internal state, but also modifies the internal state.
type KeccakState interface {
hash.Hash
Read([]byte) (int, error)
}
// Keccak256 calculates and returns the Keccak256 hash of the input data. // Keccak256 calculates and returns the Keccak256 hash of the input data.
func Keccak256(data ...[]byte) []byte { func Keccak256(data ...[]byte) []byte {
d := sha3.NewLegacyKeccak256() b := make([]byte, 32)
d := sha3.NewLegacyKeccak256().(KeccakState)
for _, b := range data { for _, b := range data {
d.Write(b) d.Write(b)
} }
return d.Sum(nil) d.Read(b)
return b
} }
// Keccak256Hash calculates and returns the Keccak256 hash of the input data, // Keccak256Hash calculates and returns the Keccak256 hash of the input data,
// converting it to an internal Hash data structure. // converting it to an internal Hash data structure.
func Keccak256Hash(data ...[]byte) (h common.Hash) { func Keccak256Hash(data ...[]byte) (h common.Hash) {
d := sha3.NewLegacyKeccak256() d := sha3.NewLegacyKeccak256().(KeccakState)
for _, b := range data { for _, b := range data {
d.Write(b) d.Write(b)
} }
d.Sum(h[:0]) d.Read(h[:])
return h return h
} }

View File

@ -156,7 +156,7 @@ type transactionsByGasPrice []*types.Transaction
func (t transactionsByGasPrice) Len() int { return len(t) } func (t transactionsByGasPrice) Len() int { return len(t) }
func (t transactionsByGasPrice) Swap(i, j int) { t[i], t[j] = t[j], t[i] } func (t transactionsByGasPrice) Swap(i, j int) { t[i], t[j] = t[j], t[i] }
func (t transactionsByGasPrice) Less(i, j int) bool { return t[i].GasPrice().Cmp(t[j].GasPrice()) < 0 } func (t transactionsByGasPrice) Less(i, j int) bool { return t[i].GasPriceCmp(t[j]) < 0 }
// getBlockPrices calculates the lowest transaction gas price in a given block // getBlockPrices calculates the lowest transaction gas price in a given block
// and sends it to the result channel. If the block is empty, price is nil. // and sends it to the result channel. If the block is empty, price is nil.

View File

@ -22,6 +22,7 @@ import (
"sync" "sync"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rlp"
"golang.org/x/crypto/sha3" "golang.org/x/crypto/sha3"
) )
@ -46,7 +47,7 @@ type leaf struct {
// processed sequentially - onleaf will never be called in parallel or out of order. // processed sequentially - onleaf will never be called in parallel or out of order.
type committer struct { type committer struct {
tmp sliceBuffer tmp sliceBuffer
sha keccakState sha crypto.KeccakState
onleaf LeafCallback onleaf LeafCallback
leafCh chan *leaf leafCh chan *leaf
@ -57,7 +58,7 @@ var committerPool = sync.Pool{
New: func() interface{} { New: func() interface{} {
return &committer{ return &committer{
tmp: make(sliceBuffer, 0, 550), // cap is as large as a full fullNode. tmp: make(sliceBuffer, 0, 550), // cap is as large as a full fullNode.
sha: sha3.NewLegacyKeccak256().(keccakState), sha: sha3.NewLegacyKeccak256().(crypto.KeccakState),
} }
}, },
} }

View File

@ -17,21 +17,13 @@
package trie package trie
import ( import (
"hash"
"sync" "sync"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rlp"
"golang.org/x/crypto/sha3" "golang.org/x/crypto/sha3"
) )
// keccakState wraps sha3.state. In addition to the usual hash methods, it also supports
// Read to get a variable amount of data from the hash state. Read is faster than Sum
// because it doesn't copy the internal state, but also modifies the internal state.
type keccakState interface {
hash.Hash
Read([]byte) (int, error)
}
type sliceBuffer []byte type sliceBuffer []byte
func (b *sliceBuffer) Write(data []byte) (n int, err error) { func (b *sliceBuffer) Write(data []byte) (n int, err error) {
@ -46,7 +38,7 @@ func (b *sliceBuffer) Reset() {
// hasher is a type used for the trie Hash operation. A hasher has some // hasher is a type used for the trie Hash operation. A hasher has some
// internal preallocated temp space // internal preallocated temp space
type hasher struct { type hasher struct {
sha keccakState sha crypto.KeccakState
tmp sliceBuffer tmp sliceBuffer
parallel bool // Whether to use paralallel threads when hashing parallel bool // Whether to use paralallel threads when hashing
} }
@ -56,7 +48,7 @@ var hasherPool = sync.Pool{
New: func() interface{} { New: func() interface{} {
return &hasher{ return &hasher{
tmp: make(sliceBuffer, 0, 550), // cap is as large as a full fullNode. tmp: make(sliceBuffer, 0, 550), // cap is as large as a full fullNode.
sha: sha3.NewLegacyKeccak256().(keccakState), sha: sha3.NewLegacyKeccak256().(crypto.KeccakState),
} }
}, },
} }