From 140384fa222684428d713d5f31cfb35957f20fc3 Mon Sep 17 00:00:00 2001 From: Kevin Krakauer Date: Wed, 12 Oct 2022 11:48:58 -0700 Subject: [PATCH] xdp: TX support PiperOrigin-RevId: 480680001 --- pkg/memutil/BUILD | 2 +- pkg/memutil/memutil.go | 16 --- pkg/memutil/memutil_unsafe.go | 44 ++++++++ pkg/memutil/mmap.go | 4 +- pkg/tcpip/link/xdp/BUILD | 2 +- pkg/tcpip/link/xdp/dispatcher.go | 125 --------------------- pkg/tcpip/link/xdp/endpoint.go | 174 ++++++++++++++++++++++++++--- pkg/xdp/BUILD | 5 + pkg/xdp/completionqueue.go | 119 ++++++++++++++++++++ pkg/xdp/fillqueue.go | 76 +++++++------ pkg/xdp/rxqueue.go | 26 ++++- pkg/xdp/txqueue.go | 116 ++++++++++++++++++++ pkg/xdp/umem.go | 54 ++++++++- pkg/xdp/xdp.go | 182 ++++++++++++++++++++++--------- pkg/xdp/xdp_unsafe.go | 45 ++++++++ runsc/boot/network.go | 1 - runsc/sandbox/network.go | 1 - tools/xdp/tcpdump.go | 24 ++-- 18 files changed, 745 insertions(+), 271 deletions(-) delete mode 100644 pkg/memutil/memutil.go create mode 100644 pkg/memutil/memutil_unsafe.go delete mode 100644 pkg/tcpip/link/xdp/dispatcher.go create mode 100644 pkg/xdp/completionqueue.go create mode 100644 pkg/xdp/txqueue.go diff --git a/pkg/memutil/BUILD b/pkg/memutil/BUILD index bea595286..50c6b8cad 100644 --- a/pkg/memutil/BUILD +++ b/pkg/memutil/BUILD @@ -6,7 +6,7 @@ go_library( name = "memutil", srcs = [ "memfd_linux_unsafe.go", - "memutil.go", + "memutil_unsafe.go", "mmap.go", ], visibility = ["//visibility:public"], diff --git a/pkg/memutil/memutil.go b/pkg/memutil/memutil.go deleted file mode 100644 index 3185882fd..000000000 --- a/pkg/memutil/memutil.go +++ /dev/null @@ -1,16 +0,0 @@ -// 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. - -// Package memutil provides utilities for working with shared memory files. -package memutil diff --git a/pkg/memutil/memutil_unsafe.go b/pkg/memutil/memutil_unsafe.go new file mode 100644 index 000000000..3c5ebd745 --- /dev/null +++ b/pkg/memutil/memutil_unsafe.go @@ -0,0 +1,44 @@ +// 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. + +// Package memutil provides utilities for working with shared memory files. +package memutil + +import ( + "reflect" + "unsafe" + + "golang.org/x/sys/unix" +) + +// MapSlice is like MapFile, but returns a slice instead of a uintptr. +func MapSlice(addr, size, prot, flags, fd, offset uintptr) ([]byte, error) { + addr, err := MapFile(addr, size, prot, flags, fd, offset) + if err != nil { + return nil, err + } + var slice []byte + hdr := (*reflect.SliceHeader)(unsafe.Pointer(&slice)) + hdr.Data = addr + hdr.Len = int(size) + hdr.Cap = int(size) + return slice, nil +} + +// UnmapSlice unmaps a mapping returned by MapSlice. +func UnmapSlice(slice []byte) error { + hdr := (*reflect.SliceHeader)(unsafe.Pointer(&slice)) + _, _, err := unix.RawSyscall6(unix.SYS_MUNMAP, uintptr(unsafe.Pointer(hdr.Data)), uintptr(hdr.Cap), 0, 0, 0, 0) + return err +} diff --git a/pkg/memutil/mmap.go b/pkg/memutil/mmap.go index 7a55d1b28..c909b8f3f 100644 --- a/pkg/memutil/mmap.go +++ b/pkg/memutil/mmap.go @@ -23,8 +23,8 @@ import ( // MapFile returns a memory mapping configured by the given options as per // mmap(2). -func MapFile(addr, len, prot, flags, fd, offset uintptr) (uintptr, error) { - m, _, e := unix.RawSyscall6(unix.SYS_MMAP, addr, len, prot, flags, fd, offset) +func MapFile(addr, size, prot, flags, fd, offset uintptr) (uintptr, error) { + m, _, e := unix.RawSyscall6(unix.SYS_MMAP, addr, size, prot, flags, fd, offset) if e != 0 { return 0, e } diff --git a/pkg/tcpip/link/xdp/BUILD b/pkg/tcpip/link/xdp/BUILD index 58d8b8531..07ec8f744 100644 --- a/pkg/tcpip/link/xdp/BUILD +++ b/pkg/tcpip/link/xdp/BUILD @@ -5,7 +5,6 @@ package(licenses = ["notice"]) go_library( name = "xdp", srcs = [ - "dispatcher.go", "endpoint.go", ], visibility = ["//visibility:public"], @@ -14,6 +13,7 @@ go_library( "//pkg/sync", "//pkg/tcpip", "//pkg/tcpip/header", + "//pkg/tcpip/link/qdisc/fifo", "//pkg/tcpip/link/rawfile", "//pkg/tcpip/link/stopfd", "//pkg/tcpip/stack", diff --git a/pkg/tcpip/link/xdp/dispatcher.go b/pkg/tcpip/link/xdp/dispatcher.go deleted file mode 100644 index 38adf652b..000000000 --- a/pkg/tcpip/link/xdp/dispatcher.go +++ /dev/null @@ -1,125 +0,0 @@ -// 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 xdp - -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" - "gvisor.dev/gvisor/pkg/xdp" -) - -// 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 - - // 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 (xd *dispatcher) init(fd int, ep *endpoint, index int) error { - stopFD, err := stopfd.New() - if err != nil { - return err - } - disp := dispatcher{ - 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 - disp.umem, disp.fillQueue, disp.rxQueue, err = xdp.ReadOnlyFromSocket(fd, uint32(index), 0 /* queueID */, opts) - if err != nil { - return fmt.Errorf("failed to create AF_XDP dispatcher: %v", err) - } - disp.fillQueue.FillAll() - return nil -} - -func (xd *dispatcher) 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) - - 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 _, ok := pkt.LinkHeader().Consume(header.EthernetMinimumSize); !ok { - panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", header.EthernetMinimumSize)) - } - xd.ep.networkDispatcher.DeliverNetworkPacket(netProto, pkt) - pkt.DecRef() - } - // Tell the kernel that we're done with these packets. - xd.rxQueue.Release(nReceived) - } - - return true, nil - } -} diff --git a/pkg/tcpip/link/xdp/endpoint.go b/pkg/tcpip/link/xdp/endpoint.go index fad8c6b34..e53bbbdab 100644 --- a/pkg/tcpip/link/xdp/endpoint.go +++ b/pkg/tcpip/link/xdp/endpoint.go @@ -22,20 +22,28 @@ import ( "fmt" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/bufferv2" "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/link/qdisc/fifo" + "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" ) +// TODO(b/240191988): Turn off GSO, GRO, and LRO. Limit veth MTU to 1500. + +// MTU is sized to ensure packets fit inside a 2048 byte XDP frame. +const MTU = 1500 + var _ stack.LinkEndpoint = (*endpoint)(nil) type endpoint struct { + // fd is the underlying AF_XDP socket. 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 @@ -46,11 +54,16 @@ type endpoint struct { // its end of the communication pipe. closed func(tcpip.Error) - inboundDispatcher dispatcher networkDispatcher stack.NetworkDispatcher // wg keeps track of running goroutines. wg sync.WaitGroup + + // control is used to control the AF_XDP socket. + control *xdp.ControlBlock + + // stopFD is used to stop the dispatch loop. + stopFD stopfd.StopFD } // Options specify the details about the fd-based endpoint to be created. @@ -58,9 +71,6 @@ 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) @@ -113,15 +123,42 @@ func New(opts *Options) (stack.LinkEndpoint, error) { 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) + stopFD, err := stopfd.New() + if err != nil { + return nil, err } + ep.stopFD = stopFD + + // Use a 2MB UMEM to match the PACKET_MMAP dispatcher. There will be + // 1024 UMEM frames, and each queue will have 512 descriptors. Having + // fewer descriptors than frames prevents RX and TX from starving each + // other. + // TODO(b/240191988): Consider different numbers of descriptors for + // different queues. + const ( + frameSize = 2048 + umemSize = 1 << 21 + nFrames = umemSize / frameSize + ) + xdpOpts := xdp.ReadOnlySocketOpts{ + NFrames: nFrames, + FrameSize: frameSize, + NDescriptors: nFrames / 2, + } + ep.control, err = xdp.ReadOnlyFromSocket(opts.FD, uint32(opts.InterfaceIndex), 0 /* queueID */, xdpOpts) + if err != nil { + return nil, fmt.Errorf("failed to create AF_XDP dispatcher: %v", err) + } + + ep.control.UMEM.Lock() + defer ep.control.UMEM.Unlock() + + ep.control.Fill.FillAll(&ep.control.UMEM) return ep, nil } @@ -134,7 +171,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) { func (ep *endpoint) Attach(networkDispatcher stack.NetworkDispatcher) { // nil means the NIC is being removed. if networkDispatcher == nil && ep.IsAttached() { - ep.inboundDispatcher.Stop() + ep.stopFD.Stop() ep.Wait() ep.networkDispatcher = nil return @@ -148,7 +185,7 @@ func (ep *endpoint) Attach(networkDispatcher stack.NetworkDispatcher) { go func() { // S/R-SAFE: See above. defer ep.wg.Done() for { - cont, err := ep.inboundDispatcher.dispatch() + cont, err := ep.dispatch() if err != nil || !cont { if ep.closed != nil { ep.closed(err) @@ -168,7 +205,7 @@ func (ep *endpoint) IsAttached() bool { // MTU implements stack.LinkEndpoint.MTU. It returns the value initialized // during construction. func (ep *endpoint) MTU() uint32 { - return ep.mtu + return MTU } // Capabilities implements stack.LinkEndpoint.Capabilities. @@ -218,5 +255,114 @@ func (ep *endpoint) ARPHardwareType() header.ARPHardwareType { // 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{} + // We expect to be called via fifo, which imposes a limit of + // fifo.BatchSize. + var preallocatedBatch [fifo.BatchSize]unix.XDPDesc + batch := preallocatedBatch[:0] + + ep.control.UMEM.Lock() + + ep.control.Completion.FreeAll(&ep.control.UMEM) + + // Reserve TX queue descriptors and umem buffers + nReserved, index := ep.control.TX.Reserve(&ep.control.UMEM, uint32(pkts.Len())) + if nReserved == 0 { + ep.control.UMEM.Unlock() + return 0, &tcpip.ErrNoBufferSpace{} + } + + // Allocate UMEM space. In order to release the UMEM lock as soon as + // possible, we allocate up-front and copy data in after releasing. + for _, pkt := range pkts.AsSlice() { + batch = append(batch, unix.XDPDesc{ + Addr: ep.control.UMEM.AllocFrame(), + Len: uint32(pkt.Size()), + }) + } + ep.control.UMEM.Unlock() + + for i, pkt := range pkts.AsSlice() { + // Copy packets into UMEM frame. + frame := ep.control.UMEM.Get(batch[i]) + offset := 0 + for _, buf := range pkt.AsSlices() { + offset += copy(frame[offset:], buf) + } + ep.control.TX.Set(index+uint32(i), batch[i]) + } + + // Notify the kernel that there're packets to write. + ep.control.TX.Notify() + + return pkts.Len(), nil +} + +func (ep *endpoint) dispatch() (bool, tcpip.Error) { + var views []*bufferv2.View + + for { + stopped, errno := rawfile.BlockingPollUntilStopped(ep.stopFD.EFD, ep.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 { + // We can receive multiple packets at once. + nReceived, rxIndex := ep.control.RX.Peek() + + if nReceived == 0 { + break + } + + // Reuse views to avoid allocating. + views = views[:0] + + // Populate views quickly so that we can release frames + // back to the kernel. + ep.control.UMEM.Lock() + for i := uint32(0); i < nReceived; i++ { + // Copy packet bytes into a view and free up the + // buffer. + descriptor := ep.control.RX.Get(rxIndex + i) + data := ep.control.UMEM.Get(descriptor) + view := bufferv2.NewViewWithData(data) + views = append(views, view) + ep.control.UMEM.FreeFrame(descriptor.Addr) + } + ep.control.Fill.FillAll(&ep.control.UMEM) + ep.control.UMEM.Unlock() + + // Process each packet. + for i := uint32(0); i < nReceived; i++ { + view := views[i] + data := view.AsSlice() + + 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), + }) + // AF_XDP packets always have a link header. + if _, ok := pkt.LinkHeader().Consume(header.EthernetMinimumSize); !ok { + panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", header.EthernetMinimumSize)) + } + ep.networkDispatcher.DeliverNetworkPacket(netProto, pkt) + pkt.DecRef() + } + // Tell the kernel that we're done with these + // descriptors in the RX queue. + ep.control.RX.Release(nReceived) + } + + return true, nil + } } diff --git a/pkg/xdp/BUILD b/pkg/xdp/BUILD index b01c55cbe..e6e8b1dbe 100644 --- a/pkg/xdp/BUILD +++ b/pkg/xdp/BUILD @@ -5,8 +5,10 @@ package(licenses = ["notice"]) go_library( name = "xdp", srcs = [ + "completionqueue.go", "fillqueue.go", "rxqueue.go", + "txqueue.go", "umem.go", "xdp.go", "xdp_unsafe.go", @@ -14,7 +16,10 @@ go_library( visibility = ["//:sandbox"], deps = [ "//pkg/atomicbitops", + "//pkg/cleanup", "//pkg/log", + "//pkg/memutil", + "//pkg/sync", "@org_golang_x_sys//unix:go_default_library", ], ) diff --git a/pkg/xdp/completionqueue.go b/pkg/xdp/completionqueue.go new file mode 100644 index 000000000..f8c69e8da --- /dev/null +++ b/pkg/xdp/completionqueue.go @@ -0,0 +1,119 @@ +// 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 amd64 || arm64 +// +build amd64 arm64 + +package xdp + +import ( + "gvisor.dev/gvisor/pkg/atomicbitops" +) + +// The CompletionQueue is how the kernel tells a process which buffers have +// been transmitted and can be reused. +// +// CompletionQueue is not thread-safe and requires external synchronization +type CompletionQueue struct { + // mem is the mmap'd area shared with the kernel. Many other fields of + // this struct point into mem. + mem []byte + + // ring is the actual ring buffer. It is a list of frame addresses + // ready to be reused. + // + // len(ring) must be a power of 2. + ring []uint64 + + // mask is used whenever indexing into ring. It is always len(ring)-1. + // It prevents index out of bounds errors while allowing the producer + // and consumer pointers to repeatedly "overflow" and loop back around + // the ring. + mask uint32 + + // producer points to the shared atomic value that indicates the last + // produced descriptor. Only the kernel updates this value. + producer *atomicbitops.Uint32 + + // consumer points to the shared atomic value that indicates the last + // consumed descriptor. Only we update this value. + consumer *atomicbitops.Uint32 + + // flags points to the shared atomic value that holds flags for the + // queue. + flags *atomicbitops.Uint32 + + // Cached values are used to avoid relatively expensive atomic + // operations. They are used, incremented, and decremented multiple + // times with non-atomic operations, and then "batch-updated" by + // reading or writing atomically to synchronize with the kernel. + + // cachedProducer is updated when we atomically read *producer. + cachedProducer uint32 + // cachedConsumer is used to atomically write *consumer. + cachedConsumer uint32 +} + +// Peek returns the number of buffers available to reuse as well as the index +// at which they start. Peek will only return a buffer once, so callers must +// process any received buffers. +func (cq *CompletionQueue) Peek() (nAvailable, index uint32) { + // Get the number of available buffers and update cachedConsumer to + // reflect that we're going to consume them. + entries := cq.free() + index = cq.cachedConsumer + cq.cachedConsumer += entries + return entries, index +} + +func (cq *CompletionQueue) free() uint32 { + // Return any buffers we know about without incurring an atomic + // operation if possible. + entries := cq.cachedProducer - cq.cachedConsumer + // If we're not aware of any completed packets, refresh the producer + // pointer to see whether the kernel enqueued anything. + if entries == 0 { + cq.cachedProducer = cq.producer.Load() + entries = cq.cachedProducer - cq.cachedConsumer + } + return entries +} + +// Release notifies the kernel that we have consumed nDone packets. +func (cq *CompletionQueue) Release(nDone uint32) { + // We don't have to use an atomic add because only we update this; the + // kernel just reads it. + cq.consumer.Store(cq.consumer.RacyLoad() + nDone) +} + +// Get gets the descriptor at index. +func (cq *CompletionQueue) Get(index uint32) uint64 { + // Use mask to avoid overflowing and loop back around the ring. + return cq.ring[index&cq.mask] +} + +// FreeAll dequeues as many buffers as possible from the queue and returns them +// to the UMEM. +// +// +checklocks:umem.mu +func (cq *CompletionQueue) FreeAll(umem *UMEM) { + available, index := cq.Peek() + if available < 1 { + return + } + for i := uint32(0); i < available; i++ { + umem.FreeFrame(cq.Get(index + i)) + } + cq.Release(available) +} diff --git a/pkg/xdp/fillqueue.go b/pkg/xdp/fillqueue.go index 0e196edab..d08b62016 100644 --- a/pkg/xdp/fillqueue.go +++ b/pkg/xdp/fillqueue.go @@ -18,29 +18,28 @@ package xdp import ( - "fmt" - "gvisor.dev/gvisor/pkg/atomicbitops" ) // The FillQueue is how a process tells the kernel which buffers are available // to be filled by incoming packets. +// +// FillQueue is not thread-safe and requires external synchronization type FillQueue struct { // mem is the mmap'd area shared with the kernel. Many other fields of // this struct point into mem. mem []byte - // umem is the UMEM (i.e. shared buffer space) to which the queue's - // descriptors point. - umem *UMEM - // ring is the actual ring buffer. It is a list of frame addresses // ready for incoming packets. + // + // len(ring) must be a power of 2. ring []uint64 - // mask is used whenever indexing into ring. It prevents index out of - // bounds errors while allowing the producer and consumer pointers to - // repeatedly "overflow" and loop back around the ring. + // mask is used whenever indexing into ring. It is always len(ring)-1. + // It prevents index out of bounds errors while allowing the producer + // and consumer pointers to repeatedly "overflow" and loop back around + // the ring. mask uint32 // producer points to the shared atomic value that indicates the last @@ -56,30 +55,26 @@ type FillQueue struct { flags *atomicbitops.Uint32 // Cached values are used to avoid relatively expensive atomic - // operations. - cachedProducer uint32 - // cachedConsumer is actually len(ring) larger than the real consumer - // value. See Free() for details. - cachedConsumer uint32 -} + // operations. They are used, incremented, and decremented multiple + // times with non-atomic operations, and then "batch-updated" by + // reading or writing atomically to synchronize with the kernel. -// Reserve reserves descriptors in the fill queue. If toReserve descriptors -// cannot be reserved, none are reserved. -func (fq *FillQueue) Reserve(toReserve uint32) (nReserved, index uint32) { - if fq.free(toReserve) < toReserve { - // Unable to free the desired number of descriptors. - return 0, 0 - } - idx := fq.cachedProducer - fq.cachedProducer += toReserve - return toReserve, idx + // cachedProducer is used to atomically write *producer. + cachedProducer uint32 + // cachedConsumer is updated when we atomically read *consumer. + // cachedConsumer is actually len(ring) larger than the real consumer + // value. See free() for details. + cachedConsumer uint32 } // free returns the number of free descriptors in the fill queue. func (fq *FillQueue) free(toReserve uint32) uint32 { + // Try to find free descriptors without incurring an atomic operation. + // // cachedConsumer is always len(fq.ring) larger than the real consumer // value. This lets us, in the common case, compute the number of free - // descriptors simply via fq.cachedConsumer - fq.cachedProducer. + // descriptors simply via fq.cachedConsumer - fq.cachedProducer without + // also adding len(fq.ring). if available := fq.cachedConsumer - fq.cachedProducer; available >= toReserve { return available } @@ -91,33 +86,36 @@ func (fq *FillQueue) free(toReserve uint32) uint32 { return fq.cachedConsumer - fq.cachedProducer } -// Notify updates the prodcer such that it is visible to the kernel. +// Notify updates the producer such that it is visible to the kernel. func (fq *FillQueue) Notify() { fq.producer.Store(fq.cachedProducer) } // Set sets the fill queue's descriptor at index to addr. func (fq *FillQueue) Set(index uint32, addr uint64) { + // Use mask to avoid overflowing and loop back around the ring. fq.ring[index&fq.mask] = addr } -// FillAll fills the queue with as many buffers as possible from the UMEM, then +// FillAll posts as many empty buffers as possible for the kernel to fill, then // notifies the kernel. -func (fq *FillQueue) FillAll() { - available := fq.free(fq.umem.nFreeFrames) - if available < 1 { +// +// +checklocks:umem.mu +func (fq *FillQueue) FillAll(umem *UMEM) { + // Figure out how many buffers and queue slots are available. + available := fq.free(umem.nFreeFrames) + if available == 0 { return } - if available > fq.umem.nFreeFrames { - available = fq.umem.nFreeFrames + if available > umem.nFreeFrames { + available = umem.nFreeFrames } - _, index := fq.Reserve(available) + + // Fill the queue as much as possible and notify ther kernel. + index := fq.cachedProducer + fq.cachedProducer += available for i := uint32(0); i < available; i++ { - addr, err := fq.umem.AllocFrame() - if err != nil { - panic(fmt.Sprintf("failed to alloc frame #%d: %v", index+i, err)) - } - fq.Set(index+i, addr) + fq.Set(index+i, umem.AllocFrame()) } fq.Notify() } diff --git a/pkg/xdp/rxqueue.go b/pkg/xdp/rxqueue.go index e36731d73..17acd6c2d 100644 --- a/pkg/xdp/rxqueue.go +++ b/pkg/xdp/rxqueue.go @@ -24,19 +24,23 @@ import ( // The RXQueue is how the kernel tells a process which buffers are full with // incoming packets. +// +// RXQueue is not thread-safe and requires external synchronization type RXQueue struct { // mem is the mmap'd area shared with the kernel. Many other fields of // this struct point into mem. mem []byte - // ring is the list of XDP descriptors shared with the kernel. // ring is the actual ring buffer. It is a list of XDP descriptors // pointing to incoming packets. + // + // len(ring) must be a power of 2. ring []unix.XDPDesc - // mask is used whenever indexing into ring. It prevents index out of - // bounds errors while allowing the producer and consumer pointers to - // repeatedly "overflow" and loop back around the ring. + // mask is used whenever indexing into ring. It is always len(ring)-1. + // It prevents index out of bounds errors while allowing the producer + // and consumer pointers to repeatedly "overflow" and loop back around + // the ring. mask uint32 // producer points to the shared atomic value that indicates the last @@ -52,8 +56,13 @@ type RXQueue struct { flags *atomicbitops.Uint32 // Cached values are used to avoid relatively expensive atomic - // operations. + // operations. They are used, incremented, and decremented multiple + // times with non-atomic operations, and then "batch-updated" by + // reading or writing atomically to synchronize with the kernel. + + // cachedProducer is updated when we atomically read *producer. cachedProducer uint32 + // cachedConsumer is used to atomically write *consumer. cachedConsumer uint32 } @@ -61,6 +70,8 @@ type RXQueue struct { // which they start. Peek will only return a packet once, so callers must // process any received packets. func (rq *RXQueue) Peek() (nReceived, index uint32) { + // Get the number of available buffers and update cachedConsumer to + // reflect that we're going to consume them. entries := rq.free() index = rq.cachedConsumer rq.cachedConsumer += entries @@ -68,6 +79,8 @@ func (rq *RXQueue) Peek() (nReceived, index uint32) { } func (rq *RXQueue) free() uint32 { + // Return any buffers we know about without incurring an atomic + // operation if possible. entries := rq.cachedProducer - rq.cachedConsumer // If we're not aware of any RX'd packets, refresh the producer pointer // to see whether the kernel enqueued anything. @@ -80,12 +93,13 @@ func (rq *RXQueue) free() uint32 { // Release notifies the kernel that we have consumed nDone packets. func (rq *RXQueue) Release(nDone uint32) { - // We don't have to use an atomic add becuase only we update this; the + // We don't have to use an atomic add because only we update this; the // kernel just reads it. rq.consumer.Store(rq.consumer.RacyLoad() + nDone) } // Get gets the descriptor at index. func (rq *RXQueue) Get(index uint32) unix.XDPDesc { + // Use mask to avoid overflowing and loop back around the ring. return rq.ring[index&rq.mask] } diff --git a/pkg/xdp/txqueue.go b/pkg/xdp/txqueue.go new file mode 100644 index 000000000..c3f73cb2b --- /dev/null +++ b/pkg/xdp/txqueue.go @@ -0,0 +1,116 @@ +// 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 amd64 || arm64 +// +build amd64 arm64 + +package xdp + +import ( + "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/atomicbitops" +) + +// The TXQueue is how a process tells the kernel which buffers are available to +// be sent via the NIC. +// +// TXQueue is not thread-safe and requires external synchronization +type TXQueue struct { + // sockfd is the underlying AF_XDP socket. + sockfd uint32 + + // mem is the mmap'd area shared with the kernel. Many other fields of + // this struct point into mem. + mem []byte + + // ring is the actual ring buffer. It is a list of XDP descriptors + // pointing to ready-to-transmit packets. + // + // len(ring) must be a power of 2. + ring []unix.XDPDesc + + // mask is used whenever indexing into ring. It is always len(ring)-1. + // It prevents index out of bounds errors while allowing the producer + // and consumer pointers to repeatedly "overflow" and loop back around + // the ring. + mask uint32 + + // producer points to the shared atomic value that indicates the last + // produced descriptor. Only we update this value. + producer *atomicbitops.Uint32 + + // consumer points to the shared atomic value that indicates the last + // consumed descriptor. Only the kernel updates this value. + consumer *atomicbitops.Uint32 + + // flags points to the shared atomic value that holds flags for the + // queue. + flags *atomicbitops.Uint32 + + // Cached values are used to avoid relatively expensive atomic + // operations. They are used, incremented, and decremented multiple + // times with non-atomic operations, and then "batch-updated" by + // reading or writing atomically to synchronize with the kernel. + + // cachedProducer is used to atomically write *producer. + cachedProducer uint32 + // cachedConsumer is updated when we atomically read *consumer. + // cachedConsumer is actually len(ring) larger than the real consumer + // value. See free() for details. + cachedConsumer uint32 +} + +// Reserve reserves descriptors in the queue. If toReserve descriptors cannot +// be reserved, none are reserved. +// +// +checklocks:umem.mu +func (tq *TXQueue) Reserve(umem *UMEM, toReserve uint32) (nReserved, index uint32) { + if umem.nFreeFrames < toReserve || tq.free(toReserve) < toReserve { + return 0, 0 + } + idx := tq.cachedProducer + tq.cachedProducer += toReserve + return toReserve, idx +} + +// free returns the number of free descriptors in the TX queue. +func (tq *TXQueue) free(toReserve uint32) uint32 { + // Try to find free descriptors without incurring an atomic operation. + // + // cachedConsumer is always len(tq.ring) larger than the real consumer + // value. This lets us, in the common case, compute the number of free + // descriptors simply via tq.cachedConsumer - tq.cachedProducer without + // also addign len(tq.ring). + if available := tq.cachedConsumer - tq.cachedProducer; available >= toReserve { + return available + } + + // If we didn't already have enough descriptors available, check + // whether the kernel has returned some to us. + tq.cachedConsumer = tq.consumer.Load() + tq.cachedConsumer += uint32(len(tq.ring)) + return tq.cachedConsumer - tq.cachedProducer +} + +// Notify updates the producer such that it is visible to the kernel. +func (tq *TXQueue) Notify() { + tq.producer.Store(tq.cachedProducer) + tq.kick() +} + +// Set sets the TX queue's descriptor at index to addr. +func (tq *TXQueue) Set(index uint32, desc unix.XDPDesc) { + // Use mask to avoid overflowing and loop back around the ring. + tq.ring[index&tq.mask] = desc +} diff --git a/pkg/xdp/umem.go b/pkg/xdp/umem.go index 98fac4336..c10e65e56 100644 --- a/pkg/xdp/umem.go +++ b/pkg/xdp/umem.go @@ -18,9 +18,19 @@ package xdp import ( + "fmt" + "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/sync" ) +// TODO(b/240191988): There's some kind of memory corruption bug that occurs +// occasionally. This occured even before TX was supported. + +// TODO(b/240191988): We can hold locks for less time if we accept a more +// obtuse API. For example, CompletionQueue.FreeAll doesn't need to hold a +// mutex for its entire duration. + // UMEM is the shared memory area that the kernel and userspace put packets in. type UMEM struct { // mem is the mmap'd area shared with the kernel. @@ -29,11 +39,20 @@ type UMEM struct { // sockfd is the underlying AF_XDP socket. sockfd uint32 + // frameMask masks the lower bits of an adderess to get the frame's + // address. + frameMask uint64 + + // mu protects frameAddresses and nFreeFrames. + mu sync.Mutex + // frameAddresses is a stack of available frame addresses. + // +checklocks:mu frameAddresses []uint64 // nFreeFrames is the number of frames available and is used to index // into frameAddresses. + // +checklocks:mu nFreeFrames uint32 } @@ -42,20 +61,47 @@ func (um *UMEM) SockFD() uint32 { return um.sockfd } +// Lock locks the UMEM. +// +// +checklocksacquire:um.mu +func (um *UMEM) Lock() { + um.mu.Lock() +} + +// Unlock unlocks the UMEM. +// +// +checklocksrelease:um.mu +func (um *UMEM) Unlock() { + um.mu.Unlock() +} + // FreeFrame returns the frame containing addr to the set of free frames. +// +// The UMEM must be locked during the call to FreeFrame. +// +// +checklocks:um.mu func (um *UMEM) FreeFrame(addr uint64) { um.frameAddresses[um.nFreeFrames] = addr um.nFreeFrames++ } // AllocFrame returns the address of a frame that can be enqueued to the fill -// or TX queue. -func (um *UMEM) AllocFrame() (uint64, error) { +// or TX queue. It will panic if there are no frames left, so callers must call +// it no more than the number of buffers reserved via TXQueue.Reserve(). +// +// The UMEM must be locked during the call to AllocFrame. +// +// +checklocks:um.mu +func (um *UMEM) AllocFrame() uint64 { um.nFreeFrames-- - return um.frameAddresses[um.nFreeFrames], nil + return um.frameAddresses[um.nFreeFrames] & um.frameMask } // Get gets the bytes of the packet pointed to by desc. func (um *UMEM) Get(desc unix.XDPDesc) []byte { - return um.mem[desc.Addr : desc.Addr+uint64(desc.Len)] + end := desc.Addr + uint64(desc.Len) + if desc.Addr&um.frameMask != (end-1)&um.frameMask { + panic(fmt.Sprintf("UMEM (%+v) access crosses frame boundaries: %+v", um, desc)) + } + return um.mem[desc.Addr:end] } diff --git a/pkg/xdp/xdp.go b/pkg/xdp/xdp.go index f3c7fc81e..0cd6f13df 100644 --- a/pkg/xdp/xdp.go +++ b/pkg/xdp/xdp.go @@ -42,11 +42,27 @@ package xdp import ( "fmt" + "math/bits" "golang.org/x/sys/unix" + "gvisor.dev/gvisor/pkg/cleanup" "gvisor.dev/gvisor/pkg/log" + "gvisor.dev/gvisor/pkg/memutil" ) +// A ControlBlock contains all the control structures necessary to use an +// AF_XDP socket. +// +// The ControlBlock and the structures it contains are meant to be used with a +// single RX goroutine and a single TX goroutine. +type ControlBlock struct { + UMEM UMEM + Fill FillQueue + RX RXQueue + TX TXQueue + Completion CompletionQueue +} + // ReadOnlySocketOpts configure a read-only AF_XDP socket. type ReadOnlySocketOpts struct { NFrames uint32 @@ -69,51 +85,69 @@ func DefaultReadOnlyOpts() ReadOnlySocketOpts { // ReadOnlySocket returns an initialized read-only AF_XDP socket bound to a // particular interface and queue. -func ReadOnlySocket(ifaceIdx, queueID uint32, opts ReadOnlySocketOpts) (*UMEM, *FillQueue, *RXQueue, error) { +func ReadOnlySocket(ifaceIdx, queueID uint32, opts ReadOnlySocketOpts) (*ControlBlock, error) { sockfd, err := unix.Socket(unix.AF_XDP, unix.SOCK_RAW, 0) if err != nil { - return nil, nil, nil, fmt.Errorf("failed to create AF_XDP socket: %v", err) + return nil, fmt.Errorf("failed to create AF_XDP socket: %v", err) } return ReadOnlyFromSocket(sockfd, ifaceIdx, queueID, opts) } // ReadOnlyFromSocket takes an AF_XDP socket, initializes it, and binds it to a // particular interface and queue. -func ReadOnlyFromSocket(sockfd int, ifaceIdx, queueID uint32, opts ReadOnlySocketOpts) (*UMEM, *FillQueue, *RXQueue, error) { +func ReadOnlyFromSocket(sockfd int, ifaceIdx, queueID uint32, opts ReadOnlySocketOpts) (*ControlBlock, error) { + if opts.FrameSize != 2048 && opts.FrameSize != 4096 { + return nil, fmt.Errorf("invalid frame size %d: must be either 2048 or 4096", opts.FrameSize) + } + if bits.OnesCount32(opts.NDescriptors) != 1 { + return nil, fmt.Errorf("invalid number of descriptors %d: must be a power of 2", opts.NDescriptors) + } + + var cb ControlBlock + // Create the UMEM area. Use mmap instead of make([[]byte) to ensure // that the UMEM is page-aligned. Aligning the UMEM keeps individual // packets from spilling over between pages. - umemMemory, err := unix.Mmap(-1, + var zerofd uintptr + umemMemory, err := memutil.MapSlice( 0, - int(opts.NFrames*opts.FrameSize), + uintptr(opts.NFrames*opts.FrameSize), unix.PROT_READ|unix.PROT_WRITE, - unix.MAP_PRIVATE|unix.MAP_ANONYMOUS) + unix.MAP_PRIVATE|unix.MAP_ANONYMOUS, + zerofd-1, + 0, + ) if err != nil { - return nil, nil, nil, fmt.Errorf("failed to mmap umem: %v", err) + return nil, fmt.Errorf("failed to mmap umem: %v", err) } + cleanup := cleanup.Make(func() { + memutil.UnmapSlice(umemMemory) + }) + if sliceBackingPointer(umemMemory)%uintptr(unix.Getpagesize()) != 0 { - return nil, nil, nil, fmt.Errorf("UMEM is not page aligned (address 0x%x)", sliceBackingPointer(umemMemory)) + return nil, fmt.Errorf("UMEM is not page aligned (address 0x%x)", sliceBackingPointer(umemMemory)) } - umem := UMEM{ + cb.UMEM = UMEM{ mem: umemMemory, sockfd: uint32(sockfd), frameAddresses: make([]uint64, opts.NFrames), nFreeFrames: opts.NFrames, + frameMask: ^(uint64(opts.FrameSize) - 1), } // Fill in each frame address. - for i := range umem.frameAddresses { - umem.frameAddresses[i] = uint64(i) * uint64(opts.FrameSize) + for i := range cb.UMEM.frameAddresses { + cb.UMEM.frameAddresses[i] = uint64(i) * uint64(opts.FrameSize) } // Check whether we're likely to fail due to RLIMIT_MEMLOCK. var rlimit unix.Rlimit if err := unix.Getrlimit(unix.RLIMIT_MEMLOCK, &rlimit); err != nil { - return nil, nil, nil, fmt.Errorf("failed to get rlimit for memlock: %v", err) + return nil, fmt.Errorf("failed to get rlimit for memlock: %v", err) } - if rlimit.Cur < uint64(len(umem.mem)) { - log.Infof("UMEM size (%d) may exceed RLIMIT_MEMLOCK (%+v) and cause registration to fail", len(umem.mem), rlimit) + if rlimit.Cur < uint64(len(cb.UMEM.mem)) { + log.Infof("UMEM size (%d) may exceed RLIMIT_MEMLOCK (%+v) and cause registration to fail", len(cb.UMEM.mem), rlimit) } reg := unix.XDPUmemReg{ @@ -126,19 +160,24 @@ func ReadOnlyFromSocket(sockfd int, ifaceIdx, queueID uint32, opts ReadOnlySocke Flags: 0, } if err := registerUMEM(sockfd, reg); err != nil { - return nil, nil, nil, fmt.Errorf("failed to register UMEM: %v", err) + return nil, fmt.Errorf("failed to register UMEM: %v", err) } // Set the number of descriptors in the fill queue. if err := unix.SetsockoptInt(sockfd, unix.SOL_XDP, unix.XDP_UMEM_FILL_RING, int(opts.NDescriptors)); err != nil { - return nil, nil, nil, fmt.Errorf("failed to register fill ring: %v", err) + return nil, fmt.Errorf("failed to register fill ring: %v", err) } - - // Set the number of descriptors in the completion queue. Note: we - // don't actually use this (the completion queue is TX-specific), but - // bind() will fail if this is left unset. + // Set the number of descriptors in the completion queue. if err := unix.SetsockoptInt(sockfd, unix.SOL_XDP, unix.XDP_UMEM_COMPLETION_RING, int(opts.NDescriptors)); err != nil { - return nil, nil, nil, fmt.Errorf("failed to register fill ring: %v", err) + return nil, fmt.Errorf("failed to register completion ring: %v", err) + } + // Set the number of descriptors in the RX queue. + if err := unix.SetsockoptInt(sockfd, unix.SOL_XDP, unix.XDP_RX_RING, int(opts.NDescriptors)); err != nil { + return nil, fmt.Errorf("failed to register RX queue: %v", err) + } + // Set the number of descriptors in the TX queue. + if err := unix.SetsockoptInt(sockfd, unix.SOL_XDP, unix.XDP_TX_RING, int(opts.NDescriptors)); err != nil { + return nil, fmt.Errorf("failed to register TX queue: %v", err) } // Get offset information for the queues. Offsets indicate where, once @@ -147,59 +186,99 @@ func ReadOnlyFromSocket(sockfd int, ifaceIdx, queueID uint32, opts ReadOnlySocke // beginning of the ring of descriptors. off, err := getOffsets(sockfd) if err != nil { - return nil, nil, nil, fmt.Errorf("failed to get offsets: %v", err) + return nil, fmt.Errorf("failed to get offsets: %v", err) } // Allocate space for the fill queue. - fillQueueMem, err := unix.Mmap(sockfd, - unix.XDP_UMEM_PGOFF_FILL_RING, - int(off.Fr.Desc+uint64(opts.NDescriptors)*sizeOfFillQueueDesc()), + fillQueueMem, err := memutil.MapSlice( + 0, + uintptr(off.Fr.Desc+uint64(opts.NDescriptors)*sizeOfFillQueueDesc()), unix.PROT_READ|unix.PROT_WRITE, - unix.MAP_SHARED|unix.MAP_POPULATE) + unix.MAP_SHARED|unix.MAP_POPULATE, + uintptr(sockfd), + unix.XDP_UMEM_PGOFF_FILL_RING, + ) if err != nil { - return nil, nil, nil, fmt.Errorf("failed to mmap fill queue: %v", err) + return nil, fmt.Errorf("failed to mmap fill queue: %v", err) } - + cleanup.Add(func() { + memutil.UnmapSlice(fillQueueMem) + }) // Setup the fillQueue with offsets into allocated memory. - fillQueue := FillQueue{ + cb.Fill = FillQueue{ mem: fillQueueMem, mask: opts.NDescriptors - 1, - umem: &umem, cachedConsumer: opts.NDescriptors, } - fillQueue.init(off, opts) + cb.Fill.init(off, opts) - // Allocate space for the (unused) completion queue. - _, err = unix.Mmap(sockfd, - unix.XDP_UMEM_PGOFF_COMPLETION_RING, - int(off.Cr.Desc+uint64(opts.NDescriptors)*sizeOfFillQueueDesc()), + // Allocate space for the completion queue. + completionQueueMem, err := memutil.MapSlice( + 0, + uintptr(off.Cr.Desc+uint64(opts.NDescriptors)*sizeOfCompletionQueueDesc()), unix.PROT_READ|unix.PROT_WRITE, - unix.MAP_SHARED|unix.MAP_POPULATE) + unix.MAP_SHARED|unix.MAP_POPULATE, + uintptr(sockfd), + unix.XDP_UMEM_PGOFF_COMPLETION_RING, + ) if err != nil { - return nil, nil, nil, fmt.Errorf("failed to mmap completion queue: %v", err) + return nil, fmt.Errorf("failed to mmap completion queue: %v", err) } - - // Set the number of descriptors in the RX queue. - if err := unix.SetsockoptInt(sockfd, unix.SOL_XDP, unix.XDP_RX_RING, int(opts.NDescriptors)); err != nil { - return nil, nil, nil, fmt.Errorf("failed to register RX queue: %v", err) + cleanup.Add(func() { + memutil.UnmapSlice(completionQueueMem) + }) + // Setup the completionQueue with offsets into allocated memory. + cb.Completion = CompletionQueue{ + mem: completionQueueMem, + mask: opts.NDescriptors - 1, } + cb.Completion.init(off, opts) // Allocate space for the RX queue. - rxQueueMem, err := unix.Mmap(sockfd, - unix.XDP_PGOFF_RX_RING, - int(off.Rx.Desc+uint64(opts.NDescriptors)*sizeOfRXQueueDesc()), + rxQueueMem, err := memutil.MapSlice( + 0, + uintptr(off.Rx.Desc+uint64(opts.NDescriptors)*sizeOfRXQueueDesc()), unix.PROT_READ|unix.PROT_WRITE, - unix.MAP_SHARED|unix.MAP_POPULATE) + unix.MAP_SHARED|unix.MAP_POPULATE, + uintptr(sockfd), + unix.XDP_PGOFF_RX_RING, + ) if err != nil { - return nil, nil, nil, fmt.Errorf("failed to mmap fill queue: %v", err) + return nil, fmt.Errorf("failed to mmap RX queue: %v", err) } - + cleanup.Add(func() { + memutil.UnmapSlice(rxQueueMem) + }) // Setup the rxQueue with offsets into allocated memory. - rxQueue := RXQueue{ + cb.RX = RXQueue{ mem: rxQueueMem, mask: opts.NDescriptors - 1, } - rxQueue.init(off, opts) + cb.RX.init(off, opts) + + // Allocate space for the TX queue. + txQueueMem, err := memutil.MapSlice( + 0, + uintptr(off.Tx.Desc+uint64(opts.NDescriptors)*sizeOfTXQueueDesc()), + unix.PROT_READ|unix.PROT_WRITE, + unix.MAP_SHARED|unix.MAP_POPULATE, + uintptr(sockfd), + unix.XDP_PGOFF_TX_RING, + ) + if err != nil { + return nil, fmt.Errorf("failed to mmap tx queue: %v", err) + } + cleanup.Add(func() { + memutil.UnmapSlice(txQueueMem) + }) + // Setup the txQueue with offsets into allocated memory. + cb.TX = TXQueue{ + sockfd: uint32(sockfd), + mem: txQueueMem, + mask: opts.NDescriptors - 1, + cachedConsumer: opts.NDescriptors, + } + cb.TX.init(off, opts) addr := unix.SockaddrXDP{ // XDP_USE_NEED_WAKEUP lets the driver sleep if there is no @@ -220,8 +299,9 @@ func ReadOnlyFromSocket(sockfd int, ifaceIdx, queueID uint32, opts ReadOnlySocke SharedUmemFD: 0, } if err := unix.Bind(sockfd, &addr); err != nil { - return nil, nil, nil, fmt.Errorf("failed to bind with addr %+v: %v", addr, err) + return nil, fmt.Errorf("failed to bind with addr %+v: %v", addr, err) } - return &umem, &fillQueue, &rxQueue, nil + cleanup.Release() + return &cb, nil } diff --git a/pkg/xdp/xdp_unsafe.go b/pkg/xdp/xdp_unsafe.go index 5c4ede51e..ab99a7f36 100644 --- a/pkg/xdp/xdp_unsafe.go +++ b/pkg/xdp/xdp_unsafe.go @@ -53,6 +53,14 @@ func sizeOfRXQueueDesc() uint64 { return uint64(unsafe.Sizeof(unix.XDPDesc{})) } +func sizeOfCompletionQueueDesc() uint64 { + return uint64(unsafe.Sizeof(uint64(0))) +} + +func sizeOfTXQueueDesc() uint64 { + return uint64(unsafe.Sizeof(unix.XDPDesc{})) +} + func (fq *FillQueue) init(off unix.XDPMmapOffsets, opts ReadOnlySocketOpts) { fillQueueRingHdr := (*reflect.SliceHeader)(unsafe.Pointer(&fq.ring)) fillQueueRingHdr.Data = uintptr(unsafe.Pointer(&fq.mem[off.Fr.Desc])) @@ -76,3 +84,40 @@ func (rq *RXQueue) init(off unix.XDPMmapOffsets, opts ReadOnlySocketOpts) { rq.cachedProducer = rq.producer.Load() rq.cachedConsumer = rq.consumer.Load() } + +func (cq *CompletionQueue) init(off unix.XDPMmapOffsets, opts ReadOnlySocketOpts) { + completionQueueRingHdr := (*reflect.SliceHeader)(unsafe.Pointer(&cq.ring)) + completionQueueRingHdr.Data = uintptr(unsafe.Pointer(&cq.mem[off.Cr.Desc])) + completionQueueRingHdr.Len = int(opts.NDescriptors) + completionQueueRingHdr.Cap = completionQueueRingHdr.Len + cq.producer = (*atomicbitops.Uint32)(unsafe.Pointer(&cq.mem[off.Cr.Producer])) + cq.consumer = (*atomicbitops.Uint32)(unsafe.Pointer(&cq.mem[off.Cr.Consumer])) + cq.flags = (*atomicbitops.Uint32)(unsafe.Pointer(&cq.mem[off.Cr.Flags])) + // These probably don't have to be atomic, but we're only loading once + // so better safe than sorry. + cq.cachedProducer = cq.producer.Load() + cq.cachedConsumer = cq.consumer.Load() +} + +func (tq *TXQueue) init(off unix.XDPMmapOffsets, opts ReadOnlySocketOpts) { + txQueueRingHdr := (*reflect.SliceHeader)(unsafe.Pointer(&tq.ring)) + txQueueRingHdr.Data = uintptr(unsafe.Pointer(&tq.mem[off.Tx.Desc])) + txQueueRingHdr.Len = int(opts.NDescriptors) + txQueueRingHdr.Cap = txQueueRingHdr.Len + tq.producer = (*atomicbitops.Uint32)(unsafe.Pointer(&tq.mem[off.Tx.Producer])) + tq.consumer = (*atomicbitops.Uint32)(unsafe.Pointer(&tq.mem[off.Tx.Consumer])) + tq.flags = (*atomicbitops.Uint32)(unsafe.Pointer(&tq.mem[off.Tx.Flags])) +} + +// kick notifies the kernel that there are packets to transmit. +func (tq *TXQueue) kick() error { + if tq.flags.RacyLoad()&unix.XDP_RING_NEED_WAKEUP == 0 { + return nil + } + + var msg unix.Msghdr + if _, _, errno := unix.Syscall6(unix.SYS_SENDMSG, uintptr(tq.sockfd), uintptr(unsafe.Pointer(&msg)), unix.MSG_DONTWAIT|unix.MSG_NOSIGNAL, 0, 0, 0); errno != 0 { + return fmt.Errorf("failed to kick TX queue via sendmsg: errno %d", errno) + } + return nil +} diff --git a/runsc/boot/network.go b/runsc/boot/network.go index 5d8c2dc7d..d0787ed7a 100644 --- a/runsc/boot/network.go +++ b/runsc/boot/network.go @@ -344,7 +344,6 @@ func (n *Network) CreateLinksAndRoutes(args *CreateLinksAndRoutesArgs, _ *struct mac := tcpip.LinkAddress(link.LinkAddress) linkEP, err := xdp.New(&xdp.Options{ FD: fd, - MTU: uint32(link.MTU), Address: mac, TXChecksumOffload: link.TXChecksumOffload, RXChecksumOffload: link.RXChecksumOffload, diff --git a/runsc/sandbox/network.go b/runsc/sandbox/network.go index 7f9aa06c3..85bb26205 100644 --- a/runsc/sandbox/network.go +++ b/runsc/sandbox/network.go @@ -253,7 +253,6 @@ func createInterfacesAndRoutesFromNS(conn *urpc.Client, nsPath string, conf *con args.XDPLinks = append(args.XDPLinks, boot.XDPLink{ Name: iface.Name, InterfaceIndex: iface.Index, - MTU: iface.MTU, Routes: routes, TXChecksumOffload: conf.TXChecksumOffload, RXChecksumOffload: conf.RXChecksumOffload, diff --git a/tools/xdp/tcpdump.go b/tools/xdp/tcpdump.go index 8f655ff61..ac651755c 100644 --- a/tools/xdp/tcpdump.go +++ b/tools/xdp/tcpdump.go @@ -106,7 +106,7 @@ func (pc *TcpdumpCommand) execute() error { } defer cleanup() - umem, fillQueue, rxQueue, err := xdp.ReadOnlySocket( + controlBlock, err := xdp.ReadOnlySocket( uint32(iface.Index), 0 /* queueID */, xdp.DefaultReadOnlyOpts()) if err != nil { return fmt.Errorf("failed to create socket: %v", err) @@ -115,18 +115,22 @@ func (pc *TcpdumpCommand) execute() error { // Insert our AF_XDP socket into the BPF map that dictates where // packets are redirected to. key := uint32(0) - val := umem.SockFD() + val := controlBlock.UMEM.SockFD() if err := objects.SockMap.Update(&key, &val, 0 /* flags */); err != nil { return fmt.Errorf("failed to insert socket into BPF map: %v", err) } log.Printf("updated key %d to value %d", key, val) // Put as many UMEM buffers into the fill queue as possible. - fillQueue.FillAll() + controlBlock.UMEM.Lock() + controlBlock.Fill.FillAll(&controlBlock.UMEM) + controlBlock.UMEM.Unlock() go func() { + controlBlock.UMEM.Lock() + defer controlBlock.UMEM.Unlock() for { - pfds := []unix.PollFd{{Fd: int32(umem.SockFD()), Events: unix.POLLIN}} + pfds := []unix.PollFd{{Fd: int32(controlBlock.UMEM.SockFD()), Events: unix.POLLIN}} _, err := unix.Poll(pfds, -1) if err != nil { if errors.Is(err, unix.EINTR) { @@ -136,19 +140,19 @@ func (pc *TcpdumpCommand) execute() error { } // How many packets did we get? - nReceived, rxIndex := rxQueue.Peek() + nReceived, rxIndex := controlBlock.RX.Peek() if nReceived == 0 { continue } // Keep the fill queue full. - fillQueue.FillAll() + controlBlock.Fill.FillAll(&controlBlock.UMEM) // Read packets one-by-one and log them. for i := uint32(0); i < nReceived; i++ { // Wrap the packet in a PacketBuffer. - descriptor := rxQueue.Get(rxIndex + i) - data := umem.Get(descriptor) + descriptor := controlBlock.RX.Get(rxIndex + i) + data := controlBlock.UMEM.Get(descriptor) pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ Payload: bufferv2.MakeWithData(data[header.EthernetMinimumSize:]), }) @@ -164,9 +168,9 @@ func (pc *TcpdumpCommand) execute() error { // problem. // // Note that this limits MTU to 4096-256 bytes. - umem.FreeFrame(descriptor.Addr) + controlBlock.UMEM.FreeFrame(descriptor.Addr) } - rxQueue.Release(nReceived) + controlBlock.RX.Release(nReceived) } }()