Add GSO support

stack_mixed.go

@@ -13,17 +13,14 @@ import (
 	"github.com/sagernet/gvisor/pkg/tcpip/transport/udp"
 	"github.com/sagernet/gvisor/pkg/waiter"
 	"github.com/sagernet/sing-tun/internal/clashtcpip"
-	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/bufio"
 	"github.com/sagernet/sing/common/canceler"
 	E "github.com/sagernet/sing/common/exceptions"
 	M "github.com/sagernet/sing/common/metadata"
-	N "github.com/sagernet/sing/common/network"
 )
 
 type Mixed struct {
 	*System
-	writer                 N.VectorisedWriter
 	endpointIndependentNat bool
 	stack                  *stack.Stack
 	endpoint               *channel.Endpoint
@@ -38,7 +35,6 @@ func NewMixed(
 	}
 	return &Mixed{
 		System:                 system.(*System),
-		writer:                 options.Tun.CreateVectorisedWriter(),
 		endpointIndependentNat: options.EndpointIndependentNat,
 	}, nil
 }
@@ -48,7 +44,7 @@ func (m *Mixed) Start() error {
 	if err != nil {
 		return err
 	}
-	endpoint := channel.New(1024, m.mtu, "")
+	endpoint := channel.New(1024, uint32(m.mtu), "")
 	ipStack, err := newGVisorStack(endpoint)
 	if err != nil {
 		return err
@@ -95,26 +91,34 @@ func (m *Mixed) tunLoop() {
 		m.wintunLoop(winTun)
 		return
 	}
+	if linuxTUN, isLinuxTUN := m.tun.(LinuxTUN); isLinuxTUN {
+		m.frontHeadroom = linuxTUN.FrontHeadroom()
+		m.txChecksumOffload = linuxTUN.TXChecksumOffload()
+		batchSize := linuxTUN.BatchSize()
+		if batchSize > 1 {
+			m.batchLoop(linuxTUN, batchSize)
+			return
+		}
+	}
 	packetBuffer := make([]byte, m.mtu+PacketOffset)
 	for {
 		n, err := m.tun.Read(packetBuffer)
 		if err != nil {
-			return
+			if E.IsClosed(err) {
+				return
+			}
+			m.logger.Error(E.Cause(err, "read packet"))
 		}
 		if n < clashtcpip.IPv4PacketMinLength {
 			continue
 		}
+		rawPacket := packetBuffer[:n]
 		packet := packetBuffer[PacketOffset:n]
-		switch ipVersion := packet[0] >> 4; ipVersion {
-		case 4:
-			err = m.processIPv4(packet)
-		case 6:
-			err = m.processIPv6(packet)
-		default:
-			err = E.New("ip: unknown version: ", ipVersion)
-		}
-		if err != nil {
-			m.logger.Trace(err)
+		if m.processPacket(packet) {
+			_, err = m.tun.Write(rawPacket)
+			if err != nil {
+				m.logger.Trace(E.Cause(err, "write packet"))
+			}
 		}
 	}
 }
@@ -129,62 +133,119 @@ func (m *Mixed) wintunLoop(winTun WinTun) {
 			release()
 			continue
 		}
-		switch ipVersion := packet[0] >> 4; ipVersion {
-		case 4:
-			err = m.processIPv4(packet)
-		case 6:
-			err = m.processIPv6(packet)
-		default:
-			err = E.New("ip: unknown version: ", ipVersion)
-		}
-		if err != nil {
-			m.logger.Trace(err)
+		if m.processPacket(packet) {
+			_, err = winTun.Write(packet)
+			if err != nil {
+				m.logger.Trace(E.Cause(err, "write packet"))
+			}
 		}
 		release()
 	}
 }
 
-func (m *Mixed) processIPv4(packet clashtcpip.IPv4Packet) error {
-	destination := packet.DestinationIP()
-	if destination == m.broadcastAddr || !destination.IsGlobalUnicast() {
-		return common.Error(m.tun.Write(packet))
+func (m *Mixed) batchLoop(linuxTUN LinuxTUN, batchSize int) {
+	packetBuffers := make([][]byte, batchSize)
+	writeBuffers := make([][]byte, batchSize)
+	packetSizes := make([]int, batchSize)
+	for i := range packetBuffers {
+		packetBuffers[i] = make([]byte, m.mtu+m.frontHeadroom)
 	}
-	switch packet.Protocol() {
-	case clashtcpip.TCP:
-		return m.processIPv4TCP(packet, packet.Payload())
-	case clashtcpip.UDP:
-		pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
-			Payload: buffer.MakeWithData(packet),
-		})
-		m.endpoint.InjectInbound(header.IPv4ProtocolNumber, pkt)
-		pkt.DecRef()
-		return nil
-	case clashtcpip.ICMP:
-		return m.processIPv4ICMP(packet, packet.Payload())
-	default:
-		return common.Error(m.tun.Write(packet))
+	for {
+		n, err := linuxTUN.BatchRead(packetBuffers, m.frontHeadroom, packetSizes)
+		if err != nil {
+			if E.IsClosed(err) {
+				return
+			}
+			m.logger.Error(E.Cause(err, "batch read packet"))
+		}
+		if n == 0 {
+			continue
+		}
+		for i := 0; i < n; i++ {
+			packetSize := packetSizes[i]
+			if packetSize < clashtcpip.IPv4PacketMinLength {
+				continue
+			}
+			packetBuffer := packetBuffers[i]
+			packet := packetBuffer[m.frontHeadroom : m.frontHeadroom+packetSize]
+			if m.processPacket(packet) {
+				writeBuffers = append(writeBuffers, packetBuffer[:m.frontHeadroom+packetSize])
+			}
+		}
+		if len(writeBuffers) > 0 {
+			err = linuxTUN.BatchWrite(writeBuffers, m.frontHeadroom)
+			if err != nil {
+				m.logger.Trace(E.Cause(err, "batch write packet"))
+			}
+			writeBuffers = writeBuffers[:0]
+		}
 	}
 }
 
-func (m *Mixed) processIPv6(packet clashtcpip.IPv6Packet) error {
-	if !packet.DestinationIP().IsGlobalUnicast() {
-		return common.Error(m.tun.Write(packet))
+func (m *Mixed) processPacket(packet []byte) bool {
+	var (
+		writeBack bool
+		err       error
+	)
+	switch ipVersion := packet[0] >> 4; ipVersion {
+	case 4:
+		writeBack, err = m.processIPv4(packet)
+	case 6:
+		writeBack, err = m.processIPv6(packet)
+	default:
+		err = E.New("ip: unknown version: ", ipVersion)
+	}
+	if err != nil {
+		m.logger.Trace(err)
+		return false
+	}
+	return writeBack
+}
+
+func (m *Mixed) processIPv4(packet clashtcpip.IPv4Packet) (writeBack bool, err error) {
+	writeBack = true
+	destination := packet.DestinationIP()
+	if destination == m.broadcastAddr || !destination.IsGlobalUnicast() {
+		return
 	}
 	switch packet.Protocol() {
 	case clashtcpip.TCP:
-		return m.processIPv6TCP(packet, packet.Payload())
+		err = m.processIPv4TCP(packet, packet.Payload())
 	case clashtcpip.UDP:
+		writeBack = false
 		pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
-			Payload: buffer.MakeWithData(packet),
+			Payload:           buffer.MakeWithData(packet),
+			IsForwardedPacket: true,
+		})
+		m.endpoint.InjectInbound(header.IPv4ProtocolNumber, pkt)
+		pkt.DecRef()
+		return
+	case clashtcpip.ICMP:
+		err = m.processIPv4ICMP(packet, packet.Payload())
+	}
+	return
+}
+
+func (m *Mixed) processIPv6(packet clashtcpip.IPv6Packet) (writeBack bool, err error) {
+	writeBack = true
+	if !packet.DestinationIP().IsGlobalUnicast() {
+		return
+	}
+	switch packet.Protocol() {
+	case clashtcpip.TCP:
+		err = m.processIPv6TCP(packet, packet.Payload())
+	case clashtcpip.UDP:
+		writeBack = false
+		pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
+			Payload:           buffer.MakeWithData(packet),
+			IsForwardedPacket: true,
 		})
 		m.endpoint.InjectInbound(header.IPv6ProtocolNumber, pkt)
 		pkt.DecRef()
-		return nil
 	case clashtcpip.ICMPv6:
-		return m.processIPv6ICMP(packet, packet.Payload())
-	default:
-		return common.Error(m.tun.Write(packet))
+		err = m.processIPv6ICMP(packet, packet.Payload())
 	}
+	return
 }
 
 func (m *Mixed) packetLoop() {
@@ -193,7 +254,7 @@ func (m *Mixed) packetLoop() {
 		if packet == nil {
 			break
 		}
-		bufio.WriteVectorised(m.writer, packet.AsSlices())
+		bufio.WriteVectorised(m.tun, packet.AsSlices())
 		packet.DecRef()
 	}
 }