Merge pull request #1151 from ydb-platform/revert-parking

Revert conns parking

Author: asmyasnikov

.github/workflows/breaking.yml

@@ -25,6 +25,8 @@ jobs:
         run: test -e ~/go/bin/gorelease || go install golang.org/x/exp/cmd/gorelease@latest
       - name: Check broken API changes
         run: gorelease -base=$GITHUB_BASE_REF 2>&1 > changes.txt | true
+      - name: Print API changes
+        run: cat changes.txt
       - name: Comment Report
         if: always()
         uses: marocchino/sticky-pull-request-comment@v2

CHANGELOG.md

@@ -4,7 +4,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

README.md

@@ -31,19 +31,20 @@ go get -u github.com/ydb-platform/ydb-go-sdk/v3
 ## Example Usage <a name="example"></a>
 
 * connect to YDB
-```golang
+```go
 db, err := ydb.Open(ctx, "grpc://localhost:2136/local")
 if err != nil {
     log.Fatal(err)
 }
 ```
-* execute `SELECT` query
- ```golang
-const query = `SELECT 42 as id, "myStr" as myStr;`
-
+* execute `SELECT` query over `Table` service client
+ ```go
 // Do retry operation on errors with best effort
-queryErr := db.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
-    _, res, err := s.Execute(ctx, table.DefaultTxControl(), query, nil)
+err := db.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) {
+    _, res, err := s.Execute(ctx, table.DefaultTxControl(), 
+		`SELECT 42 as id, "myStr" as myStr;`, 
+		nil, // empty parameters
+	)
     if err != nil {
         return err
     }
@@ -62,12 +63,61 @@ queryErr := db.Table().Do(ctx, func(ctx context.Context, s table.Session) (err e
     }
     return res.Err() // for driver retry if not nil
 })
-if queryErr != nil {
-    log.Fatal(queryErr)
+if err != nil {
+    log.Fatal(err)
+}
+```
+* execute `SELECT` query over `Query` service client
+ ```go
+// Do retry operation on errors with best effort
+err := db.Query().Do( // Do retry operation on errors with best effort
+	ctx, // context manage exiting from Do
+	func(ctx context.Context, s query.Session) (err error) { // retry operation
+		_, res, err := s.Execute(ctx, `SELECT 42 as id, "myStr" as myStr;`))
+		if err != nil {
+			return err // for auto-retry with driver
+		}
+		defer func() { _ = res.Close(ctx) }() // cleanup resources
+		for {                                 // iterate over result sets
+			rs, err := res.NextResultSet(ctx)
+			if err != nil {
+				if errors.Is(err, io.EOF) {
+					break
+				}
+
+				return err
+			}
+			for { // iterate over rows
+				row, err := rs.NextRow(ctx)
+				if err != nil {
+					if errors.Is(err, io.EOF) {
+						break
+					}
+
+					return err
+				}
+				type myStruct struct {
+					id  uint64 `sql:"id"`
+					str string `sql:"myStr"`
+				}
+				var s myStruct
+				if err = row.ScanStruct(&s); err != nil {
+					return err // generally scan error not retryable, return it for driver check error
+				}
+			}
+		}
+
+		return res.Err() // return finally result error for auto-retry with driver
+	},
+	query.WithIdempotent(),
+)
+if err != nil {
+    log.Fatal(err)
 }
 ```
+
 * usage with `database/sql` (see additional docs in [SQL.md](SQL.md) )
-```golang
+```go
 import (
     "context"
     "database/sql"
@@ -96,7 +146,7 @@ log.Printf("id = %d, myStr = \"%s\"", id, myStr)
 ```
 
 
-More examples of usage placed in [examples](https://github.com/ydb-platform/ydb-go-examples) repository.
+More examples of usage placed in [examples](./examples) directory.
 
 ## Credentials <a name="credentials"></a>
 

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

@@ -4,13 +4,15 @@ import (
 	"context"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"google.golang.org/grpc"
 	grpcCodes "google.golang.org/grpc/codes"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/closer"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/stack"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsync"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
@@ -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
 }

internal/topic/topicreaderinternal/batcher.go

@@ -176,7 +176,7 @@ func (b *batcher) Pop(ctx context.Context, opts batcherGetOptions) (_ batcherMes
 
 			findRes = b.findNeedLock(opts)
 			if findRes.Ok {
-				b.applyNeedLock(findRes)
+				b.applyNeedLock(&findRes)
 
 				return
 			}
@@ -279,7 +279,7 @@ func (b *batcher) applyForceFlagToOptions(options batcherGetOptions) batcherGetO
 	return res
 }
 
-func (b *batcher) applyNeedLock(res batcherResultCandidate) {
+func (b *batcher) applyNeedLock(res *batcherResultCandidate) {
 	if res.Rest.IsEmpty() && res.WaiterIndex >= 0 {
 		delete(b.messages, res.Key)
 	} else {

internal/topic/topicreaderinternal/batcher_test.go

@@ -407,7 +407,7 @@ func TestBatcher_Apply(t *testing.T) {
 			Key:  session,
 			Rest: batcherMessageOrderItems{newBatcherItemBatch(batch)},
 		}
-		b.applyNeedLock(foundRes)
+		b.applyNeedLock(&foundRes)
 
 		expectedMap := batcherMessagesMap{session: batcherMessageOrderItems{newBatcherItemBatch(batch)}}
 		require.Equal(t, expectedMap, b.messages)
@@ -426,7 +426,7 @@ func TestBatcher_Apply(t *testing.T) {
 
 		b.messages = batcherMessagesMap{session: batcherMessageOrderItems{newBatcherItemBatch(batch)}}
 
-		b.applyNeedLock(foundRes)
+		b.applyNeedLock(&foundRes)
 
 		require.Empty(t, b.messages)
 	})

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
 	}
 }