Senn
Senn/Decode Ethereum/
RPC
RPC

RPC

Overview

RPC (Remote Procedure Call) in Geth (Go Ethereum) is a protocol that allows a client to interact with the Ethereum node by invoking functions as if they were executing locally. This protocol facilitates communication between the client and the node, enabling operations like querying blockchain data, sending transactions, and managing accounts. The Geth RPC interface supports various protocols, including HTTP, WebSocket, and IPC (Inter-Process Communication).

Key Features of Geth's RPC

  1. JSON-RPC Protocol: Geth uses the JSON-RPC 2.0 protocol, which encodes requests and responses in JSON format, making it easy to parse and human-readable.
  1. Multiple Endpoints: Supports communication over HTTP, WebSockets, and IPC, providing flexibility for different use cases.
  1. Modular Design: Organizes RPC methods into modules, each serving specific functionalities.

How RPC Works in Geth

  1. Request Reception: The RPC server receives a JSON-encoded request from the client.
  1. Parsing: The server parses the request to extract the method name and parameters.
  1. Routing: Based on the method name, the server routes the request to the appropriate module and function.
  1. Execution: The function executes the requested operation, interacting with the Ethereum state or blockchain data as needed.
  1. Response: The function returns the result, which is then formatted as a JSON response and sent back to the client.

Example RPC Request and Response

Request: Querying the balance of an Ethereum account:
Note the method includes two parts separated by _. The first part is namespace, and the second part is the function under the namespace.
{
  "jsonrpc": "2.0",
  "method": "eth_getBalance",
  "params": ["0x407d73d8a49eeb85d32cf465507dd71d507100c1", "latest"],
  "id": 1
}
Response:
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": "0x0234c8a3397aab58" // balance in wei
}

Code

New Server

The NewServer function in Geth (Go Ethereum) is responsible for creating and initializing a new instance of an RPC (Remote Procedure Call) server. This server handles the registration of RPC methods and the processing of incoming RPC requests.
1. Server Struct Initialization
The Server struct is initialized with default values:
  • idgen: This is a function that generates unique IDs for RPC requests and subscriptions. It uses the randomIDGenerator function to create this ID generator.
  • codecs: This is a map that keeps track of active codecs (connections). A codec in this context is an abstraction that handles encoding and decoding of RPC messages.
  • httpBodyLimit: This sets a limit on the size of HTTP request bodies to prevent excessively large requests that could overwhelm the server. The defaultBodyLimit is a predefined constant setting this limit.
2. Set Server to Running State
The run field of the Server struct is an atomic boolean used to indicate whether the server is running:
This ensures that the server is initially in a running state and ready to handle requests.
3. Register Default RPC Service
The RPCService is a default service that provides meta-information about the RPC server, such as the available services and methods. It is registered under the name defined by the MetadataApi constant:
  • RPCService: This struct holds a reference to the server and provides methods that return metadata about the server.
  • RegisterName: This method registers the RPCService under a given name (rpc). This allows clients to query the server for information about available RPC methods and services.
 
Server Struct
The Server struct itself includes several important fields and methods that manage the lifecycle and functionality of the RPC server:
  • services: A registry of all services and their methods that the server can provide to clients.
  • idgen: A function for generating unique IDs.
  • mutex: A mutex to synchronize access to shared resources within the server.
  • codecs: A map to keep track of active connections/codecs.
  • run: An atomic boolean to indicate if the server is running.
  • batchItemLimit: Limits the number of items in a batch request.
  • batchResponseLimit: Limits the total size of responses for a batch request.
  • httpBodyLimit: Limits the size of HTTP request bodies.
/// ---go-ethereum/rpc/server.go---
// Server is an RPC server.
type Server struct {
	services serviceRegistry
	idgen    func() ID // a function generates a random IDs.

	mutex              sync.Mutex
	codecs             map[ServerCodec]struct{}
	run                atomic.Bool
	batchItemLimit     int
	batchResponseLimit int
	httpBodyLimit      int
}

// NewServer creates a new server instance with no registered handlers.
func NewServer() *Server {
	server := &Server{
		idgen:         randomIDGenerator(),
		codecs:        make(map[ServerCodec]struct{}),
		httpBodyLimit: defaultBodyLimit,
	}
	server.run.Store(true)
	// 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
}

// ---go-ethereum/rpc/service.go---
type serviceRegistry struct {
	mu       sync.Mutex
	services map[string]service
}

// service represents a registered object.
type service struct {
	name          string               // name for service
	callbacks     map[string]*callback // registered handlers
	subscriptions map[string]*callback // available subscriptions/notifications
}

/// ---go-ethereum/rpc/http.go---
const defaultBodyLimit = 5 * 1024 * 1024

/// ---go-ethereum/rpc/subscription.go---
// ID defines a pseudo random number that is used to identify RPC subscriptions.
type ID string

/// ---go-ethereum/rpc/types.go---
// ServerCodec implements reading, parsing and writing RPC messages for the server side of
// an RPC session. Implementations must be go-routine safe since the codec can be called in
// multiple go-routines concurrently.
type ServerCodec interface {
	peerInfo() PeerInfo
	readBatch() (msgs []*jsonrpcMessage, isBatch bool, err error)
	close()

	jsonWriter
}

Register Service

Server.RegisterName is used to register mthods for a server instance.
Function Signature
  • name: A string representing the name under which the service will be registered.
  • receiver: An interface representing the receiver object that contains the methods to be exposed as RPC methods.
The receiver’s qualified exported methods will be regisered under the name. For example, method eth_getBalance represents name eth and receiver’s getBalance function.
 
Server.RegisterName calls Server.services.registerName to register methods on Server.services.
// 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 an RPC method or a
// subscription an error is returned. Otherwise a new service is created and added to the
// service collection this server provides to clients.
func (s *Server) RegisterName(name string, receiver interface{}) error {
	return s.services.registerName(name, receiver)
}
 
For example, in the test newTestServer , it initializes objects testService and notificationTestService, then register them under name space test and nftest respectively. We can see there are many exposed methods defined on testService and notificationTestService.
Server.services.registerName will analyze all exposed methods of the testService and notificationTestService, register them on Server.services. So that client can specify the rpc name, method and parameters to call those methods.
/// ---go-ethereum/rpc/testservice_test.go---

type testService struct{}

type notificationTestService struct {
	unsubscribed            chan string
	gotHangSubscriptionReq  chan struct{}
	unblockHangSubscription chan struct{}
}

func newTestServer() *Server {
	server := NewServer()
	server.idgen = sequentialIDGenerator()
	if err := server.RegisterName("test", new(testService)); err != nil {
		panic(err)
	}
	if err := server.RegisterName("nftest", new(notificationTestService)); err != nil {
		panic(err)
	}
	return server
}

func (s *testService) PeerInfo(ctx context.Context) PeerInfo {
	return PeerInfoFromContext(ctx)
}

func (s *testService) Sleep(ctx context.Context, duration time.Duration) {
	time.Sleep(duration)
}

// ...

func (s *notificationTestService) Echo(i int) int {
	return i
}

func (s *notificationTestService) Unsubscribe(subid string) {
	if s.unsubscribed != nil {
		s.unsubscribed <- subid
	}
}

// ...
 
Server.services.registerName retgistes valid exposed method of receiver object into serviceRegistry.services.
1. Validate the Service Name
  • The function starts by converting the rcvr interface to a reflect.Value (rcvrVal).
  • It then checks if the provided service name is empty. If it is, it returns an error indicating that a service name is required.
2. Identify Suitable Callbacks
  • The suitableCallbacks function is called to iterate over the methods of the receiver object and determine which methods are suitable for RPC callbacks or subscriptions.
  • If no suitable methods are found, it returns an error indicating that the receiver object does not have any methods that can be exposed as RPC methods or subscriptions.
3. Acquire Lock and Initialize Services Map
  • A mutex lock is acquired to ensure that the registration process is thread-safe.
  • The services map is initialized if it has not been initialized already.
4. Check if Service Already Exists
  • The function checks if a service with the given name already exists in the services map.
  • If it does not exist, a new service struct is created and added to the services map.
5. Register Callbacks and Subscriptions
  • The function iterates over the identified callbacks.
  • If a callback is a subscription (determined by the isSubscribe field), it is added to the subscriptions map of the service.
  • Otherwise, it is added to the callbacks map of the service.
6. Return Success Signal Nil
/// ---go-ethereum/rpc/service.go---

type serviceRegistry struct {
	mu       sync.Mutex
	services map[string]service
}

func (r *serviceRegistry) registerName(name string, rcvr interface{}) error {
	// Validate the Service Name
	rcvrVal := reflect.ValueOf(rcvr)
	if name == "" {
		return fmt.Errorf("no service name for type %s", rcvrVal.Type().String())
	}
	
	// Identify Suitable Callbacks
	callbacks := suitableCallbacks(rcvrVal)
	if len(callbacks) == 0 {
		return fmt.Errorf("service %T doesn't have any suitable methods/subscriptions to expose", rcvr)
	}

	// Acquire Lock and Initialize Services Map
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.services == nil {
		r.services = make(map[string]service)
	}
	
	// Check if Service Already Exists
	svc, ok := r.services[name]
	if !ok {
		svc = service{
			name:          name,
			callbacks:     make(map[string]*callback),
			subscriptions: make(map[string]*callback),
		}
		r.services[name] = svc
	}
	
	// Register Callbacks and Subscriptions
	for name, cb := range callbacks {
		if cb.isSubscribe {
			svc.subscriptions[name] = cb
		} else {
			svc.callbacks[name] = cb
		}
	}
	
	// return success signal
	return nil
}
 
suitableCallbacks iterates over the methods of the given type. It determines if a method satisfies the criteria for an RPC callback or a subscription callback and adds it to the collection of callbacks.
Function Signature
  • receiver: A reflect.Value representing the receiver object whose methods are to be analyzed.
  • Returns: A map where the keys are method names (formatted) and the values are pointers to callback objects representing the suitable methods.
1.Retrieve the Type of the Receiver
  • The Type method of reflect.Value is called to get the reflect.Type of the receiver object. This type information is necessary to iterate over the methods of the receiver.
2.Initialize the Callbacks Map
  • A map is initialized to store the suitable methods. The map's keys will be the formatted method names, and the values will be pointers to callback objects representing the methods.
3.Iterate Over the Methods of the Receiver
  • A for loop is used to iterate over the methods of the receiver type. The NumMethod method returns the number of methods in the type, and Method(m) retrieves the m-th method.
4.Check if the Method is Exported
  • The PkgPath field of reflect.Method contains the package path of the method. If this field is not an empty string, the method is not exported and thus not suitable for RPC. The function continues to the next method if the current method is not exported.
5.Create a Callback Object
  • The newCallback function is called to create a callback object for the method. If newCallback returns nil, the method is deemed invalid as an RPC callback, and the function continues to the next method.
6.Format the Method Name
  • The formatName function is called to format the method name. This typically involves converting the first character of the method name to lowercase to follow a common naming convention for RPC methods.
7.Add the Callback to the Map
  • The formatted method name and the callback object are added to the callbacks map.
8.Return the Map of Callbacks
  • The map of suitable callbacks is returned.
/// ---go-ethereum/rpc/service.go---

// callback is a method callback which was registered in the server
type callback struct {
	fn          reflect.Value  // the function
	rcvr        reflect.Value  // receiver object of method, set if fn is method
	argTypes    []reflect.Type // input argument types
	hasCtx      bool           // method's first argument is a context (not included in argTypes)
	errPos      int            // err return idx, of -1 when method cannot return error
	isSubscribe bool           // true if this is a subscription callback
}

func suitableCallbacks(receiver reflect.Value) map[string]*callback {
	// Retrieve the Type of the Receiver
	typ := receiver.Type()
	
	// Initialize the Callbacks Map
	callbacks := make(map[string]*callback)
	
	// Iterate Over the Methods of the Receiver
	for m := 0; m < typ.NumMethod(); m++ {
		// Check if the Method is Exported
		method := typ.Method(m)
		if method.PkgPath != "" {
			continue // method not exported
		}
		
		// Create a Callback Object
		cb := newCallback(receiver, method.Func)
		if cb == nil {
			continue // function invalid
		}
		
		// Format the Method Name
		name := formatName(method.Name)
		
		// Add the Callback to the Map
		callbacks[name] = cb
	}
	
	// Return the Map of Callbacks
	return callbacks
}
 
The newCallback function is responsible for creating a callback object from a given function. This callback object encapsulates the necessary information for invoking the function as an RPC method.
Function Signature
  • receiver: A reflect.Value representing the receiver object of the method, if the function is a method. Otherwise, it's an invalid reflect.Value.
  • fn: A reflect.Value representing the function to be wrapped in a callback.
Return Value
  • Returns a pointer to a callback object if the function is suitable for RPC.
  • Returns nil if the function is unsuitable for RPC.
1. Determine the Function Type
  • fn.Type() retrieves the reflect.Type of the function. This type information is essential for analyzing the function's parameters and return values.
2. Initialize the Callback Struct
  • A new callback struct is created and initialized with the function fn, the receiver receiver, and default values for errPos and isSubscribe.
  • errPos is initialized to -1, indicating that the function does not return an error by default.
  • isSubscribe is determined by calling isPubSub(fntype), which checks if the function is a subscription callback.
3. Determine Parameter Types
  • c.makeArgTypes() is called to analyze and store the parameter types of the function. This method fills the argTypes slice of the callback struct with the types of the function's parameters, excluding the receiver and context.Context parameter if present.
4. Verify Return Types
  • The function's return types are retrieved and stored in the outs slice.
  • If the function returns more than two values, it is considered unsuitable for RPC, and nil is returned.
5. Check Return Types for Errors
  • If the function returns a single value, it checks if it is an error type using isErrorType. If so, it sets c.errPos to 0, indicating that the function returns an error as its only return value.
  • If the function returns two values, it checks that the first value is not an error and the second value is an error. If this condition is met, it sets c.errPos to 1, indicating that the second return value is an error.
  • If neither condition is met, the function is considered unsuitable for RPC, and nil is returned.
6. Return the Callback
/// ---go-ethereum/rpc/service.go---

// callback is a method callback which was registered in the server
type callback struct {
	fn          reflect.Value  // the function
	rcvr        reflect.Value  // receiver object of method, set if fn is method
	argTypes    []reflect.Type // input argument types
	hasCtx      bool           // method's first argument is a context (not included in argTypes)
	errPos      int            // err return idx, of -1 when method cannot return error
	isSubscribe bool           // true if this is a subscription callback
}

// newCallback turns fn (a function) into a callback object. It returns nil if the function
// is unsuitable as an RPC callback.
func newCallback(receiver, fn reflect.Value) *callback {
	// Determine the Function Type
	fntype := fn.Type()
	
	// Initialize the Callback Struct
	c := &callback{fn: fn, rcvr: receiver, errPos: -1, isSubscribe: isPubSub(fntype)}
	
	// Determine parameter types. They must all be exported or builtin types.
	c.makeArgTypes()

	// Verify return types. The function must return at most one error
	// and/or one other non-error value.
	outs := make([]reflect.Type, fntype.NumOut())
	for i := 0; i < fntype.NumOut(); i++ {
		outs[i] = fntype.Out(i)
	}
	if len(outs) > 2 {
		return nil
	}
	
	// Check Return Types for Errors
	// If an error is returned, it must be the last returned value.
	switch {
	case len(outs) == 1 && isErrorType(outs[0]):
		c.errPos = 0
	case len(outs) == 2:
		if isErrorType(outs[0]) || !isErrorType(outs[1]) {
			return nil
		}
		c.errPos = 1
	}
	return c
}

// isPubSub tests whether the given method's first argument is a context.Context and
// returns the pair (Subscription, error).
func isPubSub(methodType reflect.Type) bool {
	// numIn(0) is the receiver type
	if methodType.NumIn() < 2 || methodType.NumOut() != 2 {
		return false
	}
	return methodType.In(1) == contextType &&
		isSubscriptionType(methodType.Out(0)) &&
		isErrorType(methodType.Out(1))
}


// makeArgTypes composes the argTypes list.
func (c *callback) makeArgTypes() {
	fntype := c.fn.Type()
	// Skip receiver and context.Context parameter (if present).
	firstArg := 0
	if c.rcvr.IsValid() {
		firstArg++
	}
	if fntype.NumIn() > firstArg && fntype.In(firstArg) == contextType {
		c.hasCtx = true
		firstArg++
	}
	// Add all remaining parameters.
	c.argTypes = make([]reflect.Type, fntype.NumIn()-firstArg)
	for i := firstArg; i < fntype.NumIn(); i++ {
		c.argTypes[i-firstArg] = fntype.In(i)
	}
}

Handle RPC

Server.ServeCodec is responsible for handling incoming RPC requests using a specified codec, invoking the appropriate callbacks, and sending back responses.
A codec is an interface that abstracts the details of how RPC messages are encoded, decoded, sent, and received over a transport layer. The codec handles the specifics of communication protocols like HTTP, WebSocket, or IPC (Inter-Process Communication).
ServeCodec initializes a client based on a codec to receive and handle rpc requests continuously.
/// ---go-ethereum/rpc/server.go---

// ServeCodec reads incoming requests from codec, calls the appropriate callback and writes
// the response back using the given codec. It will block until the codec is closed or the
// server is stopped. In either case the codec is closed.
//
// Note that codec options are no longer supported.
func (s *Server) ServeCodec(codec ServerCodec, options CodecOption) {
	defer codec.close()

	if !s.trackCodec(codec) {
		return
	}
	defer s.untrackCodec(codec)

	cfg := &clientConfig{
		idgen:              s.idgen,
		batchItemLimit:     s.batchItemLimit,
		batchResponseLimit: s.batchResponseLimit,
	}
	c := initClient(codec, &s.services, cfg)
	<-codec.closed()
	c.Close()
}
/// ---go-ethereum/rpc/client.go---

func initClient(conn ServerCodec, services *serviceRegistry, cfg *clientConfig) *Client {
	_, isHTTP := conn.(*httpConn)
	c := &Client{
		isHTTP:               isHTTP,
		services:             services,
		idgen:                cfg.idgen,
		batchItemLimit:       cfg.batchItemLimit,
		batchResponseMaxSize: cfg.batchResponseLimit,
		writeConn:            conn,
		close:                make(chan struct{}),
		closing:              make(chan struct{}),
		didClose:             make(chan struct{}),
		reconnected:          make(chan ServerCodec),
		readOp:               make(chan readOp),
		readErr:              make(chan error),
		reqInit:              make(chan *requestOp),
		reqSent:              make(chan error, 1),
		reqTimeout:           make(chan *requestOp),
	}

	// Set defaults.
	if c.idgen == nil {
		c.idgen = randomIDGenerator()
	}

	// Launch the main loop.
	if !isHTTP {
		go c.dispatch(conn)
	}
	return c
}
 
Client.dispatch is the main loop for handling client operations, including reading and writing messages based on codec, handling errors, and reconnecting as necessary.
/// ---go-ethereum/rpc/client.go---

// dispatch is the main loop of the client.
// It sends read messages to waiting calls to Call and BatchCall
// and subscription notifications to registered subscriptions.
func (c *Client) dispatch(codec ServerCodec) {
	var (
		lastOp      *requestOp  // tracks last send operation
		reqInitLock = c.reqInit // nil while the send lock is held
		conn        = c.newClientConn(codec)
		reading     = true
	)
	defer func() {
		close(c.closing)
		if reading {
			conn.close(ErrClientQuit, nil)
			c.drainRead()
		}
		close(c.didClose)
	}()

	// Spawn the initial read loop.
	go c.read(codec)

	for {
		select {
		case <-c.close:
			return

		// Read path:
		case op := <-c.readOp:
			if op.batch {
				conn.handler.handleBatch(op.msgs)
			} else {
				conn.handler.handleMsg(op.msgs[0])
			}

		case err := <-c.readErr:
			conn.handler.log.Debug("RPC connection read error", "err", err)
			conn.close(err, lastOp)
			reading = false

		// Reconnect:
		case newcodec := <-c.reconnected:
			log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.remoteAddr())
			if reading {
				// Wait for the previous read loop to exit. This is a rare case which
				// happens if this loop isn't notified in time after the connection breaks.
				// In those cases the caller will notice first and reconnect. Closing the
				// handler terminates all waiting requests (closing op.resp) except for
				// lastOp, which will be transferred to the new handler.
				conn.close(errClientReconnected, lastOp)
				c.drainRead()
			}
			go c.read(newcodec)
			reading = true
			conn = c.newClientConn(newcodec)
			// Re-register the in-flight request on the new handler
			// because that's where it will be sent.
			conn.handler.addRequestOp(lastOp)

		// Send path:
		case op := <-reqInitLock:
			// Stop listening for further requests until the current one has been sent.
			reqInitLock = nil
			lastOp = op
			conn.handler.addRequestOp(op)

		case err := <-c.reqSent:
			if err != nil {
				// Remove response handlers for the last send. When the read loop
				// goes down, it will signal all other current operations.
				conn.handler.removeRequestOp(lastOp)
			}
			// Let the next request in.
			reqInitLock = c.reqInit
			lastOp = nil

		case op := <-c.reqTimeout:
			conn.handler.removeRequestOp(op)
		}
	}
}

// read decodes RPC messages from a codec, feeding them into dispatch.
func (c *Client) read(codec ServerCodec) {
	for {
		msgs, batch, err := codec.readBatch()
		if _, ok := err.(*json.SyntaxError); ok {
			msg := errorMessage(&parseError{err.Error()})
			codec.writeJSON(context.Background(), msg, true)
		}
		if err != nil {
			c.readErr <- err
			return
		}
		c.readOp <- readOp{msgs, batch}
	}
}
 
In the dispatch function, a go routine executes read function which keeps reading incoming rpc requests and stores them in the c.readOp channel.
/// ---go-ethereum/rpc/client.go---

// read decodes RPC messages from a codec, feeding them into dispatch.
func (c *Client) read(codec ServerCodec) {
	for {
		msgs, batch, err := codec.readBatch()
		if _, ok := err.(*json.SyntaxError); ok {
			msg := errorMessage(&parseError{err.Error()})
			codec.writeJSON(context.Background(), msg, true)
		}
		if err != nil {
			c.readErr <- err
			return
		}
		c.readOp <- readOp{msgs, batch}
	}
}
 
When the c.readOp channel receives message, it will call handleBatch or handleMsg to handle rpc requests in c.dispatch.
/// ---go-ethereum/rpc/client.go---

func (c *Client) dispatch(codec ServerCodec) {
	// ...
	case op := <-c.readOp:
		if op.batch {
			conn.handler.handleBatch(op.msgs)
		} else {
			conn.handler.handleMsg(op.msgs[0])
		}
	// ...
}
 
The handleMsg function is responsible for handling a single non-batch JSON-RPC message.
/// ---go-ethereum/rpc/handler.go---

// handleMsg handles a single non-batch message.
func (h *handler) handleMsg(msg *jsonrpcMessage) {
	msgs := []*jsonrpcMessage{msg}
	h.handleResponses(msgs, func(msg *jsonrpcMessage) {
		h.startCallProc(func(cp *callProc) {
			h.handleNonBatchCall(cp, msg)
		})
	})
}
/// ---go-ethereum/rpc/handler.go---

func (h *handler) handleNonBatchCall(cp *callProc, msg *jsonrpcMessage) {
	var (
		responded sync.Once
		timer     *time.Timer
		cancel    context.CancelFunc
	)
	cp.ctx, cancel = context.WithCancel(cp.ctx)
	defer cancel()

	// Cancel the request context after timeout and send an error response. Since the
	// running method might not return immediately on timeout, we must wait for the
	// timeout concurrently with processing the request.
	if timeout, ok := ContextRequestTimeout(cp.ctx); ok {
		timer = time.AfterFunc(timeout, func() {
			cancel()
			responded.Do(func() {
				resp := msg.errorResponse(&internalServerError{errcodeTimeout, errMsgTimeout})
				h.conn.writeJSON(cp.ctx, resp, true)
			})
		})
	}

	answer := h.handleCallMsg(cp, msg)
	if timer != nil {
		timer.Stop()
	}
	h.addSubscriptions(cp.notifiers)
	if answer != nil {
		responded.Do(func() {
			h.conn.writeJSON(cp.ctx, answer, false)
		})
	}
	for _, n := range cp.notifiers {
		n.activate()
	}
}
 
handleCallMsg executes a call message and returns the answer.
/// ---go-ethereum/rpc/handler.go---

func (h *handler) handleCallMsg(ctx *callProc, msg *jsonrpcMessage) *jsonrpcMessage {
	start := time.Now()
	switch {
	case msg.isNotification():
		h.handleCall(ctx, msg)
		h.log.Debug("Served "+msg.Method, "duration", time.Since(start))
		return nil

	case msg.isCall():
		resp := h.handleCall(ctx, msg)
		var ctx []interface{}
		ctx = append(ctx, "reqid", idForLog{msg.ID}, "duration", time.Since(start))
		if resp.Error != nil {
			ctx = append(ctx, "err", resp.Error.Message)
			if resp.Error.Data != nil {
				ctx = append(ctx, "errdata", resp.Error.Data)
			}
			h.log.Warn("Served "+msg.Method, ctx...)
		} else {
			h.log.Debug("Served "+msg.Method, ctx...)
		}
		return resp

	case msg.hasValidID():
		return msg.errorResponse(&invalidRequestError{"invalid request"})

	default:
		return errorMessage(&invalidRequestError{"invalid request"})
	}
}
 
handleCall processes method calls. It loads callback function from handler’s serviceRegistry and run the callback with args to get result.

/// ---go-ethereum/rpc/handler.go---

func (h *handler) handleCall(cp *callProc, msg *jsonrpcMessage) *jsonrpcMessage {
	if msg.isSubscribe() {
		return h.handleSubscribe(cp, msg)
	}
	var callb *callback
	if msg.isUnsubscribe() {
		callb = h.unsubscribeCb
	} else {
		callb = h.reg.callback(msg.Method)
	}
	if callb == nil {
		return msg.errorResponse(&methodNotFoundError{method: msg.Method})
	}

	args, err := parsePositionalArguments(msg.Params, callb.argTypes)
	if err != nil {
		return msg.errorResponse(&invalidParamsError{err.Error()})
	}
	start := time.Now()
	answer := h.runMethod(cp.ctx, msg, callb, args)

	// Collect the statistics for RPC calls if metrics is enabled.
	// We only care about pure rpc call. Filter out subscription.
	if callb != h.unsubscribeCb {
		rpcRequestGauge.Inc(1)
		if answer.Error != nil {
			failedRequestGauge.Inc(1)
		} else {
			successfulRequestGauge.Inc(1)
		}
		rpcServingTimer.UpdateSince(start)
		updateServeTimeHistogram(msg.Method, answer.Error == nil, time.Since(start))
	}

	return answer
}

// runMethod runs the Go callback for an RPC method.
func (h *handler) runMethod(ctx context.Context, msg *jsonrpcMessage, callb *callback, args []reflect.Value) *jsonrpcMessage {
	result, err := callb.call(ctx, msg.Method, args)
	if err != nil {
		return msg.errorResponse(err)
	}
	return msg.response(result)
}
 
serviceRegistry.callback cut the method to get namespace and corresponding function name.
For example, eth_getBalance will be cut and get eth and getBalance.
/// ---go-ethereum/rpc/service.go---

const serviceMethodSeparator   = "_"

// callback returns the callback corresponding to the given RPC method name.
func (r *serviceRegistry) callback(method string) *callback {
	before, after, found := strings.Cut(method, serviceMethodSeparator)
	if !found {
		return nil
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.services[before].callbacks[after]
}
 
callback.call invokes the underlying function and returns result.
/// ---go-ethereum/rpc/service.go---

// call invokes the callback.
func (c *callback) call(ctx context.Context, method string, args []reflect.Value) (res interface{}, errRes error) {
	// Create the argument slice.
	fullargs := make([]reflect.Value, 0, 2+len(args))
	if c.rcvr.IsValid() {
		fullargs = append(fullargs, c.rcvr)
	}
	if c.hasCtx {
		fullargs = append(fullargs, reflect.ValueOf(ctx))
	}
	fullargs = append(fullargs, args...)

	// Catch panic while running the callback.
	defer func() {
		if err := recover(); err != nil {
			const size = 64 << 10
			buf := make([]byte, size)
			buf = buf[:runtime.Stack(buf, false)]
			log.Error("RPC method " + method + " crashed: " + fmt.Sprintf("%v\n%s", err, buf))
			errRes = &internalServerError{errcodePanic, "method handler crashed"}
		}
	}()
	// Run the callback.
	results := c.fn.Call(fullargs)
	if len(results) == 0 {
		return nil, nil
	}
	if c.errPos >= 0 && !results[c.errPos].IsNil() {
		// Method has returned non-nil error value.
		err := results[c.errPos].Interface().(error)
		return reflect.Value{}, err
	}
	return results[0].Interface(), nil
}