diff --git a/rpc/codec/codec.go b/rpc/codec/codec.go index 5e8f38438..3177f77e4 100644 --- a/rpc/codec/codec.go +++ b/rpc/codec/codec.go @@ -12,7 +12,7 @@ type Codec int // (de)serialization support for rpc interface type ApiCoder interface { // Parse message to request from underlying stream - ReadRequest() (*shared.Request, error) + ReadRequest() ([]*shared.Request, bool, error) // Parse response message from underlying stream ReadResponse() (interface{}, error) // Encode response to encoded form in underlying stream diff --git a/rpc/codec/json.go b/rpc/codec/json.go index 31024ee74..0b1a90562 100644 --- a/rpc/codec/json.go +++ b/rpc/codec/json.go @@ -2,71 +2,130 @@ package codec import ( "encoding/json" + "fmt" "net" + "time" "github.com/ethereum/go-ethereum/rpc/shared" ) const ( - MAX_RESPONSE_SIZE = 64 * 1024 + READ_TIMEOUT = 15 // read timeout in seconds + MAX_REQUEST_SIZE = 1024 * 1024 + MAX_RESPONSE_SIZE = 1024 * 1024 ) // Json serialization support type JsonCodec struct { c net.Conn - d *json.Decoder - e *json.Encoder } // Create new JSON coder instance func NewJsonCoder(conn net.Conn) ApiCoder { return &JsonCodec{ c: conn, - d: json.NewDecoder(conn), - e: json.NewEncoder(conn), } } // Serialize obj to JSON and write it to conn -func (self *JsonCodec) ReadRequest() (*shared.Request, error) { - req := shared.Request{} - err := self.d.Decode(&req) - if err == nil { - return &req, nil +func (self *JsonCodec) ReadRequest() (requests []*shared.Request, isBatch bool, err error) { + bytesInBuffer := 0 + buf := make([]byte, MAX_REQUEST_SIZE) + + deadline := time.Now().Add(READ_TIMEOUT * time.Second) + if err := self.c.SetDeadline(deadline); err != nil { + return nil, false, err } - return nil, err + + for { + n, err := self.c.Read(buf[bytesInBuffer:]) + if err != nil { + self.c.Close() + return nil, false, err + } + + bytesInBuffer += n + + singleRequest := shared.Request{} + err = json.Unmarshal(buf[:bytesInBuffer], &singleRequest) + if err == nil { + requests := make([]*shared.Request, 1) + requests[0] = &singleRequest + return requests, false, nil + } + + requests = make([]*shared.Request, 0) + err = json.Unmarshal(buf[:bytesInBuffer], &requests) + if err == nil { + return requests, true, nil + } + } + + self.c.Close() // timeout + return nil, false, fmt.Errorf("Unable to read response") } func (self *JsonCodec) ReadResponse() (interface{}, error) { - var err error + bytesInBuffer := 0 buf := make([]byte, MAX_RESPONSE_SIZE) - n, _ := self.c.Read(buf) - var failure shared.ErrorResponse - if err = json.Unmarshal(buf[:n], &failure); err == nil && failure.Error != nil { - return failure, nil + deadline := time.Now().Add(READ_TIMEOUT * time.Second) + if err := self.c.SetDeadline(deadline); err != nil { + return nil, err } - var success shared.SuccessResponse - if err = json.Unmarshal(buf[:n], &success); err == nil { - return success, nil + for { + n, err := self.c.Read(buf[bytesInBuffer:]) + if err != nil { + return nil, err + } + bytesInBuffer += n + + var success shared.SuccessResponse + if err = json.Unmarshal(buf[:bytesInBuffer], &success); err == nil { + return success, nil + } + + var failure shared.ErrorResponse + if err = json.Unmarshal(buf[:bytesInBuffer], &failure); err == nil && failure.Error != nil { + return failure, nil + } } - return nil, err + self.c.Close() + return nil, fmt.Errorf("Unable to read response") } -// Encode response to encoded form in underlying stream +// Decode data func (self *JsonCodec) Decode(data []byte, msg interface{}) error { return json.Unmarshal(data, msg) } +// Encode message func (self *JsonCodec) Encode(msg interface{}) ([]byte, error) { return json.Marshal(msg) } // Parse JSON data from conn to obj func (self *JsonCodec) WriteResponse(res interface{}) error { - return self.e.Encode(&res) + data, err := json.Marshal(res) + if err != nil { + self.c.Close() + return err + } + + bytesWritten := 0 + + for bytesWritten < len(data) { + n, err := self.c.Write(data[bytesWritten:]) + if err != nil { + self.c.Close() + return err + } + bytesWritten += n + } + + return nil } // Close decoder and encoder diff --git a/rpc/comms/comms.go b/rpc/comms/comms.go index bfe625758..6e980149f 100644 --- a/rpc/comms/comms.go +++ b/rpc/comms/comms.go @@ -43,29 +43,49 @@ type EthereumClient interface { SupportedModules() (map[string]string, error) } -func handle(conn net.Conn, api shared.EthereumApi, c codec.Codec) { +func handle(id int, conn net.Conn, api shared.EthereumApi, c codec.Codec) { codec := c.New(conn) for { - req, err := codec.ReadRequest() + requests, isBatch, err := codec.ReadRequest() if err == io.EOF { codec.Close() return } else if err != nil { - glog.V(logger.Error).Infof("comms recv err - %v\n", err) codec.Close() + glog.V(logger.Debug).Infof("Closed IPC Conn %06d recv err - %v\n", id, err) return } - var rpcResponse interface{} - res, err := api.Execute(req) + if isBatch { + responses := make([]*interface{}, len(requests)) + responseCount := 0 + for _, req := range requests { + res, err := api.Execute(req) + if req.Id != nil { + rpcResponse := shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err) + responses[responseCount] = rpcResponse + responseCount += 1 + } + } - rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err) - err = codec.WriteResponse(rpcResponse) - if err != nil { - glog.V(logger.Error).Infof("comms send err - %v\n", err) - codec.Close() - return + err = codec.WriteResponse(responses[:responseCount]) + if err != nil { + codec.Close() + glog.V(logger.Debug).Infof("Closed IPC Conn %06d send err - %v\n", id, err) + return + } + } else { + var rpcResponse interface{} + res, err := api.Execute(requests[0]) + + rpcResponse = shared.NewRpcResponse(requests[0].Id, requests[0].Jsonrpc, res, err) + err = codec.WriteResponse(rpcResponse) + if err != nil { + codec.Close() + glog.V(logger.Debug).Infof("Closed IPC Conn %06d send err - %v\n", id, err) + return + } } } } diff --git a/rpc/comms/ipc.go b/rpc/comms/ipc.go index 068a1288f..f3dda5581 100644 --- a/rpc/comms/ipc.go +++ b/rpc/comms/ipc.go @@ -2,6 +2,7 @@ package comms import ( "fmt" + "math/rand" "net" "encoding/json" @@ -16,6 +17,7 @@ type IpcConfig struct { type ipcClient struct { endpoint string + c net.Conn codec codec.Codec coder codec.ApiCoder } @@ -94,3 +96,7 @@ func NewIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) { func StartIpc(cfg IpcConfig, codec codec.Codec, offeredApi shared.EthereumApi) error { return startIpc(cfg, codec, offeredApi) } + +func newIpcConnId() int { + return rand.Int() % 1000000 +} diff --git a/rpc/comms/ipc_unix.go b/rpc/comms/ipc_unix.go index 295eb916b..3e71c7d32 100644 --- a/rpc/comms/ipc_unix.go +++ b/rpc/comms/ipc_unix.go @@ -18,7 +18,7 @@ func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) { return nil, err } - return &ipcClient{cfg.Endpoint, codec, codec.New(c)}, nil + return &ipcClient{cfg.Endpoint, c, codec, codec.New(c)}, nil } func (self *ipcClient) reconnect() error { @@ -48,7 +48,10 @@ func startIpc(cfg IpcConfig, codec codec.Codec, api shared.EthereumApi) error { continue } - go handle(conn, api, codec) + id := newIpcConnId() + glog.V(logger.Debug).Infof("New IPC connection with id %06d started\n", id) + + go handle(id, conn, api, codec) } os.Remove(cfg.Endpoint) diff --git a/rpc/comms/ipc_windows.go b/rpc/comms/ipc_windows.go index 44c82ef8a..203cd2d7b 100644 --- a/rpc/comms/ipc_windows.go +++ b/rpc/comms/ipc_windows.go @@ -640,7 +640,7 @@ func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) { return nil, err } - return &ipcClient{cfg.Endpoint, codec, codec.New(c)}, nil + return &ipcClient{cfg.Endpoint, c, codec, codec.New(c)}, nil } func (self *ipcClient) reconnect() error { @@ -668,33 +668,13 @@ func startIpc(cfg IpcConfig, codec codec.Codec, api shared.EthereumApi) error { continue } - go func(conn net.Conn) { - codec := codec.New(conn) + id := newIpcConnId() + glog.V(logger.Debug).Infof("New IPC connection with id %06d started\n", id) - for { - req, err := codec.ReadRequest() - if err == io.EOF { - codec.Close() - return - } else if err != nil { - glog.V(logger.Error).Infof("IPC recv err - %v\n", err) - codec.Close() - return - } - - var rpcResponse interface{} - res, err := api.Execute(req) - - rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err) - err = codec.WriteResponse(rpcResponse) - if err != nil { - glog.V(logger.Error).Infof("IPC send err - %v\n", err) - codec.Close() - return - } - } - }(conn) + go handle(id, conn, api, codec) } + + os.Remove(cfg.Endpoint) }() glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint)