Arpit/add execution pool 2

builder/files/config.toml

@@ -75,13 +75,17 @@ syncmode = "full"
 #     api = ["eth", "net", "web3", "txpool", "bor"]
 #     vhosts = ["*"]
 #     corsdomain = ["*"]
+#     ep-size = 40
+#     ep-requesttimeout = "0s"
 #   [jsonrpc.ws]
 #     enabled = false
 #     port = 8546
 #     prefix = ""
 #     host = "localhost"
 #     api = ["web3", "net"]
 #     origins = ["*"]
+#     ep-size = 40
+#     ep-requesttimeout = "0s"
 #   [jsonrpc.graphql]
 #     enabled = false
 #     port = 0

cmd/clef/main.go

@@ -656,7 +656,7 @@ func signer(c *cli.Context) error {
 		vhosts := utils.SplitAndTrim(c.GlobalString(utils.HTTPVirtualHostsFlag.Name))
 		cors := utils.SplitAndTrim(c.GlobalString(utils.HTTPCORSDomainFlag.Name))
 
-		srv := rpc.NewServer()
+		srv := rpc.NewServer(0, 0)
 		err := node.RegisterApis(rpcAPI, []string{"account"}, srv, false)
 		if err != nil {
 			utils.Fatalf("Could not register API: %w", err)

docs/cli/example_config.toml

@@ -74,25 +74,30 @@ ethstats = ""                # Reporting URL of a ethstats service (nodename:sec
     api = ["eth", "net", "web3", "txpool", "bor"]  # API's offered over the HTTP-RPC interface
     vhosts = ["localhost"]                         # Comma separated list of virtual hostnames from which to accept requests (server enforced). Accepts '*' wildcard.
     corsdomain = ["localhost"]                     # Comma separated list of domains from which to accept cross origin requests (browser enforced)
+    ep-size = 40                                   # Maximum size of workers to run in rpc execution pool for HTTP requests (default: 40)
+    ep-requesttimeout = "0s"                      # Request Timeout for rpc execution pool for HTTP requests (default: 0s, 0s = disabled)
   [jsonrpc.ws]
-    enabled = false          # Enable the WS-RPC server
-    port = 8546              # WS-RPC server listening port
-    prefix = ""              # HTTP path prefix on which JSON-RPC is served. Use '/' to serve on all paths.
-    host = "localhost"       # ws.addr
-    api = ["net", "web3"]    # API's offered over the WS-RPC interface
-    origins = ["localhost"]  # Origins from which to accept websockets requests
+    enabled = false             # Enable the WS-RPC server
+    port = 8546                 # WS-RPC server listening port
+    prefix = ""                 # HTTP path prefix on which JSON-RPC is served. Use '/' to serve on all paths.
+    host = "localhost"          # ws.addr
+    api = ["net", "web3"]       # API's offered over the WS-RPC interface
+    origins = ["localhost"]     # Origins from which to accept websockets requests
+    ep-size = 40                # Maximum size of workers to run in rpc execution pool for WS requests (default: 40)
+    ep-requesttimeout = "0s"   # Request Timeout for rpc execution pool for WS requests (default: 0s, 0s = disabled)
   [jsonrpc.graphql]
     enabled = false             # Enable GraphQL on the HTTP-RPC server. Note that GraphQL can only be started if an HTTP server is started as well.
-    port = 0                    # 
-    prefix = ""                 # 
-    host = ""                   # 
+    port = 0                    #
+    prefix = ""                 #
+    host = ""                   #
     vhosts = ["localhost"]      # Comma separated list of virtual hostnames from which to accept requests (server enforced). Accepts '*' wildcard.
     corsdomain = ["localhost"]  # Comma separated list of domains from which to accept cross origin requests (browser enforced)
   [jsonrpc.timeouts]
     read = "30s"
     write = "30s"
     idle = "2m0s"
 
+
 [gpo]
   blocks = 20                 # Number of recent blocks to check for gas prices
   percentile = 60             # Suggested gas price is the given percentile of a set of recent transaction gas prices

docs/cli/server.md

@@ -120,6 +120,10 @@ The ```bor server``` command runs the Bor client.
 
 - ```http.api```: API's offered over the HTTP-RPC interface (default: eth,net,web3,txpool,bor)
 
+- ```http.ep-size```: Maximum size of workers to run in rpc execution pool for HTTP requests (default: 40)
+
+- ```http.ep-requesttimeout```: Request Timeout for rpc execution pool for HTTP requests (default: 0s)
+
 - ```ws```: Enable the WS-RPC server (default: false)
 
 - ```ws.addr```: WS-RPC server listening interface (default: localhost)
@@ -130,6 +134,10 @@ The ```bor server``` command runs the Bor client.
 
 - ```ws.api```: API's offered over the WS-RPC interface (default: net,web3)
 
+- ```ws.ep-size```: Maximum size of workers to run in rpc execution pool for WS requests (default: 40)
+
+- ```ws.ep-requesttimeout```: Request Timeout for rpc execution pool for WS requests (default: 0s)
+
 - ```graphql```: Enable GraphQL on the HTTP-RPC server. Note that GraphQL can only be started if an HTTP server is started as well. (default: false)
 
 ### P2P Options

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.3.0
 	github.com/BurntSushi/toml v1.1.0
 	github.com/JekaMas/go-grpc-net-conn v0.0.0-20220708155319-6aff21f2d13d
+	github.com/JekaMas/workerpool v1.1.5
 	github.com/VictoriaMetrics/fastcache v1.6.0
 	github.com/aws/aws-sdk-go-v2 v1.2.0
 	github.com/aws/aws-sdk-go-v2/config v1.1.1
@@ -84,6 +85,8 @@ require (
 	pgregory.net/rapid v0.4.8
 )
 
+require github.com/gammazero/deque v0.2.1 // indirect
+
 require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v0.21.1 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v0.8.3 // indirect

go.sum

@@ -31,6 +31,8 @@ github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym
 github.com/DATA-DOG/go-sqlmock v1.3.3/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM=
 github.com/JekaMas/go-grpc-net-conn v0.0.0-20220708155319-6aff21f2d13d h1:RO27lgfZF8s9lZ3pWyzc0gCE0RZC+6/PXbRjAa0CNp8=
 github.com/JekaMas/go-grpc-net-conn v0.0.0-20220708155319-6aff21f2d13d/go.mod h1:romz7UPgSYhfJkKOalzEEyV6sWtt/eAEm0nX2aOrod0=
+github.com/JekaMas/workerpool v1.1.5 h1:xmrx2Zyft95CEGiEqzDxiawptCIRZQ0zZDhTGDFOCaw=
+github.com/JekaMas/workerpool v1.1.5/go.mod h1:IoDWPpwMcA27qbuugZKeBslDrgX09lVmksuh9sjzbhc=
 github.com/Masterminds/goutils v1.1.0 h1:zukEsf/1JZwCMgHiK3GZftabmxiCw4apj3a28RPBiVg=
 github.com/Masterminds/goutils v1.1.0/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU=
 github.com/Masterminds/semver v1.5.0 h1:H65muMkzWKEuNDnfl9d70GUjFniHKHRbFPGBuZ3QEww=
@@ -157,6 +159,8 @@ github.com/fogleman/gg v1.2.1-0.20190220221249-0403632d5b90/go.mod h1:R/bRT+9gY/
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
+github.com/gammazero/deque v0.2.1 h1:qSdsbG6pgp6nL7A0+K/B7s12mcCY/5l5SIUpMOl+dC0=
+github.com/gammazero/deque v0.2.1/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZDV34tuU=
 github.com/gballet/go-libpcsclite v0.0.0-20190607065134-2772fd86a8ff h1:tY80oXqGNY4FhTFhk+o9oFHGINQ/+vhlm8HFzi6znCI=
 github.com/gballet/go-libpcsclite v0.0.0-20190607065134-2772fd86a8ff/go.mod h1:x7DCsMOv1taUwEWCzT4cmDeAkigA5/QCwUodaVOe8Ww=
 github.com/getkin/kin-openapi v0.53.0/go.mod h1:7Yn5whZr5kJi6t+kShccXS8ae1APpYTW6yheSwk8Yi4=

internal/cli/dumpconfig.go

@@ -55,6 +55,8 @@ func (c *DumpconfigCommand) Run(args []string) int {
 	userConfig.JsonRPC.HttpTimeout.ReadTimeoutRaw = userConfig.JsonRPC.HttpTimeout.ReadTimeout.String()
 	userConfig.JsonRPC.HttpTimeout.WriteTimeoutRaw = userConfig.JsonRPC.HttpTimeout.WriteTimeout.String()
 	userConfig.JsonRPC.HttpTimeout.IdleTimeoutRaw = userConfig.JsonRPC.HttpTimeout.IdleTimeout.String()
+	userConfig.JsonRPC.Http.ExecutionPoolRequestTimeoutRaw = userConfig.JsonRPC.Http.ExecutionPoolRequestTimeout.String()
+	userConfig.JsonRPC.Ws.ExecutionPoolRequestTimeoutRaw = userConfig.JsonRPC.Ws.ExecutionPoolRequestTimeout.String()
 	userConfig.TxPool.RejournalRaw = userConfig.TxPool.Rejournal.String()
 	userConfig.TxPool.LifeTimeRaw = userConfig.TxPool.LifeTime.String()
 	userConfig.Sealer.GasPriceRaw = userConfig.Sealer.GasPrice.String()

internal/cli/server/config.go

@@ -281,6 +281,13 @@ type APIConfig struct {
 
 	// Origins is the list of endpoints to accept requests from (only consumed for websockets)
 	Origins []string `hcl:"origins,optional" toml:"origins,optional"`
+
+	// ExecutionPoolSize is max size of workers to be used for rpc execution
+	ExecutionPoolSize uint64 `hcl:"ep-size,optional" toml:"ep-size,optional"`
+
+	// ExecutionPoolRequestTimeout is timeout used by execution pool for rpc execution
+	ExecutionPoolRequestTimeout    time.Duration `hcl:"-,optional" toml:"-"`
+	ExecutionPoolRequestTimeoutRaw string        `hcl:"ep-requesttimeout,optional" toml:"ep-requesttimeout,optional"`
 }
 
 // Used from rpc.HTTPTimeouts
@@ -507,21 +514,25 @@ func DefaultConfig() *Config {
 			GasCap:     ethconfig.Defaults.RPCGasCap,
 			TxFeeCap:   ethconfig.Defaults.RPCTxFeeCap,
 			Http: &APIConfig{
-				Enabled: false,
-				Port:    8545,
-				Prefix:  "",
-				Host:    "localhost",
-				API:     []string{"eth", "net", "web3", "txpool", "bor"},
-				Cors:    []string{"localhost"},
-				VHost:   []string{"localhost"},
+				Enabled:                     false,
+				Port:                        8545,
+				Prefix:                      "",
+				Host:                        "localhost",
+				API:                         []string{"eth", "net", "web3", "txpool", "bor"},
+				Cors:                        []string{"localhost"},
+				VHost:                       []string{"localhost"},
+				ExecutionPoolSize:           40,
+				ExecutionPoolRequestTimeout: 0,
 			},
 			Ws: &APIConfig{
-				Enabled: false,
-				Port:    8546,
-				Prefix:  "",
-				Host:    "localhost",
-				API:     []string{"net", "web3"},
-				Origins: []string{"localhost"},
+				Enabled:                     false,
+				Port:                        8546,
+				Prefix:                      "",
+				Host:                        "localhost",
+				API:                         []string{"net", "web3"},
+				Origins:                     []string{"localhost"},
+				ExecutionPoolSize:           40,
+				ExecutionPoolRequestTimeout: 0,
 			},
 			Graphql: &APIConfig{
 				Enabled: false,
@@ -628,6 +639,8 @@ func (c *Config) fillTimeDurations() error {
 		{"jsonrpc.timeouts.read", &c.JsonRPC.HttpTimeout.ReadTimeout, &c.JsonRPC.HttpTimeout.ReadTimeoutRaw},
 		{"jsonrpc.timeouts.write", &c.JsonRPC.HttpTimeout.WriteTimeout, &c.JsonRPC.HttpTimeout.WriteTimeoutRaw},
 		{"jsonrpc.timeouts.idle", &c.JsonRPC.HttpTimeout.IdleTimeout, &c.JsonRPC.HttpTimeout.IdleTimeoutRaw},
+		{"jsonrpc.ws.ep-requesttimeout", &c.JsonRPC.Ws.ExecutionPoolRequestTimeout, &c.JsonRPC.Ws.ExecutionPoolRequestTimeoutRaw},
+		{"jsonrpc.http.ep-requesttimeout", &c.JsonRPC.Http.ExecutionPoolRequestTimeout, &c.JsonRPC.Http.ExecutionPoolRequestTimeoutRaw},
 		{"txpool.lifetime", &c.TxPool.LifeTime, &c.TxPool.LifeTimeRaw},
 		{"txpool.rejournal", &c.TxPool.Rejournal, &c.TxPool.RejournalRaw},
 		{"cache.rejournal", &c.Cache.Rejournal, &c.Cache.RejournalRaw},
@@ -997,7 +1010,11 @@ func (c *Config) buildNode() (*node.Config, error) {
 			WriteTimeout: c.JsonRPC.HttpTimeout.WriteTimeout,
 			IdleTimeout:  c.JsonRPC.HttpTimeout.IdleTimeout,
 		},
-		RPCBatchLimit: c.RPCBatchLimit,
+		RPCBatchLimit:                          c.RPCBatchLimit,
+		WSJsonRPCExecutionPoolSize:             c.JsonRPC.Ws.ExecutionPoolSize,
+		WSJsonRPCExecutionPoolRequestTimeout:   c.JsonRPC.Ws.ExecutionPoolRequestTimeout,
+		HTTPJsonRPCExecutionPoolSize:           c.JsonRPC.Http.ExecutionPoolSize,
+		HTTPJsonRPCExecutionPoolRequestTimeout: c.JsonRPC.Http.ExecutionPoolRequestTimeout,
 	}
 
 	// dev mode

internal/cli/server/flags.go

@@ -444,6 +444,20 @@ func (c *Command) Flags() *flagset.Flagset {
 		Default: c.cliConfig.JsonRPC.Http.API,
 		Group:   "JsonRPC",
 	})
+	f.Uint64Flag(&flagset.Uint64Flag{
+		Name:    "http.ep-size",
+		Usage:   "Maximum size of workers to run in rpc execution pool for HTTP requests",
+		Value:   &c.cliConfig.JsonRPC.Http.ExecutionPoolSize,
+		Default: c.cliConfig.JsonRPC.Http.ExecutionPoolSize,
+		Group:   "JsonRPC",
+	})
+	f.DurationFlag(&flagset.DurationFlag{
+		Name:    "http.ep-requesttimeout",
+		Usage:   "Request Timeout for rpc execution pool for HTTP requests",
+		Value:   &c.cliConfig.JsonRPC.Http.ExecutionPoolRequestTimeout,
+		Default: c.cliConfig.JsonRPC.Http.ExecutionPoolRequestTimeout,
+		Group:   "JsonRPC",
+	})
 
 	// ws options
 	f.BoolFlag(&flagset.BoolFlag{
@@ -481,6 +495,20 @@ func (c *Command) Flags() *flagset.Flagset {
 		Default: c.cliConfig.JsonRPC.Ws.API,
 		Group:   "JsonRPC",
 	})
+	f.Uint64Flag(&flagset.Uint64Flag{
+		Name:    "ws.ep-size",
+		Usage:   "Maximum size of workers to run in rpc execution pool for WS requests",
+		Value:   &c.cliConfig.JsonRPC.Ws.ExecutionPoolSize,
+		Default: c.cliConfig.JsonRPC.Ws.ExecutionPoolSize,
+		Group:   "JsonRPC",
+	})
+	f.DurationFlag(&flagset.DurationFlag{
+		Name:    "ws.ep-requesttimeout",
+		Usage:   "Request Timeout for rpc execution pool for WS requests",
+		Value:   &c.cliConfig.JsonRPC.Ws.ExecutionPoolRequestTimeout,
+		Default: c.cliConfig.JsonRPC.Ws.ExecutionPoolRequestTimeout,
+		Group:   "JsonRPC",
+	})
 
 	// graphql options
 	f.BoolFlag(&flagset.BoolFlag{

internal/web3ext/web3ext.go

@@ -192,6 +192,34 @@ web3._extend({
 			name: 'stopWS',
 			call: 'admin_stopWS'
 		}),
+		new web3._extend.Method({
+			name: 'getExecutionPoolSize',
+			call: 'admin_getExecutionPoolSize'
+		}),
+		new web3._extend.Method({
+			name: 'getExecutionPoolRequestTimeout',
+			call: 'admin_getExecutionPoolRequestTimeout'
+		}),
+		// new web3._extend.Method({
+		// 	name: 'setWSExecutionPoolRequestTimeout',
+		// 	call: 'admin_setWSExecutionPoolRequestTimeout',
+		// 	params: 1
+		// }),
+		// new web3._extend.Method({
+		// 	name: 'setHttpExecutionPoolRequestTimeout',
+		// 	call: 'admin_setHttpExecutionPoolRequestTimeout',
+		// 	params: 1
+		// }),
+		new web3._extend.Method({
+			name: 'setWSExecutionPoolSize',
+			call: 'admin_setWSExecutionPoolSize',
+			params: 1
+		}),
+		new web3._extend.Method({
+			name: 'setHttpExecutionPoolSize',
+			call: 'admin_setHttpExecutionPoolSize',
+			params: 1
+		}),
 	],
 	properties: [
 		new web3._extend.Property({

node/api.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -342,3 +343,91 @@ func (s *publicWeb3API) ClientVersion() string {
 func (s *publicWeb3API) Sha3(input hexutil.Bytes) hexutil.Bytes {
 	return crypto.Keccak256(input)
 }
+
+type ExecutionPoolSize struct {
+	HttpLimit int
+	WSLimit   int
+}
+
+type ExecutionPoolRequestTimeout struct {
+	HttpLimit time.Duration
+	WSLimit   time.Duration
+}
+
+func (api *privateAdminAPI) GetExecutionPoolSize() *ExecutionPoolSize {
+	var httpLimit int
+	if api.node.http.host != "" {
+		httpLimit = api.node.http.httpHandler.Load().(*rpcHandler).server.GetExecutionPoolSize()
+	}
+
+	var wsLimit int
+	if api.node.ws.host != "" {
+		wsLimit = api.node.ws.wsHandler.Load().(*rpcHandler).server.GetExecutionPoolSize()
+	}
+
+	executionPoolSize := &ExecutionPoolSize{
+		HttpLimit: httpLimit,
+		WSLimit:   wsLimit,
+	}
+
+	return executionPoolSize
+}
+
+func (api *privateAdminAPI) GetExecutionPoolRequestTimeout() *ExecutionPoolRequestTimeout {
+	var httpLimit time.Duration
+	if api.node.http.host != "" {
+		httpLimit = api.node.http.httpHandler.Load().(*rpcHandler).server.GetExecutionPoolRequestTimeout()
+	}
+
+	var wsLimit time.Duration
+	if api.node.ws.host != "" {
+		wsLimit = api.node.ws.wsHandler.Load().(*rpcHandler).server.GetExecutionPoolRequestTimeout()
+	}
+
+	executionPoolRequestTimeout := &ExecutionPoolRequestTimeout{
+		HttpLimit: httpLimit,
+		WSLimit:   wsLimit,
+	}
+
+	return executionPoolRequestTimeout
+}
+
+// func (api *privateAdminAPI) SetWSExecutionPoolRequestTimeout(n int) *ExecutionPoolRequestTimeout {
+// 	if api.node.ws.host != "" {
+// 		api.node.ws.wsConfig.executionPoolRequestTimeout = time.Duration(n) * time.Millisecond
+// 		api.node.ws.wsHandler.Load().(*rpcHandler).server.SetExecutionPoolRequestTimeout(time.Duration(n) * time.Millisecond)
+// 		log.Warn("updating ws execution pool request timeout", "timeout", n)
+// 	}
+
+// 	return api.GetExecutionPoolRequestTimeout()
+// }
+
+// func (api *privateAdminAPI) SetHttpExecutionPoolRequestTimeout(n int) *ExecutionPoolRequestTimeout {
+// 	if api.node.http.host != "" {
+// 		api.node.http.httpConfig.executionPoolRequestTimeout = time.Duration(n) * time.Millisecond
+// 		api.node.http.httpHandler.Load().(*rpcHandler).server.SetExecutionPoolRequestTimeout(time.Duration(n) * time.Millisecond)
+// 		log.Warn("updating http execution pool request timeout", "timeout", n)
+// 	}
+
+// 	return api.GetExecutionPoolRequestTimeout()
+// }
+
+func (api *privateAdminAPI) SetWSExecutionPoolSize(n int) *ExecutionPoolSize {
+	if api.node.ws.host != "" {
+		api.node.ws.wsConfig.executionPoolSize = uint64(n)
+		api.node.ws.wsHandler.Load().(*rpcHandler).server.SetExecutionPoolSize(n)
+		log.Warn("updating ws execution pool size", "threads", n)
+	}
+
+	return api.GetExecutionPoolSize()
+}
+
+func (api *privateAdminAPI) SetHttpExecutionPoolSize(n int) *ExecutionPoolSize {
+	if api.node.http.host != "" {
+		api.node.http.httpConfig.executionPoolSize = uint64(n)
+		api.node.http.httpHandler.Load().(*rpcHandler).server.SetExecutionPoolSize(n)
+		log.Warn("updating http execution pool size", "threads", n)
+	}
+
+	return api.GetExecutionPoolSize()
+}