Merge pull request #509 from msukhotin/feat/configurable-tcp-not-sent-lowat

Make TCP_NOTSENT_LOWAT configurable

example.config.toml

@@ -221,6 +221,20 @@ http = "10s"
 idle = "5m"
 handshake = "10s"
 
+# TCP_NOTSENT_LOWAT (bytes) caps the amount of *unsent* data that the
+# kernel buffers per outbound socket. When the not-yet-sent backlog drops
+# below this threshold the socket becomes writable again, which applies
+# back-pressure to mtg's relay loop instead of letting kernel buffers
+# bloat. This reduces per-connection memory and bufferbloat, but on
+# high-BDP links (fast pipe × high RTT) it can cap single-flow upload
+# throughput at roughly value / RTT.
+#
+# Default is 128kib (good for interactive latency). If uploads through
+# the proxy feel slow on a fast link, raise this to 1mib–4mib. Suffixes
+# follow the same scheme as anti-replay max-size (kib, mib, gib).
+# Only applies to Linux and Darwin; ignored on other platforms.
+# tcp-not-sent-lowat = "128kib"
+
 # this defines a configuration for TCP keep alives. Default values are taken
 # from Golang default behavior.
 [network.keep-alive]

internal/cli/doctor.go

@@ -105,6 +105,7 @@ func (d *Doctor) Run(cli *CLI, version string) error {
 			Interval: conf.Network.KeepAlive.Interval.Get(0),
 			Count:    int(conf.Network.KeepAlive.Count.Get(0)),
 		},
+		int(conf.Network.TCPNotSentLowat.Get(network.DefaultTCPNotSentLowat)),
 	)
 
 	fmt.Println("Validate native network connectivity")

internal/cli/run_proxy.go

@@ -57,6 +57,7 @@ func makeNetwork(conf *config.Config, version string) (mtglib.Network, error) {
 			Interval: conf.Network.KeepAlive.Interval.Get(0),
 			Count:    int(conf.Network.KeepAlive.Count.Get(0)),
 		},
+		int(conf.Network.TCPNotSentLowat.Get(network.DefaultTCPNotSentLowat)),
 	)
 
 	proxyDialers := make([]mtglib.Network, len(conf.Network.Proxies))

internal/config/config.go

@@ -71,9 +71,10 @@ type Config struct {
 			Interval TypeDuration    `json:"interval"`
 			Count    TypeConcurrency `json:"count"`
 		} `json:"keepAlive"`
-		DOHIP   TypeIP         `json:"dohIp"`
-		DNS     TypeDNSURI     `json:"dns"`
-		Proxies []TypeProxyURL `json:"proxies"`
+		DOHIP            TypeIP         `json:"dohIp"`
+		DNS              TypeDNSURI     `json:"dns"`
+		Proxies          []TypeProxyURL `json:"proxies"`
+		TCPNotSentLowat  TypeBytes      `json:"tcpNotSentLowat"`
 	} `json:"network"`
 	Stats struct {
 		StatsD struct {

internal/config/parse.go

@@ -67,9 +67,10 @@ type tomlConfig struct {
 			Interval string `toml:"interval" json:"interval,omitempty"`
 			Count    uint   `toml:"count" json:"count,omitempty"`
 		} `toml:"keep-alive" json:"keepAlive,omitempty"`
-		DOHIP   string   `toml:"doh-ip" json:"dohIp,omitempty"`
-		DNS     string   `toml:"dns" json:"dns,omitempty"`
-		Proxies []string `toml:"proxies" json:"proxies,omitempty"`
+		DOHIP           string   `toml:"doh-ip" json:"dohIp,omitempty"`
+		DNS             string   `toml:"dns" json:"dns,omitempty"`
+		Proxies         []string `toml:"proxies" json:"proxies,omitempty"`
+		TCPNotSentLowat string   `toml:"tcp-not-sent-lowat" json:"tcpNotSentLowat,omitempty"`
 	} `toml:"network" json:"network,omitempty"`
 	Stats struct {
 		StatsD struct {

network/v2/base_http_test.go

@@ -25,7 +25,7 @@ func (suite *BaseHTTPTestSuite) SetupSuite() {
 }
 
 func (suite *BaseHTTPTestSuite) SetupTest() {
-	suite.client = network.New(nil, "mtg/1", 0, 0, 0, network.DefaultKeepAliveConfig).MakeHTTPClient(nil)
+	suite.client = network.New(nil, "mtg/1", 0, 0, 0, network.DefaultKeepAliveConfig, 0).MakeHTTPClient(nil)
 }
 
 func (suite *BaseHTTPTestSuite) TestGet() {

network/v2/base_network_test.go

@@ -19,7 +19,7 @@ type BaseNetworkTestSuite struct {
 func (suite *BaseNetworkTestSuite) SetupSuite() {
 	suite.EchoServerTestSuite.SetupSuite()
 
-	suite.net = network.New(nil, "agent", 0, 0, 0, network.DefaultKeepAliveConfig)
+	suite.net = network.New(nil, "agent", 0, 0, 0, network.DefaultKeepAliveConfig, 0)
 }
 
 func (suite *BaseNetworkTestSuite) TestDialUnknownNetwork() {

network/v2/init.go

@@ -56,15 +56,21 @@ const (
 	// unacknowledged data.
 	tcpLingerTimeout = 1
 
-	// tcpNotSentLowat limits the amount of unsent data queued in the
-	// kernel write buffer per socket. When the unsent data drops below
-	// this threshold, the socket becomes writable again. This reduces
-	// per-connection memory usage and bufferbloat by applying
-	// back-pressure to the relay loop instead of piling up data in
-	// kernel buffers.
-	tcpNotSentLowat = 128 * 1024
 )
 
+// DefaultTCPNotSentLowat is the default value applied to TCP_NOTSENT_LOWAT
+// on outbound sockets. It limits the amount of unsent data queued in the
+// kernel write buffer per socket. When the unsent data drops below this
+// threshold, the socket becomes writable again. This reduces per-connection
+// memory usage and bufferbloat by applying back-pressure to the relay loop
+// instead of piling up data in kernel buffers.
+//
+// The default is conservative and biased towards interactive latency. On
+// high-BDP links (fast pipe × high RTT) it can cap single-flow upload
+// throughput at roughly DefaultTCPNotSentLowat / RTT. Raise it via the
+// [network.tcp-not-sent-lowat] config option if uploads feel slow.
+const DefaultTCPNotSentLowat = 128 * 1024
+
 var (
 	ErrCannotDial = errors.New("cannot dial to any address")
 

network/v2/network.go

@@ -14,10 +14,11 @@ import (
 type network struct {
 	net.Dialer
 
-	keepAliveConfig net.KeepAliveConfig
-	httpTimeout     time.Duration
-	idleTimeout     time.Duration
-	userAgent       string
+	keepAliveConfig  net.KeepAliveConfig
+	httpTimeout      time.Duration
+	idleTimeout      time.Duration
+	userAgent        string
+	tcpNotSentLowat  int
 }
 
 func (n *network) Dial(network, address string) (essentials.Conn, error) {
@@ -38,7 +39,7 @@ func (n *network) DialContext(ctx context.Context, network, address string) (ess
 
 	tcpConn := conn.(*net.TCPConn)
 
-	return tcpConn, setCommonSocketOptions(tcpConn, n.keepAliveConfig)
+	return tcpConn, setCommonSocketOptions(tcpConn, n.keepAliveConfig, n.tcpNotSentLowat)
 }
 
 func (n *network) MakeHTTPClient(
@@ -73,6 +74,7 @@ func New(
 	httpTimeout,
 	idleTimeout time.Duration,
 	keepAliveConfig net.KeepAliveConfig,
+	tcpNotSentLowat int,
 ) mtglib.Network {
 	if dnsResolver == nil {
 		dnsResolver = net.DefaultResolver
@@ -82,6 +84,10 @@ func New(
 		userAgent = UserAgent
 	}
 
+	if tcpNotSentLowat == 0 {
+		tcpNotSentLowat = DefaultTCPNotSentLowat
+	}
+
 	return &network{
 		Dialer: net.Dialer{
 			Timeout:       tcpTimeout,
@@ -92,5 +98,6 @@ func New(
 		idleTimeout:     idleTimeout,
 		httpTimeout:     httpTimeout,
 		keepAliveConfig: keepAliveConfig,
+		tcpNotSentLowat: tcpNotSentLowat,
 	}
 }

network/v2/proxy_network_internal_test.go

@@ -27,6 +27,7 @@ func TestProxyServerDialerDropsCustomResolver(t *testing.T) {
 		0,
 		0,
 		DefaultKeepAliveConfig,
+		0,
 	)
 
 	if base.NativeDialer().Resolver != customResolver {

network/v2/sockopts.go

@@ -5,7 +5,7 @@ import (
 	"net"
 )
 
-func setCommonSocketOptions(conn *net.TCPConn, keepAliveConfig net.KeepAliveConfig) error {
+func setCommonSocketOptions(conn *net.TCPConn, keepAliveConfig net.KeepAliveConfig, tcpNotSentLowat int) error {
 	if err := applyKeepAlive(conn, keepAliveConfig); err != nil {
 		return fmt.Errorf("cannot configure TCP keepalive: %w", err)
 	}
@@ -25,7 +25,7 @@ func setCommonSocketOptions(conn *net.TCPConn, keepAliveConfig net.KeepAliveConf
 
 	setCongestionControl(rawConn)
 	setTCPUserTimeout(rawConn, keepAliveConfig)
-	setNotSentLowat(rawConn)
+	setNotSentLowat(rawConn, tcpNotSentLowat)
 
 	return nil
 }

network/v2/sockopts_lowat.go

@@ -8,13 +8,21 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-// setNotSentLowat sets TCP_NOTSENT_LOWAT which limits the amount of
-// unsent data queued in the kernel write buffer. Once unsent data drops
-// below this threshold the socket becomes writable again, applying
-// back-pressure to the relay loop instead of piling up data in kernel
-// buffers. This reduces per-connection memory and bufferbloat.
-func setNotSentLowat(conn syscall.RawConn) {
+// setNotSentLowat sets TCP_NOTSENT_LOWAT to value bytes. The option
+// limits the amount of unsent data queued in the kernel write buffer:
+// once the unsent backlog drops below the threshold, the socket becomes
+// writable again, applying back-pressure to the relay loop instead of
+// piling up data in kernel buffers. This reduces per-connection memory
+// and bufferbloat.
+//
+// A non-positive value disables the call, leaving the kernel default in
+// effect (no upper bound beyond SO_SNDBUF).
+func setNotSentLowat(conn syscall.RawConn, value int) {
+	if value <= 0 {
+		return
+	}
+
 	conn.Control(func(fd uintptr) { //nolint: errcheck
-		unix.SetsockoptInt(int(fd), unix.IPPROTO_TCP, unix.TCP_NOTSENT_LOWAT, tcpNotSentLowat) //nolint: errcheck
+		unix.SetsockoptInt(int(fd), unix.IPPROTO_TCP, unix.TCP_NOTSENT_LOWAT, value) //nolint: errcheck
 	})
 }

network/v2/sockopts_lowat_stub.go

@@ -4,4 +4,4 @@ package network
 
 import "syscall"
 
-func setNotSentLowat(conn syscall.RawConn) {}
+func setNotSentLowat(conn syscall.RawConn, value int) {}

network/v2/sockopts_test.go

@@ -65,7 +65,7 @@ func TestSetCommonSocketOptionsKeepAlive(t *testing.T) {
 
 	tcpConn := accepted.(*net.TCPConn)
 
-	err = setCommonSocketOptions(tcpConn, DefaultKeepAliveConfig)
+	err = setCommonSocketOptions(tcpConn, DefaultKeepAliveConfig, DefaultTCPNotSentLowat)
 	require.NoError(t, err)
 
 	rawConn, err := tcpConn.SyscallConn()
@@ -90,3 +90,57 @@ func TestSetCommonSocketOptionsKeepAlive(t *testing.T) {
 	})
 	require.NoError(t, err)
 }
+
+func TestSetCommonSocketOptionsNotSentLowat(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		want int
+	}{
+		{name: "default", want: DefaultTCPNotSentLowat},
+		{name: "custom", want: 4 * 1024 * 1024},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			listener, err := net.Listen("tcp", "127.0.0.1:0")
+			require.NoError(t, err)
+			defer listener.Close() //nolint: errcheck
+
+			dialDone := make(chan struct{})
+
+			go func() {
+				c, err := net.Dial("tcp", listener.Addr().String())
+				if err == nil {
+					defer c.Close() //nolint: errcheck
+				}
+				close(dialDone)
+			}()
+
+			require.NoError(t, listener.(*net.TCPListener).SetDeadline(time.Now().Add(5*time.Second)))
+
+			accepted, err := listener.Accept()
+			require.NoError(t, err)
+			defer accepted.Close() //nolint: errcheck
+
+			<-dialDone
+
+			tcpConn := accepted.(*net.TCPConn)
+
+			require.NoError(t, setCommonSocketOptions(tcpConn, DefaultKeepAliveConfig, tc.want))
+
+			rawConn, err := tcpConn.SyscallConn()
+			require.NoError(t, err)
+
+			err = rawConn.Control(func(fd uintptr) {
+				got, err := unix.GetsockoptInt(int(fd), unix.IPPROTO_TCP, unix.TCP_NOTSENT_LOWAT)
+				require.NoError(t, err)
+				require.Equal(t, tc.want, got, "TCP_NOTSENT_LOWAT should match value passed to setCommonSocketOptions")
+			})
+			require.NoError(t, err)
+		})
+	}
+}

network/v2/socks_proxy_test.go

@@ -66,7 +66,7 @@ func (suite *SocksProxyTestSuite) SetupSuite() {
 	require.NoError(suite.T(), err)
 	suite.authURL = parsed
 
-	suite.baseNetwork = network.New(nil, "mtg", 0, 0, 0, network.DefaultKeepAliveConfig)
+	suite.baseNetwork = network.New(nil, "mtg", 0, 0, 0, network.DefaultKeepAliveConfig, 0)
 }
 
 func (suite *SocksProxyTestSuite) TestIncorrectSchema() {