forked from mirror/redis
Fix sporadic pool timeouts with IdleTimeout != 0. Fixes #195.
This commit is contained in:
parent
22ed1e8f5b
commit
fb44c891dd
|
@ -146,6 +146,7 @@ func (c *ClusterClient) process(cmd Cmder) {
|
||||||
pipe.Process(NewCmd("ASKING"))
|
pipe.Process(NewCmd("ASKING"))
|
||||||
pipe.Process(cmd)
|
pipe.Process(cmd)
|
||||||
_, _ = pipe.Exec()
|
_, _ = pipe.Exec()
|
||||||
|
pipe.Close()
|
||||||
ask = false
|
ask = false
|
||||||
} else {
|
} else {
|
||||||
client.Process(cmd)
|
client.Process(cmd)
|
||||||
|
|
32
pool.go
32
pool.go
|
@ -163,9 +163,13 @@ func (p *connPool) First() *conn {
|
||||||
select {
|
select {
|
||||||
case cn := <-p.freeConns:
|
case cn := <-p.freeConns:
|
||||||
if p.isIdle(cn) {
|
if p.isIdle(cn) {
|
||||||
p.conns.Remove(cn)
|
var err error
|
||||||
|
cn, err = p.replace(cn)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("redis: replace failed: %s", err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
}
|
||||||
return cn
|
return cn
|
||||||
default:
|
default:
|
||||||
return nil
|
return nil
|
||||||
|
@ -181,9 +185,13 @@ func (p *connPool) wait() *conn {
|
||||||
select {
|
select {
|
||||||
case cn := <-p.freeConns:
|
case cn := <-p.freeConns:
|
||||||
if p.isIdle(cn) {
|
if p.isIdle(cn) {
|
||||||
p.Remove(cn)
|
var err error
|
||||||
|
cn, err = p.replace(cn)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("redis: replace failed: %s", err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
}
|
||||||
return cn
|
return cn
|
||||||
case <-deadline:
|
case <-deadline:
|
||||||
return nil
|
return nil
|
||||||
|
@ -257,17 +265,25 @@ func (p *connPool) Put(cn *conn) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *connPool) Remove(cn *conn) error {
|
func (p *connPool) replace(cn *conn) (*conn, error) {
|
||||||
// Replace existing connection with new one and unblock waiter.
|
|
||||||
newcn, err := p.new()
|
newcn, err := p.new()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("redis: new failed: %s", err)
|
_ = p.conns.Remove(cn)
|
||||||
return p.conns.Remove(cn)
|
return nil, err
|
||||||
}
|
}
|
||||||
err = p.conns.Replace(cn, newcn)
|
_ = p.conns.Replace(cn, newcn)
|
||||||
p.freeConns <- newcn
|
return newcn, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *connPool) Remove(cn *conn) error {
|
||||||
|
// Replace existing connection with new one and unblock waiter.
|
||||||
|
newcn, err := p.replace(cn)
|
||||||
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
p.freeConns <- newcn
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Len returns total number of connections.
|
// Len returns total number of connections.
|
||||||
func (p *connPool) Len() int {
|
func (p *connPool) Len() int {
|
||||||
|
|
Loading…
Reference in New Issue