[tunnel] eliminate intermediate channel in Splice Connection->TUN path

Remove the producer goroutine and pktCh channel from the zero-copy
Splice path. Packets now flow directly from muxedConn.incomingPackets
(10k-deep channel) to TUN batch writes via readPacketDirect, skipping
a channel hop and goroutine context switch per packet.

Add tryReadPacketDirect() for non-blocking batch accumulation so the
write loop drains all queued packets before issuing a TUN write syscall.

Profiling showed TUN writes at 28-35% of CPU during bulk transfer. The
channel elimination reduces Go-level overhead but TUN syscalls remain
the dominant cost.

Author: dilyevsky

pkg/tunnel/connection/muxed_conn.go

@@ -243,6 +243,24 @@ func (m *muxedConn) readPacketDirect() (*[]byte, error) {
 	return p, nil
 }
 
+// tryReadPacketDirect attempts a non-blocking read from the incoming packet
+// channel. Returns (nil, false) if no packet is immediately available.
+func (m *muxedConn) tryReadPacketDirect() (*[]byte, bool) {
+	if m.closed.Load() {
+		return nil, false
+	}
+
+	select {
+	case p, ok := <-m.incomingPackets:
+		if !ok {
+			return nil, false
+		}
+		return p, true
+	default:
+		return nil, false
+	}
+}
+
 // putPacketBuffer returns a buffer obtained from readPacketDirect to the pool.
 func (m *muxedConn) putPacketBuffer(p *[]byte) {
 	*p = (*p)[:cap(*p)]

pkg/tunnel/connection/splice.go

@@ -142,10 +142,13 @@ 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.
+	// readPacketDirect to receive pooled buffers without an extra copy, and
+	// tryReadPacketDirect for non-blocking batch accumulation — eliminating
+	// the intermediate channel and goroutine.
 	// Otherwise we fall back to the standard ReadPacket-into-local-pool path.
 	type directReader interface {
 		readPacketDirect() (*[]byte, error)
+		tryReadPacketDirect() (*[]byte, bool)
 		putPacketBuffer(*[]byte)
 	}
 	dr, zeroCopy := conn.(directReader)
@@ -155,48 +158,98 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 			slog.Debug("Stopped reading from connection")
 		}()
 
-		// 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))
-				},
+		pkts := make([][]byte, batchSize)
+
+		if zeroCopy {
+			// Zero-copy path: read directly from the muxedConn's internal
+			// packet channel without an intermediate goroutine or channel.
+			// This reduces per-packet overhead (no channel hop, no extra
+			// goroutine switch) and improves batching by draining all
+			// available packets from the 10k-deep internal queue.
+			for {
+				// Block for the first packet.
+				pkt, err := dr.readPacketDirect()
+				if err != nil {
+					if errors.Is(err, net.ErrClosed) {
+						return nil
+					}
+					slog.Error("Failed to read from connection", slog.Any("error", err))
+					return fmt.Errorf("failed to read from connection: %w", err)
+				}
+
+				if config.observer != nil && len(*pkt) > tunOffset {
+					config.observer.OnPacket(ExtractPacketInfo((*pkt)[tunOffset:], DirectionInbound))
+				}
+				if config.recalculateChecksum {
+					if err := recalculateChecksumIfNeeded((*pkt)[tunOffset:], config); err != nil {
+						slog.Debug("Failed to recalculate checksum on incoming packet", slog.Any("error", err))
+					}
+				}
+
+				pkts[0] = *pkt
+				batchCount := 1
+
+				// Drain available packets non-blockingly to maximize batch size.
+				for batchCount < batchSize {
+					pkt, ok := dr.tryReadPacketDirect()
+					if !ok {
+						break
+					}
+					if config.observer != nil && len(*pkt) > tunOffset {
+						config.observer.OnPacket(ExtractPacketInfo((*pkt)[tunOffset:], DirectionInbound))
+					}
+					if config.recalculateChecksum {
+						if err := recalculateChecksumIfNeeded((*pkt)[tunOffset:], config); err != nil {
+							slog.Debug("Failed to recalculate checksum on incoming packet", slog.Any("error", err))
+						}
+					}
+					pkts[batchCount] = *pkt
+					batchCount++
+				}
+
+				if _, err := tunDev.Write(pkts[:batchCount], tunOffset); err != nil {
+					if strings.Contains(err.Error(), "closed") {
+						slog.Debug("TUN device closed")
+						return net.ErrClosed
+					}
+					slog.Error("Failed to write to TUN", slog.Any("error", err))
+					return fmt.Errorf("failed to write to TUN: %w", err)
+				}
+
+				for i := 0; i < batchCount; i++ {
+					p := pkts[i][:cap(pkts[i])]
+					dr.putPacketBuffer(&p)
+				}
 			}
 		}
 
+		// Non-zero-copy fallback: use intermediate channel for batching.
+		pktPool := &sync.Pool{
+			New: func() any {
+				return ptr.To(make([]byte, netstack.IPv6MinMTU+tunOffset))
+			},
+		}
+
 		pktCh := make(chan *[]byte, batchSize)
 
 		g.Go(func() error {
 			defer close(pktCh)
 
 			for {
-				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]
+				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]
 
-				if config.observer != nil && len(*pkt) > tunOffset {
+				if config.observer != nil && n > 0 {
 					config.observer.OnPacket(ExtractPacketInfo((*pkt)[tunOffset:], DirectionInbound))
 				}
-
 				if config.recalculateChecksum {
-					packetData := (*pkt)[tunOffset:]
-					if err := recalculateChecksumIfNeeded(packetData, config); err != nil {
+					if err := recalculateChecksumIfNeeded((*pkt)[tunOffset:], config); err != nil {
 						slog.Debug("Failed to recalculate checksum on incoming packet", slog.Any("error", err))
 					}
 				}
@@ -205,8 +258,6 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 			}
 		})
 
-		pkts := make([][]byte, batchSize)
-
 		for {
 			select {
 			case pkt, ok := <-pktCh:
@@ -217,14 +268,12 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 				pkts[0] = *pkt
 				batchCount := 1
 
-				closed := false
 			gatherBatch:
-				for batchCount < batchSize && !closed {
+				for batchCount < batchSize {
 					select {
 					case pkt, ok := <-pktCh:
 						if !ok {
-							closed = true
-							break
+							break gatherBatch
 						}
 						pkts[batchCount] = *pkt
 						batchCount++
@@ -238,18 +287,13 @@ func Splice(tunDev tun.Device, conn Connection, opts ...SpliceOption) error {
 						slog.Debug("TUN device closed")
 						return net.ErrClosed
 					}
-
 					slog.Error("Failed to write to TUN", slog.Any("error", err))
 					return fmt.Errorf("failed to write to TUN: %w", err)
 				}
 
 				for i := 0; i < batchCount; i++ {
 					p := pkts[i][:cap(pkts[i])]
-					if zeroCopy {
-						dr.putPacketBuffer(&p)
-					} else {
-						pktPool.Put(&p)
-					}
+					pktPool.Put(&p)
 				}
 			}
 		}