grpc enable instrumentation (#84)

gointerfaces/grpcutil/utils.go

@@ -15,6 +15,7 @@ import (
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/reflection"
 )
 
 func TLS(tlsCACert, tlsCertFile, tlsKeyFile string) (credentials.TransportCredentials, error) {
@@ -60,11 +61,12 @@ func NewServer(rateLimit uint32, creds credentials.TransportCredentials) *grpc.S
 	//}
 
 	var grpcServer *grpc.Server
+	reflection.Register(grpcServer)
 	//cpus := uint32(runtime.GOMAXPROCS(-1))
 	opts := []grpc.ServerOption{
 		//grpc.NumStreamWorkers(cpus), // reduce amount of goroutines
-		grpc.WriteBufferSize(1024), // reduce buffers to save mem
+		//grpc.WriteBufferSize(1024), // reduce buffers to save mem
-		grpc.ReadBufferSize(1024),
+		//grpc.ReadBufferSize(1024),
 		grpc.MaxConcurrentStreams(rateLimit), // to force clients reduce concurrency level
 		// Don't drop the connection, settings accordign to this comment on GitHub
 		// https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779

kv/kvcache/cache.go

@@ -116,7 +116,7 @@ type CoherentCacheConfig struct {
 var DefaultCoherentCacheConfig = CoherentCacheConfig{
 	KeepViews:    50,
 	NewBlockWait: 50 * time.Millisecond,
-	KeysLimit:    400_000,
+	KeysLimit:    1_000_000,
 	MetricsLabel: "default",
 	WithStorage:  true,
 }

txpool/grpc_server.go

@@ -36,6 +36,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/reflection"
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
@@ -290,12 +291,11 @@ func StartGrpc(txPoolServer txpool_proto.TxpoolServer, miningServer txpool_proto
 	//}
 
 	var grpcServer *grpc.Server
+	reflection.Register(grpcServer) // Register reflection service on gRPC server.
 	//cpus := uint32(runtime.GOMAXPROCS(-1))
 	opts := []grpc.ServerOption{
 		//grpc.NumStreamWorkers(cpus), // reduce amount of goroutines
-		grpc.WriteBufferSize(1024), // reduce buffers to save mem
+
-		grpc.ReadBufferSize(1024),
-		grpc.MaxConcurrentStreams(kv.ReadersLimit - 128), // to force clients reduce concurrency level
 		// Don't drop the connection, settings accordign to this comment on GitHub
 		// https://github.com/grpc/grpc-go/issues/3171#issuecomment-552796779
 		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{

txpool/pool.go

@@ -331,6 +331,26 @@ func (p *TxPool) OnNewBlock(ctx context.Context, stateChanges *remote.StateChang
 		return err
 	}
 
+	for _, changesList := range stateChanges.ChangeBatch {
+		for _, change := range changesList.Changes {
+			switch change.Action {
+			case remote.Action_UPSERT, remote.Action_UPSERT_CODE:
+				if change.Incarnation > 0 {
+					continue
+				}
+				addr := gointerfaces.ConvertH160toAddress(change.Address)
+				id, ok := p.senders.id(string(addr[:]))
+				if !ok {
+					continue
+				}
+				nonce, balance, err := p.senders.info(cache, viewID, coreTx, id)
+				if err != nil {
+					return err
+				}
+				onSenderChange(id, nonce, balance, p.byNonce, protocolBaseFee, baseFee, p.pending, p.baseFee, p.queued)
+			}
+		}
+	}
 	//log.Debug("[txpool] new block", "unwinded", len(unwindTxs.txs), "mined", len(minedTxs.txs), "baseFee", baseFee, "blockHeight", blockHeight)
 	if err := addTxs(p.lastSeenBlock.Load(), cache, viewID, coreTx, p.cfg, p.senders, unwindTxs, protocolBaseFee, baseFee, p.pending, p.baseFee, p.queued, p.byNonce, p.byHash, p.addLocked, p.discardLocked); err != nil {
 		return err