From 6840864b0e3edcf2688ba705676144e91c0679b5 Mon Sep 17 00:00:00 2001 From: Kevin Krakauer Date: Tue, 6 Sep 2022 11:14:41 -0700 Subject: [PATCH] experimental xdp dispatcher PiperOrigin-RevId: 472509327 --- pkg/tcpip/BUILD | 1 + pkg/tcpip/link/fdbased/BUILD | 2 + pkg/tcpip/link/fdbased/endpoint.go | 41 +++++--- pkg/tcpip/link/fdbased/xdp.go | 146 +++++++++++++++++++++++++++++ runsc/boot/network.go | 21 ++++- runsc/sandbox/BUILD | 5 + runsc/sandbox/bpf/BUILD | 11 +++ runsc/sandbox/bpf/af_xdp.ebpf.c | 51 ++++++++++ runsc/sandbox/network.go | 80 +++++++++++++++- 9 files changed, 339 insertions(+), 19 deletions(-) create mode 100644 pkg/tcpip/link/fdbased/xdp.go create mode 100644 runsc/sandbox/bpf/BUILD create mode 100644 runsc/sandbox/bpf/af_xdp.ebpf.c diff --git a/pkg/tcpip/BUILD b/pkg/tcpip/BUILD index 9d1fce825..3408ff270 100644 --- a/pkg/tcpip/BUILD +++ b/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", diff --git a/pkg/tcpip/link/fdbased/BUILD b/pkg/tcpip/link/fdbased/BUILD index b61011a76..02b6b7747 100644 --- a/pkg/tcpip/link/fdbased/BUILD +++ b/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", ], ) diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index 9a4341b83..7437480ed 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/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 diff --git a/pkg/tcpip/link/fdbased/xdp.go b/pkg/tcpip/link/fdbased/xdp.go new file mode 100644 index 000000000..b0830165b --- /dev/null +++ b/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. +} diff --git a/runsc/boot/network.go b/runsc/boot/network.go index a64085506..ffcc41680 100644 --- a/runsc/boot/network.go +++ b/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 diff --git a/runsc/sandbox/BUILD b/runsc/sandbox/BUILD index 65d2b557f..3e08d0a9c 100644 --- a/runsc/sandbox/BUILD +++ b/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", diff --git a/runsc/sandbox/bpf/BUILD b/runsc/sandbox/bpf/BUILD new file mode 100644 index 000000000..9a7bea97a --- /dev/null +++ b/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"], +) diff --git a/runsc/sandbox/bpf/af_xdp.ebpf.c b/runsc/sandbox/bpf/af_xdp.ebpf.c new file mode 100644 index 000000000..7f3b46093 --- /dev/null +++ b/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 + +#define section(secname) __attribute__((section(secname), used)) + +char __license[] section("license") = "Apache-2.0"; + +// Helper functions are defined positionally in , 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); +} diff --git a/runsc/sandbox/network.go b/runsc/sandbox/network.go index 1cb6584e0..1081e9f5b 100644 --- a/runsc/sandbox/network.go +++ b/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