Mod: use task pool of getty

config/testdata/consumer_config.yml

@@ -56,6 +56,9 @@ protocol_conf:
     session_timeout: "20s"
     pool_size: 64
     pool_ttl: 600
+    gr_pool_size: 100
+    queue_len: 0
+    queue_number: 0
     getty_session_param:
       compress_encoding: false
       tcp_no_delay: true
@@ -63,7 +66,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

config/testdata/consumer_config_with_configcenter.yml

@@ -30,7 +30,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

config/testdata/provider_config.yml

@@ -56,6 +56,9 @@ protocol_conf:
   dubbo:
     session_number: 700
     session_timeout: "20s"
+    gr_pool_size: 100
+    queue_len: 0
+    queue_number: 0
     getty_session_param:
       compress_encoding: false
       tcp_no_delay: true
@@ -63,7 +66,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/go-client/profiles/dev/client.yml

@@ -75,7 +75,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/go-client/profiles/release/client.yml

@@ -75,7 +75,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/go-client/profiles/test/client.yml

@@ -75,7 +75,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/go-server/profiles/dev/server.yml

@@ -84,7 +84,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/go-server/profiles/release/server.yml

@@ -83,7 +83,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/go-server/profiles/test/server.yml

@@ -83,7 +83,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/with-configcenter-go-client/profiles/dev/client.yml

@@ -30,7 +30,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/with-configcenter-go-client/profiles/release/client.yml

@@ -30,7 +30,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/with-configcenter-go-client/profiles/test/client.yml

@@ -30,7 +30,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/with-configcenter-go-server/profiles/dev/server.yml

@@ -31,7 +31,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/with-configcenter-go-server/profiles/release/server.yml

@@ -31,7 +31,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

examples/dubbo/with-configcenter-go-server/profiles/test/server.yml

@@ -31,7 +31,6 @@ protocol_conf:
       keep_alive_period: "120s"
       tcp_r_buf_size: 262144
       tcp_w_buf_size: 65536
-      pkg_rq_size: 1024
       pkg_wq_size: 512
       tcp_read_timeout: "1s"
       tcp_write_timeout: "5s"

go.mod

 module github.com/apache/dubbo-go
 
 require (
-	github.com/dubbogo/getty v1.2.0
+	github.com/dubbogo/getty v1.2.2
 	github.com/dubbogo/gost v1.1.1
 	github.com/dubbogo/hessian2 v1.2.3
 	github.com/magiconair/properties v1.8.1

go.sum

 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dubbogo/getty v1.2.0 h1:LmP/R/yCYp5/ubUTm5Xa7QwE3EnFFSjguJH8DJruJ70=
-github.com/dubbogo/getty v1.2.0/go.mod h1:p9Pyk18a+5Aa0GZ546aCzlehEfvEof0jAF0+QQcEPYQ=
-github.com/dubbogo/gost v1.0.1-0.20190706005735-65c3ecbba418/go.mod h1:R7wZm1DrmrKGr50mBZVcg6C9ekG8aL5hP+sgWcIDwQg=
+github.com/dubbogo/getty v1.2.2 h1:qDC9WXjxcs5NPvWZz2ruVKBKr2r1Jjm6i0Sq//CQwbE=
+github.com/dubbogo/getty v1.2.2/go.mod h1:K4b3MkGLf7T+lMgQNFgpg0dI1Wvv1PTisFs1Psf86kU=
 github.com/dubbogo/gost v1.1.1 h1:JCM7vx5edPIjDA5ovJTuzEEXuw2t7xLyrlgi2mi5jHI=
 github.com/dubbogo/gost v1.1.1/go.mod h1:R7wZm1DrmrKGr50mBZVcg6C9ekG8aL5hP+sgWcIDwQg=
 github.com/dubbogo/hessian2 v1.2.3 h1:2ijKLGT4v/Oe/UdZzXE2cippARuCidY+02sxVfVgphU=

protocol/dubbo/client.go

@@ -25,6 +25,7 @@ import (
 
 import (
 	"github.com/dubbogo/getty"
+	"github.com/dubbogo/gost/sync"
 	"github.com/dubbogo/hessian2"
 	perrors "github.com/pkg/errors"
 	"go.uber.org/atomic"
@@ -45,7 +46,8 @@ var (
 	errClientClosed      = perrors.New("client closed")
 	errClientReadTimeout = perrors.New("client read timeout")
 
-	clientConf *ClientConfig
+	clientConf   *ClientConfig
+	clientGrpool *gxsync.TaskPool
 )
 
 func init() {
@@ -78,6 +80,7 @@ func init() {
 	}
 
 	clientConf = conf
+	setClientGrpool()
 }
 
 func SetClientConf(c ClientConfig) {
@@ -87,12 +90,20 @@ func SetClientConf(c ClientConfig) {
 		logger.Warnf("[ClientConfig CheckValidity] error: %v", err)
 		return
 	}
+	setClientGrpool()
 }
 
 func GetClientConf() ClientConfig {
 	return *clientConf
 }
 
+func setClientGrpool() {
+	if clientConf.GrPoolSize > 1 {
+		clientGrpool = gxsync.NewTaskPool(gxsync.WithTaskPoolTaskPoolSize(clientConf.GrPoolSize), gxsync.WithTaskPoolTaskQueueLength(clientConf.QueueLen),
+			gxsync.WithTaskPoolTaskQueueNumber(clientConf.QueueNumber))
+	}
+}
+
 type Options struct {
 	// connect timeout
 	ConnectTimeout time.Duration

protocol/dubbo/client_test.go

@@ -173,7 +173,6 @@ func InitTest(t *testing.T) (protocol.Protocol, common.URL) {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgRQSize:        1024,
 			PkgWQSize:        512,
 			TcpReadTimeout:   "4s",
 			TcpWriteTimeout:  "5s",
@@ -193,7 +192,6 @@ func InitTest(t *testing.T) (protocol.Protocol, common.URL) {
 			KeepAlivePeriod:  "120s",
 			TcpRBufSize:      262144,
 			TcpWBufSize:      65536,
-			PkgRQSize:        1024,
 			PkgWQSize:        512,
 			TcpReadTimeout:   "1s",
 			TcpWriteTimeout:  "5s",

protocol/dubbo/config.go

@@ -34,7 +34,6 @@ type (
 		keepAlivePeriod  time.Duration
 		TcpRBufSize      int    `default:"262144" yaml:"tcp_r_buf_size" json:"tcp_r_buf_size,omitempty"`
 		TcpWBufSize      int    `default:"65536" yaml:"tcp_w_buf_size" json:"tcp_w_buf_size,omitempty"`
-		PkgRQSize        int    `default:"1024" yaml:"pkg_rq_size" json:"pkg_rq_size,omitempty"`
 		PkgWQSize        int    `default:"1024" yaml:"pkg_wq_size" json:"pkg_wq_size,omitempty"`
 		TcpReadTimeout   string `default:"1s" yaml:"tcp_read_timeout" json:"tcp_read_timeout,omitempty"`
 		tcpReadTimeout   time.Duration
@@ -53,6 +52,11 @@ type (
 		sessionTimeout time.Duration
 		SessionNumber  int `default:"1000" yaml:"session_number" json:"session_number,omitempty"`
 
+		// grpool
+		GrPoolSize  int `default:"0" yaml:"gr_pool_size" json:"gr_pool_size,omitempty"`
+		QueueLen    int `default:"0" yaml:"queue_len" json:"queue_len,omitempty"`
+		QueueNumber int `default:"0" yaml:"queue_number" json:"queue_number,omitempty"`
+
 		// session tcp parameters
 		GettySessionParam GettySessionParam `required:"true" yaml:"getty_session_param" json:"getty_session_param,omitempty"`
 	}
@@ -76,6 +80,11 @@ type (
 		PoolSize int `default:"2" yaml:"pool_size" json:"pool_size,omitempty"`
 		PoolTTL  int `default:"180" yaml:"pool_ttl" json:"pool_ttl,omitempty"`
 
+		// grpool
+		GrPoolSize  int `default:"0" yaml:"gr_pool_size" json:"gr_pool_size,omitempty"`
+		QueueLen    int `default:"0" yaml:"queue_len" json:"queue_len,omitempty"`
+		QueueNumber int `default:"0" yaml:"queue_number" json:"queue_number,omitempty"`
+
 		// session tcp parameters
 		GettySessionParam GettySessionParam `required:"true" yaml:"getty_session_param" json:"getty_session_param,omitempty"`
 	}