Skip to content

Commit

Permalink
Make free-connection stack a LIFO.
Browse files Browse the repository at this point in the history
  • Loading branch information
dim authored and vmihailenco committed Mar 12, 2016
1 parent 7a4f8ed commit 1bb55e3
Show file tree
Hide file tree
Showing 3 changed files with 103 additions and 37 deletions.
4 changes: 4 additions & 0 deletions internal/pool/conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,10 @@ func NewConn(netConn net.Conn) *Conn {
return cn
}

func (cn *Conn) IsStale(timeout time.Duration) bool {
return timeout > 0 && time.Since(cn.UsedAt) > timeout
}

func (cn *Conn) SetNetConn(netConn net.Conn) {
cn.netConn = netConn
cn.UsedAt = time.Now()
Expand Down
72 changes: 72 additions & 0 deletions internal/pool/conn_stack.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
package pool

import (
"sync"
"time"
)

// connStack is used as a LIFO to maintain free connections
type connStack struct {
cns []*Conn
free chan struct{}
mu sync.Mutex
}

func newConnStack(max int) *connStack {
return &connStack{
cns: make([]*Conn, 0, max),
free: make(chan struct{}, max),
}
}

func (s *connStack) Len() int { return len(s.free) }

func (s *connStack) Push(cn *Conn) {
s.mu.Lock()
s.cns = append(s.cns, cn)
s.mu.Unlock()
s.free <- struct{}{}
}

func (s *connStack) ShiftStale(timeout time.Duration) *Conn {
select {
case <-s.free:
s.mu.Lock()
defer s.mu.Unlock()

if cn := s.cns[0]; cn.IsStale(timeout) {
copy(s.cns, s.cns[1:])
s.cns = s.cns[:len(s.cns)-1]
return cn
}
return nil
default:
return nil
}
}

func (s *connStack) Pop() *Conn {
select {
case <-s.free:
return s.pop()
default:
return nil
}
}

func (s *connStack) PopWithTimeout(d time.Duration) *Conn {
select {
case <-s.free:
return s.pop()
case <-time.After(d):
return nil
}
}

func (s *connStack) pop() (cn *Conn) {
s.mu.Lock()
ci := len(s.cns) - 1
cn, s.cns = s.cns[ci], s.cns[:ci]
s.mu.Unlock()
return
}
64 changes: 27 additions & 37 deletions internal/pool/pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ type ConnPool struct {
idleTimeout time.Duration

conns *connList
freeConns chan *Conn
freeConns *connStack
stats PoolStats

_closed int32
Expand All @@ -67,7 +67,7 @@ func NewConnPool(dial dialer, poolSize int, poolTimeout, idleTimeout time.Durati
idleTimeout: idleTimeout,

conns: newConnList(poolSize),
freeConns: make(chan *Conn, poolSize),
freeConns: newConnStack(poolSize),
}
if idleTimeout > 0 {
go p.reaper()
Expand All @@ -87,44 +87,33 @@ func (p *ConnPool) isIdle(cn *Conn) bool {
// there are no connections.
func (p *ConnPool) First() *Conn {
for {
select {
case cn := <-p.freeConns:
if p.isIdle(cn) {
var err error
cn, err = p.replace(cn)
if err != nil {
Logger.Printf("pool.replace failed: %s", err)
continue
}
cn := p.freeConns.Pop()
if cn != nil && cn.IsStale(p.idleTimeout) {
var err error
cn, err = p.replace(cn)
if err != nil {
Logger.Printf("pool.replace failed: %s", err)
continue
}
return cn
default:
return nil
}
return cn
}
panic("not reached")
}

// wait waits for free non-idle connection. It returns nil on timeout.
func (p *ConnPool) wait() *Conn {
deadline := time.After(p.poolTimeout)
for {
select {
case cn := <-p.freeConns:
if p.isIdle(cn) {
var err error
cn, err = p.replace(cn)
if err != nil {
Logger.Printf("pool.replace failed: %s", err)
continue
}
cn := p.freeConns.PopWithTimeout(p.poolTimeout)
if cn != nil && cn.IsStale(p.idleTimeout) {
var err error
cn, err = p.replace(cn)
if err != nil {
Logger.Printf("pool.replace failed: %s", err)
continue
}
return cn
case <-deadline:
return nil
}
return cn
}
panic("not reached")
}

func (p *ConnPool) dial() (net.Conn, error) {
Expand Down Expand Up @@ -198,7 +187,7 @@ func (p *ConnPool) Put(cn *Conn) error {
Logger.Print(err)
return p.Replace(cn, err)
}
p.freeConns <- cn
p.freeConns.Push(cn)
return nil
}

Expand All @@ -223,7 +212,7 @@ func (p *ConnPool) Replace(cn *Conn, reason error) error {
if err != nil {
return err
}
p.freeConns <- newcn
p.freeConns.Push(newcn)
return nil
}

Expand All @@ -234,7 +223,7 @@ func (p *ConnPool) Len() int {

// FreeLen returns number of free connections.
func (p *ConnPool) FreeLen() int {
return len(p.freeConns)
return p.freeConns.Len()
}

func (p *ConnPool) Stats() *PoolStats {
Expand Down Expand Up @@ -273,11 +262,12 @@ func (p *ConnPool) reaper() {
break
}

// pool.First removes idle connections from the pool and
// returns first non-idle connection. So just put returned
// connection back.
if cn := p.First(); cn != nil {
p.Put(cn)
for {
cn := p.freeConns.ShiftStale(p.idleTimeout)
if cn == nil {
break
}
_ = p.conns.Remove(cn)
}
}
}
Expand Down

0 comments on commit 1bb55e3

Please sign in to comment.