Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added options for configuring removal of idle connections #674

Merged
merged 5 commits into from
Nov 26, 2024
Merged

Added options for configuring removal of idle connections #674

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e7e11c6
Added options for configuring removal of idle connections
Nov 16, 2024
f5a1ee8
Removing idle connections in a separate goroutine
Nov 17, 2024
3646ba0
Removed lastUsage and stopChan. Used AfterFunc
Nov 18, 2024
a936dfe
Start timer only when number of idle wires is greater than minSize
Nov 21, 2024
e903089
Improvements after review
Nov 25, 2024
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
10 changes: 8 additions & 2 deletions mux.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,8 +99,14 @@ func newMux(dst string, option *ClientOption, init, dead wire, wireFn wireFn, wi
for i := 0; i < len(m.wire); i++ {
m.wire[i].Store(init)
}
m.dpool = newPool(option.BlockingPoolSize, dead, wireFn)
m.spool = newPool(option.BlockingPoolSize, dead, wireNoBgFn)

var idleConnTTL *time.Duration
if option.IdleConnTTL != 0 {
idleConnTTL = &option.IdleConnTTL
}

m.dpool = newPool(option.BlockingPoolSize, dead, idleConnTTL, option.BlockingPoolMinSize, wireFn)
m.spool = newPool(option.BlockingPoolSize, dead, idleConnTTL, option.BlockingPoolMinSize, wireNoBgFn)
return m
}

Expand Down
30 changes: 15 additions & 15 deletions pipe_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1005,7 +1005,7 @@ func TestDoStreamRecycle(t *testing.T) {
go func() {
mock.Expect("PING").ReplyString("OK")
}()
conns := newPool(1, nil, nil)
conns := newPool(1, nil, nil, 0, nil)
s := p.DoStream(context.Background(), conns, cmds.NewCompleted([]string{"PING"}))
buf := bytes.NewBuffer(nil)
if err := s.Error(); err != nil {
Expand Down Expand Up @@ -1058,7 +1058,7 @@ func TestDoStreamRecycleDestinationFull(t *testing.T) {
go func() {
mock.Expect("PING").ReplyBlobString("OK")
}()
conns := newPool(1, nil, nil)
conns := newPool(1, nil, nil, 0, nil)
s := p.DoStream(context.Background(), conns, cmds.NewCompleted([]string{"PING"}))
buf := &limitedbuffer{buf: make([]byte, 1)}
if err := s.Error(); err != nil {
Expand Down Expand Up @@ -1091,7 +1091,7 @@ func TestDoMultiStreamRecycle(t *testing.T) {
go func() {
mock.Expect("PING").Expect("PING").ReplyString("OK").ReplyString("OK")
}()
conns := newPool(1, nil, nil)
conns := newPool(1, nil, nil, 0, nil)
s := p.DoMultiStream(context.Background(), conns, cmds.NewCompleted([]string{"PING"}), cmds.NewCompleted([]string{"PING"}))
buf := bytes.NewBuffer(nil)
if err := s.Error(); err != nil {
Expand Down Expand Up @@ -1124,7 +1124,7 @@ func TestDoMultiStreamRecycleDestinationFull(t *testing.T) {
go func() {
mock.Expect("PING").Expect("PING").ReplyBlobString("OK").ReplyBlobString("OK")
}()
conns := newPool(1, nil, nil)
conns := newPool(1, nil, nil, 0, nil)
s := p.DoMultiStream(context.Background(), conns, cmds.NewCompleted([]string{"PING"}), cmds.NewCompleted([]string{"PING"}))
buf := &limitedbuffer{buf: make([]byte, 1)}
if err := s.Error(); err != nil {
Expand Down Expand Up @@ -3569,7 +3569,7 @@ func TestAlreadyCanceledContext(t *testing.T) {
t.Fatalf("unexpected err %v", err)
}

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)
if s := p.DoStream(ctx, cp, cmds.NewCompleted([]string{"GET", "a"})); !errors.Is(s.Error(), context.Canceled) {
t.Fatalf("unexpected err %v", s.Error())
}
Expand Down Expand Up @@ -3614,7 +3614,7 @@ func TestCancelContext_DoStream(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
defer cancel()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)
s := p.DoStream(ctx, cp, cmds.NewCompleted([]string{"GET", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
t.Fatalf("unexpected err %v", err)
Expand All @@ -3631,7 +3631,7 @@ func TestWriteDeadlineIsShorterThanContextDeadline_DoStream(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)
startTime := time.Now()
s := p.DoStream(ctx, cp, cmds.NewCompleted([]string{"GET", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
Expand All @@ -3652,7 +3652,7 @@ func TestWriteDeadlineIsNoShorterThanContextDeadline_DoStreamBlocked(t *testing.
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)
startTime := time.Now()
s := p.DoStream(ctx, cp, cmds.NewBlockingCompleted([]string{"BLPOP", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
Expand Down Expand Up @@ -3727,7 +3727,7 @@ func TestCancelContext_DoMultiStream(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*50)
defer cancel()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)
s := p.DoMultiStream(ctx, cp, cmds.NewCompleted([]string{"GET", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
t.Fatalf("unexpected err %v", err)
Expand All @@ -3744,7 +3744,7 @@ func TestWriteDeadlineIsShorterThanContextDeadline_DoMultiStream(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)
startTime := time.Now()
s := p.DoMultiStream(ctx, cp, cmds.NewCompleted([]string{"GET", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
Expand All @@ -3765,7 +3765,7 @@ func TestWriteDeadlineIsNoShorterThanContextDeadline_DoMultiStreamBlocked(t *tes
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)
startTime := time.Now()
s := p.DoMultiStream(ctx, cp, cmds.NewBlockingCompleted([]string{"BLPOP", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
Expand Down Expand Up @@ -3797,7 +3797,7 @@ func TestTimeout_DoStream(t *testing.T) {
defer ShouldNotLeaked(SetupLeakDetection())
p, _, _, _ := setup(t, ClientOption{ConnWriteTimeout: time.Millisecond * 30})

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)

s := p.DoStream(context.Background(), cp, cmds.NewCompleted([]string{"GET", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
Expand All @@ -3817,7 +3817,7 @@ func TestForceClose_DoStream_Block(t *testing.T) {
p.Close()
}()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)

s := p.DoStream(context.Background(), cp, cmds.NewBlockingCompleted([]string{"GET", "a"}))
if s.Error() != nil {
Expand Down Expand Up @@ -3874,7 +3874,7 @@ func TestTimeout_DoMultiStream(t *testing.T) {
defer ShouldNotLeaked(SetupLeakDetection())
p, _, _, _ := setup(t, ClientOption{ConnWriteTimeout: time.Millisecond * 30})

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)

s := p.DoMultiStream(context.Background(), cp, cmds.NewCompleted([]string{"GET", "a"}))
if err := s.Error(); err != io.EOF && !strings.Contains(err.Error(), "i/o") {
Expand All @@ -3894,7 +3894,7 @@ func TestForceClose_DoMultiStream_Block(t *testing.T) {
p.Close()
}()

cp := newPool(1, nil, nil)
cp := newPool(1, nil, nil, 0, nil)

s := p.DoMultiStream(context.Background(), cp, cmds.NewBlockingCompleted([]string{"GET", "a"}))
if s.Error() != nil {
Expand Down
100 changes: 84 additions & 16 deletions pool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,30 +1,47 @@
package rueidis

import "sync"
import (
"sync"
"time"
)

func newPool(cap int, dead wire, makeFn func() wire) *pool {
func newPool(cap int, dead wire, idleConnTTL *time.Duration, minSize int, makeFn func() wire) *pool {
if cap <= 0 {
cap = DefaultPoolSize
}

return &pool{
size: 0,
cap: cap,
dead: dead,
make: makeFn,
list: make([]wire, 0, 4),
cond: sync.NewCond(&sync.Mutex{}),
p := &pool{
size: 0,
minSize: minSize,
cap: cap,
dead: dead,
make: makeFn,
list: make([]wire, 0, 4),
cond: sync.NewCond(&sync.Mutex{}),
idleConnTTL: idleConnTTL,
lastUsage: make(map[wire]time.Time),
stopChan: make(chan struct{}, 1),
stopChanOnce: &sync.Once{},
}

go p.removeIdleConns()

return p
}

type pool struct {
dead wire
cond *sync.Cond
make func() wire
list []wire
size int
cap int
down bool
dead wire
cond *sync.Cond
make func() wire
list []wire
size int
minSize int
cap int
down bool
idleConnTTL *time.Duration
stopChan chan struct{}
stopChanOnce *sync.Once
Copy link
Collaborator

Choose a reason for hiding this comment

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

Suggested change
stopChan chan struct{}
stopChanOnce *sync.Once
timer *time.Timer

Can we just use time.AfterFunc and store the timer here? I think a timer is convenient for us because we can use .Stop() when len(p.list) <= minSize and .Reset() when len(p.list) > minSize.

lastUsage map[wire]time.Time
Copy link
Collaborator

Choose a reason for hiding this comment

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

I am wondering if we need to keep track of idle time for each connection. Can we just clear every connection in the p.list that exceeds minSize whenever the timer is fired?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Sound like a good idea. Made an update in PR

}

func (p *pool) Acquire() (v wire) {
Expand All @@ -41,6 +58,7 @@ func (p *pool) Acquire() (v wire) {
i := len(p.list) - 1
v = p.list[i]
p.list = p.list[:i]
delete(p.lastUsage, v)
}
p.cond.L.Unlock()
return v
Expand All @@ -50,6 +68,10 @@ func (p *pool) Store(v wire) {
p.cond.L.Lock()
if !p.down && v.Error() == nil {
p.list = append(p.list, v)

if p.idleConnTTL != nil {
p.lastUsage[v] = time.Now()
}
} else {
p.size--
v.Close()
Expand All @@ -60,10 +82,56 @@ func (p *pool) Store(v wire) {

func (p *pool) Close() {
p.cond.L.Lock()
p.stopChanOnce.Do(func() {
p.stopChan <- struct{}{}
})
p.down = true
for _, w := range p.list {
w.Close()
}
p.cond.L.Unlock()
p.cond.Broadcast()
}

func (p *pool) removeIdleConns() {
if p.idleConnTTL == nil {
return
}

ttl := *p.idleConnTTL
ticker := time.NewTicker(ttl)

for {
select {
case <-p.stopChan:
ticker.Stop()
return

case <-ticker.C:
p.cond.L.Lock()
if p.down {
p.cond.L.Unlock()
return
}

for p.size > p.minSize && len(p.list) > 0 {
w := p.list[0]
lastUsageTime, ok := p.lastUsage[w]
if !ok {
break
}

if time.Since(lastUsageTime) <= ttl {
break
}

w.Close()
p.size--
p.list = p.list[1:]
delete(p.lastUsage, w)
}

p.cond.L.Unlock()
}
}
}
Loading