Automated rollback of changelist 630263974

PiperOrigin-RevId: 633277180

pkg/tcpip/header/ipv4.go

@@ -346,6 +346,18 @@ 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,6 +225,18 @@ 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,13 +14,17 @@ 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,6 +42,7 @@ package fdbased
 
 import (
 	"fmt"
+	"runtime"
 
 	"golang.org/x/sys/unix"
 	"gvisor.dev/gvisor/pkg/atomicbitops"
@@ -322,6 +323,9 @@ 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 {
@@ -336,7 +340,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)
+	inboundDispatcher, err := newReadVDispatcher(fd, e, opts)
 	if err != nil {
 		return nil, fmt.Errorf("newReadVDispatcher(%d, %+v) = %v", fd, e, err)
 	}
@@ -376,7 +380,7 @@ func createInboundDispatcher(e *endpoint, fd int, isSocket bool, fID int32, opts
 
 		switch e.packetDispatchMode {
 		case PacketMMap:
-			inboundDispatcher, err = newPacketMMapDispatcher(fd, e)
+			inboundDispatcher, err = newPacketMMapDispatcher(fd, e, opts)
 			if err != nil {
 				return nil, fmt.Errorf("newPacketMMapDispatcher(%d, %+v) = %v", fd, e, err)
 			}

pkg/tcpip/link/fdbased/endpoint_test.go

@@ -582,15 +582,61 @@ func (*fakeNetworkDispatcher) DeliverLinkPacket(tcpip.NetworkProtocolNumber, *st
 func TestDispatchPacketFormat(t *testing.T) {
 	for _, test := range []struct {
 		name          string
-		newDispatcher func(fd int, e *endpoint) (linkDispatcher, error)
+		newDispatcher func(fd int, e *endpoint, opts *Options) (linkDispatcher, error)
+		ethHdr        []byte
+		netHdr        []byte
 	}{
 		{
 			name:          "readVDispatcher",
 			newDispatcher: newReadVDispatcher,
+			ethHdr: []byte{
+				1, 2, 3, 4, 5, 60,
+				1, 2, 3, 4, 5, 61,
+				8, 0,
+			},
+			netHdr: []byte{0x40, 0, 0, 0},
 		},
 		{
 			name:          "recvMMsgDispatcher",
-			newDispatcher: func(fd int, e *endpoint) (linkDispatcher, error) { return newRecvMMsgDispatcher(fd, e, &Options{}) },
+			newDispatcher: newRecvMMsgDispatcher,
+			ethHdr: []byte{
+				1, 2, 3, 4, 5, 60,
+				1, 2, 3, 4, 5, 61,
+				8, 0,
+			},
+			netHdr: []byte{0x40, 0, 0, 0},
+		},
+		{
+			name:          "readVDispatcherNoEth",
+			newDispatcher: newReadVDispatcher,
+			netHdr:        []byte{0x40, 0, 0, 0},
+		},
+		{
+			name:          "readVDispatcherNoEthIPv6",
+			newDispatcher: newReadVDispatcher,
+			netHdr: []byte{
+				0x60, 0, 0, 0, // IPv6 Preamble
+				0, 0x04, 0, 0, // IPv6 Preamble
+				byte(header.IPv6HopByHopOptionsExtHdrIdentifier), // Next header
+				0xff,                   // Hop limit
+				0, 0, 0, 0, 0, 0, 0, 0, // Src addr
+				0, 0, 0, 0, 0, 0, 0, 0,
+				0, 0, 0, 0, 0, 0, 0, 0, // Dst addr
+				0, 0, 0, 0, 0, 0, 0, 0,
+
+				// Hop by hop extension header.
+				byte(header.IPv6DestinationOptionsExtHdrIdentifier),
+				0x8,              // Length.
+				0, 0, 0, 0, 0, 0, // Padding.
+
+				// Destination options extension header.
+				0,                // Next header (none)
+				0x8,              // Length.
+				0, 0, 0, 0, 0, 0, // Padding.
+
+				// payload data.
+				1, 2, 3, 4,
+			},
 		},
 	} {
 		t.Run(test.name, func(t *testing.T) {
@@ -600,14 +646,7 @@ func TestDispatchPacketFormat(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			data := []byte{
-				// Ethernet header.
-				1, 2, 3, 4, 5, 60,
-				1, 2, 3, 4, 5, 61,
-				8, 0,
-				// Mock network header.
-				40, 41, 42, 43,
-			}
+			data := append(test.ethHdr, test.netHdr...)
 			err = unix.Sendmsg(fds[1], data, nil, nil, 0)
 			if err != nil {
 				t.Fatal(err)
@@ -616,9 +655,9 @@ func TestDispatchPacketFormat(t *testing.T) {
 			// Create and run dispatcher once.
 			sink := &fakeNetworkDispatcher{}
 			d, err := test.newDispatcher(fds[0], &endpoint{
-				hdrSize:    header.EthernetMinimumSize,
+				hdrSize:    len(test.ethHdr),
 				dispatcher: sink,
-			})
+			}, &Options{ProcessorsPerChannel: 1})
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -633,12 +672,21 @@ func TestDispatchPacketFormat(t *testing.T) {
 			}
 			pkt := sink.pkts[0]
 			defer pkt.DecRef()
-			if got, want := len(pkt.LinkHeader().Slice()), header.EthernetMinimumSize; got != want {
-				t.Errorf("pkt.LinkHeader().View().Size() = %d, want %d", got, want)
+			if len(test.ethHdr) > 0 {
+				if got, want := len(pkt.LinkHeader().Slice()), header.EthernetMinimumSize; got != want {
+					t.Errorf("pkt.LinkHeader().View().Size() = %d, want %d", got, want)
+				}
 			}
-			if got, want := pkt.Data().Size(), 4; got != want {
+			if got, want := pkt.Data().Size(), len(test.netHdr); got != want {
 				t.Errorf("pkt.Data().Size() = %d, want %d", got, want)
 			}
+			wantProto := header.IPv4ProtocolNumber
+			if header.IPVersion(test.netHdr[:]) == header.IPv6Version {
+				wantProto = header.IPv6ProtocolNumber
+			}
+			if pkt.NetworkProtocolNumber != wantProto {
+				t.Errorf("pkt.NetworkProtocolNumber = %d, want %d", pkt.NetworkProtocolNumber, wantProto)
+			}
 		})
 	}
 }

pkg/tcpip/link/fdbased/mmap.go

@@ -130,11 +130,17 @@ 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 (*packetMMapDispatcher) release() {}
+func (d *packetMMapDispatcher) release() {
+	d.mgr.close()
+}
 
-func (d *packetMMapDispatcher) readMMappedPacket() (*buffer.View, bool, tcpip.Error) {
+func (d *packetMMapDispatcher) readMMappedPackets() (stack.PacketBufferList, bool, tcpip.Error) {
+	var pkts stack.PacketBufferList
 	hdr := tPacketHdr(d.ringBuffer[d.ringOffset*tpFrameSize:])
 	for hdr.tpStatus()&tpStatusUser == 0 {
 		stopped, errno := rawfile.BlockingPollUntilStopped(d.EFD, d.fd, unix.POLLIN|unix.POLLERR)
@@ -142,10 +148,10 @@ func (d *packetMMapDispatcher) readMMappedPacket() (*buffer.View, bool, tcpip.Er
 			if errno == unix.EINTR {
 				continue
 			}
-			return nil, stopped, rawfile.TranslateErrno(errno)
+			return pkts, stopped, rawfile.TranslateErrno(errno)
 		}
 		if stopped {
-			return nil, true, nil
+			return pkts, true, nil
 		}
 		if hdr.tpStatus()&tpStatusCopy != 0 {
 			// This frame is truncated so skip it after flipping the
@@ -157,50 +163,39 @@ func (d *packetMMapDispatcher) readMMappedPacket() (*buffer.View, bool, tcpip.Er
 		}
 	}
 
-	// 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
+	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
 }
 
 // dispatch reads packets from an mmaped ring buffer and dispatches them to the
 // network stack.
 func (d *packetMMapDispatcher) dispatch() (bool, tcpip.Error) {
-	pkt, stopped, err := d.readMMappedPacket()
+	pkts, stopped, err := d.readMMappedPackets()
+	defer pkts.Reset()
 	if err != nil || stopped {
 		return false, err
 	}
-	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
+	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()
 		}
+		d.mgr.queuePacket(pkt, d.e.hdrSize > 0)
 	}
-
-	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))
-		}
+	if pkts.Len() > 0 {
+		d.mgr.wakeReady()
 	}
-	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) (linkDispatcher, error) {
+func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (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) (linkDispatcher, error) {
+func newPacketMMapDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) {
 	stopFD, err := stopfd.New()
 	if err != nil {
 		return nil, err
@@ -77,6 +77,8 @@ func newPacketMMapDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
 	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,9 +155,12 @@ 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) (linkDispatcher, error) {
+func newReadVDispatcher(fd int, e *endpoint, opts *Options) (linkDispatcher, error) {
 	stopFD, err := stopfd.New()
 	if err != nil {
 		return nil, err
@@ -169,11 +172,14 @@ func newReadVDispatcher(fd int, e *endpoint) (linkDispatcher, error) {
 	}
 	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.
@@ -188,35 +194,14 @@ 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
 		}
-		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
-		}
+		pkt.NetworkProtocolNumber = header.Ethernet(pkt.LinkHeader().Slice()).Type()
 	}
-
-	d.e.mu.RLock()
-	dsp := d.e.dispatcher
-	d.e.mu.RUnlock()
-	dsp.DeliverNetworkPacket(p, pkt)
-
+	d.mgr.queuePacket(pkt, d.e.hdrSize > 0)
+	d.mgr.wakeReady()
 	return true, nil
 }
 
@@ -244,6 +229,9 @@ type recvMMsgDispatcher struct {
 
 	// gro coalesces incoming packets to increase throughput.
 	gro gro.GRO
+
+	// mgr is the processor goroutine manager.
+	mgr *processorManager
 }
 
 const (
@@ -269,6 +257,8 @@ 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
 }
@@ -277,6 +267,7 @@ 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
@@ -318,44 +309,17 @@ 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
 			}
-			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.NetworkProtocolNumber = header.Ethernet(hdr).Type()
 		}
+		pkt.RXChecksumValidated = d.e.caps&stack.CapabilityRXChecksumOffload != 0
+		d.mgr.queuePacket(pkt, d.e.hdrSize > 0)
 	}
-	d.gro.Flush()
+	d.mgr.wakeReady()
 
 	return true, nil
 }

pkg/tcpip/link/fdbased/processors.go

@@ -0,0 +1,259 @@
+// 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
+	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 len(m.processors) == 1 || nonConnectionPkt {
+		// If the packet is not associated with an active connection, use the
+		// first processor.
+		pIdx = 0
+	} else {
+		pIdx = int(m.connectionHash(&cid)) % len(m.processors)
+	}
+	p := &m.processors[pIdx]
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	pkt.IncRef()
+	p.pkts.PushBack(pkt)
+	m.ready[pIdx] = true
+}
+
+type connectionID struct {
+	srcAddr, dstAddr []byte
+	srcPort, dstPort uint16
+	proto            tcpip.NetworkProtocolNumber
+}
+
+// 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 extension headers :(.
+			dataBuf := pkt.Data().ToBuffer()
+			dataBuf.TrimFront(header.IPv6MinimumSize)
+			it := header.MakeIPv6PayloadIterator(header.IPv6ExtensionHeaderIdentifier(ipHdr.NextHeader()), dataBuf)
+			defer it.Release()
+			for {
+				hdr, done, err := it.Next()
+				if done || err != nil {
+					break
+				}
+				hdr.Release()
+			}
+			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,6 +62,18 @@ 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", 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.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.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.