nebula/overlay/tio/queueset_gso_linux.go

//go:build linux && !android

package tio

import (
	"encoding/binary"
	"errors"
	"fmt"
	"log/slog"
	"sync/atomic"

	"golang.org/x/sys/unix"
)

type offloadQueueSet struct {
	pq []*Offload
	// pqi is exactly the same as pq, but stored as the interface type
	pqi        []Queue
	shutdownFd int
	// usoEnabled is true when newTun successfully negotiated TUN_F_USO4|6 with the kernel.
	// Queues created by Add inherit this and surface it via Offload.USOSupported so coalescers can gate USO emission.
	usoEnabled bool
	shutdown   atomic.Bool
	closed     atomic.Bool
	// l is handed to each queue for its bad-vnet-header drop logging.
	l *slog.Logger
}

// NewOffloadQueueSet creates a QueueSet that uses virtio_net_hdr to do TSO segmentation.
// usoEnabled tells downstream queues whether the kernel agreed to deliver/accept GSO_UDP_L4 superpackets.
func NewOffloadQueueSet(usoEnabled bool, l *slog.Logger) (QueueSet, error) {
	shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
	if err != nil {
		return nil, fmt.Errorf("failed to create eventfd: %w", err)
	}

	out := &offloadQueueSet{
		pq:         []*Offload{},
		pqi:        []Queue{},
		shutdownFd: shutdownFd,
		usoEnabled: usoEnabled,
		l:          l,
	}

	return out, nil
}

func (c *offloadQueueSet) Queues() []Queue {
	return c.pqi
}

func (c *offloadQueueSet) Add(fd int) error {
	if c.closed.Load() {
		return errors.New("queue set already closed")
	}
	x, err := newOffload(fd, c.shutdownFd, c.usoEnabled, c.l)
	if err != nil {
		return err
	}
	c.pq = append(c.pq, x)
	c.pqi = append(c.pqi, x)

	return nil
}

func (c *offloadQueueSet) wakeForShutdown() error {
	var buf [8]byte
	binary.NativeEndian.PutUint64(buf[:], 1)
	_, err := unix.Write(c.shutdownFd, buf[:])
	return err
}

// Shutdown wakes every reader without releasing any fds, they see POLLIN on the eventfd and return os.ErrClosed
func (c *offloadQueueSet) Shutdown() error {
	if c.shutdown.Swap(true) {
		return nil
	}
	for _, x := range c.pq {
		x.shutdown.Store(true)
	}
	return c.wakeForShutdown()
}

func (c *offloadQueueSet) Close() error {
	if c.closed.Swap(true) {
		return nil
	}

	errs := []error{}

	if err := c.Shutdown(); err != nil {
		errs = append(errs, err)
	}

	// Close the per-queue tun fds; this also unblocks any in-flight reads.
	for _, x := range c.pq {
		if err := x.Close(); err != nil {
			errs = append(errs, err)
		}
	}

	// Close the shutdown eventfd last: every reader's pollfd set references it,
	// so it must outlive the wake + per-queue teardown above.
	if err := unix.Close(c.shutdownFd); err != nil {
		errs = append(errs, err)
	}
	c.shutdownFd = -1

	return errors.Join(errs...)
}