spread N routines over n lanes

(cherry picked from commit 7043baeb3440b5ed977a54ee722287af9a6f4136)

examples/config.yml

@@ -211,7 +211,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:

handshake_manager.go

@@ -750,7 +750,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

inside.go

@@ -110,7 +110,7 @@ func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Parse
 
 	dropReason := f.firewall.Drop(fwPacket.Packet, 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) {
@@ -158,7 +158,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

interface.go

@@ -45,7 +45,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
@@ -91,6 +95,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
@@ -222,6 +227,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.MultiCoalescer, c.routines),
@@ -444,16 +450,19 @@ func (f *Interface) pinThisThread(i int) {
 	}
 }
 
-// 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 {
@@ -466,11 +475,24 @@ 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(f *Interface, i int) {
+func (tx *txQueue) flush(f *Interface) {
 	if tx.base != tx.lane {
 		f.flushSendBatch(tx.base, 0)
 	}
-	f.flushSendBatch(tx.lane, i)
+	f.flushSendBatch(tx.lane, 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) {
@@ -482,9 +504,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.ParsedPacket{}
@@ -509,10 +532,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, i)
+				tx.flush(f)
 			}
 		}
-		tx.flush(f, i)
+		tx.flush(f)
 	}
 
 	f.l.Debug("overlay reader is done", "reader", i)

lanes_test.go

@@ -343,19 +343,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, 1)
+	ifce.sendInsideMessage(base, pkt, nb, tx1)
+	tx1.flush(ifce)
 	require.Len(t, laneWriter.bufs, 1)
 	require.Empty(t, baseWriter.bufs)
 	assert.Equal(t, lane.GetRemote(), laneWriter.dsts[0])
@@ -364,9 +369,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, 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)
+	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)
 	require.Len(t, baseWriter.bufs, 1)
 	assert.Equal(t, base.GetRemote(), baseWriter.dsts[0])
 	require.NoError(t, h.Parse(baseWriter.bufs[0]))
@@ -374,10 +388,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, 1)
+	ifce.sendInsideMessage(base, pkt, nb, tx1)
+	tx1.flush(ifce)
 	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) {

main.go

@@ -376,6 +376,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),