Merge branch '1.5' of https://github.com/apache/dubbo-go into 1.5

protocol/dubbo/dubbo_invoker_test.go

@@ -106,8 +106,6 @@ func InitTest(t *testing.T) (protocol.Protocol, *common.URL) {
 		ConnectionNum:   2,
 		HeartbeatPeriod: "5s",
 		SessionTimeout:  "20s",
-		PoolTTL:         600,
-		PoolSize:        64,
 		GettySessionParam: getty.GettySessionParam{
 			CompressEncoding: false,
 			TcpNoDelay:       true,

protocol/dubbo/dubbo_protocol_test.go

@@ -63,8 +63,6 @@ func initDubboInvokerTest() {
 		ConnectionNum:   1,
 		HeartbeatPeriod: "3s",
 		SessionTimeout:  "20s",
-		PoolTTL:         600,
-		PoolSize:        64,
 		GettySessionParam: getty.GettySessionParam{
 			CompressEncoding: false,
 			TcpNoDelay:       true,

remoting/exchange_client.go

@@ -66,7 +66,7 @@ func NewExchangeClient(url *common.URL, client Client, connectTimeout time.Durat
 		address:        url.Location,
 		client:         client,
 	}
-	client.SetExchangeClient(exchangeClient)
+
 	if !lazyInit {
 		if err := exchangeClient.doInit(url); err != nil {
 			return nil

remoting/getty/config.go

@@ -95,10 +95,6 @@ type (
 		SessionTimeout string `default:"60s" yaml:"session_timeout" json:"session_timeout,omitempty"`
 		sessionTimeout time.Duration
 
-		// Connection Pool
-		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"`
@@ -116,8 +112,6 @@ func GetDefaultClientConfig() ClientConfig {
 		ConnectionNum:     16,
 		HeartbeatPeriod:   "30s",
 		SessionTimeout:    "180s",
-		PoolSize:          4,
-		PoolTTL:           600,
 		GrPoolSize:        200,
 		QueueLen:          64,
 		QueueNumber:       10,

remoting/getty/getty_client.go

@@ -18,8 +18,12 @@
 package getty
 
 import (
+	"crypto/tls"
+	"fmt"
 	"math/rand"
+	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -27,6 +31,7 @@ import (
 	"github.com/apache/dubbo-getty"
 	gxsync "github.com/dubbogo/gost/sync"
 	perrors "github.com/pkg/errors"
+	uatomic "go.uber.org/atomic"
 	"gopkg.in/yaml.v2"
 )
 
@@ -84,7 +89,6 @@ func initClient(protocol string) {
 		return
 	}
 	setClientGrpool()
-
 	rand.Seed(time.Now().UnixNano())
 }
 
@@ -114,13 +118,16 @@ type Options struct {
 
 // Client : some configuration for network communication.
 type Client struct {
-	addr           string
-	opts           Options
-	conf           ClientConfig
-	mux            sync.RWMutex
-	pool           *gettyRPCClientPool
-	codec          remoting.Codec
-	ExchangeClient *remoting.ExchangeClient
+	addr               string
+	opts               Options
+	conf               ClientConfig
+	mux                sync.RWMutex
+	sslEnabled         bool
+	clientClosed       bool
+	gettyClient        *gettyRPCClient
+	gettyClientMux     sync.RWMutex
+	gettyClientCreated uatomic.Bool
+	codec              remoting.Codec
 }
 
 // create client
@@ -134,22 +141,21 @@ func NewClient(opt Options) *Client {
 	}
 
 	c := &Client{
-		opts: opt,
+		opts:         opt,
+		clientClosed: false,
 	}
+	c.gettyClientCreated.Store(false)
 	return c
 }
 
 func (c *Client) SetExchangeClient(client *remoting.ExchangeClient) {
-	c.ExchangeClient = client
 }
 
 // init client and try to connection.
 func (c *Client) Connect(url *common.URL) error {
 	initClient(url.Protocol)
 	c.conf = *clientConf
-	// new client
-	c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))
-	c.pool.sslEnabled = url.GetParamBool(constant.SSL_ENABLED_KEY, false)
+	c.sslEnabled = url.GetParamBool(constant.SSL_ENABLED_KEY, false)
 
 	// codec
 	c.codec = remoting.GetCodec(url.Protocol)
@@ -164,11 +170,12 @@ func (c *Client) Connect(url *common.URL) error {
 // close network connection
 func (c *Client) Close() {
 	c.mux.Lock()
-	p := c.pool
-	c.pool = nil
+	client := c.gettyClient
+	c.gettyClient = nil
+	c.clientClosed = true
 	c.mux.Unlock()
-	if p != nil {
-		p.close()
+	if client != nil {
+		client.close()
 	}
 }
 
@@ -218,17 +225,325 @@ func (c *Client) IsAvailable() bool {
 func (c *Client) selectSession(addr string) (*gettyRPCClient, getty.Session, error) {
 	c.mux.RLock()
 	defer c.mux.RUnlock()
-	if c.pool == nil {
-		return nil, nil, perrors.New("client pool have been closed")
+	if c.clientClosed {
+		return nil, nil, perrors.New("client have been closed")
 	}
-	rpcClient, err := c.pool.getGettyRpcClient(addr)
-	if err != nil {
-		return nil, nil, perrors.WithStack(err)
+	if !c.gettyClientCreated.Load() {
+		c.gettyClientMux.Lock()
+		if c.gettyClient == nil {
+			rpcClientConn, rpcErr := newGettyRPCClientConn(c, addr)
+			if rpcErr != nil {
+				c.gettyClientMux.Unlock()
+				return nil, nil, perrors.WithStack(rpcErr)
+			}
+			c.gettyClientCreated.Store(true)
+			c.gettyClient = rpcClientConn
+		}
+		client := c.gettyClient
+		session := c.gettyClient.selectSession()
+		c.gettyClientMux.Unlock()
+		return client, session, nil
 	}
-	return rpcClient, rpcClient.selectSession(), nil
+	c.gettyClientMux.RLock()
+	client := c.gettyClient
+	session := c.gettyClient.selectSession()
+	c.gettyClientMux.RUnlock()
+	return client, session, nil
 }
 
 func (c *Client) transfer(session getty.Session, request *remoting.Request, timeout time.Duration) (int, int, error) {
 	totalLen, sendLen, err := session.WritePkg(request, timeout)
 	return totalLen, sendLen, perrors.WithStack(err)
 }
+
+func (c *Client) resetRpcConn() {
+	c.gettyClientMux.Lock()
+	c.gettyClient = nil
+	c.gettyClientCreated.Store(false)
+	c.gettyClientMux.Unlock()
+
+}
+
+type gettyRPCClient struct {
+	once sync.Once
+	//protocol string
+	addr   string
+	active int64 // zero, not create or be destroyed
+
+	rpcClient *Client
+
+	lock        sync.RWMutex
+	gettyClient getty.Client
+	sessions    []*rpcSession
+}
+
+func newGettyRPCClientConn(rpcClient *Client, addr string) (*gettyRPCClient, error) {
+	var (
+		gettyClient getty.Client
+		sslEnabled  bool
+	)
+	sslEnabled = rpcClient.sslEnabled
+	clientOpts := []getty.ClientOption{
+		getty.WithServerAddress(addr),
+		getty.WithConnectionNumber((int)(rpcClient.conf.ConnectionNum)),
+		getty.WithReconnectInterval(rpcClient.conf.ReconnectInterval),
+	}
+	if sslEnabled {
+		clientOpts = append(clientOpts, getty.WithClientSslEnabled(sslEnabled), getty.WithClientTlsConfigBuilder(config.GetClientTlsConfigBuilder()))
+	}
+
+	if clientGrpool != nil {
+		clientOpts = append(clientOpts, getty.WithClientTaskPool(clientGrpool))
+	}
+
+	gettyClient = getty.NewTCPClient(clientOpts...)
+	c := &gettyRPCClient{
+		addr:        addr,
+		rpcClient:   rpcClient,
+		gettyClient: gettyClient,
+	}
+	go c.gettyClient.RunEventLoop(c.newSession)
+
+	idx := 1
+	start := time.Now()
+	connectTimeout := rpcClient.opts.ConnectTimeout
+	for {
+		idx++
+		if c.isAvailable() {
+			break
+		}
+
+		if time.Since(start) > connectTimeout {
+			c.gettyClient.Close()
+			return nil, perrors.New(fmt.Sprintf("failed to create client connection to %s in %s", addr, connectTimeout))
+		}
+
+		interval := time.Millisecond * time.Duration(idx)
+		if interval > time.Duration(100e6) {
+			interval = 100e6 // 100 ms
+		}
+		time.Sleep(interval)
+	}
+	logger.Debug("client init ok")
+	c.updateActive(time.Now().Unix())
+
+	return c, nil
+}
+
+func (c *gettyRPCClient) updateActive(active int64) {
+	atomic.StoreInt64(&c.active, active)
+}
+
+func (c *gettyRPCClient) getActive() int64 {
+	return atomic.LoadInt64(&c.active)
+}
+
+func (c *gettyRPCClient) newSession(session getty.Session) error {
+	var (
+		ok         bool
+		tcpConn    *net.TCPConn
+		conf       ClientConfig
+		sslEnabled bool
+	)
+	conf = c.rpcClient.conf
+	sslEnabled = c.rpcClient.sslEnabled
+	if conf.GettySessionParam.CompressEncoding {
+		session.SetCompressType(getty.CompressZip)
+	}
+	if sslEnabled {
+		if _, ok = session.Conn().(*tls.Conn); !ok {
+			panic(fmt.Sprintf("%s, session.conn{%#v} is not tls connection\n", session.Stat(), session.Conn()))
+		}
+		session.SetName(conf.GettySessionParam.SessionName)
+		session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
+		session.SetPkgHandler(NewRpcClientPackageHandler(c.rpcClient))
+		session.SetEventListener(NewRpcClientHandler(c))
+		session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
+		session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
+		session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
+		session.SetWaitTime(conf.GettySessionParam.waitTimeout)
+		logger.Debugf("client new session:%s\n", session.Stat())
+		return nil
+	}
+	if tcpConn, ok = session.Conn().(*net.TCPConn); !ok {
+		panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
+	}
+
+	if err := tcpConn.SetNoDelay(conf.GettySessionParam.TcpNoDelay); err != nil {
+		logger.Error("tcpConn.SetNoDelay() = error:%v", err)
+	}
+	if err := tcpConn.SetKeepAlive(conf.GettySessionParam.TcpKeepAlive); err != nil {
+		logger.Error("tcpConn.SetKeepAlive() = error:%v", err)
+	}
+	if conf.GettySessionParam.TcpKeepAlive {
+		if err := tcpConn.SetKeepAlivePeriod(conf.GettySessionParam.keepAlivePeriod); err != nil {
+			logger.Error("tcpConn.SetKeepAlivePeriod() = error:%v", err)
+		}
+	}
+	if err := tcpConn.SetReadBuffer(conf.GettySessionParam.TcpRBufSize); err != nil {
+		logger.Error("tcpConn.SetReadBuffer() = error:%v", err)
+	}
+	if err := tcpConn.SetWriteBuffer(conf.GettySessionParam.TcpWBufSize); err != nil {
+		logger.Error("tcpConn.SetWriteBuffer() = error:%v", err)
+	}
+
+	session.SetName(conf.GettySessionParam.SessionName)
+	session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
+	session.SetPkgHandler(NewRpcClientPackageHandler(c.rpcClient))
+	session.SetEventListener(NewRpcClientHandler(c))
+	session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
+	session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
+	session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
+	session.SetWaitTime(conf.GettySessionParam.waitTimeout)
+	logger.Debugf("client new session:%s\n", session.Stat())
+	return nil
+}
+
+func (c *gettyRPCClient) selectSession() getty.Session {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	if c.sessions == nil {
+		return nil
+	}
+
+	count := len(c.sessions)
+	if count == 0 {
+		return nil
+	}
+	return c.sessions[rand.Int31n(int32(count))].session
+}
+
+func (c *gettyRPCClient) addSession(session getty.Session) {
+	logger.Debugf("add session{%s}", session.Stat())
+	if session == nil {
+		return
+	}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if c.sessions == nil {
+		c.sessions = make([]*rpcSession, 0, 16)
+	}
+	c.sessions = append(c.sessions, &rpcSession{session: session})
+}
+
+func (c *gettyRPCClient) removeSession(session getty.Session) {
+	if session == nil {
+		return
+	}
+
+	var removeFlag bool
+	func() {
+		c.lock.Lock()
+		defer c.lock.Unlock()
+		if c.sessions == nil {
+			return
+		}
+
+		for i, s := range c.sessions {
+			if s.session == session {
+				c.sessions = append(c.sessions[:i], c.sessions[i+1:]...)
+				logger.Debugf("delete session{%s}, its index{%d}", session.Stat(), i)
+				break
+			}
+		}
+		logger.Infof("after remove session{%s}, left session number:%d", session.Stat(), len(c.sessions))
+		if len(c.sessions) == 0 {
+			removeFlag = true
+		}
+	}()
+	if removeFlag {
+		c.rpcClient.resetRpcConn()
+		c.close()
+	}
+}
+
+func (c *gettyRPCClient) updateSession(session getty.Session) {
+	if session == nil {
+		return
+	}
+
+	var rs *rpcSession
+	func() {
+		c.lock.RLock()
+		defer c.lock.RUnlock()
+		if c.sessions == nil {
+			return
+		}
+
+		for i, s := range c.sessions {
+			if s.session == session {
+				rs = c.sessions[i]
+				break
+			}
+		}
+	}()
+	if rs != nil {
+		rs.AddReqNum(1)
+	}
+}
+
+func (c *gettyRPCClient) getClientRpcSession(session getty.Session) (rpcSession, error) {
+	var (
+		err error
+		rs  rpcSession
+	)
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+	if c.sessions == nil {
+		return rs, errClientClosed
+	}
+
+	err = errSessionNotExist
+	for _, s := range c.sessions {
+		if s.session == session {
+			rs = *s
+			err = nil
+			break
+		}
+	}
+
+	return rs, perrors.WithStack(err)
+}
+
+func (c *gettyRPCClient) isAvailable() bool {
+	return c.selectSession() != nil
+}
+
+func (c *gettyRPCClient) close() error {
+	closeErr := perrors.Errorf("close gettyRPCClient{%#v} again", c)
+	c.once.Do(func() {
+		var (
+			gettyClient getty.Client
+			sessions    []*rpcSession
+		)
+		func() {
+			c.lock.Lock()
+			defer c.lock.Unlock()
+
+			gettyClient = c.gettyClient
+			c.gettyClient = nil
+
+			sessions = make([]*rpcSession, 0, len(c.sessions))
+			sessions = append(sessions, c.sessions...)
+			c.sessions = c.sessions[:0]
+		}()
+
+		c.updateActive(0)
+
+		go func() {
+			if gettyClient != nil {
+				gettyClient.Close()
+			}
+			for _, s := range sessions {
+				logger.Infof("close client session{%s, last active:%s, request number:%d}",
+					s.session.Stat(), s.session.GetActive().String(), s.GetReqNum())
+				s.session.Close()
+			}
+		}()
+
+		closeErr = nil
+	})
+	return closeErr
+}

remoting/getty/getty_client_test.go

@@ -45,24 +45,23 @@ import (
 func TestRunSuite(t *testing.T) {
 	svr, url := InitTest(t)
 	client := getClient(url)
-	testRequestOneWay(t, svr, url, client)
-	testClient_Call(t, svr, url, client)
-	testClient_AsyncCall(t, svr, url, client)
+	assert.NotNil(t, client)
+	testRequestOneWay(t, client)
+	testClient_Call(t, client)
+	testClient_AsyncCall(t, client)
 	svr.Stop()
 }
 
-func testRequestOneWay(t *testing.T, svr *Server, url *common.URL, client *Client) {
+func testRequestOneWay(t *testing.T, client *Client) {
 
 	request := remoting.NewRequest("2.0.2")
-	up := &UserProvider{}
-	invocation := createInvocation("GetUser", nil, nil, []interface{}{[]interface{}{"1", "username"}, up},
-		[]reflect.Value{reflect.ValueOf([]interface{}{"1", "username"}), reflect.ValueOf(up)})
+	invocation := createInvocation("GetUser", nil, nil, []interface{}{"1", "username"},
+		[]reflect.Value{reflect.ValueOf("1"), reflect.ValueOf("username")})
 	attachment := map[string]string{INTERFACE_KEY: "com.ikurento.user.UserProvider"}
 	setAttachment(invocation, attachment)
 	request.Data = invocation
 	request.Event = false
 	request.TwoWay = false
-	//user := &User{}
 	err := client.Request(request, 3*time.Second, nil)
 	assert.NoError(t, err)
 }
@@ -85,16 +84,13 @@ func getClient(url *common.URL) *Client {
 		ConnectTimeout: config.GetConsumerConfig().ConnectTimeout,
 	})
 
-	exchangeClient := remoting.NewExchangeClient(url, client, 5*time.Second, false)
-	client.SetExchangeClient(exchangeClient)
 	if err := client.Connect(url); err != nil {
 		return nil
 	}
 	return client
 }
 
-func testClient_Call(t *testing.T, svr *Server, url *common.URL, c *Client) {
-	c.pool = newGettyRPCClientConnPool(c, clientConf.PoolSize, time.Duration(int(time.Second)*clientConf.PoolTTL))
+func testClient_Call(t *testing.T, c *Client) {
 
 	testGetBigPkg(t, c)
 	testGetUser(t, c)
@@ -312,7 +308,7 @@ func testGetUser61(t *testing.T, c *Client) {
 	assert.Equal(t, User{Id: "1", Name: ""}, *user)
 }
 
-func testClient_AsyncCall(t *testing.T, svr *Server, url *common.URL, client *Client) {
+func testClient_AsyncCall(t *testing.T, client *Client) {
 	user := &User{}
 	lock := sync.Mutex{}
 	request := remoting.NewRequest("2.0.2")
@@ -354,8 +350,6 @@ func InitTest(t *testing.T) (*Server, *common.URL) {
 		ConnectionNum:   2,
 		HeartbeatPeriod: "5s",
 		SessionTimeout:  "20s",
-		PoolTTL:         600,
-		PoolSize:        64,
 		GettySessionParam: GettySessionParam{
 			CompressEncoding: false,
 			TcpNoDelay:       true,

remoting/getty/listener.go

@@ -144,7 +144,7 @@ func (h *RpcClientHandler) OnCron(session getty.Session) {
 			session.Stat(), perrors.WithStack(err))
 		return
 	}
-	if h.conn.pool.rpcClient.conf.sessionTimeout.Nanoseconds() < time.Since(session.GetActive()).Nanoseconds() {
+	if h.conn.rpcClient.conf.sessionTimeout.Nanoseconds() < time.Since(session.GetActive()).Nanoseconds() {
 		logger.Warnf("session{%s} timeout{%s}, reqNum{%d}",
 			session.Stat(), time.Since(session.GetActive()).String(), rs.reqNum)
 		h.conn.removeSession(session) // -> h.conn.close() -> h.conn.pool.remove(h.conn)
@@ -164,7 +164,7 @@ func (h *RpcClientHandler) OnCron(session getty.Session) {
 		h.timeoutTimes = 0
 	}
 
-	if err := heartbeat(session, h.conn.pool.rpcClient.conf.heartbeatTimeout, heartbeatCallBack); err != nil {
+	if err := heartbeat(session, h.conn.rpcClient.conf.heartbeatTimeout, heartbeatCallBack); err != nil {
 		logger.Warnf("failed to send heartbeat, error{%v}", err)
 	}
 }

remoting/getty/pool.go

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package getty
-
-import (
-	"crypto/tls"
-	"fmt"
-	"math/rand"
-	"net"
-	"sync"
-	"sync/atomic"
-	"time"
-)
-
-import (
-	"github.com/apache/dubbo-getty"
-	perrors "github.com/pkg/errors"
-)
-
-import (
-	"github.com/apache/dubbo-go/common/logger"
-	"github.com/apache/dubbo-go/config"
-)
-
-type gettyRPCClient struct {
-	once sync.Once
-	//protocol string
-	addr   string
-	active int64 // zero, not create or be destroyed
-
-	pool *gettyRPCClientPool
-
-	lock        sync.RWMutex
-	gettyClient getty.Client
-	sessions    []*rpcSession
-}
-
-var (
-	errClientPoolClosed = perrors.New("client pool closed")
-)
-
-func newGettyRPCClientConn(pool *gettyRPCClientPool, addr string) (*gettyRPCClient, error) {
-	var (
-		gettyClient getty.Client
-		sslEnabled  bool
-	)
-	sslEnabled = pool.sslEnabled
-	clientOpts := []getty.ClientOption{
-		getty.WithServerAddress(addr),
-		getty.WithConnectionNumber((int)(pool.rpcClient.conf.ConnectionNum)),
-		getty.WithReconnectInterval(pool.rpcClient.conf.ReconnectInterval),
-	}
-	if sslEnabled {
-		clientOpts = append(clientOpts, getty.WithClientSslEnabled(pool.sslEnabled), getty.WithClientTlsConfigBuilder(config.GetClientTlsConfigBuilder()))
-	}
-
-	if clientGrpool != nil {
-		clientOpts = append(clientOpts, getty.WithClientTaskPool(clientGrpool))
-	}
-
-	gettyClient = getty.NewTCPClient(clientOpts...)
-	c := &gettyRPCClient{
-		addr:        addr,
-		pool:        pool,
-		gettyClient: gettyClient,
-	}
-	go c.gettyClient.RunEventLoop(c.newSession)
-
-	idx := 1
-	start := time.Now()
-	connectTimeout := pool.rpcClient.opts.ConnectTimeout
-	for {
-		idx++
-		if c.isAvailable() {
-			break
-		}
-
-		if time.Since(start) > connectTimeout {
-			c.gettyClient.Close()
-			return nil, perrors.New(fmt.Sprintf("failed to create client connection to %s in %s", addr, connectTimeout))
-		}
-
-		interval := time.Millisecond * time.Duration(idx)
-		if interval > time.Duration(100e6) {
-			interval = 100e6 // 100 ms
-		}
-		time.Sleep(interval)
-	}
-	logger.Debug("client init ok")
-	c.updateActive(time.Now().Unix())
-
-	return c, nil
-}
-
-func (c *gettyRPCClient) updateActive(active int64) {
-	atomic.StoreInt64(&c.active, active)
-}
-
-func (c *gettyRPCClient) getActive() int64 {
-	return atomic.LoadInt64(&c.active)
-}
-
-func (c *gettyRPCClient) newSession(session getty.Session) error {
-	var (
-		ok         bool
-		tcpConn    *net.TCPConn
-		conf       ClientConfig
-		sslEnabled bool
-	)
-	conf = c.pool.rpcClient.conf
-	sslEnabled = c.pool.sslEnabled
-	if conf.GettySessionParam.CompressEncoding {
-		session.SetCompressType(getty.CompressZip)
-	}
-	if sslEnabled {
-		if _, ok = session.Conn().(*tls.Conn); !ok {
-			panic(fmt.Sprintf("%s, session.conn{%#v} is not tls connection\n", session.Stat(), session.Conn()))
-		}
-		session.SetName(conf.GettySessionParam.SessionName)
-		session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
-		session.SetPkgHandler(NewRpcClientPackageHandler(c.pool.rpcClient))
-		session.SetEventListener(NewRpcClientHandler(c))
-		session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
-		session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
-		session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
-		session.SetWaitTime(conf.GettySessionParam.waitTimeout)
-		logger.Debugf("client new session:%s\n", session.Stat())
-		return nil
-	}
-	if tcpConn, ok = session.Conn().(*net.TCPConn); !ok {
-		panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
-	}
-
-	if err := tcpConn.SetNoDelay(conf.GettySessionParam.TcpNoDelay); err != nil {
-		logger.Error("tcpConn.SetNoDelay() = error:%v", err)
-	}
-	if err := tcpConn.SetKeepAlive(conf.GettySessionParam.TcpKeepAlive); err != nil {
-		logger.Error("tcpConn.SetKeepAlive() = error:%v", err)
-	}
-	if conf.GettySessionParam.TcpKeepAlive {
-		if err := tcpConn.SetKeepAlivePeriod(conf.GettySessionParam.keepAlivePeriod); err != nil {
-			logger.Error("tcpConn.SetKeepAlivePeriod() = error:%v", err)
-		}
-	}
-	if err := tcpConn.SetReadBuffer(conf.GettySessionParam.TcpRBufSize); err != nil {
-		logger.Error("tcpConn.SetReadBuffer() = error:%v", err)
-	}
-	if err := tcpConn.SetWriteBuffer(conf.GettySessionParam.TcpWBufSize); err != nil {
-		logger.Error("tcpConn.SetWriteBuffer() = error:%v", err)
-	}
-
-	session.SetName(conf.GettySessionParam.SessionName)
-	session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
-	session.SetPkgHandler(NewRpcClientPackageHandler(c.pool.rpcClient))
-	session.SetEventListener(NewRpcClientHandler(c))
-	session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
-	session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
-	session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
-	session.SetWaitTime(conf.GettySessionParam.waitTimeout)
-	logger.Debugf("client new session:%s\n", session.Stat())
-	return nil
-}
-
-func (c *gettyRPCClient) selectSession() getty.Session {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-
-	if c.sessions == nil {
-		return nil
-	}
-
-	count := len(c.sessions)
-	if count == 0 {
-		return nil
-	}
-	return c.sessions[rand.Int31n(int32(count))].session
-}
-
-func (c *gettyRPCClient) addSession(session getty.Session) {
-	logger.Debugf("add session{%s}", session.Stat())
-	if session == nil {
-		return
-	}
-
-	c.lock.Lock()
-	defer c.lock.Unlock()
-	if c.sessions == nil {
-		c.sessions = make([]*rpcSession, 0, 16)
-	}
-	c.sessions = append(c.sessions, &rpcSession{session: session})
-}
-
-func (c *gettyRPCClient) removeSession(session getty.Session) {
-	if session == nil {
-		return
-	}
-
-	var removeFlag bool
-	func() {
-		c.lock.Lock()
-		defer c.lock.Unlock()
-		if c.sessions == nil {
-			return
-		}
-
-		for i, s := range c.sessions {
-			if s.session == session {
-				c.sessions = append(c.sessions[:i], c.sessions[i+1:]...)
-				logger.Debugf("delete session{%s}, its index{%d}", session.Stat(), i)
-				break
-			}
-		}
-		logger.Infof("after remove session{%s}, left session number:%d", session.Stat(), len(c.sessions))
-		if len(c.sessions) == 0 {
-			removeFlag = true
-		}
-	}()
-	if removeFlag {
-		c.pool.safeRemove(c)
-		c.close()
-	}
-}
-
-func (c *gettyRPCClient) updateSession(session getty.Session) {
-	if session == nil {
-		return
-	}
-
-	var rs *rpcSession
-	func() {
-		c.lock.RLock()
-		defer c.lock.RUnlock()
-		if c.sessions == nil {
-			return
-		}
-
-		for i, s := range c.sessions {
-			if s.session == session {
-				rs = c.sessions[i]
-				break
-			}
-		}
-	}()
-	if rs != nil {
-		rs.AddReqNum(1)
-	}
-}
-
-func (c *gettyRPCClient) getClientRpcSession(session getty.Session) (rpcSession, error) {
-	var (
-		err error
-		rs  rpcSession
-	)
-	c.lock.RLock()
-	defer c.lock.RUnlock()
-	if c.sessions == nil {
-		return rs, errClientClosed
-	}
-
-	err = errSessionNotExist
-	for _, s := range c.sessions {
-		if s.session == session {
-			rs = *s
-			err = nil
-			break
-		}
-	}
-
-	return rs, perrors.WithStack(err)
-}
-
-func (c *gettyRPCClient) isAvailable() bool {
-	return c.selectSession() != nil
-}
-
-func (c *gettyRPCClient) close() error {
-	closeErr := perrors.Errorf("close gettyRPCClient{%#v} again", c)
-	c.once.Do(func() {
-		var (
-			gettyClient getty.Client
-			sessions    []*rpcSession
-		)
-		func() {
-			c.lock.Lock()
-			defer c.lock.Unlock()
-
-			gettyClient = c.gettyClient
-			c.gettyClient = nil
-
-			sessions = make([]*rpcSession, 0, len(c.sessions))
-			sessions = append(sessions, c.sessions...)
-			c.sessions = c.sessions[:0]
-		}()
-
-		c.updateActive(0)
-
-		go func() {
-			if gettyClient != nil {
-				gettyClient.Close()
-			}
-			for _, s := range sessions {
-				logger.Infof("close client session{%s, last active:%s, request number:%d}",
-					s.session.Stat(), s.session.GetActive().String(), s.GetReqNum())
-				s.session.Close()
-			}
-		}()
-
-		closeErr = nil
-	})
-	return closeErr
-}
-
-type gettyRPCClientPool struct {
-	rpcClient  *Client
-	size       int   // size of []*gettyRPCClient
-	ttl        int64 // ttl of every gettyRPCClient, it is checked when getConn
-	sslEnabled bool
-
-	sync.Mutex
-	conns []*gettyRPCClient
-}
-
-func newGettyRPCClientConnPool(rpcClient *Client, size int, ttl time.Duration) *gettyRPCClientPool {
-	return &gettyRPCClientPool{
-		rpcClient: rpcClient,
-		size:      size,
-		ttl:       int64(ttl.Seconds()),
-		// init capacity : 2
-		conns: make([]*gettyRPCClient, 0, 2),
-	}
-}
-
-func (p *gettyRPCClientPool) close() {
-	p.Lock()
-	conns := p.conns
-	p.conns = nil
-	p.Unlock()
-	for _, conn := range conns {
-		conn.close()
-	}
-}
-
-func (p *gettyRPCClientPool) getGettyRpcClient(addr string) (*gettyRPCClient, error) {
-	conn, connErr := p.get()
-	if connErr == nil && conn == nil {
-		// create new conn
-		rpcClientConn, rpcErr := newGettyRPCClientConn(p, addr)
-		if rpcErr == nil {
-			p.put(rpcClientConn)
-		}
-		return rpcClientConn, perrors.WithStack(rpcErr)
-	}
-	return conn, perrors.WithStack(connErr)
-}
-
-func (p *gettyRPCClientPool) get() (*gettyRPCClient, error) {
-	now := time.Now().Unix()
-
-	p.Lock()
-	defer p.Unlock()
-	if p.conns == nil {
-		return nil, errClientPoolClosed
-	}
-	for num := len(p.conns); num > 0; {
-		var conn *gettyRPCClient
-		if num != 1 {
-			conn = p.conns[rand.Int31n(int32(num))]
-		} else {
-			conn = p.conns[0]
-		}
-		// This will recreate gettyRpcClient for remove last position
-		//p.conns = p.conns[:len(p.conns)-1]
-
-		if d := now - conn.getActive(); d > p.ttl {
-			p.remove(conn)
-			go conn.close()
-			num = len(p.conns)
-			continue
-		}
-		conn.updateActive(now) //update active time
-		return conn, nil
-	}
-	return nil, nil
-}
-
-func (p *gettyRPCClientPool) put(conn *gettyRPCClient) {
-	if conn == nil || conn.getActive() == 0 {
-		return
-	}
-	p.Lock()
-	defer p.Unlock()
-	if p.conns == nil {
-		return
-	}
-	// check whether @conn has existed in p.conns or not.
-	for i := range p.conns {
-		if p.conns[i] == conn {
-			return
-		}
-	}
-	if len(p.conns) >= p.size {
-		// delete @conn from client pool
-		// p.remove(conn)
-		conn.close()
-		return
-	}
-	p.conns = append(p.conns, conn)
-}
-
-func (p *gettyRPCClientPool) remove(conn *gettyRPCClient) {
-	if conn == nil || conn.getActive() == 0 {
-		return
-	}
-
-	if p.conns == nil {
-		return
-	}
-
-	if len(p.conns) > 0 {
-		for idx, c := range p.conns {
-			if conn == c {
-				p.conns = append(p.conns[:idx], p.conns[idx+1:]...)
-				break
-			}
-		}
-	}
-}
-
-func (p *gettyRPCClientPool) safeRemove(conn *gettyRPCClient) {
-	p.Lock()
-	defer p.Unlock()
-
-	p.remove(conn)
-}

remoting/getty/pool_test.go

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package getty
-
-import (
-	"testing"
-	"time"
-)
-
-import (
-	"github.com/stretchr/testify/assert"
-)
-
-func TestGetConnFromPool(t *testing.T) {
-	var rpcClient Client
-
-	clientPoll := newGettyRPCClientConnPool(&rpcClient, 1, time.Duration(5*time.Second))
-
-	var conn1 gettyRPCClient
-	conn1.active = time.Now().Unix()
-	clientPoll.put(&conn1)
-	assert.Equal(t, 1, len(clientPoll.conns))
-
-	var conn2 gettyRPCClient
-	conn2.active = time.Now().Unix()
-	clientPoll.put(&conn2)
-	assert.Equal(t, 1, len(clientPoll.conns))
-	conn, err := clientPoll.get()
-	assert.Nil(t, err)
-	assert.Equal(t, &conn1, conn)
-	time.Sleep(6 * time.Second)
-	conn, err = clientPoll.get()
-	assert.Nil(t, conn)
-	assert.Nil(t, err)
-	assert.Equal(t, 0, len(clientPoll.conns))
-}

remoting/getty/readwriter_test.go

@@ -91,8 +91,6 @@ func getServer(t *testing.T) (*Server, *common.URL) {
 		ConnectionNum:   2,
 		HeartbeatPeriod: "5s",
 		SessionTimeout:  "20s",
-		PoolTTL:         600,
-		PoolSize:        64,
 		GettySessionParam: GettySessionParam{
 			CompressEncoding: false,
 			TcpNoDelay:       true,