keepalive: RT timeout parameters (pingcap#239)

.github/workflows/main.yml

@@ -63,4 +63,4 @@ jobs:
       ref: ${{ inputs.ref || github.ref }}
       debug: false
       target: "lint test"
-      all_platform: false
+      all_platform: true

lib/config/proxy.go

@@ -46,9 +46,10 @@ type Metrics struct {
 
 type KeepAlive struct {
 	Enabled bool          `yaml:"enabled,omitempty" toml:"enabled,omitempty" json:"enabled,omitempty"`
-	Cnt     int           `yaml:"cnt,omitempty" toml:"cnt,omitempty" json:"cnt,omitempty"`
 	Idle    time.Duration `yaml:"idle,omitempty" toml:"idle,omitempty" json:"idle,omitempty"`
+	Cnt     int           `yaml:"cnt,omitempty" toml:"cnt,omitempty" json:"cnt,omitempty"`
 	Intvl   time.Duration `yaml:"intvl,omitempty" toml:"intvl,omitempty" json:"intvl,omitempty"`
+	Timeout time.Duration `yaml:"timeout,omitempty" toml:"timeout,omitempty" json:"timeout,omitempty"`
 }
 
 type ProxyServerOnline struct {
@@ -123,13 +124,15 @@ type Security struct {
 func DefaultKeepAlive() (frontend, backendHealthy, backendUnhealthy KeepAlive) {
 	frontend.Enabled = true
 	backendHealthy.Enabled = true
-	backendHealthy.Cnt = 5
 	backendHealthy.Idle = 60 * time.Second
+	backendHealthy.Cnt = 5
 	backendHealthy.Intvl = 5 * time.Second
+	backendHealthy.Timeout = 30 * time.Second
 	backendUnhealthy.Enabled = true
-	backendUnhealthy.Cnt = 2
 	backendUnhealthy.Idle = 1 * time.Second
+	backendUnhealthy.Cnt = 2
 	backendUnhealthy.Intvl = 1 * time.Second
+	backendUnhealthy.Timeout = 3 * time.Second
 	return
 }
 

pkg/proxy/keepalive/keepalive.go

@@ -22,7 +22,7 @@ import (
 )
 
 var (
-	ErrKeepAlive = errors.New("failed to set keepalive")
+	ErrKeepAlive = errors.New("failed to set keepalive and timeout")
 )
 
 func SetKeepalive(conn net.Conn, cfg config.KeepAlive) error {
@@ -34,14 +34,17 @@ func SetKeepalive(conn net.Conn, cfg config.KeepAlive) error {
 	if err := tcpcn.SetKeepAlive(cfg.Enabled); err != nil {
 		return errors.Wrap(ErrKeepAlive, err)
 	}
-	if !cfg.Enabled {
-		return nil
-	}
 
 	syscn, err := tcpcn.SyscallConn()
 	if err != nil {
 		return errors.Wrap(ErrKeepAlive, err)
 	}
 
-	return setKeepalive(syscn, cfg)
+	var kerr, terr error
+	return errors.Collect(ErrKeepAlive, kerr, terr, syscn.Control(func(fd uintptr) {
+		if cfg.Enabled {
+			kerr = setKeepalive(fd, cfg)
+		}
+		terr = setTimeout(fd, cfg)
+	}))
 }

pkg/proxy/keepalive/keepalive_darwin.go

@@ -20,7 +20,6 @@ import (
 	"syscall"
 
 	"github.com/pingcap/TiProxy/lib/config"
-	"github.com/pingcap/TiProxy/lib/util/errors"
 )
 
 const (
@@ -29,26 +28,21 @@ const (
 	_TCP_KEEPCNT   = 0x102
 )
 
-func setKeepalive(syscn syscall.RawConn, cfg config.KeepAlive) error {
-	var serr error
-	return errors.Collect(ErrKeepAlive, serr, syscn.Control(func(fd uintptr) {
-		if val := cfg.Idle.Seconds(); val > 0 {
-			serr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPALIVE, int(val))
-			if serr != nil {
-				return
-			}
+func setKeepalive(fd uintptr, cfg config.KeepAlive) error {
+	if val := cfg.Idle.Seconds(); val > 0 {
+		if err := syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPALIVE, int(val)); err != nil {
+			return err
 		}
-		if cfg.Cnt > 0 {
-			serr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, _TCP_KEEPCNT, cfg.Cnt)
-			if serr != nil {
-				return
-			}
+	}
+	if cfg.Cnt > 0 {
+		if err := syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, _TCP_KEEPCNT, cfg.Cnt); err != nil {
+			return err
 		}
-		if val := cfg.Intvl.Seconds(); val > 0 {
-			serr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, _TCP_KEEPINTVL, int(val))
-			if serr != nil {
-				return
-			}
+	}
+	if val := cfg.Intvl.Seconds(); val > 0 {
+		if err := syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, _TCP_KEEPINTVL, int(val)); err != nil {
+			return err
 		}
-	}))
+	}
+	return nil
 }

pkg/proxy/keepalive/keepalive_default.go

@@ -17,11 +17,9 @@
 package keepalive
 
 import (
-	"syscall"
-
 	"github.com/pingcap/TiProxy/lib/config"
 )
 
-func setKeepalive(syscn syscall.RawConn, cfg config.KeepAlive) error {
+func setKeepalive(fd uintptr, cfg config.KeepAlive) error {
 	return nil
 }

pkg/proxy/keepalive/keepalive_unix.go

@@ -20,29 +20,23 @@ import (
 	"syscall"
 
 	"github.com/pingcap/TiProxy/lib/config"
-	"github.com/pingcap/TiProxy/lib/util/errors"
 )
 
-func setKeepalive(syscn syscall.RawConn, cfg config.KeepAlive) error {
-	var serr error
-	return errors.Collect(ErrKeepAlive, serr, syscn.Control(func(fd uintptr) {
-		if val := cfg.Idle.Seconds(); val > 0 {
-			serr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPIDLE, int(val))
-			if serr != nil {
-				return
-			}
+func setKeepalive(fd uintptr, cfg config.KeepAlive) error {
+	if val := cfg.Idle.Seconds(); val > 0 {
+		if err := syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPIDLE, int(val)); err != nil {
+			return err
 		}
-		if cfg.Cnt > 0 {
-			serr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPCNT, cfg.Cnt)
-			if serr != nil {
-				return
-			}
+	}
+	if cfg.Cnt > 0 {
+		if err := syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPCNT, cfg.Cnt); err != nil {
+			return err
 		}
-		if val := cfg.Intvl.Seconds(); val > 0 {
-			serr = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPINTVL, int(val))
-			if serr != nil {
-				return
-			}
+	}
+	if val := cfg.Intvl.Seconds(); val > 0 {
+		if err := syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPINTVL, int(val)); err != nil {
+			return err
 		}
-	}))
+	}
+	return nil
 }

pkg/proxy/keepalive/timeout_darwin.go

@@ -0,0 +1,27 @@
+//go:build darwin
+
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package keepalive
+
+import (
+	"syscall"
+
+	"github.com/pingcap/TiProxy/lib/config"
+)
+
+func setTimeout(fd uintptr, cfg config.KeepAlive) error {
+	return syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_RXT_CONNDROPTIME, int(cfg.Timeout.Seconds()))
+}

pkg/proxy/keepalive/timeout_default.go

@@ -0,0 +1,25 @@
+//go:build !(windows || linux || darwin)
+
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package keepalive
+
+import (
+	"github.com/pingcap/TiProxy/lib/config"
+)
+
+func setTimeout(fd uintptr, cfg config.KeepAlive) error {
+	return nil
+}

pkg/proxy/keepalive/timeout_linux.go

@@ -0,0 +1,29 @@
+//go:build linux
+
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package keepalive
+
+import (
+	"syscall"
+
+	"github.com/pingcap/TiProxy/lib/config"
+)
+
+const _TCP_USER_TIMEOUT = 0x12
+
+func setTimeout(fd uintptr, cfg config.KeepAlive) error {
+	return syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, _TCP_USER_TIMEOUT, int(cfg.Timeout.Milliseconds()))
+}

pkg/proxy/keepalive/timeout_windows.go

@@ -0,0 +1,29 @@
+//go:build windows
+
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package keepalive
+
+import (
+	"syscall"
+
+	"github.com/pingcap/TiProxy/lib/config"
+)
+
+const _TCP_MAXRT = 5
+
+func setTimeout(fd uintptr, cfg config.KeepAlive) error {
+	return syscall.SetsockoptInt(syscall.Handle(fd), syscall.IPPROTO_TCP, _TCP_MAXRT, int(cfg.Timeout.Seconds()))
+}

pkg/proxy/net/packetio_test.go

@@ -18,6 +18,7 @@ import (
 	"encoding/binary"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/pingcap/TiProxy/lib/config"
 	"github.com/pingcap/TiProxy/lib/util/security"
@@ -268,17 +269,29 @@ func TestKeepAlive(t *testing.T) {
 	stls, ctls, err := security.CreateTLSConfigForTest()
 	require.NoError(t, err)
 	frontend, backendHealthy, backendUnhealthy := config.DefaultKeepAlive()
+	backendUnhealthy.Timeout = 2 * time.Second
+	backendUnhealthy.Idle = time.Second
+	backendUnhealthy.Cnt = 1
+	backendUnhealthy.Intvl = time.Second
 	testTCPConn(t,
 		func(t *testing.T, cli *PacketIO) {
 			require.NoError(t, cli.SetKeepalive(frontend))
 			require.NoError(t, cli.ClientTLSHandshake(ctls))
+			time.Sleep(3 * time.Second)
+			_, err := cli.ReadPacket()
+			require.NoError(t, err)
+			require.NoError(t, cli.WritePacket([]byte{0, 1, 2}, true))
 		},
 		func(t *testing.T, srv *PacketIO) {
 			require.NoError(t, srv.SetKeepalive(backendHealthy))
 			_, err = srv.ServerTLSHandshake(stls)
 			require.NoError(t, err)
 			require.NoError(t, srv.SetKeepalive(backendUnhealthy))
+			require.NoError(t, srv.WritePacket([]byte{0, 1, 2}, true))
+			time.Sleep(3*time.Second + 100*time.Millisecond)
+			_, err := srv.ReadPacket()
+			require.NoError(t, err)
 		},
-		10,
+		1,
 	)
 }