From 597bc5f90ddddebfc63c4ba1ac9222dd5c7bfbd9 Mon Sep 17 00:00:00 2001 From: Kevin Krakauer Date: Fri, 5 Apr 2024 21:44:51 -0700 Subject: [PATCH] netstack: remove timing and locking from GRO MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Having read more GRO kernel code and NAPI, I believe the previous design was overly complex and resulted in poor performance. - Netstack GRO uses a `time.Timer` to periodically clear GRO'd packets. It's got... several problems. 1. The timer can be configured via CLI flag with arbitrary granularity, but (IIUC) `time.Timer` relies on the netpoller, which [has millisecond granularity]. 2. The timer creates new goroutines when it fires. 3. There's a complex atomic value song and dance to setting up the timer, canceling it when no packets are pending, and resuming it for incoming packets. - Linux GRO doesn't quite work this way. - Typically, [Linux flushes GRO whenever it can] (unless running with high HZ; Go preempts goroutines every 10ms so this does not apply). IIUC, receiving packets triggers the scheduling of a softirq that batch reads as many packets from a device as possible. softirqs are run as kernel threads, so this is getting scheduled with jiffy-ish granularity. - There's no special scheduling of a timer to flush GRO. In Netstack our "interrupt" is that we return from a poll, then we read multiple packets at once via recvmmsg/readv/XDP. So we've already got a delay analogous to "trigger a ksoftirq and wait for the thread to get scheduled." Thus, we should use zero-timeout GRO that does the following: - When recvmmsg/poll/etc returns a single packet, just pass it directly and immediately up the stack. - When multiple packets are returned, coalesce them with GRO and flush them without waiting. We never have 1000+ HZ situation. - We can remove all atomics and locking from GRO, as there will be one `groDispatcher` per dispatcher goroutine and no timer-spawned goroutines to synchronize with. **Performance**: The previous GRO implementation yielded a few percentage points increase in performance. This is markedly better. The following is from tcp_benchmark. It is running with host GRO/GSO disabled, as there's nothing to GRO when the host does it for us. The RecvMMsg dispatcher is used, as the PacketMMap dispatcher does not return multiple packets at once (RecvMMsg averages 8 per syscall in these benchmarks). ``` │ /tmp/old.log │ /tmp/new.log │ │ Mb/s │ Mb/s vs base │ TCP/role=server/host-gso=false/host-gro=false 1.764k ± 2% 2.139k ± 2% +21.29% (p=0.000 n=20) ``` PiperOrigin-RevId: 622366744 --- pkg/tcpip/link/fdbased/BUILD | 1 + pkg/tcpip/link/fdbased/endpoint.go | 7 +- pkg/tcpip/link/fdbased/endpoint_test.go | 2 +- pkg/tcpip/link/fdbased/packet_dispatchers.go | 22 +- pkg/tcpip/stack/BUILD | 15 - pkg/tcpip/stack/gro/BUILD | 43 +++ pkg/tcpip/stack/{ => gro}/gro.go | 288 +++++-------------- pkg/tcpip/stack/{ => gro}/gro_test.go | 2 +- test/benchmarks/tcp/tcp_proxy.go | 13 +- test/benchmarks/tcp/xdp.go | 1 + 10 files changed, 148 insertions(+), 246 deletions(-) create mode 100644 pkg/tcpip/stack/gro/BUILD rename pkg/tcpip/stack/{ => gro}/gro.go (70%) rename pkg/tcpip/stack/{ => gro}/gro_test.go (98%) diff --git a/pkg/tcpip/link/fdbased/BUILD b/pkg/tcpip/link/fdbased/BUILD index f29ab942a..301a79874 100644 --- a/pkg/tcpip/link/fdbased/BUILD +++ b/pkg/tcpip/link/fdbased/BUILD @@ -25,6 +25,7 @@ go_library( "//pkg/tcpip/link/rawfile", "//pkg/tcpip/link/stopfd", "//pkg/tcpip/stack", + "//pkg/tcpip/stack/gro", "@org_golang_x_sys//unix:go_default_library", ], ) diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index e5523c7e9..47e289af1 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/pkg/tcpip/link/fdbased/endpoint.go @@ -324,7 +324,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) { } } - inboundDispatcher, err := createInboundDispatcher(e, fd, isSocket, fid) + inboundDispatcher, err := createInboundDispatcher(e, fd, isSocket, fid, opts) if err != nil { return nil, fmt.Errorf("createInboundDispatcher(...) = %v", err) } @@ -334,7 +334,7 @@ func New(opts *Options) (stack.LinkEndpoint, error) { return e, nil } -func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32) (linkDispatcher, error) { +func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32, opts *Options) (linkDispatcher, error) { // By default use the readv() dispatcher as it works with all kinds of // FDs (tap/tun/unix domain sockets and af_packet). inboundDispatcher, err := newReadVDispatcher(fd, e) @@ -385,7 +385,7 @@ func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32) (lin // If the provided FD is a socket then we optimize // packet reads by using recvmmsg() instead of read() to // read packets in a batch. - inboundDispatcher, err = newRecvMMsgDispatcher(fd, e) + inboundDispatcher, err = newRecvMMsgDispatcher(fd, e, opts) if err != nil { return nil, fmt.Errorf("newRecvMMsgDispatcher(%d, %+v) = %v", fd, e, err) } @@ -413,6 +413,7 @@ func isSocketFD(fd int) (bool, error) { func (e *endpoint) Attach(dispatcher stack.NetworkDispatcher) { e.mu.Lock() defer e.mu.Unlock() + // nil means the NIC is being removed. if dispatcher == nil && e.dispatcher != nil { for _, dispatcher := range e.inboundDispatchers { diff --git a/pkg/tcpip/link/fdbased/endpoint_test.go b/pkg/tcpip/link/fdbased/endpoint_test.go index 3edd2f3f6..353bf4c29 100644 --- a/pkg/tcpip/link/fdbased/endpoint_test.go +++ b/pkg/tcpip/link/fdbased/endpoint_test.go @@ -590,7 +590,7 @@ func TestDispatchPacketFormat(t *testing.T) { }, { name: "recvMMsgDispatcher", - newDispatcher: newRecvMMsgDispatcher, + newDispatcher: func(fd int, e *endpoint) (linkDispatcher, error) { return newRecvMMsgDispatcher(fd, e, &Options{}) }, }, } { t.Run(test.name, func(t *testing.T) { diff --git a/pkg/tcpip/link/fdbased/packet_dispatchers.go b/pkg/tcpip/link/fdbased/packet_dispatchers.go index f8caca86f..ed418ff3d 100644 --- a/pkg/tcpip/link/fdbased/packet_dispatchers.go +++ b/pkg/tcpip/link/fdbased/packet_dispatchers.go @@ -25,6 +25,7 @@ import ( "gvisor.dev/gvisor/pkg/tcpip/link/rawfile" "gvisor.dev/gvisor/pkg/tcpip/link/stopfd" "gvisor.dev/gvisor/pkg/tcpip/stack" + "gvisor.dev/gvisor/pkg/tcpip/stack/gro" ) // BufConfig defines the shape of the buffer used to read packets from the NIC. @@ -240,6 +241,9 @@ type recvMMsgDispatcher struct { // pkts is reused to avoid allocations. pkts stack.PacketBufferList + + // gro coalesces incoming packets to increase throughput. + gro gro.GRO } const ( @@ -248,7 +252,7 @@ const ( MaxMsgsPerRecv = 8 ) -func newRecvMMsgDispatcher(fd int, e *endpoint) (linkDispatcher, error) { +func newRecvMMsgDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) { stopFD, err := stopfd.New() if err != nil { return nil, err @@ -264,6 +268,8 @@ func newRecvMMsgDispatcher(fd int, e *endpoint) (linkDispatcher, error) { for i := range d.bufs { d.bufs[i] = newIovecBuffer(BufConfig, skipsVnetHdr) } + d.gro.Init(opts.GRO) + return d, nil } @@ -292,13 +298,16 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) { if nMsgs == -1 || err != nil { return false, err } + // Process each of received packets. d.e.mu.RLock() dsp := d.e.dispatcher d.e.mu.RUnlock() + d.gro.Dispatcher = dsp defer d.pkts.Reset() + for k := 0; k < nMsgs; k++ { n := int(d.msgHdrs[k].Len) pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{ @@ -336,8 +345,17 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) { } } - dsp.DeliverNetworkPacket(p, pkt) + // Only use GRO if there's more than one packet. + if nMsgs > 1 { + pkt.NetworkProtocolNumber = p + pkt.RXChecksumValidated = d.e.caps&stack.CapabilityRXChecksumOffload != 0 + d.gro.Enqueue(pkt) + } else { + dsp.DeliverNetworkPacket(p, pkt) + return true, nil + } } + d.gro.Flush() return true, nil } diff --git a/pkg/tcpip/stack/BUILD b/pkg/tcpip/stack/BUILD index 86ced0f41..0ed67d27e 100644 --- a/pkg/tcpip/stack/BUILD +++ b/pkg/tcpip/stack/BUILD @@ -183,18 +183,6 @@ go_template_instance( }, ) -go_template_instance( - name = "gro_packet_list", - out = "gro_packet_list.go", - package = "stack", - prefix = "groPacket", - template = "//pkg/ilist:generic_list", - types = { - "Element": "*groPacket", - "Linker": "*groPacket", - }, -) - go_template_instance( name = "address_state_refs", out = "address_state_refs.go", @@ -219,8 +207,6 @@ go_library( "conn_track_mutex.go", "conntrack.go", "endpoints_by_nic_mutex.go", - "gro.go", - "gro_packet_list.go", "headertype_string.go", "hook_string.go", "icmp_rate_limit.go", @@ -329,7 +315,6 @@ go_test( srcs = [ "conntrack_test.go", "forwarding_test.go", - "gro_test.go", "iptables_test.go", "neighbor_cache_test.go", "neighbor_entry_test.go", diff --git a/pkg/tcpip/stack/gro/BUILD b/pkg/tcpip/stack/gro/BUILD new file mode 100644 index 000000000..07578b536 --- /dev/null +++ b/pkg/tcpip/stack/gro/BUILD @@ -0,0 +1,43 @@ +load("//tools:defs.bzl", "go_library", "go_test") +load("//tools/go_generics:defs.bzl", "go_template_instance") + +package( + default_applicable_licenses = ["//:license"], + licenses = ["notice"], +) + +go_template_instance( + name = "gro_packet_list", + out = "gro_packet_list.go", + package = "gro", + prefix = "groPacket", + template = "//pkg/ilist:generic_list", + types = { + "Element": "*groPacket", + "Linker": "*groPacket", + }, +) + +go_library( + name = "gro", + srcs = [ + "gro.go", + "gro_packet_list.go", + ], + visibility = ["//visibility:public"], + deps = [ + "//pkg/ilist", + "//pkg/tcpip", + "//pkg/tcpip/header", + "//pkg/tcpip/stack", + ], +) + +go_test( + name = "gro_test", + size = "small", + srcs = [ + "gro_test.go", + ], + library = ":gro", +) diff --git a/pkg/tcpip/stack/gro.go b/pkg/tcpip/stack/gro/gro.go similarity index 70% rename from pkg/tcpip/stack/gro.go rename to pkg/tcpip/stack/gro/gro.go index 53a4b9913..89856d9a1 100644 --- a/pkg/tcpip/stack/gro.go +++ b/pkg/tcpip/stack/gro/gro.go @@ -12,17 +12,16 @@ // See the License for the specific language governing permissions and // limitations under the License. -package stack +// Package gro implements generic receive offload. +package gro import ( "bytes" "fmt" - "time" - "gvisor.dev/gvisor/pkg/atomicbitops" - "gvisor.dev/gvisor/pkg/sync" "gvisor.dev/gvisor/pkg/tcpip" "gvisor.dev/gvisor/pkg/tcpip/header" + "gvisor.dev/gvisor/pkg/tcpip/stack" ) // TODO(b/256037250): Enable by default. @@ -35,8 +34,7 @@ import ( // opportunity for coalescing. // TODO(b/256037250): We're doing some header parsing here, which presents the // opportunity to skip it later. -// TODO(b/256037250): We may be able to remove locking by pairing -// groDispatchers with link endpoint dispatchers. +// TODO(b/256037250): Can we pass a packet list up the stack too? const ( // groNBuckets is the number of GRO buckets. @@ -53,39 +51,28 @@ const ( // A groBucket holds packets that are undergoing GRO. type groBucket struct { - // mu protects the fields of a bucket. - mu sync.Mutex - // count is the number of packets in the bucket. - // +checklocks:mu count int // packets is the linked list of packets. - // +checklocks:mu packets groPacketList // packetsPrealloc and allocIdxs are used to preallocate and reuse // groPacket structs and avoid allocation. - // +checklocks:mu packetsPrealloc [groBucketSize]groPacket - // +checklocks:mu allocIdxs [groBucketSize]int } -// +checklocks:gb.mu func (gb *groBucket) full() bool { return gb.count == groBucketSize } // insert inserts pkt into the bucket. -// +checklocks:gb.mu -func (gb *groBucket) insert(pkt *PacketBuffer, ipHdr []byte, tcpHdr header.TCP, ep NetworkEndpoint) { +func (gb *groBucket) insert(pkt *stack.PacketBuffer, ipHdr []byte, tcpHdr header.TCP) { groPkt := &gb.packetsPrealloc[gb.allocIdxs[gb.count]] *groPkt = groPacket{ pkt: pkt, - created: time.Now(), - ep: ep, ipHdr: ipHdr, tcpHdr: tcpHdr, initialLength: pkt.Data().Size(), // pkt.Data() contains network header. @@ -96,9 +83,8 @@ func (gb *groBucket) insert(pkt *PacketBuffer, ipHdr []byte, tcpHdr header.TCP, } // removeOldest removes the oldest packet from gb and returns the contained -// *PacketBuffer. gb must not be empty. -// +checklocks:gb.mu -func (gb *groBucket) removeOldest() *PacketBuffer { +// PacketBuffer. gb must not be empty. +func (gb *groBucket) removeOldest() *stack.PacketBuffer { pkt := gb.packets.Front() gb.packets.Remove(pkt) gb.count-- @@ -109,7 +95,6 @@ func (gb *groBucket) removeOldest() *PacketBuffer { } // removeOne removes a packet from gb. It also resets pkt to its zero value. -// +checklocks:gb.mu func (gb *groBucket) removeOne(pkt *groPacket) { gb.packets.Remove(pkt) gb.count-- @@ -120,8 +105,7 @@ func (gb *groBucket) removeOne(pkt *groPacket) { // findGROPacket4 returns the groPkt that matches ipHdr and tcpHdr, or nil if // none exists. It also returns whether the groPkt should be flushed based on // differences between the two headers. -// +checklocks:gb.mu -func (gb *groBucket) findGROPacket4(pkt *PacketBuffer, ipHdr header.IPv4, tcpHdr header.TCP, ep NetworkEndpoint) (*groPacket, bool) { +func (gb *groBucket) findGROPacket4(pkt *stack.PacketBuffer, ipHdr header.IPv4, tcpHdr header.TCP) (*groPacket, bool) { for groPkt := gb.packets.Front(); groPkt != nil; groPkt = groPkt.Next() { // Do the addresses match? groIPHdr := header.IPv4(groPkt.ipHdr) @@ -162,8 +146,7 @@ func (gb *groBucket) findGROPacket4(pkt *PacketBuffer, ipHdr header.IPv4, tcpHdr // findGROPacket6 returns the groPkt that matches ipHdr and tcpHdr, or nil if // none exists. It also returns whether the groPkt should be flushed based on // differences between the two headers. -// +checklocks:gb.mu -func (gb *groBucket) findGROPacket6(pkt *PacketBuffer, ipHdr header.IPv6, tcpHdr header.TCP, ep NetworkEndpoint) (*groPacket, bool) { +func (gb *groBucket) findGROPacket6(pkt *stack.PacketBuffer, ipHdr header.IPv6, tcpHdr header.TCP) (*groPacket, bool) { for groPkt := gb.packets.Front(); groPkt != nil; groPkt = groPkt.Next() { // Do the addresses match? groIPHdr := header.IPv6(groPkt.ipHdr) @@ -215,22 +198,18 @@ func (gb *groBucket) findGROPacket6(pkt *PacketBuffer, ipHdr header.IPv6, tcpHdr return nil, false } -// +checklocks:gb.mu -func (gb *groBucket) found(gd *groDispatcher, groPkt *groPacket, flushGROPkt bool, pkt *PacketBuffer, ipHdr []byte, tcpHdr header.TCP, ep NetworkEndpoint, updateIPHdr func([]byte, int)) { +func (gb *groBucket) found(gd *GRO, groPkt *groPacket, flushGROPkt bool, pkt *stack.PacketBuffer, ipHdr []byte, tcpHdr header.TCP, updateIPHdr func([]byte, int)) { // Flush groPkt or merge the packets. pktSize := pkt.Data().Size() flags := tcpHdr.Flags() dataOff := tcpHdr.DataOffset() tcpPayloadSize := pkt.Data().Size() - len(ipHdr) - int(dataOff) if flushGROPkt { - // Flush the existing GRO packet. Don't hold bucket.mu while - // processing the packet. + // Flush the existing GRO packet. pkt := groPkt.pkt gb.removeOne(groPkt) - gb.mu.Unlock() - ep.HandlePacket(pkt) + gd.handlePacket(pkt) pkt.DecRef() - gb.mu.Lock() groPkt = nil } else if groPkt != nil { // Merge pkt in to GRO packet. @@ -263,34 +242,24 @@ func (gb *groBucket) found(gd *groDispatcher, groPkt *groPacket, flushGROPkt boo // A merge occurred and we need to flush groPkt. pkt := groPkt.pkt gb.removeOne(groPkt) - gb.mu.Unlock() - ep.HandlePacket(pkt) + gd.handlePacket(pkt) pkt.DecRef() case flush && groPkt == nil: // No merge occurred and the incoming packet needs to be flushed. - gb.mu.Unlock() - ep.HandlePacket(pkt) + gd.handlePacket(pkt) case !flush && groPkt == nil: // New flow and we don't need to flush. Insert pkt into GRO. if gb.full() { // Head is always the oldest packet toFlush := gb.removeOldest() - gb.insert(pkt.IncRef(), ipHdr, tcpHdr, ep) - gb.mu.Unlock() - ep.HandlePacket(toFlush) + gb.insert(pkt.IncRef(), ipHdr, tcpHdr) + gd.handlePacket(toFlush) toFlush.DecRef() } else { - gb.insert(pkt.IncRef(), ipHdr, tcpHdr, ep) - gb.mu.Unlock() + gb.insert(pkt.IncRef(), ipHdr, tcpHdr) } default: // A merge occurred and we don't need to flush anything. - gb.mu.Unlock() - } - - // Schedule a timer if we never had one set before. - if gd.flushTimerState.CompareAndSwap(flushTimerUnset, flushTimerSet) { - gd.flushTimer.Reset(gd.getInterval()) } } @@ -301,7 +270,7 @@ type groPacket struct { groPacketEntry // pkt is the coalesced packet. - pkt *PacketBuffer + pkt *stack.PacketBuffer // ipHdr is the IP (v4 or v6) header for the coalesced packet. ipHdr []byte @@ -309,12 +278,6 @@ type groPacket struct { // tcpHdr is the TCP header for the coalesced packet. tcpHdr header.TCP - // created is when the packet was received. - created time.Time - - // ep is the endpoint to which the packet will be sent after GRO. - ep NetworkEndpoint - // initialLength is the length of the first packet in the flow. It is // used as a best-effort guess at MSS: senders will send MSS-sized // packets until they run out of data, so we coalesce as long as @@ -339,96 +302,47 @@ func (pk *groPacket) payloadSize() int { return pk.pkt.Data().Size() - len(pk.ipHdr) - int(pk.tcpHdr.DataOffset()) } -// Values held in groDispatcher.flushTimerState. -const ( - flushTimerUnset = iota - flushTimerSet - flushTimerClosed -) - -// groDispatcher coalesces incoming packets to increase throughput. -type groDispatcher struct { - // intervalNS is the interval in nanoseconds. - intervalNS atomicbitops.Int64 - +// GRO coalesces incoming packets to increase throughput. +type GRO struct { + enabled bool buckets [groNBuckets]groBucket - flushTimerState atomicbitops.Int32 - flushTimer *time.Timer + Dispatcher stack.NetworkDispatcher } -func (gd *groDispatcher) init(interval time.Duration) { - gd.intervalNS.Store(interval.Nanoseconds()) - +// Init initializes GRO. +func (gd *GRO) Init(enabled bool) { + gd.enabled = enabled for i := range gd.buckets { bucket := &gd.buckets[i] - bucket.mu.Lock() for j := range bucket.packetsPrealloc { bucket.allocIdxs[j] = j bucket.packetsPrealloc[j].idx = j } - bucket.mu.Unlock() - } - - // Create a timer to fire far from now and cancel it immediately. - // - // The timer will be reset when there is a need for it to fire. - gd.flushTimer = time.AfterFunc(time.Hour, func() { - if !gd.flushTimerState.CompareAndSwap(flushTimerSet, flushTimerUnset) { - // Timer was unset or GRO is closed, do nothing further. - return - } - - interval := gd.getInterval() - if interval == 0 { - gd.flushAll() - return - } - - if gd.flush() && gd.flushTimerState.CompareAndSwap(flushTimerUnset, flushTimerSet) { - // Only reset the timer if we have more packets and the timer was - // previously unset. If we have no packets left, the timer is already set - // or GRO is being closed, do not reset the timer. - gd.flushTimer.Reset(interval) - } - }) - gd.flushTimer.Stop() -} - -func (gd *groDispatcher) getInterval() time.Duration { - return time.Duration(gd.intervalNS.Load()) * time.Nanosecond -} - -// setInterval is not thread-safe and so much be protected by callers. -func (gd *groDispatcher) setInterval(interval time.Duration) { - gd.intervalNS.Store(interval.Nanoseconds()) - - if gd.flushTimerState.Load() == flushTimerSet { - // Timer was previously set, reset it. - gd.flushTimer.Reset(interval) } } -// dispatch sends pkt up the stack after it undergoes GRO coalescing. -func (gd *groDispatcher) dispatch(pkt *PacketBuffer, netProto tcpip.NetworkProtocolNumber, ep NetworkEndpoint) { - // If GRO is disabled simply pass the packet along. - if gd.getInterval() == 0 { - ep.HandlePacket(pkt) +// Enqueue the packet in GRO. This does not flush packets; Flush() must be +// called explicitly for that. +// +// pkt.NetworkProtocolNumber and pkt.RXChecksumValidated must be set. +func (gd *GRO) Enqueue(pkt *stack.PacketBuffer) { + if !gd.enabled { + gd.handlePacket(pkt) return } - switch netProto { + switch pkt.NetworkProtocolNumber { case header.IPv4ProtocolNumber: - gd.dispatch4(pkt, ep) + gd.dispatch4(pkt) case header.IPv6ProtocolNumber: - gd.dispatch6(pkt, ep) + gd.dispatch6(pkt) default: - // We can't GRO this. - ep.HandlePacket(pkt) + gd.handlePacket(pkt) } } -func (gd *groDispatcher) dispatch4(pkt *PacketBuffer, ep NetworkEndpoint) { +func (gd *GRO) dispatch4(pkt *stack.PacketBuffer) { // Immediately get the IPv4 and TCP headers. We need a way to hash the // packet into its bucket, which requires addresses and ports. Linux // simply gets a hash passed by hardware, but we're not so lucky. @@ -438,7 +352,7 @@ func (gd *groDispatcher) dispatch4(pkt *PacketBuffer, ep NetworkEndpoint) { // together. hdrBytes, ok := pkt.Data().PullUp(header.IPv4MinimumSize + header.TCPMinimumSize) if !ok { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } ipHdr := header.IPv4(hdrBytes) @@ -446,13 +360,13 @@ func (gd *groDispatcher) dispatch4(pkt *PacketBuffer, ep NetworkEndpoint) { // We don't handle fragments. That should be the vast majority of // traffic, and simplifies handling. if ipHdr.FragmentOffset() != 0 || ipHdr.Flags()&header.IPv4FlagMoreFragments != 0 { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } // We only handle TCP packets without IP options. if ipHdr.HeaderLength() != header.IPv4MinimumSize || tcpip.TransportProtocolNumber(ipHdr.Protocol()) != header.TCPProtocolNumber { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } tcpHdr := header.TCP(hdrBytes[header.IPv4MinimumSize:]) @@ -460,13 +374,13 @@ func (gd *groDispatcher) dispatch4(pkt *PacketBuffer, ep NetworkEndpoint) { dataOff := tcpHdr.DataOffset() if dataOff < header.TCPMinimumSize { // Malformed packet: will be handled further up the stack. - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } hdrBytes, ok = pkt.Data().PullUp(header.IPv4MinimumSize + int(dataOff)) if !ok { // Malformed packet: will be handled further up the stack. - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } @@ -476,13 +390,13 @@ func (gd *groDispatcher) dispatch4(pkt *PacketBuffer, ep NetworkEndpoint) { // what bits were flipped, we can't identify this packet with a flow. if !pkt.RXChecksumValidated { if !ipHdr.IsValid(pkt.Data().Size()) || !ipHdr.IsChecksumValid() { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } payloadChecksum := pkt.Data().ChecksumAtOffset(header.IPv4MinimumSize + int(dataOff)) tcpPayloadSize := pkt.Data().Size() - header.IPv4MinimumSize - int(dataOff) if !tcpHdr.IsChecksumValid(ipHdr.SourceAddress(), ipHdr.DestinationAddress(), payloadChecksum, uint16(tcpPayloadSize)) { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } // We've validated the checksum, no reason for others to do it @@ -492,19 +406,18 @@ func (gd *groDispatcher) dispatch4(pkt *PacketBuffer, ep NetworkEndpoint) { // Now we can get the bucket for the packet. bucket := &gd.buckets[gd.bucketForPacket4(ipHdr, tcpHdr)&groNBucketsMask] - bucket.mu.Lock() - groPkt, flushGROPkt := bucket.findGROPacket4(pkt, ipHdr, tcpHdr, ep) - bucket.found(gd, groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, ep, updateIPv4Hdr) + groPkt, flushGROPkt := bucket.findGROPacket4(pkt, ipHdr, tcpHdr) + bucket.found(gd, groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, updateIPv4Hdr) } -func (gd *groDispatcher) dispatch6(pkt *PacketBuffer, ep NetworkEndpoint) { +func (gd *GRO) dispatch6(pkt *stack.PacketBuffer) { // Immediately get the IPv6 and TCP headers. We need a way to hash the // packet into its bucket, which requires addresses and ports. Linux // simply gets a hash passed by hardware, but we're not so lucky. hdrBytes, ok := pkt.Data().PullUp(header.IPv6MinimumSize) if !ok { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } ipHdr := header.IPv6(hdrBytes) @@ -520,7 +433,7 @@ func (gd *groDispatcher) dispatch6(pkt *PacketBuffer, ep NetworkEndpoint) { transProto = tcpip.TransportProtocolNumber(it.NextHeaderIdentifier()) extHdr, done, err := it.Next() if err != nil { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } if done { @@ -544,28 +457,28 @@ func (gd *groDispatcher) dispatch6(pkt *PacketBuffer, ep NetworkEndpoint) { hdrBytes, ok = pkt.Data().PullUp(ipHdrSize + header.TCPMinimumSize) if !ok { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } ipHdr = header.IPv6(hdrBytes[:ipHdrSize]) // We only handle TCP packets. if transProto != header.TCPProtocolNumber { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } tcpHdr := header.TCP(hdrBytes[ipHdrSize:]) dataOff := tcpHdr.DataOffset() if dataOff < header.TCPMinimumSize { // Malformed packet: will be handled further up the stack. - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } hdrBytes, ok = pkt.Data().PullUp(ipHdrSize + int(dataOff)) if !ok { // Malformed packet: will be handled further up the stack. - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } tcpHdr = header.TCP(hdrBytes[ipHdrSize:]) @@ -574,13 +487,13 @@ func (gd *groDispatcher) dispatch6(pkt *PacketBuffer, ep NetworkEndpoint) { // what bits were flipped, we can't identify this packet with a flow. if !pkt.RXChecksumValidated { if !ipHdr.IsValid(pkt.Data().Size()) { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } payloadChecksum := pkt.Data().ChecksumAtOffset(ipHdrSize + int(dataOff)) tcpPayloadSize := pkt.Data().Size() - ipHdrSize - int(dataOff) if !tcpHdr.IsChecksumValid(ipHdr.SourceAddress(), ipHdr.DestinationAddress(), payloadChecksum, uint16(tcpPayloadSize)) { - ep.HandlePacket(pkt) + gd.handlePacket(pkt) return } // We've validated the checksum, no reason for others to do it @@ -590,12 +503,11 @@ func (gd *groDispatcher) dispatch6(pkt *PacketBuffer, ep NetworkEndpoint) { // Now we can get the bucket for the packet. bucket := &gd.buckets[gd.bucketForPacket6(ipHdr, tcpHdr)&groNBucketsMask] - bucket.mu.Lock() - groPkt, flushGROPkt := bucket.findGROPacket6(pkt, ipHdr, tcpHdr, ep) - bucket.found(gd, groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, ep, updateIPv6Hdr) + groPkt, flushGROPkt := bucket.findGROPacket6(pkt, ipHdr, tcpHdr) + bucket.found(gd, groPkt, flushGROPkt, pkt, ipHdr, tcpHdr, updateIPv6Hdr) } -func (gd *groDispatcher) bucketForPacket4(ipHdr header.IPv4, tcpHdr header.TCP) int { +func (gd *GRO) bucketForPacket4(ipHdr header.IPv4, tcpHdr header.TCP) int { // TODO(b/256037250): Use jenkins or checksum. Write a test to print // distribution. var sum int @@ -612,7 +524,7 @@ func (gd *groDispatcher) bucketForPacket4(ipHdr header.IPv4, tcpHdr header.TCP) return sum } -func (gd *groDispatcher) bucketForPacket6(ipHdr header.IPv6, tcpHdr header.TCP) int { +func (gd *GRO) bucketForPacket6(ipHdr header.IPv6, tcpHdr header.TCP) int { // TODO(b/256037250): Use jenkins or checksum. Write a test to print // distribution. var sum int @@ -629,92 +541,32 @@ func (gd *groDispatcher) bucketForPacket6(ipHdr header.IPv6, tcpHdr header.TCP) return sum } -// flush sends any packets older than interval up the stack. -// -// Returns true iff packets remain. -func (gd *groDispatcher) flush() bool { - interval := gd.intervalNS.Load() - old := time.Now().Add(-time.Duration(interval) * time.Nanosecond) - return gd.flushSinceOrEqualTo(old) -} - -// flushSinceOrEqualTo sends any packets older than or equal to the specified -// time. -// -// Returns true iff packets remain. -func (gd *groDispatcher) flushSinceOrEqualTo(old time.Time) bool { - type pair struct { - pkt *PacketBuffer - ep NetworkEndpoint - } - - hasMore := false - +// Flush sends all packets up the stack. +func (gd *GRO) Flush() { for i := range gd.buckets { - // Put packets in a slice so we don't have to hold bucket.mu - // when we call HandlePacket. - var pairsBacking [groNBuckets]pair - pairs := pairsBacking[:0] - - bucket := &gd.buckets[i] - bucket.mu.Lock() - for groPkt := bucket.packets.Front(); groPkt != nil; groPkt = groPkt.Next() { - if groPkt.created.After(old) { - // Packets are ordered by age, so we can move - // on once we find one that's too new. - hasMore = true - break - } else { - pairs = append(pairs, pair{groPkt.pkt, groPkt.ep}) - bucket.removeOne(groPkt) - } + for groPkt := gd.buckets[i].packets.Front(); groPkt != nil; groPkt = groPkt.Next() { + pkt := groPkt.pkt + gd.buckets[i].removeOne(groPkt) + gd.handlePacket(pkt) + pkt.DecRef() } - bucket.mu.Unlock() - - for _, pair := range pairs { - pair.ep.HandlePacket(pair.pkt) - pair.pkt.DecRef() - } - } - - return hasMore -} - -func (gd *groDispatcher) flushAll() { - if gd.flushSinceOrEqualTo(time.Now()) { - panic("packets unexpectedly remain after flushing all") } } -// close stops the GRO goroutine and releases any held packets. -func (gd *groDispatcher) close() { - gd.flushTimer.Stop() - // Prevent the timer from being scheduled again. - gd.flushTimerState.Store(flushTimerClosed) - - for i := range gd.buckets { - bucket := &gd.buckets[i] - bucket.mu.Lock() - for groPkt := bucket.packets.Front(); groPkt != nil; groPkt = bucket.packets.Front() { - groPkt.pkt.DecRef() - bucket.removeOne(groPkt) - } - bucket.mu.Unlock() - } +func (gd *GRO) handlePacket(pkt *stack.PacketBuffer) { + gd.Dispatcher.DeliverNetworkPacket(pkt.NetworkProtocolNumber, pkt) } // String implements fmt.Stringer. -func (gd *groDispatcher) String() string { +func (gd *GRO) String() string { ret := "GRO state: \n" for i := range gd.buckets { bucket := &gd.buckets[i] - bucket.mu.Lock() ret += fmt.Sprintf("bucket %d: %d packets: ", i, bucket.count) for groPkt := bucket.packets.Front(); groPkt != nil; groPkt = groPkt.Next() { - ret += fmt.Sprintf("%s (%d), ", groPkt.created, groPkt.pkt.Data().Size()) + ret += fmt.Sprintf("%d, ", groPkt.pkt.Data().Size()) } ret += "\n" - bucket.mu.Unlock() } return ret } diff --git a/pkg/tcpip/stack/gro_test.go b/pkg/tcpip/stack/gro/gro_test.go similarity index 98% rename from pkg/tcpip/stack/gro_test.go rename to pkg/tcpip/stack/gro/gro_test.go index 358e3d1fd..e67308601 100644 --- a/pkg/tcpip/stack/gro_test.go +++ b/pkg/tcpip/stack/gro/gro_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package stack +package gro import ( "math/bits" diff --git a/test/benchmarks/tcp/tcp_proxy.go b/test/benchmarks/tcp/tcp_proxy.go index 540e040f1..06aba544d 100644 --- a/test/benchmarks/tcp/tcp_proxy.go +++ b/test/benchmarks/tcp/tcp_proxy.go @@ -226,12 +226,13 @@ func newNetstackImpl(mode string) (impl, error) { // peer. But we do want to disable checksum verification as veth // devices do perform GRO and the linux host kernel may not // regenerate valid checksums after GRO. - TXChecksumOffload: false, - RXChecksumOffload: true, - // PacketDispatchMode: fdbased.RecvMMsg, - PacketDispatchMode: fdbased.PacketMMap, - GSOMaxSize: uint32(*gso), - GVisorGSOEnabled: *swgso, + TXChecksumOffload: false, + RXChecksumOffload: true, + PacketDispatchMode: fdbased.RecvMMsg, + // PacketDispatchMode: fdbased.PacketMMap, + GSOMaxSize: uint32(*gso), + GVisorGSOEnabled: *swgso, + GRO: *gro, }) } if err != nil { diff --git a/test/benchmarks/tcp/xdp.go b/test/benchmarks/tcp/xdp.go index aa7a0d95e..f99c2afa8 100644 --- a/test/benchmarks/tcp/xdp.go +++ b/test/benchmarks/tcp/xdp.go @@ -116,5 +116,6 @@ func newXDPEndpoint(ifaceName string, mac net.HardwareAddr) (stack.LinkEndpoint, RXChecksumOffload: true, InterfaceIndex: iface.Index, Bind: true, + GRO: *gro, }) }