Endpoint independent mapping slows down bandwidth

@Lanius-collaris

Which commit of hussainmohd-a/rethink-app can be built with the latest celzero/firestack? rethink on my phone is outdated.

Interesting. Would such an impl (essentially, half-duplex?) make #77 work better?

I don't know, but current firestack's Endpoint-Independent Mapping mode need optimization.
I tested rethink and my gvisor-playground with quic-go/perf, rethink's handling of UDP ( single flow ) seems slower.
https://github.com/quic-go/perf

Commands:
quic-go-perf --server-address=<server ip>:<server port> --upload-bytes=1G
quic-go-perf --server-address=<server ip>:<server port> --download-bytes=1G
( "upload" means from emulator to host )

rethink 63868e7 + firestack aa87903 + Endpoint-Independent Mapping disabled
upload: 91.1 MiB/s
download: 89.7 MiB/s

rethink 63868e7 + firestack aa87903 + Endpoint-Independent Mapping enabled
upload: 67.3 MiB/s
download: 60.8 MiB/s

Lanius-collaris/gvisor-playground 7dce2ae
upload: 95.6 MiB/s
download: 89.8 MiB/s

Originally posted by @Lanius-collaris in #123

Most likely due to synchronization mechanics (chans/mutexes) in udpmux.go

@Lanius-collaris

@Lanius-collaris, I'm currently cleaning up the code and will share the commit details with you in a day or two.

@Lanius-collaris

@Lanius-collaris You can use this(0a885cf) commit to get the latest.

Update 1: prevent a nil pointer dereference

This patch improves inbound throughput, but may cause new bugs:

diff --git a/intra/udp.go b/intra/udp.go
index 1add4b0..581a550 100644
--- a/intra/udp.go
+++ b/intra/udp.go
@@ -223,7 +223,23 @@ func (h *udpHandler) proxy(gconn *netstack.GUDPConn, src, dst netip.AddrPort, dm
 	h.conntracker.Track(cid, gconn, remote)
 	core.Go("udp.forward: "+cid, func() {
 		defer h.conntracker.Untrack(cid)
-		forward(gconn, &rwext{remote}, h.smmch, h.done, smm)
+		if dmx == nil {
+			forward(gconn, &rwext{remote}, h.smmch, h.done, smm)
+		} else {
+			defer gconn.Close()
+			var buf [65536]byte
+			for {
+				gconn.SetReadDeadline(time.Now().Add(udptimeout))
+				n, err := gconn.Read(buf[:])
+				if err != nil {
+					break
+				}
+				_, err = remote.Write(buf[:n])
+				if err != nil {
+					break
+				}
+			}
+		}
 	})
 	return true // ok
 }
@@ -318,7 +334,7 @@ func (h *udpHandler) Connect(gconn *netstack.GUDPConn, src, target netip.AddrPor
 	for i, dstipp := range actualTargets {
 		selectedTarget = dstipp
 		if mux { // mux is not supported by all proxies (few like Exit, Base, WG support it)
-			pc, err = h.mux.associate(cid, pid, uid, src, selectedTarget, px.Dialer().Announce, vendor(dmx))
+			pc, err = h.mux.associate(cid, pid, uid, src, selectedTarget, px.Dialer().Announce, vendor(dmx), gconn)
 		} else {
 			pc, err = px.Dialer().Dial("udp", selectedTarget.String())
 		}
diff --git a/intra/udpmux.go b/intra/udpmux.go
index 0862e6f..06dfb3a 100644
--- a/intra/udpmux.go
+++ b/intra/udpmux.go
@@ -18,6 +18,8 @@ import (
 
 	"github.com/celzero/firestack/intra/core"
 	"github.com/celzero/firestack/intra/log"
+
+	"bytes"
 )
 
 // from: github.com/pion/transport/blob/03c807b/udp/conn.go
@@ -99,6 +101,8 @@ type demuxconn struct {
 	rt  *time.Ticker  // read deadline
 	wto time.Duration // write timeout
 	rto time.Duration // read timeout
+
+	realConn net.Conn
 }
 
 // slice is a byte slice v and its recycler free.
@@ -192,17 +196,25 @@ func (x *muxer) readers() {
 		_ = x.stop() // stop muxer
 	}()
 
+	bptr := core.AllocRegion(core.B65536)
+	b := *bptr
+	b = b[:cap(b)]
+	// todo: if panics are recovered above, free() may never be called
+	free := func() {
+		*bptr = b
+		core.Recycle(bptr)
+	}
+	defer free()
+
+	var t1 [4]byte
+	oldWho := &net.UDPAddr{
+		IP:   t1[:],
+		Port: 0,
+	}
+	var dst *demuxconn = nil
 	timeouterrors := 0
 	for {
-		bptr := core.AllocRegion(core.B65536)
-		b := *bptr
-		b = b[:cap(b)]
-		// todo: if panics are recovered above, free() may never be called
-		free := func() {
-			*bptr = b
-			core.Recycle(bptr)
-		}
-
+		x.mxconn.SetReadDeadline(time.Now().Add(udptimeout))
 		n, who, err := x.mxconn.ReadFrom(b)
 
 		x.stats.tx.Add(uint32(n)) // upload
@@ -210,7 +222,6 @@ func (x *muxer) readers() {
 		if timedout(err) {
 			timeouterrors++
 			if timeouterrors < maxtimeouterrors {
-				x.extend(time.Now().Add(udptimeout))
 				log.D("udp: mux: %s read timeout(%d): %v", x.cid, timeouterrors, err)
 				continue
 			} // else: err out
@@ -224,15 +235,21 @@ func (x *muxer) readers() {
 			continue
 		}
 
-		// may be an existing route or a new route
-		if dst := x.route(addr2netip(who), ingress); dst != nil {
-			select {
-			case dst.incomingCh <- &slice{v: b[:n], free: free}: // incomingCh is never closed
-			default: // dst probably closed, but not yet unrouted
-				log.W("udp: mux: %s read: drop(sz: %d); route to %s", x.cid, n, dst.raddr)
+		who2 := who.(*net.UDPAddr)
+		if !bytes.Equal(oldWho.IP, who2.IP) || oldWho.Port != who2.Port {
+			oldWho = who2
+			// may be an existing route or a new route
+			dst = x.route(addr2netip(who), ingress)
+		}
+		if dst != nil {
+			if dst.realConn == nil {
+				return
+			}
+			_, err = dst.realConn.Write(b[:n])
+			if err != nil {
+				return
 			}
-			log.V("udp: mux: %s read: n(%d) from %v <= %v; err %v", x.cid, n, dst, who, err)
-		} // else: ignore (who is invalid or x is closed)
+		}
 	}
 }
 
@@ -352,23 +369,10 @@ func (c *demuxconn) Read(p []byte) (int, error) {
 
 // Write implements core.UDPConn.Write
 func (c *demuxconn) Write(p []byte) (n int, err error) {
-	defer c.wt.Reset(c.wto)
-	sz := len(p)
-	select {
-	case <-c.wt.C:
-		log.W("udp: mux: %s demux: write: %v => %v; timeout (sz: %d)",
-			c.remux.id(), c.laddr, c.raddr, sz)
-		return 0, os.ErrDeadlineExceeded
-	case <-c.closed:
-		log.W("udp: mux: %s demux: write: %v => %v; closed (sz: %d)",
-			c.remux.id(), c.laddr, c.raddr, sz)
-		return 0, net.ErrClosed
-	default:
-		n, err = c.remux.sendto(p, c.raddr)
-		logev(err)("udp: mux: %s demux: write: %v => %v; done(sz: %d/%d); err? %v",
-			c.remux.id(), c.laddr, c.raddr, n, sz, err)
-		return n, err
-	}
+	n, err = c.remux.sendto(p, c.raddr)
+	logev(err)("udp: mux: %s demux: write: %v => %v; done(sz: %d/%d); err? %v",
+		c.remux.id(), c.laddr, c.raddr, n, len(p), err)
+	return n, err
 }
 
 // ReadFrom implements core.UDPConn.ReadFrom (unused)
@@ -491,7 +495,7 @@ func newMuxTable() *muxTable {
 	return &muxTable{t: make(map[netip.AddrPort]*muxer)}
 }
 
-func (e *muxTable) associate(cid, pid, uid string, src, dst netip.AddrPort, mk assocFn, v vendor) (_ net.Conn, err error) {
+func (e *muxTable) associate(cid, pid, uid string, src, dst netip.AddrPort, mk assocFn, v vendor, gconn net.Conn) (_ net.Conn, err error) {
 	e.Lock() // lock
 
 	var mxr *muxer
@@ -526,6 +530,23 @@ func (e *muxTable) associate(cid, pid, uid string, src, dst netip.AddrPort, mk a
 		return nil, errUidMismatch // return
 	}
 
+	if _, ok := mxr.routes[dst]; !ok {
+		mxr.routes[dst] = &demuxconn{
+			remux:      mxr,                          // muxer
+			laddr:      mxr.mxconn.LocalAddr(),       // listen addr
+			raddr:      net.UDPAddrFromAddrPort(dst), // remote addr
+			key:        dst,                          // key (same as raddr)
+			incomingCh: make(chan *slice),            // read from muxer
+			overflowCh: make(chan *slice),            // overflow from read
+			closed:     make(chan struct{}),          // always unbuffered
+			wt:         time.NewTicker(udptimeout),
+			rt:         time.NewTicker(udptimeout),
+			wto:        udptimeout,
+			rto:        udptimeout,
+			realConn:   gconn,
+		}
+	}
+
 	e.Unlock() // unlock
 	// do not hold e.lock on calls into mxr
 	c := mxr.route(dst, egress)

Sorry, the patch I posted yesterday includes a serious DoS vulnerability.

stack.Stack has a FindTransportEndpoint() method, it may be an alternative to muxer.route() , because tcpip.Endpoint created by udp.ForwarderRequest.CreateEndpoint() has been registered in stack.Stack
https://github.com/google/gvisor/blob/738e1d995f64/pkg/tcpip/transport/udp/forwarder.go#L88

stack.Stack has a FindTransportEndpoint() method, it may be an alternative to muxer.route()

As setup, to netstack, remote addrs are locally bound. And so, when a new route appears (from a remote ip:port udpmux is seeing for the first time on a unconnected socket), FindTransportEndpoint is unlikely to find any existing route (as none would exist bound to that remote ip:port), no?

As setup, to netstack, remote addrs are locally bound. And so, when a new route appears (from a remote ip:port udpmux is seeing for the first time on a unconnected socket), FindTransportEndpoint is unlikely to find any existing route (as none would exist bound to that remote ip:port), no?

Yes. But you can construct UDP packets manually, creating an Endpoint may need longer code.

But you can construct UDP packets manually, creating an Endpoint may need longer code.

True, but why won't gonet.DialUDP work 😭
So much simpler...

Channels could be replaced with https://github.com/kelindar/event
Or: Map/Queues with https://github.com/puzpuzpuz/xsync

Lanius-collaris@510f04b improves both outbound and inbound throughput. Note that when settings.EndpointIndependentMapping == false, c439342 is even slower than aa87903 , I can't understand. 🙃

gonet.DialUDP() doesn't seem to behave like net.ListenUDP()…… If raddr is nil, laddr.IP can't be an unspecified address, and laddr.Port can't be 0, otherwise gonet.UDPConn.WriteTo() will become unusable, this means, currently users can't send UDP packets if settings.EndpointIndependentMapping == true and these packets are routed to a wireguard interface.

Note that when settings.EndpointIndependentMapping == false, c439342 is even slower than aa87903 , I can't understand.

Don't mind, the reason is some early versions of my CLI frontend can't change the log level.

True, but why won't gonet.DialUDP work 😭
So much simpler...

currently users can't send UDP packets if settings.EndpointIndependentMapping == true and these packets are routed to a wireguard interface

Hm, I kind of get what you're saying, but WireGuard doesn't use/rely on udpmux? udpmux via handler.ReverseProxy(tcp / udp) doesn't wire back any packets into WireGuard.

If raddr is nil, laddr.IP can't be an unspecified address, and laddr.Port can't be 0, otherwise gonet.UDPConn.WriteTo() will become unusable

I don't follow. gonet.DialUDP is used for ingress only when udpmux's raddr (to) (gonet's laddr) is already known (while gonet's raddr is fixed)?

currently users can't send UDP packets if settings.EndpointIndependentMapping == true and these packets are routed to a wireguard interface

Hm, I kind of get what you're saying, but WireGuard doesn't use/rely on udpmux? udpmux via handler.ReverseProxy(tcp / udp) doesn't wire back any packets into WireGuard.

You didn't get what I was saying, I mean Dialer().Announce("udp","0.0.0.0:0") or Dialer().Announce("udp","[::]:0") can't be convert into gonet.DialUDP() directly, you must choose local IP and port for it explicitly.

firestack/intra/udp.go

Line 310 in ebf2895

    
           pc, err = h.mux.associate(cid, pxid, uid, src, selectedTarget, px.Dialer().Announce, vendor(dmx))

firestack/intra/udpmux.go

Lines 616 to 628 in ebf2895

    
           proto := "udp" 
        
           anyaddr := anyaddr6 
        
           anyport := uint16(0) 
        
           if dst.Addr().Is4() && !dialers.Use6() { 
        
           	proto = "udp4" 
        
           	anyaddr = anyaddr4 
        
           } 
        
           anyaddrport := netip.AddrPortFrom(anyaddr, anyport) 
        
           if settings.PortForward.Load() || ipn.Remote(pid) { 
        
           	anyaddrport = netip.AddrPortFrom(anyaddr, src.Port()) 
        
           } 
        
           pc, err := mk(proto, anyaddrport.String())

Gotcha. The comment in the code says, src will be assigned if left nil (which it is when listening on unspecified ip:port) iff Netstack's NIC addrs are set...

firestack/intra/ipn/wgnet.go

Lines 253 to 256 in 5485aa5

    
           // if src is non-nil, addrs are acquired on wgnic; 
        
           // ep.Bind => ep.BindAndThen => ep.net.BindAndThen => ep.checkV4Mapped 
        
           // github.com/google/gvisor/blob/932d9dc6/pkg/tcpip/stack/addressable_endpoint_state.go#L644 
        
           return gonet.DialUDP(tnet.stack, src, dst, protocol)

Possible I was mistaken when I wrote that comment.

If raddr is nil, laddr.IP can't be an unspecified address, and laddr.Port can't be 0, otherwise gonet.UDPConn.WriteTo() will become unusable

Sorry, in fact gvisor can handle them, maybe I did something wrong when comparing the ipn module of firestack with the tun/netstack module of wireguard-go.

Don't mind, the reason is some early versions of my CLI frontend can't change the log level.

The default log level is Info, which shouldn't (ideally) hurt bandwidth measures that much.

Is the CLI you're talking about in this PR? #189

tcp,udp: impl missing Reset()

Navigation Menu

Repository navigation

Endpoint independent mapping slows down bandwidth #129

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	proto := "udp"
	anyaddr := anyaddr6
	anyport := uint16(0)
	if dst.Addr().Is4() && !dialers.Use6() {
	proto = "udp4"
	anyaddr = anyaddr4
	}
	anyaddrport := netip.AddrPortFrom(anyaddr, anyport)
	if settings.PortForward.Load() \|\| ipn.Remote(pid) {
	anyaddrport = netip.AddrPortFrom(anyaddr, src.Port())
	}

	pc, err := mk(proto, anyaddrport.String())

	// if src is non-nil, addrs are acquired on wgnic;
	// ep.Bind => ep.BindAndThen => ep.net.BindAndThen => ep.checkV4Mapped
	// github.com/google/gvisor/blob/932d9dc6/pkg/tcpip/stack/addressable_endpoint_state.go#L644
	return gonet.DialUDP(tnet.stack, src, dst, protocol)

Endpoint independent mapping slows down bandwidth #129

Description

Activity

hussainmohd-a commented on Feb 9, 2025

hussainmohd-a commented on Feb 15, 2025

Lanius-collaris commented on Feb 19, 2025

Lanius-collaris commented on Feb 20, 2025

Lanius-collaris commented on Mar 21, 2025

ignoramous commented on Mar 25, 2025

Lanius-collaris commented on Mar 27, 2025

ignoramous commented on Mar 27, 2025

ignoramous commented on Jul 3, 2025

Lanius-collaris commented on Jul 8, 2025

Lanius-collaris commented on Oct 17, 2025

ignoramous commented on Oct 17, 2025

Lanius-collaris commented on Oct 17, 2025

ignoramous commented on Oct 18, 2025

Lanius-collaris commented on Oct 19, 2025

ignoramous commented on Oct 22, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions