以太坊的RPC機制

1 go語言的RPC機制

    RPC(Remote Procedure Call,遠端過程呼叫)是一種通過網路從遠端計算機程式上請求服
務,而不需要了解底層網路細節的應用程式通訊協議。RPC協議構建於TCP或UDP,或者是 HTTP
之上,允許開發者直接呼叫另一臺計算機上的程式,而開發者無需額外地為這個呼叫過程編寫網

絡通訊相關程式碼,使得開發包括網路分散式程式在內的應用程式更加容易。

    go語言有net/rpc包,net/rpc包允許 RPC 客戶端程式通過網路或是其他 I/O 連線呼叫一個遠端物件的公開方法
(必須是大寫字母開頭、可外部呼叫的)。在 RPC 服務端,可將一個物件註冊為可訪問的服務,
之後該物件的公開方法就能夠以遠端的方式提供訪問。一個 RPC 服務端可以註冊多個不同型別
的物件,但不允許註冊同一型別的多個物件。

    一個物件中只有滿足如下這些條件的方法,才能被 RPC 服務端設定為可供遠端訪問:

  • 必須是在物件外部可公開呼叫的方法(首字母大寫);
  • 必須有兩個引數,且引數的型別都必須是包外部可以訪問的型別或者是Go內建支援的型別;
  • 第二個引數必須是一個指標;
  • 方法必須返回一個error型別的值。

以上4個條件,可以簡單地用如下一行程式碼表示:

func (t *T) MethodName(argType T1, replyType *T2) error

接下來,我們來看一組 RPC 服務端和客戶端互動的示例程式。

服務端:

package main;
import (
"net/rpc"
"net/http"
"log"
)
//go對RPC的支援,支援三個級別:TCP、HTTP、JSONRPC
//go的RPC只支援GO開發的伺服器與客戶端之間的互動,因為採用了gob編碼
//注意欄位必須是匯出
type Params struct {
Width, Height int;
}
type Rect struct{}
//函式必須是匯出的
//必須有兩個匯出型別引數
//第一個引數是接收引數
//第二個引數是返回給客戶端引數,必須是指標型別
//函式還要有一個返回值error
func (r *Rect) Area(p Params, ret *int) error {
*ret = p.Width * p.Height;
return nil;
}
func (r *Rect) Perimeter(p Params, ret *int) error {
*ret = (p.Width   p.Height) * 2;
return nil;
}
func main() {
rect := new(Rect);
//註冊一個rect服務
rpc.Register(rect);
//把服務處理繫結到http協議上
rpc.HandleHTTP();
err := http.ListenAndServe(":8080", nil);
if err != nil {
log.Fatal(err);
}
}

客戶端:

package main;
import (
"net/rpc"
"log"
"fmt"
)
type Params struct {
Width, Height int;
}
func main() {
//連線遠端rpc服務
rpc, err := rpc.DialHTTP("tcp", "127.0.0.1:8080");
if err != nil {
log.Fatal(err);
}
ret := 0;
//呼叫遠端方法
//注意第三個引數是指標型別
err2 := rpc.Call("Rect.Area", Params{50, 100}, &ret);
if err2 != nil {
log.Fatal(err2);
}
fmt.Println(ret);
err3 := rpc.Call("Rect.Perimeter", Params{50, 100}, &ret);
if err3 != nil {
log.Fatal(err3);
}
fmt.Println(ret);
}

2 以太坊RPC機制

以太坊啟動RPC服務

以太坊客戶端可以用下面方式來啟動RPC監聽:

geth --rpc --rpcaddr 0.0.0.0 --rpcapi db,eth,net,web3,personal --rpcport 8550

這句明命令啟動了Http-RPC服務,rpc監聽地址是任意ip地址,rcp使用的api介面包括db,eth,net,web,personal等,rpc埠是8550。

以太坊原始碼中RPC服務啟動流程

在以太坊geth的main函式裡,有函式

func geth(ctx *cli.Context) error {
node := makeFullNode(ctx)
startNode(ctx, node)
node.Wait()
return nil
}

這是geth的主執行函式,通過startNode()啟動geth節點,startNode繼續呼叫node/node.go中的Start()函式中,Start()函式中呼叫了startRPC()函式:

// startRPC is a helper method to start all the various RPC endpoint 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(services map[reflect.Type]Service) error {
// Gather all the possible APIs to surface
apis := n.apis()
for _, service := range services {
apis = append(apis, service.APIs()...)
}
// Start the various API endpoints, terminating all in case of errors
if err := n.startInProc(apis); err != nil {
return err
}
if err := n.startIPC(apis); err != nil {
n.stopInProc()
return err
}
if err := n.startHTTP(n.httpEndpoint, apis, n.config.HTTPModules, n.config.HTTPCors, n.config.HTTPVirtualHosts); err != nil {
n.stopIPC()
n.stopInProc()
return err
}
if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins, n.config.WSExposeAll); err != nil {
n.stopHTTP()
n.stopIPC()
n.stopInProc()
return err
}
// All API endpoints started successfully
n.rpcAPIs = apis
return nil
}

startRPC()收集了node中和services中所有的rpc.API型別的RPC介面,並啟動了各種RPC服務形式,包括IPC、HTTP、WS、PROC等各種形式。下面分析啟動Http方式的RPC函式startHTTP():

func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors []string, vhosts []string) error {
// Short circuit if the HTTP endpoint isn't being exposed
if endpoint == "" {
return nil
}
// Generate the whitelist based on the allowed modules
whitelist := make(map[string]bool)
for _, module := range modules {
whitelist[module] = true
}
// Register all the APIs exposed by the services
handler := rpc.NewServer()
for _, api := range apis {
if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
return err
}
n.log.Debug("HTTP registered", "service", api.Service, "namespace", api.Namespace)
}
}
// All APIs registered, start the HTTP listener
var (
listener net.Listener
err      error
)
if listener, err = net.Listen("tcp", endpoint); err != nil {
return err
}
go rpc.NewHTTPServer(cors, vhosts, handler).Serve(listener)
n.log.Info("HTTP endpoint opened", "url", fmt.Sprintf("http://%s", endpoint), "cors", strings.Join(cors, ","), "vhosts", strings.Join(vhosts, ","))
// All listeners booted successfully
n.httpEndpoint = endpoint
n.httpListener = listener
n.httpHandler = handler
return nil
}

可以看到以太坊中通過Http方式啟動RPC服務的流程跟go中的rpc包啟動方式基本一致。先是通過rpc.newServer()建立了Server,然後再通過registerName()註冊API服務,然後啟動Http監聽。不過以太坊中的RPC介面API並不是按照標準RPC介面寫的,它的基本形式是:

func (s *CalcService) Add(a, b int) (int, error)

符合以下標準的方法可用於遠端訪問:

  • 物件必須匯出
  • 方法必須匯出
  • 方法返回0,1(響應或錯誤)或2(響應和錯誤)值
  • 方法引數必須匯出或是內建型別
  • 方法返回值必須匯出或是內建型別

客戶端呼叫RPC服務

rpc/client.go中撥號函式:

/ The client reconnects automatically if the connection is lost.
func Dial(rawurl string) (*Client, error) {
return DialContext(context.Background(), rawurl)
}
// DialContext creates a new RPC client, just like Dial.
//
// The context is used to cancel or time out the initial connection establishment. It does
// not affect subsequent interactions with the client.
func DialContext(ctx context.Context, rawurl string) (*Client, error) {
u, err := url.Parse(rawurl)
if err != nil {
return nil, err
}
switch u.Scheme {
case "http", "https":
return DialHTTP(rawurl)
case "ws", "wss":
return DialWebsocket(ctx, rawurl, "")
case "":
return DialIPC(ctx, rawurl)
default:
return nil, fmt.Errorf("no known transport for URL scheme %q", u.Scheme)
}
}

呼叫RPC服務的函式:

// Call performs a JSON-RPC call with the given arguments and unmarshals into
// result if no error occurred.
//
// 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) Call(result interface{}, method string, args ...interface{}) error {
ctx := context.Background()
return c.CallContext(ctx, result, method, args...)
}
// 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 {
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 it when it quits.
switch resp, err := op.wait(ctx); {
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)
}
}

3 web3.js與控制檯呼叫RPC介面

internal/jsre/deps下有web3.js檔案,以及internal/web3ext下的web3ext.go檔案,封裝了可以在console控制檯下訪問RPC介面的方法和介面。console下面有admin.importChain方法,搜尋importChain,可以看到搜尋結果,

importChain對應的一個出現在web3ext.go中,

new web3._extend.Method({
    name: 'importChain',
    call: 'admin_importChain',
    params: 1
}),

函式定義在eth/api.go中:

// ImportChain imports a blockchain from a local file.
func (api *PrivateAdminAPI) ImportChain(file string) (bool, error) {
// Make sure the can access the file to import
in, err := os.Open(file)
if err != nil {
return false, err
}
defer in.Close()
......
}

4 自定義RPC介面

依照ImportChain介面的方法,在eth/api.go中定義函式:

func (api *PrivateAdminAPI) TestMul(a,b *int) (int, error) {
return (*a)*(*b),nil;
}

然後在web3ext.go中加入宣告:

new web3._extend.Method({
name: 'startRPC',
call: 'admin_startRPC',
params: 4,
inputFormatter: [null, null, null, null]
}),
new web3._extend.Method({
name: 'stopRPC',
call: 'admin_stopRPC'
}),
new web3._extend.Method({
name: 'startWS',
call: 'admin_startWS',
params: 4,
inputFormatter: [null, null, null, null]
}),
new web3._extend.Method({
name: 'stopWS',
call: 'admin_stopWS'
}),
new web3._extend.Method({
name: 'testMul',
call: 'admin_testMul',
params: 2
}),
],

重新編譯geth,執行,在控制檯輸入admin:

可以看到出現了testMul介面,呼叫testMul介面試一下: