Arpit/add execution pool 2 (#719)

* initial

* linters

* linters

* remove timeout

* update pool

* change pool size function

* check nil

* check nil

* fix tests

* Use execution pool from server in all handlers

* simplify things

* test fix

* add support for cli, config

* add to cli and config

* merge base branch

* debug statements

* fix bug

* atomic pointer timeout

* add apis

* update workerpool

* fix issues

* change params

* fix issues

* fix ipc issue

* remove execution pool from IPC

* revert

* fix tests

* mutex

* refactor flag and value names

* ordering fix

* refactor flag and value names

* update default ep size to 40

* fix bor start issues

* revert file changes

* debug statements

* fix bug

* update workerpool

* atomic pointer timeout

* add apis

* Merge branch 'add-execution-pool' of github.com:maticnetwork/bor into arpit/add-execution-pool

* fix issues

* change params

* fix issues

* fix ipc issue

* remove execution pool from IPC

* revert

* merge base branch

* Merge branch 'add-execution-pool' of github.com:maticnetwork/bor into arpit/add-execution-pool

* mutex

* fix tests

* Merge branch 'arpit/add-execution-pool' of github.com:maticnetwork/bor into arpit/add-execution-pool

* Change default size of execution pool to 40

* refactor flag and value names

* fix merge conflicts

* ordering fix

* refactor flag and value names

* update default ep size to 40

* fix bor start issues

* revert file changes

* fix linters

* fix go.mod

* change sec to ms

* change default value for ep timeout

* fix node api calls

* comment setter for ep timeout

---------

Co-authored-by: Evgeny Danienko <6655321@bk.ru>
Co-authored-by: Jerry <jerrycgh@gmail.com>
Co-authored-by: Manav Darji <manavdarji.india@gmail.com>

builder/files/config.toml

@@ -75,6 +75,8 @@ syncmode = "full"
 #     api = ["eth", "net", "web3", "txpool", "bor"]
 #     vhosts = ["*"]
 #     corsdomain = ["*"]
+#     ep-size = 40
+#     ep-requesttimeout = "0s"
 #   [jsonrpc.ws]
 #     enabled = false
 #     port = 8546
@@ -82,6 +84,8 @@ syncmode = "full"
 #     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,18 +74,22 @@ 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]
@@ -93,6 +97,7 @@ ethstats = ""                # Reporting URL of a ethstats service (nodename:sec
     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()
+}

node/config.go

@@ -25,6 +25,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -207,6 +208,11 @@ type Config struct {
 
 	// Maximum number of messages in a batch
 	RPCBatchLimit uint64 `toml:",omitempty"`
+	// Configs for RPC execution pool
+	WSJsonRPCExecutionPoolSize             uint64        `toml:",omitempty"`
+	WSJsonRPCExecutionPoolRequestTimeout   time.Duration `toml:",omitempty"`
+	HTTPJsonRPCExecutionPoolSize           uint64        `toml:",omitempty"`
+	HTTPJsonRPCExecutionPoolRequestTimeout time.Duration `toml:",omitempty"`
 }
 
 // IPCEndpoint resolves an IPC endpoint based on a configured value, taking into

node/node.go

@@ -105,7 +105,7 @@ func New(conf *Config) (*Node, error) {
 
 	node := &Node{
 		config:        conf,
-		inprocHandler: rpc.NewServer(),
+		inprocHandler: rpc.NewServer(0, 0),
 		eventmux:      new(event.TypeMux),
 		log:           conf.Logger,
 		stop:          make(chan struct{}),
@@ -405,10 +405,12 @@ func (n *Node) startRPC() error {
 			return err
 		}
 		if err := server.enableRPC(apis, httpConfig{
-			CorsAllowedOrigins: n.config.HTTPCors,
-			Vhosts:             n.config.HTTPVirtualHosts,
-			Modules:            n.config.HTTPModules,
-			prefix:             n.config.HTTPPathPrefix,
+			CorsAllowedOrigins:          n.config.HTTPCors,
+			Vhosts:                      n.config.HTTPVirtualHosts,
+			Modules:                     n.config.HTTPModules,
+			prefix:                      n.config.HTTPPathPrefix,
+			executionPoolSize:           n.config.HTTPJsonRPCExecutionPoolSize,
+			executionPoolRequestTimeout: n.config.HTTPJsonRPCExecutionPoolRequestTimeout,
 		}); err != nil {
 			return err
 		}
@@ -422,9 +424,11 @@ func (n *Node) startRPC() error {
 			return err
 		}
 		if err := server.enableWS(n.rpcAPIs, wsConfig{
-			Modules: n.config.WSModules,
-			Origins: n.config.WSOrigins,
-			prefix:  n.config.WSPathPrefix,
+			Modules:                     n.config.WSModules,
+			Origins:                     n.config.WSOrigins,
+			prefix:                      n.config.WSPathPrefix,
+			executionPoolSize:           n.config.WSJsonRPCExecutionPoolSize,
+			executionPoolRequestTimeout: n.config.WSJsonRPCExecutionPoolRequestTimeout,
 		}); err != nil {
 			return err
 		}

node/rpcstack.go

@@ -28,6 +28,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/rs/cors"
 
@@ -42,6 +43,10 @@ type httpConfig struct {
 	Vhosts             []string
 	prefix             string // path prefix on which to mount http handler
 	jwtSecret          []byte // optional JWT secret
+
+	// Execution pool config
+	executionPoolSize           uint64
+	executionPoolRequestTimeout time.Duration
 }
 
 // wsConfig is the JSON-RPC/Websocket configuration
@@ -50,6 +55,10 @@ type wsConfig struct {
 	Modules   []string
 	prefix    string // path prefix on which to mount ws handler
 	jwtSecret []byte // optional JWT secret
+
+	// Execution pool config
+	executionPoolSize           uint64
+	executionPoolRequestTimeout time.Duration
 }
 
 type rpcHandler struct {
@@ -284,7 +293,7 @@ func (h *httpServer) enableRPC(apis []rpc.API, config httpConfig) error {
 	}
 
 	// Create RPC server and handler.
-	srv := rpc.NewServer()
+	srv := rpc.NewServer(config.executionPoolSize, config.executionPoolRequestTimeout)
 	srv.SetRPCBatchLimit(h.RPCBatchLimit)
 	if err := RegisterApis(apis, config.Modules, srv, false); err != nil {
 		return err
@@ -316,7 +325,7 @@ func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error {
 		return fmt.Errorf("JSON-RPC over WebSocket is already enabled")
 	}
 	// Create RPC server and handler.
-	srv := rpc.NewServer()
+	srv := rpc.NewServer(config.executionPoolSize, config.executionPoolRequestTimeout)
 	srv.SetRPCBatchLimit(h.RPCBatchLimit)
 	if err := RegisterApis(apis, config.Modules, srv, false); err != nil {
 		return err

packaging/templates/mainnet-v1/archive/config.toml

@@ -67,6 +67,8 @@ gcmode = "archive"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     [jsonrpc.ws]
         enabled = true
         port = 8546
@@ -74,6 +76,8 @@ gcmode = "archive"
         # host = "localhost"
         # api = ["web3", "net"]
         origins = ["*"]
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

packaging/templates/mainnet-v1/sentry/sentry/bor/config.toml

@@ -67,6 +67,8 @@ syncmode = "full"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.ws]
         # enabled = false
         # port = 8546
@@ -74,6 +76,8 @@ syncmode = "full"
         # host = "localhost"
         # api = ["web3", "net"]
         # origins = ["*"]
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

packaging/templates/mainnet-v1/sentry/validator/bor/config.toml

@@ -69,6 +69,8 @@ syncmode = "full"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.ws]
         # enabled = false
         # port = 8546
@@ -76,6 +78,8 @@ syncmode = "full"
         # host = "localhost"
         # api = ["web3", "net"]
         # origins = ["*"]
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

packaging/templates/mainnet-v1/without-sentry/bor/config.toml

@@ -69,6 +69,8 @@ syncmode = "full"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.ws]
         # enabled = false
         # port = 8546
@@ -76,6 +78,8 @@ syncmode = "full"
         # host = "localhost"
         # api = ["web3", "net"]
         # origins = ["*"]
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

packaging/templates/testnet-v4/archive/config.toml

@@ -67,6 +67,8 @@ gcmode = "archive"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     [jsonrpc.ws]
         enabled = true
         port = 8546
@@ -74,6 +76,8 @@ gcmode = "archive"
         # host = "localhost"
         # api = ["web3", "net"]
         origins = ["*"]
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

packaging/templates/testnet-v4/sentry/sentry/bor/config.toml

@@ -67,6 +67,8 @@ syncmode = "full"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     # [jsonrpc.ws]
         # enabled = false
         # port = 8546
@@ -74,6 +76,8 @@ syncmode = "full"
         # host = "localhost"
         # api = ["web3", "net"]
         # origins = ["*"]
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

packaging/templates/testnet-v4/sentry/validator/bor/config.toml

@@ -69,6 +69,8 @@ syncmode = "full"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.ws]
         # enabled = false
         # port = 8546
@@ -76,6 +78,8 @@ syncmode = "full"
         # host = "localhost"
         # api = ["web3", "net"]
         # origins = ["*"]
+        #     ep-size = 40
+#     ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

packaging/templates/testnet-v4/without-sentry/bor/config.toml

@@ -69,6 +69,8 @@ syncmode = "full"
         vhosts = ["*"]
         corsdomain = ["*"]
         # prefix = ""
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     # [jsonrpc.ws]
         # enabled = false
         # port = 8546
@@ -76,6 +78,8 @@ syncmode = "full"
         # host = "localhost"
         # api = ["web3", "net"]
         # origins = ["*"]
+        # ep-size = 40
+        # ep-requesttimeout = "0s"
     # [jsonrpc.graphql]
         # enabled = false
         # port = 0

rpc/client.go

@@ -112,7 +112,7 @@ func (c *Client) newClientConn(conn ServerCodec) *clientConn {
 	ctx := context.Background()
 	ctx = context.WithValue(ctx, clientContextKey{}, c)
 	ctx = context.WithValue(ctx, peerInfoContextKey{}, conn.peerInfo())
-	handler := newHandler(ctx, conn, c.idgen, c.services)
+	handler := newHandler(ctx, conn, c.idgen, c.services, NewExecutionPool(100, 0))
 	return &clientConn{conn, handler}
 }
 

rpc/client_test.go

@@ -33,12 +33,14 @@ import (
 	"time"
 
 	"github.com/davecgh/go-spew/spew"
+
 	"github.com/ethereum/go-ethereum/log"
 )
 
 func TestClientRequest(t *testing.T) {
 	server := newTestServer()
 	defer server.Stop()
+
 	client := DialInProc(server)
 	defer client.Close()
 
@@ -46,6 +48,7 @@ func TestClientRequest(t *testing.T) {
 	if err := client.Call(&resp, "test_echo", "hello", 10, &echoArgs{"world"}); err != nil {
 		t.Fatal(err)
 	}
+
 	if !reflect.DeepEqual(resp, echoResult{"hello", 10, &echoArgs{"world"}}) {
 		t.Errorf("incorrect result %#v", resp)
 	}
@@ -407,7 +410,7 @@ func TestClientSubscriptionUnsubscribeServer(t *testing.T) {
 	t.Parallel()
 
 	// Create the server.
-	srv := NewServer()
+	srv := NewServer(0, 0)
 	srv.RegisterName("nftest", new(notificationTestService))
 	p1, p2 := net.Pipe()
 	recorder := &unsubscribeRecorder{ServerCodec: NewCodec(p1)}
@@ -443,7 +446,7 @@ func TestClientSubscriptionChannelClose(t *testing.T) {
 	t.Parallel()
 
 	var (
-		srv     = NewServer()
+		srv     = NewServer(0, 0)
 		httpsrv = httptest.NewServer(srv.WebsocketHandler(nil))
 		wsURL   = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:")
 	)

rpc/endpoints.go

@@ -27,7 +27,7 @@ import (
 func StartIPCEndpoint(ipcEndpoint string, apis []API) (net.Listener, *Server, error) {
 	// Register all the APIs exposed by the services.
 	var (
-		handler    = NewServer()
+		handler    = NewServer(0, 0)
 		regMap     = make(map[string]struct{})
 		registered []string
 	)

rpc/execution_pool.go

@@ -0,0 +1,99 @@
+package rpc
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/JekaMas/workerpool"
+)
+
+type SafePool struct {
+	executionPool *atomic.Pointer[workerpool.WorkerPool]
+
+	sync.RWMutex
+
+	timeout time.Duration
+	size    int
+
+	// Skip sending task to execution pool
+	fastPath bool
+}
+
+func NewExecutionPool(initialSize int, timeout time.Duration) *SafePool {
+	sp := &SafePool{
+		size:    initialSize,
+		timeout: timeout,
+	}
+
+	if initialSize == 0 {
+		sp.fastPath = true
+
+		return sp
+	}
+
+	var ptr atomic.Pointer[workerpool.WorkerPool]
+
+	p := workerpool.New(initialSize)
+	ptr.Store(p)
+	sp.executionPool = &ptr
+
+	return sp
+}
+
+func (s *SafePool) Submit(ctx context.Context, fn func() error) (<-chan error, bool) {
+	if s.fastPath {
+		go func() {
+			_ = fn()
+		}()
+
+		return nil, true
+	}
+
+	if s.executionPool == nil {
+		return nil, false
+	}
+
+	pool := s.executionPool.Load()
+	if pool == nil {
+		return nil, false
+	}
+
+	return pool.Submit(ctx, fn, s.Timeout()), true
+}
+
+func (s *SafePool) ChangeSize(n int) {
+	oldPool := s.executionPool.Swap(workerpool.New(n))
+
+	if oldPool != nil {
+		go func() {
+			oldPool.StopWait()
+		}()
+	}
+
+	s.Lock()
+	s.size = n
+	s.Unlock()
+}
+
+func (s *SafePool) ChangeTimeout(n time.Duration) {
+	s.Lock()
+	defer s.Unlock()
+
+	s.timeout = n
+}
+
+func (s *SafePool) Timeout() time.Duration {
+	s.RLock()
+	defer s.RUnlock()
+
+	return s.timeout
+}
+
+func (s *SafePool) Size() int {
+	s.RLock()
+	defer s.RUnlock()
+
+	return s.size
+}

rpc/handler.go

@@ -34,21 +34,20 @@ import (
 //
 // The entry points for incoming messages are:
 //
-//    h.handleMsg(message)
-//    h.handleBatch(message)
+//	h.handleMsg(message)
+//	h.handleBatch(message)
 //
 // Outgoing calls use the requestOp struct. Register the request before sending it
 // on the connection:
 //
-//    op := &requestOp{ids: ...}
-//    h.addRequestOp(op)
+//	op := &requestOp{ids: ...}
+//	h.addRequestOp(op)
 //
 // Now send the request, then wait for the reply to be delivered through handleMsg:
 //
-//    if err := op.wait(...); err != nil {
-//        h.removeRequestOp(op) // timeout, etc.
-//    }
-//
+//	if err := op.wait(...); err != nil {
+//	    h.removeRequestOp(op) // timeout, etc.
+//	}
 type handler struct {
 	reg            *serviceRegistry
 	unsubscribeCb  *callback
@@ -64,6 +63,8 @@ type handler struct {
 
 	subLock    sync.Mutex
 	serverSubs map[ID]*Subscription
+
+	executionPool *SafePool
 }
 
 type callProc struct {
@@ -71,7 +72,7 @@ type callProc struct {
 	notifiers []*Notifier
 }
 
-func newHandler(connCtx context.Context, conn jsonWriter, idgen func() ID, reg *serviceRegistry) *handler {
+func newHandler(connCtx context.Context, conn jsonWriter, idgen func() ID, reg *serviceRegistry, pool *SafePool) *handler {
 	rootCtx, cancelRoot := context.WithCancel(connCtx)
 	h := &handler{
 		reg:            reg,
@@ -84,11 +85,13 @@ func newHandler(connCtx context.Context, conn jsonWriter, idgen func() ID, reg *
 		allowSubscribe: true,
 		serverSubs:     make(map[ID]*Subscription),
 		log:            log.Root(),
+		executionPool:  pool,
 	}
 	if conn.remoteAddr() != "" {
 		h.log = h.log.New("conn", conn.remoteAddr())
 	}
 	h.unsubscribeCb = newCallback(reflect.Value{}, reflect.ValueOf(h.unsubscribe))
+
 	return h
 }
 
@@ -219,12 +222,16 @@ func (h *handler) cancelServerSubscriptions(err error) {
 // startCallProc runs fn in a new goroutine and starts tracking it in the h.calls wait group.
 func (h *handler) startCallProc(fn func(*callProc)) {
 	h.callWG.Add(1)
-	go func() {
-		ctx, cancel := context.WithCancel(h.rootCtx)
+
+	ctx, cancel := context.WithCancel(h.rootCtx)
+
+	h.executionPool.Submit(context.Background(), func() error {
 		defer h.callWG.Done()
 		defer cancel()
 		fn(&callProc{ctx: ctx})
-	}()
+
+		return nil
+	})
 }
 
 // handleImmediate executes non-call messages. It returns false if the message is a
@@ -261,6 +268,7 @@ func (h *handler) handleSubscriptionResult(msg *jsonrpcMessage) {
 
 // handleResponse processes method call responses.
 func (h *handler) handleResponse(msg *jsonrpcMessage) {
+
 	op := h.respWait[string(msg.ID)]
 	if op == nil {
 		h.log.Debug("Unsolicited RPC response", "reqid", idForLog{msg.ID})
@@ -281,7 +289,11 @@ func (h *handler) handleResponse(msg *jsonrpcMessage) {
 		return
 	}
 	if op.err = json.Unmarshal(msg.Result, &op.sub.subid); op.err == nil {
-		go op.sub.run()
+		h.executionPool.Submit(context.Background(), func() error {
+			op.sub.run()
+			return nil
+		})
+
 		h.clientSubs[op.sub.subid] = op.sub
 	}
 }

rpc/http_test.go

@@ -103,7 +103,7 @@ func TestHTTPResponseWithEmptyGet(t *testing.T) {
 func TestHTTPRespBodyUnlimited(t *testing.T) {
 	const respLength = maxRequestContentLength * 3
 
-	s := NewServer()
+	s := NewServer(0, 0)
 	defer s.Stop()
 	s.RegisterName("test", largeRespService{respLength})
 	ts := httptest.NewServer(s)

rpc/inproc.go

@@ -26,7 +26,13 @@ func DialInProc(handler *Server) *Client {
 	initctx := context.Background()
 	c, _ := newClient(initctx, func(context.Context) (ServerCodec, error) {
 		p1, p2 := net.Pipe()
-		go handler.ServeCodec(NewCodec(p1), 0)
+
+		//nolint:contextcheck
+		handler.executionPool.Submit(initctx, func() error {
+			handler.ServeCodec(NewCodec(p1), 0)
+			return nil
+		})
+
 		return NewCodec(p2), nil
 	})
 	return c

rpc/ipc.go

@@ -35,7 +35,11 @@ func (s *Server) ServeListener(l net.Listener) error {
 			return err
 		}
 		log.Trace("Accepted RPC connection", "conn", conn.RemoteAddr())
-		go s.ServeCodec(NewCodec(conn), 0)
+
+		s.executionPool.Submit(context.Background(), func() error {
+			s.ServeCodec(NewCodec(conn), 0)
+			return nil
+		})
 	}
 }
 

rpc/server.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"io"
 	"sync/atomic"
+	"time"
 
 	mapset "github.com/deckarep/golang-set"
 
@@ -50,12 +51,19 @@ type Server struct {
 	run      int32
 	codecs   mapset.Set
 
-	BatchLimit uint64
+	BatchLimit    uint64
+	executionPool *SafePool
 }
 
 // NewServer creates a new server instance with no registered handlers.
-func NewServer() *Server {
-	server := &Server{idgen: randomIDGenerator(), codecs: mapset.NewSet(), run: 1}
+func NewServer(executionPoolSize uint64, executionPoolRequesttimeout time.Duration) *Server {
+	server := &Server{
+		idgen:         randomIDGenerator(),
+		codecs:        mapset.NewSet(),
+		run:           1,
+		executionPool: NewExecutionPool(int(executionPoolSize), executionPoolRequesttimeout),
+	}
+
 	// Register the default service providing meta information about the RPC service such
 	// as the services and methods it offers.
 	rpcService := &RPCService{server}
@@ -67,6 +75,22 @@ func (s *Server) SetRPCBatchLimit(batchLimit uint64) {
 	s.BatchLimit = batchLimit
 }
 
+func (s *Server) SetExecutionPoolSize(n int) {
+	s.executionPool.ChangeSize(n)
+}
+
+func (s *Server) SetExecutionPoolRequestTimeout(n time.Duration) {
+	s.executionPool.ChangeTimeout(n)
+}
+
+func (s *Server) GetExecutionPoolRequestTimeout() time.Duration {
+	return s.executionPool.Timeout()
+}
+
+func (s *Server) GetExecutionPoolSize() int {
+	return s.executionPool.Size()
+}
+
 // RegisterName creates a service for the given receiver type under the given name. When no
 // methods on the given receiver match the criteria to be either a RPC method or a
 // subscription an error is returned. Otherwise a new service is created and added to the
@@ -106,7 +130,8 @@ func (s *Server) serveSingleRequest(ctx context.Context, codec ServerCodec) {
 		return
 	}
 
-	h := newHandler(ctx, codec, s.idgen, &s.services)
+	h := newHandler(ctx, codec, s.idgen, &s.services, s.executionPool)
+
 	h.allowSubscribe = false
 	defer h.close(io.EOF, nil)
 

rpc/server_test.go

@@ -29,7 +29,7 @@ import (
 )
 
 func TestServerRegisterName(t *testing.T) {
-	server := NewServer()
+	server := NewServer(0, 0)
 	service := new(testService)
 
 	if err := server.RegisterName("test", service); err != nil {

rpc/subscription_test.go

@@ -53,7 +53,7 @@ func TestSubscriptions(t *testing.T) {
 		subCount          = len(namespaces)
 		notificationCount = 3
 
-		server                 = NewServer()
+		server                 = NewServer(0, 0)
 		clientConn, serverConn = net.Pipe()
 		out                    = json.NewEncoder(clientConn)
 		in                     = json.NewDecoder(clientConn)

rpc/testservice_test.go

@@ -26,7 +26,7 @@ import (
 )
 
 func newTestServer() *Server {
-	server := NewServer()
+	server := NewServer(0, 0)
 	server.idgen = sequentialIDGenerator()
 	if err := server.RegisterName("test", new(testService)); err != nil {
 		panic(err)

rpc/websocket_test.go

@@ -203,7 +203,7 @@ func TestClientWebsocketPing(t *testing.T) {
 // This checks that the websocket transport can deal with large messages.
 func TestClientWebsocketLargeMessage(t *testing.T) {
 	var (
-		srv     = NewServer()
+		srv     = NewServer(0, 0)
 		httpsrv = httptest.NewServer(srv.WebsocketHandler(nil))
 		wsURL   = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:")
 	)