Merge pull request #73 from wemixarchive/api-performance

merge v0.10.7 to master

Author: cp-jhjin

cmd/geth/main.go

@@ -225,6 +225,9 @@ var (
 		utils.BlockMinBuildTime,
 		utils.BlockMinBuildTxs,
 		utils.BlockTrailTime,
+		utils.PublicRequestsCacheLocation,
+		utils.MaxPublicRequests,
+		utils.BootnodeCount,
 	}
 )
 

cmd/geth/usage.go

@@ -235,6 +235,9 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.BlockMinBuildTime,
 			utils.BlockMinBuildTxs,
 			utils.BlockTrailTime,
+			utils.PublicRequestsCacheLocation,
+			utils.MaxPublicRequests,
+			utils.BootnodeCount,
 		},
 	},
 	{

cmd/utils/flags.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"math"
 	"math/big"
+	"math/rand"
 	"os"
 	"path/filepath"
 	godebug "runtime/debug"
@@ -916,6 +917,21 @@ var (
 		Usage: "Time to leave for block data transfer in ms",
 		Value: params.BlockTrailTime,
 	}
+	PublicRequestsCacheLocation = cli.StringFlag{
+		Name:  "wemix.publicrequests.cache",
+		Usage: "Public requests cache location",
+		Value: params.PublicRequestsCacheLocation,
+	}
+	MaxPublicRequests = cli.Int64Flag{
+		Name:  "wemix.publicrequests.max",
+		Usage: "Max # of concurrent public requests",
+		Value: params.MaxPublicRequests,
+	}
+	BootnodeCount = cli.IntFlag{
+		Name:  "wemix.bootnodecount",
+		Usage: "Default bootnode peer count",
+		Value: params.BootnodeCount,
+	}
 )
 
 var (
@@ -1011,15 +1027,40 @@ func setNodeUserIdent(ctx *cli.Context, cfg *node.Config) {
 	}
 }
 
+// setRandomBootstrapNodes setting a random list of bootstrap nodes using the command line
+func setRandomBootstrapNodes(ctx *cli.Context, bootnodes []string) []string {
+	rand.Seed(time.Now().UnixNano())
+	bootnodeslen := len(bootnodes)
+
+	// check command line
+	if ctx.GlobalIsSet(BootnodeCount.Name) {
+		setcount := ctx.GlobalInt(BootnodeCount.Name)
+		if setcount > 0 && setcount <= bootnodeslen {
+			params.BootnodeCount = setcount
+		}
+	}
+	// select random bootnodes
+	selectcount := params.BootnodeCount
+	urls := make([]string, selectcount)
+	tempnode := make([]string, bootnodeslen)
+	copy(tempnode, bootnodes)
+	for i := 0; i < selectcount; i++ {
+		index := rand.Intn(len(tempnode))
+		urls = append(urls, tempnode[index])
+		tempnode = append(tempnode[:index], tempnode[index+1:]...)
+	}
+	return urls
+}
+
 // setBootstrapNodes creates a list of bootstrap nodes from the command line
 // flags, reverting to pre-configured ones if none have been specified.
 func setBootstrapNodes(ctx *cli.Context, cfg *p2p.Config) {
-	urls := params.WemixMainnetBootnodes
+	urls := setRandomBootstrapNodes(ctx, params.WemixMainnetBootnodes)
 	switch {
 	case ctx.GlobalIsSet(BootnodesFlag.Name):
 		urls = SplitAndTrim(ctx.GlobalString(BootnodesFlag.Name))
 	case ctx.GlobalBool(WemixTestnetFlag.Name):
-		urls = params.WemixTestnetBootnodes
+		urls = setRandomBootstrapNodes(ctx, params.WemixTestnetBootnodes)
 	case ctx.GlobalBool(RopstenFlag.Name):
 		urls = params.RopstenBootnodes
 	case ctx.GlobalBool(SepoliaFlag.Name):
@@ -1982,6 +2023,12 @@ func SetWemixConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
 	if ctx.GlobalIsSet(BlockTrailTime.Name) {
 		params.BlockTrailTime = ctx.GlobalInt64(BlockTrailTime.Name)
 	}
+	if ctx.GlobalIsSet(PublicRequestsCacheLocation.Name) {
+		params.PublicRequestsCacheLocation = ctx.GlobalString(PublicRequestsCacheLocation.Name)
+	}
+	if ctx.GlobalIsSet(MaxPublicRequests.Name) {
+		params.MaxPublicRequests = ctx.GlobalInt64(MaxPublicRequests.Name)
+	}
 
 	if params.ConsensusMethod == params.ConsensusInvalid {
 		params.ConsensusMethod = params.ConsensusPoW

eth/api.go

@@ -412,7 +412,7 @@ func (api *PrivateDebugAPI) GetBadBlocks(ctx context.Context) ([]*BadBlockArgs,
 		} else {
 			blockRlp = fmt.Sprintf("0x%x", rlpBytes)
 		}
-		if blockJSON, err = ethapi.RPCMarshalBlock(block, true, true, api.eth.APIBackend.ChainConfig()); err != nil {
+		if blockJSON, err = ethapi.RPCMarshalBlock(ctx, block, true, true, api.eth.APIBackend.ChainConfig()); err != nil {
 			blockJSON = map[string]interface{}{"error": err.Error()}
 		}
 		results = append(results, &BadBlockArgs{

eth/protocols/eth/peer.go

@@ -34,7 +34,7 @@ import (
 const (
 	// maxKnownTxs is the maximum transactions hashes to keep in the known list
 	// before starting to randomly evict them.
-	maxKnownTxs = 2000000
+	maxKnownTxs = 100000
 
 	// maxKnownBlocks is the maximum block hashes to keep in the known list
 	// before starting to randomly evict them.

internal/ethapi/api.go

@@ -20,9 +20,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"math/big"
 	"os"
+	"runtime"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/davecgh/go-spew/spew"
@@ -42,6 +45,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/crypto/vrf"
 	"github.com/ethereum/go-ethereum/eth/tracers/logger"
+	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/params"
@@ -50,6 +54,29 @@ import (
 	"github.com/tyler-smith/go-bip39"
 )
 
+var apiRequestsCache ethdb.Database
+var apiRequestsThrottle chan struct{}
+var apiRequestsTokens chan struct{}
+
+func apiRequestsEnter() {
+	if len(apiRequestsThrottle) >= int(params.MaxPublicRequests) {
+		pc, _, _, _ := runtime.Caller(1)
+		var name string
+		parts := strings.Split(runtime.FuncForPC(pc).Name(), ".")
+		if len(parts) > 0 {
+			name = parts[len(parts)-1]
+		} else {
+			name = runtime.FuncForPC(pc).Name()
+		}
+		log.Warn("Too many API requests", "func", name, "count", len(apiRequestsThrottle))
+	}
+	apiRequestsThrottle <- struct{}{}
+}
+
+func apiRequestsLeave() {
+	<-apiRequestsThrottle
+}
+
 // PublicEthereumAPI provides an API to access Ethereum related information.
 // It offers only methods that operate on public data that is freely available to anyone.
 type PublicEthereumAPI struct {
@@ -681,6 +708,20 @@ type PublicBlockChainAPI struct {
 
 // NewPublicBlockChainAPI creates a new Ethereum blockchain API.
 func NewPublicBlockChainAPI(b Backend) *PublicBlockChainAPI {
+	if len(params.PublicRequestsCacheLocation) > 0 {
+		var err error
+		apiRequestsCache, err = apiCacheOpen(params.PublicRequestsCacheLocation)
+		if err != nil {
+			panic(err)
+		}
+	}
+	apiRequestsThrottle = make(chan struct{}, params.MaxPublicRequests)
+	tokens := runtime.NumCPU() * 8 / 10
+	if tokens < 4 {
+		tokens = 4
+	}
+	apiRequestsTokens = make(chan struct{}, tokens)
+
 	return &PublicBlockChainAPI{b}
 }
 
@@ -701,6 +742,21 @@ func (s *PublicBlockChainAPI) BlockNumber() hexutil.Uint64 {
 
 // GetBlockReceipts returns all the transaction receipts for the given block hash.
 func (s *PublicBlockChainAPI) GetReceiptsByHash(ctx context.Context, blockHash common.Hash) ([]map[string]interface{}, error) {
+	apiRequestsEnter()
+	defer apiRequestsLeave()
+
+	select {
+	case <-ctx.Done():
+		return nil, io.EOF
+	default:
+	}
+
+	if apiRequestsCache != nil {
+		if fields, err := apiCacheGetReceipts(apiRequestsCache, blockHash.Bytes()); err == nil {
+			log.Debug("API Cache", "found receipts", blockHash)
+			return fields, nil
+		}
+	}
 
 	block, err1 := s.b.BlockByHash(ctx, blockHash)
 	if block == nil && err1 == nil {
@@ -723,6 +779,11 @@ func (s *PublicBlockChainAPI) GetReceiptsByHash(ctx context.Context, blockHash c
 	fieldsList := make([]map[string]interface{}, 0, len(receipts))
 
 	for index, receipt := range receipts {
+		select {
+		case <-ctx.Done():
+			return nil, io.EOF
+		default:
+		}
 
 		bigblock := new(big.Int).SetUint64(block.NumberU64())
 		signer := types.MakeSigner(s.b.ChainConfig(), bigblock)
@@ -770,6 +831,9 @@ func (s *PublicBlockChainAPI) GetReceiptsByHash(ctx context.Context, blockHash c
 
 		fieldsList = append(fieldsList, fields)
 	}
+	if apiRequestsCache != nil {
+		apiCachePutReceipts(apiRequestsCache, blockHash.Bytes(), fieldsList)
+	}
 	return fieldsList, nil
 }
 
@@ -884,6 +948,9 @@ func (s *PublicBlockChainAPI) GetHeaderByHash(ctx context.Context, hash common.H
 //   - When fullTx is true all transactions in the block are returned, otherwise
 //     only the transaction hash is returned.
 func (s *PublicBlockChainAPI) GetBlockByNumber(ctx context.Context, number rpc.BlockNumber, fullTx bool) (map[string]interface{}, error) {
+	apiRequestsEnter()
+	defer apiRequestsLeave()
+
 	block, err := s.b.BlockByNumber(ctx, number)
 	if block != nil && err == nil {
 		response, err := s.rpcMarshalBlock(ctx, block, true, fullTx)
@@ -901,6 +968,9 @@ func (s *PublicBlockChainAPI) GetBlockByNumber(ctx context.Context, number rpc.B
 // GetBlockByHash returns the requested block. When fullTx is true all transactions in the block are returned in full
 // detail, otherwise only the transaction hash is returned.
 func (s *PublicBlockChainAPI) GetBlockByHash(ctx context.Context, hash common.Hash, fullTx bool) (map[string]interface{}, error) {
+	apiRequestsEnter()
+	defer apiRequestsLeave()
+
 	block, err := s.b.BlockByHash(ctx, hash)
 	if block != nil {
 		return s.rpcMarshalBlock(ctx, block, true, fullTx)
@@ -910,6 +980,9 @@ func (s *PublicBlockChainAPI) GetBlockByHash(ctx context.Context, hash common.Ha
 
 // GetUncleByBlockNumberAndIndex returns the uncle block for the given block hash and index.
 func (s *PublicBlockChainAPI) GetUncleByBlockNumberAndIndex(ctx context.Context, blockNr rpc.BlockNumber, index hexutil.Uint) (map[string]interface{}, error) {
+	apiRequestsEnter()
+	defer apiRequestsLeave()
+
 	block, err := s.b.BlockByNumber(ctx, blockNr)
 	if block != nil {
 		uncles := block.Uncles()
@@ -925,6 +998,9 @@ func (s *PublicBlockChainAPI) GetUncleByBlockNumberAndIndex(ctx context.Context,
 
 // GetUncleByBlockHashAndIndex returns the uncle block for the given block hash and index.
 func (s *PublicBlockChainAPI) GetUncleByBlockHashAndIndex(ctx context.Context, blockHash common.Hash, index hexutil.Uint) (map[string]interface{}, error) {
+	apiRequestsEnter()
+	defer apiRequestsLeave()
+
 	block, err := s.b.BlockByHash(ctx, blockHash)
 	if block != nil {
 		uncles := block.Uncles()
@@ -940,6 +1016,9 @@ func (s *PublicBlockChainAPI) GetUncleByBlockHashAndIndex(ctx context.Context, b
 
 // GetUncleCountByBlockNumber returns number of uncles in the block for the given block number
 func (s *PublicBlockChainAPI) GetUncleCountByBlockNumber(ctx context.Context, blockNr rpc.BlockNumber) *hexutil.Uint {
+	apiRequestsEnter()
+	defer apiRequestsLeave()
+
 	if block, _ := s.b.BlockByNumber(ctx, blockNr); block != nil {
 		n := hexutil.Uint(len(block.Uncles()))
 		return &n
@@ -949,6 +1028,9 @@ func (s *PublicBlockChainAPI) GetUncleCountByBlockNumber(ctx context.Context, bl
 
 // GetUncleCountByBlockHash returns number of uncles in the block for the given block hash
 func (s *PublicBlockChainAPI) GetUncleCountByBlockHash(ctx context.Context, blockHash common.Hash) *hexutil.Uint {
+	apiRequestsEnter()
+	defer apiRequestsLeave()
+
 	if block, _ := s.b.BlockByHash(ctx, blockHash); block != nil {
 		n := hexutil.Uint(len(block.Uncles()))
 		return &n
@@ -1299,7 +1381,20 @@ func RPCMarshalHeader(head *types.Header) map[string]interface{} {
 // RPCMarshalBlock converts the given block to the RPC output which depends on fullTx. If inclTx is true transactions are
 // returned. When fullTx is true the returned block contains full transaction details, otherwise it will only contain
 // transaction hashes.
-func RPCMarshalBlock(block *types.Block, inclTx bool, fullTx bool, config *params.ChainConfig) (map[string]interface{}, error) {
+func RPCMarshalBlock(ctx context.Context, block *types.Block, inclTx bool, fullTx bool, config *params.ChainConfig) (map[string]interface{}, error) {
+	select {
+	case <-ctx.Done():
+		return nil, io.EOF
+	default:
+	}
+
+	if fullTx && apiRequestsCache != nil {
+		if fields, err := apiCacheGetBlock(apiRequestsCache, block.Hash().Bytes()); err == nil {
+			log.Debug("API Cache", "found block", block.Number())
+			return fields, nil
+		}
+	}
+
 	fields := RPCMarshalHeader(block.Header())
 	fields["size"] = hexutil.Uint64(block.Size())
 
@@ -1314,11 +1409,33 @@ func RPCMarshalBlock(block *types.Block, inclTx bool, fullTx bool, config *param
 		}
 		txs := block.Transactions()
 		transactions := make([]interface{}, len(txs))
+		var wg sync.WaitGroup
 		var err error
 		for i, tx := range txs {
-			if transactions[i], err = formatTx(tx); err != nil {
-				return nil, err
-			}
+			wg.Add(1)
+			go func(ii int, itx *types.Transaction) {
+				apiRequestsTokens <- struct{}{}
+				defer func() {
+					wg.Done()
+					<-apiRequestsTokens
+				}()
+
+				select {
+				case <-ctx.Done():
+					err = io.EOF
+					return
+				default:
+				}
+				var err2 error
+				transactions[ii], err2 = formatTx(itx)
+				if err2 != nil {
+					err = err2
+				}
+			}(i, tx)
+		}
+		wg.Wait()
+		if err != nil {
+			return nil, err
 		}
 		fields["transactions"] = transactions
 	}
@@ -1329,6 +1446,10 @@ func RPCMarshalBlock(block *types.Block, inclTx bool, fullTx bool, config *param
 	}
 	fields["uncles"] = uncleHashes
 
+	if fullTx && apiRequestsCache != nil {
+		apiCachePutBlock(apiRequestsCache, block.Hash().Bytes(), fields)
+	}
+
 	return fields, nil
 }
 
@@ -1343,7 +1464,7 @@ func (s *PublicBlockChainAPI) rpcMarshalHeader(ctx context.Context, header *type
 // rpcMarshalBlock uses the generalized output filler, then adds the total difficulty field, which requires
 // a `PublicBlockchainAPI`.
 func (s *PublicBlockChainAPI) rpcMarshalBlock(ctx context.Context, b *types.Block, inclTx bool, fullTx bool) (map[string]interface{}, error) {
-	fields, err := RPCMarshalBlock(b, inclTx, fullTx, s.b.ChainConfig())
+	fields, err := RPCMarshalBlock(ctx, b, inclTx, fullTx, s.b.ChainConfig())
 	if err != nil {
 		return nil, err
 	}