checkpt

.github/workflows/smoke-extra.yml

@@ -18,8 +18,6 @@ jobs:
     if: github.ref == 'refs/heads/master' || contains(github.event.pull_request.labels.*.name, 'smoke-test-extra')
     name: Run extra smoke tests
     runs-on: ubuntu-latest
-    env:
-      VAGRANT_DEFAULT_PROVIDER: libvirt
     steps:
 
     - uses: actions/checkout@v6
@@ -32,13 +30,11 @@ jobs:
     - name: add hashicorp source
       run: wget -O- https://apt.releases.hashicorp.com/gpg | gpg --dearmor | sudo tee /usr/share/keyrings/hashicorp-archive-keyring.gpg && echo "deb [signed-by=/usr/share/keyrings/hashicorp-archive-keyring.gpg] https://apt.releases.hashicorp.com $(lsb_release -cs) main" | sudo tee /etc/apt/sources.list.d/hashicorp.list
 
-    - name: install vagrant and libvirt
+    - name: workaround AMD-V issue  # https://github.com/cri-o/packaging/pull/306
-      run: |
+      run: sudo rmmod kvm_amd
-        sudo apt-get update && sudo apt-get install -y vagrant libvirt-daemon-system libvirt-dev
+
-        sudo chmod 666 /dev/kvm
+    - name: install vagrant
-        sudo usermod -aG libvirt $(whoami)
+      run: sudo apt-get update && sudo apt-get install -y vagrant virtualbox
-        sudo chmod 666 /var/run/libvirt/libvirt-sock
-        vagrant plugin install vagrant-libvirt
 
     - name: freebsd-amd64
       run: make smoke-vagrant/freebsd-amd64
@@ -49,19 +45,10 @@ jobs:
     - name: netbsd-amd64
       run: make smoke-vagrant/netbsd-amd64
 
+    - name: linux-386
+      run: make smoke-vagrant/linux-386
+
     - name: linux-amd64-ipv6disable
       run: make smoke-vagrant/linux-amd64-ipv6disable
 
-    # linux-386 runs last because it requires disabling KVM to use VirtualBox,
-    # which prevents libvirt (used by the other tests) from working after this point.
-    - name: install virtualbox for i386 test
-      run: |
-        sudo apt-get install -y virtualbox
-        sudo rmmod kvm_amd kvm_intel kvm 2>/dev/null || true
-
-    - name: linux-386
-      env:
-        VAGRANT_DEFAULT_PROVIDER: virtualbox
-      run: make smoke-vagrant/linux-386
-
     timeout-minutes: 30

.github/workflows/smoke/build-relay.sh

@@ -16,10 +16,8 @@ relay:
   am_relay: true
 EOF
 
-    # TEST-NET-3 placeholder IPs; smoke-relay.sh seds them to real container IPs.
+    export LIGHTHOUSES="192.168.100.1 172.17.0.2:4242"
-    # Mapping: .2 lighthouse1, .3 host2, .4 host3, .5 host4.
+    export REMOTE_ALLOW_LIST='{"172.17.0.4/32": false, "172.17.0.5/32": false}'
-    export LIGHTHOUSES="192.168.100.1 203.0.113.2:4242"
-    export REMOTE_ALLOW_LIST='{"203.0.113.4/32": false, "203.0.113.5/32": false}'
 
     HOST="host2" ../genconfig.sh >host2.yml <<EOF
 relay:
@@ -27,7 +25,7 @@ relay:
     - 192.168.100.1
 EOF
 
-    export REMOTE_ALLOW_LIST='{"203.0.113.3/32": false}'
+    export REMOTE_ALLOW_LIST='{"172.17.0.3/32": false}'
 
     HOST="host3" ../genconfig.sh >host3.yml
 

.github/workflows/smoke/build.sh

@@ -5,15 +5,9 @@ set -e -x
 rm -rf ./build
 mkdir ./build
 
-# Smoke containers run on a dedicated docker network whose subnet is allocated
+# TODO: Assumes your docker bridge network is a /24, and the first container that launches will be .1
-# at smoke time, not known at build time. Configs are written with TEST-NET-3
+# - We could make this better by launching the lighthouse first and then fetching what IP it is.
-# placeholder IPs (RFC 5737) and smoke.sh / smoke-vagrant.sh / smoke-relay.sh
+NET="$(docker network inspect bridge -f '{{ range .IPAM.Config }}{{ .Subnet }}{{ end }}' | cut -d. -f1-3)"
-# sed the real container IPs in before starting nebula.
-#
-# Placeholder mapping (last octet == fixed container slot):
-#   203.0.113.2 -> lighthouse1, 203.0.113.3 -> host2,
-#   203.0.113.4 -> host3,       203.0.113.5 -> host4.
-LIGHTHOUSE_IP="203.0.113.2"
 
 (
     cd build
@@ -31,16 +25,16 @@ LIGHTHOUSE_IP="203.0.113.2"
         ../genconfig.sh >lighthouse1.yml
 
     HOST="host2" \
-        LIGHTHOUSES="192.168.100.1 $LIGHTHOUSE_IP:4242" \
+        LIGHTHOUSES="192.168.100.1 $NET.2:4242" \
         ../genconfig.sh >host2.yml
 
     HOST="host3" \
-        LIGHTHOUSES="192.168.100.1 $LIGHTHOUSE_IP:4242" \
+        LIGHTHOUSES="192.168.100.1 $NET.2:4242" \
         INBOUND='[{"port": "any", "proto": "icmp", "group": "lighthouse"}]' \
         ../genconfig.sh >host3.yml
 
     HOST="host4" \
-        LIGHTHOUSES="192.168.100.1 $LIGHTHOUSE_IP:4242" \
+        LIGHTHOUSES="192.168.100.1 $NET.2:4242" \
         OUTBOUND='[{"port": "any", "proto": "icmp", "group": "lighthouse"}]' \
         ../genconfig.sh >host4.yml
 

.github/workflows/smoke/smoke-relay.sh

@@ -6,8 +6,6 @@ set -o pipefail
 
 mkdir -p logs
 
-NETWORK="nebula-smoke-relay"
-
 cleanup() {
     echo
     echo " *** cleanup"
@@ -18,53 +16,22 @@ cleanup() {
     then
         docker kill lighthouse1 host2 host3 host4
     fi
-    docker network rm "$NETWORK" >/dev/null 2>&1
 }
 
 trap cleanup EXIT
 
-# Create a dedicated smoke network with an explicit subnet (required for --ip
-# below). Probe a short list of candidates so a locally-used range doesn't
-# fail the whole test — we only need one to be free.
-docker network rm "$NETWORK" >/dev/null 2>&1 || true
-for candidate in 172.30.0.0/24 172.31.0.0/24 10.98.0.0/24 10.99.0.0/24 192.168.230.0/24; do
-    if docker network create --subnet "$candidate" "$NETWORK" >/dev/null 2>&1; then
-        break
-    fi
-done
-if ! docker network inspect "$NETWORK" >/dev/null 2>&1; then
-    echo "failed to create $NETWORK: every candidate subnet is in use" >&2
-    exit 1
-fi
-
-# Derive container IPs from the network's assigned subnet. Slots: .2 lighthouse1,
-# .3 host2, .4 host3, .5 host4 — matches the placeholders in build-relay.sh.
-SUBNET="$(docker network inspect -f '{{(index .IPAM.Config 0).Subnet}}' "$NETWORK")"
-PREFIX="${SUBNET%/*}"
-PREFIX="${PREFIX%.*}"
-LIGHTHOUSE_IP="$PREFIX.2"
-HOST2_IP="$PREFIX.3"
-HOST3_IP="$PREFIX.4"
-HOST4_IP="$PREFIX.5"
-
-# Sed the placeholder TEST-NET-3 IPs in the host configs to the real ones.
-for f in build/host2.yml build/host3.yml build/host4.yml; do
-    sed "s|203\.0\.113\.|$PREFIX.|g" "$f" >"$f.tmp"
-    mv "$f.tmp" "$f"
-done
-
 docker run --name lighthouse1 --rm nebula:smoke-relay -config lighthouse1.yml -test
-docker run --name host2 --rm -v "$PWD/build/host2.yml:/nebula/host2.yml:ro" nebula:smoke-relay -config host2.yml -test
+docker run --name host2 --rm nebula:smoke-relay -config host2.yml -test
-docker run --name host3 --rm -v "$PWD/build/host3.yml:/nebula/host3.yml:ro" nebula:smoke-relay -config host3.yml -test
+docker run --name host3 --rm nebula:smoke-relay -config host3.yml -test
-docker run --name host4 --rm -v "$PWD/build/host4.yml:/nebula/host4.yml:ro" nebula:smoke-relay -config host4.yml -test
+docker run --name host4 --rm nebula:smoke-relay -config host4.yml -test
 
-docker run --name lighthouse1 --network "$NETWORK" --ip "$LIGHTHOUSE_IP" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config lighthouse1.yml 2>&1 | tee logs/lighthouse1 | sed -u 's/^/  [lighthouse1]  /' &
+docker run --name lighthouse1 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config lighthouse1.yml 2>&1 | tee logs/lighthouse1 | sed -u 's/^/  [lighthouse1]  /' &
 sleep 1
-docker run --name host2 --network "$NETWORK" --ip "$HOST2_IP" -v "$PWD/build/host2.yml:/nebula/host2.yml:ro" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config host2.yml 2>&1 | tee logs/host2 | sed -u 's/^/  [host2]  /' &
+docker run --name host2 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config host2.yml 2>&1 | tee logs/host2 | sed -u 's/^/  [host2]  /' &
 sleep 1
-docker run --name host3 --network "$NETWORK" --ip "$HOST3_IP" -v "$PWD/build/host3.yml:/nebula/host3.yml:ro" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config host3.yml 2>&1 | tee logs/host3 | sed -u 's/^/  [host3]  /' &
+docker run --name host3 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config host3.yml 2>&1 | tee logs/host3 | sed -u 's/^/  [host3]  /' &
 sleep 1
-docker run --name host4 --network "$NETWORK" --ip "$HOST4_IP" -v "$PWD/build/host4.yml:/nebula/host4.yml:ro" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config host4.yml 2>&1 | tee logs/host4 | sed -u 's/^/  [host4]  /' &
+docker run --name host4 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm nebula:smoke-relay -config host4.yml 2>&1 | tee logs/host4 | sed -u 's/^/  [host4]  /' &
 sleep 1
 
 set +x
@@ -109,13 +76,7 @@ docker exec host4 sh -c 'kill 1'
 docker exec host3 sh -c 'kill 1'
 docker exec host2 sh -c 'kill 1'
 docker exec lighthouse1 sh -c 'kill 1'
-
+sleep 5
-# Wait up to 30s for all backgrounded jobs to exit rather than relying on a
-# fixed sleep.
-for _ in $(seq 1 30); do
-    [ -z "$(jobs -r)" ] && break
-    sleep 1
-done
 
 if [ "$(jobs -r)" ]
 then

.github/workflows/smoke/smoke-vagrant.sh

@@ -8,8 +8,6 @@ export VAGRANT_CWD="$PWD/vagrant-$1"
 
 mkdir -p logs
 
-NETWORK="nebula-smoke"
-
 cleanup() {
     echo
     echo " *** cleanup"
@@ -21,51 +19,32 @@ cleanup() {
         docker kill lighthouse1 host2
     fi
     vagrant destroy -f
-    docker network rm "$NETWORK" >/dev/null 2>&1
 }
 
 trap cleanup EXIT
 
-# Create a dedicated smoke network with an explicit subnet (required for --ip
-# below). Probe a short list of candidates so a locally-used range doesn't
-# fail the whole test — we only need one to be free.
-docker network rm "$NETWORK" >/dev/null 2>&1 || true
-for candidate in 172.30.0.0/24 172.31.0.0/24 10.98.0.0/24 10.99.0.0/24 192.168.230.0/24; do
-    if docker network create --subnet "$candidate" "$NETWORK" >/dev/null 2>&1; then
-        break
-    fi
-done
-if ! docker network inspect "$NETWORK" >/dev/null 2>&1; then
-    echo "failed to create $NETWORK: every candidate subnet is in use" >&2
-    exit 1
-fi
-
-# Derive container IPs from the network's assigned subnet. Slots: .2 lighthouse1,
-# .3 host2 — matches the placeholders in build.sh.
-SUBNET="$(docker network inspect -f '{{(index .IPAM.Config 0).Subnet}}' "$NETWORK")"
-PREFIX="${SUBNET%/*}"
-PREFIX="${PREFIX%.*}"
-LIGHTHOUSE_IP="$PREFIX.2"
-HOST2_IP="$PREFIX.3"
-
-# Sed the placeholder TEST-NET-3 IPs in the host configs to the real ones.
-# This must happen before `vagrant up` rsyncs build/ into the VM for host3.
-for f in build/host2.yml build/host3.yml; do
-    sed "s|203\.0\.113\.|$PREFIX.|g" "$f" >"$f.tmp"
-    mv "$f.tmp" "$f"
-done
-
 CONTAINER="nebula:${NAME:-smoke}"
 
 docker run --name lighthouse1 --rm "$CONTAINER" -config lighthouse1.yml -test
-docker run --name host2 --rm -v "$PWD/build/host2.yml:/nebula/host2.yml:ro" "$CONTAINER" -config host2.yml -test
+docker run --name host2 --rm "$CONTAINER" -config host2.yml -test
 
 vagrant up
+
+# OpenBSD: synced folders are disabled because Vagrant's rsync installer
+# uses ftp.openbsd.org which no longer hosts packages for older releases.
+# Copy build artifacts in via scp instead.
+case "$1" in
+    openbsd-*)
+        vagrant ssh -c "sudo mkdir -p /nebula" -- -T
+        tar -cf - -C build . | vagrant ssh -c "sudo tar -xf - -C /nebula && sudo chmod -R a+r /nebula" -- -T
+        ;;
+esac
+
 vagrant ssh -c "cd /nebula && /nebula/$1-nebula -config host3.yml -test" -- -T
 
-docker run --name lighthouse1 --network "$NETWORK" --ip "$LIGHTHOUSE_IP" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config lighthouse1.yml 2>&1 | tee logs/lighthouse1 | sed -u 's/^/  [lighthouse1]  /' &
+docker run --name lighthouse1 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config lighthouse1.yml 2>&1 | tee logs/lighthouse1 | sed -u 's/^/  [lighthouse1]  /' &
 sleep 1
-docker run --name host2 --network "$NETWORK" --ip "$HOST2_IP" -v "$PWD/build/host2.yml:/nebula/host2.yml:ro" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host2.yml 2>&1 | tee logs/host2 | sed -u 's/^/  [host2]  /' &
+docker run --name host2 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host2.yml 2>&1 | tee logs/host2 | sed -u 's/^/  [host2]  /' &
 sleep 1
 vagrant ssh -c "cd /nebula && sudo sh -c 'echo \$\$ >/nebula/pid && exec /nebula/$1-nebula -config host3.yml'" 2>&1 -- -T | tee logs/host3 | sed -u 's/^/  [host3]  /' &
 sleep 15
@@ -128,14 +107,7 @@ vagrant ssh -c "ping -c1 192.168.100.2" -- -T
 vagrant ssh -c "sudo xargs kill </nebula/pid" -- -T
 docker exec host2 sh -c 'kill 1'
 docker exec lighthouse1 sh -c 'kill 1'
-
+sleep 1
-# Wait up to 30s for all backgrounded jobs to exit. vagrant ssh in particular
-# takes a beat to tear down after nebula exits on the VM, so a fixed sleep is
-# racy.
-for _ in $(seq 1 30); do
-    [ -z "$(jobs -r)" ] && break
-    sleep 1
-done
 
 if [ "$(jobs -r)" ]
 then

.github/workflows/smoke/smoke.sh

@@ -6,8 +6,6 @@ set -o pipefail
 
 mkdir -p logs
 
-NETWORK="nebula-smoke"
-
 cleanup() {
     echo
     echo " *** cleanup"
@@ -18,56 +16,24 @@ cleanup() {
     then
         docker kill lighthouse1 host2 host3 host4
     fi
-    docker network rm "$NETWORK" >/dev/null 2>&1
 }
 
 trap cleanup EXIT
 
-# Create a dedicated smoke network with an explicit subnet (required for --ip
-# below). Probe a short list of candidates so a locally-used range doesn't
-# fail the whole test — we only need one to be free.
-docker network rm "$NETWORK" >/dev/null 2>&1 || true
-for candidate in 172.30.0.0/24 172.31.0.0/24 10.98.0.0/24 10.99.0.0/24 192.168.230.0/24; do
-    if docker network create --subnet "$candidate" "$NETWORK" >/dev/null 2>&1; then
-        break
-    fi
-done
-if ! docker network inspect "$NETWORK" >/dev/null 2>&1; then
-    echo "failed to create $NETWORK: every candidate subnet is in use" >&2
-    exit 1
-fi
-
-# Derive container IPs from the network's assigned subnet. Slots: .2 lighthouse1,
-# .3 host2, .4 host3, .5 host4 — matches the placeholders in build.sh.
-SUBNET="$(docker network inspect -f '{{(index .IPAM.Config 0).Subnet}}' "$NETWORK")"
-PREFIX="${SUBNET%/*}"
-PREFIX="${PREFIX%.*}"
-LIGHTHOUSE_IP="$PREFIX.2"
-HOST2_IP="$PREFIX.3"
-HOST3_IP="$PREFIX.4"
-HOST4_IP="$PREFIX.5"
-
-# Sed the placeholder TEST-NET-3 IPs in the host configs to the real ones.
-# build/lighthouse1.yml has no IPs to rewrite so it's skipped.
-for f in build/host2.yml build/host3.yml build/host4.yml; do
-    sed "s|203\.0\.113\.|$PREFIX.|g" "$f" >"$f.tmp"
-    mv "$f.tmp" "$f"
-done
-
 CONTAINER="nebula:${NAME:-smoke}"
 
 docker run --name lighthouse1 --rm "$CONTAINER" -config lighthouse1.yml -test
-docker run --name host2 --rm -v "$PWD/build/host2.yml:/nebula/host2.yml:ro" "$CONTAINER" -config host2.yml -test
+docker run --name host2 --rm "$CONTAINER" -config host2.yml -test
-docker run --name host3 --rm -v "$PWD/build/host3.yml:/nebula/host3.yml:ro" "$CONTAINER" -config host3.yml -test
+docker run --name host3 --rm "$CONTAINER" -config host3.yml -test
-docker run --name host4 --rm -v "$PWD/build/host4.yml:/nebula/host4.yml:ro" "$CONTAINER" -config host4.yml -test
+docker run --name host4 --rm "$CONTAINER" -config host4.yml -test
 
-docker run --name lighthouse1 --network "$NETWORK" --ip "$LIGHTHOUSE_IP" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config lighthouse1.yml 2>&1 | tee logs/lighthouse1 | sed -u 's/^/  [lighthouse1]  /' &
+docker run --name lighthouse1 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config lighthouse1.yml 2>&1 | tee logs/lighthouse1 | sed -u 's/^/  [lighthouse1]  /' &
 sleep 1
-docker run --name host2 --network "$NETWORK" --ip "$HOST2_IP" -v "$PWD/build/host2.yml:/nebula/host2.yml:ro" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host2.yml 2>&1 | tee logs/host2 | sed -u 's/^/  [host2]  /' &
+docker run --name host2 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host2.yml 2>&1 | tee logs/host2 | sed -u 's/^/  [host2]  /' &
 sleep 1
-docker run --name host3 --network "$NETWORK" --ip "$HOST3_IP" -v "$PWD/build/host3.yml:/nebula/host3.yml:ro" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host3.yml 2>&1 | tee logs/host3 | sed -u 's/^/  [host3]  /' &
+docker run --name host3 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host3.yml 2>&1 | tee logs/host3 | sed -u 's/^/  [host3]  /' &
 sleep 1
-docker run --name host4 --network "$NETWORK" --ip "$HOST4_IP" -v "$PWD/build/host4.yml:/nebula/host4.yml:ro" --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host4.yml 2>&1 | tee logs/host4 | sed -u 's/^/  [host4]  /' &
+docker run --name host4 --device /dev/net/tun:/dev/net/tun --cap-add NET_ADMIN --rm "$CONTAINER" -config host4.yml 2>&1 | tee logs/host4 | sed -u 's/^/  [host4]  /' &
 sleep 1
 
 # grab tcpdump pcaps for debugging
@@ -165,13 +131,7 @@ docker exec host4 sh -c 'kill 1'
 docker exec host3 sh -c 'kill 1'
 docker exec host2 sh -c 'kill 1'
 docker exec lighthouse1 sh -c 'kill 1'
-
+sleep 5
-# Wait up to 30s for all backgrounded jobs to exit rather than relying on a
-# fixed sleep.
-for _ in $(seq 1 30); do
-    [ -z "$(jobs -r)" ] && break
-    sleep 1
-done
 
 if [ "$(jobs -r)" ]
 then

.github/workflows/smoke/vagrant-linux-amd64-ipv6disable/Vagrantfile

@@ -1,7 +1,7 @@
 # -*- mode: ruby -*-
 # vi: set ft=ruby :
 Vagrant.configure("2") do |config|
-  config.vm.box = "bento/ubuntu-24.04"
+  config.vm.box = "ubuntu/jammy64"
 
   config.vm.synced_folder "../build", "/nebula"
 

.github/workflows/smoke/vagrant-openbsd-amd64/Vagrantfile

@@ -1,7 +1,7 @@
 # -*- mode: ruby -*-
 # vi: set ft=ruby :
 Vagrant.configure("2") do |config|
-  config.vm.box = "DefinedNet/openbsd78"
+  config.vm.box = "generic/openbsd7"
 
-  config.vm.synced_folder "../build", "/nebula", type: "rsync"
+  config.vm.synced_folder ".", "/vagrant", disabled: true
 end

batch.go

@@ -1,70 +0,0 @@
-package nebula
-
-import "net/netip"
-
-// sendBatchCap is the maximum number of encrypted packets accumulated before a
-// flush is forced. TSO superpackets segment to at most ~45 packets on
-// reasonable MTUs, so 128 leaves headroom without bloating the backing
-// allocation.
-const sendBatchCap = 128
-
-// sendBatch accumulates encrypted UDP packets for a single sendmmsg flush.
-// One sendBatch is owned by each listenIn goroutine; no locking is needed.
-// The backing storage holds up to batchCap packets of slotCap bytes each;
-// bufs and dsts are parallel slices of committed slots.
-type sendBatch struct {
-	bufs     [][]byte
-	dsts     []netip.AddrPort
-	backing  []byte
-	slotCap  int
-	batchCap int
-	nextSlot int
-}
-
-func newSendBatch(batchCap, slotCap int) *sendBatch {
-	return &sendBatch{
-		bufs:     make([][]byte, 0, batchCap),
-		dsts:     make([]netip.AddrPort, 0, batchCap),
-		backing:  make([]byte, batchCap*slotCap),
-		slotCap:  slotCap,
-		batchCap: batchCap,
-	}
-}
-
-// Next returns a zero-length slice with slotCap capacity over the next unused
-// slot's backing bytes. The caller writes into the returned slice and then
-// calls Commit with the final length and destination. Next returns nil when
-// the batch is full.
-func (b *sendBatch) Next() []byte {
-	if b.nextSlot >= b.batchCap {
-		return nil
-	}
-	start := b.nextSlot * b.slotCap
-	return b.backing[start : start : start+b.slotCap]
-}
-
-// Commit records the slot just returned by Next as a packet of length n
-// destined for dst.
-func (b *sendBatch) Commit(n int, dst netip.AddrPort) {
-	start := b.nextSlot * b.slotCap
-	b.bufs = append(b.bufs, b.backing[start:start+n])
-	b.dsts = append(b.dsts, dst)
-	b.nextSlot++
-}
-
-// Reset clears committed slots; backing storage is retained for reuse.
-func (b *sendBatch) Reset() {
-	b.bufs = b.bufs[:0]
-	b.dsts = b.dsts[:0]
-	b.nextSlot = 0
-}
-
-// Len returns the number of committed packets.
-func (b *sendBatch) Len() int {
-	return len(b.bufs)
-}
-
-// Cap returns the maximum number of slots in the batch.
-func (b *sendBatch) Cap() int {
-	return b.batchCap
-}

batch_test.go

@@ -1,69 +0,0 @@
-package nebula
-
-import (
-	"net/netip"
-	"testing"
-)
-
-func TestSendBatchBookkeeping(t *testing.T) {
-	b := newSendBatch(4, 32)
-	if b.Len() != 0 || b.Cap() != 4 {
-		t.Fatalf("fresh batch: len=%d cap=%d", b.Len(), b.Cap())
-	}
-
-	ap := netip.MustParseAddrPort("10.0.0.1:4242")
-	for i := 0; i < 4; i++ {
-		slot := b.Next()
-		if slot == nil {
-			t.Fatalf("slot %d: Next returned nil before cap", i)
-		}
-		if cap(slot) != 32 || len(slot) != 0 {
-			t.Fatalf("slot %d: got len=%d cap=%d want len=0 cap=32", i, len(slot), cap(slot))
-		}
-		// Write a marker byte.
-		slot = append(slot, byte(i), byte(i+1), byte(i+2))
-		b.Commit(len(slot), ap)
-	}
-	if b.Next() != nil {
-		t.Fatalf("Next should return nil when full")
-	}
-	if b.Len() != 4 {
-		t.Fatalf("Len=%d want 4", b.Len())
-	}
-	for i, buf := range b.bufs {
-		if len(buf) != 3 || buf[0] != byte(i) {
-			t.Errorf("buf %d: %x", i, buf)
-		}
-		if b.dsts[i] != ap {
-			t.Errorf("dst %d: got %v want %v", i, b.dsts[i], ap)
-		}
-	}
-
-	// Reset returns empty and Next works again.
-	b.Reset()
-	if b.Len() != 0 {
-		t.Fatalf("after Reset Len=%d want 0", b.Len())
-	}
-	slot := b.Next()
-	if slot == nil || cap(slot) != 32 {
-		t.Fatalf("after Reset Next nil or wrong cap: %v cap=%d", slot == nil, cap(slot))
-	}
-}
-
-func TestSendBatchSlotsDoNotOverlap(t *testing.T) {
-	b := newSendBatch(3, 8)
-	ap := netip.MustParseAddrPort("10.0.0.1:80")
-
-	// Fill three slots, each with its own sentinel byte.
-	for i := 0; i < 3; i++ {
-		s := b.Next()
-		s = append(s, byte(0xA0+i), byte(0xB0+i))
-		b.Commit(len(s), ap)
-	}
-
-	for i, buf := range b.bufs {
-		if buf[0] != byte(0xA0+i) || buf[1] != byte(0xB0+i) {
-			t.Errorf("slot %d corrupted: %x", i, buf)
-		}
-	}
-}

cmd/nebula-service/main.go

@@ -78,20 +78,8 @@ func main() {
 	}
 
 	if !*configTest {
-		wait, err := ctrl.Start()
+		ctrl.Start()
-		if err != nil {
+		ctrl.ShutdownBlock()
-			util.LogWithContextIfNeeded("Error while running", err, l)
-			os.Exit(1)
-		}
-
-		go ctrl.ShutdownBlock()
-
-		if err := wait(); err != nil {
-			l.WithError(err).Error("Nebula stopped due to fatal error")
-			os.Exit(2)
-		}
-
-		l.Info("Goodbye")
 	}
 
 	os.Exit(0)

cmd/nebula/main.go

@@ -72,21 +72,9 @@ func main() {
 	}
 
 	if !*configTest {
-		wait, err := ctrl.Start()
+		ctrl.Start()
-		if err != nil {
-			util.LogWithContextIfNeeded("Error while running", err, l)
-			os.Exit(1)
-		}
-
-		go ctrl.ShutdownBlock()
 		notifyReady(l)
-
+		ctrl.ShutdownBlock()
-		if err := wait(); err != nil {
-			l.WithError(err).Error("Nebula stopped due to fatal error")
-			os.Exit(2)
-		}
-
-		l.Info("Goodbye")
 	}
 
 	os.Exit(0)

connection_manager_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/flynn/noise"
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay"
 	"github.com/slackhq/nebula/test"
 	"github.com/slackhq/nebula/udp"
 	"github.com/stretchr/testify/assert"
@@ -53,7 +52,7 @@ func Test_NewConnectionManagerTest(t *testing.T) {
 	lh := newTestLighthouse()
 	ifce := &Interface{
 		hostMap:          hostMap,
-		inside:           &overlay.NoopTun{},
+		inside:           &test.NoopTun{},
 		outside:          &udp.NoopConn{},
 		firewall:         &Firewall{},
 		lightHouse:       lh,
@@ -136,7 +135,7 @@ func Test_NewConnectionManagerTest2(t *testing.T) {
 	lh := newTestLighthouse()
 	ifce := &Interface{
 		hostMap:          hostMap,
-		inside:           &overlay.NoopTun{},
+		inside:           &test.NoopTun{},
 		outside:          &udp.NoopConn{},
 		firewall:         &Firewall{},
 		lightHouse:       lh,
@@ -221,7 +220,7 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) {
 	lh := newTestLighthouse()
 	ifce := &Interface{
 		hostMap:          hostMap,
-		inside:           &overlay.NoopTun{},
+		inside:           &test.NoopTun{},
 		outside:          &udp.NoopConn{},
 		firewall:         &Firewall{},
 		lightHouse:       lh,
@@ -348,7 +347,7 @@ func Test_NewConnectionManagerTest_DisconnectInvalid(t *testing.T) {
 	lh := newTestLighthouse()
 	ifce := &Interface{
 		hostMap:          hostMap,
-		inside:           &overlay.NoopTun{},
+		inside:           &test.NoopTun{},
 		outside:          &udp.NoopConn{},
 		firewall:         &Firewall{},
 		lightHouse:       lh,

control.go

@@ -2,11 +2,9 @@ package nebula
 
 import (
 	"context"
-	"errors"
 	"net/netip"
 	"os"
 	"os/signal"
-	"sync"
 	"syscall"
 
 	"github.com/sirupsen/logrus"
@@ -15,20 +13,6 @@ import (
 	"github.com/slackhq/nebula/overlay"
 )
 
-type RunState int
-
-const (
-	StateUnknown RunState = iota
-	StateReady
-	StateStarted
-	StateStopping
-	StateStopped
-)
-
-var ErrAlreadyStarted = errors.New("nebula is already started")
-var ErrAlreadyStopped = errors.New("nebula cannot be restarted")
-var ErrUnknownState = errors.New("nebula state is invalid")
-
 // Every interaction here needs to take extra care to copy memory and not return or use arguments "as is" when touching
 // core. This means copying IP objects, slices, de-referencing pointers and taking the actual value, etc
 
@@ -42,9 +26,6 @@ type controlHostLister interface {
 }
 
 type Control struct {
-	stateLock sync.Mutex
-	state     RunState
-
 	f                      *Interface
 	l                      *logrus.Logger
 	ctx                    context.Context
@@ -68,31 +49,10 @@ type ControlHostInfo struct {
 	CurrentRelaysThroughMe []netip.Addr     `json:"currentRelaysThroughMe"`
 }
 
-// Start actually runs nebula, this is a nonblocking call.
+// Start actually runs nebula, this is a nonblocking call. To block use Control.ShutdownBlock()
-// The returned function blocks until nebula has fully stopped and returns the
+func (c *Control) Start() {
-// first fatal reader error (if any). A nil error means nebula shut down
-// gracefully; a non-nil error means a reader hit an unexpected failure that
-// triggered the shutdown.
-func (c *Control) Start() (func() error, error) {
-	c.stateLock.Lock()
-	defer c.stateLock.Unlock()
-	switch c.state {
-	case StateReady:
-		//yay!
-	case StateStopped, StateStopping:
-		return nil, ErrAlreadyStopped
-	case StateStarted:
-		return nil, ErrAlreadyStarted
-	default:
-		return nil, ErrUnknownState
-	}
-
 	// Activate the interface
-	err := c.f.activate()
+	c.f.activate()
-	if err != nil {
-		c.state = StateStopped
-		return nil, err
-	}
 
 	// Call all the delayed funcs that waited patiently for the interface to be created.
 	if c.sshStart != nil {
@@ -111,40 +71,16 @@ func (c *Control) Start() (func() error, error) {
 		c.lighthouseStart()
 	}
 
-	c.f.triggerShutdown = c.Stop
-
 	// Start reading packets.
-	out, err := c.f.run()
+	c.f.run()
-	if err != nil {
-		c.state = StateStopped
-		return nil, err
-	}
-	c.state = StateStarted
-	return out, nil
-}
-
-func (c *Control) State() RunState {
-	c.stateLock.Lock()
-	defer c.stateLock.Unlock()
-	return c.state
 }
 
 func (c *Control) Context() context.Context {
 	return c.ctx
 }
 
-// Stop is a non-blocking call that signals nebula to close all tunnels and shut down
+// Stop signals nebula to shutdown and close all tunnels, returns after the shutdown is complete
 func (c *Control) Stop() {
-	c.stateLock.Lock()
-	if c.state != StateStarted {
-		c.stateLock.Unlock()
-		// We are stopping or stopped already
-		return
-	}
-
-	c.state = StateStopping
-	c.stateLock.Unlock()
-
 	// Stop the handshakeManager (and other services), to prevent new tunnels from
 	// being created while we're shutting them all down.
 	c.cancel()
@@ -153,9 +89,7 @@ func (c *Control) Stop() {
 	if err := c.f.Close(); err != nil {
 		c.l.WithError(err).Error("Close interface failed")
 	}
-	c.stateLock.Lock()
+	c.l.Info("Goodbye")
-	c.state = StateStopped
-	c.stateLock.Unlock()
 }
 
 // ShutdownBlock will listen for and block on term and interrupt signals, calling Control.Stop() once signalled

control_test.go

@@ -79,7 +79,6 @@ func TestControl_GetHostInfoByVpnIp(t *testing.T) {
 	}, &Interface{})
 
 	c := Control{
-		state: StateReady,
 		f: &Interface{
 			hostMap: hm,
 		},

inside.go

@@ -11,7 +11,7 @@ import (
 	"github.com/slackhq/nebula/routing"
 )
 
-func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet, nb []byte, batch *sendBatch, rejectBuf []byte, q int, localCache firewall.ConntrackCache) {
+func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet, nb, out []byte, q int, localCache firewall.ConntrackCache) {
 	err := newPacket(packet, false, fwPacket)
 	if err != nil {
 		if f.l.Level >= logrus.DebugLevel {
@@ -33,7 +33,7 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 		// routes packets from the Nebula addr to the Nebula addr through the Nebula
 		// TUN device.
 		if immediatelyForwardToSelf {
-			_, err := f.readers[q].WriteReject(packet)
+			_, err := f.readers[q].Write(packet)
 			if err != nil {
 				f.l.WithError(err).Error("Failed to forward to tun")
 			}
@@ -53,7 +53,7 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 	})
 
 	if hostinfo == nil {
-		f.rejectInside(packet, rejectBuf, q)
+		f.rejectInside(packet, out, q)
 		if f.l.Level >= logrus.DebugLevel {
 			f.l.WithField("vpnAddr", fwPacket.RemoteAddr).
 				WithField("fwPacket", fwPacket).
@@ -68,10 +68,10 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 
 	dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache)
 	if dropReason == nil {
-		f.sendInsideMessage(hostinfo, packet, nb, batch, rejectBuf, q)
+		f.sendNoMetrics(header.Message, 0, hostinfo.ConnectionState, hostinfo, netip.AddrPort{}, packet, nb, out, q)
 
 	} else {
-		f.rejectInside(packet, rejectBuf, q)
+		f.rejectInside(packet, out, q)
 		if f.l.Level >= logrus.DebugLevel {
 			hostinfo.logger(f.l).
 				WithField("fwPacket", fwPacket).
@@ -81,63 +81,6 @@ func (f *Interface) consumeInsidePacket(packet []byte, fwPacket *firewall.Packet
 	}
 }
 
-// sendInsideMessage encrypts a firewall-approved inside packet into the
-// caller's batch slot for later sendmmsg flush. When hostinfo.remote is not
-// valid we fall through to the relay slow path via the unbatched sendNoMetrics
-// so relay behavior is unchanged.
-func (f *Interface) sendInsideMessage(hostinfo *HostInfo, p, nb []byte, batch *sendBatch, rejectBuf []byte, q int) {
-	ci := hostinfo.ConnectionState
-	if ci.eKey == nil {
-		return
-	}
-
-	if !hostinfo.remote.IsValid() {
-		// Slow path: relay fallback. Reuse rejectBuf as the ciphertext
-		// scratch; sendNoMetrics arranges header space for SendVia.
-		f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, p, nb, rejectBuf, q)
-		return
-	}
-
-	scratch := batch.Next()
-	if scratch == nil {
-		// Batch full: bypass batching and send this packet directly so we
-		// never drop traffic on over-subscribed iterations.
-		f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, p, nb, rejectBuf, q)
-		return
-	}
-
-	if noiseutil.EncryptLockNeeded {
-		ci.writeLock.Lock()
-	}
-	c := ci.messageCounter.Add(1)
-
-	out := header.Encode(scratch, header.Version, header.Message, 0, hostinfo.remoteIndexId, c)
-	f.connectionManager.Out(hostinfo)
-
-	if hostinfo.lastRebindCount != f.rebindCount {
-		//NOTE: there is an update hole if a tunnel isn't used and exactly 256 rebinds occur before the tunnel is
-		// finally used again. This tunnel would eventually be torn down and recreated if this action didn't help.
-		f.lightHouse.QueryServer(hostinfo.vpnAddrs[0])
-		hostinfo.lastRebindCount = f.rebindCount
-		if f.l.Level >= logrus.DebugLevel {
-			f.l.WithField("vpnAddrs", hostinfo.vpnAddrs).Debug("Lighthouse update triggered for punch due to rebind counter")
-		}
-	}
-
-	out, err := ci.eKey.EncryptDanger(out, out, p, c, nb)
-	if noiseutil.EncryptLockNeeded {
-		ci.writeLock.Unlock()
-	}
-	if err != nil {
-		hostinfo.logger(f.l).WithError(err).
-			WithField("udpAddr", hostinfo.remote).WithField("counter", c).
-			Error("Failed to encrypt outgoing packet")
-		return
-	}
-
-	batch.Commit(len(out), hostinfo.remote)
-}
-
 func (f *Interface) rejectInside(packet []byte, out []byte, q int) {
 	if !f.firewall.InSendReject {
 		return
@@ -148,7 +91,7 @@ func (f *Interface) rejectInside(packet []byte, out []byte, q int) {
 		return
 	}
 
-	_, err := f.readers[q].WriteReject(out)
+	_, err := f.readers[q].Write(out)
 	if err != nil {
 		f.l.WithError(err).Error("Failed to write to tun")
 	}

interface.go

@@ -4,8 +4,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"net/netip"
-	"sync"
+	"os"
 	"sync/atomic"
 	"time"
 
@@ -16,8 +17,6 @@ import (
 	"github.com/slackhq/nebula/firewall"
 	"github.com/slackhq/nebula/header"
 	"github.com/slackhq/nebula/overlay"
-	"github.com/slackhq/nebula/overlay/coalesce"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/udp"
 )
 
@@ -87,18 +86,7 @@ type Interface struct {
 	conntrackCacheTimeout time.Duration
 
 	writers []udp.Conn
-	readers []tio.Queue
+	readers []io.ReadWriteCloser
-	// tunCoalescers is one tcpCoalescer per tun queue, wrapping readers[i].
-	// decryptToTun sends plaintext into the coalescer; listenOut calls its
-	// Flush at the end of each UDP recvmmsg batch.
-	tunCoalescers []*coalesce.TCPCoalescer
-	wg            sync.WaitGroup
-
-	// fatalErr holds the first unexpected reader error that caused shutdown.
-	// nil means "no fatal error" (yet)
-	fatalErr atomic.Pointer[error]
-	// triggerShutdown is a function that will be run exactly once, when onFatal swaps something non-nil into fatalErr
-	triggerShutdown func()
 
 	metricHandshakes    metrics.Histogram
 	messageMetrics      *MessageMetrics
@@ -189,8 +177,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 		routines:              c.routines,
 		version:               c.version,
 		writers:               make([]udp.Conn, c.routines),
-		readers:               make([]tio.Queue, c.routines),
+		readers:               make([]io.ReadWriteCloser, c.routines),
-		tunCoalescers:         make([]*coalesce.TCPCoalescer, c.routines),
 		myVpnNetworks:         cs.myVpnNetworks,
 		myVpnNetworksTable:    cs.myVpnNetworksTable,
 		myVpnAddrs:            cs.myVpnAddrs,
@@ -222,7 +209,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 // activate creates the interface on the host. After the interface is created, any
 // other services that want to bind listeners to its IP may do so successfully. However,
 // the interface isn't going to process anything until run() is called.
-func (f *Interface) activate() error {
+func (f *Interface) activate() {
 	// actually turn on tun dev
 
 	addr, err := f.outside.LocalAddr()
@@ -245,61 +232,32 @@ func (f *Interface) activate() error {
 	metrics.GetOrRegisterGauge("routines", nil).Update(int64(f.routines))
 
 	// Prepare n tun queues
+	var reader io.ReadWriteCloser = f.inside
 	for i := 0; i < f.routines; i++ {
 		if i > 0 {
-			err = f.inside.NewMultiQueueReader()
+			reader, err = f.inside.NewMultiQueueReader()
 			if err != nil {
-				return err
+				f.l.Fatal(err)
 			}
 		}
-	}
+		f.readers[i] = reader
-	f.readers = f.inside.Readers()
-	for i := range f.readers {
-		f.tunCoalescers[i] = coalesce.NewTCPCoalescer(f.readers[i]) //todo don't always do this
 	}
 
-	f.wg.Add(1) // for us to wait on Close() to return
+	if err := f.inside.Activate(); err != nil {
-	if err = f.inside.Activate(); err != nil {
-		f.wg.Done()
 		f.inside.Close()
-		return err
+		f.l.Fatal(err)
 	}
-
-	return nil
 }
 
-func (f *Interface) run() (func() error, error) {
+func (f *Interface) run() {
 	// Launch n queues to read packets from udp
 	for i := 0; i < f.routines; i++ {
-		f.wg.Go(func() {
+		go f.listenOut(i)
-			f.listenOut(i)
-		})
 	}
 
 	// Launch n queues to read packets from tun dev
 	for i := 0; i < f.routines; i++ {
-		f.wg.Go(func() {
+		go f.listenIn(f.readers[i], i)
-			f.listenIn(f.readers[i], i)
-		})
-	}
-
-	return func() error {
-		f.wg.Wait()
-		if e := f.fatalErr.Load(); e != nil {
-			return *e
-		}
-		return nil
-	}, nil
-}
-
-// onFatal stores the first fatal reader error, and calls triggerShutdown if it was the first one
-func (f *Interface) onFatal(err error) {
-	swapped := f.fatalErr.CompareAndSwap(nil, &err)
-	if !swapped {
-		return
-	}
-	if f.triggerShutdown != nil {
-		f.triggerShutdown()
 	}
 }
 
@@ -313,75 +271,37 @@ func (f *Interface) listenOut(i int) {
 
 	ctCache := firewall.NewConntrackCacheTicker(f.conntrackCacheTimeout)
 	lhh := f.lightHouse.NewRequestHandler()
+	plaintext := make([]byte, udp.MTU)
 	h := &header.H{}
 	fwPacket := &firewall.Packet{}
 	nb := make([]byte, 12, 12)
 
-	// plaintexts is a ring of decrypt scratches, one per packet in a UDP
+	li.ListenOut(func(fromUdpAddr netip.AddrPort, payload []byte) {
-	// recvmmsg batch. The coalescer borrows payload slices from here and
+		f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintext[:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get(f.l))
-	// requires they stay valid until Flush — so we rotate each packet and
-	// reset only in the batch-end flush callback.
-	var plaintexts [][]byte
-	idx := 0
-	coalescer := f.tunCoalescers[i]
-	err := li.ListenOut(func(fromUdpAddr netip.AddrPort, payload []byte) {
-		if idx >= len(plaintexts) {
-			plaintexts = append(plaintexts, make([]byte, udp.MTU))
-		}
-		f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, plaintexts[idx][:0], payload, h, fwPacket, lhh, nb, i, ctCache.Get(f.l))
-		idx++
-	}, func() {
-		if err := coalescer.Flush(); err != nil {
-			f.l.WithError(err).Error("Failed to flush tun coalescer")
-		}
-		idx = 0
 	})
-
-	if err != nil && !f.closed.Load() {
-		f.l.WithError(err).Error("Error while reading inbound packet, closing")
-		f.onFatal(err)
-	}
-
-	f.l.Debugf("underlay reader %v is done", i)
 }
 
-func (f *Interface) listenIn(reader tio.Queue, i int) {
+func (f *Interface) listenIn(reader io.ReadWriteCloser, i int) {
-	rejectBuf := make([]byte, mtu)
+	packet := make([]byte, mtu)
-	batch := newSendBatch(sendBatchCap, udp.MTU+32)
+	out := make([]byte, mtu)
 	fwPacket := &firewall.Packet{}
 	nb := make([]byte, 12, 12)
 
 	conntrackCache := firewall.NewConntrackCacheTicker(f.conntrackCacheTimeout)
 
 	for {
-		pkts, err := reader.Read()
+		n, err := reader.Read(packet)
 		if err != nil {
-			if !f.closed.Load() {
+			if errors.Is(err, os.ErrClosed) && f.closed.Load() {
-				f.l.WithError(err).WithField("reader", i).Error("Error while reading outbound packet, closing")
+				return
-				f.onFatal(err)
 			}
-			break
+
+			f.l.WithError(err).Error("Error while reading outbound packet")
+			// This only seems to happen when something fatal happens to the fd, so exit.
+			os.Exit(2)
 		}
 
-		batch.Reset()
+		f.consumeInsidePacket(packet[:n], fwPacket, nb, out, i, conntrackCache.Get(f.l))
-		for _, pkt := range pkts {
-			if batch.Len() >= batch.Cap() {
-				f.flushBatch(batch, i)
-				batch.Reset()
-			}
-			f.consumeInsidePacket(pkt, fwPacket, nb, batch, rejectBuf, i, conntrackCache.Get(f.l))
-		}
-		if batch.Len() > 0 {
-			f.flushBatch(batch, i)
-		}
-	}
-
-	f.l.Debugf("overlay reader %v is done", i)
-}
-
-func (f *Interface) flushBatch(batch *sendBatch, q int) {
-	if err := f.writers[q].WriteBatch(batch.bufs, batch.dsts); err != nil {
-		f.l.WithError(err).WithField("writer", q).Error("Failed to write outgoing batch")
 	}
 }
 
@@ -557,23 +477,15 @@ func (f *Interface) GetCertState() *CertState {
 }
 
 func (f *Interface) Close() error {
-	var errs []error
 	f.closed.Store(true)
 
-	// Release the udp readers
+	for _, u := range f.writers {
-	for i, u := range f.writers {
 		err := u.Close()
 		if err != nil {
-			f.l.WithError(err).WithField("writer", i).Error("Error while closing udp socket")
+			f.l.WithError(err).Error("Error while closing udp socket")
-			errs = append(errs, err)
 		}
 	}
 
 	// Release the tun device (closing the tun also closes all readers)
-	closeErr := f.inside.Close()
+	return f.inside.Close()
-	if closeErr != nil {
-		errs = append(errs, closeErr)
-	}
-	f.wg.Done()
-	return errors.Join(errs...)
 }

main.go

@@ -3,10 +3,7 @@ package nebula
 import (
 	"context"
 	"fmt"
-	"log"
 	"net"
-	"net/http"
-	_ "net/http/pprof"
 	"net/netip"
 	"runtime/debug"
 	"strings"
@@ -52,11 +49,6 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 		l.Println(string(b))
 	}
 
-	//todo!!!
-	go func() {
-		log.Println(http.ListenAndServe("0.0.0.0:6060", nil))
-	}()
-
 	err := configLogger(l, c)
 	if err != nil {
 		return nil, util.ContextualizeIfNeeded("Failed to configure the logger", err)
@@ -296,16 +288,15 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 	}
 
 	return &Control{
-		state:                  StateReady,
+		ifce,
-		f:                      ifce,
+		l,
-		l:                      l,
+		ctx,
-		ctx:                    ctx,
+		cancel,
-		cancel:                 cancel,
+		sshStart,
-		sshStart:               sshStart,
+		statsStart,
-		statsStart:             statsStart,
+		dnsStart,
-		dnsStart:               dnsStart,
+		lightHouse.StartUpdateWorker,
-		lighthouseStart:        lightHouse.StartUpdateWorker,
+		connManager.Start,
-		connectionManagerStart: connManager.Start,
 	}, nil
 }
 

noise.go

@@ -15,14 +15,14 @@ type endianness interface {
 var noiseEndianness endianness = binary.BigEndian
 
 type NebulaCipherState struct {
-	c cipher.AEAD
+	c noise.Cipher
 	//k [32]byte
 	//n uint64
 }
 
 func NewNebulaCipherState(s *noise.CipherState) *NebulaCipherState {
-	x := s.Cipher()
+	return &NebulaCipherState{c: s.Cipher()}
-	return &NebulaCipherState{c: x.(cipher.AEAD)}
+
 }
 
 // EncryptDanger encrypts and authenticates a given payload.
@@ -46,7 +46,7 @@ func (s *NebulaCipherState) EncryptDanger(out, ad, plaintext []byte, n uint64, n
 		nb[2] = 0
 		nb[3] = 0
 		noiseEndianness.PutUint64(nb[4:], n)
-		out = s.c.Seal(out, nb, plaintext, ad)
+		out = s.c.(cipher.AEAD).Seal(out, nb, plaintext, ad)
 		//l.Debugf("Encryption: outlen: %d, nonce: %d, ad: %s, plainlen %d", len(out), n, ad, len(plaintext))
 		return out, nil
 	} else {
@@ -61,7 +61,7 @@ func (s *NebulaCipherState) DecryptDanger(out, ad, ciphertext []byte, n uint64,
 		nb[2] = 0
 		nb[3] = 0
 		noiseEndianness.PutUint64(nb[4:], n)
-		return s.c.Open(out, nb, ciphertext, ad)
+		return s.c.(cipher.AEAD).Open(out, nb, ciphertext, ad)
 	} else {
 		return []byte{}, nil
 	}
@@ -69,7 +69,7 @@ func (s *NebulaCipherState) DecryptDanger(out, ad, ciphertext []byte, n uint64,
 
 func (s *NebulaCipherState) Overhead() int {
 	if s != nil {
-		return s.c.Overhead()
+		return s.c.(cipher.AEAD).Overhead()
 	}
 	return 0
 }

outside.go

@@ -535,7 +535,7 @@ func (f *Interface) decryptToTun(hostinfo *HostInfo, messageCounter uint64, out
 	}
 
 	f.connectionManager.In(hostinfo)
-	err = f.tunCoalescers[q].Add(out)
+	_, err = f.readers[q].Write(out)
 	if err != nil {
 		f.l.WithError(err).Error("Failed to write to tun")
 	}

overlay/coalesce/tcp_coalesce.go

@@ -1,484 +0,0 @@
-package coalesce
-
-import (
-	"bytes"
-	"encoding/binary"
-	"io"
-
-	"github.com/slackhq/nebula/overlay/tio"
-)
-
-// ipProtoTCP is the IANA protocol number for TCP. Hardcoded instead of
-// reaching for golang.org/x/sys/unix — that package doesn't define the
-// constant on Windows, which would break cross-compiles even though this
-// file runs unchanged on every platform.
-const ipProtoTCP = 6
-
-// tcpCoalesceBufSize caps total bytes per superpacket. Mirrors the kernel's
-// sk_gso_max_size of ~64KiB; anything beyond this would be rejected anyway.
-const tcpCoalesceBufSize = 65535
-
-// tcpCoalesceMaxSegs caps how many segments we'll coalesce into a single
-// superpacket. Keeping this well below the kernel's TSO ceiling bounds
-// latency.
-const tcpCoalesceMaxSegs = 64
-
-// tcpCoalesceHdrCap is the scratch space we copy a seed's IP+TCP header
-// into. IPv6 (40) + TCP with full options (60) = 100 bytes.
-const tcpCoalesceHdrCap = 100
-
-// initialSlots is the starting capacity of the slot pool. One flow per
-// packet is the worst case so this matches a typical UDP recvmmsg batch.
-const initialSlots = 64
-
-// flowKey identifies a TCP flow by {src, dst, sport, dport, family}.
-// Comparable, so linear scans over the slot list stay tight.
-type flowKey struct {
-	src, dst     [16]byte
-	sport, dport uint16
-	isV6         bool
-}
-
-// coalesceSlot is one entry in the coalescer's ordered event queue. When
-// passthrough is true the slot holds a single borrowed packet that must be
-// emitted verbatim (non-TCP, non-admissible TCP, or oversize seed). When
-// passthrough is false the slot is an in-progress coalesced superpacket:
-// hdrBuf is a mutable copy of the seed's IP+TCP header (we patch total
-// length and pseudo-header partial at flush), and payIovs are *borrowed*
-// slices from the caller's plaintext buffers — no payload is ever copied.
-// The caller (listenOut) must keep those buffers alive until Flush.
-type coalesceSlot struct {
-	passthrough bool
-	rawPkt      []byte // borrowed when passthrough
-
-	fk       flowKey
-	hdrBuf   [tcpCoalesceHdrCap]byte
-	hdrLen   int
-	ipHdrLen int
-	isV6     bool
-	gsoSize  int
-	numSeg   int
-	totalPay int
-	nextSeq  uint32
-	// psh closes the chain: set when the last-accepted segment had PSH or
-	// was sub-gsoSize. No further appends after that.
-	psh     bool
-	payIovs [][]byte
-}
-
-// TCPCoalescer accumulates adjacent in-flow TCP data segments across
-// multiple concurrent flows and emits each flow's run as a single TSO
-// superpacket via tio.GSOWriter. All output — coalesced or not — is
-// deferred until Flush so arrival order is preserved on the wire. Owns
-// no locks; one coalescer per TUN write queue.
-type TCPCoalescer struct {
-	plainW io.Writer
-	gsoW   tio.GSOWriter // nil when the queue doesn't support TSO
-
-	// slots is the ordered event queue. Flush walks it once and emits each
-	// entry as either a WriteGSO (coalesced) or a plainW.Write (passthrough).
-	slots []*coalesceSlot
-	// openSlots maps a flow key to its most recent non-sealed slot, so new
-	// segments can extend an in-progress superpacket in O(1). Slots are
-	// removed from this map when they close (PSH or short-last-segment),
-	// when a non-admissible packet for that flow arrives, or in Flush.
-	openSlots map[flowKey]*coalesceSlot
-	pool      []*coalesceSlot // free list for reuse
-}
-
-func NewTCPCoalescer(w io.Writer) *TCPCoalescer {
-	c := &TCPCoalescer{
-		plainW:    w,
-		slots:     make([]*coalesceSlot, 0, initialSlots),
-		openSlots: make(map[flowKey]*coalesceSlot, initialSlots),
-		pool:      make([]*coalesceSlot, 0, initialSlots),
-	}
-	if gw, ok := w.(tio.GSOWriter); ok && gw.GSOSupported() {
-		c.gsoW = gw
-	}
-	return c
-}
-
-// parsedTCP holds the fields extracted from a single parse so later steps
-// (admission, slot lookup, canAppend) don't re-walk the header.
-type parsedTCP struct {
-	fk        flowKey
-	ipHdrLen  int
-	tcpHdrLen int
-	hdrLen    int
-	payLen    int
-	seq       uint32
-	flags     byte
-}
-
-// parseTCPBase extracts the flow key and IP/TCP offsets for any TCP packet,
-// regardless of whether it's admissible for coalescing. Returns ok=false
-// for non-TCP or malformed input. Accepts IPv4 (no options, no fragmentation)
-// and IPv6 (no extension headers).
-func parseTCPBase(pkt []byte) (parsedTCP, bool) {
-	var p parsedTCP
-	if len(pkt) < 20 {
-		return p, false
-	}
-	v := pkt[0] >> 4
-	switch v {
-	case 4:
-		ihl := int(pkt[0]&0x0f) * 4
-		if ihl != 20 {
-			return p, false
-		}
-		if pkt[9] != ipProtoTCP {
-			return p, false
-		}
-		// Reject actual fragmentation (MF or non-zero frag offset).
-		if binary.BigEndian.Uint16(pkt[6:8])&0x3fff != 0 {
-			return p, false
-		}
-		totalLen := int(binary.BigEndian.Uint16(pkt[2:4]))
-		if totalLen > len(pkt) || totalLen < ihl {
-			return p, false
-		}
-		p.ipHdrLen = 20
-		p.fk.isV6 = false
-		copy(p.fk.src[:4], pkt[12:16])
-		copy(p.fk.dst[:4], pkt[16:20])
-		pkt = pkt[:totalLen]
-	case 6:
-		if len(pkt) < 40 {
-			return p, false
-		}
-		if pkt[6] != ipProtoTCP {
-			return p, false
-		}
-		payloadLen := int(binary.BigEndian.Uint16(pkt[4:6]))
-		if 40+payloadLen > len(pkt) {
-			return p, false
-		}
-		p.ipHdrLen = 40
-		p.fk.isV6 = true
-		copy(p.fk.src[:], pkt[8:24])
-		copy(p.fk.dst[:], pkt[24:40])
-		pkt = pkt[:40+payloadLen]
-	default:
-		return p, false
-	}
-
-	if len(pkt) < p.ipHdrLen+20 {
-		return p, false
-	}
-	tcpOff := int(pkt[p.ipHdrLen+12]>>4) * 4
-	if tcpOff < 20 || tcpOff > 60 {
-		return p, false
-	}
-	if len(pkt) < p.ipHdrLen+tcpOff {
-		return p, false
-	}
-	p.tcpHdrLen = tcpOff
-	p.hdrLen = p.ipHdrLen + tcpOff
-	p.payLen = len(pkt) - p.hdrLen
-	p.seq = binary.BigEndian.Uint32(pkt[p.ipHdrLen+4 : p.ipHdrLen+8])
-	p.flags = pkt[p.ipHdrLen+13]
-	p.fk.sport = binary.BigEndian.Uint16(pkt[p.ipHdrLen : p.ipHdrLen+2])
-	p.fk.dport = binary.BigEndian.Uint16(pkt[p.ipHdrLen+2 : p.ipHdrLen+4])
-	return p, true
-}
-
-// coalesceable reports whether a parsed TCP segment is eligible for
-// coalescing. Accepts only ACK or ACK|PSH with a non-empty payload.
-func (p parsedTCP) coalesceable() bool {
-	const ack = 0x10
-	const psh = 0x08
-	if p.flags&^(ack|psh) != 0 || p.flags&ack == 0 {
-		return false
-	}
-	return p.payLen > 0
-}
-
-// Add borrows pkt. The caller must keep pkt valid until the next Flush,
-// whether or not the packet was coalesced — passthrough (non-admissible)
-// packets are queued and written at Flush time, not synchronously.
-func (c *TCPCoalescer) Add(pkt []byte) error {
-	if c.gsoW == nil {
-		c.addPassthrough(pkt)
-		return nil
-	}
-
-	info, ok := parseTCPBase(pkt)
-	if !ok {
-		// Non-TCP or malformed — can't possibly collide with an open flow.
-		c.addPassthrough(pkt)
-		return nil
-	}
-	if !info.coalesceable() {
-		// TCP but not admissible (SYN/FIN/RST/URG/CWR/ECE or zero-payload).
-		// Seal this flow's open slot so later in-flow packets don't extend
-		// it and accidentally reorder past this passthrough.
-		delete(c.openSlots, info.fk)
-		c.addPassthrough(pkt)
-		return nil
-	}
-
-	if open := c.openSlots[info.fk]; open != nil {
-		if c.canAppend(open, pkt, info) {
-			c.appendPayload(open, pkt, info)
-			if open.psh {
-				delete(c.openSlots, info.fk)
-			}
-			return nil
-		}
-		// Can't extend — seal it and fall through to seed a fresh slot.
-		delete(c.openSlots, info.fk)
-	}
-	c.seed(pkt, info)
-	return nil
-}
-
-// Flush emits every queued event in arrival order. Coalesced slots go out
-// via WriteGSO; passthrough slots go out via plainW.Write. Returns the
-// first error observed; keeps draining so one bad packet doesn't hold up
-// the rest. After Flush returns, borrowed payload slices may be recycled.
-func (c *TCPCoalescer) Flush() error {
-	var first error
-	for _, s := range c.slots {
-		var err error
-		if s.passthrough {
-			_, err = c.plainW.Write(s.rawPkt)
-		} else {
-			err = c.flushSlot(s)
-		}
-		if err != nil && first == nil {
-			first = err
-		}
-		c.release(s)
-	}
-	for i := range c.slots {
-		c.slots[i] = nil
-	}
-	c.slots = c.slots[:0]
-	for k := range c.openSlots {
-		delete(c.openSlots, k)
-	}
-	return first
-}
-
-func (c *TCPCoalescer) addPassthrough(pkt []byte) {
-	s := c.take()
-	s.passthrough = true
-	s.rawPkt = pkt
-	c.slots = append(c.slots, s)
-}
-
-func (c *TCPCoalescer) seed(pkt []byte, info parsedTCP) {
-	if info.hdrLen > tcpCoalesceHdrCap || info.hdrLen+info.payLen > tcpCoalesceBufSize {
-		// Pathological shape — can't fit our scratch, emit as-is.
-		c.addPassthrough(pkt)
-		return
-	}
-	s := c.take()
-	s.passthrough = false
-	s.rawPkt = nil
-	copy(s.hdrBuf[:], pkt[:info.hdrLen])
-	s.hdrLen = info.hdrLen
-	s.ipHdrLen = info.ipHdrLen
-	s.isV6 = info.fk.isV6
-	s.fk = info.fk
-	s.gsoSize = info.payLen
-	s.numSeg = 1
-	s.totalPay = info.payLen
-	s.nextSeq = info.seq + uint32(info.payLen)
-	s.psh = info.flags&0x08 != 0
-	s.payIovs = append(s.payIovs[:0], pkt[info.hdrLen:info.hdrLen+info.payLen])
-	c.slots = append(c.slots, s)
-	if !s.psh {
-		c.openSlots[info.fk] = s
-	}
-}
-
-// canAppend reports whether info's packet extends the slot's seed: same
-// header shape and stable contents, adjacent seq, not oversized, chain not
-// closed.
-func (c *TCPCoalescer) canAppend(s *coalesceSlot, pkt []byte, info parsedTCP) bool {
-	if s.psh {
-		return false
-	}
-	if info.hdrLen != s.hdrLen {
-		return false
-	}
-	if info.seq != s.nextSeq {
-		return false
-	}
-	if s.numSeg >= tcpCoalesceMaxSegs {
-		return false
-	}
-	if info.payLen > s.gsoSize {
-		return false
-	}
-	if s.hdrLen+s.totalPay+info.payLen > tcpCoalesceBufSize {
-		return false
-	}
-	if !headersMatch(s.hdrBuf[:s.hdrLen], pkt[:info.hdrLen], s.isV6, s.ipHdrLen) {
-		return false
-	}
-	return true
-}
-
-func (c *TCPCoalescer) appendPayload(s *coalesceSlot, pkt []byte, info parsedTCP) {
-	s.payIovs = append(s.payIovs, pkt[info.hdrLen:info.hdrLen+info.payLen])
-	s.numSeg++
-	s.totalPay += info.payLen
-	s.nextSeq = info.seq + uint32(info.payLen)
-	if info.payLen < s.gsoSize || info.flags&0x08 != 0 {
-		s.psh = true
-	}
-}
-
-func (c *TCPCoalescer) take() *coalesceSlot {
-	if n := len(c.pool); n > 0 {
-		s := c.pool[n-1]
-		c.pool[n-1] = nil
-		c.pool = c.pool[:n-1]
-		return s
-	}
-	return &coalesceSlot{}
-}
-
-func (c *TCPCoalescer) release(s *coalesceSlot) {
-	s.passthrough = false
-	s.rawPkt = nil
-	for i := range s.payIovs {
-		s.payIovs[i] = nil
-	}
-	s.payIovs = s.payIovs[:0]
-	s.numSeg = 0
-	s.totalPay = 0
-	s.psh = false
-	c.pool = append(c.pool, s)
-}
-
-// flushSlot patches the header and calls WriteGSO. Does not remove the
-// slot from c.slots.
-func (c *TCPCoalescer) flushSlot(s *coalesceSlot) error {
-	total := s.hdrLen + s.totalPay
-	l4Len := total - s.ipHdrLen
-	hdr := s.hdrBuf[:s.hdrLen]
-
-	if s.isV6 {
-		binary.BigEndian.PutUint16(hdr[4:6], uint16(l4Len))
-	} else {
-		binary.BigEndian.PutUint16(hdr[2:4], uint16(total))
-		hdr[10] = 0
-		hdr[11] = 0
-		binary.BigEndian.PutUint16(hdr[10:12], ipv4HdrChecksum(hdr[:s.ipHdrLen]))
-	}
-
-	var psum uint32
-	if s.isV6 {
-		psum = pseudoSumIPv6(hdr[8:24], hdr[24:40], ipProtoTCP, l4Len)
-	} else {
-		psum = pseudoSumIPv4(hdr[12:16], hdr[16:20], ipProtoTCP, l4Len)
-	}
-	tcsum := s.ipHdrLen + 16
-	binary.BigEndian.PutUint16(hdr[tcsum:tcsum+2], foldOnceNoInvert(psum))
-
-	return c.gsoW.WriteGSO(hdr, s.payIovs, uint16(s.gsoSize), s.isV6, uint16(s.ipHdrLen))
-}
-
-// headersMatch compares two IP+TCP header prefixes for byte-for-byte
-// equality on every field that must be identical across coalesced
-// segments. Size/IPID/IPCsum/seq/flags/tcpCsum are masked out.
-func headersMatch(a, b []byte, isV6 bool, ipHdrLen int) bool {
-	if len(a) != len(b) {
-		return false
-	}
-	if isV6 {
-		// IPv6: bytes [0:4] = version/TC/flow-label, [6:8] = next_hdr/hop,
-		// [8:40] = src+dst. Skip [4:6] payload length.
-		if !bytes.Equal(a[0:4], b[0:4]) {
-			return false
-		}
-		if !bytes.Equal(a[6:40], b[6:40]) {
-			return false
-		}
-	} else {
-		// IPv4: [0:2] version/IHL/TOS, [6:10] flags/fragoff/TTL/proto,
-		// [12:20] src+dst. Skip [2:4] total len, [4:6] id, [10:12] csum.
-		if !bytes.Equal(a[0:2], b[0:2]) {
-			return false
-		}
-		if !bytes.Equal(a[6:10], b[6:10]) {
-			return false
-		}
-		if !bytes.Equal(a[12:20], b[12:20]) {
-			return false
-		}
-	}
-	// TCP: compare [0:4] ports, [8:13] ack+dataoff, [14:16] window,
-	// [18:tcpHdrLen] options (incl. urgent).
-	tcp := ipHdrLen
-	if !bytes.Equal(a[tcp:tcp+4], b[tcp:tcp+4]) {
-		return false
-	}
-	if !bytes.Equal(a[tcp+8:tcp+13], b[tcp+8:tcp+13]) {
-		return false
-	}
-	if !bytes.Equal(a[tcp+14:tcp+16], b[tcp+14:tcp+16]) {
-		return false
-	}
-	if !bytes.Equal(a[tcp+18:], b[tcp+18:]) {
-		return false
-	}
-	return true
-}
-
-// ipv4HdrChecksum computes the IPv4 header checksum over hdr (which must
-// already have its checksum field zeroed) and returns the folded/inverted
-// 16-bit value to store.
-func ipv4HdrChecksum(hdr []byte) uint16 {
-	var sum uint32
-	for i := 0; i+1 < len(hdr); i += 2 {
-		sum += uint32(binary.BigEndian.Uint16(hdr[i : i+2]))
-	}
-	if len(hdr)%2 == 1 {
-		sum += uint32(hdr[len(hdr)-1]) << 8
-	}
-	for sum>>16 != 0 {
-		sum = (sum & 0xffff) + (sum >> 16)
-	}
-	return ^uint16(sum)
-}
-
-// pseudoSumIPv4 / pseudoSumIPv6 build the TCP pseudo-header partial sum
-// expected by the virtio NEEDS_CSUM kernel path: the 32-bit accumulator
-// before folding.
-func pseudoSumIPv4(src, dst []byte, proto byte, l4Len int) uint32 {
-	var sum uint32
-	sum += uint32(binary.BigEndian.Uint16(src[0:2]))
-	sum += uint32(binary.BigEndian.Uint16(src[2:4]))
-	sum += uint32(binary.BigEndian.Uint16(dst[0:2]))
-	sum += uint32(binary.BigEndian.Uint16(dst[2:4]))
-	sum += uint32(proto)
-	sum += uint32(l4Len)
-	return sum
-}
-
-func pseudoSumIPv6(src, dst []byte, proto byte, l4Len int) uint32 {
-	var sum uint32
-	for i := 0; i < 16; i += 2 {
-		sum += uint32(binary.BigEndian.Uint16(src[i : i+2]))
-		sum += uint32(binary.BigEndian.Uint16(dst[i : i+2]))
-	}
-	sum += uint32(l4Len >> 16)
-	sum += uint32(l4Len & 0xffff)
-	sum += uint32(proto)
-	return sum
-}
-
-// foldOnceNoInvert folds the 32-bit accumulator to 16 bits and returns it
-// unchanged (no one's complement). This is what virtio NEEDS_CSUM wants in
-// the L4 checksum field — the kernel will add the payload sum and invert.
-func foldOnceNoInvert(sum uint32) uint16 {
-	for sum>>16 != 0 {
-		sum = (sum & 0xffff) + (sum >> 16)
-	}
-	return uint16(sum)
-}

overlay/coalesce/tcp_coalesce_test.go

@@ -1,576 +0,0 @@
-package coalesce
-
-import (
-	"encoding/binary"
-	"testing"
-)
-
-// fakeTunWriter records plain Writes and WriteGSO calls without touching a
-// real TUN fd. WriteGSO preserves the split between hdr and borrowed pays
-// so tests can inspect each independently.
-type fakeTunWriter struct {
-	gsoEnabled bool
-	writes     [][]byte
-	gsoWrites  []fakeGSOWrite
-}
-
-type fakeGSOWrite struct {
-	hdr       []byte
-	pays      [][]byte
-	gsoSize   uint16
-	isV6      bool
-	csumStart uint16
-}
-
-// total returns hdrLen + sum of pay lens.
-func (g fakeGSOWrite) total() int {
-	n := len(g.hdr)
-	for _, p := range g.pays {
-		n += len(p)
-	}
-	return n
-}
-
-// payLen sums the pays.
-func (g fakeGSOWrite) payLen() int {
-	var n int
-	for _, p := range g.pays {
-		n += len(p)
-	}
-	return n
-}
-
-func (w *fakeTunWriter) Write(p []byte) (int, error) {
-	buf := make([]byte, len(p))
-	copy(buf, p)
-	w.writes = append(w.writes, buf)
-	return len(p), nil
-}
-
-func (w *fakeTunWriter) WriteGSO(hdr []byte, pays [][]byte, gsoSize uint16, isV6 bool, csumStart uint16) error {
-	hcopy := make([]byte, len(hdr))
-	copy(hcopy, hdr)
-	paysCopy := make([][]byte, len(pays))
-	for i, p := range pays {
-		pc := make([]byte, len(p))
-		copy(pc, p)
-		paysCopy[i] = pc
-	}
-	w.gsoWrites = append(w.gsoWrites, fakeGSOWrite{
-		hdr:       hcopy,
-		pays:      paysCopy,
-		gsoSize:   gsoSize,
-		isV6:      isV6,
-		csumStart: csumStart,
-	})
-	return nil
-}
-
-func (w *fakeTunWriter) GSOSupported() bool { return w.gsoEnabled }
-
-// buildTCPv4 constructs a minimal IPv4+TCP packet with the given payload,
-// seq, and flags. Assumes no IP options and a 20-byte TCP header.
-func buildTCPv4(seq uint32, flags byte, payload []byte) []byte {
-	return buildTCPv4Ports(1000, 2000, seq, flags, payload)
-}
-
-// buildTCPv4Ports is buildTCPv4 with caller-specified ports so tests can
-// build distinct flows.
-func buildTCPv4Ports(sport, dport uint16, seq uint32, flags byte, payload []byte) []byte {
-	const ipHdrLen = 20
-	const tcpHdrLen = 20
-	total := ipHdrLen + tcpHdrLen + len(payload)
-	pkt := make([]byte, total)
-
-	pkt[0] = 0x45
-	pkt[1] = 0x00
-	binary.BigEndian.PutUint16(pkt[2:4], uint16(total))
-	binary.BigEndian.PutUint16(pkt[4:6], 0)
-	binary.BigEndian.PutUint16(pkt[6:8], 0x4000)
-	pkt[8] = 64
-	pkt[9] = ipProtoTCP
-	copy(pkt[12:16], []byte{10, 0, 0, 1})
-	copy(pkt[16:20], []byte{10, 0, 0, 2})
-
-	binary.BigEndian.PutUint16(pkt[20:22], sport)
-	binary.BigEndian.PutUint16(pkt[22:24], dport)
-	binary.BigEndian.PutUint32(pkt[24:28], seq)
-	binary.BigEndian.PutUint32(pkt[28:32], 12345)
-	pkt[32] = 0x50
-	pkt[33] = flags
-	binary.BigEndian.PutUint16(pkt[34:36], 0xffff)
-
-	copy(pkt[40:], payload)
-	return pkt
-}
-
-const (
-	tcpAck    = 0x10
-	tcpPsh    = 0x08
-	tcpSyn    = 0x02
-	tcpFin    = 0x01
-	tcpAckPsh = tcpAck | tcpPsh
-)
-
-func TestCoalescerPassthroughWhenGSOUnavailable(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: false}
-	c := NewTCPCoalescer(w)
-	pkt := buildTCPv4(1000, tcpAck, []byte("hello"))
-	if err := c.Add(pkt); err != nil {
-		t.Fatal(err)
-	}
-	// No sync write — passthrough is deferred to Flush.
-	if len(w.writes) != 0 || len(w.gsoWrites) != 0 {
-		t.Fatalf("no Add-time writes: got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	if len(w.writes) != 1 || len(w.gsoWrites) != 0 {
-		t.Fatalf("want single plain write, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-}
-
-func TestCoalescerNonTCPPassthrough(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pkt := make([]byte, 28)
-	pkt[0] = 0x45
-	binary.BigEndian.PutUint16(pkt[2:4], 28)
-	pkt[9] = 1
-	copy(pkt[12:16], []byte{10, 0, 0, 1})
-	copy(pkt[16:20], []byte{10, 0, 0, 2})
-	if err := c.Add(pkt); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	if len(w.writes) != 1 || len(w.gsoWrites) != 0 {
-		t.Fatalf("ICMP should pass through unchanged")
-	}
-}
-
-func TestCoalescerSeedThenFlushAlone(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pkt := buildTCPv4(1000, tcpAck, make([]byte, 1000))
-	if err := c.Add(pkt); err != nil {
-		t.Fatal(err)
-	}
-	if len(w.writes) != 0 || len(w.gsoWrites) != 0 {
-		t.Fatalf("unexpected output before flush")
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	// Single-segment flush now goes through WriteGSO with GSO_NONE
-	// (virtio NEEDS_CSUM lets the kernel fill in the L4 csum).
-	if len(w.gsoWrites) != 1 || len(w.writes) != 0 {
-		t.Fatalf("single-seg flush: writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-	g := w.gsoWrites[0]
-	if g.total() != 40+1000 {
-		t.Errorf("super total=%d want %d", g.total(), 40+1000)
-	}
-	if g.payLen() != 1000 {
-		t.Errorf("payLen=%d want 1000", g.payLen())
-	}
-}
-
-func TestCoalescerCoalescesAdjacentACKs(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 1200)
-	if err := c.Add(buildTCPv4(1000, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4(2200, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4(3400, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	if len(w.gsoWrites) != 1 {
-		t.Fatalf("want 1 gso write, got %d (plain=%d)", len(w.gsoWrites), len(w.writes))
-	}
-	g := w.gsoWrites[0]
-	if g.gsoSize != 1200 {
-		t.Errorf("gsoSize=%d want 1200", g.gsoSize)
-	}
-	if len(g.hdr) != 40 {
-		t.Errorf("hdrLen=%d want 40", len(g.hdr))
-	}
-	if g.csumStart != 20 {
-		t.Errorf("csumStart=%d want 20", g.csumStart)
-	}
-	if len(g.pays) != 3 {
-		t.Errorf("pay count=%d want 3", len(g.pays))
-	}
-	if g.total() != 40+3*1200 {
-		t.Errorf("superpacket len=%d want %d", g.total(), 40+3*1200)
-	}
-	if tot := binary.BigEndian.Uint16(g.hdr[2:4]); int(tot) != g.total() {
-		t.Errorf("ip total_length=%d want %d", tot, g.total())
-	}
-}
-
-func TestCoalescerRejectsSeqGap(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 1200)
-	if err := c.Add(buildTCPv4(1000, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4(3000, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	// Each packet flushes as its own single-segment WriteGSO now.
-	if len(w.gsoWrites) != 2 || len(w.writes) != 0 {
-		t.Fatalf("seq gap: want 2 gso writes got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-}
-
-func TestCoalescerRejectsFlagMismatch(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 1200)
-	if err := c.Add(buildTCPv4(1000, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	// SYN|ACK is non-admissible. Must flush matching flow's slot (gso)
-	// and then plain-write the SYN packet itself.
-	syn := buildTCPv4(2200, tcpSyn|tcpAck, pay)
-	if err := c.Add(syn); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	if len(w.writes) != 1 || len(w.gsoWrites) != 1 {
-		t.Fatalf("flag mismatch: want 1 plain + 1 gso, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-}
-
-func TestCoalescerRejectsFIN(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	fin := buildTCPv4(1000, tcpAck|tcpFin, []byte("x"))
-	if err := c.Add(fin); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	// FIN isn't admissible — passthrough as plain, no slot, no gso.
-	if len(w.writes) != 1 || len(w.gsoWrites) != 0 {
-		t.Fatalf("FIN should be passthrough, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-}
-
-func TestCoalescerShortLastSegmentClosesChain(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	full := make([]byte, 1200)
-	half := make([]byte, 500)
-	if err := c.Add(buildTCPv4(1000, tcpAck, full)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4(2200, tcpAck, half)); err != nil {
-		t.Fatal(err)
-	}
-	// Chain now closed; next packet seeds a new slot on the same flow
-	// after flushing the old one.
-	if err := c.Add(buildTCPv4(2700, tcpAck, full)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	// Expect two gso writes: the first two packets coalesced, then the
-	// third flushed alone (single-seg via GSO_NONE).
-	if len(w.gsoWrites) != 2 {
-		t.Fatalf("want 2 gso writes got %d", len(w.gsoWrites))
-	}
-	if len(w.writes) != 0 {
-		t.Fatalf("want 0 plain writes got %d", len(w.writes))
-	}
-	if w.gsoWrites[0].gsoSize != 1200 {
-		t.Errorf("gsoSize=%d want 1200", w.gsoWrites[0].gsoSize)
-	}
-	if got, want := w.gsoWrites[0].total(), 40+1200+500; got != want {
-		t.Errorf("super len=%d want %d", got, want)
-	}
-}
-
-func TestCoalescerPSHFinalizesChain(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 1200)
-	if err := c.Add(buildTCPv4(1000, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4(2200, tcpAckPsh, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4(3400, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	// First two coalesce; the third seeds a fresh slot that flushes alone.
-	if len(w.gsoWrites) != 2 {
-		t.Fatalf("want 2 gso writes got %d", len(w.gsoWrites))
-	}
-	if len(w.writes) != 0 {
-		t.Fatalf("want 0 plain writes got %d", len(w.writes))
-	}
-}
-
-func TestCoalescerRejectsDifferentFlow(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 1200)
-	p1 := buildTCPv4(1000, tcpAck, pay)
-	p2 := buildTCPv4(2200, tcpAck, pay)
-	binary.BigEndian.PutUint16(p2[20:22], 9999)
-	if err := c.Add(p1); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(p2); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	// Two independent flows, each flushes its own single-segment WriteGSO.
-	if len(w.gsoWrites) != 2 || len(w.writes) != 0 {
-		t.Fatalf("diff flow: want 2 gso writes got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-}
-
-func TestCoalescerRejectsIPOptions(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 500)
-	pkt := buildTCPv4(1000, tcpAck, pay)
-	// Bump IHL to 6 to simulate 4 bytes of IP options. Don't actually add
-	// bytes — parser should bail before it matters.
-	pkt[0] = 0x46
-	if err := c.Add(pkt); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	// Non-admissible parse → passthrough as plain.
-	if len(w.writes) != 1 || len(w.gsoWrites) != 0 {
-		t.Fatalf("IP options should passthrough, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites))
-	}
-}
-
-func TestCoalescerCapBySegments(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 512)
-	seq := uint32(1000)
-	for i := 0; i < tcpCoalesceMaxSegs+5; i++ {
-		if err := c.Add(buildTCPv4(seq, tcpAck, pay)); err != nil {
-			t.Fatal(err)
-		}
-		seq += uint32(len(pay))
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	for _, g := range w.gsoWrites {
-		segs := len(g.pays)
-		if segs > tcpCoalesceMaxSegs {
-			t.Fatalf("super exceeded seg cap: %d > %d", segs, tcpCoalesceMaxSegs)
-		}
-	}
-}
-
-// TestCoalescerMultipleFlowsInSameBatch proves two interleaved bulk TCP
-// flows coalesce independently in a single Flush.
-func TestCoalescerMultipleFlowsInSameBatch(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 1200)
-
-	// Flow A: sport 1000. Flow B: sport 3000.
-	if err := c.Add(buildTCPv4Ports(1000, 2000, 100, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(3000, 2000, 500, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(1000, 2000, 1300, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(3000, 2000, 1700, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(1000, 2000, 2500, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(3000, 2000, 2900, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-
-	if len(w.gsoWrites) != 2 {
-		t.Fatalf("want 2 gso writes (one per flow), got %d", len(w.gsoWrites))
-	}
-	if len(w.writes) != 0 {
-		t.Fatalf("want no plain writes, got %d", len(w.writes))
-	}
-	// Each superpacket should carry 3 segments.
-	for i, g := range w.gsoWrites {
-		if len(g.pays) != 3 {
-			t.Errorf("gso[%d]: segs=%d want 3", i, len(g.pays))
-		}
-		if g.gsoSize != 1200 {
-			t.Errorf("gso[%d]: gsoSize=%d want 1200", i, g.gsoSize)
-		}
-	}
-	// Verify each superpacket carries the source port it was seeded with.
-	seenSports := map[uint16]bool{}
-	for _, g := range w.gsoWrites {
-		sp := binary.BigEndian.Uint16(g.hdr[20:22])
-		seenSports[sp] = true
-	}
-	if !seenSports[1000] || !seenSports[3000] {
-		t.Errorf("expected superpackets for sports 1000 and 3000, got %v", seenSports)
-	}
-}
-
-// TestCoalescerPreservesArrivalOrder confirms that with passthrough and
-// coalesced events both queued, Flush emits them in Add order rather than
-// writing passthrough packets synchronously.
-func TestCoalescerPreservesArrivalOrder(t *testing.T) {
-	w := &orderedFakeWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	// Sequence: coalesceable TCP, ICMP (passthrough), coalesceable TCP on
-	// a different flow. Expected emit order: gso(X), plain(ICMP), gso(Y).
-	pay := make([]byte, 1200)
-	if err := c.Add(buildTCPv4Ports(1000, 2000, 100, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	icmp := make([]byte, 28)
-	icmp[0] = 0x45
-	binary.BigEndian.PutUint16(icmp[2:4], 28)
-	icmp[9] = 1
-	copy(icmp[12:16], []byte{10, 0, 0, 1})
-	copy(icmp[16:20], []byte{10, 0, 0, 3})
-	if err := c.Add(icmp); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(3000, 2000, 500, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	// Nothing should have hit the writer synchronously.
-	if len(w.events) != 0 {
-		t.Fatalf("Add emitted events synchronously: %v", w.events)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-	if got, want := w.events, []string{"gso", "plain", "gso"}; !stringSliceEq(got, want) {
-		t.Fatalf("flush order=%v want %v", got, want)
-	}
-}
-
-// orderedFakeWriter records only the sequence of call types so tests can
-// assert arrival order without inspecting bytes.
-type orderedFakeWriter struct {
-	gsoEnabled bool
-	events     []string
-}
-
-func (w *orderedFakeWriter) Write(p []byte) (int, error) {
-	w.events = append(w.events, "plain")
-	return len(p), nil
-}
-
-func (w *orderedFakeWriter) WriteGSO(hdr []byte, pays [][]byte, gsoSize uint16, isV6 bool, csumStart uint16) error {
-	w.events = append(w.events, "gso")
-	return nil
-}
-
-func (w *orderedFakeWriter) GSOSupported() bool { return w.gsoEnabled }
-
-func stringSliceEq(a, b []string) bool {
-	if len(a) != len(b) {
-		return false
-	}
-	for i := range a {
-		if a[i] != b[i] {
-			return false
-		}
-	}
-	return true
-}
-
-// TestCoalescerInterleavedFlowsPreserveOrdering checks that a non-admissible
-// packet (SYN) mid-flow only flushes its own flow, not others.
-func TestCoalescerInterleavedFlowsPreserveOrdering(t *testing.T) {
-	w := &fakeTunWriter{gsoEnabled: true}
-	c := NewTCPCoalescer(w)
-	pay := make([]byte, 1200)
-
-	// Flow A two segments.
-	if err := c.Add(buildTCPv4Ports(1000, 2000, 100, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(1000, 2000, 1300, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	// Flow B two segments.
-	if err := c.Add(buildTCPv4Ports(3000, 2000, 500, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Add(buildTCPv4Ports(3000, 2000, 1700, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	// Flow A SYN (non-admissible) — must flush only flow A's slot.
-	syn := buildTCPv4Ports(1000, 2000, 9999, tcpSyn|tcpAck, pay)
-	if err := c.Add(syn); err != nil {
-		t.Fatal(err)
-	}
-	// Flow B continues — should still be coalesced with its seed.
-	if err := c.Add(buildTCPv4Ports(3000, 2000, 2900, tcpAck, pay)); err != nil {
-		t.Fatal(err)
-	}
-	if err := c.Flush(); err != nil {
-		t.Fatal(err)
-	}
-
-	// Expected:
-	//   - 1 gso for flow A (first 2 segments)
-	//   - 1 plain for flow A SYN
-	//   - 1 gso for flow B (3 segments)
-	if len(w.gsoWrites) != 2 {
-		t.Fatalf("want 2 gso writes, got %d", len(w.gsoWrites))
-	}
-	if len(w.writes) != 1 {
-		t.Fatalf("want 1 plain write (SYN), got %d", len(w.writes))
-	}
-	// Find the 3-segment gso (flow B) and the 2-segment gso (flow A).
-	var segCounts []int
-	for _, g := range w.gsoWrites {
-		segCounts = append(segCounts, len(g.pays))
-	}
-	if !(segCounts[0] == 2 && segCounts[1] == 3) && !(segCounts[0] == 3 && segCounts[1] == 2) {
-		t.Errorf("unexpected segment counts: %v (want 2 and 3)", segCounts)
-	}
-}

overlay/device.go

@@ -4,21 +4,15 @@ import (
 	"io"
 	"net/netip"
 
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
-// defaultBatchBufSize is the per-Queue scratch size for Read on backends
-// that don't do TSO segmentation. 65535 covers any single IP packet.
-const defaultBatchBufSize = 65535
-
 type Device interface {
-	io.Closer
+	io.ReadWriteCloser
 	Activate() error
 	Networks() []netip.Prefix
 	Name() string
 	RoutesFor(netip.Addr) routing.Gateways
-	SupportsMultiqueue() bool //todo remove?
+	SupportsMultiqueue() bool
-	NewMultiQueueReader() error
+	NewMultiQueueReader() (io.ReadWriteCloser, error)
-	Readers() []tio.Queue
 }

overlay/tio/container_gso_linux.go

@@ -1,70 +0,0 @@
-package tio
-
-import (
-	"encoding/binary"
-	"errors"
-	"fmt"
-
-	"golang.org/x/sys/unix"
-)
-
-type offloadContainer struct {
-	pq []*Offload
-	// pqi is exactly the same as pq, but stored as the interface type
-	pqi        []Queue
-	shutdownFd int
-}
-
-func NewOffloadContainer() (Container, 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 := &offloadContainer{
-		pq:         []*Offload{},
-		pqi:        []Queue{},
-		shutdownFd: shutdownFd,
-	}
-
-	return out, nil
-}
-
-func (c *offloadContainer) Queues() []Queue {
-	return c.pqi
-}
-
-func (c *offloadContainer) Add(fd int) error {
-	x, err := newOffload(fd, c.shutdownFd)
-	if err != nil {
-		return err
-	}
-	c.pq = append(c.pq, x)
-	c.pqi = append(c.pqi, x)
-
-	return nil
-}
-
-func (c *offloadContainer) wakeForShutdown() error {
-	var buf [8]byte
-	binary.NativeEndian.PutUint64(buf[:], 1)
-	_, err := unix.Write(c.shutdownFd, buf[:])
-	return err
-}
-
-func (c *offloadContainer) Close() error {
-	errs := []error{}
-
-	// Signal all readers blocked in poll to wake up and exit
-	if err := c.wakeForShutdown(); err != nil {
-		errs = append(errs, err)
-	}
-
-	for _, x := range c.pq {
-		if err := x.Close(); err != nil {
-			errs = append(errs, err)
-		}
-	}
-
-	return errors.Join(errs...)
-}

overlay/tio/container_poll_linux.go

@@ -1,69 +0,0 @@
-package tio
-
-import (
-	"encoding/binary"
-	"errors"
-	"fmt"
-
-	"golang.org/x/sys/unix"
-)
-
-type pollContainer struct {
-	pq []*Poll
-	// pqi is exactly the same as pq, but stored as the interface type
-	pqi        []Queue
-	shutdownFd int
-}
-
-func NewPollContainer() (Container, 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 := &pollContainer{
-		pq:         []*Poll{},
-		pqi:        []Queue{},
-		shutdownFd: shutdownFd,
-	}
-
-	return out, nil
-}
-
-func (c *pollContainer) Queues() []Queue {
-	return c.pqi
-}
-
-func (c *pollContainer) Add(fd int) error {
-	x, err := newPoll(fd, c.shutdownFd)
-	if err != nil {
-		return err
-	}
-	c.pq = append(c.pq, x)
-	c.pqi = append(c.pqi, x)
-
-	return nil
-}
-
-func (c *pollContainer) wakeForShutdown() error {
-	var buf [8]byte
-	binary.NativeEndian.PutUint64(buf[:], 1)
-	_, err := unix.Write(int(c.shutdownFd), buf[:])
-	return err
-}
-
-func (c *pollContainer) Close() error {
-	errs := []error{}
-
-	if err := c.wakeForShutdown(); err != nil {
-		errs = append(errs, err)
-	}
-
-	for _, x := range c.pq {
-		if err := x.Close(); err != nil {
-			errs = append(errs, err)
-		}
-	}
-
-	return errors.Join(errs...)
-}

overlay/tio/tio.go

@@ -1,63 +0,0 @@
-package tio
-
-import "io"
-
-// defaultBatchBufSize is the per-Queue scratch size for Read on backends
-// that don't do TSO segmentation. 65535 covers any single IP packet.
-const defaultBatchBufSize = 65535
-
-type Container interface {
-	Queues() []Queue
-	Add(fd int) error
-
-	io.Closer
-}
-
-// Queue is a readable/writable Poll queue. One Queue is driven by a single
-// read goroutine plus concurrent writers (see Write / WriteReject below).
-type Queue interface {
-	io.Closer
-
-	// Read returns one or more packets. The returned slices are borrowed
-	// from the Queue's internal buffer and are only valid until the next
-	// Read or Close on this Queue — callers must encrypt or copy each
-	// slice before the next call. Not safe for concurrent Reads; exactly
-	// one goroutine per Queue reads.
-	Read() ([][]byte, error)
-
-	// Write emits a single packet on the plaintext (outside→inside)
-	// delivery path. May run concurrently with WriteReject on the same
-	// Queue, but not with itself.
-	Write(p []byte) (int, error)
-
-	// WriteReject writes a single packet that originated from the inside
-	// path (reject replies or self-forward) using scratch state distinct
-	// from Write, so it can run concurrently with Write on the same Queue
-	// without a data race. On backends without a shared-scratch Write, a
-	// trivial delegation to Write is acceptable.
-	WriteReject(p []byte) (int, error)
-}
-
-// GSOWriter is implemented by Queues that can emit a TCP TSO superpacket
-// assembled from a header prefix plus one or more borrowed payload
-// fragments, in a single vectored write (writev with a leading
-// virtio_net_hdr). This lets the coalescer avoid copying payload bytes
-// between the caller's decrypt buffer and the TUN. Backends without GSO
-// support return false from GSOSupported and coalescing is skipped.
-//
-// hdr contains the IPv4/IPv6 + TCP header prefix (mutable — callers will
-// have filled in total length and pseudo-header partial). pays are
-// non-overlapping payload fragments whose concatenation is the full
-// superpacket payload; they are read-only from the writer's perspective
-// and must remain valid until the call returns. gsoSize is the MSS:
-// every segment except possibly the last is exactly that many bytes.
-// csumStart is the byte offset where the TCP header begins within hdr.
-//
-// # TODO fold into Queue
-//
-// hdr's TCP checksum field must already hold the pseudo-header partial
-// sum (single-fold, not inverted), per virtio NEEDS_CSUM semantics.
-type GSOWriter interface {
-	WriteGSO(hdr []byte, pays [][]byte, gsoSize uint16, isV6 bool, csumStart uint16) error
-	GSOSupported() bool
-}

overlay/tio/tio_gso_linux.go

@@ -1,434 +0,0 @@
-package tio
-
-import (
-	"fmt"
-	"io"
-	"os"
-	"sync/atomic"
-	"syscall"
-	"unsafe"
-
-	"golang.org/x/sys/unix"
-)
-
-// Space for segmented output. Worst case is many small segments, each paying
-// an IP+TCP header. Should be a multiple of 64KiB.
-// const tunSegBufSize = 0xffff * 8 TODO larger? config?
-const tunSegBufSize = 131072
-
-// tunSegBufCap is the total size we allocate for the per-reader segment
-// buffer. It is sized as one worst-case TSO superpacket (tunSegBufSize) plus
-// the same again as drain headroom so a Read wake can accumulate
-// additional packets after an initial big read without overflowing.
-const tunSegBufCap = tunSegBufSize * 2
-
-// tunDrainCap caps how many packets a single Read will accumulate via
-// the post-wake drain loop. Sized to soak up a burst of small ACKs while
-// bounding how much work a single caller holds before handing off.
-const tunDrainCap = 64 //256
-
-// gsoInitialPayIovs is the starting capacity (in payload fragments) of
-// Offload.gsoIovs. Sized to cover the default coalesce segment cap without
-// any reallocations.
-const gsoInitialPayIovs = 66
-
-// gsoWriteBufCap is the initial per-queue coalesce scratch capacity used by
-// WriteGSO to assemble [virtio_hdr || IP/TCP hdr || pays...] into a single
-// contiguous buffer so we can emit the superpacket via a single write()
-// instead of writev(). One worst-case TSO superpacket is bounded by the
-// virtio spec at 64KiB; 128KiB gives comfortable slack for the 10-byte
-// virtio header, the IP/TCP header, and any future size bumps. Grown on
-// demand if a superpacket exceeds this.
-const gsoWriteBufCap = tunSegBufSize
-
-// validVnetHdr is the 10-byte virtio_net_hdr we prepend to every non-GSO TUN
-// write. Only flag set is VIRTIO_NET_HDR_F_DATA_VALID, which marks the skb
-// CHECKSUM_UNNECESSARY so the receiving network stack skips L4 checksum
-// verification. All packets that reach the plain Write / WriteReject paths
-// already carry a valid L4 checksum (either supplied by a remote peer whose
-// ciphertext we AEAD-authenticated, or produced by finishChecksum during TSO
-// segmentation, or built locally by CreateRejectPacket), so trusting them is
-// safe.
-var validVnetHdr = [virtioNetHdrLen]byte{unix.VIRTIO_NET_HDR_F_DATA_VALID}
-
-// Offload wraps a TUN file descriptor with poll-based reads. The FD provided will be changed to non-blocking.
-// A shared eventfd allows Close to wake all readers blocked in poll.
-type Offload struct {
-	fd         int
-	shutdownFd int
-	readPoll   [2]unix.PollFd
-	writePoll  [2]unix.PollFd
-	closed     atomic.Bool
-	readBuf    []byte        // scratch for a single raw read (virtio hdr + superpacket)
-	segBuf     []byte        // backing store for segmented output
-	segOff     int           // cursor into segBuf for the current Read drain
-	pending    [][]byte      // segments returned from the most recent Read
-	writeIovs  [2]unix.Iovec // preallocated iovecs for Write (coalescer passthrough); iovs[0] is fixed to validVnetHdr
-	// rejectIovs is a second preallocated iovec scratch used exclusively by
-	// WriteReject (reject + self-forward from the inside path). It mirrors
-	// writeIovs but lets listenIn goroutines emit reject packets without
-	// racing with the listenOut coalescer that owns writeIovs.
-	rejectIovs [2]unix.Iovec
-
-	// gsoHdrBuf is a per-queue 10-byte scratch for the virtio_net_hdr emitted
-	// by WriteGSO. Separate from validVnetHdr so a concurrent non-GSO Write on
-	// another queue never observes a half-written header.
-	gsoHdrBuf [virtioNetHdrLen]byte
-	// gsoIovs is a legacy writev iovec scratch. No longer used by the
-	// WriteGSO path (which coalesces into gsoWriteBuf and uses a single
-	// write()) but retained for any other iovec-based path that may use it.
-	gsoIovs []unix.Iovec
-
-	// gsoWriteBuf is a per-queue scratch used by WriteGSO to coalesce the
-	// virtio_net_hdr + IP/TCP header + payload fragments into a single
-	// contiguous buffer, which is then written to the TUN fd with one
-	// write() syscall. This mirrors wireguard-go's approach and avoids
-	// triggering a kernel refcount use-after-free in skb_set_owner_w /
-	// sock_wfree observed on Linux 4.19 TUN when scatter-gather writev is
-	// combined with GSO-flagged virtio_net_hdr in the tun_chr_write_iter
-	// path. Grown on demand if a superpacket exceeds the initial cap.
-	gsoWriteBuf []byte
-}
-
-func newOffload(fd int, shutdownFd int) (*Offload, error) {
-	if err := unix.SetNonblock(fd, true); err != nil {
-		return nil, fmt.Errorf("failed to set tun fd non-blocking: %w", err)
-	}
-
-	out := &Offload{
-		fd:         fd,
-		shutdownFd: shutdownFd,
-		closed:     atomic.Bool{},
-		readBuf:    make([]byte, tunReadBufSize),
-		readPoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLIN},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
-		writePoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLOUT},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
-
-		segBuf:      make([]byte, tunSegBufCap),
-		gsoIovs:     make([]unix.Iovec, 2, 2+gsoInitialPayIovs),
-		gsoWriteBuf: make([]byte, 0, gsoWriteBufCap),
-	}
-
-	out.writeIovs[0].Base = &validVnetHdr[0]
-	out.writeIovs[0].SetLen(virtioNetHdrLen)
-	out.rejectIovs[0].Base = &validVnetHdr[0]
-	out.rejectIovs[0].SetLen(virtioNetHdrLen)
-	out.gsoIovs[0].Base = &out.gsoHdrBuf[0]
-	out.gsoIovs[0].SetLen(virtioNetHdrLen)
-
-	return out, nil
-}
-
-func (r *Offload) blockOnRead() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(r.readPoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	//always reset these!
-	tunEvents := r.readPoll[0].Revents
-	shutdownEvents := r.readPoll[1].Revents
-	r.readPoll[0].Revents = 0
-	r.readPoll[1].Revents = 0
-	//do the err check before trusting the potentially bogus bits we just got
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	} else if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
-}
-
-func (r *Offload) blockOnWrite() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(r.writePoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	//always reset these!
-	tunEvents := r.writePoll[0].Revents
-	shutdownEvents := r.writePoll[1].Revents
-	r.writePoll[0].Revents = 0
-	r.writePoll[1].Revents = 0
-	//do the err check before trusting the potentially bogus bits we just got
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	} else if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
-}
-
-func (r *Offload) readRaw(buf []byte) (int, error) {
-	for {
-		if n, err := unix.Read(r.fd, buf); err == nil {
-			return n, nil
-		} else if err == unix.EAGAIN {
-			if err = r.blockOnRead(); err != nil {
-				return 0, err
-			}
-			continue
-		} else if err == unix.EINTR {
-			continue
-		} else if err == unix.EBADF {
-			return 0, os.ErrClosed
-		} else {
-			return 0, err
-		}
-	}
-}
-
-// Read reads one or more superpackets from the tun and returns the
-// resulting packets. The first read blocks via poll; once the fd is known
-// readable we drain additional packets non-blocking until the kernel queue
-// is empty (EAGAIN), we've collected tunDrainCap packets, or we're out of
-// segBuf headroom. This amortizes the poll wake over bursts of small
-// packets (e.g. TCP ACKs). Slices point into the Offload's internal buffers
-// and are only valid until the next Read or Close on this Queue.
-func (r *Offload) Read() ([][]byte, error) {
-	r.pending = r.pending[:0]
-	r.segOff = 0
-
-	// Initial (blocking) read. Retry on decode errors so a single bad
-	// packet does not stall the reader.
-	for {
-		n, err := r.readRaw(r.readBuf)
-		if err != nil {
-			return nil, err
-		}
-		if err := r.decodeRead(n); err != nil {
-			// Drop and read again — a bad packet should not kill the reader.
-			continue
-		}
-		break
-	}
-
-	// Drain: non-blocking reads until the kernel queue is empty, the drain
-	// cap is reached, or segBuf no longer has room for another worst-case
-	// superpacket.
-	for len(r.pending) < tunDrainCap && tunSegBufCap-r.segOff >= tunSegBufSize {
-		n, err := unix.Read(r.fd, r.readBuf)
-		if err != nil {
-			// EAGAIN / EINTR / anything else: stop draining. We already
-			// have a valid batch from the first read.
-			break
-		}
-		if n <= 0 {
-			break
-		}
-		if err := r.decodeRead(n); err != nil {
-			// Drop this packet and stop the drain; we'd rather hand off
-			// what we have than keep spinning here.
-			break
-		}
-	}
-
-	return r.pending, nil
-}
-
-// decodeRead decodes the virtio header plus payload in r.readBuf[:n], appends
-// the segments to r.pending, and advances r.segOff by the total scratch used.
-// Caller must have already ensured r.vnetHdr is true.
-func (r *Offload) decodeRead(n int) error {
-	if n < virtioNetHdrLen {
-		return fmt.Errorf("short tun read: %d < %d", n, virtioNetHdrLen)
-	}
-	var hdr VirtioNetHdr
-	hdr.decode(r.readBuf[:virtioNetHdrLen])
-	before := len(r.pending)
-	if err := segmentInto(r.readBuf[virtioNetHdrLen:n], hdr, &r.pending, r.segBuf[r.segOff:]); err != nil {
-		return err
-	}
-	for k := before; k < len(r.pending); k++ {
-		r.segOff += len(r.pending[k])
-	}
-	return nil
-}
-
-func (r *Offload) Write(buf []byte) (int, error) {
-	return r.writeWithScratch(buf, &r.writeIovs)
-}
-
-// WriteReject emits a packet using a dedicated iovec scratch (rejectIovs)
-// distinct from the one used by the coalescer's Write path. This avoids a
-// data race between the inside (listenIn) goroutine emitting reject or
-// self-forward packets and the outside (listenOut) goroutine flushing TCP
-// coalescer passthroughs on the same Offload.
-func (r *Offload) WriteReject(buf []byte) (int, error) {
-	return r.writeWithScratch(buf, &r.rejectIovs)
-}
-
-func (r *Offload) writeWithScratch(buf []byte, iovs *[2]unix.Iovec) (int, error) {
-	if len(buf) == 0 {
-		return 0, nil
-	}
-	// Point the payload iovec at the caller's buffer. iovs[0] is pre-wired
-	// to validVnetHdr during Offload construction so we don't rebuild it here.
-	iovs[1].Base = &buf[0]
-	iovs[1].SetLen(len(buf))
-	return r.rawWrite(unsafe.Slice(&iovs[0], len(iovs)))
-}
-
-func (r *Offload) rawWrite(iovs []unix.Iovec) (int, error) {
-	for {
-		n, _, errno := syscall.Syscall(unix.SYS_WRITEV, uintptr(r.fd), uintptr(unsafe.Pointer(&iovs[0])), uintptr(len(iovs)))
-		if errno == 0 {
-			if int(n) < virtioNetHdrLen {
-				return 0, io.ErrShortWrite
-			}
-			return int(n) - virtioNetHdrLen, nil
-		}
-		if errno == unix.EAGAIN {
-			if err := r.blockOnWrite(); err != nil {
-				return 0, err
-			}
-			continue
-		}
-		if errno == unix.EINTR {
-			continue
-		}
-		if errno == unix.EBADF {
-			return 0, os.ErrClosed
-		}
-		return 0, errno
-	}
-}
-
-// rawWriteSingle writes buf to the TUN fd with a single write() syscall.
-// Unlike rawWrite (which uses writev), this avoids the kernel
-// scatter-gather path that triggers a use-after-free in
-// tun_chr_write_iter → sock_alloc_send_pskb → skb_set_owner_w on Linux
-// 4.19 TUN when the virtio_net_hdr requests TSO segmentation. The caller
-// is responsible for including the virtio_net_hdr prefix in buf.
-func (r *Offload) rawWriteSingle(buf []byte) (int, error) {
-	for {
-		n, err := unix.Write(r.fd, buf)
-		if err == nil {
-			if n < virtioNetHdrLen {
-				return 0, io.ErrShortWrite
-			}
-			return n - virtioNetHdrLen, nil
-		}
-		if err == unix.EAGAIN {
-			if werr := r.blockOnWrite(); werr != nil {
-				return 0, werr
-			}
-			continue
-		}
-		if err == unix.EINTR {
-			continue
-		}
-		if err == unix.EBADF {
-			return 0, os.ErrClosed
-		}
-		return 0, err
-	}
-}
-
-// GSOSupported reports whether this queue was opened with IFF_VNET_HDR and
-// can accept WriteGSO. When false, callers should fall back to per-segment
-// Write calls.
-func (r *Offload) GSOSupported() bool { return true }
-
-// WriteGSO emits a TCP TSO superpacket. hdr is the IPv4/IPv6 + TCP header
-// prefix (already finalized — total length, IP csum, and TCP pseudo-header
-// partial set by the caller). pays are payload fragments whose concatenation
-// forms the full coalesced payload. gsoSize is the MSS; every segment except
-// possibly the last is exactly gsoSize bytes. csumStart is the byte offset
-// where the TCP header begins within hdr.
-//
-// Implementation note: this path coalesces [virtio_hdr || hdr || pays...]
-// into a single contiguous scratch buffer (r.gsoWriteBuf) and emits it via
-// one write() syscall rather than writev() with a scatter-gather iovec.
-// The scatter-gather path triggered a kernel-side use-after-free on Linux
-// 4.19 TUN where tun_chr_write_iter → sock_alloc_send_pskb →
-// skb_set_owner_w could be invoked with a zero sk_wmem_alloc, crashing
-// the router. The single-write path mirrors wireguard-go's design (see
-// golang.zx2c4.com/wireguard/tun/tun_linux.go Write — it always coalesces
-// GRO-merged data into a single contiguous buffer before calling
-// tunFile.Write) and has no equivalent failure mode.
-func (r *Offload) WriteGSO(hdr []byte, pays [][]byte, gsoSize uint16, isV6 bool, csumStart uint16) error {
-	if len(hdr) == 0 || len(pays) == 0 {
-		return nil
-	}
-
-	// Build the virtio_net_hdr. When pays total to <= gsoSize the kernel
-	// would produce a single segment; keep NEEDS_CSUM semantics but skip
-	// the GSO type so the kernel doesn't spuriously mark this as TSO.
-	vhdr := VirtioNetHdr{
-		Flags:      unix.VIRTIO_NET_HDR_F_NEEDS_CSUM,
-		HdrLen:     uint16(len(hdr)),
-		GSOSize:    gsoSize,
-		CsumStart:  csumStart,
-		CsumOffset: 16, // TCP checksum field lives 16 bytes into the TCP header
-	}
-	var totalPay int
-	for _, p := range pays {
-		totalPay += len(p)
-	}
-	if totalPay > int(gsoSize) {
-		if isV6 {
-			vhdr.GSOType = unix.VIRTIO_NET_HDR_GSO_TCPV6
-		} else {
-			vhdr.GSOType = unix.VIRTIO_NET_HDR_GSO_TCPV4
-		}
-	} else {
-		vhdr.GSOType = unix.VIRTIO_NET_HDR_GSO_NONE
-		vhdr.GSOSize = 0
-	}
-	vhdr.encode(r.gsoHdrBuf[:])
-
-	// Coalesce [virtio_hdr || hdr || pays...] into a single contiguous
-	// buffer. This avoids the kernel scatter-gather write path entirely.
-	need := virtioNetHdrLen + len(hdr) + totalPay
-	if cap(r.gsoWriteBuf) < need {
-		// Grow geometrically to amortize reallocs.
-		newCap := cap(r.gsoWriteBuf) * 2
-		if newCap < need {
-			newCap = need
-		}
-		r.gsoWriteBuf = make([]byte, 0, newCap)
-	} else {
-		r.gsoWriteBuf = r.gsoWriteBuf[:0]
-	}
-	r.gsoWriteBuf = append(r.gsoWriteBuf, r.gsoHdrBuf[:]...)
-	r.gsoWriteBuf = append(r.gsoWriteBuf, hdr...)
-	for _, p := range pays {
-		r.gsoWriteBuf = append(r.gsoWriteBuf, p...)
-	}
-
-	_, err := r.rawWriteSingle(r.gsoWriteBuf)
-	return err
-}
-
-func (r *Offload) Close() error {
-	if r.closed.Swap(true) {
-		return nil
-	}
-
-	//shutdownFd is owned by the container, so we should not close it
-	var err error
-	if r.fd >= 0 {
-		err = unix.Close(r.fd)
-		r.fd = -1
-	}
-
-	return err
-}

overlay/tio/tio_poll_linux.go

@@ -1,205 +0,0 @@
-package tio
-
-import (
-	"fmt"
-	"os"
-	"sync/atomic"
-	"syscall"
-	"unsafe"
-
-	"golang.org/x/sys/unix"
-)
-
-// Maximum size we accept for a single read from a TUN with IFF_VNET_HDR. A
-// TSO superpacket can be up to 64KiB of payload plus a single L2/L3/L4 header
-// prefix plus the virtio header.
-const tunReadBufSize = 65535
-
-type Poll struct {
-	fd int
-
-	readPoll  [2]unix.PollFd
-	writePoll [2]unix.PollFd
-	closed    atomic.Bool
-
-	readBuf  []byte
-	batchRet [1][]byte
-}
-
-func newPoll(fd int, shutdownFd int) (*Poll, error) {
-	if err := unix.SetNonblock(fd, true); err != nil {
-		_ = unix.Close(fd)
-		return nil, fmt.Errorf("failed to set Poll device as nonblocking: %w", err)
-	}
-
-	out := &Poll{
-		fd:      fd,
-		readBuf: make([]byte, tunReadBufSize),
-		readPoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLIN},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
-		writePoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLOUT},
-			{Fd: int32(shutdownFd), Events: unix.POLLIN},
-		},
-	}
-	return out, nil
-}
-
-// blockOnRead waits until the Poll fd is readable or shutdown has been signaled.
-// Returns os.ErrClosed if Close was called.
-func (t *Poll) blockOnRead() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(t.readPoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	tunEvents := t.readPoll[0].Revents
-	shutdownEvents := t.readPoll[1].Revents
-	t.readPoll[0].Revents = 0
-	t.readPoll[1].Revents = 0
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	}
-	if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
-}
-
-func (t *Poll) blockOnWrite() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(t.writePoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	tunEvents := t.writePoll[0].Revents
-	shutdownEvents := t.writePoll[1].Revents
-	t.writePoll[0].Revents = 0
-	t.writePoll[1].Revents = 0
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	}
-	if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
-}
-
-func (t *Poll) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *Poll) readOne(to []byte) (int, error) {
-	// first 4 bytes is protocol family, in network byte order
-	var head [4]byte
-	iovecs := [2]syscall.Iovec{ //todo plat-specific
-		{&head[0], 4},
-		{&to[0], uint64(len(to))},
-	}
-	for {
-		n, _, errno := syscall.Syscall(syscall.SYS_READV, uintptr(t.fd), uintptr(unsafe.Pointer(&iovecs[0])), 2)
-		if errno == 0 {
-			bytesRead := int(n)
-			if bytesRead < 4 {
-				return 0, nil
-			}
-			return bytesRead - 4, nil
-		}
-		switch errno {
-		case unix.EAGAIN:
-			if err := t.blockOnRead(); err != nil {
-				return 0, err
-			}
-		case unix.EINTR:
-			// retry
-		case unix.EBADF:
-			return 0, os.ErrClosed
-		default:
-			return 0, errno
-		}
-	}
-}
-
-// Write is only valid for single threaded use
-func (t *Poll) Write(from []byte) (int, error) {
-	if len(from) <= 1 {
-		return 0, syscall.EIO
-	}
-
-	ipVer := from[0] >> 4
-	var head [4]byte
-	// first 4 bytes is protocol family, in network byte order
-	switch ipVer {
-	case 4:
-		head[3] = syscall.AF_INET
-	case 6:
-		head[3] = syscall.AF_INET6
-	default:
-		return 0, fmt.Errorf("unable to determine IP version from packet")
-	}
-
-	iovecs := [2]syscall.Iovec{ //todo plat specific
-		{&head[0], 4},
-		{&from[0], uint64(len(from))},
-	}
-	for {
-		n, _, errno := syscall.Syscall(syscall.SYS_WRITEV, uintptr(t.fd), uintptr(unsafe.Pointer(&iovecs[0])), 2)
-		if errno == 0 {
-			return int(n) - 4, nil
-		}
-		switch errno {
-		case unix.EAGAIN:
-			if err := t.blockOnWrite(); err != nil {
-				return 0, err
-			}
-		case unix.EINTR:
-			// retry
-		case unix.EBADF:
-			return 0, os.ErrClosed
-		default:
-			return 0, errno
-		}
-	}
-}
-
-func (t *Poll) Close() error {
-	if t.closed.Swap(true) {
-		return nil
-	}
-
-	//shutdownFd is owned by the container, so we should not close it
-
-	var err error
-	if t.fd >= 0 {
-		err = unix.Close(t.fd)
-		t.fd = -1
-	}
-
-	return err
-}
-
-func (t *Poll) WriteReject(p []byte) (int, error) {
-	return t.Write(p)
-}

overlay/tio/tun_file_linux_test.go

@@ -1,86 +0,0 @@
-//go:build linux && !android && !e2e_testing
-// +build linux,!android,!e2e_testing
-
-package tio
-
-import (
-	"errors"
-	"os"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-	"golang.org/x/sys/unix"
-)
-
-// newReadPipe returns a read fd. The matching write fd is registered for cleanup.
-// The caller takes ownership of the read fd (pass it to newOffload / newFriend).
-func newReadPipe(t *testing.T) int {
-	t.Helper()
-	var fds [2]int
-	if err := unix.Pipe2(fds[:], unix.O_CLOEXEC); err != nil {
-		t.Fatalf("pipe2: %v", err)
-	}
-	t.Cleanup(func() { _ = unix.Close(fds[1]) })
-	return fds[0]
-}
-
-func TestOffload_WakeForShutdown_WakesFriends(t *testing.T) {
-	pipe1 := newReadPipe(t)
-	pipe2 := newReadPipe(t)
-	parent, err := NewOffloadContainer()
-	if err != nil {
-		t.Fatalf("newOffload: %v", err)
-	}
-	require.NoError(t, parent.Add(pipe1))
-	require.NoError(t, parent.Add(pipe2))
-	t.Cleanup(func() {
-		_ = unix.Close(pipe1)
-		_ = unix.Close(pipe2)
-	})
-
-	readers := parent.Queues()
-	errs := make([]error, len(readers))
-	var wg sync.WaitGroup
-	for i, r := range readers {
-		wg.Add(1)
-		go func(i int, r Queue) {
-			defer wg.Done()
-			_, errs[i] = r.Read()
-		}(i, r)
-	}
-
-	time.Sleep(50 * time.Millisecond)
-
-	if err := parent.Close(); err != nil {
-		t.Fatalf("Close: %v", err)
-	}
-
-	done := make(chan struct{})
-	go func() { wg.Wait(); close(done) }()
-	select {
-	case <-done:
-	case <-time.After(2 * time.Second):
-		t.Fatal("readers did not wake")
-	}
-
-	for i, err := range errs {
-		if !errors.Is(err, os.ErrClosed) {
-			t.Errorf("reader %d: expected os.ErrClosed, got %v", i, err)
-		}
-	}
-}
-
-func TestTunFile_Close_Idempotent(t *testing.T) {
-	tf, err := newOffload(newReadPipe(t), 1)
-	if err != nil {
-		t.Fatalf("newOffload: %v", err)
-	}
-	if err := tf.Close(); err != nil {
-		t.Fatalf("first Close: %v", err)
-	}
-	if err := tf.Close(); err != nil {
-		t.Fatalf("second Close should be a no-op, got %v", err)
-	}
-}

overlay/tio/tun_linux_offload.go

@@ -1,281 +0,0 @@
-//go:build linux && !android && !e2e_testing
-// +build linux,!android,!e2e_testing
-
-package tio
-
-import (
-	"encoding/binary"
-	"fmt"
-
-	"golang.org/x/sys/unix"
-	"gvisor.dev/gvisor/pkg/tcpip/checksum"
-)
-
-// Protocol header size bounds used to validate / cap kernel-supplied offsets.
-const (
-	ipv4HeaderMinLen = 20 // IHL=5, no options
-	ipv4HeaderMaxLen = 60 // IHL=15, max options
-	ipv6FixedLen     = 40 // IPv6 base header; extensions would extend this
-	tcpHeaderMinLen  = 20 // data-offset=5, no options
-	tcpHeaderMaxLen  = 60 // data-offset=15, max options
-)
-
-// Byte offsets inside an IPv4 header.
-const (
-	ipv4TotalLenOff = 2
-	ipv4IDOff       = 4
-	ipv4ChecksumOff = 10
-	ipv4SrcOff      = 12
-	ipv4AddrsEnd    = 20 // end of dst address (ipv4SrcOff + 2*4)
-)
-
-// Byte offsets inside an IPv6 header.
-const (
-	ipv6PayloadLenOff = 4
-	ipv6SrcOff        = 8
-	ipv6AddrsEnd      = 40 // end of dst address (ipv6SrcOff + 2*16)
-)
-
-// Byte offsets inside a TCP header (relative to its start, i.e. csumStart).
-const (
-	tcpSeqOff      = 4
-	tcpDataOffOff  = 12 // upper nibble is header len in 32-bit words
-	tcpFlagsOff    = 13
-	tcpChecksumOff = 16
-)
-
-// tcpFinPshMask is cleared on every segment except the last of a TSO burst.
-const tcpFinPshMask = 0x09 // FIN(0x01) | PSH(0x08)
-
-// segmentInto splits a TUN-side packet described by hdr into one or more
-// IP packets, each appended to *out as a slice of scratch. scratch must be
-// sized to hold every segment (including replicated headers).
-func segmentInto(pkt []byte, hdr VirtioNetHdr, out *[][]byte, scratch []byte) error {
-	// When RSC_INFO is set the csum_start/csum_offset fields are repurposed to
-	// carry coalescing info rather than checksum offsets. A TUN writing via
-	// IFF_VNET_HDR should never emit this, but if it did we would silently
-	// miscompute the segment checksums — refuse the packet instead.
-	if hdr.Flags&unix.VIRTIO_NET_HDR_F_RSC_INFO != 0 {
-		return fmt.Errorf("virtio RSC_INFO flag not supported on TUN reads")
-	}
-
-	switch hdr.GSOType {
-	case unix.VIRTIO_NET_HDR_GSO_NONE:
-		if len(pkt) > len(scratch) {
-			return fmt.Errorf("packet larger than segment buffer: %d > %d", len(pkt), len(scratch))
-		}
-		copy(scratch, pkt)
-		seg := scratch[:len(pkt)]
-		if hdr.Flags&unix.VIRTIO_NET_HDR_F_NEEDS_CSUM != 0 {
-			if err := finishChecksum(seg, hdr); err != nil {
-				return err
-			}
-		}
-		*out = append(*out, seg)
-		return nil
-
-	case unix.VIRTIO_NET_HDR_GSO_TCPV4, unix.VIRTIO_NET_HDR_GSO_TCPV6:
-		return segmentTCP(pkt, hdr, out, scratch)
-
-	default:
-		return fmt.Errorf("unsupported virtio gso type: %d", hdr.GSOType)
-	}
-}
-
-// finishChecksum computes the L4 checksum for a non-GSO packet that the kernel
-// handed us with NEEDS_CSUM set. csum_start / csum_offset point at the 16-bit
-// checksum field; we zero it, fold a full sum (the field was pre-loaded with
-// the pseudo-header partial sum by the kernel), and store the result.
-func finishChecksum(seg []byte, hdr VirtioNetHdr) error {
-	cs := int(hdr.CsumStart)
-	co := int(hdr.CsumOffset)
-	if cs+co+2 > len(seg) {
-		return fmt.Errorf("csum offsets out of range: start=%d offset=%d len=%d", cs, co, len(seg))
-	}
-	// The kernel stores a partial pseudo-header sum at [cs+co:]; sum over the
-	// L4 region starting at cs, folding the prior partial in as the seed.
-	partial := binary.BigEndian.Uint16(seg[cs+co : cs+co+2])
-	seg[cs+co] = 0
-	seg[cs+co+1] = 0
-	binary.BigEndian.PutUint16(seg[cs+co:cs+co+2], ^checksum.Checksum(seg[cs:], partial))
-	return nil
-}
-
-// segmentTCP software-segments a TSO superpacket into one IP packet per MSS
-// chunk. The caller guarantees hdr.GSOType is TCPV4 or TCPV6.
-//
-// Hot-path shape: the per-segment loop only sums the payload chunk. The TCP
-// header, the IPv4 header, and the pseudo-header src/dst/proto contributions
-// are each summed once up front — every segment reuses those three pre-folded
-// uint32 values and combines them with small per-segment deltas (seq, flags,
-// tcpLen, ip_id, total_len) that are cheap to fold in.
-func segmentTCP(pkt []byte, hdr VirtioNetHdr, out *[][]byte, scratch []byte) error {
-	if hdr.GSOSize == 0 {
-		return fmt.Errorf("gso_size is zero")
-	}
-	if hdr.CsumStart == 0 {
-		return fmt.Errorf("csum_start is zero")
-	}
-
-	isV4 := hdr.GSOType == unix.VIRTIO_NET_HDR_GSO_TCPV4
-	csumStart := int(hdr.CsumStart)
-
-	if isV4 && csumStart < ipv4HeaderMinLen {
-		return fmt.Errorf("csum_start %d too small for IPv4", csumStart)
-	}
-	if !isV4 && csumStart < ipv6FixedLen {
-		return fmt.Errorf("csum_start %d too small for IPv6", csumStart)
-	}
-
-	// Don't trust hdr.HdrLen from the kernel: on some paths it can be set
-	// to the full length of the first packet rather than the true L3+L4 header length.
-	// Instead, read the TCP data-offset field from the packet itself and derive
-	// headerLen = csum_start + tcpHdrLen. Matches wireguard-go's approach.
-	if csumStart+tcpFlagsOff+1 > len(pkt) {
-		return fmt.Errorf("packet too short for tcp header at csum_start=%d (pkt %d)", csumStart, len(pkt))
-	}
-	tcpHdrLen := int(pkt[csumStart+tcpDataOffOff]>>4) * 4
-	if tcpHdrLen < tcpHeaderMinLen || tcpHdrLen > tcpHeaderMaxLen {
-		return fmt.Errorf("tcp data-offset out of range: %d", tcpHdrLen)
-	}
-	headerLen := csumStart + tcpHdrLen
-	if headerLen > len(pkt) {
-		return fmt.Errorf("derived hdr_len %d > pkt %d", headerLen, len(pkt))
-	}
-
-	payload := pkt[headerLen:]
-	payLen := len(payload)
-	gso := int(hdr.GSOSize)
-	numSeg := (payLen + gso - 1) / gso
-	if numSeg == 0 {
-		numSeg = 1
-	}
-
-	need := numSeg*headerLen + payLen
-	if need > len(scratch) {
-		return fmt.Errorf("scratch too small for %d segments: need %d have %d", numSeg, need, len(scratch))
-	}
-
-	origSeq := binary.BigEndian.Uint32(pkt[csumStart+tcpSeqOff : csumStart+tcpSeqOff+4])
-	origFlags := pkt[csumStart+tcpFlagsOff]
-
-	// Precompute the TCP header sum with seq/flags/csum zeroed. Copy onto
-	// the stack, zero the per-segment-varying fields, sum once.
-	var tmp [tcpHeaderMaxLen]byte
-	copy(tmp[:tcpHdrLen], pkt[csumStart:headerLen])
-	tmp[tcpSeqOff], tmp[tcpSeqOff+1], tmp[tcpSeqOff+2], tmp[tcpSeqOff+3] = 0, 0, 0, 0
-	tmp[tcpFlagsOff] = 0
-	tmp[tcpChecksumOff], tmp[tcpChecksumOff+1] = 0, 0
-	baseTcpHdrSum := uint32(checksum.Checksum(tmp[:tcpHdrLen], 0))
-
-	// Pseudo-header src+dst+proto contribution (tcpLen varies per segment).
-	var baseProtoSum uint32
-	if isV4 {
-		baseProtoSum = uint32(checksum.Checksum(pkt[ipv4SrcOff:ipv4AddrsEnd], 0))
-	} else {
-		baseProtoSum = uint32(checksum.Checksum(pkt[ipv6SrcOff:ipv6AddrsEnd], 0))
-	}
-	baseProtoSum += uint32(unix.IPPROTO_TCP)
-
-	// Precompute IPv4 header sum with total_len/id/csum zeroed.
-	var origIPID uint16
-	var ihl int
-	var baseIPHdrSum uint32
-	if isV4 {
-		origIPID = binary.BigEndian.Uint16(pkt[ipv4IDOff : ipv4IDOff+2])
-		ihl = int(pkt[0]&0x0f) * 4
-		if ihl < ipv4HeaderMinLen || ihl > csumStart {
-			return fmt.Errorf("bad IPv4 IHL: %d", ihl)
-		}
-		var ipTmp [ipv4HeaderMaxLen]byte
-		copy(ipTmp[:ihl], pkt[:ihl])
-		ipTmp[ipv4TotalLenOff], ipTmp[ipv4TotalLenOff+1] = 0, 0
-		ipTmp[ipv4IDOff], ipTmp[ipv4IDOff+1] = 0, 0
-		ipTmp[ipv4ChecksumOff], ipTmp[ipv4ChecksumOff+1] = 0, 0
-		baseIPHdrSum = uint32(checksum.Checksum(ipTmp[:ihl], 0))
-	}
-
-	off := 0
-	for i := 0; i < numSeg; i++ {
-		segStart := i * gso
-		segEnd := segStart + gso
-		if segEnd > payLen {
-			segEnd = payLen
-		}
-		segPayLen := segEnd - segStart
-
-		copy(scratch[off:], pkt[:headerLen])
-		copy(scratch[off+headerLen:], payload[segStart:segEnd])
-		seg := scratch[off : off+headerLen+segPayLen]
-		off += headerLen + segPayLen
-
-		segSeq := origSeq + uint32(segStart)
-		segFlags := origFlags
-		if i != numSeg-1 {
-			segFlags = origFlags &^ tcpFinPshMask
-		}
-		totalLen := headerLen + segPayLen
-
-		// Patch IP header and write the v4 header checksum from the precomputed base.
-		if isV4 {
-			segID := origIPID + uint16(i)
-			binary.BigEndian.PutUint16(seg[ipv4TotalLenOff:ipv4TotalLenOff+2], uint16(totalLen))
-			binary.BigEndian.PutUint16(seg[ipv4IDOff:ipv4IDOff+2], segID)
-			ipSum := baseIPHdrSum + uint32(totalLen) + uint32(segID)
-			binary.BigEndian.PutUint16(seg[ipv4ChecksumOff:ipv4ChecksumOff+2], foldComplement(ipSum))
-		} else {
-			// IPv6 payload length excludes the fixed header but includes any
-			// extension headers between [ipv6FixedLen:csumStart].
-			binary.BigEndian.PutUint16(seg[ipv6PayloadLenOff:ipv6PayloadLenOff+2], uint16(headerLen-ipv6FixedLen+segPayLen))
-		}
-
-		// Patch TCP header.
-		binary.BigEndian.PutUint32(seg[csumStart+tcpSeqOff:csumStart+tcpSeqOff+4], segSeq)
-		seg[csumStart+tcpFlagsOff] = segFlags
-		// (csum is written below; its prior contents in `seg` don't affect the
-		// computation since we never sum over the segment's own header.)
-
-		tcpLen := tcpHdrLen + segPayLen
-		paySum := uint32(checksum.Checksum(payload[segStart:segEnd], 0))
-
-		// Combine pre-folded uint32s into a wider accumulator, then fold. Using
-		// uint64 guards against overflow when segSeq's high bits set.
-		wide := uint64(baseTcpHdrSum) + uint64(paySum) + uint64(baseProtoSum)
-		wide += uint64(segSeq) + uint64(segFlags) + uint64(tcpLen)
-		wide = (wide & 0xffffffff) + (wide >> 32)
-		wide = (wide & 0xffffffff) + (wide >> 32)
-		binary.BigEndian.PutUint16(seg[csumStart+tcpChecksumOff:csumStart+tcpChecksumOff+2], foldComplement(uint32(wide)))
-
-		*out = append(*out, seg)
-	}
-
-	return nil
-}
-
-// foldComplement folds a 32-bit one's-complement partial sum to 16 bits and
-// complements it, yielding the on-wire Internet checksum value.
-func foldComplement(sum uint32) uint16 {
-	sum = (sum & 0xffff) + (sum >> 16)
-	sum = (sum & 0xffff) + (sum >> 16)
-	return ^uint16(sum)
-}
-
-// pseudoHeaderIPv4 returns the folded pseudo-header sum used to verify a TCP
-// segment's checksum in tests. src/dst are 4 bytes each.
-func pseudoHeaderIPv4(src, dst []byte, proto byte, tcpLen int) uint16 {
-	s := uint32(checksum.Checksum(src, 0)) + uint32(checksum.Checksum(dst, 0))
-	s += uint32(proto) + uint32(tcpLen)
-	s = (s & 0xffff) + (s >> 16)
-	s = (s & 0xffff) + (s >> 16)
-	return uint16(s)
-}
-
-// pseudoHeaderIPv6 returns the folded pseudo-header sum used to verify a TCP
-// segment's checksum in tests. src/dst are 16 bytes each.
-func pseudoHeaderIPv6(src, dst []byte, proto byte, tcpLen int) uint16 {
-	s := uint32(checksum.Checksum(src, 0)) + uint32(checksum.Checksum(dst, 0))
-	s += uint32(tcpLen>>16) + uint32(tcpLen&0xffff) + uint32(proto)
-	s = (s & 0xffff) + (s >> 16)
-	s = (s & 0xffff) + (s >> 16)
-	return uint16(s)
-}

overlay/tio/tun_linux_offload_test.go

@@ -1,330 +0,0 @@
-//go:build linux && !android && !e2e_testing
-// +build linux,!android,!e2e_testing
-
-package tio
-
-import (
-	"encoding/binary"
-	"os"
-	"testing"
-
-	"golang.org/x/sys/unix"
-	"gvisor.dev/gvisor/pkg/tcpip/checksum"
-)
-
-// verifyChecksum confirms that the one's-complement sum across `b`, seeded
-// with a folded pseudo-header sum, equals all-ones (valid).
-func verifyChecksum(b []byte, pseudo uint16) bool {
-	return checksum.Checksum(b, pseudo) == 0xffff
-}
-
-// buildTSOv4 builds a synthetic IPv4/TCP TSO superpacket with a payload of
-// `payLen` bytes split at `mss`.
-func buildTSOv4(t *testing.T, payLen, mss int) ([]byte, VirtioNetHdr) {
-	t.Helper()
-	const ipLen = 20
-	const tcpLen = 20
-	pkt := make([]byte, ipLen+tcpLen+payLen)
-
-	// IPv4 header
-	pkt[0] = 0x45 // version 4, IHL 5
-	// total length is meaningless for TSO but set it anyway
-	binary.BigEndian.PutUint16(pkt[2:4], uint16(ipLen+tcpLen+payLen))
-	binary.BigEndian.PutUint16(pkt[4:6], 0x4242) // original ID
-	pkt[8] = 64                                  // TTL
-	pkt[9] = unix.IPPROTO_TCP
-	copy(pkt[12:16], []byte{10, 0, 0, 1}) // src
-	copy(pkt[16:20], []byte{10, 0, 0, 2}) // dst
-
-	// TCP header
-	binary.BigEndian.PutUint16(pkt[20:22], 12345) // sport
-	binary.BigEndian.PutUint16(pkt[22:24], 80)    // dport
-	binary.BigEndian.PutUint32(pkt[24:28], 10000) // seq
-	binary.BigEndian.PutUint32(pkt[28:32], 20000) // ack
-	pkt[32] = 0x50                                // data offset 5 words
-	pkt[33] = 0x18                                // ACK | PSH
-	binary.BigEndian.PutUint16(pkt[34:36], 65535) // window
-
-	// payload
-	for i := 0; i < payLen; i++ {
-		pkt[ipLen+tcpLen+i] = byte(i & 0xff)
-	}
-
-	return pkt, VirtioNetHdr{
-		Flags:      unix.VIRTIO_NET_HDR_F_NEEDS_CSUM,
-		GSOType:    unix.VIRTIO_NET_HDR_GSO_TCPV4,
-		HdrLen:     uint16(ipLen + tcpLen),
-		GSOSize:    uint16(mss),
-		CsumStart:  uint16(ipLen),
-		CsumOffset: 16,
-	}
-}
-
-func TestSegmentTCPv4(t *testing.T) {
-	const mss = 100
-	const numSeg = 3
-	pkt, hdr := buildTSOv4(t, mss*numSeg, mss)
-
-	scratch := make([]byte, tunSegBufSize)
-	var out [][]byte
-	if err := segmentTCP(pkt, hdr, &out, scratch); err != nil {
-		t.Fatalf("segmentTCP: %v", err)
-	}
-	if len(out) != numSeg {
-		t.Fatalf("expected %d segments, got %d", numSeg, len(out))
-	}
-
-	for i, seg := range out {
-		if len(seg) != 40+mss {
-			t.Errorf("seg %d: unexpected len %d", i, len(seg))
-		}
-		totalLen := binary.BigEndian.Uint16(seg[2:4])
-		if totalLen != uint16(40+mss) {
-			t.Errorf("seg %d: total_len=%d want %d", i, totalLen, 40+mss)
-		}
-		id := binary.BigEndian.Uint16(seg[4:6])
-		if id != 0x4242+uint16(i) {
-			t.Errorf("seg %d: ip id=%#x want %#x", i, id, 0x4242+uint16(i))
-		}
-		seq := binary.BigEndian.Uint32(seg[24:28])
-		wantSeq := uint32(10000 + i*mss)
-		if seq != wantSeq {
-			t.Errorf("seg %d: seq=%d want %d", i, seq, wantSeq)
-		}
-		flags := seg[33]
-		wantFlags := byte(0x10) // ACK only, PSH cleared
-		if i == numSeg-1 {
-			wantFlags = 0x18 // ACK | PSH preserved on last
-		}
-		if flags != wantFlags {
-			t.Errorf("seg %d: flags=%#x want %#x", i, flags, wantFlags)
-		}
-		// IPv4 header checksum must verify against itself.
-		if !verifyChecksum(seg[:20], 0) {
-			t.Errorf("seg %d: bad IPv4 header checksum", i)
-		}
-		// TCP checksum must verify against the pseudo-header.
-		psum := pseudoHeaderIPv4(seg[12:16], seg[16:20], unix.IPPROTO_TCP, 20+mss)
-		if !verifyChecksum(seg[20:], psum) {
-			t.Errorf("seg %d: bad TCP checksum", i)
-		}
-	}
-}
-
-func TestSegmentTCPv4OddTail(t *testing.T) {
-	// Payload of 250 bytes with MSS 100 → segments of 100, 100, 50.
-	pkt, hdr := buildTSOv4(t, 250, 100)
-	scratch := make([]byte, tunSegBufSize)
-	var out [][]byte
-	if err := segmentTCP(pkt, hdr, &out, scratch); err != nil {
-		t.Fatalf("segmentTCP: %v", err)
-	}
-	if len(out) != 3 {
-		t.Fatalf("want 3 segments, got %d", len(out))
-	}
-	wantPayLens := []int{100, 100, 50}
-	for i, seg := range out {
-		if len(seg)-40 != wantPayLens[i] {
-			t.Errorf("seg %d: pay len %d want %d", i, len(seg)-40, wantPayLens[i])
-		}
-		if !verifyChecksum(seg[:20], 0) {
-			t.Errorf("seg %d: bad IPv4 header checksum", i)
-		}
-		psum := pseudoHeaderIPv4(seg[12:16], seg[16:20], unix.IPPROTO_TCP, 20+wantPayLens[i])
-		if !verifyChecksum(seg[20:], psum) {
-			t.Errorf("seg %d: bad TCP checksum", i)
-		}
-	}
-}
-
-func TestSegmentTCPv6(t *testing.T) {
-	const ipLen = 40
-	const tcpLen = 20
-	const mss = 120
-	const numSeg = 2
-	payLen := mss * numSeg
-	pkt := make([]byte, ipLen+tcpLen+payLen)
-
-	// IPv6 header
-	pkt[0] = 0x60 // version 6
-	binary.BigEndian.PutUint16(pkt[4:6], uint16(tcpLen+payLen))
-	pkt[6] = unix.IPPROTO_TCP
-	pkt[7] = 64
-	// src/dst fe80::1 / fe80::2
-	pkt[8] = 0xfe
-	pkt[9] = 0x80
-	pkt[23] = 1
-	pkt[24] = 0xfe
-	pkt[25] = 0x80
-	pkt[39] = 2
-
-	// TCP header
-	binary.BigEndian.PutUint16(pkt[40:42], 12345)
-	binary.BigEndian.PutUint16(pkt[42:44], 80)
-	binary.BigEndian.PutUint32(pkt[44:48], 7)
-	binary.BigEndian.PutUint32(pkt[48:52], 99)
-	pkt[52] = 0x50
-	pkt[53] = 0x19 // FIN | ACK | PSH — exercise FIN clearing too
-	binary.BigEndian.PutUint16(pkt[54:56], 65535)
-
-	for i := 0; i < payLen; i++ {
-		pkt[ipLen+tcpLen+i] = byte(i)
-	}
-
-	hdr := VirtioNetHdr{
-		Flags:      unix.VIRTIO_NET_HDR_F_NEEDS_CSUM,
-		GSOType:    unix.VIRTIO_NET_HDR_GSO_TCPV6,
-		HdrLen:     uint16(ipLen + tcpLen),
-		GSOSize:    uint16(mss),
-		CsumStart:  uint16(ipLen),
-		CsumOffset: 16,
-	}
-
-	scratch := make([]byte, tunSegBufSize)
-	var out [][]byte
-	if err := segmentTCP(pkt, hdr, &out, scratch); err != nil {
-		t.Fatalf("segmentTCP: %v", err)
-	}
-	if len(out) != numSeg {
-		t.Fatalf("want %d segments, got %d", numSeg, len(out))
-	}
-
-	for i, seg := range out {
-		if len(seg) != ipLen+tcpLen+mss {
-			t.Errorf("seg %d: len %d want %d", i, len(seg), ipLen+tcpLen+mss)
-		}
-		pl := binary.BigEndian.Uint16(seg[4:6])
-		if pl != uint16(tcpLen+mss) {
-			t.Errorf("seg %d: payload_length=%d want %d", i, pl, tcpLen+mss)
-		}
-		seq := binary.BigEndian.Uint32(seg[44:48])
-		if seq != uint32(7+i*mss) {
-			t.Errorf("seg %d: seq=%d want %d", i, seq, 7+i*mss)
-		}
-		flags := seg[53]
-		// Original flags = 0x19 (FIN|ACK|PSH). FIN(0x01)+PSH(0x08) should be
-		// cleared on all but the last; ACK(0x10) always preserved.
-		wantFlags := byte(0x10)
-		if i == numSeg-1 {
-			wantFlags = 0x19
-		}
-		if flags != wantFlags {
-			t.Errorf("seg %d: flags=%#x want %#x", i, flags, wantFlags)
-		}
-		psum := pseudoHeaderIPv6(seg[8:24], seg[24:40], unix.IPPROTO_TCP, tcpLen+mss)
-		if !verifyChecksum(seg[ipLen:], psum) {
-			t.Errorf("seg %d: bad TCP checksum", i)
-		}
-	}
-}
-
-func TestSegmentGSONonePassesThrough(t *testing.T) {
-	pkt, hdr := buildTSOv4(t, 100, 100)
-	hdr.GSOType = unix.VIRTIO_NET_HDR_GSO_NONE
-	hdr.Flags = 0 // no NEEDS_CSUM, leave packet untouched
-
-	scratch := make([]byte, tunSegBufSize)
-	var out [][]byte
-	if err := segmentInto(pkt, hdr, &out, scratch); err != nil {
-		t.Fatalf("segmentInto: %v", err)
-	}
-	if len(out) != 1 {
-		t.Fatalf("want 1 segment, got %d", len(out))
-	}
-	if len(out[0]) != len(pkt) {
-		t.Fatalf("unexpected length: %d vs %d", len(out[0]), len(pkt))
-	}
-}
-
-func TestSegmentRejectsUDP(t *testing.T) {
-	hdr := VirtioNetHdr{GSOType: unix.VIRTIO_NET_HDR_GSO_UDP}
-	var out [][]byte
-	if err := segmentInto(nil, hdr, &out, nil); err == nil {
-		t.Fatalf("expected rejection for UDP GSO")
-	}
-}
-
-func BenchmarkSegmentTCPv4(b *testing.B) {
-	sizes := []struct {
-		name   string
-		payLen int
-		mss    int
-	}{
-		{"64KiB_MSS1460", 65000, 1460},
-		{"16KiB_MSS1460", 16384, 1460},
-		{"4KiB_MSS1460", 4096, 1460},
-	}
-	for _, sz := range sizes {
-		b.Run(sz.name, func(b *testing.B) {
-			const ipLen = 20
-			const tcpLen = 20
-			pkt := make([]byte, ipLen+tcpLen+sz.payLen)
-			pkt[0] = 0x45
-			binary.BigEndian.PutUint16(pkt[2:4], uint16(ipLen+tcpLen+sz.payLen))
-			binary.BigEndian.PutUint16(pkt[4:6], 0x4242)
-			pkt[8] = 64
-			pkt[9] = unix.IPPROTO_TCP
-			copy(pkt[12:16], []byte{10, 0, 0, 1})
-			copy(pkt[16:20], []byte{10, 0, 0, 2})
-			binary.BigEndian.PutUint16(pkt[20:22], 12345)
-			binary.BigEndian.PutUint16(pkt[22:24], 80)
-			binary.BigEndian.PutUint32(pkt[24:28], 10000)
-			binary.BigEndian.PutUint32(pkt[28:32], 20000)
-			pkt[32] = 0x50
-			pkt[33] = 0x18
-			binary.BigEndian.PutUint16(pkt[34:36], 65535)
-			for i := 0; i < sz.payLen; i++ {
-				pkt[ipLen+tcpLen+i] = byte(i)
-			}
-			hdr := VirtioNetHdr{
-				Flags:      unix.VIRTIO_NET_HDR_F_NEEDS_CSUM,
-				GSOType:    unix.VIRTIO_NET_HDR_GSO_TCPV4,
-				HdrLen:     uint16(ipLen + tcpLen),
-				GSOSize:    uint16(sz.mss),
-				CsumStart:  uint16(ipLen),
-				CsumOffset: 16,
-			}
-
-			scratch := make([]byte, tunSegBufSize)
-			out := make([][]byte, 0, 64)
-
-			b.SetBytes(int64(len(pkt)))
-			b.ResetTimer()
-			for i := 0; i < b.N; i++ {
-				out = out[:0]
-				if err := segmentTCP(pkt, hdr, &out, scratch); err != nil {
-					b.Fatal(err)
-				}
-			}
-		})
-	}
-}
-
-// TestTunFileWriteVnetHdrNoAlloc verifies the IFF_VNET_HDR fast-path write is
-// allocation-free. We write to /dev/null so every call succeeds synchronously.
-func TestTunFileWriteVnetHdrNoAlloc(t *testing.T) {
-	fd, err := unix.Open("/dev/null", os.O_WRONLY, 0)
-	if err != nil {
-		t.Fatalf("open /dev/null: %v", err)
-	}
-	t.Cleanup(func() { _ = unix.Close(fd) })
-
-	tf := &Offload{fd: fd}
-	tf.writeIovs[0].Base = &validVnetHdr[0]
-	tf.writeIovs[0].SetLen(virtioNetHdrLen)
-
-	payload := make([]byte, 1400)
-	// Warm up (first call may trigger one-time internal allocations elsewhere).
-	if _, err := tf.Write(payload); err != nil {
-		t.Fatalf("Write: %v", err)
-	}
-
-	allocs := testing.AllocsPerRun(1000, func() {
-		if _, err := tf.Write(payload); err != nil {
-			t.Fatalf("Write: %v", err)
-		}
-	})
-	if allocs != 0 {
-		t.Fatalf("Write allocated %.1f times per call, want 0", allocs)
-	}
-}

overlay/tio/vnethdr_linux.go

@@ -1,39 +0,0 @@
-package tio
-
-import "encoding/binary"
-
-// Size of the legacy struct virtio_net_hdr that the kernel prepends/expects on
-// a TUN opened with IFF_VNET_HDR (TUNSETVNETHDRSZ not set).
-const virtioNetHdrLen = 10
-
-type VirtioNetHdr struct {
-	Flags      uint8
-	GSOType    uint8
-	HdrLen     uint16
-	GSOSize    uint16
-	CsumStart  uint16
-	CsumOffset uint16
-}
-
-// decode reads a virtio_net_hdr in host byte order (TUN default; we never
-// call TUNSETVNETLE so the kernel matches our endianness).
-func (h *VirtioNetHdr) decode(b []byte) {
-	h.Flags = b[0]
-	h.GSOType = b[1]
-	h.HdrLen = binary.NativeEndian.Uint16(b[2:4])
-	h.GSOSize = binary.NativeEndian.Uint16(b[4:6])
-	h.CsumStart = binary.NativeEndian.Uint16(b[6:8])
-	h.CsumOffset = binary.NativeEndian.Uint16(b[8:10])
-}
-
-// encode is the inverse of decode: writes the virtio_net_hdr fields into b
-// (must be at least virtioNetHdrLen bytes). Used to emit a TSO superpacket
-// on egress.
-func (h *VirtioNetHdr) encode(b []byte) {
-	b[0] = h.Flags
-	b[1] = h.GSOType
-	binary.NativeEndian.PutUint16(b[2:4], h.HdrLen)
-	binary.NativeEndian.PutUint16(b[4:6], h.GSOSize)
-	binary.NativeEndian.PutUint16(b[6:8], h.CsumStart)
-	binary.NativeEndian.PutUint16(b[8:10], h.CsumOffset)
-}

overlay/tun_android.go

@@ -13,45 +13,17 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 )
 
 type tun struct {
-	rwc         io.ReadWriteCloser
+	io.ReadWriteCloser
 	fd          int
 	vpnNetworks []netip.Prefix
 	Routes      atomic.Pointer[[]Route]
 	routeTree   atomic.Pointer[bart.Table[routing.Gateways]]
 	l           *logrus.Logger
-
-	readBuf  []byte
-	batchRet [1][]byte
-}
-
-func (t *tun) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.rwc.Read(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *tun) Write(p []byte) (int, error) {
-	return t.rwc.Write(p)
-}
-
-func (t *tun) WriteReject(p []byte) (int, error) {
-	return t.rwc.Write(p)
-}
-
-func (t *tun) Close() error {
-	return t.rwc.Close()
 }
 
 func newTunFromFd(c *config.C, l *logrus.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
@@ -60,10 +32,10 @@ func newTunFromFd(c *config.C, l *logrus.Logger, deviceFd int, vpnNetworks []net
 	file := os.NewFile(uintptr(deviceFd), "/dev/net/tun")
 
 	t := &tun{
-		rwc:         file,
+		ReadWriteCloser: file,
-		fd:          deviceFd,
+		fd:              deviceFd,
-		vpnNetworks: vpnNetworks,
+		vpnNetworks:     vpnNetworks,
-		l:           l,
+		l:               l,
 	}
 
 	err := t.reload(c, true)
@@ -127,6 +99,6 @@ func (t *tun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *tun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented for android")
 }

overlay/tun_darwin.go

@@ -16,7 +16,6 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -24,7 +23,7 @@ import (
 )
 
 type tun struct {
-	rwc         io.ReadWriteCloser
+	io.ReadWriteCloser
 	Device      string
 	vpnNetworks []netip.Prefix
 	DefaultMTU  int
@@ -35,9 +34,6 @@ type tun struct {
 
 	// cache out buffer since we need to prepend 4 bytes for tun metadata
 	out []byte
-
-	readBuf  []byte
-	batchRet [1][]byte
 }
 
 type ifReq struct {
@@ -128,11 +124,11 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, _ bool) (
 	}
 
 	t := &tun{
-		rwc:         os.NewFile(uintptr(fd), ""),
+		ReadWriteCloser: os.NewFile(uintptr(fd), ""),
-		Device:      name,
+		Device:          name,
-		vpnNetworks: vpnNetworks,
+		vpnNetworks:     vpnNetworks,
-		DefaultMTU:  c.GetInt("tun.mtu", DefaultMTU),
+		DefaultMTU:      c.GetInt("tun.mtu", DefaultMTU),
-		l:           l,
+		l:               l,
 	}
 
 	err = t.reload(c, true)
@@ -162,8 +158,8 @@ func newTunFromFd(_ *config.C, _ *logrus.Logger, _ int, _ []netip.Prefix) (*tun,
 }
 
 func (t *tun) Close() error {
-	if t.rwc != nil {
+	if t.ReadWriteCloser != nil {
-		return t.rwc.Close()
+		return t.ReadWriteCloser.Close()
 	}
 	return nil
 }
@@ -507,31 +503,15 @@ func delRoute(prefix netip.Prefix, gateway netroute.Addr) error {
 	return nil
 }
 
-func (t *tun) readOne(to []byte) (int, error) {
+func (t *tun) Read(to []byte) (int, error) {
 	buf := make([]byte, len(to)+4)
 
-	n, err := t.rwc.Read(buf)
+	n, err := t.ReadWriteCloser.Read(buf)
 
 	copy(to, buf[4:])
 	return n - 4, err
 }
 
-func (t *tun) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *tun) WriteReject(p []byte) (int, error) {
-	return t.Write(p)
-}
-
 // Write is only valid for single threaded use
 func (t *tun) Write(from []byte) (int, error) {
 	buf := t.out
@@ -557,7 +537,7 @@ func (t *tun) Write(from []byte) (int, error) {
 
 	copy(buf[4:], from)
 
-	n, err := t.rwc.Write(buf)
+	n, err := t.ReadWriteCloser.Write(buf)
 	return n - 4, err
 }
 
@@ -573,6 +553,6 @@ func (t *tun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *tun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented for darwin")
 }

overlay/tun_disabled.go

@@ -9,7 +9,6 @@ import (
 	"github.com/rcrowley/go-metrics"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/iputil"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
@@ -18,27 +17,9 @@ type disabledTun struct {
 	vpnNetworks []netip.Prefix
 
 	// Track these metrics since we don't have the tun device to do it for us
-	tx         metrics.Counter
+	tx metrics.Counter
-	rx         metrics.Counter
+	rx metrics.Counter
-	l          *logrus.Logger
+	l  *logrus.Logger
-	numReaders int
-
-	batchRet [1][]byte
-}
-
-func (t *disabledTun) Read() ([][]byte, error) {
-	r, ok := <-t.read
-	if !ok {
-		return nil, io.EOF
-	}
-
-	t.tx.Inc(1)
-	if t.l.Level >= logrus.DebugLevel {
-		t.l.WithField("raw", prettyPacket(r)).Debugf("Write payload")
-	}
-
-	t.batchRet[0] = r
-	return t.batchRet[:], nil
 }
 
 func newDisabledTun(vpnNetworks []netip.Prefix, queueLen int, metricsEnabled bool, l *logrus.Logger) *disabledTun {
@@ -46,7 +27,6 @@ func newDisabledTun(vpnNetworks []netip.Prefix, queueLen int, metricsEnabled boo
 		vpnNetworks: vpnNetworks,
 		read:        make(chan []byte, queueLen),
 		l:           l,
-		numReaders:  1,
 	}
 
 	if metricsEnabled {
@@ -76,6 +56,24 @@ func (*disabledTun) Name() string {
 	return "disabled"
 }
 
+func (t *disabledTun) Read(b []byte) (int, error) {
+	r, ok := <-t.read
+	if !ok {
+		return 0, io.EOF
+	}
+
+	if len(r) > len(b) {
+		return 0, fmt.Errorf("packet larger than mtu: %d > %d bytes", len(r), len(b))
+	}
+
+	t.tx.Inc(1)
+	if t.l.Level >= logrus.DebugLevel {
+		t.l.WithField("raw", prettyPacket(r)).Debugf("Write payload")
+	}
+
+	return copy(b, r), nil
+}
+
 func (t *disabledTun) handleICMPEchoRequest(b []byte) bool {
 	out := make([]byte, len(b))
 	out = iputil.CreateICMPEchoResponse(b, out)
@@ -107,25 +105,12 @@ func (t *disabledTun) Write(b []byte) (int, error) {
 	return len(b), nil
 }
 
-func (t *disabledTun) WriteReject(b []byte) (int, error) {
-	return t.Write(b)
-}
-
 func (t *disabledTun) SupportsMultiqueue() bool {
 	return true
 }
 
-func (t *disabledTun) NewMultiQueueReader() error {
+func (t *disabledTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
-	t.numReaders++
+	return t, nil
-	return nil
-}
-
-func (t *disabledTun) Readers() []tio.Queue {
-	out := make([]tio.Queue, t.numReaders)
-	for i := range t.numReaders {
-		out[i] = t
-	}
-	return out
 }
 
 func (t *disabledTun) Close() error {

overlay/tun_freebsd.go

@@ -7,9 +7,9 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"io"
 	"io/fs"
 	"net/netip"
-	"os"
 	"sync/atomic"
 	"syscall"
 	"time"
@@ -18,7 +18,6 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -94,203 +93,107 @@ type tun struct {
 	routeTree   atomic.Pointer[bart.Table[routing.Gateways]]
 	linkAddr    *netroute.LinkAddr
 	l           *logrus.Logger
-
+	devFd       int
-	fd        int
-	shutdownR int // read end of the shutdown pipe; closing the write end wakes blocked polls
-	shutdownW int // write end of the shutdown pipe; closing this signals shutdown to any blocked reader/writer
-	readPoll  [2]unix.PollFd
-	writePoll [2]unix.PollFd
-	closed    atomic.Bool
-
-	readBuf  []byte
-	batchRet [1][]byte
 }
 
-// blockOnRead waits until the tun fd is readable or shutdown has been signaled.
+func (t *tun) Read(to []byte) (int, error) {
-// Returns os.ErrClosed if Close was called.
+	// use readv() to read from the tunnel device, to eliminate the need for copying the buffer
-func (t *tun) blockOnRead() error {
+	if t.devFd < 0 {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
+		return -1, syscall.EINVAL
-	var err error
-	for {
-		_, err = unix.Poll(t.readPoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
 	}
-	tunEvents := t.readPoll[0].Revents
-	shutdownEvents := t.readPoll[1].Revents
-	t.readPoll[0].Revents = 0
-	t.readPoll[1].Revents = 0
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	}
-	if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
-}
 
-func (t *tun) blockOnWrite() error {
-	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
-	var err error
-	for {
-		_, err = unix.Poll(t.writePoll[:], -1)
-		if err != unix.EINTR {
-			break
-		}
-	}
-	tunEvents := t.writePoll[0].Revents
-	shutdownEvents := t.writePoll[1].Revents
-	t.writePoll[0].Revents = 0
-	t.writePoll[1].Revents = 0
-	if err != nil {
-		return err
-	}
-	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
-		return os.ErrClosed
-	}
-	if tunEvents&problemFlags != 0 {
-		return os.ErrClosed
-	}
-	return nil
-}
-
-func (t *tun) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *tun) WriteReject(p []byte) (int, error) {
-	return t.Write(p)
-}
-
-func (t *tun) readOne(to []byte) (int, error) {
 	// first 4 bytes is protocol family, in network byte order
-	var head [4]byte
+	head := make([]byte, 4)
-	iovecs := [2]syscall.Iovec{
+
+	iovecs := []syscall.Iovec{
 		{&head[0], 4},
 		{&to[0], uint64(len(to))},
 	}
-	for {
+
-		n, _, errno := syscall.Syscall(syscall.SYS_READV, uintptr(t.fd), uintptr(unsafe.Pointer(&iovecs[0])), 2)
+	n, _, errno := syscall.Syscall(syscall.SYS_READV, uintptr(t.devFd), uintptr(unsafe.Pointer(&iovecs[0])), uintptr(2))
-		if errno == 0 {
+
-			bytesRead := int(n)
+	var err error
-			if bytesRead < 4 {
+	if errno != 0 {
-				return 0, nil
+		err = syscall.Errno(errno)
-			}
+	} else {
-			return bytesRead - 4, nil
+		err = nil
-		}
+	}
-		switch errno {
+	// fix bytes read number to exclude header
-		case unix.EAGAIN:
+	bytesRead := int(n)
-			if err := t.blockOnRead(); err != nil {
+	if bytesRead < 0 {
-				return 0, err
+		return bytesRead, err
-			}
+	} else if bytesRead < 4 {
-		case unix.EINTR:
+		return 0, err
-			// retry
+	} else {
-		case unix.EBADF:
+		return bytesRead - 4, err
-			return 0, os.ErrClosed
-		default:
-			return 0, errno
-		}
 	}
 }
 
 // Write is only valid for single threaded use
 func (t *tun) Write(from []byte) (int, error) {
+	// use writev() to write to the tunnel device, to eliminate the need for copying the buffer
+	if t.devFd < 0 {
+		return -1, syscall.EINVAL
+	}
+
 	if len(from) <= 1 {
 		return 0, syscall.EIO
 	}
-
 	ipVer := from[0] >> 4
-	var head [4]byte
+	var head []byte
 	// first 4 bytes is protocol family, in network byte order
-	switch ipVer {
+	if ipVer == 4 {
-	case 4:
+		head = []byte{0, 0, 0, syscall.AF_INET}
-		head[3] = syscall.AF_INET
+	} else if ipVer == 6 {
-	case 6:
+		head = []byte{0, 0, 0, syscall.AF_INET6}
-		head[3] = syscall.AF_INET6
+	} else {
-	default:
 		return 0, fmt.Errorf("unable to determine IP version from packet")
 	}
-
+	iovecs := []syscall.Iovec{
-	iovecs := [2]syscall.Iovec{
 		{&head[0], 4},
 		{&from[0], uint64(len(from))},
 	}
-	for {
+
-		n, _, errno := syscall.Syscall(syscall.SYS_WRITEV, uintptr(t.fd), uintptr(unsafe.Pointer(&iovecs[0])), 2)
+	n, _, errno := syscall.Syscall(syscall.SYS_WRITEV, uintptr(t.devFd), uintptr(unsafe.Pointer(&iovecs[0])), uintptr(2))
-		if errno == 0 {
+
-			return int(n) - 4, nil
+	var err error
-		}
+	if errno != 0 {
-		switch errno {
+		err = syscall.Errno(errno)
-		case unix.EAGAIN:
+	} else {
-			if err := t.blockOnWrite(); err != nil {
+		err = nil
-				return 0, err
-			}
-		case unix.EINTR:
-			// retry
-		case unix.EBADF:
-			return 0, os.ErrClosed
-		default:
-			return 0, errno
-		}
 	}
+
+	return int(n) - 4, err
 }
 
 func (t *tun) Close() error {
-	if t.closed.Swap(true) {
+	if t.devFd >= 0 {
-		return nil
+		err := syscall.Close(t.devFd)
-	}
+		if err != nil {
-
-	// Closing the write end of the shutdown pipe causes any blocked Poll to
-	// return with POLLHUP on the shutdown fd, so readers/writers wake up and
-	// exit with os.ErrClosed.
-	if t.shutdownW >= 0 {
-		_ = unix.Close(t.shutdownW)
-		t.shutdownW = -1
-	}
-
-	if t.fd >= 0 {
-		if err := unix.Close(t.fd); err != nil {
 			t.l.WithError(err).Error("Error closing device")
 		}
-		t.fd = -1
+		t.devFd = -1
-	}
 
-	if t.shutdownR >= 0 {
+		c := make(chan struct{})
-		_ = unix.Close(t.shutdownR)
+		go func() {
-		t.shutdownR = -1
+			// destroying the interface can block if a read() is still pending. Do this asynchronously.
-	}
+			defer close(c)
+			s, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_DGRAM, syscall.IPPROTO_IP)
+			if err == nil {
+				defer syscall.Close(s)
+				ifreq := ifreqDestroy{Name: t.deviceBytes()}
+				err = ioctl(uintptr(s), syscall.SIOCIFDESTROY, uintptr(unsafe.Pointer(&ifreq)))
+			}
+			if err != nil {
+				t.l.WithError(err).Error("Error destroying tunnel")
+			}
+		}()
 
-	c := make(chan struct{})
+		// wait up to 1 second so we start blocking at the ioctl
-	go func() {
+		select {
-		// destroying the interface can block if a read() is still pending. Do this asynchronously.
+		case <-c:
-		defer close(c)
+		case <-time.After(1 * time.Second):
-		s, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_DGRAM, syscall.IPPROTO_IP)
-		if err == nil {
-			defer syscall.Close(s)
-			ifreq := ifreqDestroy{Name: t.deviceBytes()}
-			err = ioctl(uintptr(s), syscall.SIOCIFDESTROY, uintptr(unsafe.Pointer(&ifreq)))
 		}
-		if err != nil {
-			t.l.WithError(err).Error("Error destroying tunnel")
-		}
-	}()
-
-	// wait up to 1 second so we start blocking at the ioctl
-	select {
-	case <-c:
-	case <-time.After(1 * time.Second):
 	}
 
 	return nil
@@ -306,38 +209,16 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, _ bool) (
 	var err error
 	deviceName := c.GetString("tun.dev", "")
 	if deviceName != "" {
-		fd, err = unix.Open("/dev/"+deviceName, os.O_RDWR, 0)
+		fd, err = syscall.Open("/dev/"+deviceName, syscall.O_RDWR, 0)
 	}
 	if errors.Is(err, fs.ErrNotExist) || deviceName == "" {
 		// If the device doesn't already exist, request a new one and rename it
-		fd, err = unix.Open("/dev/tun", os.O_RDWR, 0)
+		fd, err = syscall.Open("/dev/tun", syscall.O_RDWR, 0)
 	}
 	if err != nil {
 		return nil, err
 	}
 
-	if err = unix.SetNonblock(fd, true); err != nil {
-		_ = unix.Close(fd)
-		return nil, fmt.Errorf("failed to set tun device as nonblocking: %w", err)
-	}
-
-	// Shutdown pipe lets Close wake any reader/writer blocked in Poll.
-	var pipeFds [2]int
-	if err = unix.Pipe2(pipeFds[:], unix.O_CLOEXEC|unix.O_NONBLOCK); err != nil {
-		_ = unix.Close(fd)
-		return nil, fmt.Errorf("failed to create shutdown pipe: %w", err)
-	}
-	shutdownR, shutdownW := pipeFds[0], pipeFds[1]
-
-	closeOnErr := true
-	defer func() {
-		if closeOnErr {
-			_ = unix.Close(fd)
-			_ = unix.Close(shutdownR)
-			_ = unix.Close(shutdownW)
-		}
-	}()
-
 	// Read the name of the interface
 	var name [16]byte
 	arg := fiodgnameArg{length: 16, buf: unsafe.Pointer(&name)}
@@ -356,7 +237,7 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, _ bool) (
 	}
 
 	if ctrlErr != nil {
-		return nil, ctrlErr
+		return nil, err
 	}
 
 	ifName := string(bytes.TrimRight(name[:], "\x00"))
@@ -372,6 +253,8 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, _ bool) (
 		}
 		defer syscall.Close(s)
 
+		fd := uintptr(s)
+
 		var fromName [16]byte
 		var toName [16]byte
 		copy(fromName[:], ifName)
@@ -383,7 +266,7 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, _ bool) (
 		}
 
 		// Set the device name
-		_ = ioctl(uintptr(s), syscall.SIOCSIFNAME, uintptr(unsafe.Pointer(&ifrr)))
+		_ = ioctl(fd, syscall.SIOCSIFNAME, uintptr(unsafe.Pointer(&ifrr)))
 	}
 
 	t := &tun{
@@ -391,24 +274,13 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, _ bool) (
 		vpnNetworks: vpnNetworks,
 		MTU:         c.GetInt("tun.mtu", DefaultMTU),
 		l:           l,
-		fd:          fd,
+		devFd:       fd,
-		shutdownR:   shutdownR,
-		shutdownW:   shutdownW,
-		readPoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLIN},
-			{Fd: int32(shutdownR), Events: unix.POLLIN},
-		},
-		writePoll: [2]unix.PollFd{
-			{Fd: int32(fd), Events: unix.POLLOUT},
-			{Fd: int32(shutdownR), Events: unix.POLLIN},
-		},
 	}
 
 	err = t.reload(c, true)
 	if err != nil {
 		return nil, err
 	}
-	closeOnErr = false
 
 	c.RegisterReloadCallback(func(c *config.C) {
 		err := t.reload(c, false)
@@ -582,7 +454,7 @@ func (t *tun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *tun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented for freebsd")
 }
 

overlay/tun_ios.go

@@ -16,44 +16,16 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 )
 
 type tun struct {
-	rwc         io.ReadWriteCloser
+	io.ReadWriteCloser
 	vpnNetworks []netip.Prefix
 	Routes      atomic.Pointer[[]Route]
 	routeTree   atomic.Pointer[bart.Table[routing.Gateways]]
 	l           *logrus.Logger
-
-	readBuf  []byte
-	batchRet [1][]byte
-}
-
-func (t *tun) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.rwc.Read(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *tun) Write(p []byte) (int, error) {
-	return t.rwc.Write(p)
-}
-
-func (t *tun) WriteReject(p []byte) (int, error) {
-	return t.rwc.Write(p)
-}
-
-func (t *tun) Close() error {
-	return t.rwc.Close()
 }
 
 func newTun(_ *config.C, _ *logrus.Logger, _ []netip.Prefix, _ bool) (*tun, error) {
@@ -63,9 +35,9 @@ func newTun(_ *config.C, _ *logrus.Logger, _ []netip.Prefix, _ bool) (*tun, erro
 func newTunFromFd(c *config.C, l *logrus.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
 	file := os.NewFile(uintptr(deviceFd), "/dev/tun")
 	t := &tun{
-		vpnNetworks: vpnNetworks,
+		vpnNetworks:     vpnNetworks,
-		rwc:         &tunReadCloser{f: file},
+		ReadWriteCloser: &tunReadCloser{f: file},
-		l:           l,
+		l:               l,
 	}
 
 	err := t.reload(c, true)
@@ -183,6 +155,6 @@ func (t *tun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *tun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented for ios")
 }

overlay/tun_linux.go

@@ -4,7 +4,9 @@
 package overlay
 
 import (
+	"encoding/binary"
 	"fmt"
+	"io"
 	"net"
 	"net/netip"
 	"os"
@@ -17,15 +19,123 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"
 )
 
+// tunFile wraps a TUN file descriptor with poll-based reads. The FD provided will be changed to non-blocking.
+// A shared eventfd allows Close to wake all readers blocked in poll.
+type tunFile struct {
+	fd      int
+	lastOne bool
+	pollFds [2]unix.PollFd
+	closed  bool
+}
+
+// newFriend makes a tunFile for a MultiQueueReader that copies the shutdown eventfd from the parent tun
+func (r *tunFile) newFriend(fd int) (*tunFile, error) {
+	if err := unix.SetNonblock(fd, true); err != nil {
+		return nil, fmt.Errorf("failed to set tun fd non-blocking: %w", err)
+	}
+	return &tunFile{
+		fd: fd,
+		pollFds: [2]unix.PollFd{
+			{Fd: int32(fd), Events: unix.POLLIN},
+			{Fd: r.pollFds[1].Fd, Events: unix.POLLIN},
+		},
+	}, nil
+}
+
+func newTunFd(fd int) (*tunFile, error) {
+	if err := unix.SetNonblock(fd, true); err != nil {
+		return nil, fmt.Errorf("failed to set tun fd non-blocking: %w", err)
+	}
+
+	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 := &tunFile{
+		fd:      fd,
+		lastOne: true,
+		pollFds: [2]unix.PollFd{
+			{Fd: int32(fd), Events: unix.POLLIN},
+			{Fd: int32(shutdownFd), Events: unix.POLLIN},
+		},
+	}
+
+	return out, nil
+}
+
+func (r *tunFile) block() error {
+	const problemFlags = unix.POLLHUP | unix.POLLNVAL | unix.POLLERR
+	var err error
+	for {
+		_, err = unix.Poll(r.pollFds[:], -1)
+		if err != unix.EINTR {
+			break
+		}
+	}
+	//always reset these!
+	tunEvents := r.pollFds[0].Revents
+	shutdownEvents := r.pollFds[1].Revents
+	r.pollFds[0].Revents = 0
+	r.pollFds[1].Revents = 0
+	//do the err check before trusting the potentially bogus bits we just got
+	if err != nil {
+		return err
+	}
+	if shutdownEvents&(unix.POLLIN|problemFlags) != 0 {
+		return os.ErrClosed
+	} else if tunEvents&problemFlags != 0 {
+		return os.ErrClosed
+	}
+	return nil
+}
+
+func (r *tunFile) Read(buf []byte) (int, error) {
+	for {
+		if n, err := unix.Read(r.fd, buf); err == nil {
+			return n, nil
+		} else if err == unix.EAGAIN {
+			if err = r.block(); err != nil {
+				return 0, err
+			}
+			continue
+		} else {
+			return 0, err
+		}
+	}
+}
+
+func (r *tunFile) Write(buf []byte) (int, error) {
+	return unix.Write(r.fd, buf)
+}
+
+func (r *tunFile) wakeForShutdown() error {
+	var buf [8]byte
+	binary.NativeEndian.PutUint64(buf[:], 1)
+	_, err := unix.Write(int(r.pollFds[1].Fd), buf[:])
+	return err
+}
+
+func (r *tunFile) Close() error {
+	if r.closed { // avoid closing more than once. Technically a fd could get re-used, which would be a problem
+		return nil
+	}
+	r.closed = true
+	if r.lastOne {
+		_ = unix.Close(int(r.pollFds[1].Fd))
+	}
+	return unix.Close(r.fd)
+}
+
 type tun struct {
-	readers     tio.Container
+	*tunFile
+	readers     []*tunFile
 	closeLock   sync.Mutex
 	Device      string
 	vpnNetworks []netip.Prefix
@@ -34,7 +144,6 @@ type tun struct {
 	TXQueueLen  int
 	deviceIndex int
 	ioctlFd     uintptr
-	vnetHdr     bool
 
 	Routes                    atomic.Pointer[[]Route]
 	routeTree                 atomic.Pointer[bart.Table[routing.Gateways]]
@@ -73,9 +182,7 @@ type ifreqQLEN struct {
 }
 
 func newTunFromFd(c *config.C, l *logrus.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) {
-	// We don't know what flags the caller opened this fd with and can't turn
+	t, err := newTunGeneric(c, l, deviceFd, vpnNetworks)
-	// on IFF_VNET_HDR after TUNSETIFF, so skip offload on inherited fds.
-	t, err := newTunGeneric(c, l, deviceFd, false, vpnNetworks)
 	if err != nil {
 		return nil, err
 	}
@@ -85,83 +192,46 @@ func newTunFromFd(c *config.C, l *logrus.Logger, deviceFd int, vpnNetworks []net
 	return t, nil
 }
 
-// openTunDev opens /dev/net/tun, creating the device node first if it's
-// missing (docker containers occasionally omit it).
-func openTunDev() (int, error) {
-	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
-	if err == nil {
-		return fd, nil
-	}
-	if !os.IsNotExist(err) {
-		return -1, err
-	}
-	if err = os.MkdirAll("/dev/net", 0755); err != nil {
-		return -1, fmt.Errorf("/dev/net/tun doesn't exist, failed to mkdir -p /dev/net: %w", err)
-	}
-	if err = unix.Mknod("/dev/net/tun", unix.S_IFCHR|0600, int(unix.Mkdev(10, 200))); err != nil {
-		return -1, fmt.Errorf("failed to create /dev/net/tun: %w", err)
-	}
-	fd, err = unix.Open("/dev/net/tun", os.O_RDWR, 0)
-	if err != nil {
-		return -1, fmt.Errorf("created /dev/net/tun, but still failed: %w", err)
-	}
-	return fd, nil
-}
-
-// tunSetIff runs TUNSETIFF with the given flags and returns the kernel-chosen
-// device name on success.
-func tunSetIff(fd int, name string, flags uint16) (string, error) {
-	var req ifReq
-	req.Flags = flags
-	copy(req.Name[:], name)
-	if err := ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
-		return "", err
-	}
-	return strings.Trim(string(req.Name[:]), "\x00"), nil
-}
-
-// tsoOffloadFlags are the TUN_F_* bits we ask the kernel to enable when a
-// TSO-capable TUN is available. CSUM is required as a prerequisite for TSO.
-const tsoOffloadFlags = unix.TUN_F_CSUM | unix.TUN_F_TSO4 | unix.TUN_F_TSO6
-
 func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueue bool) (*tun, error) {
-	baseFlags := uint16(unix.IFF_TUN | unix.IFF_NO_PI)
+	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
-	if multiqueue {
-		baseFlags |= unix.IFF_MULTI_QUEUE
-	}
-	nameStr := c.GetString("tun.dev", "")
-
-	// First try to open with IFF_VNET_HDR + TUNSETOFFLOAD so we can receive
-	// TSO superpackets. If either step fails (older kernel, unprivileged
-	// container, etc.) we close and fall back to a plain TUN.
-	fd, err := openTunDev()
 	if err != nil {
-		return nil, err
+		// If /dev/net/tun doesn't exist, try to create it (will happen in docker)
-	}
+		if os.IsNotExist(err) {
-	vnetHdr := true
+			err = os.MkdirAll("/dev/net", 0755)
-	name, err := tunSetIff(fd, nameStr, baseFlags|unix.IFF_VNET_HDR|unix.IFF_NAPI)
+			if err != nil {
-	if err != nil {
+				return nil, fmt.Errorf("/dev/net/tun doesn't exist, failed to mkdir -p /dev/net: %w", err)
-		_ = unix.Close(fd)
+			}
-		vnetHdr = false
+			err = unix.Mknod("/dev/net/tun", unix.S_IFCHR|0600, int(unix.Mkdev(10, 200)))
-	} else if err = ioctl(uintptr(fd), unix.TUNSETOFFLOAD, uintptr(tsoOffloadFlags)); err != nil {
+			if err != nil {
-		l.WithError(err).Warn("Failed to enable TUN offload (TSO); proceeding without virtio headers")
+				return nil, fmt.Errorf("failed to create /dev/net/tun: %w", err)
-		_ = unix.Close(fd)
+			}
-		vnetHdr = false
-	}
 
-	if !vnetHdr {
+			fd, err = unix.Open("/dev/net/tun", os.O_RDWR, 0)
-		fd, err = openTunDev()
+			if err != nil {
-		if err != nil {
+				return nil, fmt.Errorf("created /dev/net/tun, but still failed: %w", err)
+			}
+		} else {
 			return nil, err
 		}
-		name, err = tunSetIff(fd, nameStr, baseFlags)
-		if err != nil {
-			_ = unix.Close(fd)
-			return nil, &NameError{Name: nameStr, Underlying: err}
-		}
 	}
 
-	t, err := newTunGeneric(c, l, fd, vnetHdr, vpnNetworks)
+	var req ifReq
+	req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI)
+	if multiqueue {
+		req.Flags |= unix.IFF_MULTI_QUEUE
+	}
+	nameStr := c.GetString("tun.dev", "")
+	copy(req.Name[:], nameStr)
+	if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
+		_ = unix.Close(fd)
+		return nil, &NameError{
+			Name:       nameStr,
+			Underlying: err,
+		}
+	}
+	name := strings.Trim(string(req.Name[:]), "\x00")
+
+	t, err := newTunGeneric(c, l, fd, vpnNetworks)
 	if err != nil {
 		return nil, err
 	}
@@ -172,29 +242,16 @@ func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, multiqueu
 }
 
 // newTunGeneric does all the stuff common to different tun initialization paths. It will close your files on error.
-func newTunGeneric(c *config.C, l *logrus.Logger, fd int, vnetHdr bool, vpnNetworks []netip.Prefix) (*tun, error) {
+func newTunGeneric(c *config.C, l *logrus.Logger, fd int, vpnNetworks []netip.Prefix) (*tun, error) {
-	var container tio.Container
+	tfd, err := newTunFd(fd)
-	var err error
-	if vnetHdr {
-		container, err = tio.NewOffloadContainer()
-	} else {
-		container, err = tio.NewPollContainer()
-	}
-
 	if err != nil {
 		_ = unix.Close(fd)
 		return nil, err
 	}
-	err = container.Add(fd)
-	if err != nil {
-		_ = unix.Close(fd)
-		return nil, err
-	}
-
 	t := &tun{
-		readers:                   container,
+		tunFile:                   tfd,
+		readers:                   []*tunFile{tfd},
 		closeLock:                 sync.Mutex{},
-		vnetHdr:                   vnetHdr,
 		vpnNetworks:               vpnNetworks,
 		TXQueueLen:                c.GetInt("tun.tx_queue", 500),
 		useSystemRoutes:           c.GetBool("tun.use_system_route_table", false),
@@ -296,38 +353,32 @@ func (t *tun) SupportsMultiqueue() bool {
 	return true
 }
 
-func (t *tun) NewMultiQueueReader() error {
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	t.closeLock.Lock()
 	defer t.closeLock.Unlock()
 
 	fd, err := unix.Open("/dev/net/tun", os.O_RDWR, 0)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	flags := uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
+	var req ifReq
-	if t.vnetHdr {
+	req.Flags = uint16(unix.IFF_TUN | unix.IFF_NO_PI | unix.IFF_MULTI_QUEUE)
-		flags |= unix.IFF_VNET_HDR | unix.IFF_NAPI
+	copy(req.Name[:], t.Device)
-	}
+	if err = ioctl(uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&req))); err != nil {
-	if _, err = tunSetIff(fd, t.Device, flags); err != nil {
 		_ = unix.Close(fd)
-		return err
+		return nil, err
 	}
 
-	if t.vnetHdr {
+	out, err := t.tunFile.newFriend(fd)
-		if err = ioctl(uintptr(fd), unix.TUNSETOFFLOAD, uintptr(tsoOffloadFlags)); err != nil {
-			_ = unix.Close(fd)
-			return fmt.Errorf("failed to enable offload on multiqueue tun fd: %w", err)
-		}
-	}
-
-	err = t.readers.Add(fd)
 	if err != nil {
 		_ = unix.Close(fd)
-		return err
+		return nil, err
 	}
 
-	return nil
+	t.readers = append(t.readers, out)
+
+	return out, nil
 }
 
 func (t *tun) RoutesFor(ip netip.Addr) routing.Gateways {
@@ -757,10 +808,6 @@ func (t *tun) updateRoutes(r netlink.RouteUpdate) {
 	t.routeTree.Store(newTree)
 }
 
-func (t *tun) Readers() []tio.Queue {
-	return t.readers.Queues()
-}
-
 func (t *tun) Close() error {
 	t.closeLock.Lock()
 	defer t.closeLock.Unlock()
@@ -770,10 +817,32 @@ func (t *tun) Close() error {
 		t.routeChan = nil
 	}
 
+	// Signal all readers blocked in poll to wake up and exit
+	_ = t.tunFile.wakeForShutdown()
+
 	if t.ioctlFd > 0 {
 		_ = unix.Close(int(t.ioctlFd))
 		t.ioctlFd = 0
 	}
 
-	return t.readers.Close()
+	for i := range t.readers {
+		if i == 0 {
+			continue //we want to close the zeroth reader last
+		}
+		err := t.readers[i].Close()
+		if err != nil {
+			t.l.WithField("reader", i).WithError(err).Error("Error closing tun reader")
+		} else {
+			t.l.WithField("reader", i).Info("Closed tun reader")
+		}
+	}
+
+	//this is t.readers[0] too
+	err := t.tunFile.Close()
+	if err != nil {
+		t.l.WithField("reader", 0).WithError(err).Error("Error closing tun reader")
+	} else {
+		t.l.WithField("reader", 0).Info("Closed tun reader")
+	}
+	return err
 }

overlay/tun_linux_test.go

@@ -3,9 +3,7 @@
 
 package overlay
 
-import (
+import "testing"
-	"testing"
-)
 
 var runAdvMSSTests = []struct {
 	name     string

overlay/tun_netbsd.go

@@ -6,6 +6,7 @@ package overlay
 import (
 	"errors"
 	"fmt"
+	"io"
 	"net/netip"
 	"os"
 	"regexp"
@@ -16,7 +17,6 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -66,25 +66,6 @@ type tun struct {
 	l           *logrus.Logger
 	f           *os.File
 	fd          int
-
-	readBuf  []byte
-	batchRet [1][]byte
-}
-
-func (t *tun) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *tun) WriteReject(p []byte) (int, error) {
-	return t.Write(p)
 }
 
 var deviceNameRE = regexp.MustCompile(`^tun[0-9]+$`)
@@ -160,7 +141,7 @@ func (t *tun) Close() error {
 	return nil
 }
 
-func (t *tun) readOne(to []byte) (int, error) {
+func (t *tun) Read(to []byte) (int, error) {
 	rc, err := t.f.SyscallConn()
 	if err != nil {
 		return 0, fmt.Errorf("failed to get syscall conn for tun: %w", err)
@@ -413,7 +394,7 @@ func (t *tun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *tun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented for netbsd")
 }
 

overlay/tun_openbsd.go

@@ -6,6 +6,7 @@ package overlay
 import (
 	"errors"
 	"fmt"
+	"io"
 	"net/netip"
 	"os"
 	"regexp"
@@ -16,7 +17,6 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	netroute "golang.org/x/net/route"
@@ -59,25 +59,6 @@ type tun struct {
 	fd          int
 	// cache out buffer since we need to prepend 4 bytes for tun metadata
 	out []byte
-
-	readBuf  []byte
-	batchRet [1][]byte
-}
-
-func (t *tun) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.readOne(t.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *tun) WriteReject(p []byte) (int, error) {
-	return t.Write(p)
 }
 
 var deviceNameRE = regexp.MustCompile(`^tun[0-9]+$`)
@@ -143,7 +124,7 @@ func (t *tun) Close() error {
 	return nil
 }
 
-func (t *tun) readOne(to []byte) (int, error) {
+func (t *tun) Read(to []byte) (int, error) {
 	buf := make([]byte, len(to)+4)
 
 	n, err := t.f.Read(buf)
@@ -333,7 +314,7 @@ func (t *tun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *tun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *tun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented for openbsd")
 }
 

overlay/tun_tester.go

@@ -13,7 +13,6 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
@@ -27,17 +26,6 @@ type TestTun struct {
 	closed    atomic.Bool
 	rxPackets chan []byte // Packets to receive into nebula
 	TxPackets chan []byte // Packets transmitted outside by nebula
-
-	batchRet [1][]byte
-}
-
-func (t *TestTun) Read() ([][]byte, error) {
-	p, ok := <-t.rxPackets
-	if !ok {
-		return nil, os.ErrClosed
-	}
-	t.batchRet[0] = p
-	return t.batchRet[:], nil
 }
 
 func newTun(c *config.C, l *logrus.Logger, vpnNetworks []netip.Prefix, _ bool) (*TestTun, error) {
@@ -127,10 +115,6 @@ func (t *TestTun) Write(b []byte) (n int, err error) {
 	return len(b), nil
 }
 
-func (t *TestTun) WriteReject(b []byte) (int, error) {
-	return t.Write(b)
-}
-
 func (t *TestTun) Close() error {
 	if t.closed.CompareAndSwap(false, true) {
 		close(t.rxPackets)
@@ -139,10 +123,19 @@ func (t *TestTun) Close() error {
 	return nil
 }
 
+func (t *TestTun) Read(b []byte) (int, error) {
+	p, ok := <-t.rxPackets
+	if !ok {
+		return 0, os.ErrClosed
+	}
+	copy(b, p)
+	return len(p), nil
+}
+
 func (t *TestTun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *TestTun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *TestTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented")
 }

overlay/tun_windows.go

@@ -6,6 +6,7 @@ package overlay
 import (
 	"crypto"
 	"fmt"
+	"io"
 	"net/netip"
 	"os"
 	"path/filepath"
@@ -17,7 +18,6 @@ import (
 	"github.com/gaissmai/bart"
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 	"github.com/slackhq/nebula/util"
 	"github.com/slackhq/nebula/wintun"
@@ -36,25 +36,6 @@ type winTun struct {
 	l           *logrus.Logger
 
 	tun *wintun.NativeTun
-
-	readBuf  []byte
-	batchRet [1][]byte
-}
-
-func (t *winTun) Read() ([][]byte, error) {
-	if t.readBuf == nil {
-		t.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := t.tun.Read(t.readBuf, 0)
-	if err != nil {
-		return nil, err
-	}
-	t.batchRet[0] = t.readBuf[:n]
-	return t.batchRet[:], nil
-}
-
-func (t *winTun) WriteReject(p []byte) (int, error) {
-	return t.Write(p)
 }
 
 func newTunFromFd(_ *config.C, _ *logrus.Logger, _ int, _ []netip.Prefix) (Device, error) {
@@ -248,6 +229,10 @@ func (t *winTun) Name() string {
 	return t.Device
 }
 
+func (t *winTun) Read(b []byte) (int, error) {
+	return t.tun.Read(b, 0)
+}
+
 func (t *winTun) Write(b []byte) (int, error) {
 	return t.tun.Write(b, 0)
 }
@@ -256,7 +241,7 @@ func (t *winTun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (t *winTun) NewMultiQueueReader() (tio.Queue, error) {
+func (t *winTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
 	return nil, fmt.Errorf("TODO: multiqueue not implemented for windows")
 }
 

overlay/user.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"github.com/slackhq/nebula/config"
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
@@ -24,34 +23,17 @@ func NewUserDevice(vpnNetworks []netip.Prefix) (Device, error) {
 		outboundWriter: ow,
 		inboundReader:  ir,
 		inboundWriter:  iw,
-		numReaders:     1,
 	}, nil
 }
 
 type UserDevice struct {
 	vpnNetworks []netip.Prefix
-	numReaders  int
 
 	outboundReader *io.PipeReader
 	outboundWriter *io.PipeWriter
 
 	inboundReader *io.PipeReader
 	inboundWriter *io.PipeWriter
-
-	readBuf  []byte
-	batchRet [1][]byte
-}
-
-func (d *UserDevice) Read() ([][]byte, error) {
-	if d.readBuf == nil {
-		d.readBuf = make([]byte, defaultBatchBufSize)
-	}
-	n, err := d.outboundReader.Read(d.readBuf)
-	if err != nil {
-		return nil, err
-	}
-	d.batchRet[0] = d.readBuf[:n]
-	return d.batchRet[:], nil
 }
 
 func (d *UserDevice) Activate() error {
@@ -68,29 +50,20 @@ func (d *UserDevice) SupportsMultiqueue() bool {
 	return true
 }
 
-func (d *UserDevice) NewMultiQueueReader() error {
+func (d *UserDevice) NewMultiQueueReader() (io.ReadWriteCloser, error) {
-	d.numReaders++
+	return d, nil
-	return nil
-}
-
-func (d *UserDevice) Readers() []tio.Queue {
-	out := make([]tio.Queue, d.numReaders)
-	for i := range d.numReaders {
-		out[i] = d
-	}
-	return out
 }
 
 func (d *UserDevice) Pipe() (*io.PipeReader, *io.PipeWriter) {
 	return d.inboundReader, d.outboundWriter
 }
 
+func (d *UserDevice) Read(p []byte) (n int, err error) {
+	return d.outboundReader.Read(p)
+}
 func (d *UserDevice) Write(p []byte) (n int, err error) {
 	return d.inboundWriter.Write(p)
 }
-func (d *UserDevice) WriteReject(p []byte) (n int, err error) {
-	return d.Write(p)
-}
 func (d *UserDevice) Close() error {
 	d.inboundWriter.Close()
 	d.outboundWriter.Close()

service/service.go

@@ -44,10 +44,7 @@ type Service struct {
 }
 
 func New(control *nebula.Control) (*Service, error) {
-	wait, err := control.Start()
+	control.Start()
-	if err != nil {
-		return nil, err
-	}
 
 	ctx := control.Context()
 	eg, ctx := errgroup.WithContext(ctx)
@@ -144,12 +141,6 @@ func New(control *nebula.Control) (*Service, error) {
 		}
 	})
 
-	// Add the nebula wait function to the group so a fatal reader error
-	// propagates out through errgroup.Wait().
-	eg.Go(func() error {
-		return wait()
-	})
-
 	return &s, nil
 }
 

test/tun.go

@@ -1,10 +1,10 @@
-package overlay
+package test
 
 import (
 	"errors"
+	"io"
 	"net/netip"
 
-	"github.com/slackhq/nebula/overlay/tio"
 	"github.com/slackhq/nebula/routing"
 )
 
@@ -26,15 +26,11 @@ func (NoopTun) Name() string {
 	return "noop"
 }
 
-func (NoopTun) Read() ([][]byte, error) {
+func (NoopTun) Read([]byte) (int, error) {
-	return nil, nil
-}
-
-func (NoopTun) Write([]byte) (int, error) {
 	return 0, nil
 }
 
-func (NoopTun) WriteReject(p []byte) (int, error) {
+func (NoopTun) Write([]byte) (int, error) {
 	return 0, nil
 }
 
@@ -42,12 +38,8 @@ func (NoopTun) SupportsMultiqueue() bool {
 	return false
 }
 
-func (NoopTun) NewMultiQueueReader() error {
+func (NoopTun) NewMultiQueueReader() (io.ReadWriteCloser, error) {
-	return errors.New("unsupported")
+	return nil, errors.New("unsupported")
-}
-
-func (NoopTun) Readers() []tio.Queue {
-	return []tio.Queue{NoopTun{}}
 }
 
 func (NoopTun) Close() error {

udp/conn.go

@@ -8,12 +8,6 @@ import (
 
 const MTU = 9001
 
-// MaxWriteBatch is the largest batch any Conn.WriteBatch implementation is
-// required to accept. Callers SHOULD NOT pass more than this per call; Linux
-// backends preallocate sendmmsg scratch sized to this value, so exceeding it
-// only costs a chunked retry.
-const MaxWriteBatch = 128
-
 type EncReader func(
 	addr netip.AddrPort,
 	payload []byte,
@@ -22,19 +16,8 @@ type EncReader func(
 type Conn interface {
 	Rebind() error
 	LocalAddr() (netip.AddrPort, error)
-	// ListenOut invokes r for each received packet. On batch-capable
+	ListenOut(r EncReader)
-	// backends (recvmmsg), flush is called after each batch is fully
-	// delivered — callers use it to flush per-batch accumulators such as
-	// TUN write coalescers. Single-packet backends call flush after each
-	// packet. flush must not be nil.
-	ListenOut(r EncReader, flush func()) error
 	WriteTo(b []byte, addr netip.AddrPort) error
-	// WriteBatch sends a contiguous batch of packets, each with its own
-	// destination. bufs and addrs must have the same length. Linux uses
-	// sendmmsg(2) for a single syscall; other backends fall back to a
-	// WriteTo loop. Returns on the first error; callers may observe a
-	// partial send if some packets went out before the error.
-	WriteBatch(bufs [][]byte, addrs []netip.AddrPort) error
 	ReloadConfig(c *config.C)
 	SupportsMultipleReaders() bool
 	Close() error
@@ -48,8 +31,8 @@ func (NoopConn) Rebind() error {
 func (NoopConn) LocalAddr() (netip.AddrPort, error) {
 	return netip.AddrPort{}, nil
 }
-func (NoopConn) ListenOut(_ EncReader, _ func()) error {
+func (NoopConn) ListenOut(_ EncReader) {
-	return nil
+	return
 }
 func (NoopConn) SupportsMultipleReaders() bool {
 	return false
@@ -57,9 +40,6 @@ func (NoopConn) SupportsMultipleReaders() bool {
 func (NoopConn) WriteTo(_ []byte, _ netip.AddrPort) error {
 	return nil
 }
-func (NoopConn) WriteBatch(_ [][]byte, _ []netip.AddrPort) error {
-	return nil
-}
 func (NoopConn) ReloadConfig(_ *config.C) {
 	return
 }

udp/udp_darwin.go

@@ -140,26 +140,6 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error {
 	}
 }
 
-func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) error {
-	for i, b := range bufs {
-		if err := u.WriteTo(b, addrs[i]); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *StdConn) WriteSegmented(bufs [][]byte, addr netip.AddrPort, _ int) error {
-	for _, b := range bufs {
-		if err := u.WriteTo(b, addr); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *StdConn) SupportsGSO() bool { return false }
-
 func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
 	a := u.UDPConn.LocalAddr()
 
@@ -185,7 +165,7 @@ func NewUDPStatsEmitter(udpConns []Conn) func() {
 	return func() {}
 }
 
-func (u *StdConn) ListenOut(r EncReader, flush func()) error {
+func (u *StdConn) ListenOut(r EncReader) {
 	buffer := make([]byte, MTU)
 
 	for {
@@ -193,14 +173,14 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
 		n, rua, err := u.ReadFromUDPAddrPort(buffer)
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
-				return err
+				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+				return
 			}
 
 			u.l.WithError(err).Error("unexpected udp socket receive error")
 		}
 
 		r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])
-		flush()
 	}
 }
 

udp/udp_generic.go

@@ -44,26 +44,6 @@ func (u *GenericConn) WriteTo(b []byte, addr netip.AddrPort) error {
 	return err
 }
 
-func (u *GenericConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) error {
-	for i, b := range bufs {
-		if _, err := u.UDPConn.WriteToUDPAddrPort(b, addrs[i]); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *GenericConn) WriteSegmented(bufs [][]byte, addr netip.AddrPort, _ int) error {
-	for _, b := range bufs {
-		if _, err := u.UDPConn.WriteToUDPAddrPort(b, addr); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *GenericConn) SupportsGSO() bool { return false }
-
 func (u *GenericConn) LocalAddr() (netip.AddrPort, error) {
 	a := u.UDPConn.LocalAddr()
 
@@ -93,7 +73,7 @@ type rawMessage struct {
 	Len uint32
 }
 
-func (u *GenericConn) ListenOut(r EncReader, flush func()) error {
+func (u *GenericConn) ListenOut(r EncReader) {
 	buffer := make([]byte, MTU)
 
 	var lastRecvErr time.Time
@@ -103,7 +83,8 @@ func (u *GenericConn) ListenOut(r EncReader, flush func()) error {
 		n, rua, err := u.ReadFromUDPAddrPort(buffer)
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
-				return err
+				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+				return
 			}
 			// Dampen unexpected message warns to once per minute
 			if lastRecvErr.IsZero() || time.Since(lastRecvErr) > time.Minute {
@@ -114,7 +95,6 @@ func (u *GenericConn) ListenOut(r EncReader, flush func()) error {
 		}
 
 		r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n])
-		flush()
 	}
 }
 

udp/udp_linux.go

@@ -24,43 +24,6 @@ type StdConn struct {
 	isV4    bool
 	l       *logrus.Logger
 	batch   int
-
-	// sendmmsg scratch. Each queue has its own StdConn, so no locking is
-	// needed. Sized to MaxWriteBatch at construction; WriteBatch chunks
-	// larger inputs.
-	writeMsgs  []rawMessage
-	writeIovs  []iovec
-	writeNames [][]byte
-
-	// Per-entry UDP_SEGMENT cmsg scratch. writeCmsg is one contiguous slab
-	// of MaxWriteBatch * writeCmsgSpace bytes; each entry's cmsg header is
-	// pre-filled once in prepareWriteMessages. WriteBatch only rewrites the
-	// 2-byte gso_size payload (and toggles Hdr.Control on/off) per call.
-	writeCmsg      []byte
-	writeCmsgSpace int
-
-	// writeEntryEnd[e] is the bufs index *after* the last packet packed
-	// into mmsghdr entry e. Used to rewind `i` on partial sendmmsg success.
-	writeEntryEnd []int
-
-	// Preallocated closure + in/out slots for sendmmsg, so the hot path
-	// does not heap-allocate a fresh closure per call.
-	writeChunk int
-	writeSent  int
-	writeErrno syscall.Errno
-	writeFunc  func(fd uintptr) bool
-
-	// UDP GSO (sendmsg with UDP_SEGMENT cmsg) support. gsoSupported is
-	// probed once at socket creation. When true, WriteSegmented takes a
-	// single-syscall GSO path; otherwise it falls back to a WriteTo loop.
-	gsoSupported bool
-
-	// UDP GRO (recvmsg with UDP_GRO cmsg) support. groSupported is probed
-	// once at socket creation. When true, listenOutBatch allocates larger
-	// RX buffers and a per-entry cmsg slot so the kernel can coalesce
-	// consecutive same-flow datagrams into a single recvmmsg entry; the
-	// delivered cmsg carries the gso_size used to split them back apart.
-	groSupported bool
 }
 
 func setReusePort(network, address string, c syscall.RawConn) error {
@@ -107,102 +70,9 @@ func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch in
 	}
 	out.isV4 = af == unix.AF_INET
 
-	out.prepareWriteMessages(MaxWriteBatch)
-	out.writeFunc = out.sendmmsgRawWrite
-
-	out.prepareGSO()
-	// GRO delivers coalesced superpackets that need a cmsg to split back
-	// into segments. The single-packet RX path uses ReadFromUDPAddrPort
-	// and cannot see that cmsg, so only enable GRO for the batch path.
-	if batch > 1 {
-		out.prepareGRO()
-	}
-
 	return out, nil
 }
 
-// prepareWriteMessages allocates one mmsghdr/iovec/sockaddr/cmsg scratch
-// slot per sendmmsg entry. The iovec slab is sized to the same n so a
-// single entry can fan out to up to n iovecs (needed for UDP_SEGMENT runs
-// that coalesce consecutive bufs into one entry). Hdr.Iov / Hdr.Iovlen /
-// Hdr.Control / Hdr.Controllen are wired per call since each entry can
-// span a variable number of iovecs and may or may not carry a cmsg.
-func (u *StdConn) prepareWriteMessages(n int) {
-	u.writeMsgs = make([]rawMessage, n)
-	u.writeIovs = make([]iovec, n)
-	u.writeNames = make([][]byte, n)
-	u.writeEntryEnd = make([]int, n)
-
-	u.writeCmsgSpace = unix.CmsgSpace(2)
-	u.writeCmsg = make([]byte, n*u.writeCmsgSpace)
-	for k := 0; k < n; k++ {
-		off := k * u.writeCmsgSpace
-		h := (*unix.Cmsghdr)(unsafe.Pointer(&u.writeCmsg[off]))
-		h.Level = unix.SOL_UDP
-		h.Type = unix.UDP_SEGMENT
-		setCmsgLen(h, unix.CmsgLen(2))
-	}
-
-	for i := range u.writeMsgs {
-		u.writeNames[i] = make([]byte, unix.SizeofSockaddrInet6)
-		u.writeMsgs[i].Hdr.Name = &u.writeNames[i][0]
-	}
-}
-
-// maxGSOSegments caps the per-sendmsg GSO fan-out. Linux kernels have
-// historically capped UDP_MAX_SEGMENTS at 64; newer kernels raise it to 128
-// but we stay conservative so the same code works everywhere.
-const maxGSOSegments = 64
-
-// maxGSOBytes bounds the total payload per sendmsg() when UDP_SEGMENT is
-// set. The kernel stitches all iovecs into a single skb whose length the
-// UDP length field can represent, and also enforces sk_gso_max_size (which
-// on most devices is 65536). We use 65535 so ciphertext + headers always
-// fits, avoiding EMSGSIZE on large TSO superpackets.
-const maxGSOBytes = 65535
-
-// prepareGSO probes UDP_SEGMENT support
-func (u *StdConn) prepareGSO() {
-	var probeErr error
-	if err := u.rawConn.Control(func(fd uintptr) {
-		probeErr = unix.SetsockoptInt(int(fd), unix.IPPROTO_UDP, unix.UDP_SEGMENT, 0)
-	}); err != nil {
-		return
-	}
-	if probeErr != nil {
-		return
-	}
-	u.gsoSupported = true
-}
-
-// udpGROBufferSize sizes the per-entry recvmmsg buffer when UDP_GRO is on.
-// The kernel stitches a run of same-flow datagrams into a single skb whose
-// length is bounded by sk_gso_max_size (typically 65535); anything larger
-// would be MSG_TRUNCed. We use the maximum representable UDP length so a
-// full superpacket always lands intact.
-const udpGROBufferSize = 65535
-
-// udpGROCmsgPayload is the size of the UDP_GRO cmsg data delivered by the
-// kernel: a single int (gso_size in bytes). See udp_cmsg_recv() in
-// net/ipv4/udp.c.
-const udpGROCmsgPayload = 4
-
-// prepareGRO turns on UDP_GRO so the kernel coalesces consecutive same-flow
-// datagrams into one recvmmsg entry, with a cmsg carrying the gso_size used
-// to split them back apart on the application side.
-func (u *StdConn) prepareGRO() {
-	var probeErr error
-	if err := u.rawConn.Control(func(fd uintptr) {
-		probeErr = unix.SetsockoptInt(int(fd), unix.IPPROTO_UDP, unix.UDP_GRO, 1)
-	}); err != nil {
-		return
-	}
-	if probeErr != nil {
-		return
-	}
-	u.groSupported = true
-}
-
 func (u *StdConn) SupportsMultipleReaders() bool {
 	return true
 }
@@ -301,7 +171,7 @@ func recvmmsg(fd uintptr, msgs []rawMessage) (int, bool, error) {
 	return int(n), true, nil
 }
 
-func (u *StdConn) listenOutSingle(r EncReader, flush func()) error {
+func (u *StdConn) listenOutSingle(r EncReader) {
 	var err error
 	var n int
 	var from netip.AddrPort
@@ -310,26 +180,20 @@ func (u *StdConn) listenOutSingle(r EncReader, flush func()) error {
 	for {
 		n, from, err = u.udpConn.ReadFromUDPAddrPort(buffer)
 		if err != nil {
-			return err
+			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+			return
 		}
 		from = netip.AddrPortFrom(from.Addr().Unmap(), from.Port())
 		r(from, buffer[:n])
-		flush()
 	}
 }
 
-func (u *StdConn) listenOutBatch(r EncReader, flush func()) error {
+func (u *StdConn) listenOutBatch(r EncReader) {
 	var ip netip.Addr
 	var n int
 	var operr error
 
-	bufSize := MTU
+	msgs, buffers, names := u.PrepareRawMessages(u.batch)
-	cmsgSpace := 0
-	if u.groSupported {
-		bufSize = udpGROBufferSize
-		cmsgSpace = unix.CmsgSpace(udpGROCmsgPayload)
-	}
-	msgs, buffers, names, _ := u.PrepareRawMessages(u.batch, bufSize, cmsgSpace)
 
 	//reader needs to capture variables from this function, since it's used as a lambda with rawConn.Read
 	//defining it outside the loop so it gets re-used
@@ -339,17 +203,14 @@ func (u *StdConn) listenOutBatch(r EncReader, flush func()) error {
 	}
 
 	for {
-		if cmsgSpace > 0 {
-			for i := range msgs {
-				setMsgControllen(&msgs[i].Hdr, cmsgSpace)
-			}
-		}
 		err := u.rawConn.Read(reader)
 		if err != nil {
-			return err
+			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+			return
 		}
 		if operr != nil {
-			return operr
+			u.l.WithError(operr).Debug("operr: udp socket is closed, exiting read loop")
+			return
 		}
 
 		for i := 0; i < n; i++ {
@@ -359,72 +220,19 @@ func (u *StdConn) listenOutBatch(r EncReader, flush func()) error {
 			} else {
 				ip, _ = netip.AddrFromSlice(names[i][8:24])
 			}
-			from := netip.AddrPortFrom(ip.Unmap(), binary.BigEndian.Uint16(names[i][2:4]))
+			r(netip.AddrPortFrom(ip.Unmap(), binary.BigEndian.Uint16(names[i][2:4])), buffers[i][:msgs[i].Len])
-			payload := buffers[i][:msgs[i].Len]
-
-			segSize := 0
-			if u.groSupported {
-				segSize = parseUDPGRO(&msgs[i].Hdr)
-			}
-			if segSize <= 0 || segSize >= len(payload) {
-				// No coalescing happened (or a lone datagram).
-				r(from, payload)
-				continue
-			}
-			// GRO superpacket: the kernel guarantees every segment is
-			// exactly segSize bytes except for the final one, which may be
-			// short.
-			for off := 0; off < len(payload); off += segSize {
-				end := off + segSize
-				if end > len(payload) {
-					end = len(payload)
-				}
-				r(from, payload[off:end])
-			}
 		}
-		// End-of-batch: let callers (e.g. TUN write coalescer) flush any
-		// state they accumulated across this batch.
-		flush()
 	}
 }
 
-// parseUDPGRO walks the control buffer on hdr looking for a SOL_UDP/UDP_GRO
+func (u *StdConn) ListenOut(r EncReader) {
-// cmsg and returns the gso_size (bytes per coalesced segment) it carries.
-// Returns 0 when no UDP_GRO cmsg is present, which is the normal case for
-// lone datagrams that the kernel did not coalesce.
-func parseUDPGRO(hdr *msghdr) int {
-	controllen := int(hdr.Controllen)
-	if controllen < unix.SizeofCmsghdr || hdr.Control == nil {
-		return 0
-	}
-	ctrl := unsafe.Slice(hdr.Control, controllen)
-	off := 0
-	for off+unix.SizeofCmsghdr <= len(ctrl) {
-		ch := (*unix.Cmsghdr)(unsafe.Pointer(&ctrl[off]))
-		clen := int(ch.Len)
-		if clen < unix.SizeofCmsghdr || off+clen > len(ctrl) {
-			return 0
-		}
-		if ch.Level == unix.SOL_UDP && ch.Type == unix.UDP_GRO {
-			dataOff := off + unix.CmsgLen(0)
-			if dataOff+udpGROCmsgPayload <= len(ctrl) {
-				return int(int32(binary.NativeEndian.Uint32(ctrl[dataOff : dataOff+udpGROCmsgPayload])))
-			}
-			return 0
-		}
-		// Advance by the aligned cmsg space. CmsgSpace(n) is the stride
-		// from one header to the next (len aligned up to the platform's
-		// cmsg alignment).
-		off += unix.CmsgSpace(clen - unix.CmsgLen(0))
-	}
-	return 0
-}
-
-func (u *StdConn) ListenOut(r EncReader, flush func()) error {
 	if u.batch == 1 {
-		return u.listenOutSingle(r, flush)
+		//save some ram by not calling PrepareRawMessages for fields we won't use
+		//we could also make this path more common by calling recvmmsg with msgs[:1],
+		//but that's still the recvmmsg syscall, which would be a change
+		u.listenOutSingle(r)
 	} else {
-		return u.listenOutBatch(r, flush)
+		u.listenOutBatch(r)
 	}
 }
 
@@ -433,209 +241,6 @@ func (u *StdConn) WriteTo(b []byte, ip netip.AddrPort) error {
 	return err
 }
 
-// WriteBatch sends bufs via sendmmsg(2) using the preallocated scratch on
-// StdConn. Consecutive packets to the same destination with matching segment
-// sizes (all but possibly the last) are coalesced into a single mmsghdr entry
-// carrying a UDP_SEGMENT cmsg, so one syscall can mix runs of GSO superpackets
-// with plain one-off datagrams. Without GSO support every packet is its own
-// entry, matching the prior behaviour.
-//
-// Chunks larger than the scratch are processed across multiple syscalls. If
-// sendmmsg returns a fatal error before any entry is sent we fall back to
-// per-packet WriteTo for that chunk so the caller still gets best-effort
-// delivery.
-func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) error {
-	if len(bufs) != len(addrs) {
-		return fmt.Errorf("WriteBatch: len(bufs)=%d != len(addrs)=%d", len(bufs), len(addrs))
-	}
-
-	i := 0
-	for i < len(bufs) {
-		baseI := i
-		entry := 0
-		iovIdx := 0
-
-		for entry < len(u.writeMsgs) && i < len(bufs) {
-			iovBudget := len(u.writeIovs) - iovIdx
-			if iovBudget < 1 {
-				break
-			}
-			runLen, segSize := u.planRun(bufs, addrs, i, iovBudget)
-			if runLen == 0 {
-				break
-			}
-
-			for k := 0; k < runLen; k++ {
-				b := bufs[i+k]
-				if len(b) == 0 {
-					u.writeIovs[iovIdx+k].Base = nil
-					setIovLen(&u.writeIovs[iovIdx+k], 0)
-				} else {
-					u.writeIovs[iovIdx+k].Base = &b[0]
-					setIovLen(&u.writeIovs[iovIdx+k], len(b))
-				}
-			}
-
-			nlen, err := writeSockaddr(u.writeNames[entry], addrs[i], u.isV4)
-			if err != nil {
-				return err
-			}
-
-			hdr := &u.writeMsgs[entry].Hdr
-			hdr.Iov = &u.writeIovs[iovIdx]
-			setMsgIovlen(hdr, runLen)
-			hdr.Namelen = uint32(nlen)
-
-			if runLen >= 2 {
-				off := entry * u.writeCmsgSpace
-				dataOff := off + unix.CmsgLen(0)
-				binary.NativeEndian.PutUint16(u.writeCmsg[dataOff:dataOff+2], uint16(segSize))
-				hdr.Control = &u.writeCmsg[off]
-				setMsgControllen(hdr, u.writeCmsgSpace)
-			} else {
-				hdr.Control = nil
-				setMsgControllen(hdr, 0)
-			}
-
-			i += runLen
-			iovIdx += runLen
-			u.writeEntryEnd[entry] = i
-			entry++
-		}
-
-		if entry == 0 {
-			return fmt.Errorf("sendmmsg: no progress")
-		}
-
-		sent, serr := u.sendmmsg(entry)
-		if serr != nil && sent <= 0 {
-			// Nothing went out for this chunk; fall back to WriteTo for each
-			// packet that was queued this iteration.
-			for k := baseI; k < i; k++ {
-				if werr := u.WriteTo(bufs[k], addrs[k]); werr != nil {
-					return werr
-				}
-			}
-			continue
-		}
-		if sent == 0 {
-			return fmt.Errorf("sendmmsg made no progress")
-		}
-		// Rewind i to the end of the last successfully sent entry. For a
-		// full-success send this leaves i unchanged; for a partial send it
-		// replays the remainder on the next outer-loop iteration.
-		i = u.writeEntryEnd[sent-1]
-	}
-	return nil
-}
-
-// planRun groups consecutive packets starting at `start` that can be sent as
-// a single UDP GSO superpacket (one sendmmsg entry with UDP_SEGMENT cmsg).
-// A run of length 1 means the entry carries no cmsg and the kernel treats
-// it as a plain datagram. Returns the run length and the per-segment size
-// (which equals len(bufs[start])). Without GSO support every call returns
-// runLen=1.
-func (u *StdConn) planRun(bufs [][]byte, addrs []netip.AddrPort, start, iovBudget int) (int, int) {
-	if start >= len(bufs) || iovBudget < 1 {
-		return 0, 0
-	}
-	segSize := len(bufs[start])
-	if !u.gsoSupported || segSize == 0 || segSize > maxGSOBytes {
-		return 1, segSize
-	}
-	dst := addrs[start]
-	maxLen := maxGSOSegments
-	if iovBudget < maxLen {
-		maxLen = iovBudget
-	}
-	runLen := 1
-	total := segSize
-	for runLen < maxLen && start+runLen < len(bufs) {
-		nextLen := len(bufs[start+runLen])
-		if nextLen == 0 || nextLen > segSize {
-			break
-		}
-		if addrs[start+runLen] != dst {
-			break
-		}
-		if total+nextLen > maxGSOBytes {
-			break
-		}
-		total += nextLen
-		runLen++
-		if nextLen < segSize {
-			// A short packet must be the last in the run.
-			break
-		}
-	}
-	return runLen, segSize
-}
-
-// sendmmsgRawWrite is the preallocated callback passed to rawConn.Write. It
-// reads its input (u.writeChunk) and writes its outputs (u.writeSent,
-// u.writeErrno) through StdConn fields so the closure itself does not
-// capture per-call locals and therefore does not heap-allocate.
-func (u *StdConn) sendmmsgRawWrite(fd uintptr) bool {
-	r1, _, errno := unix.Syscall6(
-		unix.SYS_SENDMMSG,
-		fd,
-		uintptr(unsafe.Pointer(&u.writeMsgs[0])),
-		uintptr(u.writeChunk),
-		0,
-		0,
-		0,
-	)
-	if errno == syscall.EAGAIN || errno == syscall.EWOULDBLOCK {
-		return false
-	}
-	u.writeSent = int(r1)
-	u.writeErrno = errno
-	return true
-}
-
-func (u *StdConn) sendmmsg(n int) (int, error) {
-	u.writeChunk = n
-	u.writeSent = 0
-	u.writeErrno = 0
-	if err := u.rawConn.Write(u.writeFunc); err != nil {
-		return u.writeSent, err
-	}
-	if u.writeErrno != 0 {
-		return u.writeSent, &net.OpError{Op: "sendmmsg", Err: u.writeErrno}
-	}
-	return u.writeSent, nil
-}
-
-// writeSockaddr encodes addr into buf (which must be at least
-// SizeofSockaddrInet6 bytes). Returns the number of bytes used. If isV4 is
-// true and addr is not a v4 (or v4-in-v6) address, returns an error.
-func writeSockaddr(buf []byte, addr netip.AddrPort, isV4 bool) (int, error) {
-	ap := addr.Addr().Unmap()
-	if isV4 {
-		if !ap.Is4() {
-			return 0, ErrInvalidIPv6RemoteForSocket
-		}
-		// struct sockaddr_in: { sa_family_t(2), in_port_t(2, BE), in_addr(4), zero(8) }
-		// sa_family is host endian.
-		binary.NativeEndian.PutUint16(buf[0:2], unix.AF_INET)
-		binary.BigEndian.PutUint16(buf[2:4], addr.Port())
-		ip4 := ap.As4()
-		copy(buf[4:8], ip4[:])
-		for j := 8; j < 16; j++ {
-			buf[j] = 0
-		}
-		return unix.SizeofSockaddrInet4, nil
-	}
-	// struct sockaddr_in6: { sa_family_t(2), in_port_t(2, BE), flowinfo(4), in6_addr(16), scope_id(4) }
-	binary.NativeEndian.PutUint16(buf[0:2], unix.AF_INET6)
-	binary.BigEndian.PutUint16(buf[2:4], addr.Port())
-	binary.NativeEndian.PutUint32(buf[4:8], 0)
-	ip6 := addr.Addr().As16()
-	copy(buf[8:24], ip6[:])
-	binary.NativeEndian.PutUint32(buf[24:28], 0)
-	return unix.SizeofSockaddrInet6, nil
-}
-
 func (u *StdConn) ReloadConfig(c *config.C) {
 	b := c.GetInt("listen.read_buffer", 0)
 	if b > 0 {

udp/udp_linux_32.go

@@ -30,18 +30,13 @@ type rawMessage struct {
 	Len uint32
 }
 
-func (u *StdConn) PrepareRawMessages(n, bufSize, cmsgSpace int) ([]rawMessage, [][]byte, [][]byte, []byte) {
+func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
 	msgs := make([]rawMessage, n)
 	buffers := make([][]byte, n)
 	names := make([][]byte, n)
 
-	var cmsgs []byte
-	if cmsgSpace > 0 {
-		cmsgs = make([]byte, n*cmsgSpace)
-	}
-
 	for i := range msgs {
-		buffers[i] = make([]byte, bufSize)
+		buffers[i] = make([]byte, MTU)
 		names[i] = make([]byte, unix.SizeofSockaddrInet6)
 
 		vs := []iovec{
@@ -53,28 +48,7 @@ func (u *StdConn) PrepareRawMessages(n, bufSize, cmsgSpace int) ([]rawMessage, [
 
 		msgs[i].Hdr.Name = &names[i][0]
 		msgs[i].Hdr.Namelen = uint32(len(names[i]))
-
-		if cmsgSpace > 0 {
-			msgs[i].Hdr.Control = &cmsgs[i*cmsgSpace]
-			msgs[i].Hdr.Controllen = uint32(cmsgSpace)
-		}
 	}
 
-	return msgs, buffers, names, cmsgs
+	return msgs, buffers, names
-}
-
-func setIovLen(v *iovec, n int) {
-	v.Len = uint32(n)
-}
-
-func setMsgIovlen(m *msghdr, n int) {
-	m.Iovlen = uint32(n)
-}
-
-func setMsgControllen(m *msghdr, n int) {
-	m.Controllen = uint32(n)
-}
-
-func setCmsgLen(h *unix.Cmsghdr, n int) {
-	h.Len = uint32(n)
 }

udp/udp_linux_64.go

@@ -33,18 +33,13 @@ type rawMessage struct {
 	Pad0 [4]byte
 }
 
-func (u *StdConn) PrepareRawMessages(n, bufSize, cmsgSpace int) ([]rawMessage, [][]byte, [][]byte, []byte) {
+func (u *StdConn) PrepareRawMessages(n int) ([]rawMessage, [][]byte, [][]byte) {
 	msgs := make([]rawMessage, n)
 	buffers := make([][]byte, n)
 	names := make([][]byte, n)
 
-	var cmsgs []byte
-	if cmsgSpace > 0 {
-		cmsgs = make([]byte, n*cmsgSpace)
-	}
-
 	for i := range msgs {
-		buffers[i] = make([]byte, bufSize)
+		buffers[i] = make([]byte, MTU)
 		names[i] = make([]byte, unix.SizeofSockaddrInet6)
 
 		vs := []iovec{
@@ -56,28 +51,7 @@ func (u *StdConn) PrepareRawMessages(n, bufSize, cmsgSpace int) ([]rawMessage, [
 
 		msgs[i].Hdr.Name = &names[i][0]
 		msgs[i].Hdr.Namelen = uint32(len(names[i]))
-
-		if cmsgSpace > 0 {
-			msgs[i].Hdr.Control = &cmsgs[i*cmsgSpace]
-			msgs[i].Hdr.Controllen = uint64(cmsgSpace)
-		}
 	}
 
-	return msgs, buffers, names, cmsgs
+	return msgs, buffers, names
-}
-
-func setIovLen(v *iovec, n int) {
-	v.Len = uint64(n)
-}
-
-func setMsgIovlen(m *msghdr, n int) {
-	m.Iovlen = uint64(n)
-}
-
-func setMsgControllen(m *msghdr, n int) {
-	m.Controllen = uint64(n)
-}
-
-func setCmsgLen(h *unix.Cmsghdr, n int) {
-	h.Len = uint64(n)
 }

udp/udp_rio_windows.go

@@ -140,7 +140,7 @@ func (u *RIOConn) bind(l *logrus.Logger, sa windows.Sockaddr) error {
 	return nil
 }
 
-func (u *RIOConn) ListenOut(r EncReader, flush func()) error {
+func (u *RIOConn) ListenOut(r EncReader) {
 	buffer := make([]byte, MTU)
 
 	var lastRecvErr time.Time
@@ -151,7 +151,8 @@ func (u *RIOConn) ListenOut(r EncReader, flush func()) error {
 
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
-				return err
+				u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+				return
 			}
 			// Dampen unexpected message warns to once per minute
 			if lastRecvErr.IsZero() || time.Since(lastRecvErr) > time.Minute {
@@ -162,7 +163,6 @@ func (u *RIOConn) ListenOut(r EncReader, flush func()) error {
 		}
 
 		r(netip.AddrPortFrom(netip.AddrFrom16(rua.Addr).Unmap(), (rua.Port>>8)|((rua.Port&0xff)<<8)), buffer[:n])
-		flush()
 	}
 }
 
@@ -317,26 +317,6 @@ func (u *RIOConn) WriteTo(buf []byte, ip netip.AddrPort) error {
 	return winrio.SendEx(u.rq, dataBuffer, 1, nil, addressBuffer, nil, nil, 0, 0)
 }
 
-func (u *RIOConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) error {
-	for i, b := range bufs {
-		if err := u.WriteTo(b, addrs[i]); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *RIOConn) WriteSegmented(bufs [][]byte, addr netip.AddrPort, _ int) error {
-	for _, b := range bufs {
-		if err := u.WriteTo(b, addr); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *RIOConn) SupportsGSO() bool { return false }
-
 func (u *RIOConn) LocalAddr() (netip.AddrPort, error) {
 	sa, err := windows.Getsockname(u.sock)
 	if err != nil {

udp/udp_tester.go

@@ -6,7 +6,6 @@ package udp
 import (
 	"io"
 	"net/netip"
-	"os"
 	"sync/atomic"
 
 	"github.com/sirupsen/logrus"
@@ -107,34 +106,13 @@ func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
 	return nil
 }
 
-func (u *TesterConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) error {
+func (u *TesterConn) ListenOut(r EncReader) {
-	for i, b := range bufs {
-		if err := u.WriteTo(b, addrs[i]); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *TesterConn) WriteSegmented(bufs [][]byte, addr netip.AddrPort, _ int) error {
-	for _, b := range bufs {
-		if err := u.WriteTo(b, addr); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func (u *TesterConn) SupportsGSO() bool { return false }
-
-func (u *TesterConn) ListenOut(r EncReader, flush func()) error {
 	for {
 		p, ok := <-u.RxPackets
 		if !ok {
-			return os.ErrClosed
+			return
 		}
 		r(p.From, p.Data)
-		flush()
 	}
 }