Merge pull request #171 from candiduslynx/feature/fill-in-meta-in-conn

move meta call to conn exclusively

Author: asmyasnikov

.gitignore

@@ -1,2 +1,3 @@
 .idea
 .DS_Store
+vendor

CHANGELOG.md

@@ -1,3 +1,6 @@
+* Removed `WithMeta()` discovery config option
+* Moved `meta.Meta` call to conn exclusively
+
 ## v3.16.3
 * Replaced panic on cluster close to error issues
 

connection.go

@@ -282,7 +282,6 @@ func New(ctx context.Context, opts ...Option) (_ Connection, err error) {
 				discoveryConfig.WithEndpoint(c.Endpoint()),
 				discoveryConfig.WithDatabase(c.Name()),
 				discoveryConfig.WithSecure(c.Secure()),
-				discoveryConfig.WithMeta(c.config.Meta()),
 				discoveryConfig.WithOperationTimeout(c.config.OperationTimeout()),
 				discoveryConfig.WithOperationCancelAfter(c.config.OperationCancelAfter()),
 			},

discovery/config/config.go

@@ -3,7 +3,6 @@ package config
 import (
 	"time"
 
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/meta"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
@@ -42,16 +41,12 @@ type Config interface {
 	// If Interval is zero then the DefaultInterval is used.
 	// If Interval is negative, then no background discovery prepared.
 	Interval() time.Duration
-
-	// Meta is an option which contains meta information about database connection
-	Meta() meta.Meta
 }
 
 type config struct {
 	endpoint string
 	database string
 	secure   bool
-	meta     meta.Meta
 
 	operationTimeout     time.Duration
 	operationCancelAfter time.Duration
@@ -60,10 +55,6 @@ type config struct {
 	trace    trace.Discovery
 }
 
-func (c *config) Meta() meta.Meta {
-	return c.meta
-}
-
 func (c *config) OperationTimeout() time.Duration {
 	return c.operationTimeout
 }
@@ -100,12 +91,6 @@ func WithEndpoint(endpoint string) Option {
 	}
 }
 
-func WithMeta(meta meta.Meta) Option {
-	return func(c *config) {
-		c.meta = meta
-	}
-}
-
 func WithDatabase(database string) Option {
 	return func(c *config) {
 		c.database = database

internal/cluster/cluster.go

@@ -151,22 +151,17 @@ type Explorer interface {
 	Force()
 }
 
-type Locker interface {
-	Lock()
-	Unlock()
-}
-
 type CRUDExplorerLocker interface {
 	CRUD
 	Explorer
-	Locker
+	sync.Locker
 }
 
 type Cluster interface {
 	closer.Closer
 	CRUD
 	Explorer
-	Locker
+	sync.Locker
 	conn.Pessimizer
 }
 
@@ -260,8 +255,11 @@ func (c *cluster) Get(ctx context.Context, opts ...crudOption) (cc conn.Conn, er
 	ctx, cancel = context.WithTimeout(ctx, MaxGetConnTimeout)
 	defer cancel()
 
-	c.mu.RLock()
-	defer c.mu.RUnlock()
+	options := parseOptions(opts...)
+	if options.withLock {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+	}
 
 	if c.closed {
 		return nil, errors.WithStackTrace(ErrClusterClosed)
@@ -318,7 +316,7 @@ func (c *cluster) Insert(ctx context.Context, e endpoint.Endpoint, opts ...crudO
 		return nil
 	}
 
-	cc = c.pool.Get(ctx, e)
+	cc = c.pool.Get(e)
 
 	_, has := c.index[e.Address()]
 	if has {

internal/conn/conn.go

@@ -236,7 +236,7 @@ func (c *conn) changeUsages(delta int32) int32 {
 	usages := atomic.AddInt32(&c.usages, delta)
 
 	if usages < 0 {
-		panic("negative usages" + strconv.Itoa(int(usages)))
+		panic("negative usages: " + strconv.Itoa(int(usages)))
 	}
 
 	trace.DriverOnConnUsagesChange(
@@ -254,7 +254,7 @@ func (c *conn) changeStreamUsages(delta int32) {
 	usages := atomic.AddInt32(&c.streamUsages, delta)
 
 	if usages < 0 {
-		panic("negative stream usages" + strconv.Itoa(int(usages)))
+		panic("negative stream usages: " + strconv.Itoa(int(usages)))
 	}
 
 	trace.DriverOnConnStreamUsagesChange(
@@ -433,6 +433,14 @@ func (c *conn) newStream(
 		)
 	}
 
+	ctx, err = c.config.Meta().Meta(ctx)
+	if err != nil {
+		return nil, errors.NewGrpcError(
+			errors.WithStatus(grpcStatus.New(codes.Unavailable, "ydb driver conn apply meta failed")),
+			errors.WithErr(err),
+		)
+	}
+
 	c.changeStreamUsages(1)
 	defer c.changeStreamUsages(-1)
 

internal/conn/pool.go

@@ -15,14 +15,19 @@ import (
 )
 
 type Pool interface {
+	Creator
 	Getter
 	Taker
 	Releaser
 	Pessimizer
 }
 
+type Creator interface {
+	Create(endpoint endpoint.Endpoint) Conn // creates new detached conn applying pool config
+}
+
 type Getter interface {
-	Get(ctx context.Context, endpoint endpoint.Endpoint) Conn
+	Get(endpoint endpoint.Endpoint) Conn
 }
 
 type Taker interface {
@@ -51,7 +56,11 @@ type pool struct {
 	done   chan struct{}
 }
 
-func (p *pool) Get(ctx context.Context, endpoint endpoint.Endpoint) Conn {
+func (p *pool) Create(endpoint endpoint.Endpoint) Conn {
+	return newConn(endpoint, p.config)
+}
+
+func (p *pool) Get(endpoint endpoint.Endpoint) Conn {
 	p.mtx.Lock()
 	defer p.mtx.Unlock()
 

internal/db/database.go

@@ -74,7 +74,7 @@ func New(
 	}
 	defer cancel()
 
-	cc := pool.Get(ctx, endpoint.New(c.Endpoint(), endpoint.WithLocalDC(true)))
+	cc := pool.Create(endpoint.New(c.Endpoint(), endpoint.WithLocalDC(true)))
 
 	db.discovery, err = builder.New(
 		ctx,
@@ -115,11 +115,6 @@ func (db *database) Invoke(
 		return errors.WithStackTrace(err)
 	}
 
-	ctx, err = db.config.Meta().Meta(ctx)
-	if err != nil {
-		return errors.WithStackTrace(err)
-	}
-
 	defer func() {
 		if err != nil && errors.MustPessimizeEndpoint(err) {
 			db.cluster.Pessimize(ctx, cc, err)
@@ -145,11 +140,6 @@ func (db *database) NewStream(
 		return nil, errors.WithStackTrace(err)
 	}
 
-	ctx, err = db.config.Meta().Meta(ctx)
-	if err != nil {
-		return nil, errors.WithStackTrace(err)
-	}
-
 	defer func() {
 		if err != nil && errors.MustPessimizeEndpoint(err) {
 			db.cluster.Pessimize(ctx, cc, err)

internal/discovery/discovery.go

@@ -137,11 +137,6 @@ func (c *client) Discover(ctx context.Context) (endpoints []endpoint.Endpoint, e
 		onDone(location, nodes, err)
 	}()
 
-	ctx, err = c.config.Meta().Meta(ctx)
-	if err != nil {
-		return nil, errors.WithStackTrace(err)
-	}
-
 	response, err = c.service.ListEndpoints(ctx, &request)
 	if err != nil {
 		return nil, errors.WithStackTrace(err)
@@ -185,11 +180,6 @@ func (c *client) WhoAmI(ctx context.Context) (whoAmI *discovery.WhoAmI, err erro
 		}
 	}()
 
-	ctx, err = c.config.Meta().Meta(ctx)
-	if err != nil {
-		return nil, errors.WithStackTrace(err)
-	}
-
 	response, err = c.service.WhoAmI(ctx, &request)
 	if err != nil {
 		return nil, errors.WithStackTrace(err)

options.go

@@ -178,15 +178,17 @@ func WithAnonymousCredentials() Option {
 
 func WithCreateCredentialsFunc(createCredentials func(ctx context.Context) (credentials.Credentials, error)) Option {
 	return func(ctx context.Context, c *connection) error {
-		credentials, err := createCredentials(ctx)
+		creds, err := createCredentials(ctx)
 		if err != nil {
 			return errors.WithStackTrace(err)
 		}
-		c.options = append(c.options, config.WithCredentials(credentials))
+		c.options = append(c.options, config.WithCredentials(creds))
 		return nil
 	}
 }
 
+// WithCredentials in conjunction with Connection.With function prohibit reuse of conn pool.
+// Thus, Connection.With will effectively create totally separate Connection.
 func WithCredentials(c credentials.Credentials) Option {
 	return WithCreateCredentialsFunc(func(context.Context) (credentials.Credentials, error) {
 		return c, nil