xdp: TX support

PiperOrigin-RevId: 480680001

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"],

pkg/memutil/memutil.go

@@ -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

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
+}

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
 	}

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",

pkg/tcpip/link/xdp/dispatcher.go

@@ -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
-	}
-}

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
+	}
 }

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",
     ],
 )

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)
+}

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()
 }

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]
 }

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
+}

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]
 }

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
 }

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
+}

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,

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,

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)
 		}
 	}()