Merge pull request #351 from cloudwego/develop

chore: merge develop into release/v0.2.0

client/client.go

@@ -353,7 +353,7 @@ func (kc *kClient) invokeHandleEndpoint() (endpoint.Endpoint, error) {
 		if err = cli.Send(ctx, ri, sendMsg); err != nil {
 			return
 		}
-		if resp == nil || m.OneWay() {
+		if m.OneWay() {
 			cli.Recv(ctx, ri, nil)
 			return nil
 		}

client/genericclient/client.go

@@ -52,9 +52,31 @@ func NewClientWithServiceInfo(destService string, g generic.Generic, svcInfo *se
 	}
 	runtime.SetFinalizer(cli, (*genericServiceClient).Close)
 
+	svcInfo.GenericMethod = func(name string) serviceinfo.MethodInfo {
+		m := svcInfo.Methods[serviceinfo.GenericMethod]
+		n, err := g.GetMethod(nil, name)
+		if err != nil {
+			return m
+		}
+		return &methodInfo{
+			MethodInfo: m,
+			oneway:     n.Oneway,
+		}
+	}
+
 	return cli, nil
 }
 
+// methodInfo is a wrapper to update the oneway flag of a service.MethodInfo.
+type methodInfo struct {
+	serviceinfo.MethodInfo
+	oneway bool
+}
+
+func (m *methodInfo) OneWay() bool {
+	return m.oneway
+}
+
 // Client generic client
 type Client interface {
 	generic.Closer

client/option.go

@@ -35,6 +35,7 @@ import (
 	"github.com/cloudwego/kitex/pkg/loadbalance/lbcache"
 	"github.com/cloudwego/kitex/pkg/remote"
 	"github.com/cloudwego/kitex/pkg/remote/trans/netpollmux"
+	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/grpc"
 	"github.com/cloudwego/kitex/pkg/retry"
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/cloudwego/kitex/pkg/stats"
@@ -357,3 +358,55 @@ func WithCircuitBreaker(s *circuitbreak.CBSuite) Option {
 		o.CBSuite = s
 	}}
 }
+
+// WithGRPCConnPoolSize sets the value for the client connection pool size.
+// In general, you should not adjust the size of the connection pool, otherwise it may cause performance degradation.
+// You should adjust the size according to the actual situation.
+func WithGRPCConnPoolSize(s uint32) Option {
+	return Option{F: func(o *client.Options, di *utils.Slice) {
+		di.Push(fmt.Sprintf("WithGRPCConnPoolSize(%d)", s))
+		o.GRPCConnPoolSize = s
+	}}
+}
+
+// WithGRPCInitialWindowSize sets the value for initial window size on a grpc stream.
+// The lower bound for window size is 64K and any value smaller than that will be ignored.
+// It corresponds to the WithInitialWindowSize DialOption of gRPC.
+func WithGRPCInitialWindowSize(s uint32) Option {
+	return Option{F: func(o *client.Options, di *utils.Slice) {
+		di.Push(fmt.Sprintf("WithGRPCInitialWindowSize(%d)", s))
+		o.GRPCConnectOpts.InitialWindowSize = s
+	}}
+}
+
+// WithGRPCInitialConnWindowSize sets the value for initial window size on a connection of grpc.
+// The lower bound for window size is 64K and any value smaller than that will be ignored.
+// It corresponds to the WithInitialConnWindowSize DialOption of gRPC.
+func WithGRPCInitialConnWindowSize(s uint32) Option {
+	return Option{F: func(o *client.Options, di *utils.Slice) {
+		di.Push(fmt.Sprintf("WithGRPCInitialConnWindowSize(%d)", s))
+		o.GRPCConnectOpts.InitialConnWindowSize = s
+	}}
+}
+
+// WithGRPCMaxHeaderListSize returns a DialOption that specifies the maximum
+// (uncompressed) size of header list that the client is prepared to accept.
+// It corresponds to the WithMaxHeaderListSize DialOption of gRPC.
+func WithGRPCMaxHeaderListSize(s uint32) Option {
+	return Option{F: func(o *client.Options, di *utils.Slice) {
+		di.Push(fmt.Sprintf("WithGRPCMaxHeaderListSize(%d)", s))
+		o.GRPCConnectOpts.MaxHeaderListSize = &s
+	}}
+}
+
+// WithGRPCKeepaliveParams returns a DialOption that specifies keepalive parameters for the client transport.
+// It corresponds to the WithKeepaliveParams DialOption of gRPC.
+func WithGRPCKeepaliveParams(kp grpc.ClientKeepalive) Option {
+	if kp.Time < grpc.KeepaliveMinPingTime {
+		kp.Time = grpc.KeepaliveMinPingTime
+	}
+	return Option{F: func(o *client.Options, di *utils.Slice) {
+		di.Push(fmt.Sprintf("WithGRPCKeepaliveParams(%+v)", kp))
+		o.GRPCConnectOpts.KeepaliveParams = kp
+	}}
+}

client/rpctimeout.go

@@ -109,17 +109,20 @@ func rpcTimeoutMW(mwCtx context.Context) endpoint.Middleware {
 						e := panicToErr(ctx, panicInfo, ri)
 						done <- e
 					}
-					if !errors.Is(err, kerrors.ErrRPCFinish) {
+					if err == nil || !errors.Is(err, kerrors.ErrRPCFinish) {
 						// Don't regards ErrRPCFinish as normal error, it happens in retry scene,
 						// ErrRPCFinish means previous call returns first but is decoding.
 						close(done)
 					}
 				}()
 				err = next(ctx, request, response)
-				if err != nil && ctx.Err() != nil && !errors.Is(err, kerrors.ErrRPCFinish) {
-					// error occurs after the wait goroutine returns,
-					// we should log this error or it will be discarded.
-					// Specially, ErrRPCFinish can be ignored, it happens in retry scene, previous call returns first.
+				if err != nil && ctx.Err() != nil &&
+					!errors.Is(err, kerrors.ErrRPCTimeout) && !errors.Is(err, kerrors.ErrRPCFinish) {
+					// error occurs after the wait goroutine returns(RPCTimeout happens),
+					// we should log this error for troubleshooting, or it will be discarded.
+					// but ErrRPCTimeout and ErrRPCFinish can be ignored:
+					//    ErrRPCTimeout: it is same with outer timeout, here only care about non timeout err.
+					//    ErrRPCFinish: it happens in retry scene, previous call returns first.
 					var errMsg string
 					if ri.To().Address() != nil {
 						errMsg = fmt.Sprintf("KITEX: to_service=%s method=%s addr=%s error=%s",

internal/client/option.go

@@ -40,6 +40,7 @@ import (
 	"github.com/cloudwego/kitex/pkg/remote/connpool"
 	"github.com/cloudwego/kitex/pkg/remote/trans/netpoll"
 	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2"
+	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/grpc"
 	"github.com/cloudwego/kitex/pkg/retry"
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/cloudwego/kitex/pkg/serviceinfo"
@@ -98,6 +99,10 @@ type Options struct {
 	RetryContainer *retry.Container
 
 	CloseCallbacks []func() error
+
+	// GRPC
+	GRPCConnPoolSize uint32
+	GRPCConnectOpts  *grpc.ConnectOptions
 }
 
 // Apply applies all options.
@@ -128,6 +133,8 @@ func NewOptions(opts []Option) *Options {
 		Events: event.NewQueue(event.MaxEventNum),
 
 		TracerCtl: &internal_stats.Controller{},
+
+		GRPCConnectOpts: new(grpc.ConnectOptions),
 	}
 	o.Apply(opts)
 	o.MetaHandlers = append(o.MetaHandlers, transmeta.MetainfoClientHandler)
@@ -150,7 +157,7 @@ func NewOptions(opts []Option) *Options {
 
 func (o *Options) initRemoteOpt() {
 	if o.Configs.TransportProtocol()&transport.GRPC == transport.GRPC {
-		o.RemoteOpt.ConnPool = nphttp2.NewConnPool(o.Svr.ServiceName)
+		o.RemoteOpt.ConnPool = nphttp2.NewConnPool(o.Svr.ServiceName, o.GRPCConnPoolSize, *o.GRPCConnectOpts)
 		o.RemoteOpt.CliHandlerFactory = nphttp2.NewCliTransHandlerFactory()
 	}
 	if o.RemoteOpt.ConnPool == nil {

internal/server/option.go

@@ -38,6 +38,7 @@ import (
 	"github.com/cloudwego/kitex/pkg/remote/codec/thrift"
 	"github.com/cloudwego/kitex/pkg/remote/trans/detection"
 	"github.com/cloudwego/kitex/pkg/remote/trans/netpoll"
+	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/grpc"
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/cloudwego/kitex/pkg/serviceinfo"
 	"github.com/cloudwego/kitex/pkg/stats"
@@ -130,6 +131,7 @@ func newServerOption() *remote.ServerOption {
 		ExitWaitTime:          defaultExitWaitTime,
 		MaxConnectionIdleTime: defaultConnectionIdleTime,
 		AcceptFailedDelayTime: defaultAcceptFailedDelayTime,
+		GRPCCfg:               new(grpc.ServerConfig),
 	}
 }
 

licenses/LICENSE-gjson

@@ -0,0 +1,20 @@
+The MIT License (MIT)
+
+Copyright (c) 2016 Josh Baker
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.