Merge pull request #797 from wenxuwan/master
Imp: try to fix too many files open error
diff --git a/common/url.go b/common/url.go
index 7e6c719..e688450 100644
--- a/common/url.go
+++ b/common/url.go
@@ -377,7 +377,7 @@
if service != "" {
return service
} else if c.SubURL != nil {
- service = c.GetParam(constant.INTERFACE_KEY, strings.TrimPrefix(c.Path, "/"))
+ service = c.SubURL.GetParam(constant.INTERFACE_KEY, strings.TrimPrefix(c.Path, "/"))
if service != "" { //if url.path is "" then return suburl's path, special for registry url
return service
}
diff --git a/go.mod b/go.mod
index 3679b8c..bc7bc3c 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@
github.com/creasty/defaults v1.3.0
github.com/dubbogo/getty v1.3.8
github.com/dubbogo/go-zookeeper v1.0.0
- github.com/dubbogo/gost v1.9.0
+ github.com/dubbogo/gost v1.9.5
github.com/emicklei/go-restful/v3 v3.0.0
github.com/go-resty/resty/v2 v2.1.0
github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6 // indirect
@@ -29,7 +29,7 @@
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.1.0
github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b
- github.com/stretchr/testify v1.5.1
+ github.com/stretchr/testify v1.6.1
github.com/zouyx/agollo/v3 v3.4.4
go.etcd.io/bbolt v1.3.3 // indirect
go.etcd.io/etcd v3.3.13+incompatible
diff --git a/go.sum b/go.sum
index 12d5a7c..a75b445 100644
--- a/go.sum
+++ b/go.sum
@@ -132,6 +132,8 @@
github.com/dubbogo/go-zookeeper v1.0.0/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c=
github.com/dubbogo/gost v1.9.0 h1:UT+dWwvLyJiDotxJERO75jB3Yxgsdy10KztR5ycxRAk=
github.com/dubbogo/gost v1.9.0/go.mod h1:pPTjVyoJan3aPxBPNUX0ADkXjPibLo+/Ib0/fADXSG8=
+github.com/dubbogo/gost v1.9.5 h1:UeG4y0O55lR3dzgdmCm/7bMWvpKrlpR7fsfKjrcXq/g=
+github.com/dubbogo/gost v1.9.5/go.mod h1:QNM5RaeRdNWehUu8S0hUP5Qa8QUfGf6KH1JhqOVFvEI=
github.com/duosecurity/duo_api_golang v0.0.0-20190308151101-6c680f768e74 h1:2MIhn2R6oXQbgW5yHfS+d6YqyMfXiu2L55rFZC4UD/M=
github.com/duosecurity/duo_api_golang v0.0.0-20190308151101-6c680f768e74/go.mod h1:UqXY1lYT/ERa4OEAywUqdok1T4RCRdArkhic1Opuavo=
github.com/elazarl/go-bindata-assetfs v0.0.0-20160803192304-e1a2a7ec64b0 h1:ZoRgc53qJCfSLimXqJDrmBhnt5GChDsExMCK7t48o0Y=
@@ -358,6 +360,8 @@
github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
+github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88/go.mod h1:3w7q1U84EfirKl04SVQ/s7nPm1ZPhiXd34z40TNz36k=
+github.com/k0kubun/pp v3.0.1+incompatible/go.mod h1:GWse8YhT0p8pT4ir3ZgBbfZild3tgzSScAn6HmfYukg=
github.com/keybase/go-crypto v0.0.0-20180614160407-5114a9a81e1b h1:VE6r2OwP5gj+Z9aCkSKl3MlmnZbfMAjhvR5T7abKHEo=
github.com/keybase/go-crypto v0.0.0-20180614160407-5114a9a81e1b/go.mod h1:ghbZscTyKdM07+Fw3KSi0hcJm+AlEUWj8QLlPtijN/M=
github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
@@ -383,8 +387,10 @@
github.com/mailru/easyjson v0.0.0-20160728113105-d5b7844b561a/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
github.com/mattn/go-colorable v0.0.9 h1:UVL0vNpWh04HeJXV0KLcaT7r06gOH2l4OW6ddYRUIY4=
github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
+github.com/mattn/go-colorable v0.1.7/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
github.com/mattn/go-isatty v0.0.3 h1:ns/ykhmWi7G9O+8a448SecJU3nSMBXJfqQkl0upE1jI=
github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4=
+github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
github.com/miekg/dns v1.0.14 h1:9jZdLNd/P4+SfEJ0TNyxYpsK8N4GtfylBLqtbYN1sbA=
@@ -534,6 +540,7 @@
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/subosito/gotenv v1.2.0 h1:Slr1R9HxAlEKefgq5jn9U+DnETlIUa6HfgEzj0g5d7s=
github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
github.com/tebeka/strftime v0.1.3 h1:5HQXOqWKYRFfNyBMNVc9z5+QzuBtIXy03psIhtdJYto=
@@ -652,6 +659,8 @@
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3 h1:4y9KwBHBgBNwDbtu44R5o1fdOCQUEXhbk/P4A9WmJq0=
golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.0.0-20160726164857-2910a502d2bf/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -747,6 +756,7 @@
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
diff --git a/protocol/dubbo/client.go b/protocol/dubbo/client.go
index 5ec7db5..562f4ee 100644
--- a/protocol/dubbo/client.go
+++ b/protocol/dubbo/client.go
@@ -21,6 +21,7 @@
"math/rand"
"strings"
"sync"
+ "sync/atomic"
"time"
)
@@ -29,7 +30,7 @@
"github.com/dubbogo/getty"
gxsync "github.com/dubbogo/gost/sync"
perrors "github.com/pkg/errors"
- "go.uber.org/atomic"
+ uatomic "go.uber.org/atomic"
"gopkg.in/yaml.v2"
)
@@ -134,7 +135,7 @@
opts Options
conf ClientConfig
pool *gettyRPCClientPool
- sequence atomic.Uint64
+ sequence uatomic.Uint64
pendingResponses *sync.Map
}
@@ -267,11 +268,25 @@
return errSessionNotExist
}
defer func() {
+ failNumber := 0
if err == nil {
- c.pool.put(conn)
- return
+ for {
+ ok := atomic.CompareAndSwapUint32(&c.pool.pushing, 0, 1)
+ if ok {
+ c.pool.poolQueue.PushHead(conn)
+ c.pool.pushing = 0
+ c.pool.ch <- struct{}{}
+ return
+ }
+ failNumber++
+ if failNumber%10 == 0 {
+ time.Sleep(1e6)
+ }
+ }
+ } else {
+ c.pool.ch <- struct{}{}
+ conn.close()
}
- conn.close()
}()
if err = c.transfer(session, p, rsp); err != nil {
diff --git a/protocol/dubbo/config.go b/protocol/dubbo/config.go
index e2bddb0..f14d298 100644
--- a/protocol/dubbo/config.go
+++ b/protocol/dubbo/config.go
@@ -18,6 +18,7 @@
package dubbo
import (
+ "fmt"
"time"
)
@@ -84,7 +85,7 @@
sessionTimeout time.Duration
// Connection Pool
- PoolSize int `default:"2" yaml:"pool_size" json:"pool_size,omitempty"`
+ PoolSize int `default:"4" yaml:"pool_size" json:"pool_size,omitempty"`
PoolTTL int `default:"180" yaml:"pool_ttl" json:"pool_ttl,omitempty"`
// grpool
@@ -184,8 +185,12 @@
}
if c.heartbeatPeriod >= time.Duration(config.MaxWheelTimeSpan) {
- return perrors.WithMessagef(err, "heartbeat_period %s should be less than %s",
- c.HeartbeatPeriod, time.Duration(config.MaxWheelTimeSpan))
+ return perrors.New(fmt.Sprintf("heartbeat_period %s should be less than %s",
+ c.HeartbeatPeriod, time.Duration(config.MaxWheelTimeSpan)))
+ }
+
+ if c.PoolSize <= 0 || (c.PoolSize&(c.PoolSize-1) != 0) {
+ return perrors.New(fmt.Sprintf("poolsize {%#v} should be bigger than 0 and pow of 2", c.PoolSize))
}
if c.sessionTimeout, err = time.ParseDuration(c.SessionTimeout); err != nil {
diff --git a/protocol/dubbo/dubbo_invoker.go b/protocol/dubbo/dubbo_invoker.go
index 09c3725..add0dcf 100644
--- a/protocol/dubbo/dubbo_invoker.go
+++ b/protocol/dubbo/dubbo_invoker.go
@@ -72,7 +72,7 @@
err error
result protocol.RPCResult
)
- if di.reqNum < 0 {
+ if atomic.LoadInt64(&di.reqNum) < 0 {
// Generally, the case will not happen, because the invoker has been removed
// from the invoker list before destroy,so no new request will enter the destroyed invoker
logger.Warnf("this dubboInvoker is destroyed")
@@ -125,9 +125,12 @@
// Destroy ...
func (di *DubboInvoker) Destroy() {
di.quitOnce.Do(func() {
+ if di.client != nil && di.client.pool != nil {
+ close(di.client.pool.closeCh)
+ }
for {
- if di.reqNum == 0 {
- di.reqNum = -1
+ if atomic.LoadInt64(&di.reqNum) == 0 {
+ atomic.StoreInt64(&di.reqNum, -1)
logger.Infof("dubboInvoker is destroyed,url:{%s}", di.GetUrl().Key())
di.BaseInvoker.Destroy()
if di.client != nil {
diff --git a/protocol/dubbo/pool.go b/protocol/dubbo/pool.go
index 918514c..787ab8e 100644
--- a/protocol/dubbo/pool.go
+++ b/protocol/dubbo/pool.go
@@ -28,6 +28,7 @@
import (
"github.com/dubbogo/getty"
+ gxqueue "github.com/dubbogo/gost/container/queue"
perrors "github.com/pkg/errors"
)
@@ -187,7 +188,6 @@
}
}()
if removeFlag {
- c.pool.safeRemove(c)
c.close()
}
}
@@ -288,117 +288,109 @@
}
type gettyRPCClientPool struct {
- rpcClient *Client
- size int // size of []*gettyRPCClient
- ttl int64 // ttl of every gettyRPCClient, it is checked when getConn
-
- sync.Mutex
- conns []*gettyRPCClient
+ rpcClient *Client
+ maxSize int // maxSize of poolQueue
+ ttl int64 // ttl of every gettyRPCClient, it is checked when getConn
+ activeNumber uint32
+ chInitialized uint32 // set to 1 when field ch is initialized
+ ch chan struct{}
+ closeCh chan struct{}
+ poolQueue gxqueue.SPMCLockFreeQ // store *gettyRPCClient
+ pushing uint32
+ sync.RWMutex
}
func newGettyRPCClientConnPool(rpcClient *Client, size int, ttl time.Duration) *gettyRPCClientPool {
+ pq, _ := gxqueue.NewSPMCLockFreeQ(size)
return &gettyRPCClientPool{
rpcClient: rpcClient,
- size: size,
+ maxSize: size,
ttl: int64(ttl.Seconds()),
- conns: make([]*gettyRPCClient, 0, 16),
+ closeCh: make(chan struct{}, 0),
+ poolQueue: pq,
}
}
func (p *gettyRPCClientPool) close() {
p.Lock()
- conns := p.conns
- p.conns = nil
+ connPool := p.poolQueue
+ p.poolQueue = nil
p.Unlock()
- for _, conn := range conns {
- conn.close()
+ for {
+ conn, ok := connPool.PopTail()
+ if ok {
+ c := conn.(*gettyRPCClient)
+ c.close()
+ } else {
+ break
+ }
}
}
+func (p *gettyRPCClientPool) lazyInit() {
+ // Fast path.
+ if atomic.LoadUint32(&p.chInitialized) == 1 {
+ return
+ }
+ // Slow path.
+ p.Lock()
+ defer p.Unlock()
+ if p.chInitialized == 0 {
+ p.ch = make(chan struct{}, p.maxSize)
+ for i := 0; i < p.maxSize; i++ {
+ p.ch <- struct{}{}
+ }
+ atomic.StoreUint32(&p.chInitialized, 1)
+ }
+}
+
+func (p *gettyRPCClientPool) waitVacantConn() error {
+ p.lazyInit()
+ select {
+ case <-p.ch:
+ // Additionally check that close chan hasn't expired while we were waiting,
+ // because `select` picks a random `case` if several of them are "ready".
+ select {
+ case <-p.closeCh:
+ return errClientPoolClosed
+ default:
+ }
+ case <-p.closeCh:
+ return errClientPoolClosed
+ }
+ return nil
+}
+
func (p *gettyRPCClientPool) getGettyRpcClient(protocol, addr string) (*gettyRPCClient, error) {
- conn, err := p.get()
+ err := p.waitVacantConn()
+ if err != nil {
+ return nil, err
+ }
+ conn, err := p.getConnFromPool()
if err == nil && conn == nil {
- // create new conn
rpcClientConn, err := newGettyRPCClientConn(p, protocol, addr)
return rpcClientConn, perrors.WithStack(err)
+
}
return conn, perrors.WithStack(err)
}
-func (p *gettyRPCClientPool) get() (*gettyRPCClient, error) {
+func (p *gettyRPCClientPool) getConnFromPool() (*gettyRPCClient, error) {
now := time.Now().Unix()
-
- p.Lock()
- defer p.Unlock()
- if p.conns == nil {
+ if p.poolQueue == nil {
return nil, errClientPoolClosed
}
-
- for len(p.conns) > 0 {
- conn := p.conns[len(p.conns)-1]
- p.conns = p.conns[:len(p.conns)-1]
-
- if d := now - conn.getActive(); d > p.ttl {
- p.remove(conn)
- go conn.close()
- 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
+ for {
+ value, ok := p.poolQueue.PopTail()
+ if ok {
+ conn := value.(*gettyRPCClient)
+ if d := now - conn.getActive(); d > p.ttl {
+ go conn.close()
+ continue
}
+ conn.updateActive(now)
+ return conn, nil
}
+ return nil, nil
}
}
-
-func (p *gettyRPCClientPool) safeRemove(conn *gettyRPCClient) {
- p.Lock()
- defer p.Unlock()
-
- p.remove(conn)
-}