diff --git a/cmd/clef/main.go b/cmd/clef/main.go index 1bfb2610e5..541ef7ed62 100644 --- a/cmd/clef/main.go +++ b/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(0, 0) + srv := rpc.NewServer("", 0, 0) err := node.RegisterApis(rpcAPI, []string{"account"}, srv, false) if err != nil { utils.Fatalf("Could not register API: %w", err) diff --git a/go.mod b/go.mod index be2546d064..4ad48bb66b 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/BurntSushi/toml v1.1.0 github.com/JekaMas/crand v1.0.1 github.com/JekaMas/go-grpc-net-conn v0.0.0-20220708155319-6aff21f2d13d - github.com/JekaMas/workerpool v1.1.5 + github.com/JekaMas/workerpool v1.1.8 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 diff --git a/go.sum b/go.sum index cd35756589..1c2aa1f13b 100644 --- a/go.sum +++ b/go.sum @@ -54,8 +54,8 @@ github.com/JekaMas/crand v1.0.1 h1:FMPxkUQqH/hExl0aUXsr0UCGYZ4lJH9IJ5H/KbM6Y9A= github.com/JekaMas/crand v1.0.1/go.mod h1:GGzGpMCht/tbaNQ5A4kSiKSqEoNAhhyTfSDQyIENBQU= 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/JekaMas/workerpool v1.1.8 h1:IbDbTITrDgt1xAzdzTS9aLk4Q/4dCsjUOopiyFkDFZ4= +github.com/JekaMas/workerpool v1.1.8/go.mod h1:IoDWPpwMcA27qbuugZKeBslDrgX09lVmksuh9sjzbhc= github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0= github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI= github.com/Masterminds/goutils v1.1.1/go.mod h1:8cTjp+g8YejhMuvIA5y2vz3BpJxksy863GQaJW2MFNU= diff --git a/node/node.go b/node/node.go index 5cf233d17a..fb9e1f617f 100644 --- a/node/node.go +++ b/node/node.go @@ -105,7 +105,7 @@ func New(conf *Config) (*Node, error) { node := &Node{ config: conf, - inprocHandler: rpc.NewServer(0, 0), + inprocHandler: rpc.NewServer("inproc", 0, 0), eventmux: new(event.TypeMux), log: conf.Logger, stop: make(chan struct{}), diff --git a/node/rpcstack.go b/node/rpcstack.go index cba9a22f6f..ea8717310c 100644 --- a/node/rpcstack.go +++ b/node/rpcstack.go @@ -293,7 +293,7 @@ func (h *httpServer) enableRPC(apis []rpc.API, config httpConfig) error { } // Create RPC server and handler. - srv := rpc.NewServer(config.executionPoolSize, config.executionPoolRequestTimeout) + srv := rpc.NewServer("http", config.executionPoolSize, config.executionPoolRequestTimeout) srv.SetRPCBatchLimit(h.RPCBatchLimit) if err := RegisterApis(apis, config.Modules, srv, false); err != nil { return err @@ -325,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(config.executionPoolSize, config.executionPoolRequestTimeout) + srv := rpc.NewServer("ws", config.executionPoolSize, config.executionPoolRequestTimeout) srv.SetRPCBatchLimit(h.RPCBatchLimit) if err := RegisterApis(apis, config.Modules, srv, false); err != nil { return err diff --git a/rpc/client.go b/rpc/client.go index fc286fe8dc..6b29a21062 100644 --- a/rpc/client.go +++ b/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, NewExecutionPool(100, 0)) + handler := newHandler(ctx, conn, c.idgen, c.services, NewExecutionPool(100, 0, "rpcclient", true)) return &clientConn{conn, handler} } diff --git a/rpc/client_test.go b/rpc/client_test.go index 1bebd27677..c97cb79567 100644 --- a/rpc/client_test.go +++ b/rpc/client_test.go @@ -410,7 +410,7 @@ func TestClientSubscriptionUnsubscribeServer(t *testing.T) { t.Parallel() // Create the server. - srv := NewServer(0, 0) + srv := NewServer("test", 0, 0) srv.RegisterName("nftest", new(notificationTestService)) p1, p2 := net.Pipe() recorder := &unsubscribeRecorder{ServerCodec: NewCodec(p1)} @@ -446,7 +446,7 @@ func TestClientSubscriptionChannelClose(t *testing.T) { t.Parallel() var ( - srv = NewServer(0, 0) + srv = NewServer("test", 0, 0) httpsrv = httptest.NewServer(srv.WebsocketHandler(nil)) wsURL = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:") ) diff --git a/rpc/endpoints.go b/rpc/endpoints.go index 2a539d4fc5..2893a9ba10 100644 --- a/rpc/endpoints.go +++ b/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(0, 0) + handler = NewServer("", 0, 0) // we can skip reporting ipc metrics, hence empty service name regMap = make(map[string]struct{}) registered []string ) diff --git a/rpc/execution_pool.go b/rpc/execution_pool.go index d0f5ab5daa..daa431e9bc 100644 --- a/rpc/execution_pool.go +++ b/rpc/execution_pool.go @@ -7,24 +7,32 @@ import ( "time" "github.com/JekaMas/workerpool" + "github.com/ethereum/go-ethereum/metrics" ) type SafePool struct { - executionPool *atomic.Pointer[workerpool.WorkerPool] + executionPool atomic.Pointer[workerpool.WorkerPool] sync.RWMutex - timeout time.Duration size int + timeout time.Duration + + service string // the service using ep + processed atomic.Int64 // keeps count of total processed requests + + close chan struct{} // Skip sending task to execution pool fastPath bool } -func NewExecutionPool(initialSize int, timeout time.Duration) *SafePool { +func NewExecutionPool(initialSize int, timeout time.Duration, service string, report bool) *SafePool { sp := &SafePool{ size: initialSize, timeout: timeout, + service: service, + close: make(chan struct{}), } if initialSize == 0 { @@ -33,11 +41,11 @@ func NewExecutionPool(initialSize int, timeout time.Duration) *SafePool { return sp } - var ptr atomic.Pointer[workerpool.WorkerPool] + sp.executionPool.Store(workerpool.New(initialSize)) - p := workerpool.New(initialSize) - ptr.Store(p) - sp.executionPool = &ptr + if metrics.Enabled && report { + go sp.reportMetrics(3 * time.Second) + } return sp } @@ -51,10 +59,6 @@ func (s *SafePool) Submit(ctx context.Context, fn func() error) (<-chan error, b return nil, true } - if s.executionPool == nil { - return nil, false - } - pool := s.executionPool.Load() if pool == nil { return nil, false @@ -97,3 +101,42 @@ func (s *SafePool) Size() int { return s.size } + +func (s *SafePool) Stop() { + close(s.close) + + if s.executionPool.Load() != nil { + s.executionPool.Load().Stop() + } +} + +// reportMetrics reports the metrics after every `refresh` time interval +// regarding the execution pool. +func (s *SafePool) reportMetrics(refresh time.Duration) { + var ( + epWorkerCountGuage metrics.Gauge + epWaitingQueueGuage metrics.Gauge + epProcessedRequestsHistogram metrics.Histogram + ) + + ticker := time.NewTicker(refresh) + + for { + select { + case <-ticker.C: + ep := s.executionPool.Load() + + epWorkerCountGuage, epWaitingQueueGuage, epProcessedRequestsHistogram = newEpMetrics(s.service) + + epWorkerCountGuage.Update(ep.GetWorkerCount()) + epWaitingQueueGuage.Update(int64(ep.WaitingQueueSize())) + epProcessedRequestsHistogram.Update(s.processed.Load()) + + s.processed.Store(0) + case <-s.close: + ticker.Stop() + + return + } + } +} diff --git a/rpc/handler.go b/rpc/handler.go index 8b219ce792..84760d2d28 100644 --- a/rpc/handler.go +++ b/rpc/handler.go @@ -231,6 +231,8 @@ func (h *handler) startCallProc(fn func(*callProc)) { defer cancel() fn(&callProc{ctx: ctx}) + h.executionPool.processed.Add(1) + return nil }) } @@ -269,7 +271,6 @@ 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}) @@ -292,6 +293,7 @@ func (h *handler) handleResponse(msg *jsonrpcMessage) { if op.err = json.Unmarshal(msg.Result, &op.sub.subid); op.err == nil { h.executionPool.Submit(context.Background(), func() error { op.sub.run() + h.executionPool.processed.Add(1) return nil }) diff --git a/rpc/http_test.go b/rpc/http_test.go index 9737e64e91..7dc3569037 100644 --- a/rpc/http_test.go +++ b/rpc/http_test.go @@ -103,7 +103,7 @@ func TestHTTPResponseWithEmptyGet(t *testing.T) { func TestHTTPRespBodyUnlimited(t *testing.T) { const respLength = maxRequestContentLength * 3 - s := NewServer(0, 0) + s := NewServer("test", 0, 0) defer s.Stop() s.RegisterName("test", largeRespService{respLength}) ts := httptest.NewServer(s) diff --git a/rpc/inproc.go b/rpc/inproc.go index 29af5507b9..e6d302a09d 100644 --- a/rpc/inproc.go +++ b/rpc/inproc.go @@ -30,6 +30,7 @@ func DialInProc(handler *Server) *Client { //nolint:contextcheck handler.executionPool.Submit(initctx, func() error { handler.ServeCodec(NewCodec(p1), 0) + handler.executionPool.processed.Add(1) return nil }) diff --git a/rpc/ipc.go b/rpc/ipc.go index 76fbd13f92..69aa0ae8cf 100644 --- a/rpc/ipc.go +++ b/rpc/ipc.go @@ -38,6 +38,7 @@ func (s *Server) ServeListener(l net.Listener) error { s.executionPool.Submit(context.Background(), func() error { s.ServeCodec(NewCodec(conn), 0) + s.executionPool.processed.Add(1) return nil }) } diff --git a/rpc/metrics.go b/rpc/metrics.go index 7fb6fc0a17..8bd3deae2f 100644 --- a/rpc/metrics.go +++ b/rpc/metrics.go @@ -37,3 +37,11 @@ func newRPCServingTimer(method string, valid bool) metrics.Timer { m := fmt.Sprintf("rpc/duration/%s/%s", method, flag) return metrics.GetOrRegisterTimer(m, nil) } + +func newEpMetrics(service string) (metrics.Gauge, metrics.Gauge, metrics.Histogram) { + epWorkerCount := fmt.Sprintf("rpc/ep/workers/%s", service) + epWaitingQueue := fmt.Sprintf("rpc/ep/queue/%s", service) + epProcessedRequests := fmt.Sprintf("rpc/ep/processed/%s", service) + + return metrics.GetOrRegisterGauge(epWorkerCount, nil), metrics.GetOrRegisterGauge(epWaitingQueue, nil), metrics.GetOrRegisterHistogram(epProcessedRequests, nil, metrics.NewExpDecaySample(1028, 0.015)) +} diff --git a/rpc/server.go b/rpc/server.go index 04ee2dc87b..87ddcaa37d 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -56,18 +56,24 @@ type Server struct { } // NewServer creates a new server instance with no registered handlers. -func NewServer(executionPoolSize uint64, executionPoolRequesttimeout time.Duration) *Server { +func NewServer(service string, executionPoolSize uint64, executionPoolRequesttimeout time.Duration) *Server { + reportEpStats := true + if service == "" || service == "test" { + reportEpStats = false + } + server := &Server{ idgen: randomIDGenerator(), codecs: mapset.NewSet(), run: 1, - executionPool: NewExecutionPool(int(executionPoolSize), executionPoolRequesttimeout), + executionPool: NewExecutionPool(int(executionPoolSize), executionPoolRequesttimeout, service, reportEpStats), } // Register the default service providing meta information about the RPC service such // as the services and methods it offers. rpcService := &RPCService{server} server.RegisterName(MetadataApi, rpcService) + return server } @@ -166,6 +172,9 @@ func (s *Server) serveSingleRequest(ctx context.Context, codec ServerCodec) { // requests to finish, then closes all codecs which will cancel pending requests and // subscriptions. func (s *Server) Stop() { + // Stop the execution pool + s.executionPool.Stop() + if atomic.CompareAndSwapInt32(&s.run, 1, 0) { log.Debug("RPC server shutting down") s.codecs.Each(func(c interface{}) bool { diff --git a/rpc/server_test.go b/rpc/server_test.go index 166956681b..4fcdfde07f 100644 --- a/rpc/server_test.go +++ b/rpc/server_test.go @@ -29,7 +29,7 @@ import ( ) func TestServerRegisterName(t *testing.T) { - server := NewServer(0, 0) + server := NewServer("test", 0, 0) service := new(testService) if err := server.RegisterName("test", service); err != nil { diff --git a/rpc/subscription_test.go b/rpc/subscription_test.go index cfca1b24b9..f76c464718 100644 --- a/rpc/subscription_test.go +++ b/rpc/subscription_test.go @@ -53,7 +53,7 @@ func TestSubscriptions(t *testing.T) { subCount = len(namespaces) notificationCount = 3 - server = NewServer(0, 0) + server = NewServer("test", 0, 0) clientConn, serverConn = net.Pipe() out = json.NewEncoder(clientConn) in = json.NewDecoder(clientConn) diff --git a/rpc/testservice_test.go b/rpc/testservice_test.go index 2285821779..5f53a2fcb2 100644 --- a/rpc/testservice_test.go +++ b/rpc/testservice_test.go @@ -26,7 +26,7 @@ import ( ) func newTestServer() *Server { - server := NewServer(0, 0) + server := NewServer("test", 0, 0) server.idgen = sequentialIDGenerator() if err := server.RegisterName("test", new(testService)); err != nil { panic(err) diff --git a/rpc/websocket_test.go b/rpc/websocket_test.go index b805ed2023..9fb57247c7 100644 --- a/rpc/websocket_test.go +++ b/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(0, 0) + srv = NewServer("test", 0, 0) httpsrv = httptest.NewServer(srv.WebsocketHandler(nil)) wsURL = "ws:" + strings.TrimPrefix(httpsrv.URL, "http:") )