hotfix/v8_conn_pool_check_fd (redis#2431)

Author: hardeyhe

internal/pool/conncheck.go

@@ -0,0 +1,46 @@
+//go:build linux || darwin || dragonfly || freebsd || netbsd || openbsd || solaris || illumos
+// +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
+}

internal/pool/conncheck_dummy.go

@@ -0,0 +1,10 @@
+//go:build !linux && !darwin && !dragonfly && !freebsd && !netbsd && !openbsd && !solaris && !illumos
+// +build !linux,!darwin,!dragonfly,!freebsd,!netbsd,!openbsd,!solaris,!illumos
+
+package pool
+
+import "net"
+
+func connCheck(conn net.Conn) error {
+	return nil
+}

internal/pool/conncheck_test.go

@@ -0,0 +1,47 @@
+//go:build linux || darwin || dragonfly || freebsd || netbsd || openbsd || solaris || illumos
+// +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")
+		}
+	})
+}

internal/pool/main_test.go

@@ -35,19 +35,18 @@ func perform(n int, cbs ...func(int)) {
 }
 
 func dummyDialer(context.Context) (net.Conn, error) {
+	// return &net.TCPConn{}, nil
 	return newDummyConn(), nil
 }
 
 func newDummyConn() net.Conn {
 	return &dummyConn{
-		rawConn: new(dummyRawConn),
+		rawConn: &dummyRawConn{},
 	}
 }
 
-var (
-	_ net.Conn     = (*dummyConn)(nil)
-	_ syscall.Conn = (*dummyConn)(nil)
-)
+var _ net.Conn = (*dummyConn)(nil)
+var _ syscall.Conn = (*dummyConn)(nil)
 
 type dummyConn struct {
 	rawConn *dummyRawConn
@@ -95,17 +94,17 @@ func (d *dummyConn) SetWriteDeadline(t time.Time) error {
 var _ syscall.RawConn = (*dummyRawConn)(nil)
 
 type dummyRawConn struct {
-	mu     sync.Mutex
 	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.mu.Lock()
-	defer d.mu.Unlock()
+	d.mux.Lock()
+	defer d.mux.Unlock()
 	if d.closed {
 		return fmt.Errorf("dummyRawConn closed")
 	}
@@ -115,9 +114,8 @@ func (d *dummyRawConn) Read(f func(fd uintptr) (done bool)) error {
 func (d *dummyRawConn) Write(f func(fd uintptr) (done bool)) error {
 	return nil
 }
-
 func (d *dummyRawConn) Close() {
-	d.mu.Lock()
+	d.mux.Lock()
 	d.closed = true
-	d.mu.Unlock()
+	d.mux.Unlock()
 }

internal/pool/pool_test.go

@@ -323,8 +323,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()
+				case "conncheck":
+					cn.Close()
 				}
 				conns = append(conns, cn)
 				staleConns = append(staleConns, cn)
@@ -411,7 +411,7 @@ var _ = Describe("conns reaper", func() {
 
 	assert("idle")
 	assert("aged")
-	assert("connCheck")
+	assert("conncheck")
 })
 
 var _ = Describe("race", func() {

pool_test.go

@@ -72,6 +72,7 @@ var _ = Describe("pool", func() {
 			Expect(cmds).To(HaveLen(1))
 			Expect(ping.Err()).NotTo(HaveOccurred())
 			Expect(ping.Val()).To(Equal("PONG"))
+			Expect(pipe.Close()).NotTo(HaveOccurred())
 		})
 
 		pool := client.Pool()
@@ -86,11 +87,10 @@ var _ = Describe("pool", func() {
 		cn.SetNetConn(&badConn{})
 		client.Pool().Put(ctx, cn)
 
-		val, err := client.Ping(ctx).Result()
+		err = client.Ping(ctx).Err()
 		Expect(err).NotTo(HaveOccurred())
-		Expect(val).To(Equal("PONG"))
 
-		val, err = client.Ping(ctx).Result()
+		val, err := client.Ping(ctx).Result()
 		Expect(err).NotTo(HaveOccurred())
 		Expect(val).To(Equal("PONG"))
 

tx_test.go

@@ -123,7 +123,7 @@ var _ = Describe("Tx", func() {
 		Expect(num).To(Equal(int64(N)))
 	})
 
-	It("should recover from bad connection", func() {
+	It("should remove from bad connection", func() {
 		// Put bad connection in the pool.
 		cn, err := client.Pool().Get(context.Background())
 		Expect(err).NotTo(HaveOccurred())
@@ -134,14 +134,14 @@ var _ = Describe("Tx", func() {
 		do := func() error {
 			err := client.Watch(ctx, func(tx *redis.Tx) error {
 				_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
-					pipe.Ping(ctx)
-					return nil
+					return pipe.Ping(ctx).Err()
 				})
 				return err
 			})
 			return err
 		}
 
+		// connCheck will automatically remove damaged connections.
 		err = do()
 		Expect(err).NotTo(HaveOccurred())
 	})