tunnel: write inbound packets to TUN in batches

this reduces syscalls count, but only for Linux

Author: pymq

service/tunnel.go

@@ -351,24 +351,66 @@ func (vp *VpnPeer) backgroundOutboundHandler(t *Tunnel) {
 }
 
 func (vp *VpnPeer) backgroundInboundHandler(t *Tunnel) {
+	batchSize := t.device.BatchSize()
+	bytesBufs := make([][]byte, 0, batchSize)
+	packetsBufs := make([]*vpn.Packet, batchSize)
+
 	for {
-		packet, open := <-vp.inboundCh
+		firstPacket, open := <-vp.inboundCh
 		if !open {
 			return
 		}
 		localIP := *vp.localIP.Load()
-		ok := packet.Parse()
-		if !ok {
-			t.logger.Warnf("got invalid packet from peerID (%s) local ip (%s)", vp.peerID, localIP)
-			t.device.PutTempPacket(packet)
-			continue
+
+		packetsBufs[0] = firstPacket
+		packetsBatch := readBatchFromChan(vp.inboundCh, packetsBufs, 1)
+
+		newLen := 0
+		for i, packet := range packetsBatch {
+			ok := packet.Parse()
+			if !ok {
+				t.logger.Warnf("got invalid packet from peerID (%s) local ip (%s)", vp.peerID, localIP)
+				t.device.PutTempPacket(packet)
+				packetsBatch[i] = nil
+				continue
+			}
+			packetsBatch[newLen] = packet
+			newLen++
 		}
-		// TODO: add batching
-		err := t.device.WritePacket(packet, localIP)
-		if err != nil {
-			t.logger.Warnf("write packet to vpn: %v", err)
+		filteredPackets := packetsBatch[:newLen]
+
+		if len(filteredPackets) > 0 {
+			err := t.device.WritePacketsBatch(filteredPackets, bytesBufs, localIP)
+			if err != nil {
+				t.logger.Warnf("write packets batch to vpn for local ip %s: %v", localIP, err)
+			}
 		}
 
-		t.device.PutTempPacket(packet)
+		for i, packet := range packetsBatch {
+			if packet == nil {
+				continue
+			}
+			t.device.PutTempPacket(packet)
+			packetsBatch[i] = nil
+		}
+	}
+}
+
+func readBatchFromChan(ch chan *vpn.Packet, buf []*vpn.Packet, offset int) []*vpn.Packet {
+	i := offset
+	for {
+		if i == len(buf) {
+			return buf[:i]
+		}
+		select {
+		case packet, ok := <-ch:
+			if !ok {
+				return buf[:i]
+			}
+			buf[i] = packet
+			i++
+		default:
+			return buf[:i]
+		}
 	}
 }

vpn/vpn.go

@@ -70,7 +70,6 @@ func (d *Device) PutTempPacket(data *Packet) {
 	d.packetsPool.Put(data)
 }
 
-// TODO: batch write
 func (d *Device) WritePacket(data *Packet, senderIP net.IP) error {
 	if data.IsIPv6 {
 		// TODO: implement. We need to set Device.localIP ipv6 instead of ipv4
@@ -92,6 +91,41 @@ func (d *Device) WritePacket(data *Packet, senderIP net.IP) error {
 	return nil
 }
 
+// bufs should be len(bufs) == 0, cap(bufs) == len(packets)
+func (d *Device) WritePacketsBatch(packets []*Packet, bufs [][]byte, senderIP net.IP) error {
+	for _, packet := range packets {
+		if packet.IsIPv6 {
+			// TODO: implement. We need to set Device.localIP ipv6 instead of ipv4
+			continue
+		} else {
+			copy(packet.Src, senderIP)
+			copy(packet.Dst, d.localIP)
+		}
+		packet.RecalculateChecksum()
+
+		bufs = append(bufs, packet.Buffer[:tunPacketOffset+len(packet.Packet)])
+	}
+
+	defer func() {
+		for i := range bufs {
+			bufs[i] = nil
+		}
+	}()
+
+	packetsCount, err := d.tun.Write(bufs, tunPacketOffset)
+	if err != nil {
+		return fmt.Errorf("write packet to tun: %v", err)
+	} else if packetsCount < len(bufs) {
+		d.logger.Warnf("wrote %d packets, but expected %d", packetsCount, len(bufs))
+	}
+
+	return nil
+}
+
+func (d *Device) BatchSize() int {
+	return d.tun.BatchSize()
+}
+
 func (d *Device) Close() error {
 	return d.tun.Close()
 }