Spawn dequeing task via the clock

...and use manual clocks in forwarding and link resolution tests.

Fixes #5141.
Fixes #6012.

PiperOrigin-RevId: 375939167

4 files changed, 44 insertions, 34 deletions

diff --git a/pkg/tcpip/stack/forwarding_test.go b/pkg/tcpip/stack/forwarding_test.go
index d971db010..72f66441f 100644
--- a/pkg/tcpip/stack/forwarding_test.go
+++ b/pkg/tcpip/stack/forwarding_test.go
@@ -23,6 +23,7 @@ import (
 	"gvisor.dev/gvisor/pkg/sync"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/buffer"
+	"gvisor.dev/gvisor/pkg/tcpip/faketime"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 )
 
@@ -220,7 +221,7 @@ func (*fwdTestNetworkProtocol) Wait() {}
 
 func (f *fwdTestNetworkEndpoint) LinkAddressRequest(addr, _ tcpip.Address, remoteLinkAddr tcpip.LinkAddress) tcpip.Error {
 	if fn := f.proto.onLinkAddressResolved; fn != nil {
-		time.AfterFunc(f.proto.addrResolveDelay, func() {
+		f.proto.stack.clock.AfterFunc(f.proto.addrResolveDelay, func() {
 			fn(f.proto.neigh, addr, remoteLinkAddr)
 		})
 	}
@@ -354,13 +355,15 @@ func (e *fwdTestLinkEndpoint) AddHeader(tcpip.LinkAddress, tcpip.LinkAddress, tc
 	panic("not implemented")
 }
 
-func fwdTestNetFactory(t *testing.T, proto *fwdTestNetworkProtocol) (ep1, ep2 *fwdTestLinkEndpoint) {
+func fwdTestNetFactory(t *testing.T, proto *fwdTestNetworkProtocol) (*faketime.ManualClock, *fwdTestLinkEndpoint, *fwdTestLinkEndpoint) {
+	clock := faketime.NewManualClock()
 	// Create a stack with the network protocol and two NICs.
 	s := New(Options{
 		NetworkProtocols: []NetworkProtocolFactory{func(s *Stack) NetworkProtocol {
 			proto.stack = s
 			return proto
 		}},
+		Clock: clock,
 	})
 
 	protoNum := proto.Number()
@@ -369,7 +372,7 @@ func fwdTestNetFactory(t *testing.T, proto *fwdTestNetworkProtocol) (ep1, ep2 *f
 	}
 
 	// NIC 1 has the link address "a", and added the network address 1.
-	ep1 = &fwdTestLinkEndpoint{
+	ep1 := &fwdTestLinkEndpoint{
 		C:        make(chan fwdTestPacketInfo, 300),
 		mtu:      fwdTestNetDefaultMTU,
 		linkAddr: "a",
@@ -382,7 +385,7 @@ func fwdTestNetFactory(t *testing.T, proto *fwdTestNetworkProtocol) (ep1, ep2 *f
 	}
 
 	// NIC 2 has the link address "b", and added the network address 2.
-	ep2 = &fwdTestLinkEndpoint{
+	ep2 := &fwdTestLinkEndpoint{
 		C:        make(chan fwdTestPacketInfo, 300),
 		mtu:      fwdTestNetDefaultMTU,
 		linkAddr: "b",
@@ -412,7 +415,7 @@ func fwdTestNetFactory(t *testing.T, proto *fwdTestNetworkProtocol) (ep1, ep2 *f
 		s.SetRouteTable([]tcpip.Route{{Destination: subnet, NIC: 2}})
 	}
 
-	return ep1, ep2
+	return clock, ep1, ep2
 }
 
 func TestForwardingWithStaticResolver(t *testing.T) {
@@ -428,7 +431,7 @@ func TestForwardingWithStaticResolver(t *testing.T) {
 		},
 	}
 
-	ep1, ep2 := fwdTestNetFactory(t, proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, proto)
 
 	// Inject an inbound packet to address 3 on NIC 1, and see if it is
 	// forwarded to NIC 2.
@@ -440,6 +443,7 @@ func TestForwardingWithStaticResolver(t *testing.T) {
 
 	var p fwdTestPacketInfo
 
+	clock.Advance(proto.addrResolveDelay)
 	select {
 	case p = <-ep2.C:
 	default:
@@ -471,7 +475,7 @@ func TestForwardingWithFakeResolver(t *testing.T) {
 			})
 		},
 	}
-	ep1, ep2 := fwdTestNetFactory(t, &proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, &proto)
 
 	// Inject an inbound packet to address 3 on NIC 1, and see if it is
 	// forwarded to NIC 2.
@@ -483,9 +487,10 @@ func TestForwardingWithFakeResolver(t *testing.T) {
 
 	var p fwdTestPacketInfo
 
+	clock.Advance(proto.addrResolveDelay)
 	select {
 	case p = <-ep2.C:
-	case <-time.After(time.Second):
+	default:
 		t.Fatal("packet not forwarded")
 	}
 
@@ -504,7 +509,7 @@ func TestForwardingWithNoResolver(t *testing.T) {
 
 	// Whether or not we use the neighbor cache here does not matter since
 	// neither linkAddrCache nor neighborCache will be used.
-	ep1, ep2 := fwdTestNetFactory(t, proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, proto)
 
 	// inject an inbound packet to address 3 on NIC 1, and see if it is
 	// forwarded to NIC 2.
@@ -514,10 +519,11 @@ func TestForwardingWithNoResolver(t *testing.T) {
 		Data: buf.ToVectorisedView(),
 	}))
 
+	clock.Advance(proto.addrResolveDelay)
 	select {
 	case <-ep2.C:
 		t.Fatal("Packet should not be forwarded")
-	case <-time.After(time.Second):
+	default:
 	}
 }
 
@@ -529,7 +535,7 @@ func TestForwardingResolutionFailsForQueuedPackets(t *testing.T) {
 		},
 	}
 
-	ep1, ep2 := fwdTestNetFactory(t, proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, proto)
 
 	const numPackets int = 5
 	// These packets will all be enqueued in the packet queue to wait for link
@@ -543,12 +549,12 @@ func TestForwardingResolutionFailsForQueuedPackets(t *testing.T) {
 	}
 
 	// All packets should fail resolution.
-	// TODO(gvisor.dev/issue/5141): Use a fake clock.
 	for i := 0; i < numPackets; i++ {
+		clock.Advance(proto.addrResolveDelay)
 		select {
 		case got := <-ep2.C:
 			t.Fatalf("got %#v; packets should have failed resolution and not been forwarded", got)
-		case <-time.After(100 * time.Millisecond):
+		default:
 		}
 	}
 }
@@ -572,7 +578,7 @@ func TestForwardingWithFakeResolverPartialTimeout(t *testing.T) {
 			}
 		},
 	}
-	ep1, ep2 := fwdTestNetFactory(t, &proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, &proto)
 
 	// Inject an inbound packet to address 4 on NIC 1. This packet should
 	// not be forwarded.
@@ -592,9 +598,10 @@ func TestForwardingWithFakeResolverPartialTimeout(t *testing.T) {
 
 	var p fwdTestPacketInfo
 
+	clock.Advance(proto.addrResolveDelay)
 	select {
 	case p = <-ep2.C:
-	case <-time.After(time.Second):
+	default:
 		t.Fatal("packet not forwarded")
 	}
 
@@ -627,7 +634,7 @@ func TestForwardingWithFakeResolverTwoPackets(t *testing.T) {
 			})
 		},
 	}
-	ep1, ep2 := fwdTestNetFactory(t, &proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, &proto)
 
 	// Inject two inbound packets to address 3 on NIC 1.
 	for i := 0; i < 2; i++ {
@@ -641,9 +648,10 @@ func TestForwardingWithFakeResolverTwoPackets(t *testing.T) {
 	for i := 0; i < 2; i++ {
 		var p fwdTestPacketInfo
 
+		clock.Advance(proto.addrResolveDelay)
 		select {
 		case p = <-ep2.C:
-		case <-time.After(time.Second):
+		default:
 			t.Fatal("packet not forwarded")
 		}
 
@@ -677,7 +685,7 @@ func TestForwardingWithFakeResolverManyPackets(t *testing.T) {
 			})
 		},
 	}
-	ep1, ep2 := fwdTestNetFactory(t, &proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, &proto)
 
 	for i := 0; i < maxPendingPacketsPerResolution+5; i++ {
 		// Inject inbound 'maxPendingPacketsPerResolution + 5' packets on NIC 1.
@@ -693,9 +701,10 @@ func TestForwardingWithFakeResolverManyPackets(t *testing.T) {
 	for i := 0; i < maxPendingPacketsPerResolution; i++ {
 		var p fwdTestPacketInfo
 
+		clock.Advance(proto.addrResolveDelay)
 		select {
 		case p = <-ep2.C:
-		case <-time.After(time.Second):
+		default:
 			t.Fatal("packet not forwarded")
 		}
 
@@ -741,7 +750,7 @@ func TestForwardingWithFakeResolverManyResolutions(t *testing.T) {
 			})
 		},
 	}
-	ep1, ep2 := fwdTestNetFactory(t, &proto)
+	clock, ep1, ep2 := fwdTestNetFactory(t, &proto)
 
 	for i := 0; i < maxPendingResolutions+5; i++ {
 		// Inject inbound 'maxPendingResolutions + 5' packets on NIC 1.
@@ -757,9 +766,10 @@ func TestForwardingWithFakeResolverManyResolutions(t *testing.T) {
 	for i := 0; i < maxPendingResolutions; i++ {
 		var p fwdTestPacketInfo
 
+		clock.Advance(proto.addrResolveDelay)
 		select {
 		case p = <-ep2.C:
-		case <-time.After(time.Second):
+		default:
 			t.Fatal("packet not forwarded")
 		}
 
diff --git a/pkg/tcpip/stack/neighbor_entry.go b/pkg/tcpip/stack/neighbor_entry.go
index ab247db2e..0a59eecdd 100644
--- a/pkg/tcpip/stack/neighbor_entry.go
+++ b/pkg/tcpip/stack/neighbor_entry.go
@@ -166,14 +166,20 @@ func (e *neighborEntry) notifyCompletionLocked(err tcpip.Error) {
 	if ch := e.mu.done; ch != nil {
 		close(ch)
 		e.mu.done = nil
-		// Dequeue the pending packets in a new goroutine to not hold up the current
+		// Dequeue the pending packets asynchronously to not hold up the current
 		// goroutine as writing packets may be a costly operation.
 		//
 		// At the time of writing, when writing packets, a neighbor's link address
 		// is resolved (which ends up obtaining the entry's lock) while holding the
-		// link resolution queue's lock. Dequeuing packets in a new goroutine avoids
-		// a lock ordering violation.
-		go e.cache.nic.linkResQueue.dequeue(ch, e.mu.neigh.LinkAddr, err)
+		// link resolution queue's lock. Dequeuing packets asynchronously avoids a
+		// lock ordering violation.
+		//
+		// NB: this is equivalent to spawning a goroutine directly using the go
+		// keyword but allows tests that use manual clocks to deterministically
+		// wait for this work to complete.
+		e.cache.nic.stack.clock.AfterFunc(0, func() {
+			e.cache.nic.linkResQueue.dequeue(ch, e.mu.neigh.LinkAddr, err)
+		})
 	}
 }
 
diff --git a/pkg/tcpip/tests/integration/BUILD b/pkg/tcpip/tests/integration/BUILD
index 8802f36b2..181ef799e 100644
--- a/pkg/tcpip/tests/integration/BUILD
+++ b/pkg/tcpip/tests/integration/BUILD
@@ -48,7 +48,6 @@ go_test(
     size = "small",
     srcs = ["link_resolution_test.go"],
     deps = [
-        "//pkg/context",
         "//pkg/tcpip",
         "//pkg/tcpip/buffer",
         "//pkg/tcpip/checker",
diff --git a/pkg/tcpip/tests/integration/link_resolution_test.go b/pkg/tcpip/tests/integration/link_resolution_test.go
index 16fd7a99a..27caa0c28 100644
--- a/pkg/tcpip/tests/integration/link_resolution_test.go
+++ b/pkg/tcpip/tests/integration/link_resolution_test.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
-	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/buffer"
 	"gvisor.dev/gvisor/pkg/tcpip/checker"
@@ -604,8 +603,6 @@ func TestForwardingWithLinkResolutionFailure(t *testing.T) {
 
 			test.rx(incomingEndpoint, test.sourceAddr, test.destAddr)
 
-			var request channel.PacketInfo
-			var ok bool
 			nudConfigs, err := s.NUDConfigurations(outgoingNICID, test.networkProtocolNumber)
 			if err != nil {
 				t.Fatalf("s.NUDConfigurations(%d, %d): %s", outgoingNICID, test.networkProtocolNumber, err)
@@ -614,7 +611,8 @@ func TestForwardingWithLinkResolutionFailure(t *testing.T) {
 			clock.RunImmediatelyScheduledJobs()
 
 			for i := 0; i < int(nudConfigs.MaxMulticastProbes); i++ {
-				if request, ok = outgoingEndpoint.Read(); !ok {
+				request, ok := outgoingEndpoint.Read()
+				if !ok {
 					t.Fatal("expected ARP packet through outgoing NIC")
 				}
 
@@ -628,10 +626,7 @@ func TestForwardingWithLinkResolutionFailure(t *testing.T) {
 			// necessary because outgoing packets are dequeued asynchronously when
 			// link resolution fails, and this dequeue is what triggers the ICMP
 			// error.
-			//
-			// TODO(gvisor.dev/issue/6012): Replace with asynchronous read after we
-			// have integrated the stack clock with the dequeuing code.
-			reply, ok := incomingEndpoint.ReadContext(context.Background())
+			reply, ok := incomingEndpoint.Read()
 			if !ok {
 				t.Fatal("expected ICMP packet through incoming NIC")
 			}