From a7d0998bc0e87a2197159c2c48140a8f73121c2b Mon Sep 17 00:00:00 2001 From: Ethan Buchman Date: Sat, 5 Dec 2015 17:48:25 +0000 Subject: [PATCH 01/14] didSend -> willSend --- proxy/log.go | 7 +++++++ proxy/remote_app_context.go | 12 ++++++++---- 2 files changed, 15 insertions(+), 4 deletions(-) create mode 100644 proxy/log.go diff --git a/proxy/log.go b/proxy/log.go new file mode 100644 index 00000000..45d31b87 --- /dev/null +++ b/proxy/log.go @@ -0,0 +1,7 @@ +package proxy + +import ( + "github.com/tendermint/go-logger" +) + +var log = logger.New("module", "proxy") diff --git a/proxy/remote_app_context.go b/proxy/remote_app_context.go index 09a7db5a..dd96abe7 100644 --- a/proxy/remote_app_context.go +++ b/proxy/remote_app_context.go @@ -21,7 +21,7 @@ const maxResponseSize = 1048576 // 1MB // with concurrent callers. type remoteAppContext struct { QuitService - sync.Mutex + sync.Mutex // [EB]: is this even used? reqQueue chan *reqRes @@ -65,7 +65,7 @@ func (app *remoteAppContext) SetResponseCallback(resCb Callback) { func (app *remoteAppContext) StopForError(err error) { app.mtx.Lock() - fmt.Println("Stopping remoteAppContext for error:", err) + log.Error("Stopping remoteAppContext for error.", "error", err) if app.err == nil { app.err = err } @@ -89,11 +89,15 @@ func (app *remoteAppContext) sendRequestsRoutine() { case <-app.QuitService.Quit: return case reqres := <-app.reqQueue: + + app.willSendReq(reqres) + wire.WriteBinary(reqres.Request, app.bufWriter, &n, &err) if err != nil { app.StopForError(err) return } + log.Debug("Sent request", "requestType", reflect.TypeOf(reqres.Request), "request", reqres.Request) if _, ok := reqres.Request.(tmsp.RequestFlush); ok { err = app.bufWriter.Flush() if err != nil { @@ -101,7 +105,6 @@ func (app *remoteAppContext) sendRequestsRoutine() { return } } - app.didSendReq(reqres) } } } @@ -121,6 +124,7 @@ func (app *remoteAppContext) recvResponseRoutine() { case tmsp.ResponseException: app.StopForError(errors.New(res.Error)) default: + log.Debug("Received response", "responseType", reflect.TypeOf(res), "response", res) err := app.didRecvResponse(res) if err != nil { app.StopForError(err) @@ -129,7 +133,7 @@ func (app *remoteAppContext) recvResponseRoutine() { } } -func (app *remoteAppContext) didSendReq(reqres *reqRes) { +func (app *remoteAppContext) willSendReq(reqres *reqRes) { app.mtx.Lock() defer app.mtx.Unlock() app.reqSent.PushBack(reqres) From 67e74756f4457e522d95b6291fd7ffbe8eb1e2ab Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Sat, 5 Dec 2015 18:10:26 -0800 Subject: [PATCH 02/14] Remove PEX/AddrBook --- node/node.go | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/node/node.go b/node/node.go index b6311635..85de92dc 100644 --- a/node/node.go +++ b/node/node.go @@ -31,9 +31,7 @@ import _ "net/http/pprof" type Node struct { sw *p2p.Switch evsw *events.EventSwitch - book *p2p.AddrBook blockStore *bc.BlockStore - pexReactor *p2p.PEXReactor bcReactor *bc.BlockchainReactor mempoolReactor *mempl.MempoolReactor consensusState *consensus.ConsensusState @@ -74,10 +72,6 @@ func NewNode() *Node { Exit(Fmt("Failed to start switch: %v", err)) } - // Make PEXReactor - book := p2p.NewAddrBook(config.GetString("addrbook_file")) - pexReactor := p2p.NewPEXReactor(book) - // Make BlockchainReactor bcReactor := bc.NewBlockchainReactor(state.Copy(), proxyAppCtxConsensus, blockStore, config.GetBool("fast_sync")) @@ -94,7 +88,6 @@ func NewNode() *Node { // Make p2p network switch sw := p2p.NewSwitch() - sw.AddReactor("PEX", pexReactor) sw.AddReactor("MEMPOOL", mempoolReactor) sw.AddReactor("BLOCKCHAIN", bcReactor) sw.AddReactor("CONSENSUS", consensusReactor) @@ -114,9 +107,7 @@ func NewNode() *Node { return &Node{ sw: sw, evsw: eventSwitch, - book: book, blockStore: blockStore, - pexReactor: pexReactor, bcReactor: bcReactor, mempoolReactor: mempoolReactor, consensusState: consensusState, @@ -129,7 +120,6 @@ func NewNode() *Node { // Call Start() after adding the listeners. func (n *Node) Start() error { - n.book.Start() n.sw.SetNodeInfo(makeNodeInfo(n.sw, n.privKey)) n.sw.SetNodePrivKey(n.privKey) _, err := n.sw.Start() @@ -140,7 +130,6 @@ func (n *Node) Stop() { log.Notice("Stopping Node") // TODO: gracefully disconnect from peers. n.sw.Stop() - n.book.Stop() } // Add the event switch to reactors, mempool, etc. @@ -156,7 +145,6 @@ func SetFireable(evsw *events.EventSwitch, eventables ...events.Eventable) { func (n *Node) AddListener(l p2p.Listener) { log.Notice(Fmt("Added %v", l)) n.sw.AddListener(l) - n.book.AddOurAddress(l.ExternalAddress()) } // Dial a list of seeds in random order @@ -179,11 +167,9 @@ func (n *Node) dialSeed(addr *p2p.NetAddress) { peer, err := n.sw.DialPeerWithAddress(addr) if err != nil { log.Error("Error dialing seed", "error", err) - //n.book.MarkAttempt(addr) return } else { log.Notice("Connected to seed", "peer", peer) - n.book.AddAddress(addr, addr) } } From 3da76496b0e9d46e24dfe933e211e893085b02a0 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Mon, 7 Dec 2015 16:57:33 -0800 Subject: [PATCH 03/14] Fix BlockchainReactor bug w/ mismatched state.LastBlockHeight vs store.Height This is due to a non-atomic saves of state.State vs blockchain.Store. This is a simple hack. --- blockchain/reactor.go | 38 ++++++++++++++++++++------------------ 1 file changed, 20 insertions(+), 18 deletions(-) diff --git a/blockchain/reactor.go b/blockchain/reactor.go index 0a977830..473525a3 100644 --- a/blockchain/reactor.go +++ b/blockchain/reactor.go @@ -42,22 +42,24 @@ type consensusReactor interface { type BlockchainReactor struct { p2p.BaseReactor - sw *p2p.Switch - state *sm.State - proxyAppCtx proxy.AppContext // same as consensus.proxyAppCtx - store *BlockStore - pool *BlockPool - sync bool - requestsCh chan BlockRequest - timeoutsCh chan string - lastBlock *types.Block + sw *p2p.Switch + state *sm.State + proxyAppCtx proxy.AppContext // same as consensus.proxyAppCtx + store *BlockStore + pool *BlockPool + sync bool + requestsCh chan BlockRequest + timeoutsCh chan string + lastBlock *types.Block evsw events.Fireable } func NewBlockchainReactor(state *sm.State, proxyAppCtx proxy.AppContext, store *BlockStore, sync bool) *BlockchainReactor { - if state.LastBlockHeight != store.Height() && - state.LastBlockHeight != store.Height()-1 { // XXX double check this logic. + if state.LastBlockHeight == store.Height()-1 { + store.height -= 1 // XXX HACK, make this better + } + if state.LastBlockHeight != store.Height() { PanicSanity(Fmt("state (%v) and store (%v) height mismatch", state.LastBlockHeight, store.Height())) } requestsCh := make(chan BlockRequest, defaultChannelCapacity) @@ -68,13 +70,13 @@ func NewBlockchainReactor(state *sm.State, proxyAppCtx proxy.AppContext, store * timeoutsCh, ) bcR := &BlockchainReactor{ - state: state, - proxyAppCtx: proxyAppCtx, - store: store, - pool: pool, - sync: sync, - requestsCh: requestsCh, - timeoutsCh: timeoutsCh, + state: state, + proxyAppCtx: proxyAppCtx, + store: store, + pool: pool, + sync: sync, + requestsCh: requestsCh, + timeoutsCh: timeoutsCh, } bcR.BaseReactor = *p2p.NewBaseReactor(log, "BlockchainReactor", bcR) return bcR From 53f74d052fbaacac009de4d4684424180ab1db73 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Mon, 7 Dec 2015 17:54:19 -0800 Subject: [PATCH 04/14] Fix deadlock bug in websocket client impl --- benchmarks/simu/counter.go | 44 ++++++++ rpc/client/{client.go => http_client.go} | 8 +- rpc/client/log.go | 1 - rpc/client/ws_client.go | 126 +++++++++++++++++++++++ rpc/server/handlers.go | 14 +-- rpc/types/types.go | 9 ++ 6 files changed, 191 insertions(+), 11 deletions(-) create mode 100644 benchmarks/simu/counter.go rename rpc/client/{client.go => http_client.go} (82%) create mode 100644 rpc/client/ws_client.go diff --git a/benchmarks/simu/counter.go b/benchmarks/simu/counter.go new file mode 100644 index 00000000..8b1b3e5e --- /dev/null +++ b/benchmarks/simu/counter.go @@ -0,0 +1,44 @@ +package main + +import ( + "fmt" + + "github.com/gorilla/websocket" + . "github.com/tendermint/go-common" + "github.com/tendermint/go-wire" + "github.com/tendermint/tendermint/rpc/client" + // ctypes "github.com/tendermint/tendermint/rpc/core/types" + "github.com/tendermint/tendermint/rpc/types" +) + +func main() { + ws := rpcclient.NewWSClient("ws://127.0.0.1:46657/websocket") + // ws := rpcclient.NewWSClient("ws://104.236.69.128:46657/websocket") + _, err := ws.Start() + if err != nil { + Exit(err.Error()) + } + + // Read a bunch of responses + go func() { + for { + res, ok := <-ws.ResultsCh + if !ok { + break + } + fmt.Println("Received response", res) + } + }() + + // Make a bunch of requests + request := rpctypes.NewRPCRequest("fakeid", "status", nil) + for i := 0; ; i++ { + reqBytes := wire.JSONBytes(request) + err := ws.WriteMessage(websocket.TextMessage, reqBytes) + if err != nil { + Exit(err.Error()) + } + } + + ws.Stop() +} diff --git a/rpc/client/client.go b/rpc/client/http_client.go similarity index 82% rename from rpc/client/client.go rename to rpc/client/http_client.go index 80e15744..6cc275d0 100644 --- a/rpc/client/client.go +++ b/rpc/client/http_client.go @@ -9,12 +9,12 @@ import ( . "github.com/tendermint/go-common" "github.com/tendermint/go-wire" - . "github.com/tendermint/tendermint/rpc/types" + "github.com/tendermint/tendermint/rpc/types" ) -func Call(remote string, method string, params []interface{}, dest interface{}) (interface{}, error) { +func CallHTTP(remote string, method string, params []interface{}, dest interface{}) (interface{}, error) { // Make request and get responseBytes - request := RPCRequest{ + request := rpctypes.RPCRequest{ JSONRPC: "2.0", Method: method, Params: params, @@ -35,7 +35,7 @@ func Call(remote string, method string, params []interface{}, dest interface{}) log.Info(Fmt("RPC response: %v", string(responseBytes))) // Parse response into JSONResponse - response := RPCResponse{} + response := rpctypes.RPCResponse{} err = json.Unmarshal(responseBytes, &response) if err != nil { return dest, err diff --git a/rpc/client/log.go b/rpc/client/log.go index 465a5c44..8b33e2f1 100644 --- a/rpc/client/log.go +++ b/rpc/client/log.go @@ -1,4 +1,3 @@ - package rpcclient import ( diff --git a/rpc/client/ws_client.go b/rpc/client/ws_client.go new file mode 100644 index 00000000..2ca2f44f --- /dev/null +++ b/rpc/client/ws_client.go @@ -0,0 +1,126 @@ +package rpcclient + +import ( + "net/http" + "strings" + "time" + + "github.com/gorilla/websocket" + . "github.com/tendermint/go-common" + "github.com/tendermint/go-wire" + ctypes "github.com/tendermint/tendermint/rpc/core/types" + "github.com/tendermint/tendermint/rpc/types" +) + +const ( + wsEventsChannelCapacity = 10 + wsResultsChannelCapacity = 10 + wsWriteTimeoutSeconds = 10 +) + +type WSClient struct { + QuitService + Address string + *websocket.Conn + EventsCh chan ctypes.ResultEvent // closes upon WSClient.Stop() + ResultsCh chan ctypes.Result // closes upon WSClient.Stop() +} + +// create a new connection +func NewWSClient(addr string) *WSClient { + wsClient := &WSClient{ + Address: addr, + Conn: nil, + EventsCh: make(chan ctypes.ResultEvent, wsEventsChannelCapacity), + ResultsCh: make(chan ctypes.Result, wsResultsChannelCapacity), + } + wsClient.QuitService = *NewQuitService(log, "WSClient", wsClient) + return wsClient +} + +func (wsc *WSClient) OnStart() error { + wsc.QuitService.OnStart() + err := wsc.dial() + if err != nil { + return err + } + go wsc.receiveEventsRoutine() + return nil +} + +func (wsc *WSClient) dial() error { + // Dial + dialer := websocket.DefaultDialer + rHeader := http.Header{} + con, _, err := dialer.Dial(wsc.Address, rHeader) + if err != nil { + return err + } + // Set the ping/pong handlers + con.SetPingHandler(func(m string) error { + // NOTE: https://github.com/gorilla/websocket/issues/97 + go con.WriteControl(websocket.PongMessage, []byte(m), time.Now().Add(time.Second*wsWriteTimeoutSeconds)) + return nil + }) + con.SetPongHandler(func(m string) error { + // NOTE: https://github.com/gorilla/websocket/issues/97 + return nil + }) + wsc.Conn = con + return nil +} + +func (wsc *WSClient) OnStop() { + wsc.QuitService.OnStop() + // EventsCh and ResultsCh are closed in receiveEventsRoutine. +} + +func (wsc *WSClient) receiveEventsRoutine() { + for { + _, data, err := wsc.ReadMessage() + if err != nil { + log.Info("WSClient failed to read message", "error", err, "data", string(data)) + wsc.Stop() + break + } else { + var response ctypes.Response + wire.ReadJSON(&response, data, &err) + if err != nil { + log.Info("WSClient failed to parse message", "error", err) + wsc.Stop() + break + } + if strings.HasSuffix(response.ID, "#event") { + wsc.EventsCh <- *response.Result.(*ctypes.ResultEvent) + } else { + wsc.ResultsCh <- response.Result + } + } + } + + // Cleanup + close(wsc.EventsCh) + close(wsc.ResultsCh) +} + +// subscribe to an event +func (wsc *WSClient) Subscribe(eventid string) error { + err := wsc.WriteJSON(rpctypes.RPCRequest{ + JSONRPC: "2.0", + ID: "", + Method: "subscribe", + Params: []interface{}{eventid}, + }) + return err +} + +// unsubscribe from an event +func (wsc *WSClient) Unsubscribe(eventid string) error { + err := wsc.WriteJSON(rpctypes.RPCRequest{ + JSONRPC: "2.0", + ID: "", + Method: "unsubscribe", + Params: []interface{}{eventid}, + }) + return err +} diff --git a/rpc/server/handlers.go b/rpc/server/handlers.go index bdb942ff..c26b60fe 100644 --- a/rpc/server/handlers.go +++ b/rpc/server/handlers.go @@ -252,11 +252,13 @@ func (wsc *WSConnection) OnStart() error { wsc.readTimeout = time.NewTimer(time.Second * wsReadTimeoutSeconds) wsc.pingTicker = time.NewTicker(time.Second * wsPingTickerSeconds) wsc.baseConn.SetPingHandler(func(m string) error { - wsc.baseConn.WriteControl(websocket.PongMessage, []byte(m), time.Now().Add(time.Second*wsWriteTimeoutSeconds)) + // NOTE: https://github.com/gorilla/websocket/issues/97 + go wsc.baseConn.WriteControl(websocket.PongMessage, []byte(m), time.Now().Add(time.Second*wsWriteTimeoutSeconds)) wsc.readTimeout.Reset(time.Second * wsReadTimeoutSeconds) return nil }) wsc.baseConn.SetPongHandler(func(m string) error { + // NOTE: https://github.com/gorilla/websocket/issues/97 wsc.readTimeout.Reset(time.Second * wsReadTimeoutSeconds) return nil }) @@ -287,13 +289,12 @@ func (wsc *WSConnection) readTimeoutRoutine() { } } -// Attempt to write response to writeChan and record failures +// Block trying to write to writeChan until service stops. func (wsc *WSConnection) writeRPCResponse(resp RPCResponse) { select { + case <-wsc.Quit: + return case wsc.writeChan <- resp: - default: - log.Notice("Stopping connection due to writeChan overflow", "id", wsc.id) - wsc.Stop() // writeChan capacity exceeded, error. } } @@ -412,7 +413,8 @@ func (wsc *WSConnection) writeRoutine() { log.Error("Failed to marshal RPCResponse to JSON", "error", err) } else { wsc.baseConn.SetWriteDeadline(time.Now().Add(time.Second * wsWriteTimeoutSeconds)) - if err = wsc.baseConn.WriteMessage(websocket.TextMessage, buf.Bytes()); err != nil { + bufBytes := buf.Bytes() + if err = wsc.baseConn.WriteMessage(websocket.TextMessage, bufBytes); err != nil { log.Warn("Failed to write response on websocket", "error", err) wsc.Stop() return diff --git a/rpc/types/types.go b/rpc/types/types.go index c93e6735..ba6ff0d6 100644 --- a/rpc/types/types.go +++ b/rpc/types/types.go @@ -7,6 +7,15 @@ type RPCRequest struct { Params []interface{} `json:"params"` } +func NewRPCRequest(id string, method string, params []interface{}) RPCRequest { + return RPCRequest{ + JSONRPC: "2.0", + ID: id, + Method: method, + Params: params, + } +} + type RPCResponse struct { JSONRPC string `json:"jsonrpc"` ID string `json:"id"` From a62c7bfef1ec6c05a14599be940145d68a66d9fa Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Tue, 8 Dec 2015 08:11:29 -0800 Subject: [PATCH 05/14] Add benchmarking for RPC & wire --- benchmarks/simu/counter.go | 9 ++++-- benchmarks/wire_test.go | 60 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 3 deletions(-) create mode 100644 benchmarks/wire_test.go diff --git a/benchmarks/simu/counter.go b/benchmarks/simu/counter.go index 8b1b3e5e..8191e8de 100644 --- a/benchmarks/simu/counter.go +++ b/benchmarks/simu/counter.go @@ -22,22 +22,25 @@ func main() { // Read a bunch of responses go func() { for { - res, ok := <-ws.ResultsCh + _, ok := <-ws.ResultsCh if !ok { break } - fmt.Println("Received response", res) + //fmt.Println("Received response", string(wire.JSONBytes(res))) } }() // Make a bunch of requests - request := rpctypes.NewRPCRequest("fakeid", "status", nil) + request := rpctypes.NewRPCRequest("fakeid", "net_info", nil) for i := 0; ; i++ { reqBytes := wire.JSONBytes(request) err := ws.WriteMessage(websocket.TextMessage, reqBytes) if err != nil { Exit(err.Error()) } + if i%1000 == 0 { + fmt.Println(i) + } } ws.Stop() diff --git a/benchmarks/wire_test.go b/benchmarks/wire_test.go new file mode 100644 index 00000000..7da8a7f7 --- /dev/null +++ b/benchmarks/wire_test.go @@ -0,0 +1,60 @@ +package benchmarks + +import ( + "testing" + + "github.com/tendermint/go-crypto" + "github.com/tendermint/go-p2p" + "github.com/tendermint/go-wire" + ctypes "github.com/tendermint/tendermint/rpc/core/types" +) + +func BenchmarkEncodeStatus(b *testing.B) { + b.StopTimer() + pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) + status := &ctypes.ResultStatus{ + NodeInfo: &p2p.NodeInfo{ + PubKey: pubKey, + Moniker: "SOMENAME", + Network: "SOMENAME", + RemoteAddr: "SOMEADDR", + ListenAddr: "SOMEADDR", + Version: "SOMEVER", + Other: []string{"SOMESTRING", "OTHERSTRING"}, + }, + PubKey: pubKey, + LatestBlockHash: []byte("SOMEBYTES"), + LatestBlockHeight: 123, + LatestBlockTime: 1234, + } + b.StartTimer() + + counter := 0 + for i := 0; i < b.N; i++ { + jsonBytes := wire.JSONBytes(status) + counter += len(jsonBytes) + } + +} + +func BenchmarkEncodeNodeInfo(b *testing.B) { + b.StopTimer() + pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) + nodeInfo := &p2p.NodeInfo{ + PubKey: pubKey, + Moniker: "SOMENAME", + Network: "SOMENAME", + RemoteAddr: "SOMEADDR", + ListenAddr: "SOMEADDR", + Version: "SOMEVER", + Other: []string{"SOMESTRING", "OTHERSTRING"}, + } + b.StartTimer() + + counter := 0 + for i := 0; i < b.N; i++ { + jsonBytes := wire.JSONBytes(nodeInfo) + counter += len(jsonBytes) + } + +} From ac4c43c3e67ac728d4ca40cbc62bf8e1c4e4abe5 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Tue, 8 Dec 2015 15:33:26 -0800 Subject: [PATCH 06/14] Adding wire.JSON vs wire.Binary vs protobuf benchmarks --- benchmarks/codec_test.go | 110 +++ benchmarks/proto/README | 2 + benchmarks/proto/test.pb.go | 1503 +++++++++++++++++++++++++++++++++++ benchmarks/proto/test.proto | 25 + benchmarks/wire_test.go | 60 -- 5 files changed, 1640 insertions(+), 60 deletions(-) create mode 100644 benchmarks/codec_test.go create mode 100644 benchmarks/proto/README create mode 100644 benchmarks/proto/test.pb.go create mode 100644 benchmarks/proto/test.proto delete mode 100644 benchmarks/wire_test.go diff --git a/benchmarks/codec_test.go b/benchmarks/codec_test.go new file mode 100644 index 00000000..35dc591e --- /dev/null +++ b/benchmarks/codec_test.go @@ -0,0 +1,110 @@ +package benchmarks + +import ( + "testing" + + "github.com/tendermint/go-crypto" + "github.com/tendermint/go-p2p" + "github.com/tendermint/go-wire" + proto "github.com/tendermint/tendermint/benchmarks/proto" + ctypes "github.com/tendermint/tendermint/rpc/core/types" +) + +func BenchmarkEncodeStatusWire(b *testing.B) { + b.StopTimer() + pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) + status := &ctypes.ResultStatus{ + NodeInfo: &p2p.NodeInfo{ + PubKey: pubKey, + Moniker: "SOMENAME", + Network: "SOMENAME", + RemoteAddr: "SOMEADDR", + ListenAddr: "SOMEADDR", + Version: "SOMEVER", + Other: []string{"SOMESTRING", "OTHERSTRING"}, + }, + PubKey: pubKey, + LatestBlockHash: []byte("SOMEBYTES"), + LatestBlockHeight: 123, + LatestBlockTime: 1234, + } + b.StartTimer() + + counter := 0 + for i := 0; i < b.N; i++ { + jsonBytes := wire.JSONBytes(status) + counter += len(jsonBytes) + } + +} + +func BenchmarkEncodeNodeInfoWire(b *testing.B) { + b.StopTimer() + pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) + nodeInfo := &p2p.NodeInfo{ + PubKey: pubKey, + Moniker: "SOMENAME", + Network: "SOMENAME", + RemoteAddr: "SOMEADDR", + ListenAddr: "SOMEADDR", + Version: "SOMEVER", + Other: []string{"SOMESTRING", "OTHERSTRING"}, + } + b.StartTimer() + + counter := 0 + for i := 0; i < b.N; i++ { + jsonBytes := wire.JSONBytes(nodeInfo) + counter += len(jsonBytes) + } +} + +func BenchmarkEncodeNodeInfoBinary(b *testing.B) { + b.StopTimer() + pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) + nodeInfo := &p2p.NodeInfo{ + PubKey: pubKey, + Moniker: "SOMENAME", + Network: "SOMENAME", + RemoteAddr: "SOMEADDR", + ListenAddr: "SOMEADDR", + Version: "SOMEVER", + Other: []string{"SOMESTRING", "OTHERSTRING"}, + } + b.StartTimer() + + counter := 0 + for i := 0; i < b.N; i++ { + jsonBytes := wire.BinaryBytes(nodeInfo) + counter += len(jsonBytes) + } + +} + +func BenchmarkEncodeNodeInfoProto(b *testing.B) { + b.StopTimer() + pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) + pubKey2 := &proto.PubKey{Ed25519: &proto.PubKeyEd25519{Bytes: pubKey[:]}} + nodeInfo := &proto.NodeInfo{ + PubKey: pubKey2, + Moniker: "SOMENAME", + Network: "SOMENAME", + RemoteAddr: "SOMEADDR", + ListenAddr: "SOMEADDR", + Version: "SOMEVER", + Other: []string{"SOMESTRING", "OTHERSTRING"}, + } + b.StartTimer() + + counter := 0 + for i := 0; i < b.N; i++ { + bytes, err := nodeInfo.Marshal() + if err != nil { + b.Fatal(err) + return + } + //jsonBytes := wire.JSONBytes(nodeInfo) + counter += len(bytes) + } + +} diff --git a/benchmarks/proto/README b/benchmarks/proto/README new file mode 100644 index 00000000..87ece257 --- /dev/null +++ b/benchmarks/proto/README @@ -0,0 +1,2 @@ +Doing some protobuf tests here. +Using gogoprotobuf. diff --git a/benchmarks/proto/test.pb.go b/benchmarks/proto/test.pb.go new file mode 100644 index 00000000..6539cae3 --- /dev/null +++ b/benchmarks/proto/test.pb.go @@ -0,0 +1,1503 @@ +// Code generated by protoc-gen-gogo. +// source: test.proto +// DO NOT EDIT! + +/* + Package test is a generated protocol buffer package. + + It is generated from these files: + test.proto + + It has these top-level messages: + ResultStatus + NodeInfo + PubKey + PubKeyEd25519 +*/ +package test + +import proto "github.com/gogo/protobuf/proto" +import fmt "fmt" +import math "math" + +import bytes "bytes" + +import strings "strings" +import github_com_gogo_protobuf_proto "github.com/gogo/protobuf/proto" +import sort "sort" +import strconv "strconv" +import reflect "reflect" + +import io "io" + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +type ResultStatus struct { + NodeInfo *NodeInfo `protobuf:"bytes,1,opt,name=nodeInfo" json:"nodeInfo,omitempty"` + PubKey *PubKey `protobuf:"bytes,2,req,name=pubKey" json:"pubKey,omitempty"` + LatestBlockHash []byte `protobuf:"bytes,3,req,name=latestBlockHash" json:"latestBlockHash"` + LatestBlockHeight int64 `protobuf:"varint,4,req,name=latestBlockHeight" json:"latestBlockHeight"` + LatestBlocktime int64 `protobuf:"varint,5,req,name=latestBlocktime" json:"latestBlocktime"` +} + +func (m *ResultStatus) Reset() { *m = ResultStatus{} } +func (*ResultStatus) ProtoMessage() {} + +func (m *ResultStatus) GetNodeInfo() *NodeInfo { + if m != nil { + return m.NodeInfo + } + return nil +} + +func (m *ResultStatus) GetPubKey() *PubKey { + if m != nil { + return m.PubKey + } + return nil +} + +func (m *ResultStatus) GetLatestBlockHash() []byte { + if m != nil { + return m.LatestBlockHash + } + return nil +} + +func (m *ResultStatus) GetLatestBlockHeight() int64 { + if m != nil { + return m.LatestBlockHeight + } + return 0 +} + +func (m *ResultStatus) GetLatestBlocktime() int64 { + if m != nil { + return m.LatestBlocktime + } + return 0 +} + +type NodeInfo struct { + PubKey *PubKey `protobuf:"bytes,1,req,name=pubKey" json:"pubKey,omitempty"` + Moniker string `protobuf:"bytes,2,req,name=moniker" json:"moniker"` + Network string `protobuf:"bytes,3,req,name=network" json:"network"` + RemoteAddr string `protobuf:"bytes,4,req,name=remoteAddr" json:"remoteAddr"` + ListenAddr string `protobuf:"bytes,5,req,name=listenAddr" json:"listenAddr"` + Version string `protobuf:"bytes,6,req,name=version" json:"version"` + Other []string `protobuf:"bytes,7,rep,name=other" json:"other,omitempty"` +} + +func (m *NodeInfo) Reset() { *m = NodeInfo{} } +func (*NodeInfo) ProtoMessage() {} + +func (m *NodeInfo) GetPubKey() *PubKey { + if m != nil { + return m.PubKey + } + return nil +} + +func (m *NodeInfo) GetMoniker() string { + if m != nil { + return m.Moniker + } + return "" +} + +func (m *NodeInfo) GetNetwork() string { + if m != nil { + return m.Network + } + return "" +} + +func (m *NodeInfo) GetRemoteAddr() string { + if m != nil { + return m.RemoteAddr + } + return "" +} + +func (m *NodeInfo) GetListenAddr() string { + if m != nil { + return m.ListenAddr + } + return "" +} + +func (m *NodeInfo) GetVersion() string { + if m != nil { + return m.Version + } + return "" +} + +func (m *NodeInfo) GetOther() []string { + if m != nil { + return m.Other + } + return nil +} + +type PubKey struct { + Ed25519 *PubKeyEd25519 `protobuf:"bytes,1,opt,name=ed25519" json:"ed25519,omitempty"` +} + +func (m *PubKey) Reset() { *m = PubKey{} } +func (*PubKey) ProtoMessage() {} + +func (m *PubKey) GetEd25519() *PubKeyEd25519 { + if m != nil { + return m.Ed25519 + } + return nil +} + +type PubKeyEd25519 struct { + Bytes []byte `protobuf:"bytes,1,req,name=bytes" json:"bytes"` +} + +func (m *PubKeyEd25519) Reset() { *m = PubKeyEd25519{} } +func (*PubKeyEd25519) ProtoMessage() {} + +func (m *PubKeyEd25519) GetBytes() []byte { + if m != nil { + return m.Bytes + } + return nil +} + +func init() { + proto.RegisterType((*ResultStatus)(nil), "ResultStatus") + proto.RegisterType((*NodeInfo)(nil), "NodeInfo") + proto.RegisterType((*PubKey)(nil), "PubKey") + proto.RegisterType((*PubKeyEd25519)(nil), "PubKeyEd25519") +} +func (this *ResultStatus) Equal(that interface{}) bool { + if that == nil { + if this == nil { + return true + } + return false + } + + that1, ok := that.(*ResultStatus) + if !ok { + that2, ok := that.(ResultStatus) + if ok { + that1 = &that2 + } else { + return false + } + } + if that1 == nil { + if this == nil { + return true + } + return false + } else if this == nil { + return false + } + if !this.NodeInfo.Equal(that1.NodeInfo) { + return false + } + if !this.PubKey.Equal(that1.PubKey) { + return false + } + if !bytes.Equal(this.LatestBlockHash, that1.LatestBlockHash) { + return false + } + if this.LatestBlockHeight != that1.LatestBlockHeight { + return false + } + if this.LatestBlocktime != that1.LatestBlocktime { + return false + } + return true +} +func (this *NodeInfo) Equal(that interface{}) bool { + if that == nil { + if this == nil { + return true + } + return false + } + + that1, ok := that.(*NodeInfo) + if !ok { + that2, ok := that.(NodeInfo) + if ok { + that1 = &that2 + } else { + return false + } + } + if that1 == nil { + if this == nil { + return true + } + return false + } else if this == nil { + return false + } + if !this.PubKey.Equal(that1.PubKey) { + return false + } + if this.Moniker != that1.Moniker { + return false + } + if this.Network != that1.Network { + return false + } + if this.RemoteAddr != that1.RemoteAddr { + return false + } + if this.ListenAddr != that1.ListenAddr { + return false + } + if this.Version != that1.Version { + return false + } + if len(this.Other) != len(that1.Other) { + return false + } + for i := range this.Other { + if this.Other[i] != that1.Other[i] { + return false + } + } + return true +} +func (this *PubKey) Equal(that interface{}) bool { + if that == nil { + if this == nil { + return true + } + return false + } + + that1, ok := that.(*PubKey) + if !ok { + that2, ok := that.(PubKey) + if ok { + that1 = &that2 + } else { + return false + } + } + if that1 == nil { + if this == nil { + return true + } + return false + } else if this == nil { + return false + } + if !this.Ed25519.Equal(that1.Ed25519) { + return false + } + return true +} +func (this *PubKeyEd25519) Equal(that interface{}) bool { + if that == nil { + if this == nil { + return true + } + return false + } + + that1, ok := that.(*PubKeyEd25519) + if !ok { + that2, ok := that.(PubKeyEd25519) + if ok { + that1 = &that2 + } else { + return false + } + } + if that1 == nil { + if this == nil { + return true + } + return false + } else if this == nil { + return false + } + if !bytes.Equal(this.Bytes, that1.Bytes) { + return false + } + return true +} +func (this *ResultStatus) GoString() string { + if this == nil { + return "nil" + } + s := make([]string, 0, 9) + s = append(s, "&test.ResultStatus{") + if this.NodeInfo != nil { + s = append(s, "NodeInfo: "+fmt.Sprintf("%#v", this.NodeInfo)+",\n") + } + if this.PubKey != nil { + s = append(s, "PubKey: "+fmt.Sprintf("%#v", this.PubKey)+",\n") + } + s = append(s, "LatestBlockHash: "+fmt.Sprintf("%#v", this.LatestBlockHash)+",\n") + s = append(s, "LatestBlockHeight: "+fmt.Sprintf("%#v", this.LatestBlockHeight)+",\n") + s = append(s, "LatestBlocktime: "+fmt.Sprintf("%#v", this.LatestBlocktime)+",\n") + s = append(s, "}") + return strings.Join(s, "") +} +func (this *NodeInfo) GoString() string { + if this == nil { + return "nil" + } + s := make([]string, 0, 11) + s = append(s, "&test.NodeInfo{") + if this.PubKey != nil { + s = append(s, "PubKey: "+fmt.Sprintf("%#v", this.PubKey)+",\n") + } + s = append(s, "Moniker: "+fmt.Sprintf("%#v", this.Moniker)+",\n") + s = append(s, "Network: "+fmt.Sprintf("%#v", this.Network)+",\n") + s = append(s, "RemoteAddr: "+fmt.Sprintf("%#v", this.RemoteAddr)+",\n") + s = append(s, "ListenAddr: "+fmt.Sprintf("%#v", this.ListenAddr)+",\n") + s = append(s, "Version: "+fmt.Sprintf("%#v", this.Version)+",\n") + if this.Other != nil { + s = append(s, "Other: "+fmt.Sprintf("%#v", this.Other)+",\n") + } + s = append(s, "}") + return strings.Join(s, "") +} +func (this *PubKey) GoString() string { + if this == nil { + return "nil" + } + s := make([]string, 0, 5) + s = append(s, "&test.PubKey{") + if this.Ed25519 != nil { + s = append(s, "Ed25519: "+fmt.Sprintf("%#v", this.Ed25519)+",\n") + } + s = append(s, "}") + return strings.Join(s, "") +} +func (this *PubKeyEd25519) GoString() string { + if this == nil { + return "nil" + } + s := make([]string, 0, 5) + s = append(s, "&test.PubKeyEd25519{") + s = append(s, "Bytes: "+fmt.Sprintf("%#v", this.Bytes)+",\n") + s = append(s, "}") + return strings.Join(s, "") +} +func valueToGoStringTest(v interface{}, typ string) string { + rv := reflect.ValueOf(v) + if rv.IsNil() { + return "nil" + } + pv := reflect.Indirect(rv).Interface() + return fmt.Sprintf("func(v %v) *%v { return &v } ( %#v )", typ, typ, pv) +} +func extensionToGoStringTest(e map[int32]github_com_gogo_protobuf_proto.Extension) string { + if e == nil { + return "nil" + } + s := "map[int32]proto.Extension{" + keys := make([]int, 0, len(e)) + for k := range e { + keys = append(keys, int(k)) + } + sort.Ints(keys) + ss := []string{} + for _, k := range keys { + ss = append(ss, strconv.Itoa(k)+": "+e[int32(k)].GoString()) + } + s += strings.Join(ss, ",") + "}" + return s +} +func (m *ResultStatus) Marshal() (data []byte, err error) { + size := m.Size() + data = make([]byte, size) + n, err := m.MarshalTo(data) + if err != nil { + return nil, err + } + return data[:n], nil +} + +func (m *ResultStatus) MarshalTo(data []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if m.NodeInfo != nil { + data[i] = 0xa + i++ + i = encodeVarintTest(data, i, uint64(m.NodeInfo.Size())) + n1, err := m.NodeInfo.MarshalTo(data[i:]) + if err != nil { + return 0, err + } + i += n1 + } + if m.PubKey == nil { + return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("pubKey") + } else { + data[i] = 0x12 + i++ + i = encodeVarintTest(data, i, uint64(m.PubKey.Size())) + n2, err := m.PubKey.MarshalTo(data[i:]) + if err != nil { + return 0, err + } + i += n2 + } + if m.LatestBlockHash != nil { + data[i] = 0x1a + i++ + i = encodeVarintTest(data, i, uint64(len(m.LatestBlockHash))) + i += copy(data[i:], m.LatestBlockHash) + } + data[i] = 0x20 + i++ + i = encodeVarintTest(data, i, uint64(m.LatestBlockHeight)) + data[i] = 0x28 + i++ + i = encodeVarintTest(data, i, uint64(m.LatestBlocktime)) + return i, nil +} + +func (m *NodeInfo) Marshal() (data []byte, err error) { + size := m.Size() + data = make([]byte, size) + n, err := m.MarshalTo(data) + if err != nil { + return nil, err + } + return data[:n], nil +} + +func (m *NodeInfo) MarshalTo(data []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if m.PubKey == nil { + return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("pubKey") + } else { + data[i] = 0xa + i++ + i = encodeVarintTest(data, i, uint64(m.PubKey.Size())) + n3, err := m.PubKey.MarshalTo(data[i:]) + if err != nil { + return 0, err + } + i += n3 + } + data[i] = 0x12 + i++ + i = encodeVarintTest(data, i, uint64(len(m.Moniker))) + i += copy(data[i:], m.Moniker) + data[i] = 0x1a + i++ + i = encodeVarintTest(data, i, uint64(len(m.Network))) + i += copy(data[i:], m.Network) + data[i] = 0x22 + i++ + i = encodeVarintTest(data, i, uint64(len(m.RemoteAddr))) + i += copy(data[i:], m.RemoteAddr) + data[i] = 0x2a + i++ + i = encodeVarintTest(data, i, uint64(len(m.ListenAddr))) + i += copy(data[i:], m.ListenAddr) + data[i] = 0x32 + i++ + i = encodeVarintTest(data, i, uint64(len(m.Version))) + i += copy(data[i:], m.Version) + if len(m.Other) > 0 { + for _, s := range m.Other { + data[i] = 0x3a + i++ + l = len(s) + for l >= 1<<7 { + data[i] = uint8(uint64(l)&0x7f | 0x80) + l >>= 7 + i++ + } + data[i] = uint8(l) + i++ + i += copy(data[i:], s) + } + } + return i, nil +} + +func (m *PubKey) Marshal() (data []byte, err error) { + size := m.Size() + data = make([]byte, size) + n, err := m.MarshalTo(data) + if err != nil { + return nil, err + } + return data[:n], nil +} + +func (m *PubKey) MarshalTo(data []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if m.Ed25519 != nil { + data[i] = 0xa + i++ + i = encodeVarintTest(data, i, uint64(m.Ed25519.Size())) + n4, err := m.Ed25519.MarshalTo(data[i:]) + if err != nil { + return 0, err + } + i += n4 + } + return i, nil +} + +func (m *PubKeyEd25519) Marshal() (data []byte, err error) { + size := m.Size() + data = make([]byte, size) + n, err := m.MarshalTo(data) + if err != nil { + return nil, err + } + return data[:n], nil +} + +func (m *PubKeyEd25519) MarshalTo(data []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if m.Bytes != nil { + data[i] = 0xa + i++ + i = encodeVarintTest(data, i, uint64(len(m.Bytes))) + i += copy(data[i:], m.Bytes) + } + return i, nil +} + +func encodeFixed64Test(data []byte, offset int, v uint64) int { + data[offset] = uint8(v) + data[offset+1] = uint8(v >> 8) + data[offset+2] = uint8(v >> 16) + data[offset+3] = uint8(v >> 24) + data[offset+4] = uint8(v >> 32) + data[offset+5] = uint8(v >> 40) + data[offset+6] = uint8(v >> 48) + data[offset+7] = uint8(v >> 56) + return offset + 8 +} +func encodeFixed32Test(data []byte, offset int, v uint32) int { + data[offset] = uint8(v) + data[offset+1] = uint8(v >> 8) + data[offset+2] = uint8(v >> 16) + data[offset+3] = uint8(v >> 24) + return offset + 4 +} +func encodeVarintTest(data []byte, offset int, v uint64) int { + for v >= 1<<7 { + data[offset] = uint8(v&0x7f | 0x80) + v >>= 7 + offset++ + } + data[offset] = uint8(v) + return offset + 1 +} +func (m *ResultStatus) Size() (n int) { + var l int + _ = l + if m.NodeInfo != nil { + l = m.NodeInfo.Size() + n += 1 + l + sovTest(uint64(l)) + } + if m.PubKey != nil { + l = m.PubKey.Size() + n += 1 + l + sovTest(uint64(l)) + } + if m.LatestBlockHash != nil { + l = len(m.LatestBlockHash) + n += 1 + l + sovTest(uint64(l)) + } + n += 1 + sovTest(uint64(m.LatestBlockHeight)) + n += 1 + sovTest(uint64(m.LatestBlocktime)) + return n +} + +func (m *NodeInfo) Size() (n int) { + var l int + _ = l + if m.PubKey != nil { + l = m.PubKey.Size() + n += 1 + l + sovTest(uint64(l)) + } + l = len(m.Moniker) + n += 1 + l + sovTest(uint64(l)) + l = len(m.Network) + n += 1 + l + sovTest(uint64(l)) + l = len(m.RemoteAddr) + n += 1 + l + sovTest(uint64(l)) + l = len(m.ListenAddr) + n += 1 + l + sovTest(uint64(l)) + l = len(m.Version) + n += 1 + l + sovTest(uint64(l)) + if len(m.Other) > 0 { + for _, s := range m.Other { + l = len(s) + n += 1 + l + sovTest(uint64(l)) + } + } + return n +} + +func (m *PubKey) Size() (n int) { + var l int + _ = l + if m.Ed25519 != nil { + l = m.Ed25519.Size() + n += 1 + l + sovTest(uint64(l)) + } + return n +} + +func (m *PubKeyEd25519) Size() (n int) { + var l int + _ = l + if m.Bytes != nil { + l = len(m.Bytes) + n += 1 + l + sovTest(uint64(l)) + } + return n +} + +func sovTest(x uint64) (n int) { + for { + n++ + x >>= 7 + if x == 0 { + break + } + } + return n +} +func sozTest(x uint64) (n int) { + return sovTest(uint64((x << 1) ^ uint64((int64(x) >> 63)))) +} +func (this *ResultStatus) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&ResultStatus{`, + `NodeInfo:` + strings.Replace(fmt.Sprintf("%v", this.NodeInfo), "NodeInfo", "NodeInfo", 1) + `,`, + `PubKey:` + strings.Replace(fmt.Sprintf("%v", this.PubKey), "PubKey", "PubKey", 1) + `,`, + `LatestBlockHash:` + fmt.Sprintf("%v", this.LatestBlockHash) + `,`, + `LatestBlockHeight:` + fmt.Sprintf("%v", this.LatestBlockHeight) + `,`, + `LatestBlocktime:` + fmt.Sprintf("%v", this.LatestBlocktime) + `,`, + `}`, + }, "") + return s +} +func (this *NodeInfo) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&NodeInfo{`, + `PubKey:` + strings.Replace(fmt.Sprintf("%v", this.PubKey), "PubKey", "PubKey", 1) + `,`, + `Moniker:` + fmt.Sprintf("%v", this.Moniker) + `,`, + `Network:` + fmt.Sprintf("%v", this.Network) + `,`, + `RemoteAddr:` + fmt.Sprintf("%v", this.RemoteAddr) + `,`, + `ListenAddr:` + fmt.Sprintf("%v", this.ListenAddr) + `,`, + `Version:` + fmt.Sprintf("%v", this.Version) + `,`, + `Other:` + fmt.Sprintf("%v", this.Other) + `,`, + `}`, + }, "") + return s +} +func (this *PubKey) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&PubKey{`, + `Ed25519:` + strings.Replace(fmt.Sprintf("%v", this.Ed25519), "PubKeyEd25519", "PubKeyEd25519", 1) + `,`, + `}`, + }, "") + return s +} +func (this *PubKeyEd25519) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&PubKeyEd25519{`, + `Bytes:` + fmt.Sprintf("%v", this.Bytes) + `,`, + `}`, + }, "") + return s +} +func valueToStringTest(v interface{}) string { + rv := reflect.ValueOf(v) + if rv.IsNil() { + return "nil" + } + pv := reflect.Indirect(rv).Interface() + return fmt.Sprintf("*%v", pv) +} +func (m *ResultStatus) Unmarshal(data []byte) error { + var hasFields [1]uint64 + l := len(data) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: ResultStatus: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: ResultStatus: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field NodeInfo", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.NodeInfo == nil { + m.NodeInfo = &NodeInfo{} + } + if err := m.NodeInfo.Unmarshal(data[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field PubKey", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.PubKey == nil { + m.PubKey = &PubKey{} + } + if err := m.PubKey.Unmarshal(data[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + hasFields[0] |= uint64(0x00000001) + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field LatestBlockHash", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + byteLen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + byteLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.LatestBlockHash = append([]byte{}, data[iNdEx:postIndex]...) + iNdEx = postIndex + hasFields[0] |= uint64(0x00000002) + case 4: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field LatestBlockHeight", wireType) + } + m.LatestBlockHeight = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + m.LatestBlockHeight |= (int64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + hasFields[0] |= uint64(0x00000004) + case 5: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field LatestBlocktime", wireType) + } + m.LatestBlocktime = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + m.LatestBlocktime |= (int64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + hasFields[0] |= uint64(0x00000008) + default: + iNdEx = preIndex + skippy, err := skipTest(data[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthTest + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + if hasFields[0]&uint64(0x00000001) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("pubKey") + } + if hasFields[0]&uint64(0x00000002) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("latestBlockHash") + } + if hasFields[0]&uint64(0x00000004) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("latestBlockHeight") + } + if hasFields[0]&uint64(0x00000008) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("latestBlocktime") + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *NodeInfo) Unmarshal(data []byte) error { + var hasFields [1]uint64 + l := len(data) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: NodeInfo: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: NodeInfo: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field PubKey", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.PubKey == nil { + m.PubKey = &PubKey{} + } + if err := m.PubKey.Unmarshal(data[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + hasFields[0] |= uint64(0x00000001) + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Moniker", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Moniker = string(data[iNdEx:postIndex]) + iNdEx = postIndex + hasFields[0] |= uint64(0x00000002) + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Network", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Network = string(data[iNdEx:postIndex]) + iNdEx = postIndex + hasFields[0] |= uint64(0x00000004) + case 4: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field RemoteAddr", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.RemoteAddr = string(data[iNdEx:postIndex]) + iNdEx = postIndex + hasFields[0] |= uint64(0x00000008) + case 5: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ListenAddr", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.ListenAddr = string(data[iNdEx:postIndex]) + iNdEx = postIndex + hasFields[0] |= uint64(0x00000010) + case 6: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Version", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Version = string(data[iNdEx:postIndex]) + iNdEx = postIndex + hasFields[0] |= uint64(0x00000020) + case 7: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Other", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Other = append(m.Other, string(data[iNdEx:postIndex])) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipTest(data[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthTest + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + if hasFields[0]&uint64(0x00000001) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("pubKey") + } + if hasFields[0]&uint64(0x00000002) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("moniker") + } + if hasFields[0]&uint64(0x00000004) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("network") + } + if hasFields[0]&uint64(0x00000008) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("remoteAddr") + } + if hasFields[0]&uint64(0x00000010) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("listenAddr") + } + if hasFields[0]&uint64(0x00000020) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("version") + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *PubKey) Unmarshal(data []byte) error { + l := len(data) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: PubKey: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: PubKey: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Ed25519", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.Ed25519 == nil { + m.Ed25519 = &PubKeyEd25519{} + } + if err := m.Ed25519.Unmarshal(data[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipTest(data[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthTest + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *PubKeyEd25519) Unmarshal(data []byte) error { + var hasFields [1]uint64 + l := len(data) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: PubKeyEd25519: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: PubKeyEd25519: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Bytes", wireType) + } + var byteLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowTest + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + byteLen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if byteLen < 0 { + return ErrInvalidLengthTest + } + postIndex := iNdEx + byteLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Bytes = append([]byte{}, data[iNdEx:postIndex]...) + iNdEx = postIndex + hasFields[0] |= uint64(0x00000001) + default: + iNdEx = preIndex + skippy, err := skipTest(data[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthTest + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + if hasFields[0]&uint64(0x00000001) == 0 { + return github_com_gogo_protobuf_proto.NewRequiredNotSetError("bytes") + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func skipTest(data []byte) (n int, err error) { + l := len(data) + iNdEx := 0 + for iNdEx < l { + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowTest + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + wireType := int(wire & 0x7) + switch wireType { + case 0: + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowTest + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + iNdEx++ + if data[iNdEx-1] < 0x80 { + break + } + } + return iNdEx, nil + case 1: + iNdEx += 8 + return iNdEx, nil + case 2: + var length int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowTest + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + length |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + iNdEx += length + if length < 0 { + return 0, ErrInvalidLengthTest + } + return iNdEx, nil + case 3: + for { + var innerWire uint64 + var start int = iNdEx + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowTest + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := data[iNdEx] + iNdEx++ + innerWire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + innerWireType := int(innerWire & 0x7) + if innerWireType == 4 { + break + } + next, err := skipTest(data[start:]) + if err != nil { + return 0, err + } + iNdEx = start + next + } + return iNdEx, nil + case 4: + return iNdEx, nil + case 5: + iNdEx += 4 + return iNdEx, nil + default: + return 0, fmt.Errorf("proto: illegal wireType %d", wireType) + } + } + panic("unreachable") +} + +var ( + ErrInvalidLengthTest = fmt.Errorf("proto: negative length found during unmarshaling") + ErrIntOverflowTest = fmt.Errorf("proto: integer overflow") +) diff --git a/benchmarks/proto/test.proto b/benchmarks/proto/test.proto new file mode 100644 index 00000000..d1a1c94d --- /dev/null +++ b/benchmarks/proto/test.proto @@ -0,0 +1,25 @@ +message ResultStatus { + optional NodeInfo nodeInfo = 1; + required PubKey pubKey = 2; + required bytes latestBlockHash = 3; + required int64 latestBlockHeight = 4; + required int64 latestBlocktime = 5; +} + +message NodeInfo { + required PubKey pubKey = 1; + required string moniker = 2; + required string network = 3; + required string remoteAddr = 4; + required string listenAddr = 5; + required string version = 6; + repeated string other = 7; +} + +message PubKey { + optional PubKeyEd25519 ed25519 = 1; +} + +message PubKeyEd25519 { + required bytes bytes = 1; +} diff --git a/benchmarks/wire_test.go b/benchmarks/wire_test.go deleted file mode 100644 index 7da8a7f7..00000000 --- a/benchmarks/wire_test.go +++ /dev/null @@ -1,60 +0,0 @@ -package benchmarks - -import ( - "testing" - - "github.com/tendermint/go-crypto" - "github.com/tendermint/go-p2p" - "github.com/tendermint/go-wire" - ctypes "github.com/tendermint/tendermint/rpc/core/types" -) - -func BenchmarkEncodeStatus(b *testing.B) { - b.StopTimer() - pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) - status := &ctypes.ResultStatus{ - NodeInfo: &p2p.NodeInfo{ - PubKey: pubKey, - Moniker: "SOMENAME", - Network: "SOMENAME", - RemoteAddr: "SOMEADDR", - ListenAddr: "SOMEADDR", - Version: "SOMEVER", - Other: []string{"SOMESTRING", "OTHERSTRING"}, - }, - PubKey: pubKey, - LatestBlockHash: []byte("SOMEBYTES"), - LatestBlockHeight: 123, - LatestBlockTime: 1234, - } - b.StartTimer() - - counter := 0 - for i := 0; i < b.N; i++ { - jsonBytes := wire.JSONBytes(status) - counter += len(jsonBytes) - } - -} - -func BenchmarkEncodeNodeInfo(b *testing.B) { - b.StopTimer() - pubKey := crypto.GenPrivKeyEd25519().PubKey().(crypto.PubKeyEd25519) - nodeInfo := &p2p.NodeInfo{ - PubKey: pubKey, - Moniker: "SOMENAME", - Network: "SOMENAME", - RemoteAddr: "SOMEADDR", - ListenAddr: "SOMEADDR", - Version: "SOMEVER", - Other: []string{"SOMESTRING", "OTHERSTRING"}, - } - b.StartTimer() - - counter := 0 - for i := 0; i < b.N; i++ { - jsonBytes := wire.JSONBytes(nodeInfo) - counter += len(jsonBytes) - } - -} From d0b8650edf4fcbd5cb8a7e684f187e046d9f5fa3 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 9 Dec 2015 09:37:36 -0800 Subject: [PATCH 07/14] Fix tests for counter application using uin64 --- benchmarks/simu/counter.go | 12 ++++++++++-- consensus/common_test.go | 2 +- mempool/mempool_test.go | 9 ++++----- 3 files changed, 15 insertions(+), 8 deletions(-) diff --git a/benchmarks/simu/counter.go b/benchmarks/simu/counter.go index 8191e8de..5f49eb9f 100644 --- a/benchmarks/simu/counter.go +++ b/benchmarks/simu/counter.go @@ -1,6 +1,9 @@ package main import ( + "encoding/binary" + "time" + //"encoding/hex" "fmt" "github.com/gorilla/websocket" @@ -31,16 +34,21 @@ func main() { }() // Make a bunch of requests - request := rpctypes.NewRPCRequest("fakeid", "net_info", nil) + buf := make([]byte, 32) for i := 0; ; i++ { + binary.LittleEndian.PutUint64(buf, uint64(i)) + //txBytes := hex.EncodeToString(buf[:n]) + request := rpctypes.NewRPCRequest("fakeid", "broadcast_tx", Arr(buf[:8])) reqBytes := wire.JSONBytes(request) + fmt.Println("!!", string(reqBytes)) err := ws.WriteMessage(websocket.TextMessage, reqBytes) if err != nil { Exit(err.Error()) } - if i%1000 == 0 { + if i%100 == 0 { fmt.Println(i) } + time.Sleep(time.Millisecond * 10) } ws.Stop() diff --git a/consensus/common_test.go b/consensus/common_test.go index 0c75bc7f..4f05dff9 100644 --- a/consensus/common_test.go +++ b/consensus/common_test.go @@ -278,7 +278,7 @@ func simpleConsensusState(nValidators int) (*ConsensusState, []*validatorStub) { blockStore := bc.NewBlockStore(blockDB) // one for mempool, one for consensus - app := example.NewCounterApplication() + app := example.NewCounterApplication(false) appCMem := app.Open() appCCon := app.Open() proxyAppCtxMem := proxy.NewLocalAppContext(appCMem) diff --git a/mempool/mempool_test.go b/mempool/mempool_test.go index 4cc55283..a3c5cff0 100644 --- a/mempool/mempool_test.go +++ b/mempool/mempool_test.go @@ -12,9 +12,8 @@ import ( func TestSerialReap(t *testing.T) { - app := example.NewCounterApplication() + app := example.NewCounterApplication(true) appCtxMempool := app.Open() - appCtxMempool.SetOption("serial", "on") proxyAppCtx := proxy.NewLocalAppContext(appCtxMempool) mempool := NewMempool(proxyAppCtx) @@ -28,7 +27,7 @@ func TestSerialReap(t *testing.T) { // This will succeed txBytes := make([]byte, 32) - _ = binary.PutVarint(txBytes, int64(i)) + binary.LittleEndian.PutUint64(txBytes, uint64(i)) err := mempool.AppendTx(txBytes) if err != nil { t.Fatal("Error after AppendTx: %v", err) @@ -59,7 +58,7 @@ func TestSerialReap(t *testing.T) { txs := make([]types.Tx, 0) for i := start; i < end; i++ { txBytes := make([]byte, 32) - _ = binary.PutVarint(txBytes, int64(i)) + binary.LittleEndian.PutUint64(txBytes, uint64(i)) txs = append(txs, txBytes) } blockHeader := &types.Header{Height: 0} @@ -75,7 +74,7 @@ func TestSerialReap(t *testing.T) { // Append some txs. for i := start; i < end; i++ { txBytes := make([]byte, 32) - _ = binary.PutVarint(txBytes, int64(i)) + binary.LittleEndian.PutUint64(txBytes, uint64(i)) _, retCode := appCtxConsensus.AppendTx(txBytes) if retCode != tmsp.RetCodeOK { t.Error("Error committing tx", retCode) From 15b751107e3b3bbb91ffbfb31897f7c089ea2818 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 9 Dec 2015 09:38:09 -0800 Subject: [PATCH 08/14] Tweak logging; allow consensus to bypass filter --- consensus/log.go | 11 ++++++++++- consensus/state.go | 12 +++++++----- mempool/reactor.go | 2 +- 3 files changed, 18 insertions(+), 7 deletions(-) diff --git a/consensus/log.go b/consensus/log.go index 0514a66d..26f7b858 100644 --- a/consensus/log.go +++ b/consensus/log.go @@ -4,4 +4,13 @@ import ( "github.com/tendermint/go-logger" ) -var log = logger.New("module", "consensus") +var log = logger.NewBypass("module", "consensus") + +func init() { + log.SetHandler( + logger.LvlFilterHandler( + logger.LvlDebug, + logger.BypassHandler(), + ), + ) +} diff --git a/consensus/state.go b/consensus/state.go index 867f019d..5ebac079 100644 --- a/consensus/state.go +++ b/consensus/state.go @@ -154,7 +154,7 @@ func (rs *RoundState) StringShort() string { type ConsensusState struct { BaseService - proxyAppCtx proxy.AppContext + proxyAppCtx proxy.AppContext blockStore *bc.BlockStore mempool *mempl.Mempool privValidator *types.PrivValidator @@ -172,10 +172,10 @@ type ConsensusState struct { func NewConsensusState(state *sm.State, proxyAppCtx proxy.AppContext, blockStore *bc.BlockStore, mempool *mempl.Mempool) *ConsensusState { cs := &ConsensusState{ - proxyAppCtx: proxyAppCtx, - blockStore: blockStore, - mempool: mempool, - newStepCh: make(chan *RoundState, 10), + proxyAppCtx: proxyAppCtx, + blockStore: blockStore, + mempool: mempool, + newStepCh: make(chan *RoundState, 10), } cs.updateToState(state) // Don't call scheduleRound0 yet. @@ -532,8 +532,10 @@ func (cs *ConsensusState) EnterPrevote(height int, round int, timedOut bool) { // fire event for how we got here if timedOut { + log.Debug("Propose timed out") cs.evsw.FireEvent(types.EventStringTimeoutPropose(), cs.RoundStateEvent()) } else if cs.isProposalComplete() { + log.Debug("Propose is complete") cs.evsw.FireEvent(types.EventStringCompleteProposal(), cs.RoundStateEvent()) } else { // we received +2/3 prevotes for a future round diff --git a/mempool/reactor.go b/mempool/reactor.go index ed8b8b26..d6140d2d 100644 --- a/mempool/reactor.go +++ b/mempool/reactor.go @@ -63,7 +63,7 @@ func (memR *MempoolReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) { log.Warn("Error decoding message", "error", err) return } - log.Notice("MempoolReactor received message", "msg", msg) + log.Info("MempoolReactor received message", "msg", msg) switch msg := msg.(type) { case *TxMessage: From ea378c5565fc8f57d7b100dea5972919ffab4a62 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 9 Dec 2015 11:54:08 -0800 Subject: [PATCH 09/14] Improve logging --- consensus/height_vote_set.go | 1 - consensus/reactor.go | 10 +++++----- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/consensus/height_vote_set.go b/consensus/height_vote_set.go index aa87728b..3ee69391 100644 --- a/consensus/height_vote_set.go +++ b/consensus/height_vote_set.go @@ -134,7 +134,6 @@ func (hvs *HeightVoteSet) POLRound() int { } func (hvs *HeightVoteSet) getVoteSet(round int, type_ byte) *types.VoteSet { - log.Debug("getVoteSet(round)", "round", round, "type", type_) rvs, ok := hvs.roundVoteSets[round] if !ok { return nil diff --git a/consensus/reactor.go b/consensus/reactor.go index 5da5ce56..b5de47ec 100644 --- a/consensus/reactor.go +++ b/consensus/reactor.go @@ -313,7 +313,7 @@ func (conR *ConsensusReactor) sendNewRoundStepMessage(peer *p2p.Peer) { } func (conR *ConsensusReactor) gossipDataRoutine(peer *p2p.Peer, ps *PeerState) { - log := log.New("peer", peer.Key) + log := log.New("peer", peer) OUTER_LOOP: for { @@ -419,7 +419,7 @@ OUTER_LOOP: } func (conR *ConsensusReactor) gossipVotesRoutine(peer *p2p.Peer, ps *PeerState) { - log := log.New("peer", peer.Key) + log := log.New("peer", peer) // Simple hack to throttle logs upon sleep. var sleeping = 0 @@ -441,8 +441,8 @@ OUTER_LOOP: sleeping = 0 } - log.Debug("gossipVotesRoutine", "rsHeight", rs.Height, "rsRound", rs.Round, - "prsHeight", prs.Height, "prsRound", prs.Round, "prsStep", prs.Step) + //log.Debug("gossipVotesRoutine", "rsHeight", rs.Height, "rsRound", rs.Round, + // "prsHeight", prs.Height, "prsRound", prs.Round, "prsStep", prs.Step) // If height matches, then send LastCommit, Prevotes, Precommits. if rs.Height == prs.Height { @@ -749,7 +749,7 @@ func (ps *PeerState) SetHasVote(vote *types.Vote, index int) { } func (ps *PeerState) setHasVote(height int, round int, type_ byte, index int) { - log := log.New("peer", ps.Peer.Key, "peerRound", ps.Round, "height", height, "round", round) + log := log.New("peer", ps.Peer, "peerRound", ps.Round, "height", height, "round", round) if type_ != types.VoteTypePrevote && type_ != types.VoteTypePrecommit { PanicSanity("Invalid vote type") } From dfdd3cf18fc95774d747857b70c92fb010630178 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 9 Dec 2015 13:53:31 -0800 Subject: [PATCH 10/14] Tweak channel for performance test --- benchmarks/map_test.go | 9 ++------- benchmarks/simu/counter.go | 7 ++++--- consensus/reactor.go | 14 ++++++++------ 3 files changed, 14 insertions(+), 16 deletions(-) diff --git a/benchmarks/map_test.go b/benchmarks/map_test.go index 9baaba6e..ee538e0f 100644 --- a/benchmarks/map_test.go +++ b/benchmarks/map_test.go @@ -11,11 +11,11 @@ func BenchmarkSomething(b *testing.B) { numChecks := 100000 keys := make([]string, numItems) for i := 0; i < numItems; i++ { - keys[i] = RandStr(32) + keys[i] = RandStr(100) } txs := make([]string, numChecks) for i := 0; i < numChecks; i++ { - txs[i] = RandStr(32) + txs[i] = RandStr(100) } b.StartTimer() @@ -30,10 +30,5 @@ func BenchmarkSomething(b *testing.B) { counter++ } } - for _, tx := range txs { - if _, ok := foo[tx]; ok { - counter++ - } - } } } diff --git a/benchmarks/simu/counter.go b/benchmarks/simu/counter.go index 5f49eb9f..8b0ba180 100644 --- a/benchmarks/simu/counter.go +++ b/benchmarks/simu/counter.go @@ -40,15 +40,16 @@ func main() { //txBytes := hex.EncodeToString(buf[:n]) request := rpctypes.NewRPCRequest("fakeid", "broadcast_tx", Arr(buf[:8])) reqBytes := wire.JSONBytes(request) - fmt.Println("!!", string(reqBytes)) + //fmt.Println("!!", string(reqBytes)) + fmt.Print(".") err := ws.WriteMessage(websocket.TextMessage, reqBytes) if err != nil { Exit(err.Error()) } - if i%100 == 0 { + if i%1000 == 0 { fmt.Println(i) } - time.Sleep(time.Millisecond * 10) + time.Sleep(time.Microsecond * 1) } ws.Stop() diff --git a/consensus/reactor.go b/consensus/reactor.go index b5de47ec..a96790c7 100644 --- a/consensus/reactor.go +++ b/consensus/reactor.go @@ -86,14 +86,16 @@ func (conR *ConsensusReactor) GetChannels() []*p2p.ChannelDescriptor { SendQueueCapacity: 100, }, &p2p.ChannelDescriptor{ - ID: DataChannel, - Priority: 5, - SendQueueCapacity: 2, + ID: DataChannel, + Priority: 2, + SendQueueCapacity: 50, + RecvBufferCapacity: 50 * 4096, }, &p2p.ChannelDescriptor{ - ID: VoteChannel, - Priority: 5, - SendQueueCapacity: 40, + ID: VoteChannel, + Priority: 5, + SendQueueCapacity: 100, + RecvBufferCapacity: 100 * 100, }, } } From d3d7ff481877202b5dad7d2b6a6dcc44ad51f34d Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 9 Dec 2015 14:10:31 -0800 Subject: [PATCH 11/14] Add mempool cache to reduce proxyapp pressure --- mempool/mempool.go | 27 +++++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/mempool/mempool.go b/mempool/mempool.go index b135cd5c..1684a2bf 100644 --- a/mempool/mempool.go +++ b/mempool/mempool.go @@ -2,6 +2,7 @@ package mempool import ( "bytes" + "container/list" "sync" "sync/atomic" @@ -36,10 +37,10 @@ Garbage collection of old elements from mempool.txs is handlde via the DetachPrev() call, which makes old elements not reachable by peer broadcastTxRoutine() automatically garbage collected. - - */ +const cacheSize = 100000 + type Mempool struct { proxyMtx sync.Mutex proxyAppCtx proxy.AppContext @@ -47,6 +48,11 @@ type Mempool struct { counter int64 // simple incrementing counter height int // the last block Update()'d to expected *clist.CElement // pointer to .txs for next response + + // Keep a cache of already-seen txs. + // This reduces the pressure on the proxyApp. + cacheMap map[string]struct{} + cacheList *list.List } func NewMempool(proxyAppCtx proxy.AppContext) *Mempool { @@ -56,6 +62,9 @@ func NewMempool(proxyAppCtx proxy.AppContext) *Mempool { counter: 0, height: 0, expected: nil, + + cacheMap: make(map[string]struct{}, cacheSize), + cacheList: list.New(), } proxyAppCtx.SetResponseCallback(mempool.resCb) return mempool @@ -73,6 +82,20 @@ func (mem *Mempool) AppendTx(tx types.Tx) (err error) { mem.proxyMtx.Lock() defer mem.proxyMtx.Unlock() + // CACHE + if _, exists := mem.cacheMap[string(tx)]; exists { + return nil + } + if mem.cacheList.Len() >= cacheSize { + popped := mem.cacheList.Front() + poppedTx := popped.Value.(types.Tx) + delete(mem.cacheMap, string(poppedTx)) + mem.cacheList.Remove(popped) + } + mem.cacheMap[string(tx)] = struct{}{} + mem.cacheList.PushBack(tx) + // END CACHE + if err = mem.proxyAppCtx.Error(); err != nil { return err } From f534410e572e974126ef8ca6a15beb07f344fdab Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 9 Dec 2015 17:08:40 -0800 Subject: [PATCH 12/14] Fix state/execution ExecBlock() not rolling back --- state/execution.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/state/execution.go b/state/execution.go index 69b533c8..2b4d57e6 100644 --- a/state/execution.go +++ b/state/execution.go @@ -31,10 +31,7 @@ func (s *State) ExecBlock(proxyAppCtx proxy.AppContext, block *types.Block, bloc nextValSet := valSet.Copy() // First, rollback. - if err != nil { - proxyAppCtx.RollbackSync() - return err - } + proxyAppCtx.RollbackSync() // Execute, or rollback. (Does not commit) err = s.execBlockOnProxyApp(proxyAppCtx, block) From ab1e8a35c6f1bd670066c2076b9cdba95ae56f85 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Wed, 9 Dec 2015 17:09:06 -0800 Subject: [PATCH 13/14] Shorten consensus log events --- benchmarks/simu/counter.go | 2 +- types/block.go | 6 +++++- types/part_set.go | 4 ++-- types/validator.go | 2 +- 4 files changed, 9 insertions(+), 5 deletions(-) diff --git a/benchmarks/simu/counter.go b/benchmarks/simu/counter.go index 8b0ba180..25b7703e 100644 --- a/benchmarks/simu/counter.go +++ b/benchmarks/simu/counter.go @@ -49,7 +49,7 @@ func main() { if i%1000 == 0 { fmt.Println(i) } - time.Sleep(time.Microsecond * 1) + time.Sleep(time.Microsecond * 250) } ws.Stop() diff --git a/types/block.go b/types/block.go index 752d6a56..e574f289 100644 --- a/types/block.go +++ b/types/block.go @@ -343,8 +343,12 @@ func (data *Data) StringIndented(indent string) string { if data == nil { return "nil-Data" } - txStrings := make([]string, len(data.Txs)) + txStrings := make([]string, MinInt(len(data.Txs), 21)) for i, tx := range data.Txs { + if i == 20 { + txStrings[i] = fmt.Sprintf("... (%v total)", len(data.Txs)) + break + } txStrings[i] = fmt.Sprintf("Tx:%v", tx) } return fmt.Sprintf(`Data{ diff --git a/types/part_set.go b/types/part_set.go index fc082667..9ab557bd 100644 --- a/types/part_set.go +++ b/types/part_set.go @@ -49,10 +49,10 @@ func (part *Part) String() string { func (part *Part) StringIndented(indent string) string { return fmt.Sprintf(`Part{ %s Proof: %v -%s Bytes: %X +%s Bytes: %X... %s}`, indent, part.Proof.StringIndented(indent+" "), - indent, part.Bytes, + indent, Fingerprint(part.Bytes), indent) } diff --git a/types/validator.go b/types/validator.go index d7fdb3b9..f68f5892 100644 --- a/types/validator.go +++ b/types/validator.go @@ -53,7 +53,7 @@ func (v *Validator) String() string { if v == nil { return "nil-Validator" } - return fmt.Sprintf("Validator{%X %v %v-%v-%v VP:%v A:%v}", + return fmt.Sprintf("Validator{%X %v %v VP:%v A:%v}", v.Address, v.PubKey, v.LastCommitHeight, From 34f6f50a07a9c2a4a44ff1b7710bb9ef1df44478 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Thu, 10 Dec 2015 11:09:25 -0800 Subject: [PATCH 14/14] Conform to go-merkle SimpleProof.Aunts --- consensus/reactor.go | 2 +- types/part_set.go | 25 ++++++++++++++----------- types/part_set_test.go | 2 +- 3 files changed, 16 insertions(+), 13 deletions(-) diff --git a/consensus/reactor.go b/consensus/reactor.go index a96790c7..55ef1da7 100644 --- a/consensus/reactor.go +++ b/consensus/reactor.go @@ -173,7 +173,7 @@ func (conR *ConsensusReactor) Receive(chID byte, peer *p2p.Peer, msgBytes []byte case *ProposalPOLMessage: ps.ApplyProposalPOLMessage(msg) case *BlockPartMessage: - ps.SetHasProposalBlockPart(msg.Height, msg.Round, msg.Part.Proof.Index) + ps.SetHasProposalBlockPart(msg.Height, msg.Round, msg.Part.Index) _, err = conR.conS.AddProposalBlockPart(msg.Height, msg.Part) default: log.Warn(Fmt("Unknown message type %v", reflect.TypeOf(msg))) diff --git a/types/part_set.go b/types/part_set.go index 9ab557bd..0883140d 100644 --- a/types/part_set.go +++ b/types/part_set.go @@ -24,8 +24,9 @@ var ( ) type Part struct { - Proof merkle.SimpleProof `json:"proof"` + Index int `json:"index"` Bytes []byte `json:"bytes"` + Proof merkle.SimpleProof `json:"proof"` // Cache hash []byte @@ -47,12 +48,13 @@ func (part *Part) String() string { } func (part *Part) StringIndented(indent string) string { - return fmt.Sprintf(`Part{ -%s Proof: %v + return fmt.Sprintf(`Part{#%v %s Bytes: %X... +%s Proof: %v %s}`, - indent, part.Proof.StringIndented(indent+" "), + part.Index, indent, Fingerprint(part.Bytes), + indent, part.Proof.StringIndented(indent+" "), indent) } @@ -101,6 +103,7 @@ func NewPartSetFromData(data []byte) *PartSet { partsBitArray := NewBitArray(total) for i := 0; i < total; i++ { part := &Part{ + Index: i, Bytes: data[i*partSize : MinInt(len(data), (i+1)*partSize)], } parts[i] = part @@ -108,13 +111,13 @@ func NewPartSetFromData(data []byte) *PartSet { partsBitArray.SetIndex(i, true) } // Compute merkle proofs - proofs := merkle.SimpleProofsFromHashables(parts_) + root, proofs := merkle.SimpleProofsFromHashables(parts_) for i := 0; i < total; i++ { parts[i].Proof = *proofs[i] } return &PartSet{ total: total, - hash: proofs[0].RootHash, + hash: root, parts: parts, partsBitArray: partsBitArray, count: total, @@ -190,23 +193,23 @@ func (ps *PartSet) AddPart(part *Part) (bool, error) { defer ps.mtx.Unlock() // Invalid part index - if part.Proof.Index >= ps.total { + if part.Index >= ps.total { return false, ErrPartSetUnexpectedIndex } // If part already exists, return false. - if ps.parts[part.Proof.Index] != nil { + if ps.parts[part.Index] != nil { return false, nil } // Check hash proof - if !part.Proof.Verify(part.Hash(), ps.Hash()) { + if !part.Proof.Verify(part.Index, ps.total, part.Hash(), ps.Hash()) { return false, ErrPartSetInvalidProof } // Add part - ps.parts[part.Proof.Index] = part - ps.partsBitArray.SetIndex(part.Proof.Index, true) + ps.parts[part.Index] = part + ps.partsBitArray.SetIndex(part.Index, true) ps.count++ return true, nil } diff --git a/types/part_set_test.go b/types/part_set_test.go index 0260af7f..4e74d777 100644 --- a/types/part_set_test.go +++ b/types/part_set_test.go @@ -69,7 +69,7 @@ func TestWrongProof(t *testing.T) { // Test adding a part with wrong trail. part := partSet.GetPart(0) - part.Proof.InnerHashes[0][0] += byte(0x01) + part.Proof.Aunts[0][0] += byte(0x01) added, err := partSet2.AddPart(part) if added || err == nil { t.Errorf("Expected to fail adding a part with bad trail.")