Merge pull request #173 from ydb-platform/shared-pool

shared pool management with config

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,9 @@
+* Added `config.SharedPool()` setting and `config.WithSharedPool()` option
+* Added management of shared pool flag on change dial timeout and credentials 
+* Removed explicit checks of conditions for use (or not) shared pool in `ydb.With()`
+* Renamed `internal/db` interfaces
+* Changed signature of `conn.Conn.Release` (added error as result)
+
 ## v3.16.4
 * Removed `WithMeta()` discovery config option
 * Moved `meta.Meta` call to conn exclusively

config/config.go

@@ -76,6 +76,9 @@ type Config interface {
 
 	// UseDNSResolver is a flag about using dns-resolving or not
 	UseDNSResolver() bool
+
+	// SharedPool is a flag about using shared connection pool or detached
+	SharedPool() bool
 }
 
 // Config contains driver configuration options.
@@ -88,6 +91,7 @@ type config struct {
 	balancer             balancer.Balancer
 	secure               bool
 	dnsResolver          bool
+	sharedPool           bool
 	endpoint             string
 	database             string
 	requestsType         string
@@ -98,6 +102,10 @@ type config struct {
 	meta                 meta.Meta
 }
 
+func (c *config) SharedPool() bool {
+	return c.sharedPool
+}
+
 func (c *config) UseDNSResolver() bool {
 	return c.dnsResolver
 }
@@ -211,9 +219,16 @@ func WithConnectionTTL(ttl time.Duration) Option {
 	}
 }
 
+func WithSharedPool() Option {
+	return func(c *config) {
+		c.sharedPool = true
+	}
+}
+
 func WithCredentials(credentials credentials.Credentials) Option {
 	return func(c *config) {
 		c.credentials = credentials
+		c.sharedPool = false // use separated pool for another credentials
 	}
 }
 
@@ -231,6 +246,9 @@ func WithOperationCancelAfter(operationCancelAfter time.Duration) Option {
 
 func WithDialTimeout(timeout time.Duration) Option {
 	return func(c *config) {
+		if c.dialTimeout != timeout {
+			c.sharedPool = false // use separated pool for another dial timeout
+		}
 		c.dialTimeout = timeout
 	}
 }
@@ -312,8 +330,9 @@ func certPool() (certPool *x509.CertPool) {
 
 func defaultConfig() (c *config) {
 	return &config{
-		balancer: balancers.Default(),
-		secure:   true,
+		balancer:   balancers.Default(),
+		secure:     true,
+		sharedPool: true,
 		tlsConfig: &tls.Config{
 			MinVersion: tls.VersionTLS12,
 			RootCAs:    certPool(),

connection.go

@@ -34,7 +34,7 @@ import (
 // Interface and list of clients may be changed in the future
 type Connection interface {
 	closer.Closer
-	db.ConnectionInfo
+	db.Info
 	grpc.ClientConnInterface
 
 	// Table returns table client
@@ -265,7 +265,7 @@ func New(ctx context.Context, opts ...Option) (_ Connection, err error) {
 		)
 	}
 
-	if c.pool == nil {
+	if c.pool == nil || !c.config.SharedPool() {
 		c.pool = conn.NewPool(
 			ctx,
 			c.config,

internal/cluster/cluster.go

@@ -417,7 +417,9 @@ 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)
+	defer func() {
+		_ = entry.Conn.Release(ctx)
+	}()
 
 	removed = entry.RemoveFrom(c.balancer, &c.balancerMtx)
 

internal/conn/conn.go

@@ -39,7 +39,7 @@ type Conn interface {
 	GetState() State
 	SetState(State) State
 
-	Release(ctx context.Context)
+	Release(ctx context.Context) error
 }
 
 func (c *conn) Address() string {
@@ -60,19 +60,19 @@ type conn struct {
 	onClose      []func(*conn)
 }
 
-func (c *conn) Release(ctx context.Context) {
+func (c *conn) Release(ctx context.Context) (err error) {
 	var (
 		onDone = trace.DriverOnConnRelease(
 			c.config.Trace(),
 			&ctx,
 			c.endpoint.Copy(),
 		)
-		err error
+		issues []error
 	)
 	defer func() {
 		onDone(err)
 	}()
-	var issues []error
+
 	if c.changeUsages(-1) == 0 {
 		if usages := atomic.LoadInt32(&c.streamUsages); usages > 0 {
 			issues = append(issues, fmt.Errorf("conn in stream use: usages=%d", usages))
@@ -81,9 +81,12 @@ func (c *conn) Release(ctx context.Context) {
 			issues = append(issues, closeErr)
 		}
 	}
+
 	if len(issues) > 0 {
-		err = errors.NewWithIssues("conn released with issues", issues...)
+		return errors.NewWithIssues("conn released with issues", issues...)
 	}
+
+	return nil
 }
 
 func (c *conn) LastUsage() time.Time {

internal/db/connection.go

@@ -16,11 +16,11 @@ type Cluster interface {
 	closer.Closer
 }
 
-type ConnectionDiscovery interface {
+type Discoverer interface {
 	Discovery() discovery.Client
 }
 
-type ConnectionInfo interface {
+type Info interface {
 	// Endpoint returns initial endpoint
 	Endpoint() string
 
@@ -33,6 +33,6 @@ type ConnectionInfo interface {
 
 type Connection interface {
 	Cluster
-	ConnectionInfo
-	ConnectionDiscovery
+	Info
+	Discoverer
 }

internal/db/database.go

@@ -20,6 +20,7 @@ type database struct {
 	config    config.Config
 	cluster   cluster.Cluster
 	discovery discovery.Client
+	cc        conn.Conn
 }
 
 func (db *database) Discovery() discovery.Client {
@@ -64,6 +65,7 @@ func New(
 	db := &database{
 		config:  c,
 		cluster: cluster.New(ctx, c, pool, c.Balancer()),
+		cc:      pool.Create(endpoint.New(c.Endpoint(), endpoint.WithLocalDC(true))),
 	}
 
 	var cancel context.CancelFunc
@@ -74,17 +76,14 @@ func New(
 	}
 	defer cancel()
 
-	cc := pool.Create(endpoint.New(c.Endpoint(), endpoint.WithLocalDC(true)))
-
 	db.discovery, err = builder.New(
 		ctx,
-		cc,
+		db.cc,
 		db.cluster,
 		db.config.Trace(),
 		opts...,
 	)
 	if err != nil {
-		cc.Release(ctx)
 		return nil, err
 	}
 

internal/discovery/discovery.go

@@ -197,6 +197,5 @@ func (c *client) WhoAmI(ctx context.Context) (whoAmI *discovery.WhoAmI, err erro
 }
 
 func (c *client) Close(ctx context.Context) error {
-	c.cc.Release(ctx)
-	return nil
+	return c.cc.Release(ctx)
 }

test/connection_test.go

@@ -83,6 +83,7 @@ func TestConnection(t *testing.T) {
 		ydb.WithUserAgent(userAgent),
 		ydb.WithRequestsType(requestType),
 		ydb.With(
+			config.WithInternalDNSResolver(),
 			config.WithGrpcOptions(
 				grpc.WithUnaryInterceptor(func(
 					ctx context.Context,
@@ -236,7 +237,7 @@ func TestConnection(t *testing.T) {
 		var childDB ydb.Connection
 		childDB, err = db.With(
 			ctx,
-			ydb.WithAccessTokenCredentials(""),
+			ydb.WithDialTimeout(time.Second*5),
 		)
 		if err != nil {
 			t.Fatalf("failed to open sub-connection: %v", err)

with.go

@@ -10,43 +10,41 @@ import (
 var nextID = uint64(0)
 
 func (c *connection) With(ctx context.Context, opts ...Option) (Connection, error) {
-	if len(opts) == 0 {
-		return c, nil
-	}
+	id := atomic.AddUint64(&nextID, 1)
 
 	opts = append(
 		opts,
 		WithBalancer(
 			c.config.Balancer().Create(),
 		),
-	)
-
-	id := atomic.AddUint64(&nextID, 1)
-
-	opts = append(
-		opts,
 		withOnClose(func(child *connection) {
 			c.childrenMtx.Lock()
 			defer c.childrenMtx.Unlock()
 
 			delete(c.children, id)
 		}),
+		withConnPool(c.pool),
 	)
-	// check if credentials have been overridden
-	tmp := new(connection)
-	for _, o := range opts {
-		_ = o(ctx, tmp)
-	}
-	tmpCfg := config.New(tmp.options...)
-	if tmpCfg.Credentials() == nil {
-		// use previous credentials, so we can share conn pool
-		opts = append(opts, withConnPool(c.pool))
-	}
 
 	child, err := New(
 		ctx,
 		append(
-			c.opts,
+			append(
+				c.opts,
+				With(
+					config.WithSharedPool(), // force set to true prefer shared pool
+					config.WithBalancer(
+						c.config.Balancer().Create(),
+					),
+				),
+				withOnClose(func(child *connection) {
+					c.childrenMtx.Lock()
+					defer c.childrenMtx.Unlock()
+
+					delete(c.children, id)
+				}),
+				withConnPool(c.pool),
+			),
 			opts...,
 		)...,
 	)