[tunnel] eliminate per-packet copy in muxedConn ReadPacket path

Add tunOffset headroom to muxedConn pooled buffers so Splice can receive
them directly via readPacketDirect/putPacketBuffer without an extra copy.
Splice detects this via interface assertion and uses the zero-copy path,
eliminating both the copy and its own redundant sync.Pool.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: dilyevsky

pkg/tunnel/connection/muxed_conn.go

@@ -26,19 +26,30 @@ type muxedConn struct {
 	incomingPackets  chan *[]byte
 	packetBufferPool sync.Pool
 
+	// headroom is the number of bytes reserved before packet data in pooled
+	// buffers. This allows callers using readPacketDirect to receive buffers
+	// with pre-allocated headroom (e.g. for TUN transport headers), avoiding
+	// an extra copy.
+	headroom int
+
 	closeOnce sync.Once
 	closed    atomic.Bool
 }
 
+const defaultMTU = 1500
+
 // newMuxedConn creates a new *muxedConn.
 func newMuxedConn() *muxedConn {
+	headroom := tunOffset
+	bufSize := headroom + defaultMTU
 	return &muxedConn{
 		conns:           iptrie.NewTrie(),
 		prefixes:        make(map[netip.Prefix]Connection),
 		incomingPackets: make(chan *[]byte, 10000),
+		headroom:        headroom,
 		packetBufferPool: sync.Pool{
 			New: func() interface{} {
-				b := make([]byte, 1500)
+				b := make([]byte, bufSize)
 				return &b
 			},
 		},
@@ -48,10 +59,10 @@ func newMuxedConn() *muxedConn {
 func (m *muxedConn) readFromConn(src netip.Prefix, conn Connection) {
 	for {
 		pkt := m.packetBufferPool.Get().(*[]byte)
-		// Reset the buffer to its original size.
+		// Reset the buffer to its full capacity.
 		*pkt = (*pkt)[:cap(*pkt)]
 
-		n, err := conn.ReadPacket(*pkt)
+		n, err := conn.ReadPacket((*pkt)[m.headroom:])
 		if err != nil {
 			// If the connection is closed, remove it from the multiplexer and quit
 			// the read loop. Otherwise, treat it as transient error and just log it.
@@ -79,7 +90,7 @@ func (m *muxedConn) readFromConn(src netip.Prefix, conn Connection) {
 			continue
 		}
 
-		*pkt = (*pkt)[:n]
+		*pkt = (*pkt)[:n+m.headroom]
 		select {
 		case m.incomingPackets <- pkt:
 		default: // Channel is closed or full, return the buffer to the pool
@@ -209,13 +220,35 @@ func (m *muxedConn) ReadPacket(pkt []byte) (int, error) {
 		return 0, net.ErrClosed
 	}
 
-	n := copy(pkt, *p)
+	n := copy(pkt, (*p)[m.headroom:])
 
 	m.packetBufferPool.Put(p)
 
 	return n, nil
 }
 
+// readPacketDirect returns the next packet's pooled buffer directly, avoiding
+// a copy. Packet data starts at buf[m.headroom:]. The caller must call
+// putPacketBuffer when done with the buffer.
+func (m *muxedConn) readPacketDirect() (*[]byte, error) {
+	if m.closed.Load() {
+		return nil, net.ErrClosed
+	}
+
+	p, ok := <-m.incomingPackets
+	if !ok {
+		return nil, net.ErrClosed
+	}
+
+	return p, nil
+}
+
+// putPacketBuffer returns a buffer obtained from readPacketDirect to the pool.
+func (m *muxedConn) putPacketBuffer(p *[]byte) {
+	*p = (*p)[:cap(*p)]
+	m.packetBufferPool.Put(p)
+}
+
 func (m *muxedConn) writeToConn(addr netip.Addr, pkt []byte) []byte {
 	if !addr.IsValid() || !addr.IsGlobalUnicast() {
 		slog.Warn("Invalid IP", slog.String("ip", addr.String()))

pkg/tunnel/connection/splice.go

@@ -21,39 +21,39 @@ const (
 	tunOffset = device.MessageTransportHeaderSize
 )
 
-// SpliceConfig holds configuration options for splice operations
+// SpliceConfig holds configuration options for splice operations.
 type SpliceConfig struct {
 	recalculateChecksum bool
 	verifyChecksum      bool
 	logChecksumErrors   bool
 	observer            PacketObserver
 }
 
-// SpliceOption is a function that configures splice behavior
+// SpliceOption is a function that configures splice behavior.
 type SpliceOption func(*SpliceConfig)
 
-// WithChecksumRecalculation enables TCP checksum recalculation
+// WithChecksumRecalculation enables TCP checksum recalculation.
 func WithChecksumRecalculation() SpliceOption {
 	return func(c *SpliceConfig) {
 		c.recalculateChecksum = true
 	}
 }
 
-// WithChecksumVerification enables TCP checksum verification (for debugging)
+// WithChecksumVerification enables TCP checksum verification (for debugging).
 func WithChecksumVerification() SpliceOption {
 	return func(c *SpliceConfig) {
 		c.verifyChecksum = true
 	}
 }
 
-// WithChecksumErrorLogging enables detailed logging of checksum errors
+// WithChecksumErrorLogging enables detailed logging of checksum errors.
 func WithChecksumErrorLogging() SpliceOption {
 	return func(c *SpliceConfig) {
 		c.logChecksumErrors = true
 	}
 }
 
-// WithPacketObserver sets a packet observer for traffic monitoring
+// WithPacketObserver sets a packet observer for traffic monitoring.
 func WithPacketObserver(obs PacketObserver) SpliceOption {
 	return func(c *SpliceConfig) {
 		c.observer = obs
@@ -111,12 +111,10 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 			for i := 0; i < n; i++ {
 				packetData := pkts[i][:sizes[i]]
 
-				// Notify observer if set
 				if config.observer != nil {
 					config.observer.OnPacket(ExtractPacketInfo(packetData, DirectionOutbound))
 				}
 
-				// Recalculate TCP checksum if enabled and needed
 				if config.recalculateChecksum {
 					if err := recalculateChecksumIfNeeded(packetData, config); err != nil {
 						slog.Debug("Failed to recalculate checksum", slog.Any("error", err))
@@ -142,15 +140,29 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 	})
 
 	// Connection -> TUN path
+	//
+	// When the connection is a muxedConn with matching headroom, we use
+	// readPacketDirect to receive pooled buffers without an extra copy.
+	// Otherwise we fall back to the standard ReadPacket-into-local-pool path.
+	type directReader interface {
+		readPacketDirect() (*[]byte, error)
+		putPacketBuffer(*[]byte)
+	}
+	dr, zeroCopy := conn.(directReader)
+
 	g.Go(func() error {
 		defer func() {
 			slog.Debug("Stopped reading from connection")
 		}()
 
-		var pktPool = sync.Pool{
-			New: func() any {
-				return ptr.To(make([]byte, netstack.IPv6MinMTU+tunOffset))
-			},
+		// Fallback pool only allocated when zero-copy is not available.
+		var pktPool *sync.Pool
+		if !zeroCopy {
+			pktPool = &sync.Pool{
+				New: func() any {
+					return ptr.To(make([]byte, netstack.IPv6MinMTU+tunOffset))
+				},
+			}
 		}
 
 		pktCh := make(chan *[]byte, batchSize)
@@ -159,26 +171,33 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 			defer close(pktCh)
 
 			for {
-				pkt := pktPool.Get().(*[]byte)
-				n, err := conn.ReadPacket((*pkt)[tunOffset:])
-				if err != nil {
-					slog.Error("Failed to read from connection", slog.Any("error", err))
-					return fmt.Errorf("failed to read from connection: %w", err)
+				var pkt *[]byte
+				if zeroCopy {
+					var err error
+					pkt, err = dr.readPacketDirect()
+					if err != nil {
+						slog.Error("Failed to read from connection", slog.Any("error", err))
+						return fmt.Errorf("failed to read from connection: %w", err)
+					}
+				} else {
+					pkt = pktPool.Get().(*[]byte)
+					*pkt = (*pkt)[:cap(*pkt)]
+					n, err := conn.ReadPacket((*pkt)[tunOffset:])
+					if err != nil {
+						slog.Error("Failed to read from connection", slog.Any("error", err))
+						return fmt.Errorf("failed to read from connection: %w", err)
+					}
+					*pkt = (*pkt)[:n+tunOffset]
 				}
 
-				// Notify observer if set
-				if config.observer != nil && n > 0 {
-					config.observer.OnPacket(ExtractPacketInfo((*pkt)[tunOffset:tunOffset+n], DirectionInbound))
+				if config.observer != nil && len(*pkt) > tunOffset {
+					config.observer.OnPacket(ExtractPacketInfo((*pkt)[tunOffset:], DirectionInbound))
 				}
 
-				*pkt = (*pkt)[:n+tunOffset]
-
-				// Recalculate TCP checksum if enabled and needed
 				if config.recalculateChecksum {
 					packetData := (*pkt)[tunOffset:]
 					if err := recalculateChecksumIfNeeded(packetData, config); err != nil {
 						slog.Debug("Failed to recalculate checksum on incoming packet", slog.Any("error", err))
-						// Continue processing - not all packets are TCP
 					}
 				}
 
@@ -225,8 +244,12 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 				}
 
 				for i := 0; i < batchCount; i++ {
-					pkt := pkts[i][:cap(pkts[i])]
-					pktPool.Put(&pkt)
+					p := pkts[i][:cap(pkts[i])]
+					if zeroCopy {
+						dr.putPacketBuffer(&p)
+					} else {
+						pktPool.Put(&p)
+					}
 				}
 			}
 		}
@@ -247,17 +270,14 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 	return nil
 }
 
-// recalculateChecksumIfNeeded recalculates TCP checksum if the packet is TCP
+// recalculateChecksumIfNeeded recalculates TCP checksum if the packet is TCP.
 func recalculateChecksumIfNeeded(packetData []byte, config *SpliceConfig) error {
 	if len(packetData) < 20 {
-		return nil // Packet too short to be IP
+		return nil // Packet too short to be an IP packet.
 	}
 
-	// Check if this is a TCP packet
-	version := packetData[0] >> 4
 	isTCP := false
-
-	switch version {
+	switch packetData[0] >> 4 {
 	case 4:
 		if len(packetData) >= 20 {
 			ihl := int(packetData[0]&0x0F) * 4
@@ -270,14 +290,12 @@ func recalculateChecksumIfNeeded(packetData []byte, config *SpliceConfig) error
 			isTCP = true
 		}
 	default:
-		return nil // Unknown IP version
+		return nil // Unknown IP version.
 	}
-
 	if !isTCP {
-		return nil // Not a TCP packet
+		return nil // Not a TCP packet.
 	}
 
-	// Verify checksum before recalculation if enabled
 	if config.verifyChecksum {
 		valid, err := tunnet.VerifyTCPChecksum(packetData)
 		if err != nil {
@@ -291,7 +309,6 @@ func recalculateChecksumIfNeeded(packetData []byte, config *SpliceConfig) error
 		}
 	}
 
-	// Recalculate the checksum
 	if err := tunnet.RecalculateTCPChecksum(packetData); err != nil {
 		if config.logChecksumErrors {
 			slog.Error("Failed to recalculate TCP checksum", slog.Any("error", err))