From c4deb5fc1c3c4128352a7d1ea264b35785743233 Mon Sep 17 00:00:00 2001 From: JackDoan Date: Thu, 14 May 2026 13:40:40 -0500 Subject: [PATCH] checkpt --- inside.go | 16 ++++++++-------- interface.go | 6 +++--- overlay/batch/batch.go | 28 ---------------------------- overlay/batch/rx_batch.go | 12 ++++++++++++ overlay/batch/tx_batch.go | 11 +++-------- overlay/batch/tx_batch_test.go | 8 +++++--- overlay/tio/segment.go | 12 ------------ wire/wire.go | 6 ++++++ 8 files changed, 37 insertions(+), 62 deletions(-) delete mode 100644 overlay/batch/batch.go create mode 100644 overlay/batch/rx_batch.go delete mode 100644 overlay/tio/segment.go diff --git a/inside.go b/inside.go index 223a1c02..319d887e 100644 --- a/inside.go +++ b/inside.go @@ -11,11 +11,11 @@ import ( "github.com/slackhq/nebula/iputil" "github.com/slackhq/nebula/noiseutil" "github.com/slackhq/nebula/overlay/batch" - "github.com/slackhq/nebula/overlay/tio" "github.com/slackhq/nebula/routing" + "github.com/slackhq/nebula/wire" ) -func (f *Interface) consumeInsidePacket(pkt wire.Packet, fwPacket *firewall.Packet, nb []byte, sendBatch batch.TxBatcher, rejectBuf []byte, q int, localCache firewall.ConntrackCache) { +func (f *Interface) consumeInsidePacket(pkt wire.TunPacket, fwPacket *firewall.Packet, nb []byte, sendBatch *batch.SendBatch, rejectBuf []byte, q int, localCache firewall.ConntrackCache) { // pkt.Bytes is either one IP datagram (GSO zero) or a TSO/USO // superpacket. In both cases the L3+L4 headers at the start describe // the same 5-tuple every segment will share, so a single newPacket / @@ -45,7 +45,7 @@ func (f *Interface) consumeInsidePacket(pkt wire.Packet, fwPacket *firewall.Pack // routes packets from the Nebula addr to the Nebula addr through the Nebula // TUN device. if immediatelyForwardToSelf { - err := tio.SegmentSuperpacket(pkt, func(seg []byte) error { + err := pkt.PerSegment(func(seg []byte) error { _, werr := f.readers[q].Write(seg) return werr }) @@ -67,7 +67,7 @@ func (f *Interface) consumeInsidePacket(pkt wire.Packet, fwPacket *firewall.Pack // borrowed: SegmentSuperpacket builds each segment in the kernel-supplied pkt // bytes underneath. cachePacket explicitly copies its argument (handshake_manager.go cachePacket), // so retaining segments past the loop is safe. - err := tio.SegmentSuperpacket(pkt, func(seg []byte) error { + err := pkt.PerSegment(func(seg []byte) error { hh.cachePacket(f.l, header.Message, 0, seg, f.sendMessageNow, f.cachedPacketMetrics) return nil }) @@ -139,10 +139,10 @@ func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, s // segment of a TSO/USO superpacket) into the caller's batch slot for // later sendmmsg flush. Segmentation is fused with encryption here so the // kernel-supplied superpacket bytes never get written into a separate -// scratch arena: SegmentSuperpacket builds each segment's plaintext in +// scratch arena: PerSegment builds each segment's plaintext in // segScratch[:segLen] in turn, and we encrypt directly into a fresh // SendBatch slot. -func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, sendBatch batch.TxBatcher, rejectBuf []byte, q int) { +func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt wire.TunPacket, nb []byte, sendBatch *batch.SendBatch) { ci := hostinfo.ConnectionState if ci.eKey == nil { return @@ -183,7 +183,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return } - err = tio.SegmentSuperpacket(pkt, func(seg []byte) error { + err = pkt.PerSegment(func(seg []byte) error { //relay header + header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305) + relay tag scratch := sendBatch.Reserve(header.Len + header.Len + len(seg) + 16 + 16) @@ -212,7 +212,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b return } - err := tio.SegmentSuperpacket(pkt, func(seg []byte) error { + err := pkt.PerSegment(func(seg []byte) error { // header + plaintext + AEAD tag (16 bytes for both AES-GCM and ChaCha20-Poly1305) scratch := sendBatch.Reserve(header.Len + len(seg) + 16) diff --git a/interface.go b/interface.go index d903a8e7..100500d5 100644 --- a/interface.go +++ b/interface.go @@ -289,10 +289,10 @@ func (f *Interface) activate() error { // is on, everything else (and either lane disabled) falls // through to passthrough so non-IP / non-TCP-UDP traffic still // reaches the TUN. - arena := batch.NewArena(max(f.batchSize, 1) * 65535) + arena := util.NewArena(max(f.batchSize, 1) * 65535) f.batchers[i] = batch.NewMultiCoalescer(f.readers[i], f.l, arena, caps.TSO, caps.USO) } else { - arena := batch.NewArena(max(f.batchSize, 1) * udp.MTU) + arena := util.NewArena(max(f.batchSize, 1) * udp.MTU) f.batchers[i] = batch.NewPassthrough(f.readers[i], f.batchSize, arena) } } @@ -389,7 +389,7 @@ func (f *Interface) listenIn(reader tio.Queue, q int) { rejectBuf := make([]byte, mtu) arenaSize := batch.SendBatchCap * (udp.MTU + 32) - sb := batch.NewSendBatch(f.writers[q], batch.SendBatchCap, arenaSize) + sb := batch.NewSendBatch(f.writers[q], batch.SendBatchCap, util.NewArena(arenaSize)) fwPacket := &firewall.Packet{} nb := make([]byte, 12, 12) diff --git a/overlay/batch/batch.go b/overlay/batch/batch.go deleted file mode 100644 index d171d136..00000000 --- a/overlay/batch/batch.go +++ /dev/null @@ -1,28 +0,0 @@ -package batch - -import "net/netip" - -type RxBatcher interface { - // Reserve creates a pkt to borrow - Reserve(sz int) []byte - // Commit borrows pkt. The caller must keep pkt valid until the next Flush - Commit(pkt []byte) error - // Flush emits every queued packet in arrival order. Returns the - // first error observed; keeps draining so one bad packet doesn't hold up - // the rest. After Flush returns, borrowed payload slices may be recycled. - Flush() error -} - -type TxBatcher interface { - // Reserve creates a pkt to borrow - Reserve(sz int) []byte - // Commit borrows pkt and records its destination plus the 2-bit - // IP-level ECN codepoint to set on the outer (carrier) header. The - // caller must keep pkt valid until the next Flush. Pass 0 (Not-ECT) - // to leave the outer ECN field unset. - Commit(pkt []byte, dst netip.AddrPort, outerECN byte) - // Flush emits every queued packet via the underlying batch writer in - // arrival order. Returns an errors.Join of one or more errors. After Flush returns, - // borrowed payload slices may be recycled. - Flush() error -} diff --git a/overlay/batch/rx_batch.go b/overlay/batch/rx_batch.go new file mode 100644 index 00000000..a58a39f0 --- /dev/null +++ b/overlay/batch/rx_batch.go @@ -0,0 +1,12 @@ +package batch + +type RxBatcher interface { + // Reserve creates a pkt to borrow + Reserve(sz int) []byte + // Commit borrows pkt. The caller must keep pkt valid until the next Flush + Commit(pkt []byte) error + // Flush emits every queued packet in arrival order. Returns the + // first error observed; keeps draining so one bad packet doesn't hold up + // the rest. After Flush returns, borrowed payload slices may be recycled. + Flush() error +} diff --git a/overlay/batch/tx_batch.go b/overlay/batch/tx_batch.go index 599bc306..dcf928b9 100644 --- a/overlay/batch/tx_batch.go +++ b/overlay/batch/tx_batch.go @@ -3,16 +3,11 @@ package batch import ( "net/netip" - "github.com/slackhq/nebula/udp" + "github.com/slackhq/nebula/util" ) const SendBatchCap = 128 -// DefaultSendBatchArenaCap is the recommended arena capacity for a -// standalone SendBatch: 128 slots × (udp.MTU + 32) ≈ 1.1 MiB. The +32 covers -// the nebula header + AEAD tag tacked onto each plaintext segment. -const DefaultSendBatchArenaCap = SendBatchCap * (udp.MTU + 32) - // batchWriter is the minimal subset of udp.Conn needed by SendBatch to flush. type batchWriter interface { WriteBatch(bufs [][]byte, addrs []netip.AddrPort, outerECNs []byte) error @@ -27,11 +22,11 @@ type SendBatch struct { bufs [][]byte dsts []netip.AddrPort ecns []byte - arena *Arena + arena *util.Arena } // NewSendBatch makes a SendBatch with batchCap slots backed by arena. -func NewSendBatch(out batchWriter, batchCap int, arena *Arena) *SendBatch { +func NewSendBatch(out batchWriter, batchCap int, arena *util.Arena) *SendBatch { return &SendBatch{ out: out, bufs: make([][]byte, 0, batchCap), diff --git a/overlay/batch/tx_batch_test.go b/overlay/batch/tx_batch_test.go index 59b06e58..f8e8bd51 100644 --- a/overlay/batch/tx_batch_test.go +++ b/overlay/batch/tx_batch_test.go @@ -3,6 +3,8 @@ package batch import ( "net/netip" "testing" + + "github.com/slackhq/nebula/util" ) type fakeBatchWriter struct { @@ -27,7 +29,7 @@ func (w *fakeBatchWriter) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns func TestSendBatchReserveCommitFlush(t *testing.T) { fw := &fakeBatchWriter{} - b := NewSendBatch(fw, 4, NewArena(32)) + b := NewSendBatch(fw, 4, util.NewArena(32)) ap := netip.MustParseAddrPort("10.0.0.1:4242") for i := 0; i < 4; i++ { @@ -71,7 +73,7 @@ func TestSendBatchReserveCommitFlush(t *testing.T) { func TestSendBatchSlotsDoNotOverlap(t *testing.T) { fw := &fakeBatchWriter{} - b := NewSendBatch(fw, 3, NewArena(8)) + b := NewSendBatch(fw, 3, util.NewArena(8)) ap := netip.MustParseAddrPort("10.0.0.1:80") for i := 0; i < 3; i++ { @@ -93,7 +95,7 @@ func TestSendBatchSlotsDoNotOverlap(t *testing.T) { func TestSendBatchGrowPreservesCommitted(t *testing.T) { fw := &fakeBatchWriter{} // Tiny initial backing forces a grow on the second Reserve. - b := NewSendBatch(fw, 1, NewArena(4)) + b := NewSendBatch(fw, 1, util.NewArena(4)) ap := netip.MustParseAddrPort("10.0.0.1:80") s1 := b.Reserve(4) diff --git a/overlay/tio/segment.go b/overlay/tio/segment.go deleted file mode 100644 index 67648ad2..00000000 --- a/overlay/tio/segment.go +++ /dev/null @@ -1,12 +0,0 @@ -package tio - -import "fmt" - -// SegmentSuperpacket invokes fn once per segment of pkt. -// This is a stub implementation that does not actually support segmentation -func SegmentSuperpacket(pkt Packet, fn func(seg []byte) error) error { - if pkt.GSO.IsSuperpacket() { - return fmt.Errorf("tio: GSO superpacket on platform without segmentation support") - } - return fn(pkt.Bytes) -} diff --git a/wire/wire.go b/wire/wire.go index acc20dd8..9f580fd8 100644 --- a/wire/wire.go +++ b/wire/wire.go @@ -9,3 +9,9 @@ type TunPacket struct { // Fields in Meta should be as portable/platform-agnostic as possible. Meta struct{} } + +// PerSegment invokes fn once per segment of pkt. +// This is a stub implementation that does not actually support segmentation +func (t *TunPacket) PerSegment(fn func(seg []byte) error) error { + return fn(t.Bytes) +}