diff --git a/pkg/tcpip/link/fdbased/BUILD b/pkg/tcpip/link/fdbased/BUILD index 02b6b7747..f7f7073c8 100644 --- a/pkg/tcpip/link/fdbased/BUILD +++ b/pkg/tcpip/link/fdbased/BUILD @@ -11,7 +11,6 @@ go_library( "mmap_stub.go", "mmap_unsafe.go", "packet_dispatchers.go", - "xdp.go", ], visibility = ["//visibility:public"], deps = [ @@ -21,8 +20,8 @@ go_library( "//pkg/tcpip", "//pkg/tcpip/header", "//pkg/tcpip/link/rawfile", + "//pkg/tcpip/link/stopfd", "//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 7437480ed..05c96ada2 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/pkg/tcpip/link/fdbased/endpoint.go @@ -56,7 +56,7 @@ import ( // linkDispatcher reads packets from the link FD and dispatches them to the // NetworkDispatcher. type linkDispatcher interface { - stop() + Stop() dispatch() (bool, tcpip.Error) release() } @@ -88,9 +88,6 @@ const ( // primary use-case for this is runsc which uses an AF_PACKET FD to // receive packets from the veth device. PacketMMap - // AFXDP utilizes an AF_XDP socket to receive packets. AFXDP requires that - // the underlying FD be an AF_XDP socket. - AFXDP ) func (p PacketDispatchMode) String() string { @@ -101,8 +98,6 @@ func (p PacketDispatchMode) String() string { return "RecvMMsg" case PacketMMap: return "PacketMMap" - case AFXDP: - return "AFXDP" default: return fmt.Sprintf("unknown packet dispatch mode '%d'", p) } @@ -323,12 +318,6 @@ func New(opts *Options) (stack.LinkEndpoint, error) { } } - // 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) @@ -336,15 +325,6 @@ 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 } @@ -420,12 +400,15 @@ func isSocketFD(fd int) (bool, error) { } // Attach launches the goroutine that reads packets from the file descriptor and -// dispatches them via the provided dispatcher. +// dispatches them via the provided dispatcher. If one is already attached, +// then nothing happens. +// +// Attach implements stack.LinkEndpoint.Attach. func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) { // nil means the NIC is being removed. if dispatcher == nil && e.dispatcher != nil { for _, dispatcher := range e.inboundDispatchers { - dispatcher.stop() + dispatcher.Stop() } e.Wait() e.dispatcher = nil @@ -739,12 +722,7 @@ func (e *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) return sentPackets, nil } -// viewsEqual tests whether v1 and v2 refer to the same backing bytes. -func viewsEqual(vs1, vs2 []bufferv2.View) bool { - return len(vs1) == len(vs2) && (len(vs1) == 0 || &vs1[0] == &vs2[0]) -} - -// InjectOutobund implements stack.InjectableEndpoint.InjectOutbound. +// InjectOutbound implements stack.InjectableEndpoint.InjectOutbound. func (e *endpoint) InjectOutbound(dest tcpip.Address, packet *bufferv2.View) tcpip.Error { return rawfile.NonBlockingWrite(e.fds[0].fd, packet.AsSlice()) } diff --git a/pkg/tcpip/link/fdbased/mmap.go b/pkg/tcpip/link/fdbased/mmap.go index ebb9ad778..743bb912c 100644 --- a/pkg/tcpip/link/fdbased/mmap.go +++ b/pkg/tcpip/link/fdbased/mmap.go @@ -26,6 +26,7 @@ import ( "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" "gvisor.dev/gvisor/pkg/tcpip/link/rawfile" + "gvisor.dev/gvisor/pkg/tcpip/link/stopfd" "gvisor.dev/gvisor/pkg/tcpip/stack" ) @@ -115,7 +116,7 @@ func (t tPacketHdr) Payload() []byte { // packetMMapDispatcher uses PACKET_RX_RING's to read/dispatch inbound packets. // See: mmap_amd64_unsafe.go for implementation details. type packetMMapDispatcher struct { - stopFd + stopfd.StopFD // fd is the file descriptor used to send and receive packets. fd int @@ -136,7 +137,7 @@ func (*packetMMapDispatcher) release() {} func (d *packetMMapDispatcher) readMMappedPacket() (*bufferv2.View, bool, tcpip.Error) { hdr := tPacketHdr(d.ringBuffer[d.ringOffset*tpFrameSize:]) for hdr.tpStatus()&tpStatusUser == 0 { - stopped, errno := rawfile.BlockingPollUntilStopped(d.efd, d.fd, unix.POLLIN|unix.POLLERR) + stopped, errno := rawfile.BlockingPollUntilStopped(d.EFD, d.fd, unix.POLLIN|unix.POLLERR) if errno != 0 { if errno == unix.EINTR { continue diff --git a/pkg/tcpip/link/fdbased/mmap_unsafe.go b/pkg/tcpip/link/fdbased/mmap_unsafe.go index 7d62aa2a6..abe07a5e1 100644 --- a/pkg/tcpip/link/fdbased/mmap_unsafe.go +++ b/pkg/tcpip/link/fdbased/mmap_unsafe.go @@ -23,6 +23,7 @@ import ( "golang.org/x/sys/unix" "gvisor.dev/gvisor/pkg/atomicbitops" + "gvisor.dev/gvisor/pkg/tcpip/link/stopfd" ) // tPacketHdrlen is the TPACKET_HDRLEN variable defined in . @@ -47,12 +48,12 @@ func (t tPacketHdr) setTPStatus(status uint32) { } func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) { - stopFd, err := newStopFd() + stopFD, err := stopfd.New() if err != nil { return nil, err } d := &packetMMapDispatcher{ - stopFd: stopFd, + StopFD: stopFD, fd: fd, e: e, } diff --git a/pkg/tcpip/link/fdbased/packet_dispatchers.go b/pkg/tcpip/link/fdbased/packet_dispatchers.go index 0ab67212e..13bbf494e 100644 --- a/pkg/tcpip/link/fdbased/packet_dispatchers.go +++ b/pkg/tcpip/link/fdbased/packet_dispatchers.go @@ -18,13 +18,12 @@ 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/link/stopfd" "gvisor.dev/gvisor/pkg/tcpip/stack" ) @@ -143,36 +142,10 @@ func (b *iovecBuffer) release() { } } -// stopFd is an eventfd used to signal the stop of a dispatcher. -type stopFd struct { - efd int -} - -func newStopFd() (stopFd, error) { - efd, err := unix.Eventfd(0, unix.EFD_NONBLOCK) - if err != nil { - return stopFd{efd: -1}, fmt.Errorf("failed to create eventfd: %w", err) - } - return stopFd{efd: efd}, nil -} - -// stop writes to the eventfd and notifies the dispatcher to stop. It does not -// block. -func (s *stopFd) stop() { - increment := []byte{1, 0, 0, 0, 0, 0, 0, 0} - if n, err := unix.Write(s.efd, increment); n != len(increment) || err != nil { - // There are two possible errors documented in eventfd(2) for writing: - // 1. We are writing 8 bytes and not 0xffffffffffffff, thus no EINVAL. - // 2. stop is only supposed to be called once, it can't reach the limit, - // thus no EAGAIN. - panic(fmt.Sprintf("write(efd) = (%d, %s), want (%d, nil)", n, err, len(increment))) - } -} - // readVDispatcher uses readv() system call to read inbound packets and // dispatches them. type readVDispatcher struct { - stopFd + stopfd.StopFD // fd is the file descriptor used to send and receive packets. fd int @@ -184,12 +157,12 @@ type readVDispatcher struct { } func newReadVDispatcher(fd int, e *endpoint) (linkDispatcher, error) { - stopFd, err := newStopFd() + stopFD, err := stopfd.New() if err != nil { return nil, err } d := &readVDispatcher{ - stopFd: stopFd, + StopFD: stopFD, fd: fd, e: e, } @@ -204,7 +177,7 @@ func (d *readVDispatcher) release() { // dispatch reads one packet from the file descriptor and dispatches it. func (d *readVDispatcher) dispatch() (bool, tcpip.Error) { - n, err := rawfile.BlockingReadvUntilStopped(d.efd, d.fd, d.buf.nextIovecs()) + n, err := rawfile.BlockingReadvUntilStopped(d.EFD, d.fd, d.buf.nextIovecs()) if n <= 0 || err != nil { return false, err } @@ -247,7 +220,7 @@ func (d *readVDispatcher) dispatch() (bool, tcpip.Error) { // recvMMsgDispatcher uses the recvmmsg system call to read inbound packets and // dispatches them. type recvMMsgDispatcher struct { - stopFd + stopfd.StopFD // fd is the file descriptor used to send and receive packets. fd int @@ -271,12 +244,12 @@ const ( ) func newRecvMMsgDispatcher(fd int, e *endpoint) (linkDispatcher, error) { - stopFd, err := newStopFd() + stopFD, err := stopfd.New() if err != nil { return nil, err } d := &recvMMsgDispatcher{ - stopFd: stopFd, + StopFD: stopFD, fd: fd, e: e, bufs: make([]*iovecBuffer, MaxMsgsPerRecv), @@ -310,7 +283,7 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) { d.msgHdrs[k].Msg.SetIovlen(iovLen) } - nMsgs, err := rawfile.BlockingRecvMMsgUntilStopped(d.efd, d.fd, d.msgHdrs) + nMsgs, err := rawfile.BlockingRecvMMsgUntilStopped(d.EFD, d.fd, d.msgHdrs) if nMsgs == -1 || err != nil { return false, err } diff --git a/pkg/tcpip/link/stopfd/BUILD b/pkg/tcpip/link/stopfd/BUILD new file mode 100644 index 000000000..a299f1e03 --- /dev/null +++ b/pkg/tcpip/link/stopfd/BUILD @@ -0,0 +1,14 @@ +load("//tools:defs.bzl", "go_library") + +package(licenses = ["notice"]) + +go_library( + name = "stopfd", + srcs = [ + "stopfd.go", + ], + visibility = ["//visibility:public"], + deps = [ + "@org_golang_x_sys//unix:go_default_library", + ], +) diff --git a/pkg/tcpip/link/stopfd/stopfd.go b/pkg/tcpip/link/stopfd/stopfd.go new file mode 100644 index 000000000..cc46baa77 --- /dev/null +++ b/pkg/tcpip/link/stopfd/stopfd.go @@ -0,0 +1,52 @@ +// 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 stopfd provides an type that can be used to signal the stop of a dispatcher. +package stopfd + +import ( + "fmt" + + "golang.org/x/sys/unix" +) + +// StopFD is an eventfd used to signal the stop of a dispatcher. +type StopFD struct { + EFD int +} + +// New returns a new, initialized StopFD. +func New() (StopFD, error) { + efd, err := unix.Eventfd(0, unix.EFD_NONBLOCK) + if err != nil { + return StopFD{EFD: -1}, fmt.Errorf("failed to create eventfd: %w", err) + } + return StopFD{EFD: efd}, nil +} + +// Stop writes to the eventfd and notifies the dispatcher to stop. It does not +// block. +func (sf *StopFD) Stop() { + increment := []byte{1, 0, 0, 0, 0, 0, 0, 0} + if n, err := unix.Write(sf.EFD, increment); n != len(increment) || err != nil { + // There are two possible errors documented in eventfd(2) for writing: + // 1. We are writing 8 bytes and not 0xffffffffffffff, thus no EINVAL. + // 2. stop is only supposed to be called once, it can't reach the limit, + // thus no EAGAIN. + panic(fmt.Sprintf("write(EFD) = (%d, %s), want (%d, nil)", n, err, len(increment))) + } +} diff --git a/pkg/tcpip/link/xdp/BUILD b/pkg/tcpip/link/xdp/BUILD new file mode 100644 index 000000000..58d8b8531 --- /dev/null +++ b/pkg/tcpip/link/xdp/BUILD @@ -0,0 +1,23 @@ +load("//tools:defs.bzl", "go_library") + +package(licenses = ["notice"]) + +go_library( + name = "xdp", + srcs = [ + "dispatcher.go", + "endpoint.go", + ], + visibility = ["//visibility:public"], + deps = [ + "//pkg/bufferv2", + "//pkg/sync", + "//pkg/tcpip", + "//pkg/tcpip/header", + "//pkg/tcpip/link/rawfile", + "//pkg/tcpip/link/stopfd", + "//pkg/tcpip/stack", + "//pkg/xdp", + "@org_golang_x_sys//unix:go_default_library", + ], +) diff --git a/pkg/tcpip/link/fdbased/xdp.go b/pkg/tcpip/link/xdp/dispatcher.go similarity index 60% rename from pkg/tcpip/link/fdbased/xdp.go rename to pkg/tcpip/link/xdp/dispatcher.go index b0830165b..38adf652b 100644 --- a/pkg/tcpip/link/fdbased/xdp.go +++ b/pkg/tcpip/link/xdp/dispatcher.go @@ -15,7 +15,7 @@ //go:build (linux && amd64) || (linux && arm64) // +build linux,amd64 linux,arm64 -package fdbased +package xdp import ( "fmt" @@ -25,16 +25,16 @@ import ( "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" "gvisor.dev/gvisor/pkg/tcpip/link/rawfile" + "gvisor.dev/gvisor/pkg/tcpip/link/stopfd" "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 +// TODO(b/240191988): Handle the XDP-limited MTU. + +// dispatcher utilizes AF_XDP to dispatch incoming packets. +type dispatcher struct { + stopfd.StopFD // ep is the endpoint this dispatcher is attached to. ep *endpoint @@ -48,13 +48,13 @@ type xdpDispatcher struct { rxQueue *xdp.RXQueue } -func newAFXDPDispatcher(fd int, ep *endpoint, index int) (linkDispatcher, error) { - stopFd, err := newStopFd() +func (xd *dispatcher) init(fd int, ep *endpoint, index int) error { + stopFD, err := stopfd.New() if err != nil { - return nil, err + return err } - dispatcher := xdpDispatcher{ - stopFd: stopFd, + disp := dispatcher{ + StopFD: stopFD, fd: fd, ep: ep, } @@ -62,17 +62,17 @@ func newAFXDPDispatcher(fd int, ep *endpoint, index int) (linkDispatcher, error) // 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) + disp.umem, disp.fillQueue, disp.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) + return fmt.Errorf("failed to create AF_XDP dispatcher: %v", err) } - dispatcher.fillQueue.FillAll() - return &dispatcher, nil + disp.fillQueue.FillAll() + return nil } -func (xd *xdpDispatcher) dispatch() (bool, tcpip.Error) { +func (xd *dispatcher) dispatch() (bool, tcpip.Error) { for { - stopped, errno := rawfile.BlockingPollUntilStopped(xd.efd, xd.fd, unix.POLLIN|unix.POLLERR) + stopped, errno := rawfile.BlockingPollUntilStopped(xd.EFD, xd.fd, unix.POLLIN|unix.POLLERR) if errno != 0 { if errno == unix.EINTR { continue @@ -104,33 +104,16 @@ func (xd *xdpDispatcher) dispatch() (bool, tcpip.Error) { 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 - } - } + netProto := header.Ethernet(data).Type() // 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)) - } + if _, ok := pkt.LinkHeader().Consume(header.EthernetMinimumSize); !ok { + panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", header.EthernetMinimumSize)) } - xd.ep.dispatcher.DeliverNetworkPacket(netProto, pkt) + xd.ep.networkDispatcher.DeliverNetworkPacket(netProto, pkt) pkt.DecRef() } // Tell the kernel that we're done with these packets. @@ -140,7 +123,3 @@ func (xd *xdpDispatcher) dispatch() (bool, tcpip.Error) { return true, nil } } - -func (*xdpDispatcher) release() { - // Noop: let the kernel clean up. -} diff --git a/pkg/tcpip/link/xdp/endpoint.go b/pkg/tcpip/link/xdp/endpoint.go new file mode 100644 index 000000000..b532a052f --- /dev/null +++ b/pkg/tcpip/link/xdp/endpoint.go @@ -0,0 +1,222 @@ +// Copyright 2018 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 +// +build linux + +// Package xdp provides link layer endpoints backed by AF_XDP sockets. +package xdp + +import ( + "fmt" + + "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/sync" + "gvisor.dev/gvisor/pkg/tcpip" + "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/stack" +) + +var _ stack.LinkEndpoint = (*endpoint)(nil) + +type endpoint struct { + fd int + + // mtu (maximum transmission unit) is the maximum size of a packet. + mtu uint32 + + // addr is the address of the endpoint. + addr tcpip.LinkAddress + + // caps holds the endpoint capabilities. + caps stack.LinkEndpointCapabilities + + // closed is a function to be called when the FD's peer (if any) closes + // its end of the communication pipe. + closed func(tcpip.Error) + + inboundDispatcher dispatcher + networkDispatcher stack.NetworkDispatcher + + // wg keeps track of running goroutines. + wg sync.WaitGroup +} + +// Options specify the details about the fd-based endpoint to be created. +type Options struct { + // FD is used to read/write packets. + FD int + + // MTU is the mtu to use for this endpoint. + MTU uint32 + + // ClosedFunc is a function to be called when an endpoint's peer (if + // any) closes its end of the communication pipe. + ClosedFunc func(tcpip.Error) + + // Address is the link address for this endpoint. + Address tcpip.LinkAddress + + // SaveRestore if true, indicates that this NIC capability set should + // include CapabilitySaveRestore + SaveRestore bool + + // DisconnectOk if true, indicates that this NIC capability set should + // include CapabilityDisconnectOk. + DisconnectOk bool + + // TXChecksumOffload if true, indicates that this endpoints capability + // set should include CapabilityTXChecksumOffload. + TXChecksumOffload bool + + // RXChecksumOffload if true, indicates that this endpoints capability + // set should include CapabilityRXChecksumOffload. + RXChecksumOffload bool + + // InterfaceIndex is the interface index of the underlying device. + InterfaceIndex int +} + +// New creates a new endpoint from an AF_XDP socket. +func New(opts *Options) (stack.LinkEndpoint, error) { + caps := stack.CapabilityResolutionRequired + if opts.RXChecksumOffload { + caps |= stack.CapabilityRXChecksumOffload + } + + if opts.TXChecksumOffload { + caps |= stack.CapabilityTXChecksumOffload + } + + if opts.SaveRestore { + caps |= stack.CapabilitySaveRestore + } + + if opts.DisconnectOk { + caps |= stack.CapabilityDisconnectOk + } + + if err := unix.SetNonblock(opts.FD, true); err != nil { + return nil, fmt.Errorf("unix.SetNonblock(%v) failed: %v", opts.FD, err) + } + + ep := &endpoint{ + fd: opts.FD, + mtu: opts.MTU, + caps: caps, + closed: opts.ClosedFunc, + addr: opts.Address, + } + + if err := ep.inboundDispatcher.init(opts.FD, ep, opts.InterfaceIndex); err != nil { + return nil, fmt.Errorf("ep.inboundDispatcher.init(%d, %+v) = %v", opts.FD, ep, err) + } + + return ep, nil +} + +// Attach launches the goroutine that reads packets from the file descriptor and +// dispatches them via the provided dispatcher. If one is already attached, +// then nothing happens. +// +// Attach implements stack.LinkEndpoint.Attach. +func (ep *endpoint) Attach(networkDispatcher stack.NetworkDispatcher) { + // nil means the NIC is being removed. + if networkDispatcher == nil && ep.IsAttached() { + ep.inboundDispatcher.Stop() + ep.Wait() + ep.networkDispatcher = nil + return + } + if networkDispatcher != nil && ep.networkDispatcher == nil { + ep.networkDispatcher = networkDispatcher + // Link endpoints are not savable. When transportation endpoints are + // saved, they stop sending outgoing packets and all incoming packets + // are rejected. + ep.wg.Add(1) + go func() { // S/R-SAFE: See above. + defer ep.wg.Done() + for { + cont, err := ep.inboundDispatcher.dispatch() + if err != nil || !cont { + if ep.closed != nil { + ep.closed(err) + } + return + } + } + }() + } +} + +// IsAttached implements stack.LinkEndpoint.IsAttached. +func (ep *endpoint) IsAttached() bool { + return ep.networkDispatcher != nil +} + +// MTU implements stack.LinkEndpoint.MTU. It returns the value initialized +// during construction. +func (ep *endpoint) MTU() uint32 { + return ep.mtu +} + +// Capabilities implements stack.LinkEndpoint.Capabilities. +func (ep *endpoint) Capabilities() stack.LinkEndpointCapabilities { + return ep.caps +} + +// MaxHeaderLength returns the maximum size of the link-layer header. +func (ep *endpoint) MaxHeaderLength() uint16 { + return uint16(header.EthernetMinimumSize) +} + +// LinkAddress returns the link address of this endpoint. +func (ep *endpoint) LinkAddress() tcpip.LinkAddress { + return ep.addr +} + +// Wait implements stack.LinkEndpoint.Wait. It waits for the endpoint to stop +// reading from its FD. +func (ep *endpoint) Wait() { + ep.wg.Wait() +} + +// AddHeader implements stack.LinkEndpoint.AddHeader. +func (ep *endpoint) AddHeader(pkt *stack.PacketBuffer) { + // Add ethernet header if needed. + eth := header.Ethernet(pkt.LinkHeader().Push(header.EthernetMinimumSize)) + eth.Encode(&header.EthernetFields{ + SrcAddr: pkt.EgressRoute.LocalLinkAddress, + DstAddr: pkt.EgressRoute.RemoteLinkAddress, + Type: pkt.NetworkProtocolNumber, + }) +} + +// ARPHardwareType implements stack.LinkEndpoint.ARPHardwareType. +func (ep *endpoint) ARPHardwareType() header.ARPHardwareType { + return header.ARPHardwareEther +} + +// WritePackets writes outbound packets to the underlying file descriptors. If +// one is not currently writable, the packet is dropped. +// +// Each packet in pkts should have the following fields populated: +// - pkt.EgressRoute +// - pkt.NetworkProtocolNumber +// +// The following should not be populated, as GSO is not supported with XDP. +// - pkt.GSOOptions +func (ep *endpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Error) { + return 0, &tcpip.ErrNotSupported{} +} diff --git a/runsc/boot/BUILD b/runsc/boot/BUILD index 5507ce0b4..f21039bb2 100644 --- a/runsc/boot/BUILD +++ b/runsc/boot/BUILD @@ -100,6 +100,7 @@ go_library( "//pkg/tcpip/link/packetsocket", "//pkg/tcpip/link/qdisc/fifo", "//pkg/tcpip/link/sniffer", + "//pkg/tcpip/link/xdp", "//pkg/tcpip/network/arp", "//pkg/tcpip/network/ipv4", "//pkg/tcpip/network/ipv6", diff --git a/runsc/boot/network.go b/runsc/boot/network.go index 18cb022cb..5d8c2dc7d 100644 --- a/runsc/boot/network.go +++ b/runsc/boot/network.go @@ -31,6 +31,7 @@ import ( "gvisor.dev/gvisor/pkg/tcpip/link/packetsocket" "gvisor.dev/gvisor/pkg/tcpip/link/qdisc/fifo" "gvisor.dev/gvisor/pkg/tcpip/link/sniffer" + "gvisor.dev/gvisor/pkg/tcpip/link/xdp" "gvisor.dev/gvisor/pkg/tcpip/network/ipv4" "gvisor.dev/gvisor/pkg/tcpip/network/ipv6" "gvisor.dev/gvisor/pkg/tcpip/stack" @@ -106,6 +107,24 @@ type FDBasedLink struct { NumChannels int } +// XDPLink configures an XDP link. +type XDPLink struct { + Name string + InterfaceIndex int + MTU int + Addresses []IPWithPrefix + Routes []Route + TXChecksumOffload bool + RXChecksumOffload bool + LinkAddress net.HardwareAddr + QDisc config.QueueingDiscipline + Neighbors []Neighbor + + // NumChannels controls how many underlying FDs are to be used to + // create this endpoint. + NumChannels int +} + // LoopbackLink configures a loopback li nk. type LoopbackLink struct { Name string @@ -122,12 +141,11 @@ type CreateLinksAndRoutesArgs struct { LoopbackLinks []LoopbackLink FDBasedLinks []FDBasedLink + XDPLinks []XDPLink Defaultv4Gateway DefaultRoute Defaultv6Gateway DefaultRoute - AFXDP bool - // PCAP indicates that FilePayload also contains a PCAP log file. PCAP bool } @@ -165,18 +183,21 @@ func (r *Route) toTcpipRoute(id tcpip.NICID) (tcpip.Route, error) { // CreateLinksAndRoutes creates links and routes in a network stack. It should // only be called once. func (n *Network) CreateLinksAndRoutes(args *CreateLinksAndRoutesArgs, _ *struct{}) error { + if len(args.FDBasedLinks) > 0 && len(args.XDPLinks) > 0 { + return fmt.Errorf("received both fdbased and XDP links, but only one can be used at a time") + } wantFDs := 0 for _, l := range args.FDBasedLinks { wantFDs += l.NumChannels } - if args.AFXDP { + if len(args.XDPLinks) > 0 { wantFDs += 4 } if args.PCAP { wantFDs++ } 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, PCAP is %t", got, wantFDs, args.AFXDP, args.PCAP) + return fmt.Errorf("args.FilePayload.Files has %d FDs but we need %d entries based on FDBasedLinks, XDPLinks, and PCAP", got, wantFDs) } var nicID tcpip.NICID @@ -208,79 +229,126 @@ func (n *Network) CreateLinksAndRoutes(args *CreateLinksAndRoutesArgs, _ *struct } } - // Choose a dispatch mode. - dispatchMode := fdbased.RecvMMsg - version, err := hostos.KernelVersion() - if err != nil { - return err - } - if version.AtLeast(5, 6) { - dispatchMode = fdbased.PacketMMap - } else { - log.Infof("Host kernel version < 5.6, falling back to RecvMMsg dispatch") - } - if args.AFXDP { - dispatchMode = fdbased.AFXDP - } + // Setup fdbased or XDP links. + if len(args.FDBasedLinks) > 0 { + // Choose a dispatch mode. + dispatchMode := fdbased.RecvMMsg + version, err := hostos.KernelVersion() + if err != nil { + return err + } + if version.AtLeast(5, 6) { + dispatchMode = fdbased.PacketMMap + } else { + log.Infof("Host kernel version < 5.6, falling back to RecvMMsg dispatch") + } - fdOffset := 0 - for _, link := range args.FDBasedLinks { + fdOffset := 0 + for _, link := range args.FDBasedLinks { + nicID++ + nicids[link.Name] = nicID + + FDs := make([]int, 0, link.NumChannels) + for j := 0; j < link.NumChannels; j++ { + // Copy the underlying FD. + oldFD := args.FilePayload.Files[fdOffset].Fd() + newFD, err := unix.Dup(int(oldFD)) + if err != nil { + return fmt.Errorf("failed to dup FD %v: %v", oldFD, err) + } + FDs = append(FDs, newFD) + fdOffset++ + } + + mac := tcpip.LinkAddress(link.LinkAddress) + log.Infof("gso max size is: %d", link.GSOMaxSize) + + linkEP, err := fdbased.New(&fdbased.Options{ + FDs: FDs, + MTU: uint32(link.MTU), + EthernetHeader: mac != "", + Address: mac, + PacketDispatchMode: dispatchMode, + GSOMaxSize: link.GSOMaxSize, + GvisorGSOEnabled: link.GvisorGSOEnabled, + TXChecksumOffload: link.TXChecksumOffload, + RXChecksumOffload: link.RXChecksumOffload, + }) + if err != nil { + return err + } + + // Wrap linkEP in a sniffer to enable packet logging. + sniffEP := sniffer.New(packetsocket.New(linkEP)) + + var qDisc stack.QueueingDiscipline + switch link.QDisc { + case config.QDiscNone: + case config.QDiscFIFO: + log.Infof("Enabling FIFO QDisc on %q", link.Name) + qDisc = fifo.New(sniffEP, runtime.GOMAXPROCS(0), 1000) + } + + log.Infof("Enabling interface %q with id %d on addresses %+v (%v) w/ %d channels", link.Name, nicID, link.Addresses, mac, link.NumChannels) + opts := stack.NICOptions{ + Name: link.Name, + QDisc: qDisc, + } + if err := n.createNICWithAddrs(nicID, sniffEP, opts, link.Addresses); err != nil { + return err + } + + // Collect the routes from this link. + for _, r := range link.Routes { + route, err := r.toTcpipRoute(nicID) + if err != nil { + return err + } + routes = append(routes, route) + } + + for _, neigh := range link.Neighbors { + proto, tcpipAddr := ipToAddressAndProto(neigh.IP) + n.Stack.AddStaticNeighbor(nicID, proto, tcpipAddr, tcpip.LinkAddress(neigh.HardwareAddr)) + } + } + } else if len(args.XDPLinks) > 0 { + if nlinks := len(args.XDPLinks); nlinks > 1 { + return fmt.Errorf("XDP only supports one link device, but got %d", nlinks) + } + link := args.XDPLinks[0] nicID++ nicids[link.Name] = nicID - FDs := make([]int, 0, link.NumChannels) - for j := 0; j < link.NumChannels; j++ { - // Copy the underlying FD. - oldFD := args.FilePayload.Files[fdOffset].Fd() - newFD, err := unix.Dup(int(oldFD)) - if err != nil { - return fmt.Errorf("failed to dup FD %v: %v", oldFD, err) - } - FDs = append(FDs, newFD) - fdOffset++ + // Get the AF_XDP socket. + fdOffset := 0 + oldFD := args.FilePayload.Files[fdOffset].Fd() + fd, err := unix.Dup(int(oldFD)) + if err != nil { + return fmt.Errorf("failed to dup AF_XDP fd %v: %v", oldFD, err) } + fdOffset++ - // If AFXDP is enabled, we perform RX via AF_XDP and TX via - // AF_PACKET. - var AFXDPFD *int - if args.AFXDP { - // Get the AF_XDP socket. + // 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() - newFD, err := unix.Dup(int(oldFD)) - if err != nil { - return fmt.Errorf("failed to dup AF_XDP fd %v: %v", oldFD, err) + if _, err := unix.Dup(int(oldFD)); err != nil { + return fmt.Errorf("failed to dup %s with FD %d: %v", fdName, oldFD, err) } - 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) - log.Infof("gso max size is: %d", link.GSOMaxSize) - - linkEP, err := fdbased.New(&fdbased.Options{ - FDs: FDs, - AFXDPFD: AFXDPFD, - MTU: uint32(link.MTU), - EthernetHeader: mac != "", - Address: mac, - PacketDispatchMode: dispatchMode, - GSOMaxSize: link.GSOMaxSize, - GvisorGSOEnabled: link.GvisorGSOEnabled, - TXChecksumOffload: link.TXChecksumOffload, - RXChecksumOffload: link.RXChecksumOffload, - InterfaceIndex: link.InterfaceIndex, + linkEP, err := xdp.New(&xdp.Options{ + FD: fd, + MTU: uint32(link.MTU), + Address: mac, + TXChecksumOffload: link.TXChecksumOffload, + RXChecksumOffload: link.RXChecksumOffload, + InterfaceIndex: link.InterfaceIndex, }) if err != nil { return err diff --git a/runsc/sandbox/network.go b/runsc/sandbox/network.go index 1168a3a63..7f9aa06c3 100644 --- a/runsc/sandbox/network.go +++ b/runsc/sandbox/network.go @@ -217,66 +217,19 @@ func createInterfacesAndRoutesFromNS(conn *urpc.Client, nsPath string, conf *con args.Defaultv6Gateway.Name = iface.Name } - args.AFXDP = conf.AFXDP - - link := boot.FDBasedLink{ - Name: iface.Name, - InterfaceIndex: iface.Index, - MTU: iface.MTU, - Routes: routes, - TXChecksumOffload: conf.TXChecksumOffload, - RXChecksumOffload: conf.RXChecksumOffload, - NumChannels: conf.NumNetworkChannels, - QDisc: conf.QDisc, - Neighbors: neighbors, - } - // Get the link for the interface. ifaceLink, err := netlink.LinkByName(iface.Name) if err != nil { return fmt.Errorf("getting link for interface %q: %w", iface.Name, err) } - link.LinkAddress = ifaceLink.Attrs().HardwareAddr - - log.Debugf("Setting up network channels") - // Create the socket for the device. - for i := 0; i < link.NumChannels; i++ { - log.Debugf("Creating Channel %d", i) - socketEntry, err := createSocket(iface, ifaceLink, conf.HostGSO, conf.AFXDP) - if err != nil { - return fmt.Errorf("failed to createSocket for %s : %w", iface.Name, err) - } - if i == 0 { - link.GSOMaxSize = socketEntry.gsoMaxSize - } else { - if link.GSOMaxSize != socketEntry.gsoMaxSize { - return fmt.Errorf("inconsistent gsoMaxSize %d and %d when creating multiple channels for same interface: %s", - link.GSOMaxSize, socketEntry.gsoMaxSize, iface.Name) - } - } - args.FilePayload.Files = append(args.FilePayload.Files, socketEntry.deviceFile) - } - - // If enabled, create an RX socket for AF_XDP. - if conf.AFXDP { - xdpSockFDs, err := createSocketXDP(iface) - if err != nil { - return fmt.Errorf("failed to create XDP socket: %v", err) - } - args.FilePayload.Files = append(args.FilePayload.Files, xdpSockFDs...) - } - - if link.GSOMaxSize == 0 && conf.GvisorGSO { - // Host GSO is disabled. Let's enable gVisor GSO. - link.GSOMaxSize = stack.GvisorGSOMaxSize - link.GvisorGSOEnabled = true - } + linkAddress := ifaceLink.Attrs().HardwareAddr // Collect the addresses for the interface, enable forwarding, // and remove them from the host. + var addresses []boot.IPWithPrefix for _, addr := range ipAddrs { prefix, _ := addr.Mask.Size() - link.Addresses = append(link.Addresses, boot.IPWithPrefix{Address: addr.IP, PrefixLen: prefix}) + addresses = append(addresses, boot.IPWithPrefix{Address: addr.IP, PrefixLen: prefix}) // Steal IP address from NIC. if err := removeAddress(ifaceLink, addr.String()); err != nil { @@ -291,7 +244,66 @@ func createInterfacesAndRoutesFromNS(conn *urpc.Client, nsPath string, conf *con } } - args.FDBasedLinks = append(args.FDBasedLinks, link) + if conf.AFXDP { + xdpSockFDs, err := createSocketXDP(iface) + if err != nil { + return fmt.Errorf("failed to create XDP socket: %v", err) + } + args.FilePayload.Files = append(args.FilePayload.Files, xdpSockFDs...) + args.XDPLinks = append(args.XDPLinks, boot.XDPLink{ + Name: iface.Name, + InterfaceIndex: iface.Index, + MTU: iface.MTU, + Routes: routes, + TXChecksumOffload: conf.TXChecksumOffload, + RXChecksumOffload: conf.RXChecksumOffload, + NumChannels: conf.NumNetworkChannels, + QDisc: conf.QDisc, + Neighbors: neighbors, + LinkAddress: linkAddress, + Addresses: addresses, + }) + } else { + link := boot.FDBasedLink{ + Name: iface.Name, + MTU: iface.MTU, + Routes: routes, + TXChecksumOffload: conf.TXChecksumOffload, + RXChecksumOffload: conf.RXChecksumOffload, + NumChannels: conf.NumNetworkChannels, + QDisc: conf.QDisc, + Neighbors: neighbors, + LinkAddress: linkAddress, + Addresses: addresses, + } + + log.Debugf("Setting up network channels") + // Create the socket for the device. + for i := 0; i < link.NumChannels; i++ { + log.Debugf("Creating Channel %d", i) + socketEntry, err := createSocket(iface, ifaceLink, conf.HostGSO) + if err != nil { + return fmt.Errorf("failed to createSocket for %s : %w", iface.Name, err) + } + if i == 0 { + link.GSOMaxSize = socketEntry.gsoMaxSize + } else { + if link.GSOMaxSize != socketEntry.gsoMaxSize { + return fmt.Errorf("inconsistent gsoMaxSize %d and %d when creating multiple channels for same interface: %s", + link.GSOMaxSize, socketEntry.gsoMaxSize, iface.Name) + } + } + args.FilePayload.Files = append(args.FilePayload.Files, socketEntry.deviceFile) + } + + if link.GSOMaxSize == 0 && conf.GvisorGSO { + // Host GSO is disabled. Let's enable gVisor GSO. + link.GSOMaxSize = stack.GvisorGSOMaxSize + link.GvisorGSOEnabled = true + } + + args.FDBasedLinks = append(args.FDBasedLinks, link) + } } // Pass PCAP log file if present. @@ -342,7 +354,7 @@ type socketEntry struct { // createSocket creates an underlying AF_PACKET socket and configures it for // use by the sentry and returns an *os.File that wraps the underlying socket // fd. -func createSocket(iface net.Interface, ifaceLink netlink.Link, enableGSO bool, AFXDP bool) (*socketEntry, error) { +func createSocket(iface net.Interface, ifaceLink netlink.Link, enableGSO bool) (*socketEntry, error) { // Create the socket. const protocol = 0x0300 // htons(ETH_P_ALL) fd, err := unix.Socket(unix.AF_PACKET, unix.SOCK_RAW, protocol)