diff --git a/examples/config.yml b/examples/config.yml index a6aeda74..c8201ffd 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -198,7 +198,9 @@ listen: #enabled: true # How many lanes to run, counting the base tunnel as lane 0. 0 (default) # means one per routine. Lowering this bounds how many extra tunnels each - # peer pair maintains; routines without a lane use the base tunnel. + # peer pair maintains (useful on a big server with many peers); routines + # beyond the lane count share the configured lanes round-robin, so TX still + # spreads across `lanes` underlay flows rather than piling onto the base. #lanes: 0 punchy: diff --git a/handshake_manager.go b/handshake_manager.go index 2b2b0685..93645884 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -744,7 +744,7 @@ func (hm *HandshakeManager) maybeAllocLaneState(hostinfo *HostInfo, result *hand if len(hm.f.myVpnAddrs) > 0 && len(hostinfo.vpnAddrs) > 0 { offset = lanePortOffset(hm.f.myVpnAddrs[0], hostinfo.vpnAddrs[0], uint16(peerPorts)) } - hostinfo.lanes = newLaneState(hm.f.routines, uint16(peerPorts), uint16(result.PeerBasePort), offset) + hostinfo.lanes = newLaneState(hm.config.laneCount, uint16(peerPorts), uint16(result.PeerBasePort), offset) } // EnsureLanes starts lane handshakes for every empty, non-pending, retry-due diff --git a/inside.go b/inside.go index 49b92305..a6ccbd6a 100644 --- a/inside.go +++ b/inside.go @@ -106,7 +106,7 @@ func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Packe dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache) if dropReason == nil { - f.sendInsideMessage(hostinfo, pkt, nb, tx, q) + f.sendInsideMessage(hostinfo, pkt, nb, tx) } else { f.rejectInside(packet, rejectBuf, q) if f.l.Enabled(context.Background(), slog.LevelDebug) { @@ -157,7 +157,7 @@ func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, s // when routine q has an established lane to this peer, the direct path swaps // to the lane's session and socket below. Relay and base traffic stays on // tx.base (socket 0). -func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, tx *txQueue, q int) { +func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, tx *txQueue) { ci := hostinfo.ConnectionState if ci.eKey == nil { return @@ -233,8 +233,8 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b // The pointer is only published once the lane's ConnectionState is fully // populated, so a non-nil Load is always usable. On lane death the slot // CAS-clears and traffic falls back to the base tunnel instantly. - if ls := hostinfo.lanes; ls != nil && q < len(ls.txLanes) { - if lane := ls.txLanes[q].Load(); lane != nil { + if ls := hostinfo.lanes; ls != nil && tx.laneSlot < len(ls.txLanes) { + if lane := ls.txLanes[tx.laneSlot].Load(); lane != nil { if lci := lane.ConnectionState; lci != nil && lci.eKey != nil { hostinfo = lane ci = lci diff --git a/interface.go b/interface.go index 1783179c..85159806 100644 --- a/interface.go +++ b/interface.go @@ -44,7 +44,11 @@ type InterfaceConfig struct { routines int // Multiport means writers[i] is bound to listen.port+i (not a shared // SO_REUSEPORT port) and lane tunnels are negotiated with capable peers. - Multiport bool + Multiport bool + // LaneCount is the number of lanes counting the base tunnel as lane 0 + // (multiport.lanes, clamped to routines). Routines at or beyond it share + // the configured lanes round-robin. + LaneCount int MessageMetrics *MessageMetrics version string relayManager *relayManager @@ -90,6 +94,7 @@ type Interface struct { dropMulticast bool routines int multiport bool + laneCount int disconnectInvalid atomic.Bool closed atomic.Bool // cpuAffinity, when non-empty, names the CPUs each TUN reader goroutine @@ -227,6 +232,7 @@ func NewInterface(ctx context.Context, c *InterfaceConfig) (*Interface, error) { dropMulticast: c.DropMulticast, routines: c.routines, multiport: c.Multiport, + laneCount: c.LaneCount, version: c.version, writers: make([]udp.Conn, c.routines), batchers: make([]batch.RxBatcher, c.routines), @@ -407,16 +413,19 @@ func (f *Interface) listenOut(i int) { f.l.Debug("underlay reader is done", "reader", i) } -// txQueue is the per-routine TX state owned by one listenIn goroutine. lane -// is bound to the routine's own socket and carries lane-tunnel data; base is -// bound to socket 0 and carries base-tunnel and relay data, which must keep -// the base source port (a vanilla peer would otherwise see per-routine source -// ports and roam-thrash). The two alias when multiport is off or on routine 0. -// Concurrent sendmmsg on the shared socket-0 fd is safe: a flow is pinned to -// one routine by tun steering, so per-flow wire order still holds. +// txQueue is the per-routine TX state owned by one listenIn goroutine. +// laneSlot is the lane this routine's traffic rides (laneSlotFor); lane is +// bound to that slot's socket and carries lane-tunnel data; base is bound to +// socket 0 and carries base-tunnel and relay data, which must keep the base +// source port (a vanilla peer would otherwise see per-routine source ports +// and roam-thrash). The two alias when multiport is off or laneSlot is 0. +// Concurrent sendmmsg on a shared fd (socket 0, or a lane socket shared by +// overflow routines) is safe: a flow is pinned to one routine by tun +// steering, so per-flow wire order still holds. type txQueue struct { - lane *batch.SendBatch - base *batch.SendBatch + laneSlot int + lane *batch.SendBatch + base *batch.SendBatch } func (tx *txQueue) full() bool { @@ -429,17 +438,30 @@ func (tx *txQueue) full() bool { // flush drains base before lane so that when a flow moves from the base // tunnel onto a freshly established lane mid-window, its packets still leave // this host in encryption order. -func (tx *txQueue) flush(l *slog.Logger, i int) { +func (tx *txQueue) flush(l *slog.Logger) { if tx.base != tx.lane { if err := tx.base.Flush(); err != nil { l.Error("Failed to write outgoing batch", "error", err, "writer", 0) } } if err := tx.lane.Flush(); err != nil { - l.Error("Failed to write outgoing batch", "error", err, "writer", i) + l.Error("Failed to write outgoing batch", "error", err, "writer", tx.laneSlot) } } +// laneSlotFor maps a routine index to the lane its traffic rides. When +// multiport.lanes is below routines, overflow routines share the configured +// lanes round-robin instead of all falling back onto the base tunnel's +// single underlay flow. Sharers use the lane's own socket and 4-tuple, so +// this is vanilla-style same-flow sharing: no cross-path replay skew, and +// per-flow ordering still holds (a flow stays pinned to one routine). +func (f *Interface) laneSlotFor(i int) int { + if f.multiport && f.laneCount > 0 { + return i % f.laneCount + } + return i +} + func (f *Interface) listenIn(queue tio.Queue, i int) { // Pinning this thread (and goroutine) to a single CPU keeps every sendmmsg from this goroutine going through the // same TX ring on the nic, so the wire sees per-flow order. Skip entirely when tun.pin_threads is false. @@ -464,9 +486,10 @@ func (f *Interface) listenIn(queue tio.Queue, i int) { rejectBuf := make([]byte, mtu) arenaSize := batch.SendBatchCap * (udp.MTU + 32) - sb := batch.NewSendBatch(f.writers[i], batch.SendBatchCap, arenaSize) - tx := &txQueue{lane: sb, base: sb} - if f.multiport && i != 0 { + laneSlot := f.laneSlotFor(i) + sb := batch.NewSendBatch(f.writers[laneSlot], batch.SendBatchCap, arenaSize) + tx := &txQueue{laneSlot: laneSlot, lane: sb, base: sb} + if f.multiport && laneSlot != 0 { tx.base = batch.NewSendBatch(f.writers[0], batch.SendBatchCap, arenaSize) } fwPacket := &firewall.Packet{} @@ -491,10 +514,10 @@ func (f *Interface) listenIn(queue tio.Queue, i int) { // accumulated so the first packets of a deep read drain // hit the wire while the rest are still being encrypted. if tx.full() { - tx.flush(f.l, i) + tx.flush(f.l) } } - tx.flush(f.l, i) + tx.flush(f.l) } f.l.Debug("overlay reader is done", "reader", i) diff --git a/lanes_test.go b/lanes_test.go index 128888f9..b7722953 100644 --- a/lanes_test.go +++ b/lanes_test.go @@ -340,19 +340,24 @@ func TestSendInsideMessageLaneSwap(t *testing.T) { baseWriter := &recordingBatchWriter{} laneWriter := &recordingBatchWriter{} - tx := &txQueue{ - base: batch.NewSendBatch(baseWriter, batch.SendBatchCap, 1<<16), - lane: batch.NewSendBatch(laneWriter, batch.SendBatchCap, 1<<16), + newTx := func(laneSlot int) *txQueue { + return &txQueue{ + laneSlot: laneSlot, + base: batch.NewSendBatch(baseWriter, batch.SendBatchCap, 1<<16), + lane: batch.NewSendBatch(laneWriter, batch.SendBatchCap, 1<<16), + } } + tx1 := newTx(1) + tx2 := newTx(2) pkt := tio.Packet{Bytes: []byte{0x45, 0, 0, 4, 1, 2, 3, 4}} nb := make([]byte, 12) - // With the lane published, routine 1's traffic uses the lane session and - // the lane batch. + // With the lane published, slot-1 traffic uses the lane session and the + // lane batch. base.lanes.txLanes[1].Store(lane) - ifce.sendInsideMessage(base, pkt, nb, tx, 1) - tx.flush(ifce.l, 1) + ifce.sendInsideMessage(base, pkt, nb, tx1) + tx1.flush(ifce.l) require.Len(t, laneWriter.bufs, 1) require.Empty(t, baseWriter.bufs) assert.Equal(t, lane.GetRemote(), laneWriter.dsts[0]) @@ -361,9 +366,18 @@ func TestSendInsideMessageLaneSwap(t *testing.T) { require.NoError(t, h.Parse(laneWriter.bufs[0])) assert.Equal(t, lane.remoteIndexId, h.RemoteIndex) - // Routine 2 has no lane: base tunnel, base batch. - ifce.sendInsideMessage(base, pkt, nb, tx, 2) - tx.flush(ifce.l, 1) + // An overflow routine sharing slot 1 (multiport.lanes < routines) rides + // the same lane session. + tx1b := newTx(1) + ifce.sendInsideMessage(base, pkt, nb, tx1b) + tx1b.flush(ifce.l) + require.Len(t, laneWriter.bufs, 2) + require.NoError(t, h.Parse(laneWriter.bufs[1])) + assert.Equal(t, lane.remoteIndexId, h.RemoteIndex) + + // Slot 2 has no lane: base tunnel, base batch. + ifce.sendInsideMessage(base, pkt, nb, tx2) + tx2.flush(ifce.l) require.Len(t, baseWriter.bufs, 1) assert.Equal(t, base.GetRemote(), baseWriter.dsts[0]) require.NoError(t, h.Parse(baseWriter.bufs[0])) @@ -371,10 +385,30 @@ func TestSendInsideMessageLaneSwap(t *testing.T) { // Lane death: slot cleared, instant fallback to base. base.lanes.txLanes[1].Store(nil) - ifce.sendInsideMessage(base, pkt, nb, tx, 1) - tx.flush(ifce.l, 1) + ifce.sendInsideMessage(base, pkt, nb, tx1) + tx1.flush(ifce.l) require.Len(t, baseWriter.bufs, 2) - require.Len(t, laneWriter.bufs, 1) + require.Len(t, laneWriter.bufs, 2) +} + +func TestLaneSlotFor(t *testing.T) { + // Overflow routines wrap onto the configured lanes round-robin. + f := &Interface{multiport: true, laneCount: 2} + for i, want := range []int{0, 1, 0, 1, 0, 1} { + assert.Equal(t, want, f.laneSlotFor(i), "routine %d", i) + } + + // Full lane count: identity mapping, one lane per routine. + f = &Interface{multiport: true, laneCount: 4} + for i := range 4 { + assert.Equal(t, i, f.laneSlotFor(i), "routine %d", i) + } + + // Multiport off: identity, each routine keeps its own writer. + f = &Interface{multiport: false, laneCount: 0} + for i := range 4 { + assert.Equal(t, i, f.laneSlotFor(i), "routine %d", i) + } } func TestCompleteLaneResponder(t *testing.T) { diff --git a/main.go b/main.go index fbcc41b8..a1e80fbd 100644 --- a/main.go +++ b/main.go @@ -310,6 +310,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev DropMulticast: c.GetBool("tun.drop_multicast", false), routines: routines, Multiport: multiport, + LaneCount: handshakeConfig.laneCount, MessageMetrics: messageMetrics, version: buildVersion, relayManager: NewRelayManager(ctx, l, hostMap, c),