crazy multiport stuff

interface.go

@@ -42,10 +42,13 @@ type InterfaceConfig struct {
 	DropLocalBroadcast bool
 	DropMulticast      bool
 	routines           int
-	MessageMetrics     *MessageMetrics
-	version            string
-	relayManager       *relayManager
-	punchy             *Punchy
+	// Multiport means writers[i] is bound to listen.port+i (not a shared
+	// SO_REUSEPORT port) and lane tunnels are negotiated with capable peers.
+	Multiport      bool
+	MessageMetrics *MessageMetrics
+	version        string
+	relayManager   *relayManager
+	punchy         *Punchy
 
 	tryPromoteEvery uint32
 	reQueryEvery    uint32
@@ -86,6 +89,7 @@ type Interface struct {
 	dropLocalBroadcast    bool
 	dropMulticast         bool
 	routines              int
+	multiport             bool
 	disconnectInvalid     atomic.Bool
 	closed                atomic.Bool
 	// cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine
@@ -222,6 +226,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) {
 		dropLocalBroadcast:    c.DropLocalBroadcast,
 		dropMulticast:         c.DropMulticast,
 		routines:              c.routines,
+		multiport:             c.Multiport,
 		version:               c.version,
 		writers:               make([]udp.Conn, c.routines),
 		batchers:              make([]batch.RxBatcher, c.routines),
@@ -277,7 +282,10 @@ func (f *Interface) activate() error {
 		"boringcrypto", boringEnabled(),
 	)
 
-	if f.routines > 1 && !f.outside.SupportsMultipleReaders() {
+	// Under multiport each socket has exactly one reader on its own port, so
+	// the shared-port multi-reader capability is irrelevant (and main.go
+	// already hard-errored on unsupported platforms).
+	if f.routines > 1 && !f.multiport && !f.outside.SupportsMultipleReaders() {
 		f.routines = 1
 		f.l.Warn("multiple udp readers are not supported on this platform, falling back to a single routine")
 	}
@@ -290,6 +298,11 @@ func (f *Interface) activate() error {
 		return err
 	}
 	if len(queues) < f.routines {
+		if f.multiport {
+			// The lane sockets are already bound one-per-routine; shrinking
+			// the routine count would leave bound ports with no reader.
+			return fmt.Errorf("multiport requires %d tun queues, device provided %d", f.routines, len(queues))
+		}
 		f.l.Warn("tun multiqueue is not supported on this platform, falling back to fewer routines",
 			"requested", f.routines, "opened", len(queues))
 		f.routines = len(queues)
@@ -372,7 +385,7 @@ func (f *Interface) listenOut(i int) {
 	scratch := make([]byte, mtu)
 
 	listener := func(fromUdpAddr netip.AddrPort, payload []byte, meta udp.RxMeta) {
-		f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr}, scratch, payload, h, fwPacket, lhh, nb, i, ctCache.Get(), meta)
+		f.readOutsidePackets(ViaSender{UdpAddr: fromUdpAddr, SockIdx: i}, scratch, payload, h, fwPacket, lhh, nb, i, ctCache.Get(), meta)
 	}
 
 	flusher := func() {
@@ -394,6 +407,39 @@ func (f *Interface) listenOut(i int) {
 	f.l.Debug("underlay reader is done", "reader", i)
 }
 
+// txQueue is the per-routine TX state owned by one listenIn goroutine. lane
+// is bound to the routine's own socket and carries lane-tunnel data; base is
+// bound to socket 0 and carries base-tunnel and relay data, which must keep
+// the base source port (a vanilla peer would otherwise see per-routine source
+// ports and roam-thrash). The two alias when multiport is off or on routine 0.
+// Concurrent sendmmsg on the shared socket-0 fd is safe: a flow is pinned to
+// one routine by tun steering, so per-flow wire order still holds.
+type txQueue struct {
+	lane *batch.SendBatch
+	base *batch.SendBatch
+}
+
+func (tx *txQueue) full() bool {
+	if tx.lane.Len() >= batch.SendBatchCap {
+		return true
+	}
+	return tx.base != tx.lane && tx.base.Len() >= batch.SendBatchCap
+}
+
+// flush drains base before lane so that when a flow moves from the base
+// tunnel onto a freshly established lane mid-window, its packets still leave
+// this host in encryption order.
+func (tx *txQueue) flush(l *slog.Logger, i int) {
+	if tx.base != tx.lane {
+		if err := tx.base.Flush(); err != nil {
+			l.Error("Failed to write outgoing batch", "error", err, "writer", 0)
+		}
+	}
+	if err := tx.lane.Flush(); err != nil {
+		l.Error("Failed to write outgoing batch", "error", err, "writer", i)
+	}
+}
+
 func (f *Interface) listenIn(queue tio.Queue, i int) {
 	// Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the
 	// same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false.
@@ -419,6 +465,10 @@ func (f *Interface) listenIn(queue tio.Queue, i int) {
 	rejectBuf := make([]byte, mtu)
 	arenaSize := batch.SendBatchCap * (udp.MTU + 32)
 	sb := batch.NewSendBatch(f.writers[i], batch.SendBatchCap, arenaSize)
+	tx := &txQueue{lane: sb, base: sb}
+	if f.multiport && i != 0 {
+		tx.base = batch.NewSendBatch(f.writers[0], batch.SendBatchCap, arenaSize)
+	}
 	fwPacket := &firewall.Packet{}
 	nb := make([]byte, 12, 12)
 
@@ -436,19 +486,15 @@ func (f *Interface) listenIn(queue tio.Queue, i int) {
 		}
 
 		for _, pkt := range pkts {
-			f.consumeInsidePacket(pkt, fwPacket, nb, sb, rejectBuf, i, conntrackCache.Get())
+			f.consumeInsidePacket(pkt, fwPacket, nb, tx, rejectBuf, i, conntrackCache.Get())
 			// Flush incrementally once a full sendmmsg batch has
 			// accumulated so the first packets of a deep read drain
 			// hit the wire while the rest are still being encrypted.
-			if sb.Len() >= batch.SendBatchCap {
-				if err := sb.Flush(); err != nil {
-					f.l.Error("Failed to write outgoing batch", "error", err, "writer", i)
-				}
+			if tx.full() {
+				tx.flush(f.l, i)
 			}
 		}
-		if err := sb.Flush(); err != nil {
-			f.l.Error("Failed to write outgoing batch", "error", err, "writer", i)
-		}
+		tx.flush(f.l, i)
 	}
 
 	f.l.Debug("overlay reader is done", "reader", i)