remove node session counting

Author: asmyasnikov

connection.go

@@ -3,7 +3,6 @@ package ydb
 import (
 	"context"
 	"errors"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
 	"os"
 	"sync"
 
@@ -226,9 +225,6 @@ func (c *connection) Table() table.Client {
 				)...,
 			),
 		)
-		c.balancer.OnUpdateEndpoints(func(nodes []endpoint.Info) {
-			c.table.UpdateNodes(nodes)
-		})
 		return c.table.Close
 	})
 	// may be nil if driver closed early

internal/balancer/balancer.go

@@ -3,6 +3,7 @@ package balancer
 import (
 	"context"
 	"fmt"
+
 	"google.golang.org/grpc"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/config"
@@ -30,14 +31,6 @@ type Balancer struct {
 
 	mu               xsync.RWMutex
 	connectionsState *connectionsState
-
-	onUpdateEndpoints []func(endpoints []endpoint.Info)
-}
-
-func (b *Balancer) OnUpdateEndpoints(cb func(nodes []endpoint.Info)) {
-	b.mu.WithLock(func() {
-		b.onUpdateEndpoints = append(b.onUpdateEndpoints, cb)
-	})
 }
 
 func (b *Balancer) clusterDiscovery(ctx context.Context) (err error) {
@@ -92,13 +85,6 @@ func (b *Balancer) applyDiscoveredEndpoints(ctx context.Context, endpoints []end
 
 	b.mu.WithLock(func() {
 		b.connectionsState = state
-		nodes := make([]endpoint.Info, len(endpoints))
-		for i := range endpoints {
-			nodes[i] = endpoints[i]
-		}
-		for _, cb := range b.onUpdateEndpoints {
-			cb(nodes)
-		}
 	})
 }
 

internal/balancer/ctx.go

@@ -14,16 +14,14 @@ func WithEndpoint(ctx context.Context, endpoint Endpoint) context.Context {
 	return context.WithValue(ctx, ctxEndpointKey{}, endpoint)
 }
 
-type nodeIDer struct {
-	nodeID uint32
-}
+type nodeID uint32
 
-func (n nodeIDer) NodeID() uint32 {
-	return n.nodeID
+func (n nodeID) NodeID() uint32 {
+	return uint32(n)
 }
 
-func WithNodeID(ctx context.Context, nodeID uint32) context.Context {
-	return WithEndpoint(ctx, nodeIDer{nodeID: nodeID})
+func WithNodeID(ctx context.Context, id uint32) context.Context {
+	return WithEndpoint(ctx, nodeID(id))
 }
 
 func ContextEndpoint(ctx context.Context) (e Endpoint, ok bool) {

internal/table/client.go

@@ -4,9 +4,6 @@ import (
 	"container/list"
 	"context"
 	"fmt"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
-	"math"
-	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -49,14 +46,13 @@ func newClient(
 		onDone = trace.TableOnInit(config.Trace(), &ctx)
 	)
 	c := &Client{
-		config:      config,
-		cc:          cc,
-		build:       builder,
-		index:       make(map[*session]sessionInfo),
-		nodeLoading: make(map[uint32]int),
-		idle:        list.New(),
-		waitq:       list.New(),
-		limit:       config.SizeLimit(),
+		config: config,
+		cc:     cc,
+		build:  builder,
+		index:  make(map[*session]sessionInfo),
+		idle:   list.New(),
+		waitq:  list.New(),
+		limit:  config.SizeLimit(),
 		waitChPool: sync.Pool{
 			New: func() interface{} {
 				ch := make(chan *session)
@@ -83,11 +79,10 @@ type Client struct {
 	cc                grpc.ClientConnInterface
 	config            config.Config
 	index             map[*session]sessionInfo
-	createInProgress  int        // KIKIMR-9163: in-create-process counter
-	limit             int        // Upper bound for Client size.
-	idle              *list.List // list<table.session>
-	waitq             *list.List // list<*chan table.session>
-	nodeLoading       map[uint32]int
+	createInProgress  int           // KIKIMR-9163: in-create-process counter
+	limit             int           // Upper bound for Client size.
+	idle              *list.List    // list<table.session>
+	waitq             *list.List    // list<*chan table.session>
 	keeperWake        chan struct{} // Set by internalPoolKeeper.
 	keeperStop        chan struct{}
 	keeperDone        chan struct{}
@@ -127,8 +122,6 @@ func (c *Client) CreateSession(ctx context.Context, opts ...table.Option) (_ tab
 				defer cancel()
 			}
 
-			createSessionCtx = balancer.WithNodeID(createSessionCtx, c.nextNodeID())
-
 			s, err = c.build(createSessionCtx)
 
 			select {
@@ -161,21 +154,6 @@ func (c *Client) CreateSession(ctx context.Context, opts ...table.Option) (_ tab
 		}
 	}
 	var s *session
-	defer func() {
-		if s != nil {
-			c.mu.WithLock(func() {
-				c.nodeLoading[s.NodeID()]++
-			})
-			s.onClose = append(s.onClose, func(s *session) {
-				c.mu.WithLock(func() {
-					nodeID := s.NodeID()
-					if _, has := c.nodeLoading[nodeID]; has {
-						c.nodeLoading[nodeID]--
-					}
-				})
-			})
-		}
-	}()
 	if !c.config.AutoRetry() {
 		s, err = createSession(ctx)
 		if err != nil {
@@ -219,28 +197,10 @@ func (c *Client) isClosed() bool {
 	return atomic.LoadUint32(&c.closed) != 0
 }
 
-func (c *Client) nextNodeID() (nodeID uint32) {
-	nodeID = math.MaxUint32
-	c.mu.WithLock(func() {
-		min := math.MaxInt
-		for id, count := range c.nodeLoading {
-			if count < min {
-				min = count
-				nodeID = id
-			}
-		}
-		fmt.Println("nextNodeID =", nodeID, "min =", min)
-	})
-	return nodeID
-}
-
 // c.mu must NOT be held.
 func (c *Client) internalPoolCreateSession(ctx context.Context) (s *session, err error) {
 	defer func() {
 		if s != nil {
-			c.mu.WithLock(func() {
-				c.nodeLoading[s.NodeID()]++
-			})
 			s.onClose = append(s.onClose, func(s *session) {
 				c.spawnedGoroutines.Start("onClose", func(ctx context.Context) {
 					c.mu.WithLock(func() {
@@ -259,11 +219,6 @@ func (c *Client) internalPoolCreateSession(ctx context.Context) (s *session, err
 						if info.idle != nil {
 							c.idle.Remove(info.idle)
 						}
-
-						nodeID := s.NodeID()
-						if _, has := c.nodeLoading[nodeID]; has {
-							c.nodeLoading[nodeID]--
-						}
 					})
 				})
 			})
@@ -308,8 +263,6 @@ func (c *Client) internalPoolCreateSession(ctx context.Context) (s *session, err
 			defer cancel()
 		}
 
-		createSessionCtx = balancer.WithNodeID(createSessionCtx, c.nextNodeID())
-
 		s, err = c.build(createSessionCtx)
 		if s == nil && err == nil {
 			panic("ydb: abnormal result of session build")
@@ -321,8 +274,6 @@ func (c *Client) internalPoolCreateSession(ctx context.Context) (s *session, err
 				c.index[s] = sessionInfo{}
 				trace.TableOnPoolSessionAdd(c.config.Trace(), s)
 				trace.TableOnPoolStateChange(c.config.Trace(), len(c.index), "append")
-				c.nodeLoading[s.NodeID()]++
-				fmt.Printf("c.nodeLoading: %v", c.nodeLoading)
 			}
 		})
 
@@ -356,28 +307,6 @@ func (c *Client) internalPoolCreateSession(ctx context.Context) (s *session, err
 	}
 }
 
-func (c *Client) UpdateNodes(nodes []endpoint.Info) {
-	sort.Slice(nodes, func(i, j int) bool {
-		return nodes[i].NodeID() < nodes[j].NodeID()
-	})
-	c.mu.WithLock(func() {
-		for _, node := range nodes {
-			nodeID := node.NodeID()
-			if _, has := c.nodeLoading[nodeID]; !has {
-				c.nodeLoading[nodeID] = 0
-			}
-		}
-		nodeLoading := c.nodeLoading
-		for nodeID := range nodeLoading {
-			if sort.Search(len(nodes), func(i int) bool {
-				return nodes[i].NodeID() <= nodeID
-			}) < len(nodes) {
-				delete(c.nodeLoading, nodeID)
-			}
-		}
-	})
-}
-
 type getOptions struct {
 	t trace.Table
 }