experimental xdp dispatcher

PiperOrigin-RevId: 472509327

pkg/tcpip/BUILD

@@ -66,6 +66,7 @@ deps_test(
         "//pkg/state/wire",
         "//pkg/sync",
         "//pkg/waiter",
+        "//pkg/xdp",
 
         # Other deps.
         "@com_github_google_btree//:go_default_library",

pkg/tcpip/link/fdbased/BUILD

@@ -11,6 +11,7 @@ go_library(
         "mmap_stub.go",
         "mmap_unsafe.go",
         "packet_dispatchers.go",
+        "xdp.go",
     ],
     visibility = ["//visibility:public"],
     deps = [
@@ -21,6 +22,7 @@ go_library(
         "//pkg/tcpip/header",
         "//pkg/tcpip/link/rawfile",
         "//pkg/tcpip/stack",
+        "//pkg/xdp",
         "@org_golang_x_sys//unix:go_default_library",
     ],
 )

pkg/tcpip/link/fdbased/endpoint.go

@@ -41,7 +41,6 @@
 package fdbased
 
 import (
-	"errors"
 	"fmt"
 
 	"golang.org/x/sys/unix"
@@ -66,12 +65,12 @@ type linkDispatcher interface {
 // dispatching packets from the underlying FD.
 type PacketDispatchMode int
 
-const (
-	// BatchSize is the number of packets to write in each syscall. It is 47
-	// because when GvisorGSO is in use then a single 65KB TCP segment can get
-	// split into 46 segments of 1420 bytes and a single 216 byte segment.
-	BatchSize = 47
+// BatchSize is the number of packets to write in each syscall. It is 47
+// because when GvisorGSO is in use then a single 65KB TCP segment can get
+// split into 46 segments of 1420 bytes and a single 216 byte segment.
+const BatchSize = 47
 
+const (
 	// Readv is the default dispatch mode and is the least performant of the
 	// dispatch options but the one that is supported by all underlying FD
 	// types.
@@ -227,7 +226,10 @@ type Options struct {
 	// AFXDPFD is used with the experimental AF_XDP mode.
 	// TODO(b/240191988): Use multiple sockets.
 	// TODO(b/240191988): How do we handle the MTU issue?
-	AFXDPFD int
+	AFXDPFD *int
+
+	// InterfaceIndex is the interface index of the underlying device.
+	InterfaceIndex int
 }
 
 // fanoutID is used for AF_PACKET based endpoints to enable PACKET_FANOUT
@@ -295,8 +297,8 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 		}
 	}
 
-	// Increment fanoutID to ensure that we don't re-use the same fanoutID for
-	// the next endpoint.
+	// Increment fanoutID to ensure that we don't re-use the same fanoutID
+	// for the next endpoint.
 	fid := fanoutID.Add(1)
 
 	// Create per channel dispatchers.
@@ -320,6 +322,13 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 				e.gsoMaxSize = opts.GSOMaxSize
 			}
 		}
+
+		// TODO(b/240191988): Remove this check once we support
+		// multiple AF_XDP sockets.
+		if opts.AFXDPFD != nil {
+			continue
+		}
+
 		inboundDispatcher, err := createInboundDispatcher(e, fd, isSocket, fid)
 		if err != nil {
 			return nil, fmt.Errorf("createInboundDispatcher(...) = %v", err)
@@ -327,6 +336,15 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 		e.inboundDispatchers = append(e.inboundDispatchers, inboundDispatcher)
 	}
 
+	if opts.AFXDPFD != nil {
+		fd := *opts.AFXDPFD
+		inboundDispatcher, err := newAFXDPDispatcher(fd, e, opts.InterfaceIndex)
+		if err != nil {
+			return nil, fmt.Errorf("newAFXDPDispatcher(%d, %+v) = %v", fd, e, err)
+		}
+		e.inboundDispatchers = append(e.inboundDispatchers, inboundDispatcher)
+	}
+
 	return e, nil
 }
 
@@ -385,8 +403,9 @@ func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32) (lin
 			if err != nil {
 				return nil, fmt.Errorf("newRecvMMsgDispatcher(%d, %+v) = %v", fd, e, err)
 			}
-		case AFXDP:
-			return nil, errors.New("AFXDP not yet implemented")
+		case Readv:
+		default:
+			return nil, fmt.Errorf("unknown dispatch mode %d", e.packetDispatchMode)
 		}
 	}
 	return inboundDispatcher, nil

pkg/tcpip/link/fdbased/xdp.go

@@ -0,0 +1,146 @@
+// Copyright 2022 The gVisor Authors.
+//
+// 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.
+
+//go:build (linux && amd64) || (linux && arm64)
+// +build linux,amd64 linux,arm64
+
+package fdbased
+
+import (
+	"fmt"
+
+	"golang.org/x/sys/unix"
+	"gvisor.dev/gvisor/pkg/bufferv2"
+	"gvisor.dev/gvisor/pkg/tcpip"
+	"gvisor.dev/gvisor/pkg/tcpip/header"
+	"gvisor.dev/gvisor/pkg/tcpip/link/rawfile"
+	"gvisor.dev/gvisor/pkg/tcpip/stack"
+	"gvisor.dev/gvisor/pkg/xdp"
+)
+
+// xdpDispatcher utilizes AF_XDP to dispatch incoming packets.
+//
+// xdpDispatcher is experimental and should not be used in production.
+type xdpDispatcher struct {
+	// stopFd enables the dispatched to be stopped via stop().
+	stopFd
+
+	// ep is the endpoint this dispatcher is attached to.
+	ep *endpoint
+
+	// fd is the AF_XDP socket FD.
+	fd int
+
+	// The following control the AF_XDP socket.
+	umem      *xdp.UMEM
+	fillQueue *xdp.FillQueue
+	rxQueue   *xdp.RXQueue
+}
+
+func newAFXDPDispatcher(fd int, ep *endpoint, index int) (linkDispatcher, error) {
+	stopFd, err := newStopFd()
+	if err != nil {
+		return nil, err
+	}
+	dispatcher := xdpDispatcher{
+		stopFd: stopFd,
+		fd:     fd,
+		ep:     ep,
+	}
+
+	// Use a 2MB UMEM to match the PACKET_MMAP dispatcher.
+	opts := xdp.DefaultReadOnlyOpts()
+	opts.NFrames = (1 << 21) / opts.FrameSize
+	dispatcher.umem, dispatcher.fillQueue, dispatcher.rxQueue, err = xdp.ReadOnlyFromSocket(fd, uint32(index), 0 /* queueID */, opts)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create AF_XDP dispatcher: %v", err)
+	}
+	dispatcher.fillQueue.FillAll()
+	return &dispatcher, nil
+}
+
+func (xd *xdpDispatcher) dispatch() (bool, tcpip.Error) {
+	for {
+		stopped, errno := rawfile.BlockingPollUntilStopped(xd.efd, xd.fd, unix.POLLIN|unix.POLLERR)
+		if errno != 0 {
+			if errno == unix.EINTR {
+				continue
+			}
+			return !stopped, rawfile.TranslateErrno(errno)
+		}
+		if stopped {
+			return true, nil
+		}
+
+		// Avoid the cost of the poll syscall if possible by peeking
+		// until there are no packets left.
+		for {
+			xd.fillQueue.FillAll()
+
+			// We can receive multiple packets at once.
+			nReceived, rxIndex := xd.rxQueue.Peek()
+
+			if nReceived == 0 {
+				break
+			}
+
+			for i := uint32(0); i < nReceived; i++ {
+				// Copy packet bytes into a view and free up the
+				// buffer.
+				descriptor := xd.rxQueue.Get(rxIndex + i)
+				data := xd.umem.Get(descriptor)
+				view := bufferv2.NewView(int(descriptor.Len))
+				view.Write(data)
+				xd.umem.FreeFrame(descriptor.Addr)
+
+				// Determine the network protocol.
+				var netProto tcpip.NetworkProtocolNumber
+				if xd.ep.hdrSize > 0 {
+					netProto = header.Ethernet(data).Type()
+				} else {
+					// We don't get any indication of what the packet is, so try to guess
+					// if it's an IPv4 or IPv6 packet.
+					switch header.IPVersion(data) {
+					case header.IPv4Version:
+						netProto = header.IPv4ProtocolNumber
+					case header.IPv6Version:
+						netProto = header.IPv6ProtocolNumber
+					default:
+						return true, nil
+					}
+				}
+
+				// Wrap the packet in a PacketBuffer and send it up the stack.
+				pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
+					Payload: bufferv2.MakeWithView(view),
+				})
+				if xd.ep.hdrSize > 0 {
+					if _, ok := pkt.LinkHeader().Consume(xd.ep.hdrSize); !ok {
+						panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", xd.ep.hdrSize))
+					}
+				}
+				xd.ep.dispatcher.DeliverNetworkPacket(netProto, pkt)
+				pkt.DecRef()
+			}
+			// Tell the kernel that we're done with these packets.
+			xd.rxQueue.Release(nReceived)
+		}
+
+		return true, nil
+	}
+}
+
+func (*xdpDispatcher) release() {
+	// Noop: let the kernel clean up.
+}

runsc/boot/network.go

@@ -88,6 +88,7 @@ type Neighbor struct {
 // FDBasedLink configures an fd-based link.
 type FDBasedLink struct {
 	Name              string
+	InterfaceIndex    int
 	MTU               int
 	Addresses         []IPWithPrefix
 	Routes            []Route
@@ -165,7 +166,7 @@ func (n *Network) CreateLinksAndRoutes(args *CreateLinksAndRoutesArgs, _ *struct
 		wantFDs += l.NumChannels
 	}
 	if args.AFXDP {
-		wantFDs++
+		wantFDs += 4
 	}
 	if got := len(args.FilePayload.Files); got != wantFDs {
 		return fmt.Errorf("args.FilePayload.Files has %d FDs but we need %d entries based on FDBasedLinks. AFXDP is %t", got, wantFDs, args.AFXDP)
@@ -234,15 +235,28 @@ func (n *Network) CreateLinksAndRoutes(args *CreateLinksAndRoutesArgs, _ *struct
 
 		// If AFXDP is enabled, we perform RX via AF_XDP and TX via
 		// AF_PACKET.
-		AFXDPFD := -1
+		var AFXDPFD *int
 		if args.AFXDP {
+			// Get the AF_XDP socket.
 			oldFD := args.FilePayload.Files[fdOffset].Fd()
 			newFD, err := unix.Dup(int(oldFD))
 			if err != nil {
 				return fmt.Errorf("failed to dup AF_XDP fd %v: %v", oldFD, err)
 			}
-			AFXDPFD = newFD
+			AFXDPFD = &newFD
 			fdOffset++
+
+			// The parent process sends several other FDs in order
+			// to keep them open and alive. These are for BPF
+			// programs and maps that, if closed, will break the
+			// dispatcher.
+			for _, fdName := range []string{"program-fd", "sockmap-fd", "link-fd"} {
+				oldFD := args.FilePayload.Files[fdOffset].Fd()
+				if _, err := unix.Dup(int(oldFD)); err != nil {
+					return fmt.Errorf("failed to dup %s with FD %d: %v", fdName, oldFD, err)
+				}
+				fdOffset++
+			}
 		}
 
 		mac := tcpip.LinkAddress(link.LinkAddress)
@@ -259,6 +273,7 @@ func (n *Network) CreateLinksAndRoutes(args *CreateLinksAndRoutesArgs, _ *struct
 			GvisorGSOEnabled:   link.GvisorGSOEnabled,
 			TXChecksumOffload:  link.TXChecksumOffload,
 			RXChecksumOffload:  link.RXChecksumOffload,
+			InterfaceIndex:     link.InterfaceIndex,
 		})
 		if err != nil {
 			return err

runsc/sandbox/BUILD

@@ -10,6 +10,9 @@ go_library(
         "network_unsafe.go",
         "sandbox.go",
     ],
+    embedsrcs = [
+        "//runsc/sandbox/bpf:af_xdp_ebpf.o",  # keep
+    ],
     visibility = [
         "//runsc:__subpackages__",
     ],
@@ -35,6 +38,8 @@ go_library(
         "//runsc/donation",
         "//runsc/specutils",
         "@com_github_cenkalti_backoff//:go_default_library",
+        "@com_github_cilium_ebpf//:go_default_library",
+        "@com_github_cilium_ebpf//link:go_default_library",
         "@com_github_opencontainers_runtime_spec//specs-go:go_default_library",
         "@com_github_syndtr_gocapability//capability:go_default_library",
         "@com_github_vishvananda_netlink//:go_default_library",

runsc/sandbox/bpf/BUILD

@@ -0,0 +1,11 @@
+load("//tools:defs.bzl", "bpf_program")
+
+package(licenses = ["notice"])
+
+bpf_program(
+    name = "af_xdp_ebpf",
+    src = "af_xdp.ebpf.c",
+    hdrs = [],
+    bpf_object = "af_xdp_ebpf.o",
+    visibility = ["//:sandbox"],
+)

runsc/sandbox/bpf/af_xdp.ebpf.c

@@ -0,0 +1,51 @@
+// Copyright 2022 The gVisor Authors.
+//
+// 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.
+
+#include <linux/bpf.h>
+
+#define section(secname) __attribute__((section(secname), used))
+
+char __license[] section("license") = "Apache-2.0";
+
+// Helper functions are defined positionally in <linux/bpf.h>, and their
+// signatures are scattered throughout the kernel. They can be found via the
+// defining macro BPF_CALL_[0-5].
+// TODO(b/240191988): Use vmlinux instead of this.
+static int (*bpf_redirect_map)(void *bpf_map, __u32 iface_index,
+                               __u64 flags) = (void *)51;
+
+struct bpf_map_def {
+  unsigned int type;
+  unsigned int key_size;
+  unsigned int value_size;
+  unsigned int max_entries;
+  unsigned int map_flags;
+};
+
+// A map of RX queue number to AF_XDP socket. We only ever use one key: 0.
+struct bpf_map_def section("maps") sock_map = {
+    .type = BPF_MAP_TYPE_XSKMAP,  // Note: "XSK" means AF_XDP socket.
+    .key_size = sizeof(int),
+    .value_size = sizeof(int),
+    .max_entries = 1,
+};
+
+section("xdp") int xdp_prog(struct xdp_md *ctx) {
+  // Lookup the socket for the current RX queue. Veth devices by default have
+  // only one RX queue. If one is found, redirect the packet to that socket.
+  // Otherwise pass it on to the kernel network stack.
+  //
+  // TODO: We can support multiple sockets with a fancier hash-based handoff.
+  return bpf_redirect_map(&sock_map, ctx->rx_queue_index, XDP_PASS);
+}

runsc/sandbox/network.go

@@ -15,6 +15,8 @@
 package sandbox
 
 import (
+	"bytes"
+	_ "embed"
 	"fmt"
 	"net"
 	"os"
@@ -22,6 +24,8 @@ import (
 	"runtime"
 	"strconv"
 
+	"github.com/cilium/ebpf"
+	"github.com/cilium/ebpf/link"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
@@ -217,6 +221,7 @@ func createInterfacesAndRoutesFromNS(conn *urpc.Client, nsPath string, conf *con
 
 		link := boot.FDBasedLink{
 			Name:              iface.Name,
+			InterfaceIndex:    iface.Index,
 			MTU:               iface.MTU,
 			Routes:            routes,
 			TXChecksumOffload: conf.TXChecksumOffload,
@@ -254,11 +259,11 @@ func createInterfacesAndRoutesFromNS(conn *urpc.Client, nsPath string, conf *con
 
 		// If enabled, create an RX socket for AF_XDP.
 		if conf.AFXDP {
-			xdpSock, err := createSocketXDP(iface)
+			xdpSockFDs, err := createSocketXDP(iface)
 			if err != nil {
 				return fmt.Errorf("failed to create XDP socket: %v", err)
 			}
-			args.FilePayload.Files = append(args.FilePayload.Files, xdpSock)
+			args.FilePayload.Files = append(args.FilePayload.Files, xdpSockFDs...)
 		}
 
 		if link.GSOMaxSize == 0 && conf.GvisorGSO {
@@ -387,15 +392,80 @@ func createSocket(iface net.Interface, ifaceLink netlink.Link, enableGSO bool, A
 	return &socketEntry{deviceFile, gsoMaxSize}, nil
 }
 
-func createSocketXDP(iface net.Interface) (*os.File, error) {
+// program is the BPF program to attach to the socket.
+//
+//go:embed bpf/af_xdp_ebpf.o
+var program []byte
+
+func createSocketXDP(iface net.Interface) ([]*os.File, error) {
 	// Create an XDP socket. The sentry will mmap memory for the various
 	// rings and bind to the device.
 	fd, err := unix.Socket(unix.AF_XDP, unix.SOCK_RAW, 0)
 	if err != nil {
 		return nil, fmt.Errorf("unable to create AF_XDP socket: %v", err)
 	}
-	deviceFile := os.NewFile(uintptr(fd), "xdp-fd")
-	return deviceFile, nil
+
+	// We also need to, before dropping privileges, attach a program to the
+	// device and insert our socket into its map.
+
+	// Load into the kernel.
+	spec, err := ebpf.LoadCollectionSpecFromReader(bytes.NewReader(program))
+	if err != nil {
+		return nil, fmt.Errorf("failed to load spec: %v", err)
+	}
+
+	var objects struct {
+		Program *ebpf.Program `ebpf:"xdp_prog"`
+		SockMap *ebpf.Map     `ebpf:"sock_map"`
+	}
+	if err := spec.LoadAndAssign(&objects, nil); err != nil {
+		return nil, fmt.Errorf("failed to load program: %v", err)
+	}
+
+	rawLink, err := link.AttachRawLink(link.RawLinkOptions{
+		Program: objects.Program,
+		Attach:  ebpf.AttachXDP,
+		Target:  iface.Index,
+		// By not setting the Flag field, the kernel will choose the
+		// fastest mode. In order those are:
+		// - Offloaded onto the NIC.
+		// - Running directly in the driver.
+		// - Generic mode, which works with any NIC/driver but lacks
+		//   much of the XDP performance boost.
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to attach BPF program: %v", err)
+	}
+
+	// Insert our AF_XDP socket into the BPF map that dictates where
+	// packets are redirected to.
+	key := uint32(0)
+	val := uint32(fd)
+	if err := objects.SockMap.Update(&key, &val, 0 /* flags */); err != nil {
+		return nil, fmt.Errorf("failed to insert socket into BPF map: %v", err)
+	}
+
+	// We need to keep the Program, SockMap, and link FDs open until they
+	// can be passed to the sandbox process.
+	progFD, err := unix.Dup(objects.Program.FD())
+	if err != nil {
+		return nil, fmt.Errorf("failed to dup BPF program: %v", err)
+	}
+	sockMapFD, err := unix.Dup(objects.SockMap.FD())
+	if err != nil {
+		return nil, fmt.Errorf("failed to dup BPF map: %v", err)
+	}
+	linkFD, err := unix.Dup(rawLink.FD())
+	if err != nil {
+		return nil, fmt.Errorf("failed to dup BPF link: %v", err)
+	}
+
+	return []*os.File{
+		os.NewFile(uintptr(fd), "xdp-fd"),            // The socket.
+		os.NewFile(uintptr(progFD), "program-fd"),    // The XDP program.
+		os.NewFile(uintptr(sockMapFD), "sockmap-fd"), // The XDP map.
+		os.NewFile(uintptr(linkFD), "link-fd"),       // The XDP link.
+	}, nil
 }
 
 // loopbackLink returns the link with addresses and routes for a loopback