Example of client send task and monitor task

benchmarks/codec_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 
 	proto "github.com/tendermint/tendermint/benchmarks/proto"
@@ -32,7 +32,7 @@ func BenchmarkEncodeStatusWire(b *testing.B) {
 			LatestBlockTime:   time.Unix(0, 1234),
 		},
 		ValidatorInfo: ctypes.ValidatorInfo{
-			PubKey:            nodeKey.PubKey(),
+			PubKey: nodeKey.PubKey(),
 		},
 	}
 	b.StartTimer()

benchmarks/experiments/Makefile

@@ -0,0 +1,26 @@
+DIST_DIRS := find * -type d -exec
+VERSION := $(shell perl -ne '/^var version.*"([^"]+)".*$$/ && print "v$$1\n"' main.go)
+GOTOOLS = \
+					github.com/mitchellh/gox
+
+tools:
+	go get $(GOTOOLS)
+
+get_vendor_deps:
+	@hash glide 2>/dev/null || go get github.com/Masterminds/glide
+	glide install
+
+build:
+	go build
+
+install:
+	go install
+
+test:
+	go test -race
+
+clean:
+	rm -f ./experiments
+	rm -rf ./dist
+
+.PHONY: tools get_vendor_deps build install test clean

benchmarks/experiments/glide.yaml

@@ -0,0 +1,12 @@
+package: github.com/tendermint/tendermint/benchmarks/experiments
+import:
+- package: github.com/tendermint/tendermint
+  version: v0.16.0
+  subpackages:
+  - rpc/client
+  - rpc/lib/types
+  - types
+- package: github.com/tendermint/tmlibs
+  version: v0.7.0
+  subpackages:
+  - log

benchmarks/experiments/main.go

@@ -0,0 +1,126 @@
+package main
+
+import (
+	"encoding/binary"
+	"fmt"
+	"math/rand"
+	"os"
+	"sync"
+	"time"
+
+	"context"
+
+	"github.com/tendermint/tendermint/rpc/client"
+	"github.com/tendermint/tendermint/types"
+	"github.com/tendermint/tmlibs/log"
+)
+
+var logger = log.NewNopLogger()
+var finishedTasks = 0
+var mutex = &sync.Mutex{}
+
+func main() {
+
+	var endpoint = "tcp://0.0.0.0:46657"
+
+	var httpClient = getHTTPClient(endpoint)
+
+	var res, err = httpClient.Status()
+	if err != nil {
+		logger.Info("something wrong happens", err)
+	}
+	logger.Info("received status", res)
+
+	go monitorTask(endpoint)
+
+	txCount := 10
+	var clientNumber = 10
+	for i := 0; i < clientNumber; i++ {
+		go clientTask(i, txCount, endpoint)
+	}
+	for finishedTasks < clientNumber+1 {
+	}
+	fmt.Printf("Done: %d\n", finishedTasks)
+}
+
+func clientTask(id, txCount int, endpoint string) {
+	var httpClient = getHTTPClient(endpoint)
+	for i := 0; i < txCount; i++ {
+		var _, err = httpClient.BroadcastTxSync(generateTx(id, rand.Int()))
+		if err != nil {
+			fmt.Printf("Something wrong happened: %s\n", err)
+		}
+	}
+	fmt.Printf("Finished client task: %d\n", id)
+
+	mutex.Lock()
+	finishedTasks++
+	mutex.Unlock()
+}
+
+func getHTTPClient(rpcAddr string) *client.HTTP {
+	return client.NewHTTP(rpcAddr, "/websocket")
+}
+
+func generateTx(i, valI int) []byte {
+	// a tx encodes the validator index, the tx number, and some random junk
+	tx := make([]byte, 250)
+	binary.PutUvarint(tx[:32], uint64(valI))
+	binary.PutUvarint(tx[32:64], uint64(i))
+	if _, err := rand.Read(tx[65:]); err != nil {
+		fmt.Println("err reading from crypto/rand", err)
+		os.Exit(1)
+	}
+	return tx
+}
+
+func monitorTask(endpoint string) {
+	fmt.Println("Monitor task started...")
+	var duration = 5 * time.Second
+
+	const subscriber = "monitor"
+	ctx, cancel := context.WithTimeout(context.Background(), duration)
+	defer cancel()
+	evts := make(chan interface{})
+
+	var httpClient = getHTTPClient(endpoint)
+	httpClient.Start()
+
+	evtTyp := types.EventNewBlockHeader
+
+	// register for the next event of this type
+	query := types.QueryForEvent(evtTyp)
+	err := httpClient.Subscribe(ctx, subscriber, query, evts)
+	if err != nil {
+		fmt.Println("error when subscribing", err)
+	}
+
+	// make sure to unregister after the test is over
+	defer httpClient.UnsubscribeAll(ctx, subscriber)
+
+	totalNumOfCommittedTxs := int64(0)
+
+	for {
+		fmt.Println("Starting main loop", err)
+		select {
+		case evt := <-evts:
+			event := evt.(types.TMEventData)
+			header, ok := event.Unwrap().(types.EventDataNewBlockHeader)
+			if ok {
+				fmt.Println("received header\n", header.Header.StringIndented(""))
+			} else {
+				fmt.Println("not able to unwrap header")
+			}
+			// Do some metric computation with header
+			totalNumOfCommittedTxs += header.Header.NumTxs
+
+		case <-ctx.Done():
+			fmt.Printf("Finished monitor task. Received %d transactions \n", totalNumOfCommittedTxs)
+
+			mutex.Lock()
+			finishedTasks++
+			mutex.Unlock()
+			return
+		}
+	}
+}

blockchain/reactor.go

@@ -5,7 +5,7 @@ import (
 	"reflect"
 	"time"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tendermint/p2p"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"

blockchain/wire.go

@@ -1,7 +1,7 @@
 package blockchain
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

cmd/tendermint/commands/show_node_id.go

@@ -6,7 +6,6 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/tendermint/tendermint/p2p"
-
 )
 
 // ShowNodeIDCmd dumps node's ID to the standard output.

cmd/tendermint/commands/show_validator.go

@@ -2,6 +2,7 @@ package commands
 
 import (
 	"fmt"
+
 	"github.com/spf13/cobra"
 
 	privval "github.com/tendermint/tendermint/types/priv_validator"

cmd/tendermint/commands/wire.go

@@ -1,7 +1,7 @@
 package commands
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

consensus/wal.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tendermint/types"
 	auto "github.com/tendermint/tmlibs/autofile"
 	cmn "github.com/tendermint/tmlibs/common"

consensus/wire.go

@@ -1,7 +1,7 @@
 package consensus
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

evidence/reactor.go

@@ -5,7 +5,7 @@ import (
 	"reflect"
 	"time"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tmlibs/log"
 
 	"github.com/tendermint/tendermint/p2p"

evidence/wire.go

@@ -1,7 +1,7 @@
 package evidence
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 	"github.com/tendermint/tendermint/types"
 )

lite/files/wire.go

@@ -1,7 +1,7 @@
 package files
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

lite/proxy/proxy.go

@@ -3,7 +3,7 @@ package proxy
 import (
 	"net/http"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tmlibs/log"
 
 	rpcclient "github.com/tendermint/tendermint/rpc/client"

mempool/reactor.go

@@ -6,7 +6,7 @@ import (
 	"time"
 
 	abci "github.com/tendermint/abci/types"
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tmlibs/clist"
 	"github.com/tendermint/tmlibs/log"
 

mempool/wire.go

@@ -1,8 +1,6 @@
 package mempool
 
-import (
+import amino "github.com/tendermint/go-amino"
-	"github.com/tendermint/go-amino"
-)
 
 var cdc = amino.NewCodec()
 

p2p/conn/connection_test.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/tmlibs/log"
 )
 

p2p/conn/wire.go

@@ -1,7 +1,7 @@
 package conn
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

p2p/pex/pex_reactor.go

@@ -7,7 +7,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	cmn "github.com/tendermint/tmlibs/common"
 
 	"github.com/tendermint/tendermint/p2p"

p2p/pex/wire.go

@@ -1,8 +1,6 @@
 package pex
 
-import (
+import amino "github.com/tendermint/go-amino"
-	"github.com/tendermint/go-amino"
-)
 
 var cdc *amino.Codec = amino.NewCodec()
 

p2p/wire.go

@@ -1,7 +1,7 @@
 package p2p
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

rpc/client/helpers_test.go

@@ -32,7 +32,7 @@ func TestWaitForHeight(t *testing.T) {
 
 	// now set current block height to 10
 	m.Call = mock.Call{
-		Response: &ctypes.ResultStatus{SyncInfo: ctypes.SyncInfo{LatestBlockHeight: 10} },
+		Response: &ctypes.ResultStatus{SyncInfo: ctypes.SyncInfo{LatestBlockHeight: 10}},
 	}
 
 	// we will not wait for more than 10 blocks

rpc/core/types/wire.go

@@ -1,7 +1,7 @@
 package core_types
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 	"github.com/tendermint/tendermint/types"
 )

rpc/lib/client/args_test.go

@@ -5,8 +5,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
+	amino "github.com/tendermint/go-amino"
-	"github.com/tendermint/go-amino"
 )
 
 type Tx []byte

rpc/lib/client/http_client.go

@@ -12,8 +12,8 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
-	"github.com/tendermint/go-amino"
 
+	amino "github.com/tendermint/go-amino"
 	types "github.com/tendermint/tendermint/rpc/lib/types"
 )
 

rpc/lib/client/ws_client.go

@@ -13,7 +13,7 @@ import (
 	"github.com/pkg/errors"
 	metrics "github.com/rcrowley/go-metrics"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	types "github.com/tendermint/tendermint/rpc/lib/types"
 	cmn "github.com/tendermint/tmlibs/common"
 )

rpc/lib/rpc_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	cmn "github.com/tendermint/tmlibs/common"
 	"github.com/tendermint/tmlibs/log"
 

rpc/lib/server/handlers.go

@@ -17,7 +17,7 @@ import (
 	"github.com/gorilla/websocket"
 	"github.com/pkg/errors"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	types "github.com/tendermint/tendermint/rpc/lib/types"
 	cmn "github.com/tendermint/tmlibs/common"
 	"github.com/tendermint/tmlibs/log"

rpc/lib/server/handlers_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	rs "github.com/tendermint/tendermint/rpc/lib/server"
 	types "github.com/tendermint/tendermint/rpc/lib/types"
 	"github.com/tendermint/tmlibs/log"

rpc/lib/server/parse_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	cmn "github.com/tendermint/tmlibs/common"
 )
 

rpc/lib/test/main.go

@@ -5,7 +5,7 @@ import (
 	"net/http"
 	"os"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	rpcserver "github.com/tendermint/tendermint/rpc/lib/server"
 	cmn "github.com/tendermint/tmlibs/common"
 	"github.com/tendermint/tmlibs/log"

rpc/lib/types/types.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	tmpubsub "github.com/tendermint/tmlibs/pubsub"
 )
 

rpc/lib/types/types_test.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 )
 
 type SampleResult struct {

scripts/wire2amino.go

@@ -8,7 +8,7 @@ import (
 	"path/filepath"
 	"time"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	crypto "github.com/tendermint/go-crypto"
 	cmn "github.com/tendermint/tmlibs/common"
 

state/txindex/kv/wire.go

@@ -1,8 +1,6 @@
 package kv
 
-import (
+import amino "github.com/tendermint/go-amino"
-	"github.com/tendermint/go-amino"
-)
 
 var cdc = amino.NewCodec()
 

state/wire.go

@@ -1,7 +1,7 @@
 package state
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

types/canonical_json.go

@@ -3,7 +3,7 @@ package types
 import (
 	"time"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	cmn "github.com/tendermint/tmlibs/common"
 )
 

types/events.go

@@ -3,7 +3,7 @@ package types
 import (
 	"fmt"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	tmpubsub "github.com/tendermint/tmlibs/pubsub"
 	tmquery "github.com/tendermint/tmlibs/pubsub/query"
 )

types/evidence.go

@@ -4,7 +4,7 @@ import (
 	"bytes"
 	"fmt"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 	"github.com/tendermint/tmlibs/merkle"
 )

types/genesis_test.go

@@ -1,9 +1,10 @@
 package types
 
 import (
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/tendermint/go-crypto"
-	"testing"
 )
 
 func TestGenesisBad(t *testing.T) {

types/priv_validator/socket.go

@@ -7,7 +7,7 @@ import (
 	"net"
 	"time"
 
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 	cmn "github.com/tendermint/tmlibs/common"
 	"github.com/tendermint/tmlibs/log"

types/priv_validator/wire.go

@@ -1,7 +1,7 @@
 package privval
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )
 

types/wire.go

@@ -1,7 +1,7 @@
 package types
 
 import (
-	"github.com/tendermint/go-amino"
+	amino "github.com/tendermint/go-amino"
 	"github.com/tendermint/go-crypto"
 )