Merge pull request #155 from ydb-platform/auto-remove-conn-from-pool

v3.14.4: auto-removing `conn.Conn` from `conn.Pool`

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 3.14.4
+* Implemented auto-removing `conn.Conn` from `conn.Pool` with counting usages of `conn.Conn`
+* Refactored naming of source files which declares service client interfaces
+
 ## 3.14.3
 * Fixed bug with update balancer element with nil handle
 

coordination/coordination.go renamed to coordination/client.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/closer"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 )
 
@@ -18,7 +19,8 @@ func (w WhoAmI) String() string {
 }
 
 type Client interface {
+	closer.Closer
+
 	Discover(ctx context.Context) ([]endpoint.Endpoint, error)
 	WhoAmI(ctx context.Context) (*WhoAmI, error)
-	Close(ctx context.Context) error
 }

discovery/client.go

@@ -102,10 +102,6 @@ func (c *cluster) Unlock() {
 	c.mu.Unlock()
 }
 
-func (c *cluster) GetConn(endpoint endpoint.Endpoint) conn.Conn {
-	return c.pool.GetConn(endpoint)
-}
-
 func (c *cluster) Force() {
 	c.explorer.Force()
 }
@@ -171,7 +167,6 @@ type Cluster interface {
 	CRUD
 	Explorer
 	Locker
-	conn.PoolGetter
 	conn.Pessimizer
 }
 
@@ -211,10 +206,40 @@ func (c *cluster) Close(ctx context.Context) (err error) {
 	if c.explorer != nil {
 		c.explorer.Stop()
 	}
+
+	for _, entry := range c.index {
+		c.Remove(
+			ctx,
+			entry.Conn.Endpoint(),
+			WithoutLock(),
+		)
+	}
+
 	c.closed = true
 
-	c.index = nil
-	c.endpoints = nil
+	if len(c.index) > 0 {
+		panic(fmt.Sprintf(
+			"non empty index after remove all entries: %v",
+			func() (endpoints []string) {
+				for e := range c.index {
+					endpoints = append(endpoints, e)
+				}
+				return endpoints
+			}(),
+		))
+	}
+
+	if len(c.endpoints) > 0 {
+		panic(fmt.Sprintf(
+			"non empty nodes after remove all entries: %v",
+			func() (nodes []uint32) {
+				for e := range c.endpoints {
+					nodes = append(nodes, e)
+				}
+				return nodes
+			}(),
+		))
+	}
 
 	return c.pool.Release(ctx)
 }
@@ -284,7 +309,7 @@ func (c *cluster) Insert(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		return nil
 	}
 
-	cc = c.pool.GetConn(e)
+	cc = c.pool.Get(ctx, e)
 
 	_, has := c.index[e.Address()]
 	if has {
@@ -335,7 +360,12 @@ func (c *cluster) Update(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		panic("ydb: cluster entry with nil conn")
 	}
 
-	entry.Conn.Endpoint().Touch()
+	entry.Conn.Endpoint().Touch(
+		endpoint.WithLocation(e.Location()),
+		endpoint.WithID(e.NodeID()),
+		endpoint.WithLoadFactor(e.LoadFactor()),
+		endpoint.WithLocalDC(e.LocalDC()),
+	)
 
 	delete(c.endpoints, e.NodeID())
 	c.index[e.Address()] = entry
@@ -380,6 +410,8 @@ func (c *cluster) Remove(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		panic("ydb: can't remove not-existing endpoint")
 	}
 
+	defer entry.Conn.Release(ctx)
+
 	removed = entry.RemoveFrom(c.balancer, &c.balancerMtx)
 
 	delete(c.index, e.Address())

internal/cluster/cluster.go

@@ -3,6 +3,7 @@ package conn
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -16,7 +17,6 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/errors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/response"
-	"github.com/ydb-platform/ydb-go-sdk/v3/testutil/timeutil"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
@@ -32,11 +32,13 @@ type Conn interface {
 
 	Endpoint() endpoint.Endpoint
 
-	TTL() <-chan time.Time
+	LastUsage() time.Time
 
 	IsState(states ...State) bool
 	GetState() State
 	SetState(State) State
+
+	Release(ctx context.Context)
 }
 
 func (c *conn) Address() string {
@@ -45,15 +47,28 @@ func (c *conn) Address() string {
 
 type conn struct {
 	sync.RWMutex
-	config   Config // ro access
-	cc       *grpc.ClientConn
-	done     chan struct{}
-	endpoint endpoint.Endpoint // ro access
-	closed   bool
-	state    State
-	usages   int32
-	ttl      timeutil.Timer
-	onClose  []func(*conn)
+	config    Config // ro access
+	cc        *grpc.ClientConn
+	done      chan struct{}
+	endpoint  endpoint.Endpoint // ro access
+	closed    bool
+	state     State
+	usages    int32
+	lastUsage time.Time
+	onClose   []func(*conn)
+}
+
+func (c *conn) Release(ctx context.Context) {
+	if c.decUsages() == 0 {
+		_ = c.Close(ctx)
+	}
+}
+
+func (c *conn) LastUsage() time.Time {
+	c.RLock()
+	defer c.RUnlock()
+
+	return c.lastUsage
 }
 
 func (c *conn) IsState(states ...State) bool {
@@ -69,7 +84,7 @@ func (c *conn) IsState(states ...State) bool {
 	return false
 }
 
-func (c *conn) Park(ctx context.Context) (err error) {
+func (c *conn) park(ctx context.Context) (err error) {
 	c.Lock()
 	defer c.Unlock()
 
@@ -135,19 +150,6 @@ func (c *conn) GetState() (s State) {
 	return c.state
 }
 
-func (c *conn) TTL() <-chan time.Time {
-	if c.config.ConnectionTTL() == 0 {
-		return nil
-	}
-	if c.isClosed() {
-		return nil
-	}
-	if atomic.LoadInt32(&c.usages) > 0 {
-		return nil
-	}
-	return c.ttl.C()
-}
-
 func (c *conn) take(ctx context.Context) (cc *grpc.ClientConn, err error) {
 	onDone := trace.DriverOnConnTake(
 		trace.ContextDriver(ctx).Compose(c.config.Trace()),
@@ -192,25 +194,30 @@ func (c *conn) take(ctx context.Context) (cc *grpc.ClientConn, err error) {
 	return c.cc, nil
 }
 
-func (c *conn) changeUsages(delta int32) {
-	trace.DriverOnConnUsagesChange(
-		c.config.Trace(),
-		c.endpoint.Copy(),
-		int(atomic.AddInt32(&c.usages, delta)),
-	)
+func (c *conn) changeUsages(delta int32) int32 {
+	if usages := atomic.AddInt32(&c.usages, delta); usages < 0 {
+		panic("negative usages" + strconv.Itoa(int(usages)))
+	} else {
+		trace.DriverOnConnUsagesChange(
+			c.config.Trace(),
+			c.endpoint.Copy(),
+			int(usages),
+		)
+		return usages
+	}
 }
 
 func (c *conn) incUsages() {
+	c.Lock()
+	defer c.Unlock()
 	c.changeUsages(1)
 }
 
-func (c *conn) decUsages() {
+func (c *conn) decUsages() int32 {
 	c.Lock()
 	defer c.Unlock()
-	if ttl := c.config.ConnectionTTL(); ttl > 0 {
-		c.ttl.Reset(ttl)
-	}
-	c.changeUsages(-1)
+	c.lastUsage = time.Now()
+	return c.changeUsages(-1)
 }
 
 func isBroken(raw *grpc.ClientConn) bool {
@@ -433,6 +440,7 @@ func withOnClose(onClose func(*conn)) option {
 
 func newConn(e endpoint.Endpoint, config Config, opts ...option) *conn {
 	c := &conn{
+		usages:   1,
 		state:    Created,
 		endpoint: e,
 		config:   config,
@@ -442,9 +450,6 @@ func newConn(e endpoint.Endpoint, config Config, opts ...option) *conn {
 	for _, o := range opts {
 		o(c)
 	}
-	if ttl := config.ConnectionTTL(); ttl > 0 {
-		c.ttl = timeutil.NewTimer(ttl)
-	}
 	return c
 }
 

internal/conn/conn.go

@@ -44,10 +44,12 @@ func (s *grpcClientStream) Context() context.Context {
 func (s *grpcClientStream) SendMsg(m interface{}) (err error) {
 	s.c.incUsages()
 	defer s.c.decUsages()
+
 	err = s.s.SendMsg(m)
 	if err != nil && s.wrapping {
 		err = errors.Error(errors.MapGRPCError(err))
 	}
+
 	return
 }