conn, device, tun: implement vectorized I/O on Linux

This commit changes the tun.Device and conn.Bind interfaces to accept
packet vectors for reading and writing. Internal plumbing between
these interfaces now passes a vector of packets. Vectors move untouched
between these interfaces, i.e. if 128 packets are received from
conn.Bind.Read(), 128 packets are passed to tun.Device.Write(). There
is no internal buffering.

Platform-specific implementations of tun.Device have been updated to
the new tun.Device interface, but only Linux supports passing more than
one packet for now. The Linux tun.Device implementation accomplishes
this via TSO and GRO, which is made possible by virtio extensions in
the TUN driver.

conn.LinuxSocketEndpoint has been deleted in favor of a collapsed
conn.StdNetBind. conn.StdNetBind makes use of recvmmsg() and sendmmsg()
on Linux. All platforms fall under conn.StdNetBind, except for Windows,
which remains in conn.WinRingBind. Sticky sockets support has been
refactored as part of this work to eventually be applicable on
platforms other than just Linux, however Linux remains the sole
platform that fully implements it. The conn.Bind UDP socket buffer is
now being sized to 7MB, whereas it was previously inheriting the system
default.

Signed-off-by: Jordan Whited <jordan@tailscale.com>
Signed-off-by: James Tucker <james@tailscale.com>
Co-authored-by: James Tucker <james@tailscale.com>

Author: jwhited

conn/bind_linux.go

@@ -321,6 +321,11 @@ func (bind *WinRingBind) Close() error {
 	return nil
 }
 
+func (bind *WinRingBind) BatchSize() int {
+	// TODO: implement batching in and out of the ring
+	return 1
+}
+
 func (bind *WinRingBind) SetMark(mark uint32) error {
 	return nil
 }
@@ -409,16 +414,22 @@ retry:
 	return n, &ep, nil
 }
 
-func (bind *WinRingBind) receiveIPv4(buf []byte) (int, Endpoint, error) {
+func (bind *WinRingBind) receiveIPv4(buffs [][]byte, sizes []int, eps []Endpoint) (int, error) {
 	bind.mu.RLock()
 	defer bind.mu.RUnlock()
-	return bind.v4.Receive(buf, &bind.isOpen)
+	n, ep, err := bind.v4.Receive(buffs[0], &bind.isOpen)
+	sizes[0] = n
+	eps[0] = ep
+	return 1, err
 }
 
-func (bind *WinRingBind) receiveIPv6(buf []byte) (int, Endpoint, error) {
+func (bind *WinRingBind) receiveIPv6(buffs [][]byte, sizes []int, eps []Endpoint) (int, error) {
 	bind.mu.RLock()
 	defer bind.mu.RUnlock()
-	return bind.v6.Receive(buf, &bind.isOpen)
+	n, ep, err := bind.v6.Receive(buffs[0], &bind.isOpen)
+	sizes[0] = n
+	eps[0] = ep
+	return 1, err
 }
 
 func (bind *afWinRingBind) Send(buf []byte, nend *WinRingEndpoint, isOpen *atomic.Uint32) error {
@@ -473,32 +484,38 @@ func (bind *afWinRingBind) Send(buf []byte, nend *WinRingEndpoint, isOpen *atomi
 	return winrio.SendEx(bind.rq, dataBuffer, 1, nil, addressBuffer, nil, nil, 0, 0)
 }
 
-func (bind *WinRingBind) Send(buf []byte, endpoint Endpoint) error {
+func (bind *WinRingBind) Send(buffs [][]byte, endpoint Endpoint) error {
 	nend, ok := endpoint.(*WinRingEndpoint)
 	if !ok {
 		return ErrWrongEndpointType
 	}
 	bind.mu.RLock()
 	defer bind.mu.RUnlock()
-	switch nend.family {
-	case windows.AF_INET:
-		if bind.v4.blackhole {
-			return nil
-		}
-		return bind.v4.Send(buf, nend, &bind.isOpen)
-	case windows.AF_INET6:
-		if bind.v6.blackhole {
-			return nil
+	for _, buf := range buffs {
+		switch nend.family {
+		case windows.AF_INET:
+			if bind.v4.blackhole {
+				continue
+			}
+			if err := bind.v4.Send(buf, nend, &bind.isOpen); err != nil {
+				return err
+			}
+		case windows.AF_INET6:
+			if bind.v6.blackhole {
+				continue
+			}
+			if err := bind.v6.Send(buf, nend, &bind.isOpen); err != nil {
+				return err
+			}
 		}
-		return bind.v6.Send(buf, nend, &bind.isOpen)
 	}
 	return nil
 }
 
-func (bind *StdNetBind) BindSocketToInterface4(interfaceIndex uint32, blackhole bool) error {
-	bind.mu.Lock()
-	defer bind.mu.Unlock()
-	sysconn, err := bind.ipv4.SyscallConn()
+func (s *StdNetBind) BindSocketToInterface4(interfaceIndex uint32, blackhole bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	sysconn, err := s.ipv4.SyscallConn()
 	if err != nil {
 		return err
 	}
@@ -511,14 +528,14 @@ func (bind *StdNetBind) BindSocketToInterface4(interfaceIndex uint32, blackhole
 	if err != nil {
 		return err
 	}
-	bind.blackhole4 = blackhole
+	s.blackhole4 = blackhole
 	return nil
 }
 
-func (bind *StdNetBind) BindSocketToInterface6(interfaceIndex uint32, blackhole bool) error {
-	bind.mu.Lock()
-	defer bind.mu.Unlock()
-	sysconn, err := bind.ipv6.SyscallConn()
+func (s *StdNetBind) BindSocketToInterface6(interfaceIndex uint32, blackhole bool) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	sysconn, err := s.ipv6.SyscallConn()
 	if err != nil {
 		return err
 	}
@@ -531,7 +548,7 @@ func (bind *StdNetBind) BindSocketToInterface6(interfaceIndex uint32, blackhole
 	if err != nil {
 		return err
 	}
-	bind.blackhole6 = blackhole
+	s.blackhole6 = blackhole
 	return nil
 }
 

conn/bind_std.go

@@ -89,32 +89,39 @@ func (c *ChannelBind) Close() error {
 	return nil
 }
 
+func (c *ChannelBind) BatchSize() int { return 1 }
+
 func (c *ChannelBind) SetMark(mark uint32) error { return nil }
 
 func (c *ChannelBind) makeReceiveFunc(ch chan []byte) conn.ReceiveFunc {
-	return func(b []byte) (n int, ep conn.Endpoint, err error) {
+	return func(buffs [][]byte, sizes []int, eps []conn.Endpoint) (n int, err error) {
 		select {
 		case <-c.closeSignal:
-			return 0, nil, net.ErrClosed
+			return 0, net.ErrClosed
 		case rx := <-ch:
-			return copy(b, rx), c.target6, nil
+			copied := copy(buffs[0], rx)
+			sizes[0] = copied
+			eps[0] = c.target6
+			return 1, nil
 		}
 	}
 }
 
-func (c *ChannelBind) Send(b []byte, ep conn.Endpoint) error {
-	select {
-	case <-c.closeSignal:
-		return net.ErrClosed
-	default:
-		bc := make([]byte, len(b))
-		copy(bc, b)
-		if ep.(ChannelEndpoint) == c.target4 {
-			*c.tx4 <- bc
-		} else if ep.(ChannelEndpoint) == c.target6 {
-			*c.tx6 <- bc
-		} else {
-			return os.ErrInvalid
+func (c *ChannelBind) Send(buffs [][]byte, ep conn.Endpoint) error {
+	for _, b := range buffs {
+		select {
+		case <-c.closeSignal:
+			return net.ErrClosed
+		default:
+			bc := make([]byte, len(b))
+			copy(bc, b)
+			if ep.(ChannelEndpoint) == c.target4 {
+				*c.tx4 <- bc
+			} else if ep.(ChannelEndpoint) == c.target6 {
+				*c.tx6 <- bc
+			} else {
+				return os.ErrInvalid
+			}
 		}
 	}
 	return nil

conn/bind_windows.go

@@ -5,8 +5,8 @@
 
 package conn
 
-func (bind *StdNetBind) PeekLookAtSocketFd4() (fd int, err error) {
-	sysconn, err := bind.ipv4.SyscallConn()
+func (s *StdNetBind) PeekLookAtSocketFd4() (fd int, err error) {
+	sysconn, err := s.ipv4.SyscallConn()
 	if err != nil {
 		return -1, err
 	}
@@ -19,8 +19,8 @@ func (bind *StdNetBind) PeekLookAtSocketFd4() (fd int, err error) {
 	return
 }
 
-func (bind *StdNetBind) PeekLookAtSocketFd6() (fd int, err error) {
-	sysconn, err := bind.ipv6.SyscallConn()
+func (s *StdNetBind) PeekLookAtSocketFd6() (fd int, err error) {
+	sysconn, err := s.ipv6.SyscallConn()
 	if err != nil {
 		return -1, err
 	}

conn/bindtest/bindtest.go

@@ -15,10 +15,17 @@ import (
 	"strings"
 )
 
-// A ReceiveFunc receives a single inbound packet from the network.
-// It writes the data into b. n is the length of the packet.
-// ep is the remote endpoint.
-type ReceiveFunc func(b []byte) (n int, ep Endpoint, err error)
+const (
+	DefaultBatchSize = 128 // maximum number of packets handled per read and write
+)
+
+// A ReceiveFunc receives at least one packet from the network and writes them
+// into packets. On a successful read it returns the number of elements of
+// sizes, packets, and endpoints that should be evaluated. Some elements of
+// sizes may be zero, and callers should ignore them. Callers must pass a sizes
+// and eps slice with a length greater than or equal to the length of packets.
+// These lengths must not exceed the length of the associated Bind.BatchSize().
+type ReceiveFunc func(packets [][]byte, sizes []int, eps []Endpoint) (n int, err error)
 
 // A Bind listens on a port for both IPv6 and IPv4 UDP traffic.
 //
@@ -38,11 +45,16 @@ type Bind interface {
 	// This mark is passed to the kernel as the socket option SO_MARK.
 	SetMark(mark uint32) error
 
-	// Send writes a packet b to address ep.
-	Send(b []byte, ep Endpoint) error
+	// Send writes one or more packets in buffs to address ep. The length of
+	// buffs must not exceed BatchSize().
+	Send(buffs [][]byte, ep Endpoint) error
 
 	// ParseEndpoint creates a new endpoint from a string.
 	ParseEndpoint(s string) (Endpoint, error)
+
+	// BatchSize is the number of buffers expected to be passed to
+	// the ReceiveFuncs, and the maximum expected to be passed to SendBatch.
+	BatchSize() int
 }
 
 // BindSocketToInterface is implemented by Bind objects that support being

conn/boundif_android.go

@@ -0,0 +1,24 @@
+/* SPDX-License-Identifier: MIT
+ *
+ * Copyright (C) 2017-2023 WireGuard LLC. All Rights Reserved.
+ */
+
+package conn
+
+import (
+	"testing"
+)
+
+func TestPrettyName(t *testing.T) {
+	var (
+		recvFunc ReceiveFunc = func(buffs [][]byte, sizes []int, eps []Endpoint) (n int, err error) { return }
+	)
+
+	const want = "TestPrettyName"
+
+	t.Run("ReceiveFunc.PrettyName", func(t *testing.T) {
+		if got := recvFunc.PrettyName(); got != want {
+			t.Errorf("PrettyName() = %v, want %v", got, want)
+		}
+	})
+}

conn/conn.go

@@ -0,0 +1,43 @@
+/* SPDX-License-Identifier: MIT
+ *
+ * Copyright (C) 2017-2023 WireGuard LLC. All Rights Reserved.
+ */
+
+package conn
+
+import (
+	"net"
+	"syscall"
+)
+
+// UDP socket read/write buffer size (7MB). The value of 7MB is chosen as it is
+// the max supported by a default configuration of macOS. Some platforms will
+// silently clamp the value to other maximums, such as linux clamping to
+// net.core.{r,w}mem_max (see _linux.go for additional implementation that works
+// around this limitation)
+const socketBufferSize = 7 << 20
+
+// controlFn is the callback function signature from net.ListenConfig.Control.
+// It is used to apply platform specific configuration to the socket prior to
+// bind.
+type controlFn func(network, address string, c syscall.RawConn) error
+
+// controlFns is a list of functions that are called from the listen config
+// that can apply socket options.
+var controlFns = []controlFn{}
+
+// listenConfig returns a net.ListenConfig that applies the controlFns to the
+// socket prior to bind. This is used to apply socket buffer sizing and packet
+// information OOB configuration for sticky sockets.
+func listenConfig() *net.ListenConfig {
+	return &net.ListenConfig{
+		Control: func(network, address string, c syscall.RawConn) error {
+			for _, fn := range controlFns {
+				if err := fn(network, address, c); err != nil {
+					return err
+				}
+			}
+			return nil
+		},
+	}
+}

conn/conn_test.go

@@ -0,0 +1,56 @@
+/* SPDX-License-Identifier: MIT
+ *
+ * Copyright (C) 2017-2023 WireGuard LLC. All Rights Reserved.
+ */
+
+package conn
+
+import (
+	"fmt"
+	"syscall"
+
+	"golang.org/x/sys/unix"
+)
+
+func init() {
+	controlFns = append(controlFns,
+
+		// Attempt to set the socket buffer size beyond net.core.{r,w}mem_max by
+		// using SO_*BUFFORCE. This requires CAP_NET_ADMIN, and is allowed here to
+		// fail silently - the result of failure is lower performance on very fast
+		// links or high latency links.
+		func(network, address string, c syscall.RawConn) error {
+			return c.Control(func(fd uintptr) {
+				// Set up to *mem_max
+				_ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, socketBufferSize)
+				_ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_SNDBUF, socketBufferSize)
+				// Set beyond *mem_max if CAP_NET_ADMIN
+				_ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, socketBufferSize)
+				_ = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, socketBufferSize)
+			})
+		},
+
+		// Enable receiving of the packet information (IP_PKTINFO for IPv4,
+		// IPV6_PKTINFO for IPv6) that is used to implement sticky socket support.
+		func(network, address string, c syscall.RawConn) error {
+			var err error
+			switch network {
+			case "udp4":
+				c.Control(func(fd uintptr) {
+					err = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_PKTINFO, 1)
+				})
+			case "udp6":
+				c.Control(func(fd uintptr) {
+					err = unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_RECVPKTINFO, 1)
+					if err != nil {
+						return
+					}
+					err = unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_V6ONLY, 1)
+				})
+			default:
+				err = fmt.Errorf("unhandled network: %s: %w", network, unix.EINVAL)
+			}
+			return err
+		},
+	)
+}