fallback if not found items by node id

Author: asmyasnikov

internal/pool/pool.go

@@ -444,7 +444,7 @@ func (p *Pool[PT, T]) Close(ctx context.Context) (finalErr error) {
 	}
 }
 
-// getWaitCh returns pointer to a channel of sessions.
+// getWaitCh returns pointer to a channel of items.
 //
 // Note that returning a pointer reduces allocations on sync.Pool usage –
 // sync.Client.Get() returns empty interface, which leads to allocation for
@@ -494,9 +494,7 @@ func (p *Pool[PT, T]) peekFirstIdleByNodeID(nodeID uint32) (item PT, touched tim
 	return item, info.lastUsage
 }
 
-// removes first session from idle and resets the keepAliveCount
-// to prevent session from dying in the internalPoolGC after it was returned
-// to be used only in outgoing functions that make session busy.
+// removes first item from idle to use only in outgoing functions that make item busy.
 // p.mu must be held.
 func (p *Pool[PT, T]) removeFirstIdle() PT {
 	idle, _ := p.peekFirstIdle()
@@ -508,9 +506,7 @@ func (p *Pool[PT, T]) removeFirstIdle() PT {
 	return idle
 }
 
-// removes first session from idle and resets the keepAliveCount
-// to prevent session from dying in the internalPoolGC after it was returned
-// to be used only in outgoing functions that make session busy.
+// removes first item with preferred nodeID from idle to use only in outgoing functions that make item busy.
 // p.mu must be held.
 func (p *Pool[PT, T]) removeIdleByNodeID(nodeID uint32) PT {
 	idle, _ := p.peekFirstIdleByNodeID(nodeID)
@@ -525,7 +521,7 @@ func (p *Pool[PT, T]) removeIdleByNodeID(nodeID uint32) PT {
 // p.mu must be held.
 func (p *Pool[PT, T]) notifyAboutIdle(idle PT) (notified bool) {
 	for el := p.waitQ.Front(); el != nil; el = p.waitQ.Front() {
-		// Some goroutine is waiting for a session.
+		// Some goroutine is waiting for a item.
 		//
 		// It could be in this states:
 		//   1) Reached the select code and awaiting for a value in channel.
@@ -566,7 +562,7 @@ func (p *Pool[PT, T]) notifyAboutIdle(idle PT) (notified bool) {
 func (p *Pool[PT, T]) removeIdle(item PT) itemInfo[PT, T] {
 	info, has := p.index[item]
 	if !has || info.idle == nil {
-		panic("inconsistent session client index")
+		panic("inconsistent item client index")
 	}
 
 	p.changeState(func() Stats {
@@ -630,7 +626,15 @@ func (p *Pool[PT, T]) getItem(ctx context.Context) (item PT, finalErr error) { /
 
 		if item := xsync.WithLock(&p.mu, func() PT { //nolint:nestif
 			if hasPreferredNodeID {
-				return p.removeIdleByNodeID(preferredNodeID)
+				item := p.removeIdleByNodeID(preferredNodeID)
+				if item != nil {
+					return item
+				}
+
+				if len(p.index)+p.createInProgress < p.config.limit {
+					// for create item with preferred nodeID
+					return nil
+				}
 			}
 
 			return p.removeFirstIdle()
@@ -746,15 +750,15 @@ func (p *Pool[PT, T]) waitFromCh(ctx context.Context) (item PT, finalErr error)
 
 	case item, ok := <-*ch:
 		// Note that race may occur and some goroutine may try to write
-		// session into channel after it was enqueued but before it being
+		// item into channel after it was enqueued but before it being
 		// read here. In that case we will receive nil here and will retry.
 		//
-		// The same way will work when some session become deleted - the
+		// The same way will work when some item become deleted - the
 		// nil value will be sent into the channel.
 		if ok {
 			// Put only filled and not closed channel back to the Client.
 			// That is, we need to avoid races on filling reused channel
-			// for the next waiter – session could be lost for a long time.
+			// for the next waiter – item could be lost for a long time.
 			p.putWaitCh(ch)
 		}