golang RPC通訊讀寫超時設置

golang RPC通訊中，有時候就怕讀寫hang住。

那是否能夠設置讀寫超時呢？

1.方案一： 設置鏈接的讀寫超時

1.1 client

RPC通訊基於底層網絡通訊，能夠經過設置connection的讀寫超時時間，達到RPC讀寫超時的目的。更多細節可參考golang網絡通訊超時設置.

下面以client端的讀超時爲例，介紹設置方法。

server端和client端代碼以下。
server
一個簡單的json RPC server。

package main

import (
    "fmt"
    "log"
    "net"
    "net/rpc"
    "net/rpc/jsonrpc"
    "time"
)

type Counter struct {
    Sum int
}

func (this *Counter) Add(i int, r *int) error {
    //sleep
    time.Sleep(3*time.Second)

    if i < 0 {
        return fmt.Errorf("data format incorrect")
    }

    this.Sum += i
    *r = this.Sum
    log.Printf("i: %v\n", i)
    return nil
}

func NewJsonRpcSocketServer() {

    rpc.Register(new(Counter))

    l, err := net.Listen("tcp", ":3333")
    if err != nil {
        log.Printf("Listener tcp err: %s", err)
        return
    }

    for {
        log.Println("waiting...")
        conn, err := l.Accept()
        if err != nil {
            log.Printf("accept connection err: %s\n", conn)
            continue
        }

        go jsonrpc.ServeConn(conn)
    }

}

func main() {
    NewJsonRpcSocketServer()
}

client

json RPC client。

鏈接創建後，會設置connection的讀超時時間。

package main

import (
    "log"
    "net"
    "net/rpc/jsonrpc"
    "time"
)


func main() {

     NewJsonRpcSocketClient()
}

func NewJsonRpcSocketClient() {
    timeout := time.Second*30
    conn, err := net.DialTimeout("tcp", "127.0.0.1:3333", timeout)
    if err != nil {
        log.Printf("create client err:%s\n", err)
        return
    }
    defer conn.Close()

    // timeout
    readAndWriteTimeout := 2*time.Second
    err = conn.SetDeadline(time.Now().Add(readAndWriteTimeout))
    if err != nil {
        log.Println("SetDeadline failed:", err)
    }

    client := jsonrpc.NewClient(conn)

    var reply int

    err = client.Call("Counter.Add", 2, &reply)
    if err != nil {
        log.Println("error:", err, "reply:", reply)
        return
    }

    log.Printf("reply: %d, err: %v\n", reply, err)

}

client輸出：

2019/05/12 22:52:57 error: read tcp 127.0.0.1:55226->127.0.0.1:3333: i/o timeout reply: 0

1.2 server

一般狀況下，RPC server端的代碼以下：

server := rpc.NewServer()
    ... ....
    for {
        conn, err := l.Accept()
        if err != nil {
            log.Println("listener accept fail:", err)
            time.Sleep(time.Duration(100) * time.Millisecond)
            continue
        }
        
        // timeout
        timeout := 10*time.Second
        conn.SetDeadline(time.Now().Add(timeout))
        
        go server.ServeCodec(jsonrpc.NewServerCodec(conn))
    }

這樣，若是設置超時，只有效的影響一次。

對於server來講，都是屢次讀寫。因此，暫時沒有方法設置。

2.方案二：定時器

經過定時器的方式，若是RPC調用在指定時間內沒有完成，觸發定時器，返回超時錯誤，關閉鏈接。

2.1 client端

func  RpcCall(method string, args interface{}, reply interface{}) error {
    ... ...

    timeout := time.Duration(10 * time.Second)
    done := make(chan error, 1)

    go func() {
        err := rpcClient.Call(method, args, reply)
        done <- err
    }()

    select {
    case <-time.After(timeout):
        log.Printf("[WARN] rpc call timeout %v => %v", rpcClient, RpcServer)
        rpcClient.Close()
        return fmt.Errorf("timeout")
    case err := <-done:
        if err != nil {
            rpcClient.Close()
            return err
        }
    }

    return nil
}

2.2 server端

不管是gobServerCodec 或者 jsonrpc ，都是實現了ServerCodec接口。

gobServerCodec文件路徑：/usr/local/go/src/net/rpc/server.go

jsonrpc文件路徑：/usr/local/go/src/net/rpc/server.go

要給server端RPC讀寫加上超時機制，須要從新定義ServerCodec接口，加上超時的控制。

// A ServerCodec implements reading of RPC requests and writing of
// RPC responses for the server side of an RPC session.
// The server calls ReadRequestHeader and ReadRequestBody in pairs
// to read requests from the connection, and it calls WriteResponse to
// write a response back. The server calls Close when finished with the
// connection. ReadRequestBody may be called with a nil
// argument to force the body of the request to be read and discarded.
type ServerCodec interface {
    ReadRequestHeader(*Request) error
    ReadRequestBody(interface{}) error
    // WriteResponse must be safe for concurrent use by multiple goroutines.
    WriteResponse(*Response, interface{}) error

    Close() error
}

目前，已經有現成的實現方式，可參考Golang標準庫RPC實踐及改進

3.參考

Does RPC have a timeout mechanism?

net/rpc: expected Timeout based alternatives to functions for rpc.Dial, rpc.DialHTTP, rpc.DialHTTPPath [proposal]. #15236