Endpoint independent mapping slows down bandwidth

[ignoramous]

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

[hussainmohd-a]

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

[hussainmohd-a]

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

[Lanius-collaris]

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)

[Lanius-collaris]

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

[Lanius-collaris]

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

[ignoramous]

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?

[Lanius-collaris]

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.

[ignoramous]

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...

[ignoramous]

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

[Lanius-collaris]

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. 🙃

[Lanius-collaris]

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.

[ignoramous]

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)?

[Lanius-collaris]

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())

[ignoramous]

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.