mirror of
https://github.com/slackhq/nebula.git
synced 2026-11-15 12:36:52 +01:00
Ryan HuberandClaude Opus 5.5
64bec56404
On darwin, listenIn encrypts into one SendBatch while a sender goroutine writes the other (batch.SendPipeline). The sender takes whatever built up during its last write, so writes overlap tun reads and packets queued behind a write leave in one WriteBatch, which a connected socket sends as one sendmsg_x. Two batches bound memory; a full producer waits for the swap rather than dropping. One producer and one sender keep per-peer order equal to message-counter order. Everywhere else the branch is a false constant and listenIn is unchanged. -tags send_pipeline builds the pipeline on any platform, so its tests and the e2e suite can run through it on linux. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
59 lines
2.1 KiB
Go
59 lines
2.1 KiB
Go
//go:build darwin || send_pipeline
|
|
|
|
package nebula
|
|
|
|
import (
|
|
"github.com/slackhq/nebula/firewall"
|
|
"github.com/slackhq/nebula/overlay/batch"
|
|
"github.com/slackhq/nebula/overlay/tio"
|
|
"github.com/slackhq/nebula/udp"
|
|
)
|
|
|
|
// sendPipelinePlatform lets listenIn take the pipelined send path. It is on for darwin, and for any platform built
|
|
// with -tags send_pipeline so the pipeline's tests (and the e2e suite through it) run on linux too.
|
|
const sendPipelinePlatform = true
|
|
|
|
// listenInPipelined is f.listenIn with the underlay writes moved to a sender goroutine (batch.SendPipeline). This
|
|
// goroutine still reads the tun, runs the firewall and encrypts, straight into the pipeline's batch. The sender
|
|
// writes whatever has built up each time its previous write returns, so under load the packets read while one
|
|
// write is in the kernel leave together in the next one, and the next tun read overlaps the write.
|
|
func listenInPipelined(f *Interface, queue tio.Queue, i int) {
|
|
if f.pinThreads {
|
|
f.pinThisThread(i)
|
|
}
|
|
|
|
rejectBuf := make([]byte, mtu)
|
|
arenaSize := batch.SendBatchCap * (udp.MTU + 32)
|
|
onWrite := func(queued, written int, err error) {
|
|
f.accountSendBatch(queued, written, err, i)
|
|
}
|
|
p := batch.NewSendPipeline(f.writers[i], batch.SendBatchCap, arenaSize, udp.MaxWriteBatch, onWrite)
|
|
defer p.Close()
|
|
fwPacket := &firewall.ParsedPacket{}
|
|
nb := make([]byte, 12, 12)
|
|
|
|
conntrackCache := firewall.NewConntrackCacheTicker(f.ctx, f.l, f.conntrackCacheTimeout)
|
|
|
|
for {
|
|
pkts, err := queue.Read()
|
|
if err != nil {
|
|
// Same shutdown noise handling as listenOut
|
|
if !f.closed.Load() && f.ctx.Err() == nil {
|
|
f.l.Error("Error while reading outbound packet, closing", "error", err, "reader", i)
|
|
f.onFatal(err)
|
|
}
|
|
break
|
|
}
|
|
|
|
for j, pkt := range pkts {
|
|
// One Begin/End per packet holds the batch only briefly, so a sender that finishes a write mid-read
|
|
// takes what is ready then. An idle sender is woken once, at the end of the read, to take it whole.
|
|
sb := p.Begin()
|
|
f.consumeInsidePacket(pkt, fwPacket, nb, sb, rejectBuf, i, conntrackCache.Get())
|
|
p.End(j == len(pkts)-1)
|
|
}
|
|
}
|
|
|
|
f.l.Debug("overlay reader is done", "reader", i)
|
|
}
|