Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 39 additions & 13 deletions go/pools/smartconnpool/pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ type ConnPool[C Connection] struct {
// workers is a waitgroup for all the currently running worker goroutines
workers sync.WaitGroup
close chan struct{}
capacityMu sync.Mutex
capacityMu sync.RWMutex

config struct {
// connect is the callback to create a new connection for the pool
Expand Down Expand Up @@ -430,18 +430,40 @@ func (pool *ConnPool[C]) tryReturnConn(conn *Pooled[C]) bool {
if pool.wait.tryReturnConn(conn) {
return true
}
if pool.closeOnIdleLimitReached(conn) {

for {
if pool.capacity.Load() < pool.active.Load() {
conn.Close()
pool.closedConn()
return true
}

if pool.closeOnIdleLimitReached(conn) {
return false
}

if !pool.capacityMu.TryRLock() {
// If we can't get a read lock here, it means that the pool is being closed. Retry and check `capacity` again.
continue
}
defer pool.capacityMu.RUnlock()

if pool.capacity.Load() < pool.active.Load() {
conn.Close()
pool.closedConn()
return true
}

connSetting := conn.Conn.Setting()
if connSetting == nil {
pool.clean.Push(conn)
} else {
stack := connSetting.bucket & stackMask
pool.settings[stack].Push(conn)
pool.freshSettingsStack.Store(int64(stack))
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder, would an alternative approach work to avoid having to lock? What if we still optimistically put it into the stack, but then check after we've completed that if we have since been closed and then remove it again?

So you only pay the price during closing mostly and then having to do any locking or other synchronization?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm definitely open to alternative approaches. For your suggestion, don't we have to take a lock on capacityMu to correctly decide whether the pool is closed or not?

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can think about optimistic concurrency approach with epoch validation.
ConnPool can have epoch atomic.Uint64. We invalidated with increment whenever there is a change like Close or Capacity.
In tryReturnConn we read the epoch value and then read the capacity. Before returning the connection back to the stack we check the epoch again If it is changed we continue otherwise return to the stack.
After returning we check again and if there is a change we try to get the connection back, if we receive the connection continue again.

We do not expect capacity to change that frequently.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am working on an additional change that uses a pool generation number to allow Close() to return instantly and basically break the link between checked out connections and the pool.

I think that could be used for the optimistic concurrency check. I'll try this out and see if I can make this work.

}
return false
}
connSetting := conn.Conn.Setting()
if connSetting == nil {
pool.clean.Push(conn)
} else {
stack := connSetting.bucket & stackMask
pool.settings[stack].Push(conn)
pool.freshSettingsStack.Store(int64(stack))
}
return false
}

func (pool *ConnPool[C]) pop(stack *connStack[C]) *Pooled[C] {
Expand Down Expand Up @@ -595,7 +617,9 @@ func (pool *ConnPool[C]) get(ctx context.Context) (*Pooled[C], error) {
// to other clients, wait until one of the connections is returned
if conn == nil {
start := time.Now()
conn, err = pool.wait.waitForConn(ctx, nil)
conn, err = pool.wait.waitForConn(ctx, nil, func() bool {
return pool.close == nil || pool.capacity.Load() == 0
})
if err != nil {
return nil, ErrTimeout
}
Expand Down Expand Up @@ -652,7 +676,9 @@ func (pool *ConnPool[C]) getWithSetting(ctx context.Context, setting *Setting) (
// wait for one of them
if conn == nil {
start := time.Now()
conn, err = pool.wait.waitForConn(ctx, setting)
conn, err = pool.wait.waitForConn(ctx, setting, func() bool {
return pool.close == nil || pool.capacity.Load() == 0
})
if err != nil {
return nil, ErrTimeout
}
Expand Down
8 changes: 7 additions & 1 deletion go/pools/smartconnpool/waitlist.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,11 +50,17 @@ type waitlist[C Connection] struct {
// The returned connection may _not_ have the requested Setting. This function can
// also return a `nil` connection even if our context has expired, if the pool has
// forced an expiration of all waiters in the waitlist.
func (wl *waitlist[C]) waitForConn(ctx context.Context, setting *Setting) (*Pooled[C], error) {
func (wl *waitlist[C]) waitForConn(ctx context.Context, setting *Setting, isClosed func() bool) (*Pooled[C], error) {
elem := wl.nodes.Get().(*list.Element[waiter[C]])
elem.Value = waiter[C]{setting: setting, conn: nil, ctx: ctx}

wl.mu.Lock()
if isClosed() {
// if the pool is closed, we can't wait for a connection, so return an error
wl.nodes.Put(elem)
Comment on lines +58 to +60
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can the pool be re-opened between these two calls?

wl.mu.Unlock()
return nil, ErrConnPoolClosed
}
// add ourselves as a waiter at the end of the waitlist
wl.list.PushBackValue(elem)
wl.mu.Unlock()
Expand Down
6 changes: 5 additions & 1 deletion go/pools/smartconnpool/waitlist_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,11 @@ func TestWaitlistExpireWithMultipleWaiters(t *testing.T) {

for i := 0; i < waiterCount; i++ {
go func() {
_, err := wait.waitForConn(ctx, nil)
_, err := wait.waitForConn(ctx, nil, func() bool {
// This function is called to check if the pool is closed.
return ctx.Err() != nil
})

if err != nil {
expireCount.Add(1)
}
Expand Down
Loading