From 89df5a681c004a3772facc08de73d930636227be Mon Sep 17 00:00:00 2001 From: Ghanan Gowripalan Date: Thu, 21 Jan 2021 14:51:24 -0800 Subject: Queue packets in WritePackets when resolving link address Test: integration_test.TestWritePacketsLinkResolution Fixes #4458. PiperOrigin-RevId: 353108826 --- pkg/tcpip/stack/nic.go | 37 +++++++++++++++++++++++---- pkg/tcpip/stack/pending_packets.go | 51 +++++++++++++++++++++++++++++++------- 2 files changed, 74 insertions(+), 14 deletions(-) (limited to 'pkg/tcpip/stack') diff --git a/pkg/tcpip/stack/nic.go b/pkg/tcpip/stack/nic.go index d19150a20..447b5c99d 100644 --- a/pkg/tcpip/stack/nic.go +++ b/pkg/tcpip/stack/nic.go @@ -358,16 +358,43 @@ func (n *NIC) writePacket(r RouteInfo, gso *GSO, protocol tcpip.NetworkProtocolN // WritePackets implements NetworkLinkEndpoint. func (n *NIC) WritePackets(r *Route, gso *GSO, pkts PacketBufferList, protocol tcpip.NetworkProtocolNumber) (int, *tcpip.Error) { - // TODO(gvisor.dev/issue/4458): Queue packets whie link address resolution - // is being peformed like WritePacket. - routeInfo := r.Fields() + // As per relevant RFCs, we should queue packets while we wait for link + // resolution to complete. + // + // RFC 1122 section 2.3.2.2 (for IPv4): + // The link layer SHOULD save (rather than discard) at least + // one (the latest) packet of each set of packets destined to + // the same unresolved IP address, and transmit the saved + // packet when the address has been resolved. + // + // RFC 4861 section 7.2.2 (for IPv6): + // While waiting for address resolution to complete, the sender MUST, for + // each neighbor, retain a small queue of packets waiting for address + // resolution to complete. The queue MUST hold at least one packet, and MAY + // contain more. However, the number of queued packets per neighbor SHOULD + // be limited to some small value. When a queue overflows, the new arrival + // SHOULD replace the oldest entry. Once address resolution completes, the + // node transmits any queued packets. + if ch, err := r.Resolve(nil); err != nil { + if err == tcpip.ErrWouldBlock { + r.Acquire() + n.linkResQueue.enqueue(ch, r, protocol, &pkts) + return pkts.Len(), nil + } + return 0, err + } + + return n.writePackets(r.Fields(), gso, protocol, pkts) +} + +func (n *NIC) writePackets(r RouteInfo, gso *GSO, protocol tcpip.NetworkProtocolNumber, pkts PacketBufferList) (int, *tcpip.Error) { for pkt := pkts.Front(); pkt != nil; pkt = pkt.Next() { - pkt.EgressRoute = routeInfo + pkt.EgressRoute = r pkt.GSOOptions = gso pkt.NetworkProtocolNumber = protocol } - writtenPackets, err := n.LinkEndpoint.WritePackets(routeInfo, gso, pkts, protocol) + writtenPackets, err := n.LinkEndpoint.WritePackets(r, gso, pkts, protocol) n.stats.Tx.Packets.IncrementBy(uint64(writtenPackets)) writtenBytes := 0 for i, pb := 0, pkts.Front(); i < writtenPackets && pb != nil; i, pb = i+1, pb.Next() { diff --git a/pkg/tcpip/stack/pending_packets.go b/pkg/tcpip/stack/pending_packets.go index 81d8ff6e8..3ac039c7d 100644 --- a/pkg/tcpip/stack/pending_packets.go +++ b/pkg/tcpip/stack/pending_packets.go @@ -28,10 +28,26 @@ const ( maxPendingPacketsPerResolution = 256 ) +// pendingPacketBuffer is a pending packet buffer. +// +// TODO(gvisor.dev/issue/5331): Drop this when we drop WritePacket and only use +// WritePackets so we can use a PacketBufferList everywhere. +type pendingPacketBuffer interface { + len() int +} + +func (*PacketBuffer) len() int { + return 1 +} + +func (p *PacketBufferList) len() int { + return p.Len() +} + type pendingPacket struct { route *Route proto tcpip.NetworkProtocolNumber - pkt *PacketBuffer + pkt pendingPacketBuffer } // packetsPendingLinkResolution is a queue of packets pending link resolution. @@ -54,16 +70,17 @@ func (f *packetsPendingLinkResolution) init() { f.packets = make(map[<-chan struct{}][]pendingPacket) } -func incrementOutgoingPacketErrors(r *Route, proto tcpip.NetworkProtocolNumber) { - r.Stats().IP.OutgoingPacketErrors.Increment() +func incrementOutgoingPacketErrors(r *Route, proto tcpip.NetworkProtocolNumber, pkt pendingPacketBuffer) { + n := uint64(pkt.len()) + r.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 { - ipEndpointStats.IPStats().OutgoingPacketErrors.Increment() + ipEndpointStats.IPStats().OutgoingPacketErrors.IncrementBy(n) } } -func (f *packetsPendingLinkResolution) enqueue(ch <-chan struct{}, r *Route, proto tcpip.NetworkProtocolNumber, pkt *PacketBuffer) { +func (f *packetsPendingLinkResolution) enqueue(ch <-chan struct{}, r *Route, proto tcpip.NetworkProtocolNumber, pkt pendingPacketBuffer) { f.Lock() defer f.Unlock() @@ -73,7 +90,7 @@ func (f *packetsPendingLinkResolution) enqueue(ch <-chan struct{}, r *Route, pro packets[0] = pendingPacket{} packets = packets[1:] - incrementOutgoingPacketErrors(r, proto) + incrementOutgoingPacketErrors(r, proto, p.pkt) p.route.Release() } @@ -113,13 +130,29 @@ func (f *packetsPendingLinkResolution) enqueue(ch <-chan struct{}, r *Route, pro for _, p := range packets { if cancelled || p.route.IsResolutionRequired() { - incrementOutgoingPacketErrors(r, proto) + incrementOutgoingPacketErrors(r, proto, p.pkt) if linkResolvableEP, ok := p.route.outgoingNIC.getNetworkEndpoint(p.route.NetProto).(LinkResolvableNetworkEndpoint); ok { - linkResolvableEP.HandleLinkResolutionFailure(pkt) + 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 { - p.route.outgoingNIC.writePacket(p.route.Fields(), nil /* gso */, p.proto, p.pkt) + switch pkt := p.pkt.(type) { + case *PacketBuffer: + p.route.outgoingNIC.writePacket(p.route.Fields(), nil /* gso */, p.proto, pkt) + case *PacketBufferList: + p.route.outgoingNIC.writePackets(p.route.Fields(), nil /* gso */, p.proto, *pkt) + default: + panic(fmt.Sprintf("unrecognized pending packet buffer type = %T", p.pkt)) + } } p.route.Release() } -- cgit v1.2.3