以太坊RPC机制

geth --networkid 666 --datadir /home/ubuntu/Private_eth/eth1 --identity "node1" --rpc --rpcport "8545" --rpcaddr "192.168.174.212" --nodiscover --rpcapi "eth,net,web3,txpool,debug,miner" 

#!/usr/bin/env python3
import requests
URL = "http://192.168.174.212:8545/"
data = {
   "jsonrpc": "2.0", 
   "method": "eth_getBalance", 
   "params":["0x578efd53cf8342f4f5acfb6ee0ce9c7b3cfe2252", "latest"],
   "id":0 
}
response = requests.post(url=URL,json=data)
print(response.json())

// filedir:go-ethereum-1.10.2\cmd\geth\main.go  L308
// geth is the main entry point into the system if no special subcommand is ran.
// It creates a default node based on the command line arguments and runs it in
// blocking mode, waiting for it to be shut down.
func geth(ctx *cli.Context) error {
  if args := ctx.Args(); len(args) > 0 {
    return fmt.Errorf("invalid command: %q", args[0])
  }

  prepare(ctx)
  stack, backend := makeFullNode(ctx)
  defer stack.Close()

  startNode(ctx, stack, backend)
  stack.Wait()
  return nil
}

// filedir:go-ethereum-1.10.2\cmd\geth\main.go  L325
// startNode boots up the system node and all registered protocols, after which
// it unlocks any requested accounts, and starts the RPC/IPC interfaces and the
// miner.
func startNode(ctx *cli.Context, stack *node.Node, backend ethapi.Backend) {
  debug.Memsize.Add("node", stack)

  // Start up the node itself
  utils.StartNode(ctx, stack)
......

func StartNode(ctx *cli.Context, stack *node.Node) {
  if err := stack.Start(); err != nil {
    Fatalf("Error starting protocol stack: %v", err)
  }
  go func() {
    sigc := make(chan os.Signal, 1)
    signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM)
    defer signal.Stop(sigc)

    minFreeDiskSpace := ethconfig.Defaults.TrieDirtyCache
    if ctx.GlobalIsSet(MinFreeDiskSpaceFlag.Name) {
      minFreeDiskSpace = ctx.GlobalInt(MinFreeDiskSpaceFlag.Name)
    } else if ctx.GlobalIsSet(CacheFlag.Name) || ctx.GlobalIsSet(CacheGCFlag.Name) {
      minFreeDiskSpace = ctx.GlobalInt(CacheFlag.Name) * ctx.GlobalInt(CacheGCFlag.Name) / 100
    }
    if minFreeDiskSpace > 0 {
      go monitorFreeDiskSpace(sigc, stack.InstanceDir(), uint64(minFreeDiskSpace)*1024*1024)
    }

    <-sigc
    log.Info("Got interrupt, shutting down...")
    go stack.Close()
    for i := 10; i > 0; i-- {
      <-sigc
      if i > 1 {
        log.Warn("Already shutting down, interrupt more to panic.", "times", i-1)
      }
    }
    debug.Exit() // ensure trace and CPU profile data is flushed.
    debug.LoudPanic("boom")
  }()
}

// filedir: go-ethereum-1.10.2\node\node.go
// Start starts all registered lifecycles, RPC services and p2p networking.
// Node can only be started once.
func (n *Node) Start() error {
  n.startStopLock.Lock()
  defer n.startStopLock.Unlock()

  n.lock.Lock()
  switch n.state {
  case runningState:
    n.lock.Unlock()
    return ErrNodeRunning
  case closedState:
    n.lock.Unlock()
    return ErrNodeStopped
  }
  n.state = runningState
  // open networking and RPC endpoints
  err := n.openEndpoints()
  lifecycles := make([]Lifecycle, len(n.lifecycles))
  copy(lifecycles, n.lifecycles)
  n.lock.Unlock()

  // Check if endpoint startup failed.
  if err != nil {
    n.doClose(nil)
    return err
  }
  // Start all registered lifecycles.
  var started []Lifecycle
  for _, lifecycle := range lifecycles {
    if err = lifecycle.Start(); err != nil {
      break
    }
    started = append(started, lifecycle)
  }
  // Check if any lifecycle failed to start.
  if err != nil {
    n.stopServices(started)
    n.doClose(nil)
  }
  return err
}

// filedir:go-ethereum-1.10.2\node\node.go   L260
// openEndpoints starts all network and RPC endpoints.
func (n *Node) openEndpoints() error {
  // start networking endpoints
  n.log.Info("Starting peer-to-peer node", "instance", n.server.Name)
  if err := n.server.Start(); err != nil {
    return convertFileLockError(err)
  }
  // start RPC endpoints
  err := n.startRPC()
  if err != nil {
    n.stopRPC()
    n.server.Stop()
  }
  return err
}

// filedir: go-ethereum-1.10.2\node\node.go
// configureRPC is a helper method to configure all the various RPC endpoints during node
// startup. It's not meant to be called at any time afterwards as it makes certain
// assumptions about the state of the node.
func (n *Node) startRPC() error {
  if err := n.startInProc(); err != nil {
    return err
  }

  // Configure IPC.
  if n.ipc.endpoint != "" {
    if err := n.ipc.start(n.rpcAPIs); err != nil {
      return err
    }
  }

  // Configure HTTP.
  if n.config.HTTPHost != "" {
    config := httpConfig{
      CorsAllowedOrigins: n.config.HTTPCors,
      Vhosts:             n.config.HTTPVirtualHosts,
      Modules:            n.config.HTTPModules,
      prefix:             n.config.HTTPPathPrefix,
    }
    if err := n.http.setListenAddr(n.config.HTTPHost, n.config.HTTPPort); err != nil {
      return err
    }
    if err := n.http.enableRPC(n.rpcAPIs, config); err != nil {
      return err
    }
  }

  // Configure WebSocket.
  if n.config.WSHost != "" {
    server := n.wsServerForPort(n.config.WSPort)
    config := wsConfig{
      Modules: n.config.WSModules,
      Origins: n.config.WSOrigins,
      prefix:  n.config.WSPathPrefix,
    }
    if err := server.setListenAddr(n.config.WSHost, n.config.WSPort); err != nil {
      return err
    }
    if err := server.enableWS(n.rpcAPIs, config); err != nil {
      return err
    }
  }

  if err := n.http.start(); err != nil {
    return err
  }
  return n.ws.start()
}

// startInProc registers all RPC APIs on the inproc server.
func (n *Node) startInProc() error {
  for _, api := range n.rpcAPIs {
    if err := n.inprocHandler.RegisterName(api.Namespace, api.Service); err != nil {
      return err
    }
  }
  return nil
}

// filedir: go-ethereum-1.10.2\node\rpcstack.go
// Start starts the httpServer's http.Server
func (is *ipcServer) start(apis []rpc.API) error {
  is.mu.Lock()
  defer is.mu.Unlock()

  if is.listener != nil {
    return nil // already running
  }
  listener, srv, err := rpc.StartIPCEndpoint(is.endpoint, apis)
  if err != nil {
    is.log.Warn("IPC opening failed", "url", is.endpoint, "error", err)
    return err
  }
  is.log.Info("IPC endpoint opened", "url", is.endpoint)
  is.listener, is.srv = listener, srv
  return nil
}

// filedir:go-ethereum-1.10.2\node\rpcstack.go    L272
// enableRPC turns on JSON-RPC over HTTP on the server.
func (h *httpServer) enableRPC(apis []rpc.API, config httpConfig) error {
  h.mu.Lock()
  defer h.mu.Unlock()

  if h.rpcAllowed() {
    return fmt.Errorf("JSON-RPC over HTTP is already enabled")
  }

  // Create RPC server and handler.
  srv := rpc.NewServer()
  if err := RegisterApisFromWhitelist(apis, config.Modules, srv, false); err != nil {
    return err
  }
  h.httpConfig = config
  h.httpHandler.Store(&rpcHandler{
    Handler: NewHTTPHandlerStack(srv, config.CorsAllowedOrigins, config.Vhosts),
    server:  srv,
  })
  return nil
}

// filedir: go-ethereum-1.10.2\node\rpcstack.go
// NewHTTPHandlerStack returns wrapped http-related handlers
func NewHTTPHandlerStack(srv http.Handler, cors []string, vhosts []string) http.Handler {
  // Wrap the CORS-handler within a host-handler
  handler := newCorsHandler(srv, cors)
  handler = newVHostHandler(vhosts, handler)
  return newGzipHandler(handler)
}

// filedir:go-ethereum-1.10.2\node\rpcstack.go
func newGzipHandler(next http.Handler) http.Handler {
  return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    if !strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") {
      next.ServeHTTP(w, r)
      return
    }

    w.Header().Set("Content-Encoding", "gzip")

    gz := gzPool.Get().(*gzip.Writer)
    defer gzPool.Put(gz)

    gz.Reset(w)
    defer gz.Close()

    next.ServeHTTP(&gzipResponseWriter{ResponseWriter: w, Writer: gz}, r)
  })
}
// filedir: go-ethereum-1.10.2\rpc\http.go
// ServeHTTP serves JSON-RPC requests over HTTP.
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  // Permit dumb empty requests for remote health-checks (AWS)
  if r.Method == http.MethodGet && r.ContentLength == 0 && r.URL.RawQuery == "" {
    w.WriteHeader(http.StatusOK)
    return
  }
  if code, err := validateRequest(r); err != nil {
    http.Error(w, err.Error(), code)
    return
  }
  // All checks passed, create a codec that reads directly from the request body
  // until EOF, writes the response to w, and orders the server to process a
  // single request.
  ctx := r.Context()
  ctx = context.WithValue(ctx, "remote", r.RemoteAddr)
  ctx = context.WithValue(ctx, "scheme", r.Proto)
  ctx = context.WithValue(ctx, "local", r.Host)
  if ua := r.Header.Get("User-Agent"); ua != "" {
    ctx = context.WithValue(ctx, "User-Agent", ua)
  }
  if origin := r.Header.Get("Origin"); origin != "" {
    ctx = context.WithValue(ctx, "Origin", origin)
  }

  w.Header().Set("content-type", contentType)
  codec := newHTTPServerConn(r, w)
  defer codec.close()
  s.serveSingleRequest(ctx, codec)
}

// filedir:go-ethereum-1.10.2\node\rpcstack.go
// enableWS turns on JSON-RPC over WebSocket on the server.
func (h *httpServer) enableWS(apis []rpc.API, config wsConfig) error {
  h.mu.Lock()
  defer h.mu.Unlock()

  if h.wsAllowed() {
    return fmt.Errorf("JSON-RPC over WebSocket is already enabled")
  }

  // Create RPC server and handler.
  srv := rpc.NewServer()
  if err := RegisterApisFromWhitelist(apis, config.Modules, srv, false); err != nil {
    return err
  }
  h.wsConfig = config
  h.wsHandler.Store(&rpcHandler{
    Handler: srv.WebsocketHandler(config.Origins),
    server:  srv,
  })
  return nil
}

// filedir:  go-ethereum-1.10.2\rpc\websocket.go L45
// WebsocketHandler returns a handler that serves JSON-RPC to WebSocket connections.
//
// allowedOrigins should be a comma-separated list of allowed origin URLs.
// To allow connections with any origin, pass "*".
func (s *Server) WebsocketHandler(allowedOrigins []string) http.Handler {
  var upgrader = websocket.Upgrader{
    ReadBufferSize:  wsReadBuffer,
    WriteBufferSize: wsWriteBuffer,
    WriteBufferPool: wsBufferPool,
    CheckOrigin:     wsHandshakeValidator(allowedOrigins),
  }
  return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
      log.Debug("WebSocket upgrade failed", "err", err)
      return
    }
    codec := newWebsocketCodec(conn)
    s.ServeCodec(codec, 0)
  })
}

// filedir: go-ethereum-1.10.2\rpc\websocket.go L241
func newWebsocketCodec(conn *websocket.Conn) ServerCodec {
  conn.SetReadLimit(wsMessageSizeLimit)
  wc := &websocketCodec{
    jsonCodec: NewFuncCodec(conn, conn.WriteJSON, conn.ReadJSON).(*jsonCodec),
    conn:      conn,
    pingReset: make(chan struct{}, 1),
  }
  wc.wg.Add(1)
  go wc.pingLoop()
  return wc
}
// filedir: go-ethereum-1.10.2\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()

  // Don't serve if server is stopped.
  if atomic.LoadInt32(&s.run) == 0 {
    return
  }

  // Add the codec to the set so it can be closed by Stop.
  s.codecs.Add(codec)
  defer s.codecs.Remove(codec)

  c := initClient(codec, s.idgen, &s.services)
  <-codec.closed()
  c.Close()
}

// start starts the HTTP server if it is enabled and not already running.
func (h *httpServer) start() error {
  h.mu.Lock()
  defer h.mu.Unlock()

  if h.endpoint == "" || h.listener != nil {
    return nil // already running or not configured
  }

  // Initialize the server.
  h.server = &http.Server{Handler: h}
  if h.timeouts != (rpc.HTTPTimeouts{}) {
    CheckTimeouts(&h.timeouts)
    h.server.ReadTimeout = h.timeouts.ReadTimeout
    h.server.WriteTimeout = h.timeouts.WriteTimeout
    h.server.IdleTimeout = h.timeouts.IdleTimeout
  }

  // Start the server.
  listener, err := net.Listen("tcp", h.endpoint)
  if err != nil {
    // If the server fails to start, we need to clear out the RPC and WS
    // configuration so they can be configured another time.
    h.disableRPC()
    h.disableWS()
    return err
  }
  h.listener = listener
  go h.server.Serve(listener)

  if h.wsAllowed() {
    url := fmt.Sprintf("ws://%v", listener.Addr())
    if h.wsConfig.prefix != "" {
      url += h.wsConfig.prefix
    }
    h.log.Info("WebSocket enabled", "url", url)
  }
  // if server is websocket only, return after logging
  if !h.rpcAllowed() {
    return nil
  }
  // Log http endpoint.
  h.log.Info("HTTP server started",
    "endpoint", listener.Addr(),
    "prefix", h.httpConfig.prefix,
    "cors", strings.Join(h.httpConfig.CorsAllowedOrigins, ","),
    "vhosts", strings.Join(h.httpConfig.Vhosts, ","),
  )

  // Log all handlers mounted on server.
  var paths []string
  for path := range h.handlerNames {
    paths = append(paths, path)
  }
  sort.Strings(paths)
  logged := make(map[string]bool, len(paths))
  for _, path := range paths {
    name := h.handlerNames[path]
    if !logged[name] {
      log.Info(name+" enabled", "url", "http://"+listener.Addr().String()+path)
      logged[name] = true
    }
  }
  return nil
}

// filedir:go-ethereum-1.10.2\ethclient\ethclient.go  L351
// BalanceAt returns the wei balance of the given account.
// The block number can be nil, in which case the balance is taken from the latest known block.
func (ec *Client) BalanceAt(ctx context.Context, account common.Address, blockNumber *big.Int) (*big.Int, error) {
  var result hexutil.Big
  err := ec.c.CallContext(ctx, &result, "eth_getBalance", account, toBlockNumArg(blockNumber))
  return (*big.Int)(&result), err
}

// filedir: go-ethereum-1.10.2\rpc\client.go  L286
// CallContext performs a JSON-RPC call with the given arguments. If the context is
// canceled before the call has successfully returned, CallContext returns immediately.
//
// The result must be a pointer so that package json can unmarshal into it. You
// can also pass nil, in which case the result is ignored.
func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
  if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr {
    return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result)
  }
  msg, err := c.newMessage(method, args...)
  if err != nil {
    return err
  }
  op := &requestOp{ids: []json.RawMessage{msg.ID}, resp: make(chan *jsonrpcMessage, 1)}

  if c.isHTTP {
    err = c.sendHTTP(ctx, op, msg)
  } else {
    err = c.send(ctx, op, msg)
  }
  if err != nil {
    return err
  }

  // dispatch has accepted the request and will close the channel when it quits.
  switch resp, err := op.wait(ctx, c); {
  case err != nil:
    return err
  case resp.Error != nil:
    return resp.Error
  case len(resp.Result) == 0:
    return ErrNoResult
  default:
    return json.Unmarshal(resp.Result, &result)
  }
}

// filedir: go-ethereum-1.10.2\rpc\client.go
func (c *Client) newMessage(method string, paramsIn ...interface{}) (*jsonrpcMessage, error) {
  msg := &jsonrpcMessage{Version: vsn, ID: c.nextID(), Method: method}
  if paramsIn != nil { // prevent sending "params":null
    var err error
    if msg.Params, err = json.Marshal(paramsIn); err != nil {
      return nil, err
    }
  }
  return msg, nil
}

#!/usr/bin/env python3
import requests
URL = "http://192.168.174.212:8545/"
data = {
   "jsonrpc": "2.0", 
   "method": "eth_getBalance", 
   "params":["0x578efd53cf8342f4f5acfb6ee0ce9c7b3cfe2252", "latest"],
   "id":0 
}
response = requests.post(url=URL,json=data)
print(response.json())

// filedir: go-ethereum-1.10.2\rpc\http.go
func (c *Client) sendHTTP(ctx context.Context, op *requestOp, msg interface{}) error {
  hc := c.writeConn.(*httpConn)
  respBody, err := hc.doRequest(ctx, msg)
  if respBody != nil {
    defer respBody.Close()
  }

  if err != nil {
    if respBody != nil {
      buf := new(bytes.Buffer)
      if _, err2 := buf.ReadFrom(respBody); err2 == nil {
        return fmt.Errorf("%v: %v", err, buf.String())
      }
    }
    return err
  }
  var respmsg jsonrpcMessage
  if err := json.NewDecoder(respBody).Decode(&respmsg); err != nil {
    return err
  }
  op.resp <- &respmsg
  return nil
}

// filedir: go-ethereum-1.10.2\rpc\http.go  L175
func (hc *httpConn) doRequest(ctx context.Context, msg interface{}) (io.ReadCloser, error) {
  body, err := json.Marshal(msg)
  if err != nil {
    return nil, err
  }
  req, err := http.NewRequestWithContext(ctx, "POST", hc.url, ioutil.NopCloser(bytes.NewReader(body)))
  if err != nil {
    return nil, err
  }
  req.ContentLength = int64(len(body))

  // set headers
  hc.mu.Lock()
  req.Header = hc.headers.Clone()
  hc.mu.Unlock()

  // do request
  resp, err := hc.client.Do(req)
  if err != nil {
    return nil, err
  }
  if resp.StatusCode < 200 || resp.StatusCode >= 300 {
    return resp.Body, errors.New(resp.Status)
  }
  return resp.Body, nil
}

func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
  ......
    
  // dispatch has accepted the request and will close the channel when it quits.
  switch resp, err := op.wait(ctx, c); {
  case err != nil:
    return err
  case resp.Error != nil:
    return resp.Error
  case len(resp.Result) == 0:
    return ErrNoResult
  default:
    return json.Unmarshal(resp.Result, &result)
  }
}

func (op *requestOp) wait(ctx context.Context, c *Client) (*jsonrpcMessage, error) {
  select {
  case <-ctx.Done():
    // Send the timeout to dispatch so it can remove the request IDs.
    if !c.isHTTP {
      select {
      case c.reqTimeout <- op:
      case <-c.closing:
      }
    }
    return nil, ctx.Err()
  case resp := <-op.resp:
    return resp, op.err
  }
}

// filedir: go-ethereum-1.10.2\rpc\http.go  L225
// ServeHTTP serves JSON-RPC requests over HTTP.
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
  // Permit dumb empty requests for remote health-checks (AWS)
  if r.Method == http.MethodGet && r.ContentLength == 0 && r.URL.RawQuery == "" {
    w.WriteHeader(http.StatusOK)
    return
  }
  if code, err := validateRequest(r); err != nil {
    http.Error(w, err.Error(), code)
    return
  }
  // All checks passed, create a codec that reads directly from the request body
  // until EOF, writes the response to w, and orders the server to process a
  // single request.
  ctx := r.Context()
  ctx = context.WithValue(ctx, "remote", r.RemoteAddr)
  ctx = context.WithValue(ctx, "scheme", r.Proto)
  ctx = context.WithValue(ctx, "local", r.Host)
  if ua := r.Header.Get("User-Agent"); ua != "" {
    ctx = context.WithValue(ctx, "User-Agent", ua)
  }
  if origin := r.Header.Get("Origin"); origin != "" {
    ctx = context.WithValue(ctx, "Origin", origin)
  }

  w.Header().Set("content-type", contentType)
  codec := newHTTPServerConn(r, w)
  defer codec.close()
  s.serveSingleRequest(ctx, codec)
}

// validateRequest returns a non-zero response code and error message if the
// request is invalid.
func validateRequest(r *http.Request) (int, error) {
  if r.Method == http.MethodPut || r.Method == http.MethodDelete {
    return http.StatusMethodNotAllowed, errors.New("method not allowed")
  }
  if r.ContentLength > maxRequestContentLength {
    err := fmt.Errorf("content length too large (%d>%d)", r.ContentLength, maxRequestContentLength)
    return http.StatusRequestEntityTooLarge, err
  }
  // Allow OPTIONS (regardless of content-type)
  if r.Method == http.MethodOptions {
    return 0, nil
  }
  // Check content-type
  if mt, _, err := mime.ParseMediaType(r.Header.Get("content-type")); err == nil {
    for _, accepted := range acceptedContentTypes {
      if accepted == mt {
        return 0, nil
      }
    }
  }
  // Invalid content-type
  err := fmt.Errorf("invalid content type, only %s is supported", contentType)
  return http.StatusUnsupportedMediaType, err
}

// filedir:go-ethereum-1.10.2\rpc\server.go  L94
// serveSingleRequest reads and processes a single RPC request from the given codec. This
// is used to serve HTTP connections. Subscriptions and reverse calls are not allowed in
// this mode.
func (s *Server) serveSingleRequest(ctx context.Context, codec ServerCodec) {
  // Don't serve if server is stopped.
  if atomic.LoadInt32(&s.run) == 0 {
    return
  }

  h := newHandler(ctx, codec, s.idgen, &s.services)
  h.allowSubscribe = false
  defer h.close(io.EOF, nil)

  reqs, batch, err := codec.readBatch()
  if err != nil {
    if err != io.EOF {
      codec.writeJSON(ctx, errorMessage(&invalidMessageError{"parse error"}))
    }
    return
  }
  if batch {
    h.handleBatch(reqs)
  } else {
    h.handleMsg(reqs[0])
  }
}

func newHandler(connCtx context.Context, conn jsonWriter, idgen func() ID, reg *serviceRegistry) *handler {
  rootCtx, cancelRoot := context.WithCancel(connCtx)
  h := &handler{
    reg:            reg,
    idgen:          idgen,
    conn:           conn,
    respWait:       make(map[string]*requestOp),
    clientSubs:     make(map[string]*ClientSubscription),
    rootCtx:        rootCtx,
    cancelRoot:     cancelRoot,
    allowSubscribe: true,
    serverSubs:     make(map[ID]*Subscription),
    log:            log.Root(),
  }
  if conn.remoteAddr() != "" {
    h.log = h.log.New("conn", conn.remoteAddr())
  }
  h.unsubscribeCb = newCallback(reflect.Value{}, reflect.ValueOf(h.unsubscribe))
  return h
}

// filedir:go-ethereum-1.10.2\rpc\service.go
// 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 {
  fntype := fn.Type()
  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
  }
  // 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
}

// filedir:go-ethereum-1.10.2\rpc\handler.go L219
// 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)
    defer h.callWG.Done()
    defer cancel()
    fn(&callProc{ctx: ctx})
  }()
}

// filedir:go-ethereum-1.10.2\rpc\handler.go  L290
// handleCallMsg executes a call message and returns the answer.
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, "t", time.Since(start))
    return nil
  case msg.isCall():
    resp := h.handleCall(ctx, msg)
    var ctx []interface{}
    ctx = append(ctx, "reqid", idForLog{msg.ID}, "t", 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"})
  }
}

// filedir:go-ethereum-1.10.2\rpc\handler.go  L318
// handleCall processes method calls.
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 {
      failedReqeustGauge.Inc(1)
    } else {
      successfulRequestGauge.Inc(1)
    }
    rpcServingTimer.UpdateSince(start)
    newRPCServingTimer(msg.Method, answer.Error == nil).UpdateSince(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)
}

// 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 = errors.New("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
}

// CallContext performs a JSON-RPC call with the given arguments. If the context is
// canceled before the call has successfully returned, CallContext returns immediately.
//
// The result must be a pointer so that package json can unmarshal into it. You
// can also pass nil, in which case the result is ignored.
func (c *Client) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
  if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr {
    return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result)
  }
  msg, err := c.newMessage(method, args...)
  if err != nil {
    return err
  }
  op := &requestOp{ids: []json.RawMessage{msg.ID}, resp: make(chan *jsonrpcMessage, 1)}

  if c.isHTTP {
    err = c.sendHTTP(ctx, op, msg)
  } else {
    err = c.send(ctx, op, msg)
  }
  if err != nil {
    return err
  }

  // dispatch has accepted the request and will close the channel when it quits.
  switch resp, err := op.wait(ctx, c); {
  case err != nil:
    return err
  case resp.Error != nil:
    return resp.Error
  case len(resp.Result) == 0:
    return ErrNoResult
  default:
    return json.Unmarshal(resp.Result, &result)
  }
}

// filedir:go-ethereum-1.10.2\rpc\client.go   L478
// send registers op with the dispatch loop, then sends msg on the connection.
// if sending fails, op is deregistered.
func (c *Client) send(ctx context.Context, op *requestOp, msg interface{}) error {
  select {
  case c.reqInit <- op:
    err := c.write(ctx, msg, false)
    c.reqSent <- err
    return err
  case <-ctx.Done():
    // This can happen if the client is overloaded or unable to keep up with
    // subscription notifications.
    return ctx.Err()
  case <-c.closing:
    return ErrClientQuit
  }
}

func (c *Client) write(ctx context.Context, msg interface{}, retry bool) error {
  // The previous write failed. Try to establish a new connection.
  if c.writeConn == nil {
    if err := c.reconnect(ctx); err != nil {
      return err
    }
  }
  err := c.writeConn.writeJSON(ctx, msg)
  if err != nil {
    c.writeConn = nil
    if !retry {
      return c.write(ctx, msg, true)
    }
  }
  return err
}

// filedir ：go-ethereum-1.10.2\rpc\client.go
func (op *requestOp) wait(ctx context.Context, c *Client) (*jsonrpcMessage, error) {
  select {
  case <-ctx.Done():
    // Send the timeout to dispatch so it can remove the request IDs.
    if !c.isHTTP {
      select {
      case c.reqTimeout <- op:
      case <-c.closing:
      }
    }
    return nil, ctx.Err()
  case resp := <-op.resp:
    return resp, op.err
  }
}

// filedir: go-ethereum-1.10.2\rpc\client.go L538
// 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)
    }
  }
}

// filedir：go-ethereum-1.10.2\rpc\client.go  L630
// 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 {
      codec.writeJSON(context.Background(), errorMessage(&parseError{err.Error()}))
    }
    if err != nil {
      c.readErr <- err
      return
    }
    c.readOp <- readOp{msgs, batch}
  }
}