netstack: remove timing and locking from GRO

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

@@ -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",
     ],
 )

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 {

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) {

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
 }

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",

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",
+)

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"

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 {

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,
 	})
 }