Repository navigation
Endpoint independent mapping slows down bandwidth #129
Description
Activity
@Lanius-collaris, I'm currently cleaning up the code and will share the commit details with you in a day or two.
Reacted by ignoramous@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)Reacted by ignoramousSorry, the patch I posted yesterday includes a serious DoS vulnerability.
Reacted by ignoramousstack.Stackhas aFindTransportEndpoint()method, it may be an alternative tomuxer.route(), becausetcpip.Endpointcreated byudp.ForwarderRequest.CreateEndpoint()has been registered instack.Stack
https://github.com/google/gvisor/blob/738e1d995f64/pkg/tcpip/transport/udp/forwarder.go#L88Reacted by ignoramousstack.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. 🙃Reacted by ignoramousgonet.DialUDP()doesn't seem to behave likenet.ListenUDP()…… Ifraddris nil,laddr.IPcan't be an unspecified address, andladdr.Portcan't be 0, otherwisegonet.UDPConn.WriteTo()will become unusable, this means, currently users can't send UDP packets ifsettings.EndpointIndependentMapping == trueand 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.
Reacted by ignoramousTrue, 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 meanDialer().Announce("udp","0.0.0.0:0")orDialer().Announce("udp","[::]:0")can't be convert intogonet.DialUDP()directly, you must choose local IP and port for it explicitly.
Line 310 in ebf2895
Lines 616 to 628 in ebf2895
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...Lines 253 to 256 in 5485aa5
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
ipnmodule of firestack with thetun/netstackmodule of wireguard-go.Reacted by ignoramousDon'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
- added a commit that references this issue
on Aug 1, 2026
Originally posted by @Lanius-collaris in #123
Most likely due to synchronization mechanics (chans/mutexes) in
udpmux.go