Skip to content

Commit 346bfaf

Browse files
hiduhidumonkey92t
authored
ConnPool check fd for bad conns (#1824)
* conncheck for badconn (#1821) * format imports * fix ut: pool with badconn * fix unstable ut: should facilitate failover * Revert "fix unstable ut: should facilitate failover" This reverts commit c7eeca2. * fix test error Signed-off-by: monkey92t <golang@88.com> Co-authored-by: hidu <duv123+github@gmail.com> Co-authored-by: monkey92t <golang@88.com>
1 parent 62fc2c8 commit 346bfaf

File tree

10 files changed

+204
-18
lines changed

10 files changed

+204
-18
lines changed

‎internal/pool/conncheck.go‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
// +build linux darwin dragonfly freebsd netbsd openbsd solaris illumos
2+
3+
package pool
4+
5+
import (
6+
"errors"
7+
"io"
8+
"net"
9+
"syscall"
10+
)
11+
12+
var errUnexpectedRead = errors.New("unexpected read from socket")
13+
14+
func connCheck(conn net.Conn) error {
15+
sysConn, ok := conn.(syscall.Conn)
16+
if !ok {
17+
return nil
18+
}
19+
rawConn, err := sysConn.SyscallConn()
20+
if err != nil {
21+
return err
22+
}
23+
24+
var sysErr error
25+
err = rawConn.Read(func(fd uintptr) bool {
26+
var buf [1]byte
27+
n, err := syscall.Read(int(fd), buf[:])
28+
switch {
29+
case n == 0 && err == nil:
30+
sysErr = io.EOF
31+
case n > 0:
32+
sysErr = errUnexpectedRead
33+
case err == syscall.EAGAIN || err == syscall.EWOULDBLOCK:
34+
sysErr = nil
35+
default:
36+
sysErr = err
37+
}
38+
return true
39+
})
40+
if err != nil {
41+
return err
42+
}
43+
44+
return sysErr
45+
}

‎internal/pool/conncheck_dummy.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
// +build !linux,!darwin,!dragonfly,!freebsd,!netbsd,!openbsd,!solaris,!illumos
2+
3+
package pool
4+
5+
import "net"
6+
7+
func connCheck(conn net.Conn) error {
8+
return nil
9+
}

‎internal/pool/conncheck_test.go‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
// +build linux darwin dragonfly freebsd netbsd openbsd solaris illumos
2+
3+
package pool
4+
5+
import (
6+
"net"
7+
"net/http/httptest"
8+
"testing"
9+
"time"
10+
)
11+
12+
func Test_connCheck(t *testing.T) {
13+
// tests with real conns
14+
ts := httptest.NewServer(nil)
15+
defer ts.Close()
16+
17+
t.Run("good conn", func(t *testing.T) {
18+
conn, err := net.DialTimeout(ts.Listener.Addr().Network(), ts.Listener.Addr().String(), time.Second)
19+
if err != nil {
20+
t.Fatalf(err.Error())
21+
}
22+
defer conn.Close()
23+
if err = connCheck(conn); err != nil {
24+
t.Fatalf(err.Error())
25+
}
26+
conn.Close()
27+
28+
if err = connCheck(conn); err == nil {
29+
t.Fatalf("expect has error")
30+
}
31+
})
32+
33+
t.Run("bad conn 2", func(t *testing.T) {
34+
conn, err := net.DialTimeout(ts.Listener.Addr().Network(), ts.Listener.Addr().String(), time.Second)
35+
if err != nil {
36+
t.Fatalf(err.Error())
37+
}
38+
defer conn.Close()
39+
40+
ts.Close()
41+
42+
if err = connCheck(conn); err == nil {
43+
t.Fatalf("expect has err")
44+
}
45+
})
46+
}

‎internal/pool/main_test.go‎

Lines changed: 86 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,12 @@ package pool_test
22

33
import (
44
"context"
5+
"fmt"
56
"net"
67
"sync"
8+
"syscall"
79
"testing"
10+
"time"
811

912
. "github.com/onsi/ginkgo"
1013
. "github.com/onsi/gomega"
@@ -32,5 +35,87 @@ func perform(n int, cbs ...func(int)) {
3235
}
3336

3437
func dummyDialer(context.Context) (net.Conn, error) {
35-
return &net.TCPConn{}, nil
38+
// return &net.TCPConn{}, nil
39+
return newDummyConn(), nil
40+
}
41+
42+
func newDummyConn() net.Conn {
43+
return &dummyConn{
44+
rawConn: &dummyRawConn{},
45+
}
46+
}
47+
48+
var _ net.Conn = (*dummyConn)(nil)
49+
var _ syscall.Conn = (*dummyConn)(nil)
50+
51+
type dummyConn struct {
52+
rawConn *dummyRawConn
53+
}
54+
55+
func (d *dummyConn) SyscallConn() (syscall.RawConn, error) {
56+
return d.rawConn, nil
57+
}
58+
59+
var errDummy = fmt.Errorf("dummyConn err")
60+
61+
func (d *dummyConn) Read(b []byte) (n int, err error) {
62+
return 0, errDummy
63+
}
64+
65+
func (d *dummyConn) Write(b []byte) (n int, err error) {
66+
return 0, errDummy
67+
}
68+
69+
func (d *dummyConn) Close() error {
70+
d.rawConn.Close()
71+
return nil
72+
}
73+
74+
func (d *dummyConn) LocalAddr() net.Addr {
75+
return &net.TCPAddr{}
76+
}
77+
78+
func (d *dummyConn) RemoteAddr() net.Addr {
79+
return &net.TCPAddr{}
80+
}
81+
82+
func (d *dummyConn) SetDeadline(t time.Time) error {
83+
return nil
84+
}
85+
86+
func (d *dummyConn) SetReadDeadline(t time.Time) error {
87+
return nil
88+
}
89+
90+
func (d *dummyConn) SetWriteDeadline(t time.Time) error {
91+
return nil
92+
}
93+
94+
var _ syscall.RawConn = (*dummyRawConn)(nil)
95+
96+
type dummyRawConn struct {
97+
closed bool
98+
mux sync.Mutex
99+
}
100+
101+
func (d *dummyRawConn) Control(f func(fd uintptr)) error {
102+
return nil
103+
}
104+
105+
func (d *dummyRawConn) Read(f func(fd uintptr) (done bool)) error {
106+
d.mux.Lock()
107+
defer d.mux.Unlock()
108+
if d.closed {
109+
return fmt.Errorf("dummyRawConn closed")
110+
}
111+
return nil
112+
}
113+
114+
func (d *dummyRawConn) Write(f func(fd uintptr) (done bool)) error {
115+
return nil
116+
}
117+
func (d *dummyRawConn) Close() {
118+
d.mux.Lock()
119+
d.closed = true
120+
d.mux.Unlock()
36121
}

‎internal/pool/pool.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -520,7 +520,7 @@ func (p *ConnPool) reapStaleConn() *Conn {
520520

521521
func (p *ConnPool) isStaleConn(cn *Conn) bool {
522522
if p.opt.IdleTimeout == 0 && p.opt.MaxConnAge == 0 {
523-
return false
523+
return connCheck(cn.netConn) != nil
524524
}
525525

526526
now := time.Now()
@@ -531,5 +531,5 @@ func (p *ConnPool) isStaleConn(cn *Conn) bool {
531531
return true
532532
}
533533

534-
return false
534+
return connCheck(cn.netConn) != nil
535535
}

‎internal/pool/pool_test.go‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,10 @@ import (
66
"testing"
77
"time"
88

9-
"github.com/go-redis/redis/v8/internal/pool"
10-
119
. "github.com/onsi/ginkgo"
1210
. "github.com/onsi/gomega"
11+
12+
"github.com/go-redis/redis/v8/internal/pool"
1313
)
1414

1515
var _ = Describe("ConnPool", func() {
@@ -285,6 +285,8 @@ var _ = Describe("conns reaper", func() {
285285
cn.SetUsedAt(time.Now().Add(-2 * idleTimeout))
286286
case "aged":
287287
cn.SetCreatedAt(time.Now().Add(-2 * maxAge))
288+
case "conncheck":
289+
cn.Close()
288290
}
289291
conns = append(conns, cn)
290292
staleConns = append(staleConns, cn)
@@ -371,6 +373,7 @@ var _ = Describe("conns reaper", func() {
371373

372374
assert("idle")
373375
assert("aged")
376+
assert("conncheck")
374377
})
375378

376379
var _ = Describe("race", func() {

‎main_test.go‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,10 @@ import (
1212
"testing"
1313
"time"
1414

15-
"github.com/go-redis/redis/v8"
16-
1715
. "github.com/onsi/ginkgo"
1816
. "github.com/onsi/gomega"
17+
18+
"github.com/go-redis/redis/v8"
1919
)
2020

2121
const (
@@ -117,7 +117,7 @@ func TestGinkgoSuite(t *testing.T) {
117117
RunSpecs(t, "go-redis")
118118
}
119119

120-
//------------------------------------------------------------------------------
120+
// ------------------------------------------------------------------------------
121121

122122
func redisOptions() *redis.Options {
123123
return &redis.Options{
@@ -364,7 +364,7 @@ func startSentinel(port, masterName, masterPort string) (*redisProcess, error) {
364364
return p, nil
365365
}
366366

367-
//------------------------------------------------------------------------------
367+
// ------------------------------------------------------------------------------
368368

369369
type badConnError string
370370

@@ -409,7 +409,7 @@ func (cn *badConn) Write([]byte) (int, error) {
409409
return 0, badConnError("bad connection")
410410
}
411411

412-
//------------------------------------------------------------------------------
412+
// ------------------------------------------------------------------------------
413413

414414
type hook struct {
415415
beforeProcess func(ctx context.Context, cmd redis.Cmder) (context.Context, error)

‎pool_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,8 +87,9 @@ var _ = Describe("pool", func() {
8787
cn.SetNetConn(&badConn{})
8888
client.Pool().Put(ctx, cn)
8989

90+
// connCheck will automatically remove damaged connections.
9091
err = client.Ping(ctx).Err()
91-
Expect(err).To(MatchError("bad connection"))
92+
Expect(err).NotTo(HaveOccurred())
9293

9394
val, err := client.Ping(ctx).Result()
9495
Expect(err).NotTo(HaveOccurred())

‎sentinel_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -191,7 +191,7 @@ var _ = Describe("NewFailoverClusterClient", func() {
191191
err = master.Shutdown(ctx).Err()
192192
Expect(err).NotTo(HaveOccurred())
193193
Eventually(func() error {
194-
return sentinelMaster.Ping(ctx).Err()
194+
return master.Ping(ctx).Err()
195195
}, "15s", "100ms").Should(HaveOccurred())
196196

197197
// Check that client picked up new master.

‎tx_test.go‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,7 @@ var _ = Describe("Tx", func() {
123123
Expect(num).To(Equal(int64(N)))
124124
})
125125

126-
It("should recover from bad connection", func() {
126+
It("should remove from bad connection", func() {
127127
// Put bad connection in the pool.
128128
cn, err := client.Pool().Get(context.Background())
129129
Expect(err).NotTo(HaveOccurred())
@@ -134,17 +134,14 @@ var _ = Describe("Tx", func() {
134134
do := func() error {
135135
err := client.Watch(ctx, func(tx *redis.Tx) error {
136136
_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
137-
pipe.Ping(ctx)
138-
return nil
137+
return pipe.Ping(ctx).Err()
139138
})
140139
return err
141140
})
142141
return err
143142
}
144143

145-
err = do()
146-
Expect(err).To(MatchError("bad connection"))
147-
144+
// connCheck will automatically remove damaged connections.
148145
err = do()
149146
Expect(err).NotTo(HaveOccurred())
150147
})

0 commit comments

Comments
 (0)