Automated rollback of changelist 629815554

PiperOrigin-RevId: 630263974

pkg/tcpip/header/ipv4.go

@@ -346,18 +346,6 @@ func (b IPv4) DestinationAddress() tcpip.Address {
 	return tcpip.AddrFrom4([4]byte(b[dstAddr : dstAddr+IPv4AddressSize]))
 }
 
-// SourceAddressSlice returns the "source address" field of the IPv4 header as a
-// byte slice.
-func (b IPv4) SourceAddressSlice() []byte {
-	return []byte(b[srcAddr : srcAddr+IPv4AddressSize])
-}
-
-// DestinationAddressSlice returns the "destination address" field of the IPv4
-// header as a byte slice.
-func (b IPv4) DestinationAddressSlice() []byte {
-	return []byte(b[dstAddr : dstAddr+IPv4AddressSize])
-}
-
 // SetSourceAddressWithChecksumUpdate implements ChecksummableNetwork.
 func (b IPv4) SetSourceAddressWithChecksumUpdate(new tcpip.Address) {
 	b.SetChecksum(^checksumUpdate2ByteAlignedAddress(^b.Checksum(), b.SourceAddress(), new))

pkg/tcpip/header/ipv6.go

@@ -225,18 +225,6 @@ func (b IPv6) DestinationAddress() tcpip.Address {
 	return tcpip.AddrFrom16([16]byte(b[v6DstAddr:][:IPv6AddressSize]))
 }
 
-// SourceAddressSlice returns the "source address" field of the ipv6 header as a
-// byte slice.
-func (b IPv6) SourceAddressSlice() []byte {
-	return []byte(b[v6SrcAddr:][:IPv6AddressSize])
-}
-
-// DestinationAddressSlice returns the "destination address" field of the ipv6
-// header as a byte slice.
-func (b IPv6) DestinationAddressSlice() []byte {
-	return []byte(b[v6DstAddr:][:IPv6AddressSize])
-}
-
 // Checksum implements Network.Checksum. Given that IPv6 doesn't have a
 // checksum, it just returns 0.
 func (IPv6) Checksum() uint16 {

pkg/tcpip/link/fdbased/BUILD

@@ -14,17 +14,13 @@ go_library(
         "mmap_stub.go",
         "mmap_unsafe.go",
         "packet_dispatchers.go",
-        "processors.go",
     ],
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/atomicbitops",
         "//pkg/buffer",
-        "//pkg/rand",
-        "//pkg/sleep",
         "//pkg/sync",
         "//pkg/tcpip",
-        "//pkg/tcpip/hash/jenkins",
         "//pkg/tcpip/header",
         "//pkg/tcpip/link/rawfile",
         "//pkg/tcpip/link/stopfd",

pkg/tcpip/link/fdbased/endpoint.go

@@ -42,7 +42,6 @@ package fdbased
 
 import (
 	"fmt"
-	"runtime"
 
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/atomicbitops"
@@ -323,9 +322,6 @@ func New(opts *Options) (stack.LinkEndpoint, error) {
 				e.gsoMaxSize = opts.GSOMaxSize
 			}
 		}
-		if opts.ProcessorsPerChannel == 0 {
-			opts.ProcessorsPerChannel = max(1, runtime.GOMAXPROCS(0)/len(opts.FDs))
-		}
 
 		inboundDispatcher, err := createInboundDispatcher(e, fd, isSocket, fid, opts)
 		if err != nil {
@@ -340,7 +336,7 @@ func New(opts *Options) (stack.LinkEndpoint, 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, opts)
+	inboundDispatcher, err := newReadVDispatcher(fd, e)
 	if err != nil {
 		return nil, fmt.Errorf("newReadVDispatcher(%d, %+v) = %v", fd, e, err)
 	}
@@ -380,7 +376,7 @@ func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32, opts
 
 		switch e.packetDispatchMode {
 		case PacketMMap:
-			inboundDispatcher, err = newPacketMMapDispatcher(fd, e, opts)
+			inboundDispatcher, err = newPacketMMapDispatcher(fd, e)
 			if err != nil {
 				return nil, fmt.Errorf("newPacketMMapDispatcher(%d, %+v) = %v", fd, e, err)
 			}

pkg/tcpip/link/fdbased/endpoint_test.go

@@ -582,7 +582,7 @@ func (*fakeNetworkDispatcher) DeliverLinkPacket(tcpip.NetworkProtocolNumber, *st
 func TestDispatchPacketFormat(t *testing.T) {
 	for _, test := range []struct {
 		name          string
-		newDispatcher func(fd int, e *endpoint, opts *Options) (linkDispatcher, error)
+		newDispatcher func(fd int, e *endpoint) (linkDispatcher, error)
 	}{
 		{
 			name:          "readVDispatcher",
@@ -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) {
@@ -618,7 +618,7 @@ func TestDispatchPacketFormat(t *testing.T) {
 			d, err := test.newDispatcher(fds[0], &endpoint{
 				hdrSize:    header.EthernetMinimumSize,
 				dispatcher: sink,
-			}, &Options{ProcessorsPerChannel: 1})
+			})
 			if err != nil {
 				t.Fatal(err)
 			}

pkg/tcpip/link/fdbased/mmap.go

@@ -130,17 +130,11 @@ type packetMMapDispatcher struct {
 	// ringOffset is the current offset into the ring buffer where the next
 	// inbound packet will be placed by the kernel.
 	ringOffset int
-
-	// mgr is the processor goroutine manager.
-	mgr *processorManager
 }
 
-func (d *packetMMapDispatcher) release() {
-	d.mgr.close()
-}
+func (*packetMMapDispatcher) release() {}
 
-func (d *packetMMapDispatcher) readMMappedPackets() (stack.PacketBufferList, bool, tcpip.Error) {
-	var pkts stack.PacketBufferList
+func (d *packetMMapDispatcher) readMMappedPacket() (*buffer.View, bool, tcpip.Error) {
 	hdr := tPacketHdr(d.ringBuffer[d.ringOffset*tpFrameSize:])
 	for hdr.tpStatus()&tpStatusUser == 0 {
 		stopped, errno := rawfile.BlockingPollUntilStopped(d.EFD, d.fd, unix.POLLIN|unix.POLLERR)
@@ -148,10 +142,10 @@ func (d *packetMMapDispatcher) readMMappedPackets() (stack.PacketBufferList, boo
 			if errno == unix.EINTR {
 				continue
 			}
-			return pkts, stopped, rawfile.TranslateErrno(errno)
+			return nil, stopped, rawfile.TranslateErrno(errno)
 		}
 		if stopped {
-			return pkts, true, nil
+			return nil, true, nil
 		}
 		if hdr.tpStatus()&tpStatusCopy != 0 {
 			// This frame is truncated so skip it after flipping the
@@ -163,39 +157,50 @@ func (d *packetMMapDispatcher) readMMappedPackets() (stack.PacketBufferList, boo
 		}
 	}
 
-	for hdr.tpStatus()&tpStatusUser == 1 {
-		// Copy out the packet from the mmapped frame to a locally owned buffer.
-		pkts.PushBack(stack.NewPacketBuffer(stack.PacketBufferOptions{
-			Payload: buffer.MakeWithView(buffer.NewViewWithData(hdr.Payload())),
-		}))
-		// Release packet to kernel.
-		hdr.setTPStatus(tpStatusKernel)
-		d.ringOffset = (d.ringOffset + 1) % tpFrameNR
-		hdr = tPacketHdr(d.ringBuffer[d.ringOffset*tpFrameSize:])
-	}
-	return pkts, false, nil
+	// Copy out the packet from the mmapped frame to a locally owned buffer.
+	pkt := buffer.NewView(int(hdr.tpSnapLen()))
+	pkt.Write(hdr.Payload())
+	// Release packet to kernel.
+	hdr.setTPStatus(tpStatusKernel)
+	d.ringOffset = (d.ringOffset + 1) % tpFrameNR
+	return pkt, false, nil
 }
 
 // dispatch reads packets from an mmaped ring buffer and dispatches them to the
 // network stack.
 func (d *packetMMapDispatcher) dispatch() (bool, tcpip.Error) {
-	pkts, stopped, err := d.readMMappedPackets()
-	defer pkts.Reset()
+	pkt, stopped, err := d.readMMappedPacket()
 	if err != nil || stopped {
 		return false, err
 	}
-	for _, pkt := range pkts.AsSlice() {
-		if d.e.hdrSize > 0 {
-			hdr, ok := pkt.LinkHeader().Consume(d.e.hdrSize)
-			if !ok {
-				panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", d.e.hdrSize))
-			}
-			pkt.NetworkProtocolNumber = header.Ethernet(hdr).Type()
+	var p tcpip.NetworkProtocolNumber
+	if d.e.hdrSize > 0 {
+		p = header.Ethernet(pkt.AsSlice()).Type()
+	} else {
+		// We don't get any indication of what the packet is, so try to guess
+		// if it's an IPv4 or IPv6 packet.
+		switch header.IPVersion(pkt.AsSlice()) {
+		case header.IPv4Version:
+			p = header.IPv4ProtocolNumber
+		case header.IPv6Version:
+			p = header.IPv6ProtocolNumber
+		default:
+			return true, nil
 		}
-		d.mgr.queuePacket(pkt, d.e.hdrSize > 0)
 	}
-	if pkts.Len() > 0 {
-		d.mgr.wakeReady()
+
+	pbuf := stack.NewPacketBuffer(stack.PacketBufferOptions{
+		Payload: buffer.MakeWithView(pkt),
+	})
+	defer pbuf.DecRef()
+	if d.e.hdrSize > 0 {
+		if _, ok := pbuf.LinkHeader().Consume(d.e.hdrSize); !ok {
+			panic(fmt.Sprintf("LinkHeader().Consume(%d) must succeed", d.e.hdrSize))
+		}
 	}
+	d.e.mu.RLock()
+	dsp := d.e.dispatcher
+	d.e.mu.RUnlock()
+	dsp.DeliverNetworkPacket(p, pbuf)
 	return true, nil
 }

pkg/tcpip/link/fdbased/mmap_stub.go

@@ -19,6 +19,6 @@ package fdbased
 
 // Stubbed out version for non-linux/non-amd64/non-arm64 platforms.
 
-func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) {
+func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
 	return nil, nil
 }

pkg/tcpip/link/fdbased/mmap_unsafe.go

@@ -47,7 +47,7 @@ func (t tPacketHdr) setTPStatus(status uint32) {
 	(*atomicbitops.Uint32)(statusPtr).Store(status)
 }
 
-func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) {
+func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
 	stopFD, err := stopfd.New()
 	if err != nil {
 		return nil, err
@@ -77,8 +77,6 @@ func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher
 	if err != nil {
 		return nil, fmt.Errorf("unix.Mmap(...,0, %v, ...) failed = %v", sz, err)
 	}
-	d.mgr = newProcessorManager(opts, e)
-	d.mgr.start()
 	d.ringBuffer = buf
 	return d, nil
 }

pkg/tcpip/link/fdbased/packet_dispatchers.go

@@ -155,12 +155,9 @@ type readVDispatcher struct {
 
 	// buf is the iovec buffer that contains the packet contents.
 	buf *iovecBuffer
-
-	// mgr is the processor goroutine manager.
-	mgr *processorManager
 }
 
-func newReadVDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) {
+func newReadVDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
 	stopFD, err := stopfd.New()
 	if err != nil {
 		return nil, err
@@ -172,14 +169,11 @@ func newReadVDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, err
 	}
 	skipsVnetHdr := d.e.gsoKind == stack.HostGSOSupported
 	d.buf = newIovecBuffer(BufConfig, skipsVnetHdr)
-	d.mgr = newProcessorManager(opts, e)
-	d.mgr.start()
 	return d, nil
 }
 
 func (d *readVDispatcher) release() {
 	d.buf.release()
-	d.mgr.close()
 }
 
 // dispatch reads one packet from the file descriptor and dispatches it.
@@ -194,14 +188,35 @@ func (d *readVDispatcher) dispatch() (bool, tcpip.Error) {
 	})
 	defer pkt.DecRef()
 
+	var p tcpip.NetworkProtocolNumber
 	if d.e.hdrSize > 0 {
 		if !d.e.parseHeader(pkt) {
 			return false, nil
 		}
-		pkt.NetworkProtocolNumber = header.Ethernet(pkt.LinkHeader().Slice()).Type()
+		p = header.Ethernet(pkt.LinkHeader().Slice()).Type()
+	} else {
+		// We don't get any indication of what the packet is, so try to guess
+		// if it's an IPv4 or IPv6 packet.
+		// IP version information is at the first octet, so pulling up 1 byte.
+		h, ok := pkt.Data().PullUp(1)
+		if !ok {
+			return true, nil
+		}
+		switch header.IPVersion(h) {
+		case header.IPv4Version:
+			p = header.IPv4ProtocolNumber
+		case header.IPv6Version:
+			p = header.IPv6ProtocolNumber
+		default:
+			return true, nil
+		}
 	}
-	d.mgr.queuePacket(pkt, d.e.hdrSize > 0)
-	d.mgr.wakeReady()
+
+	d.e.mu.RLock()
+	dsp := d.e.dispatcher
+	d.e.mu.RUnlock()
+	dsp.DeliverNetworkPacket(p, pkt)
+
 	return true, nil
 }
 
@@ -229,9 +244,6 @@ type recvMMsgDispatcher struct {
 
 	// gro coalesces incoming packets to increase throughput.
 	gro gro.GRO
-
-	// mgr is the processor goroutine manager.
-	mgr *processorManager
 }
 
 const (
@@ -257,8 +269,6 @@ func newRecvMMsgDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher,
 		d.bufs[i] = newIovecBuffer(BufConfig, skipsVnetHdr)
 	}
 	d.gro.Init(opts.GRO)
-	d.mgr = newProcessorManager(opts, e)
-	d.mgr.start()
 
 	return d, nil
 }
@@ -267,7 +277,6 @@ func (d *recvMMsgDispatcher) release() {
 	for _, iov := range d.bufs {
 		iov.release()
 	}
-	d.mgr.close()
 }
 
 // recvMMsgDispatch reads more than one packet at a time from the file
@@ -309,17 +318,44 @@ func (d *recvMMsgDispatcher) dispatch() (bool, tcpip.Error) {
 		// Mark that this iovec has been processed.
 		d.msgHdrs[k].Msg.Iovlen = 0
 
+		var p tcpip.NetworkProtocolNumber
 		if d.e.hdrSize > 0 {
 			hdr, ok := pkt.LinkHeader().Consume(d.e.hdrSize)
 			if !ok {
 				return false, nil
 			}
-			pkt.NetworkProtocolNumber = header.Ethernet(hdr).Type()
+			p = header.Ethernet(hdr).Type()
+		} else {
+			// We don't get any indication of what the packet is, so try to guess
+			// if it's an IPv4 or IPv6 packet.
+			// IP version information is at the first octet, so pulling up 1 byte.
+			h, ok := pkt.Data().PullUp(1)
+			if !ok {
+				// Skip this packet.
+				continue
+			}
+			switch header.IPVersion(h) {
+			case header.IPv4Version:
+				p = header.IPv4ProtocolNumber
+			case header.IPv6Version:
+				p = header.IPv6ProtocolNumber
+			default:
+				// Skip this packet.
+				continue
+			}
+		}
+
+		// 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
 		}
-		pkt.RXChecksumValidated = d.e.caps&stack.CapabilityRXChecksumOffload != 0
-		d.mgr.queuePacket(pkt, d.e.hdrSize > 0)
 	}
-	d.mgr.wakeReady()
+	d.gro.Flush()
 
 	return true, nil
 }

pkg/tcpip/link/fdbased/processors.go

@@ -1,263 +0,0 @@
-// Copyright 2024 The gVisor Authors.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//go:build linux
-// +build linux
-
-package fdbased
-
-import (
-	"encoding/binary"
-
-	"gvisor.dev/gvisor/pkg/rand"
-	"gvisor.dev/gvisor/pkg/sleep"
-	"gvisor.dev/gvisor/pkg/sync"
-	"gvisor.dev/gvisor/pkg/tcpip"
-	"gvisor.dev/gvisor/pkg/tcpip/hash/jenkins"
-	"gvisor.dev/gvisor/pkg/tcpip/header"
-	"gvisor.dev/gvisor/pkg/tcpip/stack"
-	"gvisor.dev/gvisor/pkg/tcpip/stack/gro"
-)
-
-type processor struct {
-	mu sync.Mutex
-	// +checklocks:mu
-	pkts stack.PacketBufferList
-
-	e           *endpoint
-	gro         gro.GRO
-	sleeper     sleep.Sleeper
-	packetWaker sleep.Waker
-	closeWaker  sleep.Waker
-}
-
-func (p *processor) start(wg *sync.WaitGroup) {
-	defer wg.Done()
-	defer p.sleeper.Done()
-	for {
-		switch w := p.sleeper.Fetch(true); {
-		case w == &p.packetWaker:
-			p.deliverPackets()
-		case w == &p.closeWaker:
-			p.mu.Lock()
-			p.pkts.Reset()
-			p.mu.Unlock()
-			return
-		}
-	}
-}
-
-func (p *processor) deliverPackets() {
-	p.e.mu.RLock()
-	p.gro.Dispatcher = p.e.dispatcher
-	p.e.mu.RUnlock()
-
-	p.mu.Lock()
-	for p.pkts.Len() > 0 {
-		pkt := p.pkts.PopFront()
-		p.mu.Unlock()
-		p.gro.Enqueue(pkt)
-		pkt.DecRef()
-		p.mu.Lock()
-	}
-	p.mu.Unlock()
-	p.gro.Flush()
-}
-
-// processorManager handles starting, closing, and queuing packets on processor
-// goroutines.
-type processorManager struct {
-	processors []processor
-	seed       uint32
-	wg         sync.WaitGroup
-	e          *endpoint
-	ready      []bool
-}
-
-// newProcessorManager creates a new processor manager.
-func newProcessorManager(opts *Options, e *endpoint) *processorManager {
-	m := &processorManager{}
-	m.seed = rand.Uint32()
-	m.ready = make([]bool, opts.ProcessorsPerChannel)
-	m.processors = make([]processor, opts.ProcessorsPerChannel)
-	m.e = e
-	m.wg.Add(opts.ProcessorsPerChannel)
-
-	for i := range m.processors {
-		p := &m.processors[i]
-		p.sleeper.AddWaker(&p.packetWaker)
-		p.sleeper.AddWaker(&p.closeWaker)
-		p.gro.Init(opts.GRO)
-		p.e = e
-	}
-
-	return m
-}
-
-// start starts the processor goroutines if the processor manager is configured
-// with more than one processor.
-func (m *processorManager) start() {
-	for i := range m.processors {
-		p := &m.processors[i]
-		// Only start processor in a separate goroutine if we have multiple of them.
-		if len(m.processors) > 1 {
-			go p.start(&m.wg)
-		}
-	}
-}
-
-func (m *processorManager) connectionHash(cid *connectionID) uint32 {
-	var payload [4]byte
-	binary.LittleEndian.PutUint16(payload[0:], cid.srcPort)
-	binary.LittleEndian.PutUint16(payload[2:], cid.dstPort)
-
-	h := jenkins.Sum32(m.seed)
-	h.Write(payload[:])
-	h.Write(cid.srcAddr)
-	h.Write(cid.dstAddr)
-	return h.Sum32()
-}
-
-// queuePacket queues a packet to be delivered to the appropriate processor.
-func (m *processorManager) queuePacket(pkt *stack.PacketBuffer, hasEthHeader bool) {
-	var pIdx int
-	if len(m.processors) > 1 {
-		cid, nonConnectionPkt := tcpipConnectionID(pkt)
-		if !hasEthHeader {
-			if nonConnectionPkt {
-				// If there's no eth header this should be a standard tcpip packet. If
-				// it isn't the packet is invalid so drop it.
-				return
-			}
-			pkt.NetworkProtocolNumber = cid.proto
-		}
-		// If the packet is not associated with an active connection, use the
-		// first processor.
-		if nonConnectionPkt {
-			pIdx = 0
-		} else {
-			pIdx = int(m.connectionHash(&cid)) % len(m.processors)
-		}
-	} else {
-		pIdx = 0
-	}
-	p := &m.processors[pIdx]
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	pkt.IncRef()
-	p.pkts.PushBack(pkt)
-	m.ready[pIdx] = true
-}
-
-type connectionID struct {
-	proto            tcpip.NetworkProtocolNumber
-	srcAddr, dstAddr []byte
-	srcPort, dstPort uint16
-}
-
-// tcpipConnectionID returns a tcpip connection id tuple based on the data found
-// in the packet. It returns true if the packet is not associated with an active
-// connection (e.g ARP, NDP, etc). The method assumes link headers have already
-// been processed if they were present.
-func tcpipConnectionID(pkt *stack.PacketBuffer) (connectionID, bool) {
-	var cid connectionID
-	h, ok := pkt.Data().PullUp(1)
-	if !ok {
-		// Skip this packet.
-		return cid, true
-	}
-
-	const tcpSrcDstPortLen = 4
-	switch header.IPVersion(h) {
-	case header.IPv4Version:
-		hdrLen := header.IPv4(h).HeaderLength()
-		h, ok = pkt.Data().PullUp(int(hdrLen) + tcpSrcDstPortLen)
-		if !ok {
-			return cid, true
-		}
-		ipHdr := header.IPv4(h[:hdrLen])
-		tcpHdr := header.TCP(h[hdrLen:][:tcpSrcDstPortLen])
-
-		cid.srcAddr = ipHdr.SourceAddressSlice()
-		cid.dstAddr = ipHdr.DestinationAddressSlice()
-		cid.srcPort = tcpHdr.SourcePort()
-		cid.dstPort = tcpHdr.DestinationPort()
-		cid.proto = header.IPv4ProtocolNumber
-	case header.IPv6Version:
-		h, ok = pkt.Data().PullUp(header.IPv6FixedHeaderSize + tcpSrcDstPortLen)
-		if !ok {
-			return cid, true
-		}
-		ipHdr := header.IPv6(h)
-
-		var tcpHdr header.TCP
-		if tcpip.TransportProtocolNumber(ipHdr.NextHeader()) == header.TCPProtocolNumber {
-			tcpHdr = header.TCP(h[header.IPv6FixedHeaderSize:][:tcpSrcDstPortLen])
-		} else {
-			// Slow path for IPv6 options :(.
-			dataBuf := pkt.Data().ToBuffer()
-			dataBuf.TrimFront(header.IPv6MinimumSize)
-			it := header.MakeIPv6PayloadIterator(header.IPv6ExtensionHeaderIdentifier(ipHdr.NextHeader()), dataBuf)
-			defer it.Release()
-			for {
-				_, done, err := it.Next()
-				if done || err != nil {
-					break
-				}
-			}
-			h, ok = pkt.Data().PullUp(int(it.HeaderOffset()) + tcpSrcDstPortLen)
-			if !ok {
-				return cid, true
-			}
-			tcpHdr = header.TCP(h[it.HeaderOffset():][:tcpSrcDstPortLen])
-		}
-		cid.srcAddr = ipHdr.SourceAddressSlice()
-		cid.dstAddr = ipHdr.DestinationAddressSlice()
-		cid.srcPort = tcpHdr.SourcePort()
-		cid.dstPort = tcpHdr.DestinationPort()
-		cid.proto = header.IPv6ProtocolNumber
-
-	default:
-		return cid, true
-	}
-	return cid, false
-}
-
-func (m *processorManager) close() {
-	if len(m.processors) < 2 {
-		return
-	}
-	for i := range m.processors {
-		p := &m.processors[i]
-		p.closeWaker.Assert()
-	}
-}
-
-// wakeReady wakes up all processors that have a packet queued. If there is only
-// one processor, the method delivers the packet inline without waking a
-// goroutine.
-func (m *processorManager) wakeReady() {
-	for i, ready := range m.ready {
-		if !ready {
-			continue
-		}
-		p := &m.processors[i]
-		if len(m.processors) > 1 {
-			p.packetWaker.Assert()
-		} else {
-			p.deliverPackets()
-		}
-		m.ready[i] = false
-	}
-}

pkg/tcpip/stack/packet_buffer_list.go

@@ -62,18 +62,6 @@ func (pl *PacketBufferList) PushBack(pb *PacketBuffer) {
 	pl.pbs = append(pl.pbs, pb)
 }
 
-// PopFront removes the first element in the list if it exists and returns it.
-//
-//go:nosplit
-func (pl *PacketBufferList) PopFront() *PacketBuffer {
-	if len(pl.pbs) == 0 {
-		return nil
-	}
-	pkt := pl.pbs[0]
-	pl.pbs = pl.pbs[1:]
-	return pkt
-}
-
 // DecRef decreases the reference count on each PacketBuffer
 // stored in the list.
 //

runsc/config/flags.go

@@ -119,7 +119,7 @@ func RegisterFlags(flagSet *flag.FlagSet) {
 	flagSet.Bool("rx-checksum-offload", true, "enable RX checksum offload.")
 	flagSet.Var(queueingDisciplinePtr(QDiscFIFO), "qdisc", "specifies which queueing discipline to apply by default to the non loopback nics used by the sandbox.")
 	flagSet.Int("num-network-channels", 1, "number of underlying channels(FDs) to use for network link endpoints.")
-	flagSet.Int("network-processors-per-channel", 0, "number of goroutines in each channel for processng inbound packets. If 0, the link endpoint will divide GOMAXPROCS evenly among the number of channels specified by num-network-channels.")
+	flagSet.Int("network-processors-per-channel", 1, "number of goroutines in each channel for processng inbound packets. If 0, the link endpoint will divide GOMAXPROCS evenly among the number of channels specified by num-network-channels.")
 	flagSet.Bool("buffer-pooling", true, "DEPRECATED: this flag has no effect. Buffer pooling is always enabled.")
 	flagSet.Var(&xdpConfig, "EXPERIMENTAL-xdp", `whether and how to use XDP. Can be one of: "off" (default), "ns", "redirect:<device name>", or "tunnel:<device name>"`)
 	flagSet.Bool("EXPERIMENTAL-xdp-need-wakeup", true, "EXPERIMENTAL. Use XDP_USE_NEED_WAKEUP with XDP sockets.") // TODO(b/240191988): Figure out whether this helps and remove it as a flag.