tunnel: remove OutboundChan from vpn.Device, use callback instead

this drastically reduces scheduler overhead on high bandwidth networks

Author: pymq

application.go

@@ -120,6 +120,7 @@ func (a *Application) Init(ctx context.Context, tunDevice tun.Device) error {
 	a.Dns = NewDNSService(a.Conf, a.Eventbus, a.ctx, a.logger)
 	a.AuthStatus = service.NewAuthStatus(a.P2p, a.Conf, a.Eventbus)
 	a.Tunnel = service.NewTunnel(a.P2p, vpnDevice, a.Conf)
+	go vpnDevice.ReadTUNPackets(a.Tunnel.HandleReadPackets)
 	a.SOCKS5, err = service.NewSOCKS5(a.P2p, a.Conf)
 	if err != nil {
 		return fmt.Errorf("failed to init socks5: %v", err)

service/tunnel.go

@@ -24,26 +24,32 @@ const (
 )
 
 type Tunnel struct {
-	p2p          P2p
-	conf         *config.Config
-	device       *vpn.Device
-	logger       *log.ZapEventLogger
-	peersLock    sync.RWMutex
-	peerIDToPeer map[peer.ID]*VpnPeer
-	netIPToPeer  map[string]*VpnPeer
+	p2p    P2p
+	conf   *config.Config
+	device *vpn.Device
+	logger *log.ZapEventLogger
+
+	isClosed         atomic.Bool
+	peersLock        sync.RWMutex
+	peerIDToPeer     map[peer.ID]*VpnPeer
+	netIPToPeer      map[string]*VpnPeer
+	udpBroadcastAddr net.IP
 }
 
 func NewTunnel(p2pService P2p, device *vpn.Device, conf *config.Config) *Tunnel {
+	localIP, netMask := conf.VPNLocalIPMask()
+	udpBroadcastAddr := vpn.GetIPv4BroadcastAddress(&net.IPNet{IP: localIP, Mask: netMask})
+
 	tunnel := &Tunnel{
-		p2p:          p2pService,
-		conf:         conf,
-		device:       device,
-		logger:       log.Logger("awl/service/tunnel"),
-		peerIDToPeer: make(map[peer.ID]*VpnPeer),
-		netIPToPeer:  make(map[string]*VpnPeer),
+		p2p:              p2pService,
+		conf:             conf,
+		device:           device,
+		logger:           log.Logger("awl/service/tunnel"),
+		peerIDToPeer:     make(map[peer.ID]*VpnPeer),
+		netIPToPeer:      make(map[string]*VpnPeer),
+		udpBroadcastAddr: udpBroadcastAddr,
 	}
 	tunnel.RefreshPeersList()
-	go tunnel.backgroundReadPackets()
 
 	return tunnel
 }
@@ -160,6 +166,8 @@ func (t *Tunnel) Close() {
 	t.peersLock.Lock()
 	defer t.peersLock.Unlock()
 
+	t.isClosed.Store(true)
+
 	for _, vpnPeer := range t.peerIDToPeer {
 		localIP := *vpnPeer.localIP.Load()
 		vpnPeer.Close(t)
@@ -168,17 +176,24 @@ func (t *Tunnel) Close() {
 	}
 }
 
-func (t *Tunnel) backgroundReadPackets() {
-	localIP, netMask := t.conf.VPNLocalIPMask()
-	broadcastAddr := vpn.GetIPv4BroadcastAddress(&net.IPNet{IP: localIP, Mask: netMask})
+// HandleReadPackets for successfully handled packets it sets packet in slice as nil
+func (t *Tunnel) HandleReadPackets(packets []*vpn.Packet) {
+	t.peersLock.RLock()
+	defer t.peersLock.RUnlock()
+
+	if t.isClosed.Load() {
+		return
+	}
+
+	for i, packet := range packets {
+		if packet == nil {
+			continue
+		}
 
-	// TODO: batch read
-	for packet := range t.device.OutboundChan() {
 		// TODO: ipv6 support
-		if packet.Dst.Equal(broadcastAddr) || packet.Dst.Equal(net.IPv4bcast) {
+		if packet.Dst.Equal(t.udpBroadcastAddr) || packet.Dst.Equal(net.IPv4bcast) {
 			// udp broadcast
 
-			t.peersLock.RLock()
 			for _, vpnPeer := range t.netIPToPeer {
 				// TODO: replace with event-based check OnConnected/OnDisconnected to improve performance
 				if !t.p2p.IsConnected(vpnPeer.peerID) {
@@ -195,26 +210,19 @@ func (t *Tunnel) backgroundReadPackets() {
 				}
 			}
 
-			t.device.PutTempPacket(packet)
-			t.peersLock.RUnlock()
-
 			continue
 		}
 
-		t.peersLock.RLock()
 		vpnPeer, ok := t.netIPToPeer[string(packet.Dst)]
 		if !ok {
-			t.device.PutTempPacket(packet)
-			t.peersLock.RUnlock()
 			continue
 		}
 
 		select {
 		case vpnPeer.outboundCh <- packet:
+			packets[i] = nil
 		default:
-			t.device.PutTempPacket(packet)
 		}
-		t.peersLock.RUnlock()
 	}
 }
 

vpn/vpn.go

@@ -9,24 +9,22 @@ import (
 	"sync/atomic"
 
 	"github.com/ipfs/go-log/v2"
+	"go.uber.org/zap"
 	"golang.zx2c4.com/wireguard/tun"
 )
 
 const (
 	InterfaceMTU   = 3500
-	maxContentSize = InterfaceMTU * 2 // TODO: determine real size
-	outboundChCap  = 50
+	maxContentSize = InterfaceMTU + 100
 	// internal tun header. see offset in tun_darwin (4) and tun_linux (virtioNetHdrLen, currently 10)
 	tunPacketOffset = 14
 )
 
 type Device struct {
-	tun        tun.Device
-	mtu        int64
-	localIP    net.IP
-	outboundCh chan *Packet
+	tun     tun.Device
+	mtu     int64
+	localIP net.IP
 
-	closeCh     chan struct{}
 	packetsPool sync.Pool
 	logger      *log.ZapEventLogger
 }
@@ -49,19 +47,16 @@ func NewDevice(existingTun tun.Device, interfaceName string, localIP net.IP, ipM
 	}
 
 	dev := &Device{
-		tun:        tunDevice,
-		mtu:        int64(realMtu),
-		localIP:    localIP,
-		outboundCh: make(chan *Packet, outboundChCap),
+		tun:     tunDevice,
+		mtu:     int64(realMtu),
+		localIP: localIP,
 		packetsPool: sync.Pool{
 			New: func() interface{} {
 				return new(Packet)
 			}},
-		logger:  log.Logger("awl/vpn"),
-		closeCh: make(chan struct{}),
+		logger: log.Logger("awl/vpn"),
 	}
 	go dev.tunEventsReader()
-	go dev.tunPacketsReader()
 
 	return dev, nil
 }
@@ -97,12 +92,7 @@ func (d *Device) WritePacket(data *Packet, senderIP net.IP) error {
 	return nil
 }
 
-func (d *Device) OutboundChan() <-chan *Packet {
-	return d.outboundCh
-}
-
 func (d *Device) Close() error {
-	close(d.closeCh)
 	return d.tun.Close()
 }
 
@@ -137,9 +127,7 @@ func (d *Device) tunEventsReader() {
 	}
 }
 
-func (d *Device) tunPacketsReader() {
-	defer close(d.outboundCh)
-
+func (d *Device) ReadTUNPackets(packetsHandler func([]*Packet)) {
 	batchSize := d.tun.BatchSize()
 	packets := make([]*Packet, batchSize)
 	bufs := make([][]byte, batchSize)
@@ -167,16 +155,19 @@ func (d *Device) tunPacketsReader() {
 			data.Packet = data.Buffer[tunPacketOffset : size+tunPacketOffset]
 			okay := data.Parse()
 			if !okay {
+				d.logger.Error("Failed to parse packet",
+					zap.ByteString("packet", data.Packet[:min(10, len(data.Packet))]),
+				)
+
+				packets[i] = nil
+				d.PutTempPacket(data)
 				continue
 			}
+		}
 
-			select {
-			case <-d.closeCh:
-				return
-			case d.outboundCh <- data:
-				// ok
-			}
-			packets[i] = nil
+		if packetsCount > 0 {
+			// packetsHandler skips nil packets in slice and sets packet to nil after successful processing
+			packetsHandler(packets[:packetsCount])
 		}
 
 		if errors.Is(err, tun.ErrTooManySegments) {