aboutsummaryrefslogtreecommitdiffstats
path: root/rpc
diff options
context:
space:
mode:
Diffstat (limited to 'rpc')
-rw-r--r--rpc/api/debug.go100
-rw-r--r--rpc/api/debug_args.go24
-rw-r--r--rpc/api/debug_js.go7
-rw-r--r--rpc/api/eth.go8
-rw-r--r--rpc/api/parsing.go18
-rw-r--r--rpc/codec/codec.go2
-rw-r--r--rpc/codec/json.go103
-rw-r--r--rpc/comms/comms.go42
-rw-r--r--rpc/comms/ipc.go6
-rw-r--r--rpc/comms/ipc_unix.go7
-rw-r--r--rpc/comms/ipc_windows.go34
-rw-r--r--rpc/xeth.go52
12 files changed, 324 insertions, 79 deletions
diff --git a/rpc/api/debug.go b/rpc/api/debug.go
index b451d8662..f16f62d2e 100644
--- a/rpc/api/debug.go
+++ b/rpc/api/debug.go
@@ -2,6 +2,8 @@ package api
import (
"fmt"
+ "strings"
+ "time"
"github.com/ethereum/ethash"
"github.com/ethereum/go-ethereum/core/state"
@@ -11,6 +13,7 @@ import (
"github.com/ethereum/go-ethereum/rpc/codec"
"github.com/ethereum/go-ethereum/rpc/shared"
"github.com/ethereum/go-ethereum/xeth"
+ "github.com/rcrowley/go-metrics"
)
const (
@@ -26,6 +29,7 @@ var (
"debug_processBlock": (*debugApi).ProcessBlock,
"debug_seedHash": (*debugApi).SeedHash,
"debug_setHead": (*debugApi).SetHead,
+ "debug_metrics": (*debugApi).Metrics,
}
)
@@ -171,3 +175,99 @@ func (self *debugApi) SeedHash(req *shared.Request) (interface{}, error) {
return nil, err
}
}
+
+func (self *debugApi) Metrics(req *shared.Request) (interface{}, error) {
+ args := new(MetricsArgs)
+ if err := self.codec.Decode(req.Params, &args); err != nil {
+ return nil, shared.NewDecodeParamError(err.Error())
+ }
+ // Create a rate formatter
+ units := []string{"", "K", "M", "G", "T", "E", "P"}
+ round := func(value float64, prec int) string {
+ unit := 0
+ for value >= 1000 {
+ unit, value, prec = unit+1, value/1000, 2
+ }
+ return fmt.Sprintf(fmt.Sprintf("%%.%df%s", prec, units[unit]), value)
+ }
+ format := func(total float64, rate float64) string {
+ return fmt.Sprintf("%s (%s/s)", round(total, 0), round(rate, 2))
+ }
+ // Iterate over all the metrics, and just dump for now
+ counters := make(map[string]interface{})
+ metrics.DefaultRegistry.Each(func(name string, metric interface{}) {
+ // Create or retrieve the counter hierarchy for this metric
+ root, parts := counters, strings.Split(name, "/")
+ for _, part := range parts[:len(parts)-1] {
+ if _, ok := root[part]; !ok {
+ root[part] = make(map[string]interface{})
+ }
+ root = root[part].(map[string]interface{})
+ }
+ name = parts[len(parts)-1]
+
+ // Fill the counter with the metric details, formatting if requested
+ if args.Raw {
+ switch metric := metric.(type) {
+ case metrics.Meter:
+ root[name] = map[string]interface{}{
+ "AvgRate01Min": metric.Rate1(),
+ "AvgRate05Min": metric.Rate5(),
+ "AvgRate15Min": metric.Rate15(),
+ "MeanRate": metric.RateMean(),
+ "Overall": float64(metric.Count()),
+ }
+
+ case metrics.Timer:
+ root[name] = map[string]interface{}{
+ "AvgRate01Min": metric.Rate1(),
+ "AvgRate05Min": metric.Rate5(),
+ "AvgRate15Min": metric.Rate15(),
+ "MeanRate": metric.RateMean(),
+ "Overall": float64(metric.Count()),
+ "Percentiles": map[string]interface{}{
+ "5": metric.Percentile(0.05),
+ "20": metric.Percentile(0.2),
+ "50": metric.Percentile(0.5),
+ "80": metric.Percentile(0.8),
+ "95": metric.Percentile(0.95),
+ },
+ }
+
+ default:
+ root[name] = "Unknown metric type"
+ }
+ } else {
+ switch metric := metric.(type) {
+ case metrics.Meter:
+ root[name] = map[string]interface{}{
+ "Avg01Min": format(metric.Rate1()*60, metric.Rate1()),
+ "Avg05Min": format(metric.Rate5()*300, metric.Rate5()),
+ "Avg15Min": format(metric.Rate15()*900, metric.Rate15()),
+ "Overall": format(float64(metric.Count()), metric.RateMean()),
+ }
+
+ case metrics.Timer:
+ root[name] = map[string]interface{}{
+ "Avg01Min": format(metric.Rate1()*60, metric.Rate1()),
+ "Avg05Min": format(metric.Rate5()*300, metric.Rate5()),
+ "Avg15Min": format(metric.Rate15()*900, metric.Rate15()),
+ "Overall": format(float64(metric.Count()), metric.RateMean()),
+ "Maximum": time.Duration(metric.Max()).String(),
+ "Minimum": time.Duration(metric.Min()).String(),
+ "Percentiles": map[string]interface{}{
+ "5": time.Duration(metric.Percentile(0.05)).String(),
+ "20": time.Duration(metric.Percentile(0.2)).String(),
+ "50": time.Duration(metric.Percentile(0.5)).String(),
+ "80": time.Duration(metric.Percentile(0.8)).String(),
+ "95": time.Duration(metric.Percentile(0.95)).String(),
+ },
+ }
+
+ default:
+ root[name] = "Unknown metric type"
+ }
+ }
+ })
+ return counters, nil
+}
diff --git a/rpc/api/debug_args.go b/rpc/api/debug_args.go
index b9b5aa27e..b72fb03ae 100644
--- a/rpc/api/debug_args.go
+++ b/rpc/api/debug_args.go
@@ -4,6 +4,7 @@ import (
"encoding/json"
"fmt"
"math/big"
+ "reflect"
"github.com/ethereum/go-ethereum/rpc/shared"
)
@@ -45,3 +46,26 @@ func (args *WaitForBlockArgs) UnmarshalJSON(b []byte) (err error) {
return nil
}
+
+type MetricsArgs struct {
+ Raw bool
+}
+
+func (args *MetricsArgs) UnmarshalJSON(b []byte) (err error) {
+ var obj []interface{}
+ if err := json.Unmarshal(b, &obj); err != nil {
+ return shared.NewDecodeParamError(err.Error())
+ }
+ if len(obj) > 1 {
+ return fmt.Errorf("metricsArgs needs 0, 1 arguments")
+ }
+ // default values when not provided
+ if len(obj) >= 1 && obj[0] != nil {
+ if value, ok := obj[0].(bool); !ok {
+ return fmt.Errorf("invalid argument %v", reflect.TypeOf(obj[0]))
+ } else {
+ args.Raw = value
+ }
+ }
+ return nil
+}
diff --git a/rpc/api/debug_js.go b/rpc/api/debug_js.go
index 35fecb75f..bd3a6dfb2 100644
--- a/rpc/api/debug_js.go
+++ b/rpc/api/debug_js.go
@@ -46,6 +46,13 @@ web3._extend({
params: 1,
inputFormatter: [web3._extend.formatters.formatInputInt],
outputFormatter: function(obj) { return obj; }
+ }),
+ new web3._extend.Method({
+ name: 'metrics',
+ call: 'debug_metrics',
+ params: 1,
+ inputFormatter: [web3._extend.formatters.formatInputBool],
+ outputFormatter: function(obj) { return obj; }
})
],
properties:
diff --git a/rpc/api/eth.go b/rpc/api/eth.go
index 0dff138c6..962c8d0f9 100644
--- a/rpc/api/eth.go
+++ b/rpc/api/eth.go
@@ -348,14 +348,6 @@ func (self *ethApi) GetBlockByNumber(req *shared.Request) (interface{}, error) {
block := self.xeth.EthBlockByNumber(args.BlockNumber)
br := NewBlockRes(block, args.IncludeTxs)
- // If request was for "pending", nil nonsensical fields
- if args.BlockNumber == -2 {
- br.BlockHash = nil
- br.BlockNumber = nil
- br.Miner = nil
- br.Nonce = nil
- br.LogsBloom = nil
- }
return br, nil
}
diff --git a/rpc/api/parsing.go b/rpc/api/parsing.go
index 85a9165e5..632462c31 100644
--- a/rpc/api/parsing.go
+++ b/rpc/api/parsing.go
@@ -270,29 +270,31 @@ func NewBlockRes(block *types.Block, fullTx bool) *BlockRes {
res.BlockHash = newHexData(block.Hash())
res.ParentHash = newHexData(block.ParentHash())
res.Nonce = newHexData(block.Nonce())
- res.Sha3Uncles = newHexData(block.Header().UncleHash)
+ res.Sha3Uncles = newHexData(block.UncleHash())
res.LogsBloom = newHexData(block.Bloom())
- res.TransactionRoot = newHexData(block.Header().TxHash)
+ res.TransactionRoot = newHexData(block.TxHash())
res.StateRoot = newHexData(block.Root())
- res.Miner = newHexData(block.Header().Coinbase)
+ res.Miner = newHexData(block.Coinbase())
res.Difficulty = newHexNum(block.Difficulty())
res.TotalDifficulty = newHexNum(block.Td)
res.Size = newHexNum(block.Size().Int64())
- res.ExtraData = newHexData(block.Header().Extra)
+ res.ExtraData = newHexData(block.Extra())
res.GasLimit = newHexNum(block.GasLimit())
res.GasUsed = newHexNum(block.GasUsed())
res.UnixTimestamp = newHexNum(block.Time())
- res.Transactions = make([]*TransactionRes, len(block.Transactions()))
- for i, tx := range block.Transactions() {
+ txs := block.Transactions()
+ res.Transactions = make([]*TransactionRes, len(txs))
+ for i, tx := range txs {
res.Transactions[i] = NewTransactionRes(tx)
res.Transactions[i].BlockHash = res.BlockHash
res.Transactions[i].BlockNumber = res.BlockNumber
res.Transactions[i].TxIndex = newHexNum(i)
}
- res.Uncles = make([]*UncleRes, len(block.Uncles()))
- for i, uncle := range block.Uncles() {
+ uncles := block.Uncles()
+ res.Uncles = make([]*UncleRes, len(uncles))
+ for i, uncle := range uncles {
res.Uncles[i] = NewUncleRes(uncle)
}
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
+ }
+
+ 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
+ }
}
- return nil, err
+
+ 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)
-
- 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)
+ 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)
}()
glog.V(logger.Info).Infof("IPC service started (%s)\n", cfg.Endpoint)
diff --git a/rpc/xeth.go b/rpc/xeth.go
new file mode 100644
index 000000000..b3e844380
--- /dev/null
+++ b/rpc/xeth.go
@@ -0,0 +1,52 @@
+package rpc
+
+import (
+ "encoding/json"
+ "fmt"
+ "reflect"
+ "sync/atomic"
+
+ "github.com/ethereum/go-ethereum/rpc/comms"
+ "github.com/ethereum/go-ethereum/rpc/shared"
+)
+
+// Xeth is a native API interface to a remote node.
+type Xeth struct {
+ client comms.EthereumClient
+ reqId uint32
+}
+
+// NewXeth constructs a new native API interface to a remote node.
+func NewXeth(client comms.EthereumClient) *Xeth {
+ return &Xeth{
+ client: client,
+ }
+}
+
+// Call invokes a method with the given parameters are the remote node.
+func (self *Xeth) Call(method string, params []interface{}) (map[string]interface{}, error) {
+ // Assemble the json RPC request
+ data, err := json.Marshal(params)
+ if err != nil {
+ return nil, err
+ }
+ req := &shared.Request{
+ Id: atomic.AddUint32(&self.reqId, 1),
+ Jsonrpc: "2.0",
+ Method: method,
+ Params: data,
+ }
+ // Send the request over and process the response
+ if err := self.client.Send(req); err != nil {
+ return nil, err
+ }
+ res, err := self.client.Recv()
+ if err != nil {
+ return nil, err
+ }
+ value, ok := res.(map[string]interface{})
+ if !ok {
+ return nil, fmt.Errorf("Invalid response type: have %v, want %v", reflect.TypeOf(res), reflect.TypeOf(make(map[string]interface{})))
+ }
+ return value, nil
+}
generated by cgit v1.2.3 (git 2.25.1) at 2025-01-15 18:03:37 +0800