1 files changed, 157 insertions, 90 deletions

diff --git a/pkg/tcpip/stack/pending_packets.go b/pkg/tcpip/stack/pending_packets.go
index 3ac039c7d..22dfc7960 100644
--- a/pkg/tcpip/stack/pending_packets.go
+++ b/pkg/tcpip/stack/pending_packets.go
@@ -45,135 +45,202 @@ func (p *PacketBufferList) len() int {
 }
 
 type pendingPacket struct {
-	route *Route
-	proto tcpip.NetworkProtocolNumber
-	pkt   pendingPacketBuffer
+	routeInfo RouteInfo
+	gso       *GSO
+	proto     tcpip.NetworkProtocolNumber
+	pkt       pendingPacketBuffer
 }
 
 // packetsPendingLinkResolution is a queue of packets pending link resolution.
 //
 // Once link resolution completes successfully, the packets will be written.
 type packetsPendingLinkResolution struct {
-	sync.Mutex
+	nic *NIC
 
-	// The packets to send once the resolver completes.
-	packets map[<-chan struct{}][]pendingPacket
+	mu struct {
+		sync.Mutex
 
-	// FIFO of channels used to cancel the oldest goroutine waiting for
-	// link-address resolution.
-	cancelChans []chan struct{}
-}
+		// The packets to send once the resolver completes.
+		//
+		// The link resolution channel is used as the key for this map.
+		packets map[<-chan struct{}][]pendingPacket
 
-func (f *packetsPendingLinkResolution) init() {
-	f.Lock()
-	defer f.Unlock()
-	f.packets = make(map[<-chan struct{}][]pendingPacket)
+		// FIFO of channels used to cancel the oldest goroutine waiting for
+		// link-address resolution.
+		//
+		// cancelChans holds the same channels that are used as keys to packets.
+		cancelChans []<-chan struct{}
+	}
 }
 
-func incrementOutgoingPacketErrors(r *Route, proto tcpip.NetworkProtocolNumber, pkt pendingPacketBuffer) {
+func (f *packetsPendingLinkResolution) incrementOutgoingPacketErrors(proto tcpip.NetworkProtocolNumber, pkt pendingPacketBuffer) {
 	n := uint64(pkt.len())
-	r.Stats().IP.OutgoingPacketErrors.IncrementBy(n)
+	f.nic.stack.stats.IP.OutgoingPacketErrors.IncrementBy(n)
 
-	// ok may be false if the endpoint's stats do not collect IP-related data.
-	if ipEndpointStats, ok := r.outgoingNIC.getNetworkEndpoint(proto).Stats().(IPNetworkEndpointStats); ok {
+	if ipEndpointStats, ok := f.nic.getNetworkEndpoint(proto).Stats().(IPNetworkEndpointStats); ok {
 		ipEndpointStats.IPStats().OutgoingPacketErrors.IncrementBy(n)
 	}
 }
 
-func (f *packetsPendingLinkResolution) enqueue(ch <-chan struct{}, r *Route, proto tcpip.NetworkProtocolNumber, pkt pendingPacketBuffer) {
-	f.Lock()
-	defer f.Unlock()
+func (f *packetsPendingLinkResolution) init(nic *NIC) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.nic = nic
+	f.mu.packets = make(map[<-chan struct{}][]pendingPacket)
+}
 
-	packets, ok := f.packets[ch]
-	if len(packets) == maxPendingPacketsPerResolution {
-		p := packets[0]
-		packets[0] = pendingPacket{}
-		packets = packets[1:]
+// dequeue any pending packets associated with ch.
+//
+// If success is true, packets will be written and sent to the given remote link
+// address.
+func (f *packetsPendingLinkResolution) dequeue(ch <-chan struct{}, linkAddr tcpip.LinkAddress, success bool) {
+	f.mu.Lock()
+	packets, ok := f.mu.packets[ch]
+	delete(f.mu.packets, ch)
+
+	if ok {
+		for i, cancelChan := range f.mu.cancelChans {
+			if cancelChan == ch {
+				f.mu.cancelChans = append(f.mu.cancelChans[:i], f.mu.cancelChans[i+1:]...)
+				break
+			}
+		}
+	}
+
+	f.mu.Unlock()
 
-		incrementOutgoingPacketErrors(r, proto, p.pkt)
+	if ok {
+		f.dequeuePackets(packets, linkAddr, success)
+	}
+}
 
-		p.route.Release()
+func (f *packetsPendingLinkResolution) writePacketBuffer(r RouteInfo, gso *GSO, proto tcpip.NetworkProtocolNumber, pkt pendingPacketBuffer) (int, *tcpip.Error) {
+	switch pkt := pkt.(type) {
+	case *PacketBuffer:
+		if err := f.nic.writePacket(r, gso, proto, pkt); err != nil {
+			return 0, err
+		}
+		return 1, nil
+	case *PacketBufferList:
+		return f.nic.writePackets(r, gso, proto, *pkt)
+	default:
+		panic(fmt.Sprintf("unrecognized pending packet buffer type = %T", pkt))
 	}
+}
 
-	if l := len(packets); l >= maxPendingPacketsPerResolution {
-		panic(fmt.Sprintf("max pending packets for resolution reached; got %d packets, max = %d", l, maxPendingPacketsPerResolution))
+// enqueue a packet to be sent once link resolution completes.
+//
+// If the maximum number of pending resolutions is reached, the packets
+// associated with the oldest link resolution will be dequeued as if they failed
+// link resolution.
+func (f *packetsPendingLinkResolution) enqueue(r *Route, gso *GSO, proto tcpip.NetworkProtocolNumber, pkt pendingPacketBuffer) (int, *tcpip.Error) {
+	f.mu.Lock()
+	// Make sure we attempt resolution while holding f's lock so that we avoid
+	// a race where link resolution completes before we enqueue the packets.
+	//
+	//   A @ T1: Call ResolvedFields (get link resolution channel)
+	//   B @ T2: Complete link resolution, dequeue pending packets
+	//   C @ T1: Enqueue packet that already completed link resolution (which will
+	//       never dequeue)
+	//
+	// To make sure B does not interleave with A and C, we make sure A and C are
+	// done while holding the lock.
+	routeInfo, ch, err := r.ResolvedFields(nil)
+	switch err {
+	case nil:
+		// The route resolved immediately, so we don't need to wait for link
+		// resolution to send the packet.
+		f.mu.Unlock()
+		return f.writePacketBuffer(routeInfo, gso, proto, pkt)
+	case tcpip.ErrWouldBlock:
+		// We need to wait for link resolution to complete.
+	default:
+		f.mu.Unlock()
+		return 0, err
 	}
 
-	f.packets[ch] = append(packets, pendingPacket{
-		route: r,
-		proto: proto,
-		pkt:   pkt,
+	defer f.mu.Unlock()
+
+	packets, ok := f.mu.packets[ch]
+	packets = append(packets, pendingPacket{
+		routeInfo: routeInfo,
+		gso:       gso,
+		proto:     proto,
+		pkt:       pkt,
 	})
 
+	if len(packets) > maxPendingPacketsPerResolution {
+		f.incrementOutgoingPacketErrors(packets[0].proto, packets[0].pkt)
+		packets[0] = pendingPacket{}
+		packets = packets[1:]
+
+		if numPackets := len(packets); numPackets != maxPendingPacketsPerResolution {
+			panic(fmt.Sprintf("holding more queued packets than expected; got = %d, want <= %d", numPackets, maxPendingPacketsPerResolution))
+		}
+	}
+
+	f.mu.packets[ch] = packets
+
 	if ok {
-		return
+		return pkt.len(), nil
 	}
 
-	// Wait for the link-address resolution to complete.
-	cancel := f.newCancelChannelLocked()
-	go func() {
-		cancelled := false
-		select {
-		case <-ch:
-		case <-cancel:
-			cancelled = true
-		}
+	cancelledPackets := f.newCancelChannelLocked(ch)
 
-		f.Lock()
-		packets, ok := f.packets[ch]
-		delete(f.packets, ch)
-		f.Unlock()
+	if len(cancelledPackets) != 0 {
+		// Dequeue the pending packets in a new goroutine to not hold up the current
+		// goroutine as handing link resolution failures may be a costly operation.
+		go f.dequeuePackets(cancelledPackets, "" /* linkAddr */, false /* success */)
+	}
 
-		if !ok {
-			panic(fmt.Sprintf("link-resolution goroutine woke up but no entry exists in the queue of packets"))
-		}
+	return pkt.len(), nil
+}
 
-		for _, p := range packets {
-			if cancelled || p.route.IsResolutionRequired() {
-				incrementOutgoingPacketErrors(r, proto, p.pkt)
-
-				if linkResolvableEP, ok := p.route.outgoingNIC.getNetworkEndpoint(p.route.NetProto).(LinkResolvableNetworkEndpoint); ok {
-					switch pkt := p.pkt.(type) {
-					case *PacketBuffer:
-						linkResolvableEP.HandleLinkResolutionFailure(pkt)
-					case *PacketBufferList:
-						for pb := pkt.Front(); pb != nil; pb = pb.Next() {
-							linkResolvableEP.HandleLinkResolutionFailure(pb)
-						}
-					default:
-						panic(fmt.Sprintf("unrecognized pending packet buffer type = %T", p.pkt))
-					}
-				}
-			} else {
+// newCancelChannelLocked appends the link resolution channel to a FIFO. If the
+// maximum number of pending resolutions is reached, the oldest channel will be
+// removed and its associated pending packets will be returned.
+func (f *packetsPendingLinkResolution) newCancelChannelLocked(newCH <-chan struct{}) []pendingPacket {
+	f.mu.cancelChans = append(f.mu.cancelChans, newCH)
+	if len(f.mu.cancelChans) <= maxPendingResolutions {
+		return nil
+	}
+
+	ch := f.mu.cancelChans[0]
+	f.mu.cancelChans[0] = nil
+	f.mu.cancelChans = f.mu.cancelChans[1:]
+	if l := len(f.mu.cancelChans); l > maxPendingResolutions {
+		panic(fmt.Sprintf("max pending resolutions reached; got %d active resolutions, max = %d", l, maxPendingResolutions))
+	}
+
+	packets, ok := f.mu.packets[ch]
+	if !ok {
+		panic("must have a packet queue for an uncancelled channel")
+	}
+	delete(f.mu.packets, ch)
+
+	return packets
+}
+
+func (f *packetsPendingLinkResolution) dequeuePackets(packets []pendingPacket, linkAddr tcpip.LinkAddress, success bool) {
+	for _, p := range packets {
+		if success {
+			p.routeInfo.RemoteLinkAddress = linkAddr
+			_, _ = f.writePacketBuffer(p.routeInfo, p.gso, p.proto, p.pkt)
+		} else {
+			f.incrementOutgoingPacketErrors(p.proto, p.pkt)
+
+			if linkResolvableEP, ok := f.nic.getNetworkEndpoint(p.proto).(LinkResolvableNetworkEndpoint); ok {
 				switch pkt := p.pkt.(type) {
 				case *PacketBuffer:
-					p.route.outgoingNIC.writePacket(p.route.Fields(), nil /* gso */, p.proto, pkt)
+					linkResolvableEP.HandleLinkResolutionFailure(pkt)
 				case *PacketBufferList:
-					p.route.outgoingNIC.writePackets(p.route.Fields(), nil /* gso */, p.proto, *pkt)
+					for pb := pkt.Front(); pb != nil; pb = pb.Next() {
+						linkResolvableEP.HandleLinkResolutionFailure(pb)
+					}
 				default:
 					panic(fmt.Sprintf("unrecognized pending packet buffer type = %T", p.pkt))
 				}
 			}
-			p.route.Release()
 		}
-	}()
-}
-
-// newCancelChannel creates a channel that can cancel a pending forwarding
-// activity. The oldest channel is closed if the number of open channels would
-// exceed maxPendingResolutions.
-func (f *packetsPendingLinkResolution) newCancelChannelLocked() chan struct{} {
-	if len(f.cancelChans) == maxPendingResolutions {
-		ch := f.cancelChans[0]
-		f.cancelChans[0] = nil
-		f.cancelChans = f.cancelChans[1:]
-		close(ch)
 	}
-	if l := len(f.cancelChans); l >= maxPendingResolutions {
-		panic(fmt.Sprintf("max pending resolutions reached; got %d active resolutions, max = %d", l, maxPendingResolutions))
-	}
-
-	ch := make(chan struct{})
-	f.cancelChans = append(f.cancelChans, ch)
-	return ch
 }