From f22b38daba9006cefe120c8b7fc64ad7e6069105 Mon Sep 17 00:00:00 2001 From: hidu Date: Thu, 15 Jul 2021 22:18:31 +0800 Subject: [PATCH] conncheck for badconn (#1821) --- internal/pool/conncheck.go | 45 +++++++++++++++ internal/pool/conncheck_dummy.go | 9 +++ internal/pool/conncheck_test.go | 46 ++++++++++++++++ internal/pool/main_test.go | 94 ++++++++++++++++++++++++++++++-- internal/pool/pool.go | 4 +- internal/pool/pool_test.go | 8 ++- 6 files changed, 196 insertions(+), 10 deletions(-) create mode 100644 internal/pool/conncheck.go create mode 100644 internal/pool/conncheck_dummy.go create mode 100644 internal/pool/conncheck_test.go diff --git a/internal/pool/conncheck.go b/internal/pool/conncheck.go new file mode 100644 index 0000000..5dd60df --- /dev/null +++ b/internal/pool/conncheck.go @@ -0,0 +1,45 @@ +// +build linux darwin dragonfly freebsd netbsd openbsd solaris illumos + +package pool + +import ( + "errors" + "io" + "net" + "syscall" +) + +var errUnexpectedRead = errors.New("unexpected read from socket") + +func connCheck(conn net.Conn) error { + sysConn, ok := conn.(syscall.Conn) + if !ok { + return nil + } + rawConn, err := sysConn.SyscallConn() + if err != nil { + return err + } + + var sysErr error + err = rawConn.Read(func(fd uintptr) bool { + var buf [1]byte + n, err := syscall.Read(int(fd), buf[:]) + switch { + case n == 0 && err == nil: + sysErr = io.EOF + case n > 0: + sysErr = errUnexpectedRead + case err == syscall.EAGAIN || err == syscall.EWOULDBLOCK: + sysErr = nil + default: + sysErr = err + } + return true + }) + if err != nil { + return err + } + + return sysErr +} diff --git a/internal/pool/conncheck_dummy.go b/internal/pool/conncheck_dummy.go new file mode 100644 index 0000000..1daf986 --- /dev/null +++ b/internal/pool/conncheck_dummy.go @@ -0,0 +1,9 @@ +// +build !linux,!darwin,!dragonfly,!freebsd,!netbsd,!openbsd,!solaris,!illumos + +package pool + +import "net" + +func connCheck(conn net.Conn) error { + return nil +} diff --git a/internal/pool/conncheck_test.go b/internal/pool/conncheck_test.go new file mode 100644 index 0000000..0332174 --- /dev/null +++ b/internal/pool/conncheck_test.go @@ -0,0 +1,46 @@ +// +build linux darwin dragonfly freebsd netbsd openbsd solaris illumos + +package pool + +import ( + "net" + "net/http/httptest" + "testing" + "time" +) + +func Test_connCheck(t *testing.T) { + // tests with real conns + ts := httptest.NewServer(nil) + defer ts.Close() + + t.Run("good conn", func(t *testing.T) { + conn, err := net.DialTimeout(ts.Listener.Addr().Network(), ts.Listener.Addr().String(), time.Second) + if err != nil { + t.Fatalf(err.Error()) + } + defer conn.Close() + if err = connCheck(conn); err != nil { + t.Fatalf(err.Error()) + } + conn.Close() + + if err = connCheck(conn); err == nil { + t.Fatalf("expect has error") + } + }) + + t.Run("bad conn 2", func(t *testing.T) { + conn, err := net.DialTimeout(ts.Listener.Addr().Network(), ts.Listener.Addr().String(), time.Second) + if err != nil { + t.Fatalf(err.Error()) + } + defer conn.Close() + + ts.Close() + + if err = connCheck(conn); err == nil { + t.Fatalf("expect has err") + } + }) +} diff --git a/internal/pool/main_test.go b/internal/pool/main_test.go index 2365dbc..1a57ed1 100644 --- a/internal/pool/main_test.go +++ b/internal/pool/main_test.go @@ -2,12 +2,14 @@ package pool_test import ( "context" - "net" - "sync" - "testing" - + "fmt" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" + "net" + "sync" + "syscall" + "testing" + "time" ) func TestGinkgoSuite(t *testing.T) { @@ -32,5 +34,87 @@ func perform(n int, cbs ...func(int)) { } func dummyDialer(context.Context) (net.Conn, error) { - return &net.TCPConn{}, nil + // return &net.TCPConn{}, nil + return newDummyConn(), nil +} + +func newDummyConn() net.Conn { + return &dummyConn{ + rawConn: &dummyRawConn{}, + } +} + +var _ net.Conn = (*dummyConn)(nil) +var _ syscall.Conn = (*dummyConn)(nil) + +type dummyConn struct { + rawConn *dummyRawConn +} + +func (d *dummyConn) SyscallConn() (syscall.RawConn, error) { + return d.rawConn, nil +} + +var errDummy = fmt.Errorf("dummyConn err") + +func (d *dummyConn) Read(b []byte) (n int, err error) { + return 0, errDummy +} + +func (d *dummyConn) Write(b []byte) (n int, err error) { + return 0, errDummy +} + +func (d *dummyConn) Close() error { + d.rawConn.Close() + return nil +} + +func (d *dummyConn) LocalAddr() net.Addr { + return &net.TCPAddr{} +} + +func (d *dummyConn) RemoteAddr() net.Addr { + return &net.TCPAddr{} +} + +func (d *dummyConn) SetDeadline(t time.Time) error { + return nil +} + +func (d *dummyConn) SetReadDeadline(t time.Time) error { + return nil +} + +func (d *dummyConn) SetWriteDeadline(t time.Time) error { + return nil +} + +var _ syscall.RawConn = (*dummyRawConn)(nil) + +type dummyRawConn struct { + closed bool + mux sync.Mutex +} + +func (d *dummyRawConn) Control(f func(fd uintptr)) error { + return nil +} + +func (d *dummyRawConn) Read(f func(fd uintptr) (done bool)) error { + d.mux.Lock() + defer d.mux.Unlock() + if d.closed { + return fmt.Errorf("dummyRawConn closed") + } + return nil +} + +func (d *dummyRawConn) Write(f func(fd uintptr) (done bool)) error { + return nil +} +func (d *dummyRawConn) Close() { + d.mux.Lock() + d.closed = true + d.mux.Unlock() } diff --git a/internal/pool/pool.go b/internal/pool/pool.go index a8f0a6d..6d087bc 100644 --- a/internal/pool/pool.go +++ b/internal/pool/pool.go @@ -511,7 +511,7 @@ func (p *ConnPool) reapStaleConn() *Conn { func (p *ConnPool) isStaleConn(cn *Conn) bool { if p.opt.IdleTimeout == 0 && p.opt.MaxConnAge == 0 { - return false + return connCheck(cn.netConn) != nil } now := time.Now() @@ -522,5 +522,5 @@ func (p *ConnPool) isStaleConn(cn *Conn) bool { return true } - return false + return connCheck(cn.netConn) != nil } diff --git a/internal/pool/pool_test.go b/internal/pool/pool_test.go index 795aef3..7dcbf69 100644 --- a/internal/pool/pool_test.go +++ b/internal/pool/pool_test.go @@ -2,14 +2,13 @@ package pool_test import ( "context" + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" "sync" "testing" "time" "github.com/go-redis/redis/v8/internal/pool" - - . "github.com/onsi/ginkgo" - . "github.com/onsi/gomega" ) var _ = Describe("ConnPool", func() { @@ -285,6 +284,8 @@ var _ = Describe("conns reaper", func() { cn.SetUsedAt(time.Now().Add(-2 * idleTimeout)) case "aged": cn.SetCreatedAt(time.Now().Add(-2 * maxAge)) + case "conncheck": + cn.Close() } conns = append(conns, cn) staleConns = append(staleConns, cn) @@ -371,6 +372,7 @@ var _ = Describe("conns reaper", func() { assert("idle") assert("aged") + assert("conncheck") }) var _ = Describe("race", func() {