personal_interest_mirrors/nebula
1
0
Fork 0
mirror of https://github.com/slackhq/nebula.git synced 2026-08-15 03:07:01 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

spread N routines over n lanes

Browse Source
This commit is contained in:
JackDoan
2026-07-21 14:24:20 -05:00
parent 405a78f415
commit 7043baeb34
6 changed files with 96 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/config.yml
+3 -1
View File
@@ -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:
handshake_manager.go
+1 -1
View File
@@ -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
inside.go
+4 -4
View File
@@ -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
interface.go
+40 -17
View File
@@ -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)
lanes_test.go
+47 -13
View File
@@ -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) {
main.go
+1
View File
@@ -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),