drop TxBatcher interface

This commit is contained in:
JackDoan
2026-07-31 13:58:33 -05:00
parent ddb90ad4b7
commit d7bcfb5d6b
2 changed files with 7 additions and 26 deletions
+6 -10
View File
@@ -15,7 +15,7 @@ import (
"github.com/slackhq/nebula/routing" "github.com/slackhq/nebula/routing"
) )
func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Packet, nb []byte, sendBatch batch.TxBatcher, rejectBuf []byte, q int, localCache firewall.ConntrackCache) { func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Packet, nb []byte, sendBatch *batch.SendBatch, rejectBuf []byte, q int, localCache firewall.ConntrackCache) {
// borrowed: pkt.Bytes is owned by the originating tio.Queue and is // borrowed: pkt.Bytes is owned by the originating tio.Queue and is
// only valid until the next Read on that queue. Every consumer below // only valid until the next Read on that queue. Every consumer below
// (parse, self-forward, handshake cache, sendInsideMessage) reads it // (parse, self-forward, handshake cache, sendInsideMessage) reads it
@@ -107,7 +107,7 @@ func (f *Interface) consumeInsidePacket(pkt tio.Packet, fwPacket *firewall.Packe
dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache) dropReason := f.firewall.Drop(*fwPacket, false, hostinfo, f.pki.GetCAPool(), localCache)
if dropReason == nil { if dropReason == nil {
f.sendInsideMessage(hostinfo, pkt, nb, sendBatch, rejectBuf, q) f.sendInsideMessage(hostinfo, pkt, nb, sendBatch)
} else { } else {
f.rejectInside(packet, rejectBuf, q) f.rejectInside(packet, rejectBuf, q)
if f.l.Enabled(context.Background(), slog.LevelDebug) { if f.l.Enabled(context.Background(), slog.LevelDebug) {
@@ -137,8 +137,7 @@ func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, s
"udpAddr", hostinfo.GetRemote(), "udpAddr", hostinfo.GetRemote(),
"counter", c, "counter", c,
) )
// Skip this segment; the rest of the superpacket can still // Skip this segment; the rest of the superpacket can still go out. TCP will retransmit anything we drop here.
// go out — TCP will retransmit anything we drop here.
return nil return nil
} }
@@ -150,9 +149,8 @@ func (f *Interface) sendInsideEncrypt(hostinfo *HostInfo, ci *ConnectionState, s
// later sendmmsg flush. Segmentation is fused with encryption here so the // later sendmmsg flush. Segmentation is fused with encryption here so the
// kernel-supplied superpacket bytes never get written into a separate // kernel-supplied superpacket bytes never get written into a separate
// scratch arena: SegmentSuperpacket builds each segment's plaintext in // scratch arena: SegmentSuperpacket builds each segment's plaintext in
// segScratch[:segLen] in turn, and we encrypt directly into a fresh // segScratch[:segLen] in turn, and we encrypt directly into a fresh SendBatch slot.
// SendBatch slot. func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, sendBatch *batch.SendBatch) {
func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []byte, sendBatch batch.TxBatcher, rejectBuf []byte, q int) {
ci := hostinfo.ConnectionState ci := hostinfo.ConnectionState
if ci.eKey == nil { if ci.eKey == nil {
return return
@@ -236,9 +234,7 @@ func (f *Interface) sendInsideMessage(hostinfo *HostInfo, pkt tio.Packet, nb []b
return nil return nil
}) })
if err != nil { if err != nil {
hostinfo.logger(f.l).Error("Failed to segment superpacket for send", hostinfo.logger(f.l).Error("Failed to segment superpacket for send", "error", err)
"error", err,
)
} }
} }
+1 -16
View File
@@ -1,7 +1,5 @@
package batch package batch
import "net/netip"
type RxBatcher interface { type RxBatcher interface {
// Commit commits pkt to be flushed by the batch. The caller must keep pkt valid until the next Flush, and not re-use it. // Commit commits pkt to be flushed by the batch. The caller must keep pkt valid until the next Flush, and not re-use it.
Commit(pkt []byte) error Commit(pkt []byte) error
@@ -9,21 +7,8 @@ type RxBatcher interface {
// a flow's payload-bearing packets are never reordered relative to each // a flow's payload-bearing packets are never reordered relative to each
// other. Cross-flow and cross-lane order is not preserved, and two shapes // other. Cross-flow and cross-lane order is not preserved, and two shapes
// may legally be overtaken by later same-flow data: pure ACKs (by design, // may legally be overtaken by later same-flow data: pure ACKs (by design,
// stale ACKs are ignored) and unparseable shapes such as fragments (an // stale ACKs are ignored) and unparseable shapes such as fragments (an accepted tradeoff; see MultiCoalescer).
// accepted tradeoff; see MultiCoalescer).
// Returns the first error observed; keeps draining so one bad packet doesn't hold up the rest. // Returns the first error observed; keeps draining so one bad packet doesn't hold up the rest.
// After Flush returns, committed payload slices may be recycled. // After Flush returns, committed payload slices may be recycled.
Flush() error Flush() error
} }
type TxBatcher interface {
// Reserve creates a pkt to borrow
Reserve(sz int) []byte
// Commit borrows pkt and records its destination. The caller must
// keep pkt valid until the next Flush.
Commit(pkt []byte, dst netip.AddrPort)
// Flush emits every queued packet via the underlying batch writer in arrival order and reports how many were
// actually written. A short count means some destinations were undeliverable, not that the batch failed.
// After Flush returns, borrowed payload slices may be recycled.
Flush() (int, error)
}