From a67dd1062313dbfcb8d12f58cc470f0062247b9e Mon Sep 17 00:00:00 2001 From: Lucas Manning Date: Mon, 13 May 2024 11:29:51 -0700 Subject: [PATCH] Automated rollback of changelist 630263974 PiperOrigin-RevId: 633277180 --- 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 | 78 ++++-- 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 | 259 +++++++++++++++++++ pkg/tcpip/stack/packet_buffer_list.go | 12 + runsc/config/flags.go | 2 +- 12 files changed, 426 insertions(+), 114 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 3168f1f7a..84a5b1d77 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 43e205da7..c9474eec4 100644 --- a/pkg/tcpip/link/fdbased/endpoint_test.go +++ b/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) + } }) } } 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..03b51fb4a --- /dev/null +++ b/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 + } +} 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 53b4cb024..96d6bd1f3 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.