Skip to content

Commit f7b3619

Browse files
hardeyhehardeyhe
authored
hotfix/v8_conn_pool_check_fd (#2431)
1 parent c888488 commit f7b3619

9 files changed

+199
-11
lines changed

‎internal/pool/conncheck.go

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

‎internal/pool/conncheck_dummy.go

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

‎internal/pool/conncheck_test.go

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

‎internal/pool/main_test.go

+86-1
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

+2-2
Original file line numberDiff line numberDiff line change
@@ -542,7 +542,7 @@ func (p *ConnPool) reapStaleConn() *Conn {
542542

543543
func (p *ConnPool) isStaleConn(cn *Conn) bool {
544544
if p.opt.IdleTimeout == 0 && p.opt.MaxConnAge == 0 {
545-
return false
545+
return connCheck(cn.netConn) != nil
546546
}
547547

548548
now := time.Now()
@@ -553,5 +553,5 @@ func (p *ConnPool) isStaleConn(cn *Conn) bool {
553553
return true
554554
}
555555

556-
return false
556+
return connCheck(cn.netConn) != nil
557557
}

‎internal/pool/pool_test.go

+3
Original file line numberDiff line numberDiff line change
@@ -323,6 +323,8 @@ var _ = Describe("conns reaper", func() {
323323
cn.SetUsedAt(time.Now().Add(-2 * idleTimeout))
324324
case "aged":
325325
cn.SetCreatedAt(time.Now().Add(-2 * maxAge))
326+
case "conncheck":
327+
cn.Close()
326328
}
327329
conns = append(conns, cn)
328330
staleConns = append(staleConns, cn)
@@ -409,6 +411,7 @@ var _ = Describe("conns reaper", func() {
409411

410412
assert("idle")
411413
assert("aged")
414+
assert("conncheck")
412415
})
413416

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

‎pool_test.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ var _ = Describe("pool", func() {
8888
client.Pool().Put(ctx, cn)
8989

9090
err = client.Ping(ctx).Err()
91-
Expect(err).To(MatchError("bad connection"))
91+
Expect(err).NotTo(HaveOccurred())
9292

9393
val, err := client.Ping(ctx).Result()
9494
Expect(err).NotTo(HaveOccurred())

‎sentinel_test.go

+1-1
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

+3-6
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)