Skip to content

Commit e64625a

Browse files
obscurenobscuren
committed
Merge pull request #1332 from bas-vk/ipcbatch
IPC interface improvements
2 parents 6b2a03f + 6622850 commit e64625a

File tree

6 files changed

+131
-63
lines changed

6 files changed

+131
-63
lines changed

‎rpc/codec/codec.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ type Codec int
1212
// (de)serialization support for rpc interface
1313
type ApiCoder interface {
1414
// Parse message to request from underlying stream
15-
ReadRequest() (*shared.Request, error)
15+
ReadRequest() ([]*shared.Request, bool, error)
1616
// Parse response message from underlying stream
1717
ReadResponse() (interface{}, error)
1818
// Encode response to encoded form in underlying stream

‎rpc/codec/json.go‎

Lines changed: 81 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -2,71 +2,130 @@ package codec
22

33
import (
44
"encoding/json"
5+
"fmt"
56
"net"
7+
"time"
68

79
"github.com/ethereum/go-ethereum/rpc/shared"
810
)
911

1012
const (
11-
MAX_RESPONSE_SIZE = 64 * 1024
13+
READ_TIMEOUT = 15 // read timeout in seconds
14+
MAX_REQUEST_SIZE = 1024 * 1024
15+
MAX_RESPONSE_SIZE = 1024 * 1024
1216
)
1317

1418
// Json serialization support
1519
type JsonCodec struct {
1620
c net.Conn
17-
d *json.Decoder
18-
e *json.Encoder
1921
}
2022

2123
// Create new JSON coder instance
2224
func NewJsonCoder(conn net.Conn) ApiCoder {
2325
return &JsonCodec{
2426
c: conn,
25-
d: json.NewDecoder(conn),
26-
e: json.NewEncoder(conn),
2727
}
2828
}
2929

3030
// Serialize obj to JSON and write it to conn
31-
func (self *JsonCodec) ReadRequest() (*shared.Request, error) {
32-
req := shared.Request{}
33-
err := self.d.Decode(&req)
34-
if err == nil {
35-
return &req, nil
31+
func (self *JsonCodec) ReadRequest() (requests []*shared.Request, isBatch bool, err error) {
32+
bytesInBuffer := 0
33+
buf := make([]byte, MAX_REQUEST_SIZE)
34+
35+
deadline := time.Now().Add(READ_TIMEOUT * time.Second)
36+
if err := self.c.SetDeadline(deadline); err != nil {
37+
return nil, false, err
38+
}
39+
40+
for {
41+
n, err := self.c.Read(buf[bytesInBuffer:])
42+
if err != nil {
43+
self.c.Close()
44+
return nil, false, err
45+
}
46+
47+
bytesInBuffer += n
48+
49+
singleRequest := shared.Request{}
50+
err = json.Unmarshal(buf[:bytesInBuffer], &singleRequest)
51+
if err == nil {
52+
requests := make([]*shared.Request, 1)
53+
requests[0] = &singleRequest
54+
return requests, false, nil
55+
}
56+
57+
requests = make([]*shared.Request, 0)
58+
err = json.Unmarshal(buf[:bytesInBuffer], &requests)
59+
if err == nil {
60+
return requests, true, nil
61+
}
3662
}
37-
return nil, err
63+
64+
self.c.Close() // timeout
65+
return nil, false, fmt.Errorf("Unable to read response")
3866
}
3967

4068
func (self *JsonCodec) ReadResponse() (interface{}, error) {
41-
var err error
69+
bytesInBuffer := 0
4270
buf := make([]byte, MAX_RESPONSE_SIZE)
43-
n, _ := self.c.Read(buf)
4471

45-
var failure shared.ErrorResponse
46-
if err = json.Unmarshal(buf[:n], &failure); err == nil && failure.Error != nil {
47-
return failure, nil
72+
deadline := time.Now().Add(READ_TIMEOUT * time.Second)
73+
if err := self.c.SetDeadline(deadline); err != nil {
74+
return nil, err
4875
}
4976

50-
var success shared.SuccessResponse
51-
if err = json.Unmarshal(buf[:n], &success); err == nil {
52-
return success, nil
77+
for {
78+
n, err := self.c.Read(buf[bytesInBuffer:])
79+
if err != nil {
80+
return nil, err
81+
}
82+
bytesInBuffer += n
83+
84+
var success shared.SuccessResponse
85+
if err = json.Unmarshal(buf[:bytesInBuffer], &success); err == nil {
86+
return success, nil
87+
}
88+
89+
var failure shared.ErrorResponse
90+
if err = json.Unmarshal(buf[:bytesInBuffer], &failure); err == nil && failure.Error != nil {
91+
return failure, nil
92+
}
5393
}
5494

55-
return nil, err
95+
self.c.Close()
96+
return nil, fmt.Errorf("Unable to read response")
5697
}
5798

58-
// Encode response to encoded form in underlying stream
99+
// Decode data
59100
func (self *JsonCodec) Decode(data []byte, msg interface{}) error {
60101
return json.Unmarshal(data, msg)
61102
}
62103

104+
// Encode message
63105
func (self *JsonCodec) Encode(msg interface{}) ([]byte, error) {
64106
return json.Marshal(msg)
65107
}
66108

67109
// Parse JSON data from conn to obj
68110
func (self *JsonCodec) WriteResponse(res interface{}) error {
69-
return self.e.Encode(&res)
111+
data, err := json.Marshal(res)
112+
if err != nil {
113+
self.c.Close()
114+
return err
115+
}
116+
117+
bytesWritten := 0
118+
119+
for bytesWritten < len(data) {
120+
n, err := self.c.Write(data[bytesWritten:])
121+
if err != nil {
122+
self.c.Close()
123+
return err
124+
}
125+
bytesWritten += n
126+
}
127+
128+
return nil
70129
}
71130

72131
// Close decoder and encoder

‎rpc/comms/comms.go‎

Lines changed: 31 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -43,29 +43,49 @@ type EthereumClient interface {
4343
SupportedModules() (map[string]string, error)
4444
}
4545

46-
func handle(conn net.Conn, api shared.EthereumApi, c codec.Codec) {
46+
func handle(id int, conn net.Conn, api shared.EthereumApi, c codec.Codec) {
4747
codec := c.New(conn)
4848

4949
for {
50-
req, err := codec.ReadRequest()
50+
requests, isBatch, err := codec.ReadRequest()
5151
if err == io.EOF {
5252
codec.Close()
5353
return
5454
} else if err != nil {
55-
glog.V(logger.Error).Infof("comms recv err - %v\n", err)
5655
codec.Close()
56+
glog.V(logger.Debug).Infof("Closed IPC Conn %06d recv err - %v\n", id, err)
5757
return
5858
}
5959

60-
var rpcResponse interface{}
61-
res, err := api.Execute(req)
60+
if isBatch {
61+
responses := make([]*interface{}, len(requests))
62+
responseCount := 0
63+
for _, req := range requests {
64+
res, err := api.Execute(req)
65+
if req.Id != nil {
66+
rpcResponse := shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)
67+
responses[responseCount] = rpcResponse
68+
responseCount += 1
69+
}
70+
}
6271

63-
rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)
64-
err = codec.WriteResponse(rpcResponse)
65-
if err != nil {
66-
glog.V(logger.Error).Infof("comms send err - %v\n", err)
67-
codec.Close()
68-
return
72+
err = codec.WriteResponse(responses[:responseCount])
73+
if err != nil {
74+
codec.Close()
75+
glog.V(logger.Debug).Infof("Closed IPC Conn %06d send err - %v\n", id, err)
76+
return
77+
}
78+
} else {
79+
var rpcResponse interface{}
80+
res, err := api.Execute(requests[0])
81+
82+
rpcResponse = shared.NewRpcResponse(requests[0].Id, requests[0].Jsonrpc, res, err)
83+
err = codec.WriteResponse(rpcResponse)
84+
if err != nil {
85+
codec.Close()
86+
glog.V(logger.Debug).Infof("Closed IPC Conn %06d send err - %v\n", id, err)
87+
return
88+
}
6989
}
7090
}
7191
}

‎rpc/comms/ipc.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package comms
22

33
import (
44
"fmt"
5+
"math/rand"
56
"net"
67

78
"encoding/json"
@@ -16,6 +17,7 @@ type IpcConfig struct {
1617

1718
type ipcClient struct {
1819
endpoint string
20+
c net.Conn
1921
codec codec.Codec
2022
coder codec.ApiCoder
2123
}
@@ -94,3 +96,7 @@ func NewIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) {
9496
func StartIpc(cfg IpcConfig, codec codec.Codec, offeredApi shared.EthereumApi) error {
9597
return startIpc(cfg, codec, offeredApi)
9698
}
99+
100+
func newIpcConnId() int {
101+
return rand.Int() % 1000000
102+
}

‎rpc/comms/ipc_unix.go‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) {
1818
return nil, err
1919
}
2020

21-
return &ipcClient{cfg.Endpoint, codec, codec.New(c)}, nil
21+
return &ipcClient{cfg.Endpoint, c, codec, codec.New(c)}, nil
2222
}
2323

2424
func (self *ipcClient) reconnect() error {
@@ -48,7 +48,10 @@ func startIpc(cfg IpcConfig, codec codec.Codec, api shared.EthereumApi) error {
4848
continue
4949
}
5050

51-
go handle(conn, api, codec)
51+
id := newIpcConnId()
52+
glog.V(logger.Debug).Infof("New IPC connection with id %06d started\n", id)
53+
54+
go handle(id, conn, api, codec)
5255
}
5356

5457
os.Remove(cfg.Endpoint)

‎rpc/comms/ipc_windows.go‎

Lines changed: 7 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -640,7 +640,7 @@ func newIpcClient(cfg IpcConfig, codec codec.Codec) (*ipcClient, error) {
640640
return nil, err
641641
}
642642

643-
return &ipcClient{cfg.Endpoint, codec, codec.New(c)}, nil
643+
return &ipcClient{cfg.Endpoint, c, codec, codec.New(c)}, nil
644644
}
645645

646646
func (self *ipcClient) reconnect() error {
@@ -668,33 +668,13 @@ func startIpc(cfg IpcConfig, codec codec.Codec, api shared.EthereumApi) error {
668668
continue
669669
}
670670

671-
go func(conn net.Conn) {
672-
codec := codec.New(conn)
673-
674-
for {
675-
req, err := codec.ReadRequest()
676-
if err == io.EOF {
677-
codec.Close()
678-
return
679-
} else if err != nil {
680-
glog.V(logger.Error).Infof("IPC recv err - %v\n", err)
681-
codec.Close()
682-
return
683-
}
684-
685-
var rpcResponse interface{}
686-
res, err := api.Execute(req)
687-
688-
rpcResponse = shared.NewRpcResponse(req.Id, req.Jsonrpc, res, err)
689-
err = codec.WriteResponse(rpcResponse)
690-
if err != nil {
691-
glog.V(logger.Error).Infof("IPC send err - %v\n", err)
692-
codec.Close()
693-
return
694-
}
695-
}
696-
}(conn)
671+
id := newIpcConnId()
672+
glog.V(logger.Debug).Infof("New IPC connection with id %06d started\n", id)
673+
674+
go handle(id, conn, api, codec)
697675
}
676+
677+
os.Remove(cfg.Endpoint)
698678
}()
699679

700680
glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint)

0 commit comments

Comments
 (0)