Revert commit of conns parking

Author: asmyasnikov

CHANGELOG.md

@@ -3,7 +3,6 @@
 * Added query pool metrics
 * Fixed logic of query session pool
 * Changed initialization of internal driver clients to lazy
-* Disabled the logic of background grpc-connection parking 
 * Removed `ydb.WithSessionPoolSizeLimit()` option
 * Added async put session into pool if external context is done
 * Dropped intermediate callbacks from `trace.{Table,Retry,Query}` events

config/config.go

@@ -21,6 +21,7 @@ type Config struct {
 
 	trace          *trace.Driver
 	dialTimeout    time.Duration
+	connectionTTL  time.Duration
 	balancerConfig *balancerConfig.Config
 	secure         bool
 	endpoint       string
@@ -56,6 +57,13 @@ func (c *Config) Meta() *meta.Meta {
 	return c.meta
 }
 
+// ConnectionTTL defines interval for parking grpc connections.
+//
+// If ConnectionTTL is zero - connections are not park.
+func (c *Config) ConnectionTTL() time.Duration {
+	return c.connectionTTL
+}
+
 // Secure is a flag for secure connection
 func (c *Config) Secure() bool {
 	return c.secure
@@ -169,6 +177,12 @@ func WithUserAgent(userAgent string) Option {
 	}
 }
 
+func WithConnectionTTL(ttl time.Duration) Option {
+	return func(c *Config) {
+		c.connectionTTL = ttl
+	}
+}
+
 func WithCredentials(credentials credentials.Credentials) Option {
 	return func(c *Config) {
 		c.credentials = credentials

internal/conn/config.go

@@ -10,6 +10,7 @@ import (
 
 type Config interface {
 	DialTimeout() time.Duration
+	ConnectionTTL() time.Duration
 	Trace() *trace.Driver
 	GrpcDialOptions() []grpc.DialOption
 }

internal/conn/conn.go

@@ -102,6 +102,36 @@ func (c *conn) NodeID() uint32 {
 	return 0
 }
 
+func (c *conn) park(ctx context.Context) (err error) {
+	onDone := trace.DriverOnConnPark(
+		c.config.Trace(), &ctx,
+		stack.FunctionID(""),
+		c.Endpoint(),
+	)
+	defer func() {
+		onDone(err)
+	}()
+
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	if c.closed {
+		return nil
+	}
+
+	if c.cc == nil {
+		return nil
+	}
+
+	err = c.close(ctx)
+
+	if err != nil {
+		return c.wrapError(err)
+	}
+
+	return nil
+}
+
 func (c *conn) Endpoint() endpoint.Endpoint {
 	if c != nil {
 		return c.endpoint

internal/conn/pool.go

@@ -2,8 +2,10 @@ package conn
 
 import (
 	"context"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"google.golang.org/grpc"
 	grpcCodes "google.golang.org/grpc/codes"
@@ -194,6 +196,39 @@ func (p *Pool) Release(ctx context.Context) (finalErr error) {
 	return nil
 }
 
+func (p *Pool) connParker(ctx context.Context, ttl, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-p.done:
+			return
+		case <-ticker.C:
+			for _, c := range p.collectConns() {
+				if time.Since(c.LastUsage()) > ttl {
+					switch c.GetState() {
+					case Online, Banned:
+						_ = c.park(ctx)
+					default:
+						// nop
+					}
+				}
+			}
+		}
+	}
+}
+
+func (p *Pool) collectConns() []*conn {
+	p.mtx.RLock()
+	defer p.mtx.RUnlock()
+	conns := make([]*conn, 0, len(p.conns))
+	for _, c := range p.conns {
+		conns = append(conns, c)
+	}
+
+	return conns
+}
+
 func NewPool(ctx context.Context, config Config) *Pool {
 	onDone := trace.DriverOnPoolNew(config.Trace(), &ctx, stack.FunctionID(""))
 	defer onDone()
@@ -206,5 +241,9 @@ func NewPool(ctx context.Context, config Config) *Pool {
 		done:   make(chan struct{}),
 	}
 
+	if ttl := config.ConnectionTTL(); ttl > 0 {
+		go p.connParker(xcontext.ValueOnly(ctx), ttl, ttl/2)
+	}
+
 	return p
 }

options.go

@@ -108,10 +108,10 @@ func WithConnectionString(connectionString string) Option {
 }
 
 // WithConnectionTTL defines duration for parking idle connections
-//
-// Deprecated: background connection parking not available
-func WithConnectionTTL(time.Duration) Option {
+func WithConnectionTTL(ttl time.Duration) Option {
 	return func(ctx context.Context, c *Driver) error {
+		c.options = append(c.options, config.WithConnectionTTL(ttl))
+
 		return nil
 	}
 }

trace/driver.go

@@ -37,6 +37,7 @@ type (
 		OnConnDial            func(DriverConnDialStartInfo) func(DriverConnDialDoneInfo)
 		OnConnBan             func(DriverConnBanStartInfo) func(DriverConnBanDoneInfo)
 		OnConnAllow           func(DriverConnAllowStartInfo) func(DriverConnAllowDoneInfo)
+		OnConnPark            func(DriverConnParkStartInfo) func(DriverConnParkDoneInfo)
 		OnConnClose           func(DriverConnCloseStartInfo) func(DriverConnCloseDoneInfo)
 
 		// Repeater events
@@ -225,6 +226,18 @@ type (
 	DriverConnDialDoneInfo struct {
 		Error error
 	}
+	DriverConnParkStartInfo struct {
+		// Context make available context in trace callback function.
+		// Pointer to context provide replacement of context in trace callback function.
+		// Warning: concurrent access to pointer on client side must be excluded.
+		// Safe replacement of context are provided only inside callback function
+		Context  *context.Context
+		Call     call
+		Endpoint EndpointInfo
+	}
+	DriverConnParkDoneInfo struct {
+		Error error
+	}
 	DriverConnCloseStartInfo struct {
 		// Context make available context in trace callback function.
 		// Pointer to context provide replacement of context in trace callback function.