From 19c7ca8c3bd9dd8bfe8c657845d79c752c9f3ff6 Mon Sep 17 00:00:00 2001 From: Lucas Manning Date: Wed, 1 May 2024 13:14:35 -0700 Subject: [PATCH] Add fdbased processor goroutines. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Results from iperf microbenchmark: ``` name old speed new speed delta IperfOneConnection/operation.Upload-16 226GB/s ± 1% 240GB/s ± 2% +6.44% (p=0.008 n=5+5) IperfOneConnection/operation.Download-16 330GB/s ± 3% 350GB/s ± 2% +6.12% (p=0.016 n=5+4) IperfManyConnections/operation.Upload/parallel.4-16 617GB/s ± 2% 644GB/s ± 1% +4.43% (p=0.016 n=5+4) IperfManyConnections/operation.Download/parallel.4-16 302GB/s ± 1% 337GB/s ± 3% +11.85% (p=0.008 n=5+5) IperfManyConnections/operation.Upload/parallel.16-16 702GB/s ± 3% 640GB/s ± 4% -8.82% (p=0.008 n=5+5) IperfManyConnections/operation.Download/parallel.16-16 267GB/s ± 1% 305GB/s ± 1% +14.17% (p=0.008 n=5+5) IperfManyConnections/operation.Upload/parallel.64-16 486GB/s ± 7% 655GB/s ± 1% +34.88% (p=0.008 n=5+5) IperfManyConnections/operation.Download/parallel.64-16 181GB/s ± 1% 215GB/s ± 1% +18.78% (p=0.008 n=5+5) ``` PiperOrigin-RevId: 629815554 --- pkg/tcpip/header/ipv4.go | 12 + pkg/tcpip/header/ipv6.go | 12 + pkg/tcpip/link/fdbased/BUILD | 4 + pkg/tcpip/link/fdbased/endpoint.go | 8 +- pkg/tcpip/link/fdbased/endpoint_test.go | 6 +- pkg/tcpip/link/fdbased/mmap.go | 71 +++-- pkg/tcpip/link/fdbased/mmap_stub.go | 2 +- pkg/tcpip/link/fdbased/mmap_unsafe.go | 4 +- pkg/tcpip/link/fdbased/packet_dispatchers.go | 76 ++---- pkg/tcpip/link/fdbased/processors.go | 263 +++++++++++++++++++ pkg/tcpip/stack/packet_buffer_list.go | 12 + runsc/config/flags.go | 2 +- 12 files changed, 370 insertions(+), 102 deletions(-) create mode 100644 pkg/tcpip/link/fdbased/processors.go diff --git a/pkg/tcpip/header/ipv4.go b/pkg/tcpip/header/ipv4.go index 4fb77df23..6e566205d 100644 --- a/pkg/tcpip/header/ipv4.go +++ b/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)) diff --git a/pkg/tcpip/header/ipv6.go b/pkg/tcpip/header/ipv6.go index ed30f77b3..4260095c6 100644 --- a/pkg/tcpip/header/ipv6.go +++ b/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 { diff --git a/pkg/tcpip/link/fdbased/BUILD b/pkg/tcpip/link/fdbased/BUILD index 301a79874..8bb1ca63d 100644 --- a/pkg/tcpip/link/fdbased/BUILD +++ b/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", diff --git a/pkg/tcpip/link/fdbased/endpoint.go b/pkg/tcpip/link/fdbased/endpoint.go index fa64a703b..bff2d0475 100644 --- a/pkg/tcpip/link/fdbased/endpoint.go +++ b/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) } diff --git a/pkg/tcpip/link/fdbased/endpoint_test.go b/pkg/tcpip/link/fdbased/endpoint_test.go index 353bf4c29..1bbb01953 100644 --- a/pkg/tcpip/link/fdbased/endpoint_test.go +++ b/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) (linkDispatcher, error) + newDispatcher func(fd int, e *endpoint, opts *Options) (linkDispatcher, error) }{ { name: "readVDispatcher", @@ -590,7 +590,7 @@ func TestDispatchPacketFormat(t *testing.T) { }, { name: "recvMMsgDispatcher", - newDispatcher: func(fd int, e *endpoint) (linkDispatcher, error) { return newRecvMMsgDispatcher(fd, e, &Options{}) }, + newDispatcher: newRecvMMsgDispatcher, }, } { 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) } diff --git a/pkg/tcpip/link/fdbased/mmap.go b/pkg/tcpip/link/fdbased/mmap.go index a136a9fdd..6473e95a3 100644 --- a/pkg/tcpip/link/fdbased/mmap.go +++ b/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 } diff --git a/pkg/tcpip/link/fdbased/mmap_stub.go b/pkg/tcpip/link/fdbased/mmap_stub.go index 9d8679502..c76c4b655 100644 --- a/pkg/tcpip/link/fdbased/mmap_stub.go +++ b/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 } diff --git a/pkg/tcpip/link/fdbased/mmap_unsafe.go b/pkg/tcpip/link/fdbased/mmap_unsafe.go index abe07a5e1..c324d1b4b 100644 --- a/pkg/tcpip/link/fdbased/mmap_unsafe.go +++ b/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 } diff --git a/pkg/tcpip/link/fdbased/packet_dispatchers.go b/pkg/tcpip/link/fdbased/packet_dispatchers.go index ed418ff3d..95d19dd97 100644 --- a/pkg/tcpip/link/fdbased/packet_dispatchers.go +++ b/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 } diff --git a/pkg/tcpip/link/fdbased/processors.go b/pkg/tcpip/link/fdbased/processors.go new file mode 100644 index 000000000..1bb7c298c --- /dev/null +++ b/pkg/tcpip/link/fdbased/processors.go @@ -0,0 +1,263 @@ +// 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 + } +} diff --git a/pkg/tcpip/stack/packet_buffer_list.go b/pkg/tcpip/stack/packet_buffer_list.go index 226b3e495..363059a9b 100644 --- a/pkg/tcpip/stack/packet_buffer_list.go +++ b/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. // diff --git a/runsc/config/flags.go b/runsc/config/flags.go index ae7a2546a..a3e44a9e5 100644 --- a/runsc/config/flags.go +++ b/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:", or "tunnel:"`) 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.