Vendor update

vendor/gopkg.in/redis.v5/internal/pool/conn.go

@@ -2,56 +2,77 @@ package pool
 
 import (
 	"net"
+	"sync/atomic"
 	"time"
 
 	"gopkg.in/redis.v5/internal/proto"
 )
 
-const defaultBufSize = 4096
-
 var noDeadline = time.Time{}
 
 type Conn struct {
-	NetConn net.Conn
-	Rd      *proto.Reader
-	Wb      *proto.WriteBuffer
+	netConn net.Conn
+
+	Rd *proto.Reader
+	Wb *proto.WriteBuffer
 
 	Inited bool
-	UsedAt time.Time
+	usedAt atomic.Value
 }
 
 func NewConn(netConn net.Conn) *Conn {
 	cn := &Conn{
-		NetConn: netConn,
+		netConn: netConn,
 		Wb:      proto.NewWriteBuffer(),
-
-		UsedAt: time.Now(),
 	}
-	cn.Rd = proto.NewReader(cn.NetConn)
+	cn.Rd = proto.NewReader(cn.netConn)
+	cn.SetUsedAt(time.Now())
 	return cn
 }
 
+func (cn *Conn) UsedAt() time.Time {
+	return cn.usedAt.Load().(time.Time)
+}
+
+func (cn *Conn) SetUsedAt(tm time.Time) {
+	cn.usedAt.Store(tm)
+}
+
+func (cn *Conn) SetNetConn(netConn net.Conn) {
+	cn.netConn = netConn
+	cn.Rd.Reset(netConn)
+}
+
 func (cn *Conn) IsStale(timeout time.Duration) bool {
-	return timeout > 0 && time.Since(cn.UsedAt) > timeout
+	return timeout > 0 && time.Since(cn.UsedAt()) > timeout
 }
 
 func (cn *Conn) SetReadTimeout(timeout time.Duration) error {
-	cn.UsedAt = time.Now()
+	now := time.Now()
+	cn.SetUsedAt(now)
 	if timeout > 0 {
-		return cn.NetConn.SetReadDeadline(cn.UsedAt.Add(timeout))
+		return cn.netConn.SetReadDeadline(now.Add(timeout))
 	}
-	return cn.NetConn.SetReadDeadline(noDeadline)
-
+	return cn.netConn.SetReadDeadline(noDeadline)
 }
 
 func (cn *Conn) SetWriteTimeout(timeout time.Duration) error {
-	cn.UsedAt = time.Now()
+	now := time.Now()
+	cn.SetUsedAt(now)
 	if timeout > 0 {
-		return cn.NetConn.SetWriteDeadline(cn.UsedAt.Add(timeout))
+		return cn.netConn.SetWriteDeadline(now.Add(timeout))
 	}
-	return cn.NetConn.SetWriteDeadline(noDeadline)
+	return cn.netConn.SetWriteDeadline(noDeadline)
+}
+
+func (cn *Conn) Write(b []byte) (int, error) {
+	return cn.netConn.Write(b)
+}
+
+func (cn *Conn) RemoteAddr() net.Addr {
+	return cn.netConn.RemoteAddr()
 }
 
 func (cn *Conn) Close() error {
-	return cn.NetConn.Close()
+	return cn.netConn.Close()
 }

vendor/gopkg.in/redis.v5/internal/pool/pool.go

@@ -19,7 +19,9 @@ var (
 
 var timers = sync.Pool{
 	New: func() interface{} {
-		return time.NewTimer(0)
+		t := time.NewTimer(time.Hour)
+		t.Stop()
+		return t
 	},
 }
 
@@ -41,7 +43,6 @@ type Pooler interface {
 	FreeLen() int
 	Stats() *Stats
 	Close() error
-	Closed() bool
 }
 
 type dialer func() (net.Conn, error)
@@ -96,12 +97,13 @@ func (p *ConnPool) NewConn() (*Conn, error) {
 
 func (p *ConnPool) PopFree() *Conn {
 	timer := timers.Get().(*time.Timer)
-	if !timer.Reset(p.poolTimeout) {
-		<-timer.C
-	}
+	timer.Reset(p.poolTimeout)
 
 	select {
 	case p.queue <- struct{}{}:
+		if !timer.Stop() {
+			<-timer.C
+		}
 		timers.Put(timer)
 	case <-timer.C:
 		timers.Put(timer)
@@ -132,19 +134,20 @@ func (p *ConnPool) popFree() *Conn {
 
 // Get returns existed connection from the pool or creates a new one.
 func (p *ConnPool) Get() (*Conn, bool, error) {
-	if p.Closed() {
+	if p.closed() {
 		return nil, false, ErrClosed
 	}
 
 	atomic.AddUint32(&p.stats.Requests, 1)
 
 	timer := timers.Get().(*time.Timer)
-	if !timer.Reset(p.poolTimeout) {
-		<-timer.C
-	}
+	timer.Reset(p.poolTimeout)
 
 	select {
 	case p.queue <- struct{}{}:
+		if !timer.Stop() {
+			<-timer.C
+		}
 		timers.Put(timer)
 	case <-timer.C:
 		timers.Put(timer)
@@ -241,7 +244,7 @@ func (p *ConnPool) Stats() *Stats {
 	}
 }
 
-func (p *ConnPool) Closed() bool {
+func (p *ConnPool) closed() bool {
 	return atomic.LoadInt32(&p._closed) == 1
 }
 
@@ -318,7 +321,7 @@ func (p *ConnPool) reaper(frequency time.Duration) {
 	defer ticker.Stop()
 
 	for _ = range ticker.C {
-		if p.Closed() {
+		if p.closed() {
 			break
 		}
 		n, err := p.ReapStaleConns()

vendor/gopkg.in/redis.v5/internal/pool/pool_single.go

@@ -12,10 +12,6 @@ func NewSingleConnPool(cn *Conn) *SingleConnPool {
 	}
 }
 
-func (p *SingleConnPool) First() *Conn {
-	return p.cn
-}
-
 func (p *SingleConnPool) Get() (*Conn, bool, error) {
 	return p.cn, false, nil
 }
@@ -49,7 +45,3 @@ func (p *SingleConnPool) Stats() *Stats {
 func (p *SingleConnPool) Close() error {
 	return nil
 }
-
-func (p *SingleConnPool) Closed() bool {
-	return false
-}

vendor/gopkg.in/redis.v5/internal/pool/pool_sticky.go

@@ -11,7 +11,7 @@ type StickyConnPool struct {
 
 	cn     *Conn
 	closed bool
-	mx     sync.Mutex
+	mu     sync.Mutex
 }
 
 var _ Pooler = (*StickyConnPool)(nil)
@@ -23,16 +23,9 @@ func NewStickyConnPool(pool *ConnPool, reusable bool) *StickyConnPool {
 	}
 }
 
-func (p *StickyConnPool) First() *Conn {
-	p.mx.Lock()
-	cn := p.cn
-	p.mx.Unlock()
-	return cn
-}
-
 func (p *StickyConnPool) Get() (*Conn, bool, error) {
-	defer p.mx.Unlock()
-	p.mx.Lock()
+	p.mu.Lock()
+	defer p.mu.Unlock()
 
 	if p.closed {
 		return nil, false, ErrClosed
@@ -56,14 +49,12 @@ func (p *StickyConnPool) putUpstream() (err error) {
 }
 
 func (p *StickyConnPool) Put(cn *Conn) error {
-	defer p.mx.Unlock()
-	p.mx.Lock()
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if p.closed {
 		return ErrClosed
 	}
-	if p.cn != cn {
-		panic("p.cn != cn")
-	}
 	return nil
 }
 
@@ -74,23 +65,19 @@ func (p *StickyConnPool) removeUpstream(reason error) error {
 }
 
 func (p *StickyConnPool) Remove(cn *Conn, reason error) error {
-	defer p.mx.Unlock()
-	p.mx.Lock()
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if p.closed {
 		return nil
 	}
-	if p.cn == nil {
-		panic("p.cn == nil")
-	}
-	if cn != nil && p.cn != cn {
-		panic("p.cn != cn")
-	}
 	return p.removeUpstream(reason)
 }
 
 func (p *StickyConnPool) Len() int {
-	defer p.mx.Unlock()
-	p.mx.Lock()
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if p.cn == nil {
 		return 0
 	}
@@ -98,8 +85,9 @@ func (p *StickyConnPool) Len() int {
 }
 
 func (p *StickyConnPool) FreeLen() int {
-	defer p.mx.Unlock()
-	p.mx.Lock()
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if p.cn == nil {
 		return 1
 	}
@@ -111,8 +99,9 @@ func (p *StickyConnPool) Stats() *Stats {
 }
 
 func (p *StickyConnPool) Close() error {
-	defer p.mx.Unlock()
-	p.mx.Lock()
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	if p.closed {
 		return ErrClosed
 	}
@@ -128,10 +117,3 @@ func (p *StickyConnPool) Close() error {
 	}
 	return err
 }
-
-func (p *StickyConnPool) Closed() bool {
-	p.mx.Lock()
-	closed := p.closed
-	p.mx.Unlock()
-	return closed
-}

vendor/gopkg.in/redis.v5/internal/proto/reader.go

@@ -29,10 +29,14 @@ type Reader struct {
 func NewReader(rd io.Reader) *Reader {
 	return &Reader{
 		src: bufio.NewReader(rd),
-		buf: make([]byte, 0, bufferSize),
+		buf: make([]byte, 4096),
 	}
 }
 
+func (r *Reader) Reset(rd io.Reader) {
+	r.src.Reset(rd)
+}
+
 func (p *Reader) PeekBuffered() []byte {
 	if n := p.src.Buffered(); n != 0 {
 		b, _ := p.src.Peek(n)
@@ -42,7 +46,12 @@ func (p *Reader) PeekBuffered() []byte {
 }
 
 func (p *Reader) ReadN(n int) ([]byte, error) {
-	return readN(p.src, p.buf, n)
+	b, err := readN(p.src, p.buf, n)
+	if err != nil {
+		return nil, err
+	}
+	p.buf = b
+	return b, nil
 }
 
 func (p *Reader) ReadLine() ([]byte, error) {
@@ -72,11 +81,11 @@ func (p *Reader) ReadReply(m MultiBulkParse) (interface{}, error) {
 	case ErrorReply:
 		return nil, ParseErrorReply(line)
 	case StatusReply:
-		return parseStatusValue(line)
+		return parseStatusValue(line), nil
 	case IntReply:
 		return parseInt(line[1:], 10, 64)
 	case StringReply:
-		return p.readBytesValue(line)
+		return p.readTmpBytesValue(line)
 	case ArrayReply:
 		n, err := parseArrayLen(line)
 		if err != nil {
@@ -111,9 +120,9 @@ func (p *Reader) ReadTmpBytesReply() ([]byte, error) {
 	case ErrorReply:
 		return nil, ParseErrorReply(line)
 	case StringReply:
-		return p.readBytesValue(line)
+		return p.readTmpBytesValue(line)
 	case StatusReply:
-		return parseStatusValue(line)
+		return parseStatusValue(line), nil
 	default:
 		return nil, fmt.Errorf("redis: can't parse string reply: %.100q", line)
 	}
@@ -210,7 +219,7 @@ func (p *Reader) ReadScanReply() ([]string, uint64, error) {
 	return keys, cursor, err
 }
 
-func (p *Reader) readBytesValue(line []byte) ([]byte, error) {
+func (p *Reader) readTmpBytesValue(line []byte) ([]byte, error) {
 	if isNilReply(line) {
 		return nil, internal.Nil
 	}
@@ -297,8 +306,8 @@ func ParseErrorReply(line []byte) error {
 	return internal.RedisError(string(line[1:]))
 }
 
-func parseStatusValue(line []byte) ([]byte, error) {
-	return line[1:], nil
+func parseStatusValue(line []byte) []byte {
+	return line[1:]
 }
 
 func parseArrayLen(line []byte) (int64, error) {

vendor/gopkg.in/redis.v5/internal/proto/scan.go

@@ -3,6 +3,7 @@ package proto
 import (
 	"encoding"
 	"fmt"
+	"reflect"
 
 	"gopkg.in/redis.v5/internal"
 )
@@ -105,3 +106,26 @@ func Scan(b []byte, v interface{}) error {
 			"redis: can't unmarshal %T (consider implementing BinaryUnmarshaler)", v)
 	}
 }
+
+func ScanSlice(data []string, slice interface{}) error {
+	v := reflect.ValueOf(slice)
+	if !v.IsValid() {
+		return fmt.Errorf("redis: ScanSlice(nil)")
+	}
+	if v.Kind() != reflect.Ptr {
+		return fmt.Errorf("redis: ScanSlice(non-pointer %T)", slice)
+	}
+	v = v.Elem()
+	if v.Kind() != reflect.Slice {
+		return fmt.Errorf("redis: ScanSlice(non-slice %T)", slice)
+	}
+
+	for i, s := range data {
+		elem := internal.SliceNextElem(v)
+		if err := Scan([]byte(s), elem.Addr().Interface()); err != nil {
+			return fmt.Errorf("redis: ScanSlice(index=%d value=%q) failed: %s", i, s, err)
+		}
+	}
+
+	return nil
+}

vendor/gopkg.in/redis.v5/internal/proto/write_buffer.go

@@ -8,11 +8,13 @@ import (
 
 const bufferSize = 4096
 
-type WriteBuffer struct{ b []byte }
+type WriteBuffer struct {
+	b []byte
+}
 
 func NewWriteBuffer() *WriteBuffer {
 	return &WriteBuffer{
-		b: make([]byte, 0, bufferSize),
+		b: make([]byte, 0, 4096),
 	}
 }
 

vendor/gopkg.in/redis.v5/internal/util.go

@@ -1,5 +1,7 @@
 package internal
 
+import "reflect"
+
 func ToLower(s string) string {
 	if isLower(s) {
 		return s
@@ -25,3 +27,21 @@ func isLower(s string) bool {
 	}
 	return true
 }
+
+func SliceNextElem(v reflect.Value) reflect.Value {
+	if v.Len() < v.Cap() {
+		v.Set(v.Slice(0, v.Len()+1))
+		return v.Index(v.Len() - 1)
+	}
+
+	elemType := v.Type().Elem()
+
+	if elemType.Kind() == reflect.Ptr {
+		elem := reflect.New(elemType.Elem())
+		v.Set(reflect.Append(v, elem))
+		return elem.Elem()
+	}
+
+	v.Set(reflect.Append(v, reflect.Zero(elemType)))
+	return v.Index(v.Len() - 1)
+}