#743 refactor to improve the performance

2025-09-26 20:21:14 +08:00 · 2022-07-10 22:44:47 +08:00
parent 0f372c4f24
commit f5fc9d3a91
7 changed files with 121 additions and 80 deletions
--- a/client/client.go
+++ b/client/client.go
@@ -34,7 +34,10 @@ const (
 )

 // ServiceError is an error from server.
-type ServiceError error
+type ServiceError interface {
+	Error() string
+	IsServiceError() bool
+}

 var ClientErrorFunc func(e string) ServiceError

@@ -44,6 +47,10 @@ func (s strErr) Error() string {
 	return string(s)
 }

+func (s strErr) IsServiceError() bool {
+	return true
+}
+
 // DefaultOption is a common option configuration for client.
 var DefaultOption = Option{
 	Retries:             3,
--- a/client/connection_nonquic.go
+++ b/client/connection_nonquic.go
@@ -1,3 +1,4 @@
+//go:build !quic
 // +build !quic

 package client
--- a/client/connection_quic.go
+++ b/client/connection_quic.go
@@ -1,3 +1,4 @@
+//go:build quic
 // +build quic

 package client
@@ -21,9 +22,7 @@ func newDirectQuicConn(c *Client, network, address string) (net.Conn, error) {
 		tlsConf.NextProtos = []string{"rpcx"}
 	}

-	quicConfig := &quic.Config{
-		KeepAlive: c.option.Heartbeat,
-	}
+	quicConfig := &quic.Config{}

 	return quick.Dial(address, tlsConf, quicConfig)
 }
--- a/go.mod
+++ b/go.mod
@@ -38,12 +38,13 @@ require (
 	github.com/xtaci/kcp-go v5.4.20+incompatible
 	go.opentelemetry.io/otel v1.7.0
 	go.opentelemetry.io/otel/trace v1.7.0
-	golang.org/x/net v0.0.0-20220630215102-69896b714898
+	golang.org/x/net v0.0.0-20220708220712-1185a9018129
 	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f
 	google.golang.org/protobuf v1.28.0
 )

 require (
+	github.com/alphadose/itogami v0.0.0-20220705100819-134f04183c42 // indirect
 	github.com/armon/go-metrics v0.4.0 // indirect
 	github.com/cenk/backoff v2.2.1+incompatible // indirect
 	github.com/cenkalti/backoff v2.2.1+incompatible // indirect
@@ -93,7 +94,7 @@ require (
 	github.com/xtaci/lossyconn v0.0.0-20200209145036-adba10fffc37 // indirect
 	golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d // indirect
 	golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect
-	golang.org/x/sys v0.0.0-20220702020025-31831981b65f // indirect
+	golang.org/x/sys v0.0.0-20220708085239-5a0f0661e09d // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.11 // indirect
 	golang.org/x/xerrors v0.0.0-20220609144429-65e65417b02f // indirect
--- a/go.sum
+++ b/go.sum
@@ -17,6 +17,8 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
+github.com/alphadose/itogami v0.0.0-20220705100819-134f04183c42 h1:GLC/6Lfd6l1CRfkeUnR973fZBETOPQAtoaYGGl+Ho10=
+github.com/alphadose/itogami v0.0.0-20220705100819-134f04183c42/go.mod h1:QDsatlDSUJB4sXxZsJEpawGnTDwSdvX27ZXqtuZY3WA=
 github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
 github.com/apache/thrift v0.16.0 h1:qEy6UW60iVOlUy+b9ZR0d5WzUWYGOo4HfopoyBaNmoY=
 github.com/apache/thrift v0.16.0/go.mod h1:PHK3hniurgQaNMZYaCLEqXKsYK8upmhPbmdP2FXSqgU=
@@ -505,6 +507,8 @@ golang.org/x/net v0.0.0-20211216030914-fe4d6282115f/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220630215102-69896b714898 h1:K7wO6V1IrczY9QOQ2WkVpw4JQSwCd52UsxVEirZUfiw=
 golang.org/x/net v0.0.0-20220630215102-69896b714898/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.0.0-20220708220712-1185a9018129 h1:vucSRfWwTsoXro7P+3Cjlr6flUMtzCwzlvkxEQtHHB0=
+golang.org/x/net v0.0.0-20220708220712-1185a9018129/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181017192945-9dcd33a902f4/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181203162652-d668ce993890/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -564,6 +568,8 @@ golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220702020025-31831981b65f h1:xdsejrW/0Wf2diT5CPp3XmKUNbr7Xvw8kYilQ+6qjRY=
 golang.org/x/sys v0.0.0-20220702020025-31831981b65f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220708085239-5a0f0661e09d h1:/m5NbqQelATgoSPVC2Z23sR4kVNokFwDDyWh/3rGY+I=
+golang.org/x/sys v0.0.0-20220708085239-5a0f0661e09d/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
--- a/server/option.go
+++ b/server/option.go
@@ -3,6 +3,8 @@ package server
 import (
 	"crypto/tls"
 	"time"
+
+	"github.com/alphadose/itogami"
 )

 // OptionFn configures options of server.
@@ -37,3 +39,19 @@ func WithWriteTimeout(writeTimeout time.Duration) OptionFn {
 		s.writeTimeout = writeTimeout
 	}
 }
+
+// WithPool sets goroutine pool.
+func WithPool(n uint64) OptionFn {
+	return func(s *Server) {
+		if n > 0 {
+			s.pool = itogami.NewPool(n)
+		}
+	}
+}
+
+// WithAsyncWrite sets AsyncWrite to true.
+func WithAsyncWrite() OptionFn {
+	return func(s *Server) {
+		s.AsyncWrite = true
+	}
+}
--- a/server/server.go
+++ b/server/server.go
@@ -20,6 +20,7 @@ import (
 	"sync/atomic"
 	"time"

+	"github.com/alphadose/itogami"
 	"github.com/smallnest/rpcx/log"
 	"github.com/smallnest/rpcx/protocol"
 	"github.com/smallnest/rpcx/share"
@@ -78,6 +79,7 @@ type Server struct {
 	DisableHTTPGateway bool // should disable http invoke or not.
 	DisableJSONRPC     bool // should disable json rpc or not.
 	AsyncWrite         bool // set true if your server only serves few clients
+	pool               *itogami.Pool

 	serviceMapMu sync.RWMutex
 	serviceMap   map[string]*service
@@ -487,98 +489,105 @@ func (s *Server) serveConn(conn net.Conn) {
 			continue
 		}

-		// handle biz logics
-		go func() {
-			defer func() {
-				if r := recover(); r != nil {
-					log.Errorf("failed to handle the request: %v", r)
-				}
-			}()
+		if s.pool != nil {
+			s.pool.Submit(func() {
+				s.processOneRequest(ctx, req, conn, writeCh)
+			})
+		} else {
+			go s.processOneRequest(ctx, req, conn, writeCh)
+		}
+	}
+}

-			atomic.AddInt32(&s.handlerMsgNum, 1)
-			defer atomic.AddInt32(&s.handlerMsgNum, -1)
+func (s *Server) processOneRequest(ctx *share.Context, req *protocol.Message, conn net.Conn, writeCh chan *[]byte) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Errorf("failed to handle the request: %v", r)
+		}
+	}()

-			// 心跳请求，直接处理返回
-			if req.IsHeartbeat() {
-				s.Plugins.DoHeartbeatRequest(ctx, req)
-				req.SetMessageType(protocol.Response)
-				data := req.EncodeSlicePointer()
+	atomic.AddInt32(&s.handlerMsgNum, 1)
+	defer atomic.AddInt32(&s.handlerMsgNum, -1)

-				if s.writeTimeout != 0 {
-					conn.SetWriteDeadline(time.Now().Add(s.writeTimeout))
-				}
-				conn.Write(*data)
+	// 心跳请求，直接处理返回
+	if req.IsHeartbeat() {
+		s.Plugins.DoHeartbeatRequest(ctx, req)
+		req.SetMessageType(protocol.Response)
+		data := req.EncodeSlicePointer()

-				protocol.PutData(data)
-				protocol.FreeMsg(req)
+		if s.writeTimeout != 0 {
+			conn.SetWriteDeadline(time.Now().Add(s.writeTimeout))
+		}
+		conn.Write(*data)

-				return
-			}
+		protocol.PutData(data)
+		protocol.FreeMsg(req)

-			cancelFunc := parseServerTimeout(ctx, req)
-			if cancelFunc != nil {
-				defer cancelFunc()
-			}
+		return
+	}

-			resMetadata := make(map[string]string)
-			if req.Metadata == nil {
-				req.Metadata = make(map[string]string)
-			}
-			ctx = share.WithLocalValue(share.WithLocalValue(ctx, share.ReqMetaDataKey, req.Metadata),
-				share.ResMetaDataKey, resMetadata)
+	cancelFunc := parseServerTimeout(ctx, req)
+	if cancelFunc != nil {
+		defer cancelFunc()
+	}

-			s.Plugins.DoPreHandleRequest(ctx, req)
+	resMetadata := make(map[string]string)
+	if req.Metadata == nil {
+		req.Metadata = make(map[string]string)
+	}
+	ctx = share.WithLocalValue(share.WithLocalValue(ctx, share.ReqMetaDataKey, req.Metadata),
+		share.ResMetaDataKey, resMetadata)

-			if share.Trace {
-				log.Debugf("server handle request %+v from conn: %v", req, conn.RemoteAddr().String())
-			}
+	s.Plugins.DoPreHandleRequest(ctx, req)

-			// use handlers first
-			if handler, ok := s.router[req.ServicePath+"."+req.ServiceMethod]; ok {
-				sctx := NewContext(ctx, conn, req, writeCh)
-				err := handler(sctx)
-				if err != nil {
-					log.Errorf("[handler internal error]: servicepath: %s, servicemethod, err: %v", req.ServicePath, req.ServiceMethod, err)
-				}
+	if share.Trace {
+		log.Debugf("server handle request %+v from conn: %v", req, conn.RemoteAddr().String())
+	}

-				protocol.FreeMsg(req)
-				return
-			}
+	// use handlers first
+	if handler, ok := s.router[req.ServicePath+"."+req.ServiceMethod]; ok {
+		sctx := NewContext(ctx, conn, req, writeCh)
+		err := handler(sctx)
+		if err != nil {
+			log.Errorf("[handler internal error]: servicepath: %s, servicemethod, err: %v", req.ServicePath, req.ServiceMethod, err)
+		}

-			res, err := s.handleRequest(ctx, req)
-			if err != nil {
-				if s.HandleServiceError != nil {
-					s.HandleServiceError(err)
-				} else {
-					log.Warnf("rpcx: failed to handle request: %v", err)
-				}
-			}
+		protocol.FreeMsg(req)
+		return
+	}

-			if !req.IsOneway() {
-				if len(resMetadata) > 0 { // copy meta in context to responses
-					meta := res.Metadata
-					if meta == nil {
-						res.Metadata = resMetadata
-					} else {
-						for k, v := range resMetadata {
-							if meta[k] == "" {
-								meta[k] = v
-							}
-						}
+	res, err := s.handleRequest(ctx, req)
+	if err != nil {
+		if s.HandleServiceError != nil {
+			s.HandleServiceError(err)
+		} else {
+			log.Warnf("rpcx: failed to handle request: %v", err)
+		}
+	}
+
+	if !req.IsOneway() {
+		if len(resMetadata) > 0 { // copy meta in context to responses
+			meta := res.Metadata
+			if meta == nil {
+				res.Metadata = resMetadata
+			} else {
+				for k, v := range resMetadata {
+					if meta[k] == "" {
+						meta[k] = v
 					}
 				}
-
-				s.sendResponse(ctx, conn, writeCh, err, req, res)
 			}
+		}

-			if share.Trace {
-				log.Debugf("server write response %+v for an request %+v from conn: %v", res, req, conn.RemoteAddr().String())
-			}
-
-			protocol.FreeMsg(req)
-			protocol.FreeMsg(res)
-		}()
+		s.sendResponse(ctx, conn, writeCh, err, req, res)
 	}
+
+	if share.Trace {
+		log.Debugf("server write response %+v for an request %+v from conn: %v", res, req, conn.RemoteAddr().String())
+	}
+
+	protocol.FreeMsg(req)
+	protocol.FreeMsg(res)
 }

 func (s *Server) serveAsyncWrite(conn net.Conn, writeCh chan *[]byte) {