backport feat: Modify grpc gateway to be concurrent (#11234) from cos…

client/context.go

@@ -10,6 +10,8 @@ import (
 
 	"gopkg.in/yaml.v2"
 
+	"google.golang.org/grpc"
+
 	"github.com/gogo/protobuf/proto"
 	"github.com/pkg/errors"
 	rpcclient "github.com/tendermint/tendermint/rpc/client"
@@ -28,6 +30,7 @@ type Context struct {
 	ChainID     string
 	// Deprecated: Codec codec will be changed to Codec: codec.Codec
 	JSONCodec         codec.JSONCodec
+	GRPCClient        *grpc.ClientConn
 	Codec             codec.Codec
 	InterfaceRegistry codectypes.InterfaceRegistry
 	Input             io.Reader
@@ -140,6 +143,13 @@ func (ctx Context) WithClient(client rpcclient.Client) Context {
 	return ctx
 }
 
+// WithGRPCClient returns a copy of the context with an updated GRPC client
+// instance.
+func (ctx Context) WithGRPCClient(grpcClient *grpc.ClientConn) Context {
+	ctx.GRPCClient = grpcClient
+	return ctx
+}
+
 // WithUseLedger returns a copy of the context with an updated UseLedger flag.
 func (ctx Context) WithUseLedger(useLedger bool) Context {
 	ctx.UseLedger = useLedger

client/grpc_query.go

@@ -27,7 +27,8 @@ var protoCodec = encoding.GetCodec(proto.Name)
 func (ctx Context) Invoke(grpcCtx gocontext.Context, method string, req, reply interface{}, opts ...grpc.CallOption) (err error) {
 	// Two things can happen here:
 	// 1. either we're broadcasting a Tx, in which call we call Tendermint's broadcast endpoint directly,
-	// 2. or we are querying for state, in which case we call ABCI's Query.
+	// 2-1. or we are querying for state, in which case we call grpc if grpc client set.
+	// 2-2. or we are querying for state, in which case we call ABCI's Query if grpc client not set.
 
 	// In both cases, we don't allow empty request args (it will panic unexpectedly).
 	if reflect.ValueOf(req).IsNil() {
@@ -50,7 +51,12 @@ func (ctx Context) Invoke(grpcCtx gocontext.Context, method string, req, reply i
 		return err
 	}
 
-	// Case 2. Querying state.
+	if ctx.GRPCClient != nil {
+		// Case 2-1. Invoke grpc.
+		return ctx.GRPCClient.Invoke(grpcCtx, method, req, reply, opts...)
+	}
+
+	// Case 2-2. Querying state via abci query.
 	reqBz, err := protoCodec.Marshal(req)
 	if err != nil {
 		return err

server/start.go

@@ -4,6 +4,7 @@ package server
 
 import (
 	"fmt"
+	"net"
 	"net/http"
 	"os"
 	"runtime/pprof"
@@ -25,6 +26,8 @@ import (
 
 	"github.com/cosmos/cosmos-sdk/server/rosetta"
 	crgserver "github.com/cosmos/cosmos-sdk/server/rosetta/lib/server"
+	"google.golang.org/grpc/encoding"
+	"google.golang.org/grpc/encoding/proto"
 
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/client/flags"
@@ -301,6 +304,25 @@ func startInProcess(ctx *Context, clientCtx client.Context, appCreator types.App
 			WithHomeDir(home).
 			WithChainID(genDoc.ChainID)
 
+		if config.GRPC.Enable {
+			_, port, err := net.SplitHostPort(config.GRPC.Address)
+			if err != nil {
+				return err
+			}
+			grpcAddress := fmt.Sprintf("127.0.0.1:%s", port)
+			// If grpc is enabled, configure grpc client for grpc gateway.
+			grpcClient, err := grpc.Dial(
+				grpcAddress,
+				grpc.WithInsecure(),
+				grpc.WithDefaultCallOptions(grpc.ForceCodec(encoding.GetCodec(proto.Name))),
+			)
+			if err != nil {
+				return err
+			}
+			clientCtx = clientCtx.WithGRPCClient(grpcClient)
+			ctx.Logger.Debug("grpc client assigned to client context", "target", grpcAddress)
+		}
+
 		apiSrv = api.New(clientCtx, ctx.Logger.With("module", "api-server"))
 		app.RegisterAPIRoutes(apiSrv, config.API)
 		errCh := make(chan error)